Create creator outside of creation

Author: timorleph

consensus/src/consensus/mod.rs

@@ -6,7 +6,8 @@ use crate::{
         handler::Consensus,
         service::{Service, IO as ConsensusIO},
     },
-    creation, handle_task_termination,
+    creation::{run as run_creation, Creator, IO as CreationIO},
+    handle_task_termination,
     interface::LocalIO,
     network::{Hub as NetworkHub, NetworkData},
     units::Validator,
@@ -97,14 +98,16 @@ pub async fn run_session<
     let (starting_round_for_creator, starting_round_from_collection) = oneshot::channel();
 
     debug!(target: LOG_TARGET, "Spawning creator.");
+    let creator = Creator::new(index, config.n_members());
     let creator_terminator = terminator.add_offspring_connection("creator");
     let creator_config = config.clone();
     let creator_keychain = keychain.clone();
     let creator_handle = spawn_handle
         .spawn_essential("consensus/creator", async move {
-            creation::run(
+            run_creation(
+                creator,
                 creator_config,
-                creation::IO {
+                CreationIO {
                     outgoing_units: new_units_for_service,
                     incoming_parents: parents_from_service,
                     data_provider,

consensus/src/creation/mod.rs

@@ -109,14 +109,15 @@ async fn keep_processing_units_until<U: Unit>(
 /// We refer to the documentation https://cardinal-cryptography.github.io/AlephBFT/internals.html
 /// Section 5.1 for a discussion of this component.
 pub async fn run<U: Unit, MK: MultiKeychain, DP: DataProvider>(
+    creator: Creator<U::Hasher>,
     conf: Config,
     mut io: IO<U, MK, DP>,
     keychain: MK,
     mut starting_round: oneshot::Receiver<Option<Round>>,
     mut terminator: Terminator,
 ) {
     futures::select! {
-        _ = read_starting_round_and_run_creator(conf, &mut io, keychain, &mut starting_round).fuse() =>
+        _ = read_starting_round_and_run_creator(creator, conf, &mut io, keychain, &mut starting_round).fuse() =>
             debug!(target: LOG_TARGET, "Creator is about to finish."),
         _ = terminator.get_exit().fuse() =>
             debug!(target: LOG_TARGET, "Received an exit signal."),
@@ -126,6 +127,7 @@ pub async fn run<U: Unit, MK: MultiKeychain, DP: DataProvider>(
 }
 
 async fn read_starting_round_and_run_creator<U: Unit, MK: MultiKeychain, DP: DataProvider>(
+    creator: Creator<U::Hasher>,
     conf: Config,
     io: &mut IO<U, MK, DP>,
     keychain: MK,
@@ -144,7 +146,7 @@ async fn read_starting_round_and_run_creator<U: Unit, MK: MultiKeychain, DP: Dat
         }
     };
 
-    if let Err(err) = run_creator(conf, io, keychain, starting_round).await {
+    if let Err(err) = run_creator(creator, conf, io, keychain, starting_round).await {
         match err {
             CreatorError::OutChannelClosed(e) => {
                 warn!(target: LOG_TARGET, "Notification send error: {}. Exiting.", e)
@@ -157,17 +159,15 @@ async fn read_starting_round_and_run_creator<U: Unit, MK: MultiKeychain, DP: Dat
 }
 
 async fn run_creator<U: Unit, MK: MultiKeychain, DP: DataProvider>(
+    mut creator: Creator<U::Hasher>,
     conf: Config,
     io: &mut IO<U, MK, DP>,
     keychain: MK,
     starting_round: Round,
 ) -> anyhow::Result<(), CreatorError> {
-    let node_id = conf.node_ix();
-    let n_members = conf.n_members();
     let create_delay = conf.delay_config().unit_creation_delay.clone();
     let max_round = conf.max_round();
     let session_id = conf.session_id();
-    let mut creator = Creator::new(node_id, n_members);
     let packer = Packer::new(keychain, session_id);
     let incoming_parents = &mut io.incoming_parents;
     let outgoing_units = &io.outgoing_units;

consensus/src/testing/creation.rs

@@ -1,5 +1,5 @@
 use crate::{
-    creation::{run, IO},
+    creation::{run, Creator, IO},
     testing::{gen_config, gen_delay_config},
     units::{SignedUnit as GenericSignedUnit, Unit as GenericUnit},
     NodeCount, Receiver, Round, Sender, Terminator,
@@ -77,6 +77,7 @@ fn setup_test(n_members: NodeCount) -> TestSetup {
             data_provider: DataProvider::new(),
         };
         let config = gen_config(node_ix, n_members, gen_delay_config());
+        let creator = Creator::new(node_ix, n_members);
         let (starting_round_for_consensus, starting_round) = oneshot::channel();
 
         units_for_creators.push(parents_for_creator);
@@ -87,6 +88,7 @@ fn setup_test(n_members: NodeCount) -> TestSetup {
 
         let handle = tokio::spawn(async move {
             run(
+                creator,
                 config,
                 io,
                 keychain,