Initial: Offchain Workers (#1942)

* Refactor state-machine stuff. * Fix tests. * WiP * WiP2 * Service support for offchain workers. * Service support for offchain workers. * Testing offchain worker. * Initial version working. * Pass side effects in call. * Pass OffchainExt in context. * Submit extrinsics to the pool. * Support inherents. * Insert to inherents pool. * Inserting to the pool asynchronously. * Add test to offchain worker. * Implement convenience syntax for modules. * Dispatching offchain worker through executive. * Fix offchain test. * Remove offchain worker from timestamp. * Update Cargo.lock. * Address review comments. * Use latest patch version for futures. * Add CLI parameter for offchain worker. * Fix compilation. * Fix test. * Fix extrinsics format for tests. * Fix RPC test. * Bump spec version. * Fix executive. * Fix support macro. * Address grumbles. * Bump runtime
2026-06-12 08:51:09 +00:00 · 2019-03-25 23:22:11 +01:00
parent 3ee0e69463
commit e2f5e40876
58 changed files with 1158 additions and 178 deletions
@@ -0,0 +1,99 @@
+// Copyright 2019 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Substrate is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// 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::sync::Arc;
+use futures::{Stream, Future, sync::mpsc};
+use inherents::pool::InherentsPool;
+use log::{info, debug, warn};
+use parity_codec::Decode;
+use primitives::OffchainExt;
+use runtime_primitives::{
+	generic::BlockId,
+	traits::{self, Extrinsic},
+};
+use transaction_pool::txpool::{Pool, ChainApi};
+
+/// A message between the offchain extension and the processing thread.
+enum ExtMessage {
+	SubmitExtrinsic(Vec<u8>),
+}
+
+/// Asynchronous offchain API.
+///
+/// NOTE this is done to prevent recursive calls into the runtime (which are not supported currently).
+pub(crate) struct AsyncApi(mpsc::UnboundedSender<ExtMessage>);
+
+impl OffchainExt for AsyncApi {
+	fn submit_extrinsic(&mut self, ext: Vec<u8>) {
+		let _ = self.0.unbounded_send(ExtMessage::SubmitExtrinsic(ext));
+	}
+}
+
+/// Offchain extensions implementation API
+pub(crate) struct Api<A: ChainApi> {
+	receiver: Option<mpsc::UnboundedReceiver<ExtMessage>>,
+	transaction_pool: Arc<Pool<A>>,
+	inherents_pool: Arc<InherentsPool<<A::Block as traits::Block>::Extrinsic>>,
+	at: BlockId<A::Block>,
+}
+
+impl<A: ChainApi> Api<A> {
+	pub fn new(
+		transaction_pool: Arc<Pool<A>>,
+		inherents_pool: Arc<InherentsPool<<A::Block as traits::Block>::Extrinsic>>,
+		at: BlockId<A::Block>,
+	) -> (AsyncApi, Self) {
+		let (tx, rx) = mpsc::unbounded();
+		let api = Self {
+			receiver: Some(rx),
+			transaction_pool,
+			inherents_pool,
+			at,
+		};
+		(AsyncApi(tx), api)
+	}
+
+	/// Run a processing task for the API
+	pub fn process(mut self) -> impl Future<Item=(), Error=()> {
+		let receiver = self.receiver.take().expect("Take invoked only once.");
+
+		receiver.for_each(move |msg| {
+			match msg {
+				ExtMessage::SubmitExtrinsic(ext) => self.submit_extrinsic(ext),
+			}
+			Ok(())
+		})
+	}
+
+	fn submit_extrinsic(&mut self, ext: Vec<u8>) {
+		let xt = match <A::Block as traits::Block>::Extrinsic::decode(&mut &*ext) {
+			Some(xt) => xt,
+			None => {
+				warn!("Unable to decode extrinsic: {:?}", ext);
+				return
+			},
+		};
+
+		info!("Submitting to the pool: {:?} (isSigned: {:?})", xt, xt.is_signed());
+		match self.transaction_pool.submit_one(&self.at, xt.clone()) {
+			Ok(hash) => debug!("[{:?}] Offchain transaction added to the pool.", hash),
+			Err(_) => {
+				debug!("Offchain inherent added to the pool.");
+				self.inherents_pool.add(xt);
+			},
+		}
+	}
+}
@@ -0,0 +1,132 @@
+// Copyright 2019 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Substrate is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Substrate offchain workers.
+//!
+//! The offchain workers is a special function of the runtime that
+//! gets executed after block is imported. During execution
+//! it's able to asynchronously submit extrinsics that will either
+//! be propagated to other nodes (transactions) or will be
+//! added to the next block produced by the node as inherents.
+//!
+//! Offchain workers can be used for computation-heavy tasks
+//! that are not feasible for execution during regular block processing.
+//! It can either be tasks that no consensus is required for,
+//! or some form of consensus over the data can be built on-chain
+//! for instance via:
+//! 1. Challenge period for incorrect computations
+//! 2. Majority voting for results
+//! 3. etc
+
+#![warn(missing_docs)]
+
+use std::{
+	marker::PhantomData,
+	sync::Arc,
+};
+
+use client::runtime_api::ApiExt;
+use inherents::pool::InherentsPool;
+use log::{debug, warn};
+use primitives::ExecutionContext;
+use runtime_primitives::{
+	generic::BlockId,
+	traits::{self, ProvideRuntimeApi},
+};
+use tokio::runtime::TaskExecutor;
+use transaction_pool::txpool::{Pool, ChainApi};
+
+mod api;
+
+pub use offchain_primitives::OffchainWorkerApi;
+
+/// An offchain workers manager.
+#[derive(Debug)]
+pub struct OffchainWorkers<C, Block: traits::Block> {
+	client: Arc<C>,
+	inherents_pool: Arc<InherentsPool<<Block as traits::Block>::Extrinsic>>,
+	executor: TaskExecutor,
+	_block: PhantomData<Block>,
+}
+
+impl<C, Block: traits::Block> OffchainWorkers<C, Block> {
+	/// Creates new `OffchainWorkers`.
+	pub fn new(
+		client: Arc<C>,
+		inherents_pool: Arc<InherentsPool<<Block as traits::Block>::Extrinsic>>,
+		executor: TaskExecutor,
+	) -> Self {
+		Self {
+			client,
+			inherents_pool,
+			executor,
+			_block: PhantomData,
+		}
+	}
+}
+
+impl<C, Block> OffchainWorkers<C, Block> where
+	Block: traits::Block,
+	C: ProvideRuntimeApi,
+	C::Api: OffchainWorkerApi<Block>,
+{
+	/// Start the offchain workers after given block.
+	pub fn on_block_imported<A>(
+		&self,
+		number: &<Block::Header as traits::Header>::Number,
+		pool: &Arc<Pool<A>>,
+	) where
+		A: ChainApi<Block=Block> + 'static,
+	{
+		let runtime = self.client.runtime_api();
+		let at = BlockId::number(*number);
+		let has_api = runtime.has_api::<OffchainWorkerApi<Block>>(&at);
+		debug!("Checking offchain workers at {:?}: {:?}", at, has_api);
+
+		if has_api.unwrap_or(false) {
+			let (api, runner) = api::Api::new(pool.clone(), self.inherents_pool.clone(), at.clone());
+			self.executor.spawn(runner.process());
+
+			debug!("Running offchain workers at {:?}", at);
+			let api = Box::new(api);
+			runtime.offchain_worker_with_context(&at, ExecutionContext::OffchainWorker(api), *number).unwrap();
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use futures::Future;
+
+	#[test]
+	fn should_call_into_runtime_and_produce_extrinsic() {
+		// given
+		let _ = env_logger::try_init();
+		let runtime = tokio::runtime::Runtime::new().unwrap();
+		let client = Arc::new(test_client::new());
+		let pool = Arc::new(Pool::new(Default::default(), ::transaction_pool::ChainApi::new(client.clone())));
+		let inherents = Arc::new(InherentsPool::default());
+
+		// when
+		let offchain = OffchainWorkers::new(client, inherents.clone(), runtime.executor());
+		offchain.on_block_imported(&0u64, &pool);
+
+		// then
+		runtime.shutdown_on_idle().wait().unwrap();
+		assert_eq!(inherents.drain().len(), 1);
+	}
+}