WIP

Forked at: 85cd556
Parent branch: origin/master

client/cli/src/runner.rs

@@ -264,7 +264,7 @@ impl<C: SubstrateCli> Runner<C> {
 		F: FnOnce(Configuration) -> std::result::Result<T, sc_service::error::Error>,
 		T: AbstractService + Unpin,
 	{
-		let service = service_builder(self.config)?;
+		let mut service = service_builder(self.config)?;
 
 		let informant_future = sc_informant::build(&service, sc_informant::OutputFormat::Coloured);
 		let _informant_handle = self.tokio_runtime.spawn(informant_future);
@@ -275,12 +275,13 @@ impl<C: SubstrateCli> Runner<C> {
 		let _telemetry = service.telemetry();
 
 		{
-			let f = service.fuse();
+			let f = service.future().fuse();
 			self.tokio_runtime
 				.block_on(main(f))
 				.map_err(|e| e.to_string())?;
 		}
 
+		drop(service);
 		// The `service` **must** have been destroyed here for the shutdown signal to propagate
 		// to all the tasks. Dropping `tokio_runtime` will block the thread until all tasks have
 		// shut down.

client/service/src/lib.rs

@@ -42,7 +42,7 @@ use std::net::SocketAddr;
 use std::collections::HashMap;
 use std::time::Duration;
 use wasm_timer::Instant;
-use std::task::{Poll, Context};
+use std::task::Poll;
 use parking_lot::Mutex;
 
 use client::Client;
@@ -132,7 +132,7 @@ pub struct Service<TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> {
 impl<TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> Unpin for Service<TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> {}
 
 /// Abstraction over a Substrate service.
-pub trait AbstractService: Future<Output = Result<(), Error>> + Send + Unpin + Spawn + 'static {
+pub trait AbstractService: Send + Unpin + Spawn + 'static {
 	/// Type of block of this chain.
 	type Block: BlockT;
 	/// Backend storage for the client.
@@ -210,6 +210,9 @@ pub trait AbstractService: Future<Output = Result<(), Error>> + Send + Unpin + S
 
 	/// Get the prometheus metrics registry, if available.
 	fn prometheus_registry(&self) -> Option<prometheus_endpoint::Registry>;
+
+	/// Return a future that will end if an essential task fails
+	fn future<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = Result<(), Error>> + 'a>>;
 }
 
 impl<TBl, TBackend, TExec, TRtApi, TSc, TExPool, TOc> AbstractService for
@@ -310,27 +313,13 @@ where
 	fn prometheus_registry(&self) -> Option<prometheus_endpoint::Registry> {
 		self.prometheus_registry.clone()
 	}
-}
-
-impl<TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> Future for
-	Service<TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc>
-{
-	type Output = Result<(), Error>;
-
-	fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
-		let this = Pin::into_inner(self);
 
-		match Pin::new(&mut this.essential_failed_rx).poll_next(cx) {
-			Poll::Pending => {},
-			Poll::Ready(_) => {
-				// Ready(None) should not be possible since we hold a live
-				// sender.
-				return Poll::Ready(Err(Error::Other("Essential task failed.".into())));
-			}
-		}
+	fn future<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = Result<(), Error>> + 'a>> {
+		Box::pin(async move {
+			self.essential_failed_rx.next().await;
 
-		// The service future never ends.
-		Poll::Pending
+			Err(Error::Other("Essential task failed.".into()))
+		})
 	}
 }