Implement graceful shutdown (#2812)

* Remove stop status mutex
* remove some deadlocks
* Rewrite stop channel handling
* fix deadlock in peers object
* add missing test fixes

api/src/types.rs

@@ -434,8 +434,7 @@ impl<'de> serde::de::Deserialize<'de> for OutputPrintable {
 				}
 
 				if output_type.is_none()
-					|| commit.is_none()
-					|| spent.is_none()
+					|| commit.is_none() || spent.is_none()
 					|| proof_hash.is_none()
 					|| mmr_index.is_none()
 				{

chain/src/chain.rs

@@ -33,7 +33,7 @@ use crate::types::{
 	BlockStatus, ChainAdapter, NoStatus, Options, Tip, TxHashSetRoots, TxHashsetWriteStatus,
 };
 use crate::util::secp::pedersen::{Commitment, RangeProof};
-use crate::util::{Mutex, RwLock, StopState};
+use crate::util::RwLock;
 use grin_store::Error::NotFoundErr;
 use std::collections::HashMap;
 use std::fs::{self, File};
@@ -152,7 +152,6 @@ pub struct Chain {
 	// POW verification function
 	pow_verifier: fn(&BlockHeader) -> Result<(), pow::Error>,
 	archive_mode: bool,
-	stop_state: Arc<Mutex<StopState>>,
 	genesis: BlockHeader,
 }
 
@@ -167,16 +166,7 @@ impl Chain {
 		pow_verifier: fn(&BlockHeader) -> Result<(), pow::Error>,
 		verifier_cache: Arc<RwLock<dyn VerifierCache>>,
 		archive_mode: bool,
-		stop_state: Arc<Mutex<StopState>>,
 	) -> Result<Chain, Error> {
-		// Note: We take a lock on the stop_state here and do not release it until
-		// we have finished chain initialization.
-		let stop_state_local = stop_state.clone();
-		let stop_lock = stop_state_local.lock();
-		if stop_lock.is_stopped() {
-			return Err(ErrorKind::Stopped.into());
-		}
-
 		let store = Arc::new(store::ChainStore::new(&db_root)?);
 
 		// open the txhashset, creating a new one if necessary
@@ -194,7 +184,6 @@ impl Chain {
 			pow_verifier,
 			verifier_cache,
 			archive_mode,
-			stop_state,
 			genesis: genesis.header.clone(),
 		})
 	}
@@ -283,15 +272,6 @@ impl Chain {
 	/// or false if it has added to a fork (or orphan?).
 	fn process_block_single(&self, b: Block, opts: Options) -> Result<Option<Tip>, Error> {
 		let (maybe_new_head, prev_head) = {
-			// Note: We take a lock on the stop_state here and do not release it until
-			// we have finished processing this single block.
-			// We take care to write both the txhashset *and* the batch while we
-			// have the stop_state lock.
-			let stop_lock = self.stop_state.lock();
-			if stop_lock.is_stopped() {
-				return Err(ErrorKind::Stopped.into());
-			}
-
 			let mut txhashset = self.txhashset.write();
 			let batch = self.store.batch()?;
 			let mut ctx = self.new_ctx(opts, batch, &mut txhashset)?;
@@ -381,15 +361,6 @@ impl Chain {
 	/// This is only ever used during sync and is based on sync_head.
 	/// We update header_head here if our total work increases.
 	pub fn sync_block_headers(&self, headers: &[BlockHeader], opts: Options) -> Result<(), Error> {
-		// Note: We take a lock on the stop_state here and do not release it until
-		// we have finished processing this single block.
-		// We take care to write both the txhashset *and* the batch while we
-		// have the stop_state lock.
-		let stop_lock = self.stop_state.lock();
-		if stop_lock.is_stopped() {
-			return Err(ErrorKind::Stopped.into());
-		}
-
 		let mut txhashset = self.txhashset.write();
 		let batch = self.store.batch()?;
 		let mut ctx = self.new_ctx(opts, batch, &mut txhashset)?;
@@ -1098,14 +1069,6 @@ impl Chain {
 			}
 		}
 
-		// Note: We take a lock on the stop_state here and do not release it until
-		// we have finished processing this chain compaction operation.
-		// We want to avoid shutting the node down in the middle of compacting the data.
-		let stop_lock = self.stop_state.lock();
-		if stop_lock.is_stopped() {
-			return Err(ErrorKind::Stopped.into());
-		}
-
 		// Take a write lock on the txhashet and start a new writeable db batch.
 		let mut txhashset = self.txhashset.write();
 		let mut batch = self.store.batch()?;

chain/src/store.rs

@@ -384,7 +384,6 @@ impl<'a> Batch<'a> {
 		let key = to_key(BLOCK_PREFIX, &mut "".to_string().into_bytes());
 		self.db.iter(&key)
 	}
-
 }
 
 /// An iterator on blocks, from latest to earliest, specialized to return

chain/tests/data_file_integrity.rs

@@ -21,7 +21,7 @@ use self::core::libtx;
 use self::core::pow::{self, Difficulty};
 use self::core::{consensus, genesis};
 use self::keychain::{ExtKeychain, ExtKeychainPath, Keychain};
-use self::util::{Mutex, RwLock, StopState};
+use self::util::RwLock;
 use chrono::Duration;
 use grin_chain as chain;
 use grin_core as core;
@@ -47,7 +47,6 @@ fn setup(dir_name: &str) -> Chain {
 		pow::verify_size,
 		verifier_cache,
 		false,
-		Arc::new(Mutex::new(StopState::new())),
 	)
 	.unwrap()
 }
@@ -61,7 +60,6 @@ fn reload_chain(dir_name: &str) -> Chain {
 		pow::verify_size,
 		verifier_cache,
 		false,
-		Arc::new(Mutex::new(StopState::new())),
 	)
 	.unwrap()
 }

chain/tests/mine_simple_chain.rs

@@ -23,7 +23,7 @@ use self::core::libtx::{self, build, reward};
 use self::core::pow::Difficulty;
 use self::core::{consensus, global, pow};
 use self::keychain::{ExtKeychain, ExtKeychainPath, Keychain};
-use self::util::{Mutex, RwLock, StopState};
+use self::util::{RwLock, StopState};
 use chrono::Duration;
 use grin_chain as chain;
 use grin_core as core;
@@ -47,7 +47,6 @@ fn setup(dir_name: &str, genesis: Block) -> Chain {
 		pow::verify_size,
 		verifier_cache,
 		false,
-		Arc::new(Mutex::new(StopState::new())),
 	)
 	.unwrap()
 }
@@ -565,7 +564,6 @@ fn actual_diff_iter_output() {
 		pow::verify_size,
 		verifier_cache,
 		false,
-		Arc::new(Mutex::new(StopState::new())),
 	)
 	.unwrap();
 	let iter = chain.difficulty_iter().unwrap();

chain/tests/test_coinbase_maturity.rs

@@ -20,7 +20,7 @@ use self::core::libtx::{self, build};
 use self::core::pow::Difficulty;
 use self::core::{consensus, pow};
 use self::keychain::{ExtKeychain, ExtKeychainPath, Keychain};
-use self::util::{Mutex, RwLock, StopState};
+use self::util::{RwLock, StopState};
 use chrono::Duration;
 use env_logger;
 use grin_chain as chain;
@@ -53,7 +53,6 @@ fn test_coinbase_maturity() {
 			pow::verify_size,
 			verifier_cache,
 			false,
-			Arc::new(Mutex::new(StopState::new())),
 		)
 		.unwrap();
 

etc/gen_gen/src/bin/gen_gen.rs

@@ -281,7 +281,7 @@ fn setup_chain(dir_name: &str, genesis: core::core::Block) -> chain::Chain {
 		core::pow::verify_size,
 		verifier_cache,
 		false,
-		Arc::new(util::Mutex::new(util::StopState::new())),
+		Arc::new(util::StopState::new()),
 	)
 	.unwrap()
 }