feat(hydroflow_plus): add paxos

hydroflow_plus/src/builder/mod.rs

@@ -262,6 +262,23 @@ impl<'a> FlowBuilder<'a> {
         )
     }
 
+    pub fn source_interval_delayed<L: Location>(
+        &self,
+        on: &L,
+        delay: impl Quoted<'a, Duration> + Copy + 'a,
+        interval: impl Quoted<'a, Duration> + Copy + 'a,
+    ) -> Stream<'a, hydroflow::tokio::time::Instant, Async, L> {
+        self.source_stream(
+            on,
+            q!(hydroflow::tokio_stream::wrappers::IntervalStream::new(
+                hydroflow::tokio::time::interval_at(
+                    hydroflow::tokio::time::Instant::now() + delay,
+                    interval
+                )
+            )),
+        )
+    }
+
     pub fn cycle<T, W, L: Location>(&self, on: &L) -> (HfCycle<'a, T, W, L>, Stream<'a, T, W, L>) {
         let next_id = {
             let on_id = match on.id() {

hydroflow_plus_test/examples/paxos.rs

@@ -0,0 +1,65 @@
+use hydro_deploy::Deployment;
+use hydroflow_plus_cli_integration::TrybuildHost;
+
+#[tokio::main]
+async fn main() {
+    let mut deployment = Deployment::new();
+    let localhost = deployment.Localhost();
+
+    let builder = hydroflow_plus::FlowBuilder::new();
+    let f = 1;
+    let num_clients = 1;
+    let num_clients_per_node = 1; // Change based on experiment between 1, 50, 100.
+    let median_latency_window_size = 1000;
+    let checkpoint_frequency = 1000; // Num log entries
+    let i_am_leader_send_timeout = 5; // Sec
+    let i_am_leader_check_timeout = 10; // Sec
+    let i_am_leader_check_timeout_delay_multiplier = 15;
+
+    let (proposers, acceptors, clients, replicas) = hydroflow_plus_test::cluster::paxos::paxos(
+        &builder,
+        f,
+        num_clients_per_node,
+        median_latency_window_size,
+        checkpoint_frequency,
+        i_am_leader_send_timeout,
+        i_am_leader_check_timeout,
+        i_am_leader_check_timeout_delay_multiplier,
+    );
+
+    let rustflags = "-C opt-level=3 -C codegen-units=1 -C strip=none -C debuginfo=2 -C lto=off";
+
+    let _nodes = builder
+        .with_default_optimize()
+        .with_cluster(
+            &proposers,
+            (0..f + 1)
+                .map(|_| TrybuildHost::new(localhost.clone()).rustflags(rustflags))
+                .collect::<Vec<_>>(),
+        )
+        .with_cluster(
+            &acceptors,
+            (0..2 * f + 1)
+                .map(|_| TrybuildHost::new(localhost.clone()).rustflags(rustflags))
+                .collect::<Vec<_>>(),
+        )
+        .with_cluster(
+            &clients,
+            (0..num_clients)
+                .map(|_| TrybuildHost::new(localhost.clone()).rustflags(rustflags))
+                .collect::<Vec<_>>(),
+        )
+        .with_cluster(
+            &replicas,
+            (0..f + 1)
+                .map(|_| TrybuildHost::new(localhost.clone()).rustflags(rustflags))
+                .collect::<Vec<_>>(),
+        )
+        .deploy(&mut deployment);
+
+    deployment.deploy().await.unwrap();
+
+    deployment.start().await.unwrap();
+
+    tokio::signal::ctrl_c().await.unwrap();
+}

hydroflow_plus_test/src/cluster/mod.rs

@@ -1,4 +1,5 @@
 pub mod compute_pi;
 pub mod many_to_many;
 pub mod map_reduce;
+pub mod paxos;
 pub mod simple_cluster;