refactor import queue to use explicit block import references

substrate/core/network/src/import_queue.rs

@@ -34,14 +34,16 @@ use primitives::AuthorityId;
 use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, NumberFor, Zero};
 
 pub use blocks::BlockData;
-use chain::Client;
+use client::error::Error as ClientError;
 use error::{ErrorKind, Error};
 use protocol::Context;
 use service::ExecuteInContext;
 use sync::ChainSync;
 
-pub use consensus::{ImportBlock, ImportResult, BlockOrigin};
+pub use consensus::{ImportBlock, BlockImport, ImportResult, BlockOrigin};
 
+/// Shared block import struct used by the queue.
+pub type SharedBlockImport<B> = Arc<dyn BlockImport<B,Error=ClientError> + Send + Sync>;
 
 #[cfg(any(test, feature = "test-helpers"))]
 use std::cell::RefCell;
@@ -98,6 +100,7 @@ pub struct BasicQueue<B: BlockT, V: 'static + Verifier<B>> {
 	handle: Mutex<Option<::std::thread::JoinHandle<()>>>,
 	data: Arc<AsyncImportQueueData<B>>,
 	verifier: Arc<V>,
+	block_import: SharedBlockImport<B>,
 }
 
 /// Locks order: queue, queue_blocks, best_importing_number
