Initial: Offchain Workers

core/basic-authorship/src/basic_authorship.rs

@@ -20,23 +20,22 @@
 //
 use std::{self, time, sync::Arc};
 
-use log::{info, debug};
+use log::{info, debug, warn};
 
 use client::{
 	self, error, Client as SubstrateClient, CallExecutor,
 	block_builder::api::BlockBuilder as BlockBuilderApi, runtime_api::Core,
 };
 use codec::Decode;
 use consensus_common::{self, evaluation};
-use primitives::{H256, Blake2Hasher};
+use primitives::{H256, Blake2Hasher, ExecutionContext};
 use runtime_primitives::traits::{
 	Block as BlockT, Hash as HashT, Header as HeaderT, ProvideRuntimeApi, AuthorityIdFor
 };
-use runtime_primitives::ExecutionContext;
 use runtime_primitives::generic::BlockId;
 use runtime_primitives::ApplyError;
 use transaction_pool::txpool::{self, Pool as TransactionPool};
-use inherents::InherentData;
+use inherents::{InherentData, pool::InherentsPool};
 
 /// Build new blocks.
 pub trait BlockBuilder<Block: BlockT> {
@@ -114,6 +113,8 @@ pub struct ProposerFactory<C, A> where A: txpool::ChainApi {
 	pub client: Arc<C>,
 	/// The transaction pool.
 	pub transaction_pool: Arc<TransactionPool<A>>,
+	/// The inherents pool
+	pub inherents_pool: Arc<InherentsPool<<A::Block as BlockT>::Extrinsic>>,
 }
 
 impl<C, A> consensus_common::Environment<<C as AuthoringApi>::Block> for ProposerFactory<C, A> where
@@ -143,6 +144,7 @@ impl<C, A> consensus_common::Environment<<C as AuthoringApi>::Block> for Propose
 			parent_id: id,
 			parent_number: *parent_header.number(),
 			transaction_pool: self.transaction_pool.clone(),
+			inherents_pool: self.inherents_pool.clone(),
 			now: Box::new(time::Instant::now),
 		};
 
@@ -157,6 +159,7 @@ pub struct Proposer<Block: BlockT, C, A: txpool::ChainApi> {
 	parent_id: BlockId<Block>,
 	parent_number: <<Block as BlockT>::Header as HeaderT>::Number,
 	transaction_pool: Arc<TransactionPool<A>>,
+	inherents_pool: Arc<InherentsPool<<Block as BlockT>::Extrinsic>>,
 	now: Box<Fn() -> time::Instant>,
 }
 
