rebase

benchmark/benchmark/full_demo.py

@@ -93,13 +93,8 @@ def run(self, debug=False):
             self.node_parameters.print(PathMaker.parameters_file())
 
             # Run the clients (they will wait for the nodes to be ready).
-<< << << < HEAD
-            workers_addresses = committee.workers_addresses(self.faults)
-            rate_share = ceil(rate / committee.workers())
-== == == =
             workers_addresses = worker_cache.workers_addresses(self.faults)
             rate_share = ceil(rate / worker_cache.workers())
->>>>>> > cd36ad6d(Fix local benchmark, client demo & tests)
             for i, addresses in enumerate(workers_addresses):
                 for (id, address) in addresses:
                     cmd = CommandMaker.run_client(

benchmark/benchmark/remote.py

@@ -149,7 +149,7 @@ def _update(self, hosts, bench_parameters):
             mem_profiling=bench_parameters.mem_profile))
         cmd = [
             f'(cd {self.settings.repo_name} && git fetch -f)',
-            f'(cd {self.settings.repo_name} && git checkout -f {self.settings.branch})',
+            f'(cd {self.settings.repo_name} && git checkout -f {self.settings.branch} --)',
             f'(cd {self.settings.repo_name} && git pull -f)',
             'source $HOME/.cargo/env',
             f'(cd {self.settings.repo_name}/node && {compile_cmd})',

executor/src/lib.rs

@@ -24,7 +24,7 @@ use crate::{
     batch_loader::BatchLoader, core::Core, metrics::ExecutorMetrics, subscriber::Subscriber,
 };
 use async_trait::async_trait;
-use config::SharedWorkerCache;
+use config::{SharedCommittee, SharedWorkerCache};
 use consensus::ConsensusOutput;
 use crypto::PublicKey;
 use prometheus::Registry;
@@ -100,6 +100,7 @@ impl Executor {
     /// Spawn a new client subscriber.
     pub async fn spawn<State>(
         name: PublicKey,
+        committee: SharedCommittee,
         worker_cache: SharedWorkerCache,
         store: Store<BatchDigest, SerializedBatchMessage>,
         execution_state: Arc<State>,

node/src/lib.rs

@@ -290,6 +290,7 @@ impl Node {
         // subscriber handler if it missed some transactions.
         let executor_handles = Executor::spawn(
             name,
+            committee,
             worker_cache,
             store.batch_store.clone(),
             execution_state,

primary/tests/epoch_change.rs

@@ -458,6 +458,7 @@ async fn test_simple_committee_update() {
     // The configuration of epoch 0.
     let keys_0 = keys(None);
     let committee_0 = pure_committee_from_keys(&keys_0);
+    let worker_cache_0 = worker_cache_from_keys(&keys_0);
 
     // Spawn the committee of epoch 0.
     let mut rx_channels = Vec::new();
@@ -482,6 +483,7 @@ async fn test_simple_committee_update() {
             name,
             signer,
             Arc::new(ArcSwap::from_pointee(committee_0.clone())),
+            Arc::new(ArcSwap::from_pointee(worker_cache_0.clone())),
             parameters.clone(),
             store.header_store.clone(),
             store.certificate_store.clone(),

primary/tests/nodes_bootstrapping_tests.rs

@@ -112,7 +112,7 @@ async fn test_loss_of_liveness_without_recovery() {
     let node_advance_delay = Duration::from_secs(60);
 
     // A cluster of 4 nodes will be created
-    let mut cluster = Cluster::new(None, None, true);
+    let mut cluster = Cluster::new(None, None, None, true);
 
     // ===== Start the cluster ====
     cluster.start(Some(4), Some(1), None).await;
@@ -163,7 +163,7 @@ async fn test_loss_of_liveness_with_recovery() {
     let node_advance_delay = Duration::from_secs(60);
 
     // A cluster of 4 nodes will be created
-    let mut cluster = Cluster::new(None, None, true);
+    let mut cluster = Cluster::new(None, None, None, true);
 
     // ===== Start the cluster ====
     cluster.start(Some(4), Some(1), None).await;

test_utils/src/lib.rs

@@ -14,7 +14,7 @@ use indexmap::IndexMap;
 use multiaddr::Multiaddr;
 use rand::{rngs::StdRng, Rng, SeedableRng as _};
 use std::{
-    collections::{BTreeSet, HashMap, VecDeque},
+    collections::{BTreeSet, VecDeque},
     ops::RangeInclusive,
     pin::Pin,
     sync::Arc,

worker/src/tests/quorum_waiter_tests.rs

@@ -3,7 +3,6 @@
 // SPDX-License-Identifier: Apache-2.0
 use super::*;
 use crate::worker::WorkerMessage;
-use crypto::traits::KeyPair;
 use test_utils::{batch, resolve_name_and_committee_and_worker_cache, WorkerToWorkerMockServer};
 
 #[tokio::test]