@@ -111,11 +114,12 @@ struct AsyncImportQueueData<B: BlockT> {
 
 impl<B: BlockT, V: Verifier<B>> BasicQueue<B, V> {
 	/// Instantiate a new basic queue, with given verifier.
-	pub fn new(verifier: Arc<V>) -> Self {
+	pub fn new(verifier: Arc<V>, block_import: SharedBlockImport<B>) -> Self {
 		Self {
 			handle: Mutex::new(None),
 			data: Arc::new(AsyncImportQueueData::new()),
 			verifier,
+			block_import,
 		}
 	}
 }
@@ -141,8 +145,9 @@ impl<B: BlockT, V: 'static + Verifier<B>> ImportQueue<B> for BasicQueue<B, V> {
 
 		let qdata = self.data.clone();
 		let verifier = self.verifier.clone();
+		let block_import = self.block_import.clone();
 		*self.handle.lock() = Some(::std::thread::Builder::new().name("ImportQueue".into()).spawn(move || {
-			import_thread(link, qdata, verifier)
+			import_thread(block_import, link, qdata, verifier)
 		}).map_err(|err| Error::from(ErrorKind::Io(err)))?);
 		Ok(())
 	}
@@ -205,6 +210,7 @@ impl<B: BlockT, V: 'static + Verifier<B>> Drop for BasicQueue<B, V> {
 
 /// Blocks import thread.
 fn import_thread<B: BlockT, L: Link<B>, V: Verifier<B>>(
+	block_import: SharedBlockImport<B>,
 	link: L,
 	qdata: Arc<AsyncImportQueueData<B>>,
 	verifier: Arc<V>
@@ -229,6 +235,7 @@ fn import_thread<B: BlockT, L: Link<B>, V: Verifier<B>>(
 
 		let blocks_hashes: Vec<B::Hash> = new_blocks.1.iter().map(|b| b.block.hash.clone()).collect();
 		if !import_many_blocks(
+			&*block_import,
 			&link,
 			Some(&*qdata),
 			new_blocks,
@@ -249,8 +256,6 @@ fn import_thread<B: BlockT, L: Link<B>, V: Verifier<B>>(
 /// Hooks that the verification queue can use to influence the synchronization
 /// algorithm.
 pub trait Link<B: BlockT>: Send {
-	/// Get chain reference.
-	fn chain(&self) -> &Client<B>;
 	/// Block imported.
 	fn block_imported(&self, _hash: &B::Hash, _number: NumberFor<B>) { }
 	/// Maintain sync.
@@ -265,8 +270,6 @@ pub trait Link<B: BlockT>: Send {
 
 /// A link implementation that connects to the network.
 pub struct NetworkLink<B: BlockT, E: ExecuteInContext<B>> {
-	/// The client handle.
-	pub(crate) client: Arc<Client<B>>,
 	/// The chain-sync handle
 	pub(crate) sync: Weak<RwLock<ChainSync<B>>>,
 	/// Network context.
@@ -286,10 +289,6 @@ impl<B: BlockT, E: ExecuteInContext<B>> NetworkLink<B, E> {
 }
 
 impl<B: BlockT, E: ExecuteInContext<B>> Link<B> for NetworkLink<B, E> {
-	fn chain(&self) -> &Client<B> {
-		&*self.client
-	}
-
 	fn block_imported(&self, hash: &B::Hash, number: NumberFor<B>) {
 		self.with_sync(|sync, _| sync.block_imported(&hash, number))
 	}
@@ -342,6 +341,7 @@ enum BlockImportError {
 
 /// Import a bunch of blocks.
 fn import_many_blocks<'a, B: BlockT, L: Link<B>, V: Verifier<B>>(
+	import_handle: &BlockImport<B, Error=ClientError>,
 	link: &L,
 	qdata: Option<&AsyncImportQueueData<B>>,
 	blocks: (BlockOrigin, Vec<BlockData<B>>),
@@ -365,7 +365,7 @@ fn import_many_blocks<'a, B: BlockT, L: Link<B>, V: Verifier<B>>(
 	// Blocks in the response/drain should be in ascending order.
 	for block in blocks {
 		let import_result = import_single_block(
-			link.chain(),
+			import_handle,
 			blocks_origin.clone(),
 			block,
 			verifier.clone(),
@@ -389,7 +389,7 @@ fn import_many_blocks<'a, B: BlockT, L: Link<B>, V: Verifier<B>>(
 
 /// Single block import function.
 fn import_single_block<B: BlockT, V: Verifier<B>>(
-	chain: &Client<B>,
+	import_handle: &BlockImport<B,Error=ClientError>,
 	block_origin: BlockOrigin,
 	block: BlockData<B>,
 	verifier: Arc<V>
@@ -431,7 +431,7 @@ fn import_single_block<B: BlockT, V: Verifier<B>>(
 			BlockImportError::VerificationFailed(peer, msg)
 		})?;
 
-	match chain.import(import_block, new_authorities) {
+	match import_handle.import_block(import_block, new_authorities) {
 		Ok(ImportResult::AlreadyInChain) => {
 			trace!(target: "sync", "Block already in chain {}: {:?}", number, hash);
 			Ok(BlockImportResult::ImportedKnown(hash, number))
@@ -558,15 +558,25 @@ impl<B: BlockT> Verifier<B> for PassThroughVerifier {
 	}
 }
 
-#[cfg(any(test, feature = "test-helpers"))]
 /// Blocks import queue that is importing blocks in the same thread.
 /// The boolean value indicates whether blocks should be imported without instant finality.
-pub struct SyncImportQueue<B: BlockT, V: Verifier<B>>(Arc<V>, ImportCB<B>);
+#[cfg(any(test, feature = "test-helpers"))]
+pub struct SyncImportQueue<B: BlockT, V: Verifier<B>> {
+	verifier: Arc<V>,
+	link: ImportCB<B>,
+	block_import: SharedBlockImport<B>,
+}
+
 #[cfg(any(test, feature = "test-helpers"))]
 impl<B: BlockT, V: Verifier<B>> SyncImportQueue<B, V> {
-	/// Create a new SyncImportQueue wrapping the given Verifier
-	pub fn new(verifier: Arc<V>) -> Self {
-		SyncImportQueue(verifier, ImportCB::new())
+	/// Create a new SyncImportQueue wrapping the given Verifier and block import
+	/// handle.
+	pub fn new(verifier: Arc<V>, block_import: SharedBlockImport<B>) -> Self {
+		SyncImportQueue {
+			verifier,
+			link: ImportCB::new(),
+			block_import,
+		}
 	}
 }
 
@@ -577,10 +587,12 @@ impl<B: 'static + BlockT, V: 'static + Verifier<B>> ImportQueue<B> for SyncImpor
 		&self,
 		link: L,
 	) -> Result<(), Error> {
-		let v = self.0.clone();
-		self.1.set(Box::new(move |origin, new_blocks| {
+		let v = self.verifier.clone();
+		let import_handle = self.block_import.clone();
+		self.link.set(Box::new(move |origin, new_blocks| {
 			let verifier = v.clone();
 			import_many_blocks(
+				&*import_handle,
 				&link,
 				None,
 				(origin, new_blocks),
@@ -605,7 +617,7 @@ impl<B: 'static + BlockT, V: 'static + Verifier<B>> ImportQueue<B> for SyncImpor
 	}
 
 	fn import_blocks(&self, origin: BlockOrigin, blocks: Vec<BlockData<B>>) {
-		self.1.call(origin, blocks);
+		self.link.call(origin, blocks);
 	}
 }
 
@@ -644,7 +656,6 @@ pub mod tests {
 	}
 
 	impl Link<Block> for TestLink {
-		fn chain(&self) -> &Client<Block> { &*self.chain }
 		fn block_imported(&self, _hash: &Hash, _number: NumberFor<Block>) {
 			self.imported.set(self.imported.get() + 1);
 		}
@@ -762,7 +773,9 @@ pub mod tests {
 		let qdata = AsyncImportQueueData::new();
 		let verifier = Arc::new(PassThroughVerifier(true));
 		qdata.is_stopping.store(true, Ordering::SeqCst);
+		let client = test_client::new();
 		assert!(!import_many_blocks(
+			&client,
 			&mut TestLink::new(),
 			Some(&qdata),
 			(BlockOrigin::File, vec![block.clone(), block]),

substrate/core/network/src/service.rs

@@ -130,7 +130,6 @@ impl<B: BlockT + 'static, S: Specialization<B>, H: ExHashT> Service<B, S, H> {
 		protocol_id: ProtocolId,
 		import_queue: I,
 	) -> Result<Arc<Service<B, S, H>>, Error> {
-		let chain = params.chain.clone();
 		let import_queue = Arc::new(import_queue);
 		let handler = Arc::new(Protocol::new(
 			params.config,
@@ -153,7 +152,6 @@ impl<B: BlockT + 'static, S: Specialization<B>, H: ExHashT> Service<B, S, H> {
 
 		// connect the import-queue to the network service.
 		let link = ::import_queue::NetworkLink {
-			client: chain,
 			sync: Arc::downgrade(service.handler.sync()),
 			context: Arc::downgrade(&service),
 		};

substrate/core/network/src/test/mod.rs

@@ -24,6 +24,7 @@ use std::sync::Arc;
 
 use parking_lot::RwLock;
 use client;
+use client::error::Error as ClientError;
 use client::block_builder::BlockBuilder;
 use runtime_primitives::generic::BlockId;
 use io::SyncIo;
@@ -38,7 +39,7 @@ use import_queue::{SyncImportQueue, PassThroughVerifier, Verifier};
 use consensus::BlockOrigin;
 use specialization::Specialization;
 use consensus_gossip::ConsensusGossip;
-use import_queue::ImportQueue;
+use import_queue::{BlockImport, ImportQueue};
 use service::ExecuteInContext;
 use test_client;
 
@@ -163,7 +164,6 @@ impl<V: 'static + Verifier<Block>> Peer<V> {
 		let info = self.client.info().expect("In-mem client does not fail");
 		let header = self.client.header(&BlockId::Hash(info.chain.best_hash)).unwrap().unwrap();
 		let network_link = ::import_queue::NetworkLink {
-			client: self.sync.context_data().chain.clone(),
 			sync: Arc::downgrade(self.sync.sync()),
 			context: Arc::downgrade(&self.executor),
 		};
@@ -235,16 +235,31 @@ impl<V: 'static + Verifier<Block>> Peer<V> {
 	pub fn generate_blocks<F>(&self, count: usize, origin: BlockOrigin, mut edit_block: F)
 	where F: FnMut(&mut BlockBuilder<test_client::Backend, test_client::Executor, Block, Blake2Hasher>)
 	{
-		for _ in 0 .. count {
+		use blocks::BlockData;
+
+		for _  in 0..count {
 			let mut builder = self.client.new_block().unwrap();
 			edit_block(&mut builder);
 			let block = builder.bake().unwrap();
 			let hash = block.header.hash();
 			trace!("Generating {}, (#{}, parent={})", hash, block.header.number, block.header.parent_hash);
 			let header = block.header.clone();
-			self.client.justify_and_import(origin, block).unwrap();
-			self.sync.on_block_imported(&mut TestIo::new(&self.queue, None), hash, &header);
+
+			// NOTE: if we use a non-synchronous queue in the test-net in the future,
+			// this may not work.
+		 	self.import_queue.import_blocks(origin, vec![BlockData {
+				origin: None,
+				block: ::message::BlockData::<Block> {
+					hash,
+					header: Some(header),
+					body: Some(block.extrinsics),
+					receipt: None,
+					message_queue: None,
+					justification: None,
+				},
+			}]);
 		}
+
 	}
 
 	/// Push blocks to the peer (simplified: with or without a TX)
@@ -308,7 +323,12 @@ pub trait TestNetFactory: Sized {
 	fn mut_peers<F: Fn(&mut Vec<Arc<Peer<Self::Verifier>>>)>(&mut self, closure: F );
 
 	fn started(&self) -> bool;
-	fn set_started(&mut self, now: bool); 
+	fn set_started(&mut self, now: bool);
+
+	/// Get custom block import handle for fresh client.
+	fn make_block_import(&self, client: Arc<PeersClient>) -> Arc<BlockImport<Block,Error=ClientError> + Send + Sync> {
+		client
+	}
 
 	fn default_config() -> ProtocolConfig {
 		ProtocolConfig::default()
@@ -330,7 +350,9 @@ pub trait TestNetFactory: Sized {
 		let client = Arc::new(test_client::new());
 		let tx_pool = Arc::new(EmptyTransactionPool);
 		let verifier = self.make_verifier(client.clone(), config);
-		let import_queue = Arc::new(SyncImportQueue::new(verifier));
+		let block_import = self.make_block_import(client.clone());
+
+		let import_queue = Arc::new(SyncImportQueue::new(verifier, block_import));
 		let specialization = DummySpecialization {
 			gossip: ConsensusGossip::new(),
 		};
@@ -471,7 +493,7 @@ impl TestNetFactory for TestNet {
 			started: false
 		}
 	}
-	
+
 	fn make_verifier(&self, _client: Arc<PeersClient>, _config: &ProtocolConfig)
 		-> Arc<Self::Verifier>
 	{