Instantiate environment with asynchronous API (#768)

* point to in-progress Substrate branch

* instantiate environment async

* Fix futures

* Bump runtime

* Fix collation tests

* point to polkadot-master again

* point to polkadot-master again

* update deps

Co-authored-by: Ashley <ashley.ruglys@gmail.com>
This commit is contained in:
Robert Habermeier
2020-01-16 16:09:06 +01:00
committed by GitHub
parent 9c59e8a271
commit a5d9645bf4
8 changed files with 252 additions and 258 deletions
+220 -220
View File
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -23,7 +23,7 @@ sc-client-db = { git = "https://github.com/paritytech/substrate", branch = "polk
sc-executor = { git = "https://github.com/paritytech/substrate", branch = "polkadot-master" } sc-executor = { git = "https://github.com/paritytech/substrate", branch = "polkadot-master" }
service = { package = "polkadot-service", path = "../service", default-features = false } service = { package = "polkadot-service", path = "../service", default-features = false }
tokio = { version = "0.1.22", optional = true } tokio = { version = "0.2", features = ["rt-threaded"], optional = true }
wasm-bindgen = { version = "0.2.57", optional = true } wasm-bindgen = { version = "0.2.57", optional = true }
wasm-bindgen-futures = { version = "0.4.7", optional = true } wasm-bindgen-futures = { version = "0.4.7", optional = true }
+8 -21
View File
@@ -24,9 +24,7 @@ mod chain_spec;
mod browser; mod browser;
use chain_spec::ChainSpec; use chain_spec::ChainSpec;
use futures::{ use futures::{Future, future::{select, Either}, channel::oneshot};
Future, FutureExt, TryFutureExt, future::select, channel::oneshot, compat::Future01CompatExt,
};
#[cfg(feature = "cli")] #[cfg(feature = "cli")]
use tokio::runtime::Runtime; use tokio::runtime::Runtime;
use log::info; use log::info;
@@ -217,33 +215,22 @@ pub fn run_until_exit(
) -> error::Result<()> { ) -> error::Result<()> {
let (exit_send, exit) = oneshot::channel(); let (exit_send, exit) = oneshot::channel();
let executor = runtime.executor();
let informant = sc_cli::informant::build(&service); let informant = sc_cli::informant::build(&service);
let future = select(exit, informant)
.map(|_| Ok(()))
.compat();
executor.spawn(future); let handle = runtime.spawn(select(exit, informant));
// we eagerly drop the service so that the internal exit future is fired, // 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 // but we need to keep holding a reference to the global telemetry guard
let _telemetry = service.telemetry(); let _telemetry = service.telemetry();
let service_res = { let service_res = runtime.block_on(select(service, e));
let service = service
.map_err(|err| error::Error::Service(err))
.compat();
let select = select(service, e)
.map(|_| Ok(()))
.compat();
runtime.block_on(select)
};
let _ = exit_send.send(()); let _ = exit_send.send(());
use futures01::Future; runtime.block_on(handle);
// TODO [andre]: timeout this future substrate/#1318
let _ = runtime.shutdown_on_idle().wait();
service_res match service_res {
Either::Left((res, _)) => res.map_err(error::Error::Service),
Either::Right((_, _)) => Ok(())
}
} }
+1 -1
View File
@@ -21,7 +21,7 @@ polkadot-network = { path = "../network" }
polkadot-validation = { path = "../validation" } polkadot-validation = { path = "../validation" }
polkadot-service = { path = "../service" } polkadot-service = { path = "../service" }
log = "0.4.8" log = "0.4.8"
tokio = "0.1.22" tokio = "0.2"
futures-timer = "1.0" futures-timer = "1.0"
codec = { package = "parity-scale-codec", version = "1.1.0" } codec = { package = "parity-scale-codec", version = "1.1.0" }
+3 -3
View File
@@ -425,7 +425,7 @@ fn run_collator_node<S, E, P, Extrinsic>(
); );
let exit = inner_exit_2.clone(); let exit = inner_exit_2.clone();
tokio::spawn(future::select(res.boxed(), exit).map(drop).map(|_| Ok(())).compat()); tokio::spawn(future::select(res.boxed(), exit).map(drop));
}) })
}); });
@@ -449,11 +449,11 @@ fn run_collator_node<S, E, P, Extrinsic>(
inner_exit.clone() inner_exit.clone()
).map(drop); ).map(drop);
tokio::spawn(future.map(|_| Ok(())).compat()); tokio::spawn(future);
future::ready(()) future::ready(())
}); });
service.spawn_essential_task(work.map(|_| Ok::<_, ()>(())).compat()); service.spawn_essential_task(work);
polkadot_cli::run_until_exit(runtime, service, exit) polkadot_cli::run_until_exit(runtime, service, exit)
} }
+1 -1
View File
@@ -77,7 +77,7 @@ pub const VERSION: RuntimeVersion = RuntimeVersion {
spec_name: create_runtime_str!("kusama"), spec_name: create_runtime_str!("kusama"),
impl_name: create_runtime_str!("parity-kusama"), impl_name: create_runtime_str!("parity-kusama"),
authoring_version: 2, authoring_version: 2,
spec_version: 1040, spec_version: 1041,
impl_version: 2, impl_version: 2,
apis: RUNTIME_API_VERSIONS, apis: RUNTIME_API_VERSIONS,
}; };
+9 -5
View File
@@ -18,7 +18,11 @@
pub mod chain_spec; pub mod chain_spec;
use futures::{FutureExt, TryFutureExt, task::{Spawn, SpawnError, FutureObj}}; use futures::{
FutureExt, TryFutureExt,
task::{Spawn, SpawnError, FutureObj},
compat::Future01CompatExt,
};
use sc_client::LongestChain; use sc_client::LongestChain;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
@@ -455,9 +459,7 @@ pub fn new_full<Runtime, Dispatch, Extrinsic>(config: Configuration)
service.keystore(), service.keystore(),
dht_event_stream, dht_event_stream,
); );
let future01_authority_discovery = authority_discovery.map(|x| Ok(x)).compat(); service.spawn_task(authority_discovery);
service.spawn_task(future01_authority_discovery);
} }
} }
@@ -498,7 +500,9 @@ pub fn new_full<Runtime, Dispatch, Extrinsic>(config: Configuration)
executor: service.spawn_task_handle(), executor: service.spawn_task_handle(),
}; };
service.spawn_essential_task(grandpa::run_grandpa_voter(grandpa_config)?); service.spawn_essential_task(
grandpa::run_grandpa_voter(grandpa_config)?.compat().map(drop)
);
} else { } else {
grandpa::setup_disabled_grandpa( grandpa::setup_disabled_grandpa(
service.client(), service.client(),
+9 -6
View File
@@ -569,23 +569,24 @@ impl<C, N, P, SC, TxPool, B> consensus::Environment<Block> for ProposerFactory<C
// Rust bug: https://github.com/rust-lang/rust/issues/24159 // Rust bug: https://github.com/rust-lang/rust/issues/24159
sp_api::StateBackendFor<P, Block>: sp_api::StateBackend<HasherFor<Block>> + Send, sp_api::StateBackendFor<P, Block>: sp_api::StateBackend<HasherFor<Block>> + Send,
{ {
type CreateProposer = Pin<Box<
dyn Future<Output = Result<Self::Proposer, Self::Error>> + Send + Unpin + 'static
>>;
type Proposer = Proposer<P, TxPool, B>; type Proposer = Proposer<P, TxPool, B>;
type Error = Error; type Error = Error;
fn init( fn init(
&mut self, &mut self,
parent_header: &Header, parent_header: &Header,
) -> Result<Self::Proposer, Error> { ) -> Self::CreateProposer {
let parent_hash = parent_header.hash(); let parent_hash = parent_header.hash();
let parent_id = BlockId::hash(parent_hash); let parent_id = BlockId::hash(parent_hash);
let tracker = self.parachain_validation.get_or_instantiate( let maybe_proposer = self.parachain_validation.get_or_instantiate(
parent_hash, parent_hash,
&self.keystore, &self.keystore,
self.max_block_data_size, self.max_block_data_size,
)?; ).map(|tracker| Proposer {
Ok(Proposer {
client: self.parachain_validation.client.clone(), client: self.parachain_validation.client.clone(),
tracker, tracker,
parent_hash, parent_hash,
@@ -594,7 +595,9 @@ impl<C, N, P, SC, TxPool, B> consensus::Environment<Block> for ProposerFactory<C
transaction_pool: self.transaction_pool.clone(), transaction_pool: self.transaction_pool.clone(),
slot_duration: self.babe_slot_duration, slot_duration: self.babe_slot_duration,
backend: self.backend.clone(), backend: self.backend.clone(),
}) });
Box::pin(future::ready(maybe_proposer))
} }
} }