FromOverseer and ToOverseer msgs and stopping

overseer/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2018"
 [dependencies]
 futures = "0.3.5"
 log = "0.4.8"
+futures-timer = "3.0.2"
 
 [dev-dependencies]
 futures = { version = "0.3.5", features = ["thread-pool"] }

overseer/examples/minimal-example.rs

@@ -39,7 +39,7 @@ impl Subsystem1 {
 		loop {
 			match ctx.try_recv().await {
 				Ok(Some(msg)) => {
-					log::info!("Subsystem1 received message {}", msg);
+					log::info!("Subsystem1 received message {:?}", msg);
 				}
 				Ok(None) => (),
 				Err(_) => {}
@@ -77,7 +77,7 @@ impl Subsystem2 {
 		loop {
 			match ctx.try_recv().await {
 				Ok(Some(msg)) => {
-					log::info!("Subsystem2 received message {}", msg);
+					log::info!("Subsystem2 received message {:?}", msg);
 					continue;
 				}
 				Ok(None) => { pending!(); }

overseer/src/lib.rs

@@ -62,15 +62,17 @@ use std::fmt::Debug;
 use std::pin::Pin;
 use std::collections::HashMap;
 use std::task::Poll;
+use std::time::Duration;
 
 use futures::channel::{mpsc, oneshot};
 use futures::{
-	pending, poll,
+	pending, poll, select,
 	future::{BoxFuture, RemoteHandle},
 	stream::FuturesUnordered,
 	task::{Spawn, SpawnExt},
-	Future, SinkExt, StreamExt,
+	Future, FutureExt, SinkExt, StreamExt,
 };
+use futures_timer::Delay;
 
 /// An error type that describes faults that may happen
 ///
@@ -101,7 +103,7 @@ const CHANNEL_CAPACITY: usize = 1024;
 /// It is also generic over `I` that is entended to be a type identifying
 /// different subsystems, again most likely this is a one large `enum`
 /// covering all possible subsystem kinds.
-enum OverseerMessage<M: Debug, I> {
+enum ToOverseer<M: Debug, I> {
 	/// This is a message generated by a `Subsystem`.
 	/// Wraps the messge itself and has an optional `to` of
 	/// someone who can receive this message.
@@ -150,38 +152,53 @@ impl OverseerHandler {
 	}
 }
 
-impl<M: Debug, I: Debug> Debug for OverseerMessage<M, I> {
+impl<M: Debug, I: Debug> Debug for ToOverseer<M, I> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
 		match self {
-			OverseerMessage::SubsystemMessage { to, msg } => {
+			ToOverseer::SubsystemMessage { to, msg } => {
 				write!(f, "OverseerMessage::SubsystemMessage{{ to: {:?}, msg: {:?} }}", to, msg)
 			}
-			OverseerMessage::SpawnJob { .. } => write!(f, "OverseerMessage::Spawn(..)")
+			ToOverseer::SpawnJob { .. } => write!(f, "OverseerMessage::Spawn(..)")
 		}
-	}
-}
+	} }
 
 /// A running instance of some `Subsystem`.
 struct SubsystemInstance<M: Debug, I> {
 	/// We talk to the `Overseer` over this channel.
-	rx: mpsc::Receiver<OverseerMessage<M, I>>,
+	rx: mpsc::Receiver<ToOverseer<M, I>>,
 	/// The `Overseer` talks to use over this channel.
-	tx: mpsc::Sender<M>,
+	tx: mpsc::Sender<FromOverseer<M>>,
 }
 
 /// A context type that is given to the `Subsystem` upon spawning
 /// that can be used by `Subsystem` to communicate with the outside world.
 pub struct SubsystemContext<M: Debug, I>{
-	rx: mpsc::Receiver<M>,
-	tx: mpsc::Sender<OverseerMessage<M, I>>,
+	rx: mpsc::Receiver<FromOverseer<M>>,
+	tx: mpsc::Sender<ToOverseer<M, I>>,
+}
+
+/// A signal used by `Overseer` to communicate with the `Subsystems`.
+#[derive(Debug)]
+pub enum OverseerSignal {
+	StartWork,
+	StopWork,
+}
+
+/// A message type that a `Subsystem` receives from the `Overseer`.
+/// It wraps siglans from the `Oveseer` and messages that are communicated
+/// between subsystems.
+#[derive(Debug)]
+pub enum FromOverseer<M: Debug> {
+	Signal(OverseerSignal),
+	Communication(M),
 }
 
 impl<M: Debug, I> SubsystemContext<M, I> {
 	/// Try to asyncronously receive a message.
 	///
 	/// This has to be used with caution, if you loop over this without
 	/// using `pending!()` macro you will end up with a busy loop!
-	pub async fn try_recv(&mut self) -> Result<Option<M>, ()> {
+	pub async fn try_recv(&mut self) -> Result<Option<FromOverseer<M>>, ()> {
 		match poll!(self.rx.next()) {
 			Poll::Ready(Some(msg)) => Ok(Some(msg)),
 			Poll::Ready(None) => Err(()),
@@ -190,7 +207,7 @@ impl<M: Debug, I> SubsystemContext<M, I> {
 	}
 
 	/// Receive a message.
-	pub async fn recv(&mut self) -> SubsystemResult<M> {
+	pub async fn recv(&mut self) -> SubsystemResult<FromOverseer<M>> {
 		self.rx.next().await.ok_or(SubsystemError)
 	}
 
@@ -199,7 +216,7 @@ impl<M: Debug, I> SubsystemContext<M, I> {
 	/// The message will be broadcasted to all other `Subsystem`s that can
 	/// receive it.
 	pub async fn broadcast_msg(&mut self, msg: M) {
-		let _ = self.tx.send(OverseerMessage::SubsystemMessage{
+		let _ = self.tx.send(ToOverseer::SubsystemMessage{
 			to: None,
 			msg,
 		}).await;
@@ -208,7 +225,7 @@ impl<M: Debug, I> SubsystemContext<M, I> {
 	/// Spawn a child `Subsystem` on the executor and get it's `I`d upon success.
 	pub async fn spawn(&mut self, s: Pin<Box<dyn Future<Output = ()> + Send>>) -> SubsystemResult<()> {
 		let (tx, rx) = oneshot::channel();
-		let _ = self.tx.send(OverseerMessage::SpawnJob {
+		let _ = self.tx.send(ToOverseer::SpawnJob {
 			s,
 			res: tx,
 		}).await;
@@ -218,13 +235,13 @@ impl<M: Debug, I> SubsystemContext<M, I> {
 
 	/// Send a direct message to some other `Subsystem` you know the `I`d of.
 	pub async fn send_msg(&mut self, to: I, msg: M) {
-		let _ = self.tx.send(OverseerMessage::SubsystemMessage{
+		let _ = self.tx.send(ToOverseer::SubsystemMessage{
 			to: Some(to),
 			msg,
 		}).await;
 	}
 
-	fn new(rx: mpsc::Receiver<M>, tx: mpsc::Sender<OverseerMessage<M, I>>) -> Self {
+	fn new(rx: mpsc::Receiver<FromOverseer<M>>, tx: mpsc::Sender<ToOverseer<M, I>>) -> Self {
 		Self {
 			rx,
 			tx,
@@ -334,6 +351,29 @@ where
 		}
 	}
 
+	// Stop the overseer.
+	async fn stop(mut self) {
+		for s in self.subsystems.iter_mut() {
+			if let Some(ref mut s) = s.1.instance {
+				let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::StopWork)).await;
+			}
+		}
+
+		let mut stop_delay = Delay::new(Duration::from_secs(1)).fuse();
+
+		loop {
+			select! {
+				_ = self.running_subsystems.next() => {
+					if self.running_subsystems.is_empty() {
+						break;
+					}
+				},
+				_ = stop_delay => break,
+				complete => break,
+			}
+		}
+	}
+
 	/// Run the `Overseer`.
 	// TODO: we have to
 	//   * Give out to the user some handler to communicate with the `Overseer`
@@ -347,10 +387,7 @@ where
 
 			while let Poll::Ready(Some(msg)) = poll!(&mut self.events_rx.next()) {
 				if let Event::Stop = msg {
-					// TODO: We should send stop messages to all subsystems, join them
-					// and wait for some timeout for them to gracefully shutdown and then
-					// just drop their handlers.
-					return
+					return self.stop().await;
 				}
 			}
 
@@ -368,12 +405,12 @@ where
 			// TODO: this desperately need refactoring.
 			for msg in msgs.into_iter() {
 				match msg.1 {
-					OverseerMessage::SubsystemMessage{ to, msg: m } => {
+					ToOverseer::SubsystemMessage{ to, msg: m } => {
 						match to {
 							Some(to) => {
 								if let Some(subsystem) = self.subsystems.get_mut(&to) {
 									if let Some(ref mut i) = subsystem.instance {
-										let _ = i.tx.send(m).await;
+										let _ = i.tx.send(FromOverseer::Communication(m)).await;
 									}
 								}
 							}
@@ -386,14 +423,14 @@ where
 
 									if s.subsystem.can_recv_msg(&m) {
 										if let Some(ref mut i) = s.instance {
-											let _ = i.tx.send(m.clone()).await;
+											let _ = i.tx.send(FromOverseer::Communication(m.clone())).await;
 										}
 									}
 								}
 							}
 						}
 					}
-					OverseerMessage::SpawnJob { s, res } => {
+					ToOverseer::SpawnJob { s, res } => {
 						log::info!("Spawn message");
 
 						let s = self.spawn_job(s);
@@ -466,11 +503,13 @@ mod tests {
 			SpawnedSubsystem(Box::pin(async move {
 				loop {
 					match ctx.recv().await {
-						Ok(msg) => {
+						Ok(FromOverseer::Communication(msg)) => {
 							let _ = sender.send(msg).await;
 							continue;
 						}
+						Ok(FromOverseer::Signal(OverseerSignal::StopWork)) => return,
 					    Err(_) => return,
+						_ => (),
 					}
 				}
 			}))
@@ -490,6 +529,9 @@ mod tests {
 						continue;
 					}
 					match ctx.try_recv().await {
+						Ok(Some(FromOverseer::Signal(OverseerSignal::StopWork))) => {
+							break;
+						}
 						Ok(Some(_)) => {
 							continue;
 						}
@@ -537,9 +579,8 @@ mod tests {
 				// just stay around for longer
 				loop {
 					match ctx.try_recv().await {
-						Ok(Some(_)) => {
-							continue;
-						}
+						Ok(Some(FromOverseer::Signal(OverseerSignal::StopWork))) => break,
+						Ok(Some(_)) => continue,
 						Err(_) => return,
 						_ => (),
 					}