Add SnapshotNotifySender to Tendermint worker

Author: foriequal0

codechain/run_node.rs

@@ -366,10 +366,11 @@ pub fn run_node(matches: &ArgMatches) -> Result<(), String> {
     }
 
     let _snapshot_service = {
+        let client = client.client();
+        let (tx, rx) = snapshot_notify::create();
+        client.engine().register_snapshot_notify_sender(tx);
+
         if !config.snapshot.disable.unwrap() {
-            let client = client.client();
-            let (tx, rx) = snapshot_notify::create();
-            client.engine().register_snapshot_notify_sender(tx);
             let service = Arc::new(SnapshotService::new(client, rx, config.snapshot.path.unwrap()));
             Some(service)
         } else {

core/src/consensus/tendermint/engine.rs

@@ -37,6 +37,7 @@ use super::worker;
 use super::{ChainNotify, Tendermint, SEAL_FIELDS};
 use crate::account_provider::AccountProvider;
 use crate::block::*;
+use crate::client::snapshot_notify::NotifySender as SnapshotNotifySender;
 use crate::client::{Client, ConsensusClient};
 use crate::codechain_machine::CodeChainMachine;
 use crate::consensus::tendermint::params::TimeGapParams;
@@ -359,6 +360,10 @@ impl ConsensusEngine for Tendermint {
         client.add_notify(Arc::downgrade(&self.chain_notify) as Weak<dyn ChainNotify>);
     }
 
+    fn register_snapshot_notify_sender(&self, sender: SnapshotNotifySender) {
+        self.snapshot_notify_sender_initializer.send(sender).unwrap();
+    }
+
     fn complete_register(&self) {
         let (result, receiver) = crossbeam::bounded(1);
         self.inner.send(worker::Event::Restore(result)).unwrap();

core/src/consensus/tendermint/mod.rs

@@ -41,6 +41,7 @@ pub use self::types::{Height, Step, View};
 use super::{stake, ValidatorSet};
 use crate::client::ConsensusClient;
 use crate::codechain_machine::CodeChainMachine;
+use crate::snapshot_notify::NotifySender as SnapshotNotifySender;
 use crate::ChainNotify;
 
 /// Timer token representing the consensus step timeouts.
@@ -58,6 +59,7 @@ pub struct Tendermint {
     client: RwLock<Option<Weak<dyn ConsensusClient>>>,
     external_params_initializer: crossbeam::Sender<TimeGapParams>,
     extension_initializer: crossbeam::Sender<(crossbeam::Sender<network::Event>, Weak<dyn ConsensusClient>)>,
+    snapshot_notify_sender_initializer: crossbeam::Sender<SnapshotNotifySender>,
     timeouts: TimeoutParams,
     join: Option<JoinHandle<()>>,
     quit_tendermint: crossbeam::Sender<()>,
@@ -93,15 +95,22 @@ impl Tendermint {
         let timeouts = our_params.timeouts;
         let machine = Arc::new(machine);
 
-        let (join, external_params_initializer, extension_initializer, inner, quit_tendermint) =
-            worker::spawn(our_params.validators);
+        let (
+            join,
+            external_params_initializer,
+            extension_initializer,
+            snapshot_notify_sender_initializer,
+            inner,
+            quit_tendermint,
+        ) = worker::spawn(our_params.validators);
         let action_handlers: Vec<Arc<dyn ActionHandler>> = vec![stake.clone()];
         let chain_notify = Arc::new(TendermintChainNotify::new(inner.clone()));
 
         Arc::new(Tendermint {
             client: Default::default(),
             external_params_initializer,
             extension_initializer,
+            snapshot_notify_sender_initializer,
             timeouts,
             join: Some(join),
             quit_tendermint,

core/src/consensus/tendermint/worker.rs

@@ -50,6 +50,7 @@ use crate::consensus::validator_set::{DynamicValidator, ValidatorSet};
 use crate::consensus::{EngineError, Seal};
 use crate::encoded;
 use crate::error::{BlockError, Error};
+use crate::snapshot_notify::NotifySender as SnapshotNotifySender;
 use crate::transaction::{SignedTransaction, UnverifiedTransaction};
 use crate::views::BlockView;
 use crate::BlockId;
@@ -59,6 +60,7 @@ type SpawnResult = (
     JoinHandle<()>,
     crossbeam::Sender<TimeGapParams>,
     crossbeam::Sender<(crossbeam::Sender<network::Event>, Weak<dyn ConsensusClient>)>,
+    crossbeam::Sender<SnapshotNotifySender>,
     crossbeam::Sender<Event>,
     crossbeam::Sender<()>,
 );
@@ -97,6 +99,7 @@ struct Worker {
     time_gap_params: TimeGapParams,
     timeout_token_nonce: usize,
     vote_regression_checker: VoteRegressionChecker,
+    snapshot_notify_sender: SnapshotNotifySender,
 }
 
 pub enum Event {
@@ -180,6 +183,7 @@ impl Worker {
         extension: EventSender<network::Event>,
         client: Weak<dyn ConsensusClient>,
         time_gap_params: TimeGapParams,
+        snapshot_notify_sender: SnapshotNotifySender,
     ) -> Self {
         Worker {
             client,
@@ -198,6 +202,7 @@ impl Worker {
             time_gap_params,
             timeout_token_nonce: ENGINE_TIMEOUT_TOKEN_NONCE_BASE,
             vote_regression_checker: VoteRegressionChecker::new(),
+            snapshot_notify_sender,
         }
     }
 
@@ -206,6 +211,7 @@ impl Worker {
         let (quit, quit_receiver) = crossbeam::bounded(1);
         let (external_params_initializer, external_params_receiver) = crossbeam::bounded(1);
         let (extension_initializer, extension_receiver) = crossbeam::bounded(1);
+        let (snapshot_notify_sender_initializer, snapshot_notify_sender_receiver) = crossbeam::bounded(1);
         let join = Builder::new()
             .name("tendermint".to_string())
             .spawn(move || {
@@ -249,8 +255,29 @@ impl Worker {
                     return
                 }
                 };
+                // TODO: Make initialization steps to order insensitive.
+                let snapshot_notify_sender = crossbeam::select! {
+                recv(snapshot_notify_sender_receiver) -> msg => {
+                    match msg {
+                        Ok(sender) => sender,
+                        Err(crossbeam::RecvError) => {
+                            cerror!(ENGINE, "The tendermint extension is not initalized.");
+                            return
+                        }
+                    }
+                }
+                recv(quit_receiver) -> msg => {
+                    match msg {
+                        Ok(()) => {},
+                        Err(crossbeam::RecvError) => {
+                            cerror!(ENGINE, "The quit channel for tendermint thread had been closed.");
+                        }
+                    }
+                    return
+                }
+                };
                 validators.register_client(Weak::clone(&client));
-                let mut inner = Self::new(validators, extension, client, time_gap_params);
+                let mut inner = Self::new(validators, extension, client, time_gap_params, snapshot_notify_sender);
                 loop {
                     crossbeam::select! {
                     recv(receiver) -> msg => {
@@ -374,7 +401,7 @@ impl Worker {
                 }
             })
             .unwrap();
-        (join, external_params_initializer, extension_initializer, sender, quit)
+        (join, external_params_initializer, extension_initializer, snapshot_notify_sender_initializer, sender, quit)
     }
 
     /// The client is a thread-safe struct. Using it in multi-threads is safe.