Voting and AutoVoting improvements

benchmarks/sigmod23/autovoting_lt_dedalus/benchmark.py

@@ -13,7 +13,7 @@ def cluster_spec(self) -> Dict[str, Dict[str, int]]:
             return {
                 '1': {
                     'leaders': 1,
-                    'replicas': 10, # Max across any benchmark
+                    'replicas': 21, # Max across any benchmark
                     'collectors': 3, # Max across any benchmark
                     'broadcasters': 3, # Max across any benchmark
                     'clients': 1,
@@ -47,11 +47,11 @@ def gen_input(clients: int, replica_groups: int, replica_partitions: int, collec
             return [
                 gen_input(client_procs, replica_groups, replica_partitions, collectors, broadcasters)
 
-                for client_procs in [100, 300, 500]
-                for replica_groups in [3,5,10]
-                for replica_partitions in [1]
-                for collectors in [3]
-                for broadcasters in [3]
+                for client_procs in [10, 25, 40, 50, 60, 75, 100, 125, 150, 175, 200, 275, 400]
+                for replica_groups in [7]
+                for replica_partitions in [1,3]
+                for collectors in [1]
+                for broadcasters in [1]
             ]#*3
 
         def summary(self, input: Input, output: Output) -> str:

benchmarks/sigmod23/voting_lt_dedalus/benchmark.py

@@ -45,8 +45,8 @@ def gen_input(clients: int, replicas: int) -> Input:
                 # for client_procs in [1, 10, 25, 40, 50, 60, 75, 100, 125, 150, 175]
                 # for num_replicas in [3, 5]
 
-                for client_procs in [100, 300, 500]
-                for num_replicas in [3,5,10]
+                for client_procs in [10, 25, 40, 50, 60, 75, 100, 125, 150, 175, 200, 275, 400]
+                for num_replicas in [7]
             ]#*3
 
         def summary(self, input: Input, output: Output) -> str:

benchmarks/voting/smoke.py

@@ -12,7 +12,7 @@ def cluster_spec(self) -> Dict[str, Dict[str, int]]:
         return {
             '1': {
                 'leaders': 1,
-                'replicas': 5, # Max across any benchmark
+                'replicas': 7, # Max across any benchmark
                 'clients': 1,
             },
         }
@@ -39,7 +39,7 @@ def gen_input(clients: int, replicas: int) -> Input:
 
         return [
             # gen_input(10, 5),
-            gen_input(100, 5),
+            gen_input(100, 7),
         ]
 
         # return [

benchmarks/voting/voting.py

@@ -65,7 +65,7 @@ class Placement(NamedTuple):
         replicas: List[host.Endpoint]
 
     def prom_placement(self) -> Placement:
-        ports = itertools.count(40001, 100)
+        ports = itertools.count(30001, 100)
 
         def portify_one(e: host.PartialEndpoint) -> host.Endpoint:
             return host.Endpoint(e.host, next(ports) if self._input.monitored else -1)

rust/examples/autovoting/main.rs

@@ -38,9 +38,8 @@ struct Args {
 #[tokio::main]
 async fn main() {
     let args = Args::parse();
-    let ports = hydroflow::util::cli::init().await;
-
     frankenpaxos::serve_prometheus(args.prometheus_host, args.prometheus_port);
+    let ports = hydroflow::util::cli::init().await;
 
     match args.service {
         Service::Leader => leader::run(args.leader, ports).await,

rust/examples/voting/main.rs

@@ -28,9 +28,8 @@ struct Args {
 #[tokio::main]
 async fn main() {
     let args = Args::parse();
-    let ports = hydroflow::util::cli::init().await;
-
     frankenpaxos::serve_prometheus(args.prometheus_host, args.prometheus_port);
+    let ports = hydroflow::util::cli::init().await;
 
     match args.service {
         Service::Leader => leader::run(args.leader, ports).await,

rust/src/lib.rs

@@ -3,7 +3,8 @@ pub fn serve_prometheus(host: String, port: i32) {
         return;
     }
     let addr = format!("{}:{}", host, port).parse().unwrap();
-    prometheus_exporter::start(addr).expect("failed to start Prometheus");
+    prometheus_exporter::start(addr)
+        .expect(format!("failed to start Prometheus for host {host}, port {port}").as_str());
 }
 
 pub mod echo_proto {

shared/src/main/scala/frankenpaxos/voting/Leader.scala

@@ -100,14 +100,15 @@ class Leader[Transport <: frankenpaxos.Transport[Transport]](
             VoteRequest(id = request.id, clientAddress = request.clientAddress)
           )
         )
-        if (numMessagesSinceLastFlush >= flushEveryN) {
-          replicas.foreach(_.flush())
-          clients.foreach(_._2.flush())
-          numMessagesSinceLastFlush = 0
-        }
 
       case LeaderInbound.Request.Empty =>
         logger.fatal("Empty LeaderInbound encountered.")
     }
+
+    if (numMessagesSinceLastFlush >= flushEveryN) {
+      replicas.foreach(_.flush())
+      clients.foreach(_._2.flush())
+      numMessagesSinceLastFlush = 0
+    }
   }
 }