External sender queue implementation

examples/simulation.rs

@@ -26,6 +26,7 @@ use signifix::{metric, TryFrom};
 
 use hbbft::dynamic_honey_badger::DynamicHoneyBadger;
 use hbbft::queueing_honey_badger::{Batch, QueueingHoneyBadger};
+use hbbft::sender_queue::{Message, SenderQueue};
 use hbbft::{DistAlgorithm, NetworkInfo, Step, Target};
 
 const VERSION: &str = env!("CARGO_PKG_VERSION");
@@ -347,14 +348,16 @@ struct EpochInfo {
     nodes: BTreeMap<NodeId, (Duration, Batch<Transaction, NodeId>)>,
 }
 
+type QHB = SenderQueue<QueueingHoneyBadger<Transaction, NodeId, Vec<Transaction>>>;
+
 impl EpochInfo {
     /// Adds a batch to this epoch. Prints information if the epoch is complete.
     fn add(
         &mut self,
         id: NodeId,
         time: Duration,
         batch: &Batch<Transaction, NodeId>,
-        network: &TestNetwork<QueueingHoneyBadger<Transaction, NodeId, Vec<Transaction>>>,
+        network: &TestNetwork<QHB>,
     ) {
         if self.nodes.contains_key(&id) {
             return;
@@ -385,9 +388,7 @@ impl EpochInfo {
 }
 
 /// Proposes `num_txs` values and expects nodes to output and order them.
-fn simulate_honey_badger(
-    mut network: TestNetwork<QueueingHoneyBadger<Transaction, NodeId, Vec<Transaction>>>,
-) {
+fn simulate_honey_badger(mut network: TestNetwork<QHB>) {
     // Handle messages until all nodes have output all transactions.
     println!(
         "{}",
@@ -436,11 +437,20 @@ fn main() {
         .map(|_| Transaction::new(args.flag_tx_size))
         .collect();
     let new_honey_badger = |netinfo: NetworkInfo<NodeId>| {
-        let dyn_hb = DynamicHoneyBadger::builder().build(netinfo);
-        QueueingHoneyBadger::builder(dyn_hb)
+        let our_id = *netinfo.our_id();
+        let peer_ids: Vec<_> = netinfo
+            .all_ids()
+            .filter(|&&them| them != our_id)
+            .cloned()
+            .collect();
+        let dhb = DynamicHoneyBadger::builder().build(netinfo);
+        let (qhb, qhb_step) = QueueingHoneyBadger::builder(dhb)
             .batch_size(args.flag_b)
             .build_with_transactions(txs.clone(), rand::thread_rng().gen::<Isaac64Rng>())
-            .expect("instantiate QueueingHoneyBadger")
+            .expect("instantiate QueueingHoneyBadger");
+        let (sq, mut step) = SenderQueue::builder(qhb, peer_ids.into_iter()).build(our_id);
+        step.extend_with(qhb_step, Message::from);
+        (sq, step)
     };
     let hw_quality = HwQuality {
         latency: Duration::from_millis(args.flag_lag),

src/dynamic_honey_badger/batch.rs

@@ -10,6 +10,8 @@ use {NetworkInfo, NodeIdT};
 pub struct Batch<C, N> {
     /// The sequence number: there is exactly one batch in each epoch.
     pub(super) epoch: u64,
+    /// The current `DynamicHoneyBadger` era.
+    pub(super) era: u64,
     /// The user contributions committed in this epoch.
     pub(super) contributions: BTreeMap<N, C>,
     /// The current state of adding or removing a node: whether any is in progress, or completed
@@ -22,10 +24,16 @@ pub struct Batch<C, N> {
 }
 
 impl<C, N: NodeIdT> Batch<C, N> {
+    /// Returns the linear epoch of this `DynamicHoneyBadger` batch.
     pub fn epoch(&self) -> u64 {
         self.epoch
     }
 
+    /// Returns the `DynamicHoneyBadger` era of the batch.
+    pub fn era(&self) -> u64 {
+        self.era
+    }
+
     /// Returns whether any change to the set of participating nodes is in progress or was
     /// completed in this epoch.
     pub fn change(&self) -> &ChangeState<N> {
@@ -89,7 +97,7 @@ impl<C, N: NodeIdT> Batch<C, N> {
             return None;
         }
         Some(JoinPlan {
-            epoch: self.epoch + 1,
+            era: self.epoch + 1,
             change: self.change.clone(),
             pub_key_set: self.netinfo.public_key_set().clone(),
             pub_keys: self.netinfo.public_key_map().clone(),
@@ -104,6 +112,7 @@ impl<C, N: NodeIdT> Batch<C, N> {
         C: PartialEq,
     {
         self.epoch == other.epoch
+            && self.era == other.era
             && self.contributions == other.contributions
             && self.change == other.change
             && self.netinfo.public_key_set() == other.netinfo.public_key_set()

src/dynamic_honey_badger/builder.rs

@@ -16,10 +16,10 @@ use {Contribution, NetworkInfo, NodeIdT};
 /// A Dynamic Honey Badger builder, to configure the parameters and create new instances of
 /// `DynamicHoneyBadger`.
 pub struct DynamicHoneyBadgerBuilder<C, N> {
-    /// Start in this epoch.
-    epoch: u64,
+    /// Start in this era.
+    era: u64,
     /// The maximum number of future epochs for which we handle messages simultaneously.
-    max_future_epochs: usize,
+    max_future_epochs: u64,
     /// Random number generator passed on to algorithm instance for key generation. Also used to
     /// instantiate `HoneyBadger`.
     rng: Box<dyn rand::Rng>,
@@ -30,11 +30,14 @@ pub struct DynamicHoneyBadgerBuilder<C, N> {
     _phantom: PhantomData<(C, N)>,
 }
 
-impl<C, N> Default for DynamicHoneyBadgerBuilder<C, N> {
+impl<C, N> Default for DynamicHoneyBadgerBuilder<C, N>
+where
+    N: Ord,
+{
     fn default() -> Self {
         // TODO: Use the defaults from `HoneyBadgerBuilder`.
         DynamicHoneyBadgerBuilder {
-            epoch: 0,
+            era: 0,
             max_future_epochs: 3,
             rng: Box::new(rand::thread_rng()),
             subset_handling_strategy: SubsetHandlingStrategy::Incremental,
@@ -55,14 +58,14 @@ where
         Self::default()
     }
 
-    /// Sets the starting epoch to the given value.
-    pub fn epoch(&mut self, epoch: u64) -> &mut Self {
-        self.epoch = epoch;
+    /// Sets the starting era to the given value.
+    pub fn era(&mut self, era: u64) -> &mut Self {
+        self.era = era;
         self
     }
 
     /// Sets the maximum number of future epochs for which we handle messages simultaneously.
-    pub fn max_future_epochs(&mut self, max_future_epochs: usize) -> &mut Self {
+    pub fn max_future_epochs(&mut self, max_future_epochs: u64) -> &mut Self {
         self.max_future_epochs = max_future_epochs;
         self
     }
@@ -91,18 +94,18 @@ where
     /// Creates a new Dynamic Honey Badger instance with an empty buffer.
     pub fn build(&mut self, netinfo: NetworkInfo<N>) -> DynamicHoneyBadger<C, N> {
         let DynamicHoneyBadgerBuilder {
-            epoch,
+            era,
             max_future_epochs,
             rng,
             subset_handling_strategy,
             encryption_schedule,
             _phantom,
         } = self;
-        let epoch = *epoch;
+        let era = *era;
         let max_future_epochs = *max_future_epochs;
         let arc_netinfo = Arc::new(netinfo.clone());
         let honey_badger = HoneyBadger::builder(arc_netinfo.clone())
-            .session_id(epoch)
+            .session_id(era)
             .max_future_epochs(max_future_epochs)
             .rng(rng.sub_rng())
             .subset_handling_strategy(subset_handling_strategy.clone())
@@ -111,12 +114,11 @@ where
         DynamicHoneyBadger {
             netinfo,
             max_future_epochs,
-            start_epoch: epoch,
+            era,
             vote_counter: VoteCounter::new(arc_netinfo, 0),
             key_gen_msg_buffer: Vec::new(),
             honey_badger,
             key_gen_state: None,
-            incoming_queue: Vec::new(),
             rng: Box::new(rng.sub_rng()),
         }
     }
@@ -155,17 +157,16 @@ where
         let mut dhb = DynamicHoneyBadger {
             netinfo,
             max_future_epochs: self.max_future_epochs,
-            start_epoch: join_plan.epoch,
-            vote_counter: VoteCounter::new(arc_netinfo, join_plan.epoch),
+            era: join_plan.era,
+            vote_counter: VoteCounter::new(arc_netinfo, join_plan.era),
             key_gen_msg_buffer: Vec::new(),
             honey_badger,
             key_gen_state: None,
-            incoming_queue: Vec::new(),
             rng: Box::new(self.rng.sub_rng()),
         };
         let step = match join_plan.change {
             ChangeState::InProgress(ref change) => match change {
-                Change::NodeChange(change) => dhb.update_key_gen(join_plan.epoch, change)?,
+                Change::NodeChange(change) => dhb.update_key_gen(join_plan.era, change)?,
                 _ => Step::default(),
             },
             ChangeState::None | ChangeState::Complete(..) => Step::default(),