Fix import queue thread pool shutdown (#4929)

* Fix import queue thread pool shutdown

* Make sure runtime is disposed before telemetry

* Close channel istead of sending a message

* Fixed test

substrate/client/cli/src/runtime.rs

@@ -128,12 +128,14 @@ where
 
 	// we eagerly drop the service so that the internal exit future is fired,
 	// but we need to keep holding a reference to the global telemetry guard
+	// and drop the runtime first.
 	let _telemetry = service.telemetry();
 
 	let f = service.fuse();
 	pin_mut!(f);
 
 	runtime.block_on(main(f)).map_err(|e| e.to_string())?;
+	drop(runtime);
 
 	Ok(())
 }

substrate/client/consensus/babe/src/tests.rs

@@ -229,7 +229,7 @@ impl Verifier<TestBlock> for TestVerifier {
 	) -> Result<(BlockImportParams<TestBlock, ()>, Option<Vec<(CacheKeyId, Vec<u8>)>>), String> {
 		// apply post-sealing mutations (i.e. stripping seal, if desired).
 		(self.mutator)(&mut header, Stage::PostSeal);
-		Ok(self.inner.verify(origin, header, justification, body).expect("verification failed!"))
+		self.inner.verify(origin, header, justification, body)
 	}
 }
 
@@ -423,7 +423,14 @@ fn run_one_test(
 	}
 
 	runtime.spawn(futures01::future::poll_fn(move || {
-		net.lock().poll();
+		let mut net = net.lock();
+		net.poll();
+		for p in net.peers() {
+			for (h, e) in p.failed_verifications() {
+				panic!("Verification failed for {:?}: {}", h, e);
+			}
+		}
+
 		Ok::<_, ()>(futures01::Async::NotReady::<()>)
 	}));
 

substrate/client/network/test/src/lib.rs

@@ -183,7 +183,7 @@ pub struct Peer<D, S: NetworkSpecialization<Block>> {
 	client: PeersClient,
 	/// We keep a copy of the verifier so that we can invoke it for locally-generated blocks,
 	/// instead of going through the import queue.
-	verifier: VerifierAdapter<dyn Verifier<Block>>,
+	verifier: VerifierAdapter<Block>,
 	/// We keep a copy of the block_import so that we can invoke it for locally-generated blocks,
 	/// instead of going through the import queue.
 	block_import: BlockImportAdapter<()>,
@@ -360,6 +360,11 @@ impl<D, S: NetworkSpecialization<Block>> Peer<D, S> {
 			|backend| backend.blocks_count()
 		).unwrap_or(0)
 	}
+
+	/// Return a collection of block hashes that failed verification
+	pub fn failed_verifications(&self) -> HashMap<<Block as BlockT>::Hash, String> {
+		self.verifier.failed_verifications.lock().clone()
+	}
 }
 
 pub struct EmptyTransactionPool;
@@ -485,15 +490,13 @@ impl<Transaction> BlockImport<Block> for BlockImportAdapter<Transaction> {
 }
 
 /// Implements `Verifier` on an `Arc<Mutex<impl Verifier>>`. Used internally.
