refactor(paxos): minor additional cleanups

hydro-project · Oct 31, 2024 · c06cd6e · c06cd6e
1 parent 275a0ed
commit c06cd6e
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 17 deletions.
diff --git a/hydroflow_plus_test/src/cluster/paxos.rs b/hydroflow_plus_test/src/cluster/paxos.rs
@@ -90,13 +90,14 @@ pub fn paxos_core<'a, P: PaxosPayload, R>(
     let proposers = flow.cluster::<Proposer>();
     let acceptors = flow.cluster::<Acceptor>();
 
-    let c_to_proposers = c_to_proposers(&proposers);
-
-    // Proposers.
     proposers
         .source_iter(q!(["Proposers say hello"]))
         .for_each(q!(|s| println!("{}", s)));
 
+    acceptors
+        .source_iter(q!(["Acceptors say hello"]))
+        .for_each(q!(|s| println!("{}", s)));
+
     let (a_to_proposers_p2b_complete_cycle, a_to_proposers_p2b_forward_reference) =
         proposers.forward_ref::<Stream<_, _, _, _>>();
     let (a_log_complete_cycle, a_log_forward_reference) =
@@ -114,6 +115,8 @@ pub fn paxos_core<'a, P: PaxosPayload, R>(
             a_log_forward_reference,
         );
 
+    let c_to_proposers = c_to_proposers(&proposers);
+
     let (p_to_clients_new_leader_elected, p_to_replicas, a_log, a_to_proposers_p2b) =
         sequence_payload(
             &proposers,
@@ -549,9 +552,6 @@ fn sequence_payload<'a, P: PaxosPayload, R>(
         .all_ticks();
 
     // Acceptors.
-    acceptors
-        .source_iter(q!(["Acceptors say hello"]))
-        .for_each(q!(|s| println!("{}", s)));
     let r_to_acceptors_checkpoint = r_to_acceptors_checkpoint(acceptors);
 
     // p_to_acceptors_p2a.clone().for_each(q!(|p2a: P2a| println!("Acceptor received P2a: {:?}", p2a)));

diff --git a/...est/src/cluster/snapshots/hydroflow_plus_test__cluster__paxos_bench__tests__paxos_ir.snap b/...est/src/cluster/snapshots/hydroflow_plus_test__cluster__paxos_bench__tests__paxos_ir.snap
@@ -14,6 +14,17 @@ expression: built.ir()
             ),
         },
     },
+    ForEach {
+        f: stageleft :: runtime_support :: fn1_type_hint :: < & str , () > ({ use crate :: __staged :: cluster :: paxos :: * ; | s | println ! ("{}" , s) }),
+        input: Source {
+            source: Iter(
+                { use crate :: __staged :: cluster :: paxos :: * ; ["Acceptors say hello"] },
+            ),
+            location_kind: Cluster(
+                3,
+            ),
+        },
+    },
     CycleSink {
         ident: Ident {
             sym: cycle_4,
@@ -655,17 +666,6 @@ expression: built.ir()
             ),
         ),
     },
-    ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < & str , () > ({ use crate :: __staged :: cluster :: paxos :: * ; | s | println ! ("{}" , s) }),
-        input: Source {
-            source: Iter(
-                { use crate :: __staged :: cluster :: paxos :: * ; ["Acceptors say hello"] },
-            ),
-            location_kind: Cluster(
-                3,
-            ),
-        },
-    },
     CycleSink {
         ident: Ident {
             sym: cycle_6,