@@ -200,11 +203,23 @@ impl<Block, C, A> Proposer<Block, C, A>	where
 			&self.parent_id,
 			inherent_data,
 			|block_builder| {
+				// Add inherents from the internal pool
+
+				let inherents = self.inherents_pool.drain();
+				debug!("Pushing {} queued inherents.", inherents.len());
+				for i in inherents {
+					if let Err(e) = block_builder.push_extrinsic(i) {
+						warn!("Error while pushing inherent extrinsic from the pool: {:?}", e);
+					}
+				}
+
+				// proceed with transactions
 				let mut is_first = true;
 				let mut skipped = 0;
 				let mut unqueue_invalid = Vec::new();
 				let pending_iterator = self.transaction_pool.ready();
 
+				debug!("Attempting to push transactions from the pool.");
 				for pending in pending_iterator {
 					if (self.now)() > deadline {
 						debug!("Consensus deadline reached when pushing block transactions, proceeding with proposing.");
@@ -299,6 +314,7 @@ mod tests {
 		let proposer_factory = ProposerFactory {
 			client: client.clone(),
 			transaction_pool: txpool.clone(),
+			inherents_pool: Default::default(),
 		};
 
 		let mut proposer = proposer_factory.init(
@@ -321,4 +337,32 @@ mod tests {
 		assert_eq!(txpool.ready().count(), 2);
 	}
 
+	#[test]
+	fn should_include_inherents_from_the_pool() {
+		// given
+		let client = Arc::new(test_client::new());
+		let chain_api = transaction_pool::ChainApi::new(client.clone());
+		let txpool = Arc::new(TransactionPool::new(Default::default(), chain_api));
+		let inpool = Arc::new(InherentsPool::default());
+
+		let proposer_factory = ProposerFactory {
+			client: client.clone(),
+			transaction_pool: txpool.clone(),
+			inherents_pool: inpool.clone(),
+		};
+
+		inpool.add(extrinsic(0));
+
+		let proposer = proposer_factory.init(
+			&client.header(&BlockId::number(0)).unwrap().unwrap(),
+			&[]
+		).unwrap();
+
+		// when
+		let deadline = time::Duration::from_secs(3);
+		let block = proposer.propose(Default::default(), deadline).unwrap();
+
+		// then
+		assert_eq!(block.extrinsics().len(), 1);
+	}
 }

core/cli/src/lib.rs

@@ -25,7 +25,7 @@ mod params;
 pub mod error;
 pub mod informant;
 
-use client::ExecutionStrategies;
+use client::{ExecutionStrategies, ExecutionStrategy};
 use runtime_primitives::traits::As;
 use service::{
 	ServiceFactory, FactoryFullConfiguration, RuntimeGenesis,
@@ -364,6 +364,7 @@ where
 		syncing: cli.syncing_execution.into(),
 		importing: cli.importing_execution.into(),
 		block_construction: cli.block_construction_execution.into(),
+		offchain_worker: ExecutionStrategy::NativeWhenPossible,
 		other: cli.other_execution.into(),
 	};
 

core/client/src/block_builder/block_builder.rs

@@ -17,15 +17,15 @@
 use super::api::BlockBuilder as BlockBuilderApi;
 use std::vec::Vec;
 use parity_codec::Encode;
-use crate::blockchain::HeaderBackend;
+use runtime_primitives::ApplyOutcome;
+use runtime_primitives::generic::BlockId;
 use runtime_primitives::traits::{
 	Header as HeaderT, Hash, Block as BlockT, One, HashFor, ProvideRuntimeApi, ApiRef
 };
-use primitives::H256;
-use runtime_primitives::generic::BlockId;
+use primitives::{H256, ExecutionContext};
+use crate::blockchain::HeaderBackend;
 use crate::runtime_api::Core;
 use crate::error;
-use runtime_primitives::{ApplyOutcome, ExecutionContext};
 
 
 /// Utility for building new (valid) blocks from a stream of extrinsics.

core/client/src/call_executor.rs

@@ -19,12 +19,12 @@ use parity_codec::{Encode, Decode};
 use runtime_primitives::generic::BlockId;
 use runtime_primitives::traits::Block as BlockT;
 use state_machine::{
-	self, OverlayedChanges, Ext, CodeExecutor, ExecutionManager, ExecutionStrategy
+	self, OverlayedChanges, Ext, CodeExecutor, ExecutionManager, ExecutionStrategy, NeverOffchainExt,
 };
 use executor::{RuntimeVersion, RuntimeInfo, NativeVersion};
 use hash_db::Hasher;
 use trie::MemoryDB;
-use primitives::{H256, Blake2Hasher, NativeOrEncoded, NeverNativeValue};
+use primitives::{H256, Blake2Hasher, NativeOrEncoded, NeverNativeValue, OffchainExt};
 
 use crate::backend;
 use crate::error;
@@ -42,12 +42,15 @@ where
 	/// Execute a call to a contract on top of state in a block of given hash.
 	///
 	/// No changes are made.
-	fn call(
+	fn call<
+		O: OffchainExt,
+	>(
 		&self,
 		id: &BlockId<B>,
 		method: &str,
 		call_data: &[u8],
 		strategy: ExecutionStrategy,
+		side_effects_handler: Option<&mut O>,
 	) -> Result<Vec<u8>, error::Error>;
 
 	/// Execute a contextual call on top of state in a block of a given hash.
@@ -56,6 +59,7 @@ where
 	/// Before executing the method, passed header is installed as the current header
 	/// of the execution context.
 	fn contextual_call<
+		O: OffchainExt,
 		PB: Fn() -> error::Result<B::Header>,
 		EM: Fn(
 			Result<NativeOrEncoded<R>, Self::Error>,
@@ -73,6 +77,7 @@ where
 		prepare_environment_block: PB,
 		execution_manager: ExecutionManager<EM>,
 		native_call: Option<NC>,
+		side_effects_handler: Option<&mut O>,
 	) -> error::Result<NativeOrEncoded<R>> where ExecutionManager<EM>: Clone;
 
 	/// Extract RuntimeVersion of given block
@@ -84,6 +89,7 @@ where
 	///
 	/// No changes are made.
 	fn call_at_state<
+		O: OffchainExt,
 		S: state_machine::Backend<H>,
 		F: FnOnce(
 			Result<NativeOrEncoded<R>, Self::Error>,
@@ -98,6 +104,7 @@ where
 		call_data: &[u8],
 		manager: ExecutionManager<F>,
 		native_call: Option<NC>,
+		side_effects_handler: Option<&mut O>,
 	) -> Result<(NativeOrEncoded<R>, S::Transaction, Option<MemoryDB<H>>), error::Error>;
 
 	/// Execute a call to a contract on top of given state, gathering execution proof.
@@ -140,7 +147,10 @@ pub struct LocalCallExecutor<B, E> {
 impl<B, E> LocalCallExecutor<B, E> {
 	/// Creates new instance of local call executor.
 	pub fn new(backend: Arc<B>, executor: E) -> Self {
-		LocalCallExecutor { backend, executor }
+		LocalCallExecutor {
+			backend,
+			executor,
+		}
 	}
 }
 
@@ -161,17 +171,19 @@ where
 {
 	type Error = E::Error;
 
-	fn call(&self,
+	fn call<O: OffchainExt>(&self,
 		id: &BlockId<Block>,
 		method: &str,
 		call_data: &[u8],
-		strategy: ExecutionStrategy
+		strategy: ExecutionStrategy,
+		side_effects_handler: Option<&mut O>,
 	) -> error::Result<Vec<u8>> {
 		let mut changes = OverlayedChanges::default();
 		let state = self.backend.state_at(*id)?;
 		let return_data = state_machine::new(
 			&state,
 			self.backend.changes_trie_storage(),
+			side_effects_handler,
 			&mut changes,
 			&self.executor,
 			method,
@@ -187,6 +199,7 @@ where
 	}
 
 	fn contextual_call<
+		O: OffchainExt,
 		PB: Fn() -> error::Result<Block::Header>,
 		EM: Fn(
 			Result<NativeOrEncoded<R>, Self::Error>,
@@ -204,13 +217,15 @@ where
 		prepare_environment_block: PB,
 		execution_manager: ExecutionManager<EM>,
 		native_call: Option<NC>,
+		mut side_effects_handler: Option<&mut O>,
 	) -> Result<NativeOrEncoded<R>, error::Error> where ExecutionManager<EM>: Clone {
 		let state = self.backend.state_at(*at)?;
 		if method != "Core_initialise_block" && initialised_block.map(|id| id != *at).unwrap_or(true) {
 			let header = prepare_environment_block()?;
 			state_machine::new(
 				&state,
 				self.backend.changes_trie_storage(),
+				side_effects_handler.as_mut().map(|x| &mut **x),
 				changes,
 				&self.executor,
 				"Core_initialise_block",
@@ -226,6 +241,7 @@ where
 		let result = state_machine::new(
 			&state,
 			self.backend.changes_trie_storage(),
+			side_effects_handler,
 			changes,
 			&self.executor,
 			method,
@@ -248,12 +264,13 @@ where
 	fn runtime_version(&self, id: &BlockId<Block>) -> error::Result<RuntimeVersion> {
 		let mut overlay = OverlayedChanges::default();
 		let state = self.backend.state_at(*id)?;
-		let mut ext = Ext::new(&mut overlay, &state, self.backend.changes_trie_storage());
+		let mut ext = Ext::new(&mut overlay, &state, self.backend.changes_trie_storage(), NeverOffchainExt::new());
 		self.executor.runtime_version(&mut ext)
 			.ok_or(error::ErrorKind::VersionInvalid.into())
 	}
 
 	fn call_at_state<
+		O: OffchainExt,
 		S: state_machine::Backend<Blake2Hasher>,
 		F: FnOnce(
 			Result<NativeOrEncoded<R>, Self::Error>,
@@ -268,10 +285,12 @@ where
 		call_data: &[u8],
 		manager: ExecutionManager<F>,
 		native_call: Option<NC>,
+		side_effects_handler: Option<&mut O>,
 	) -> error::Result<(NativeOrEncoded<R>, S::Transaction, Option<MemoryDB<Blake2Hasher>>)> {
 		state_machine::new(
 			state,
 			self.backend.changes_trie_storage(),
+			side_effects_handler,
 			changes,
 			&self.executor,
 			method,