-struct VerifierAdapter<T: ?Sized>(Arc<Mutex<Box<T>>>);
-
-impl<T: ?Sized> Clone for VerifierAdapter<T> {
-	fn clone(&self) -> Self {
-		VerifierAdapter(self.0.clone())
-	}
+#[derive(Clone)]
+struct VerifierAdapter<B: BlockT> {
+	verifier: Arc<Mutex<Box<dyn Verifier<B>>>>,
+	failed_verifications: Arc<Mutex<HashMap<B::Hash, String>>>,
 }
 
-impl<B: BlockT, T: ?Sized + Verifier<B>> Verifier<B> for VerifierAdapter<T> {
+impl<B: BlockT> Verifier<B> for VerifierAdapter<B> {
 	fn verify(
 		&mut self,
 		origin: BlockOrigin,
@@ -501,7 +504,20 @@ impl<B: BlockT, T: ?Sized + Verifier<B>> Verifier<B> for VerifierAdapter<T> {
 		justification: Option<Justification>,
 		body: Option<Vec<B::Extrinsic>>
 	) -> Result<(BlockImportParams<B, ()>, Option<Vec<(CacheKeyId, Vec<u8>)>>), String> {
-		self.0.lock().verify(origin, header, justification, body)
+		let hash = header.hash();
+		self.verifier.lock().verify(origin, header, justification, body).map_err(|e| {
+			self.failed_verifications.lock().insert(hash, e.clone());
+			e
+		})
+	}
+}
+
+impl<B: BlockT> VerifierAdapter<B> {
+	fn new(verifier: Arc<Mutex<Box<dyn Verifier<B>>>>) -> VerifierAdapter<B> {
+		VerifierAdapter {
+			verifier,
+			failed_verifications: Default::default(),
+		}
 	}
 }
 
@@ -592,7 +608,7 @@ pub trait TestNetFactory: Sized {
 			config,
 			&data,
 		);
-		let verifier = VerifierAdapter(Arc::new(Mutex::new(Box::new(verifier) as Box<_>)));
+		let verifier = VerifierAdapter::new(Arc::new(Mutex::new(Box::new(verifier) as Box<_>)));
 
 		let import_queue = Box::new(BasicQueue::new(
 			verifier.clone(),
@@ -668,7 +684,7 @@ pub trait TestNetFactory: Sized {
 			&config,
 			&data,
 		);
-		let verifier = VerifierAdapter(Arc::new(Mutex::new(Box::new(verifier) as Box<_>)));
+		let verifier = VerifierAdapter::new(Arc::new(Mutex::new(Box::new(verifier) as Box<_>)));
 
 		let import_queue = Box::new(BasicQueue::new(
 			verifier.clone(),

substrate/primitives/consensus/common/src/import_queue/basic_queue.rs

@@ -14,9 +14,10 @@
 // You should have received a copy of the GNU General Public License
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
-use std::{mem, pin::Pin, time::Duration, marker::PhantomData};
+use std::{mem, pin::Pin, time::Duration, marker::PhantomData, sync::Arc};
 use futures::{prelude::*, channel::mpsc, task::Context, task::Poll};
 use futures_timer::Delay;
+use parking_lot::{Mutex, Condvar};
 use sp_runtime::{Justification, traits::{Block as BlockT, Header as HeaderT, NumberFor}};
 
 use crate::block_import::BlockOrigin;
@@ -40,9 +41,28 @@ pub struct BasicQueue<B: BlockT, Transaction> {
 	manual_poll: Option<Pin<Box<dyn Future<Output = ()> + Send>>>,
 	/// A thread pool where the background worker is being run.
 	pool: Option<futures::executor::ThreadPool>,
+	pool_guard: Arc<(Mutex<usize>, Condvar)>,
 	_phantom: PhantomData<Transaction>,
 }
 
+impl<B: BlockT, Transaction> Drop for BasicQueue<B, Transaction> {
+	fn drop(&mut self) {
+		self.pool = None;
+		// Flush the queue and close the receiver to terminate the future.
+		self.sender.close_channel();
+		self.result_port.close();
+
+		// Make sure all pool threads terminate.
+		// https://github.com/rust-lang/futures-rs/issues/1470
+		// https://github.com/rust-lang/futures-rs/issues/1349
+		let (ref mutex, ref condvar) = *self.pool_guard;
+		let mut lock = mutex.lock();
+		while *lock != 0 {
+			condvar.wait(&mut lock);
+		}
+	}
+}
+
 impl<B: BlockT, Transaction: Send + 'static> BasicQueue<B, Transaction> {
 	/// Instantiate a new basic queue, with given verifier.
 	///
@@ -63,9 +83,22 @@ impl<B: BlockT, Transaction: Send + 'static> BasicQueue<B, Transaction> {
 			finality_proof_import,
 		);
 
+		let guard = Arc::new((Mutex::new(0usize), Condvar::new()));
+		let guard_start = guard.clone();
+		let guard_end = guard.clone();
+
 		let mut pool = futures::executor::ThreadPool::builder()
 			.name_prefix("import-queue-worker-")
 			.pool_size(1)
+			.after_start(move |_| *guard_start.0.lock() += 1)
+			.before_stop(move |_| {
+				let (ref mutex, ref condvar) = *guard_end;
+				let mut lock = mutex.lock();
+				*lock -= 1;
+				if *lock == 0 {
+					condvar.notify_one();
+				}
+			})
 			.create()
 			.ok();
 
@@ -82,6 +115,7 @@ impl<B: BlockT, Transaction: Send + 'static> BasicQueue<B, Transaction> {
 			result_port,
 			manual_poll,
 			pool,
+			pool_guard: guard,
 			_phantom: PhantomData,
 		}
 	}

substrate/primitives/consensus/common/src/import_queue/buffered_link.rs

@@ -157,6 +157,11 @@ impl<B: BlockT> BufferedLinkReceiver<B> {
 			}
 		}
 	}
+
+	/// Close the channel.
+	pub fn close(&mut self) {
+		self.rx.close()
+	}
 }
 
 #[cfg(test)]