Shutdown like a Sir (#1545)

## Issue Addressed
#1494 

## Proposed Changes
- Give the TaskExecutor the sender side of a channel that a task can clone to request shutting down
- The receiver side of this channel is in environment and now we block until ctrl+c or an internal shutdown signal is received
- The swarm now informs when it has reached 0 listeners
- The network receives this message and requests the shutdown

beacon_node/eth2_libp2p/src/service.rs

@@ -36,6 +36,8 @@ pub enum Libp2pEvent<TSpec: EthSpec> {
     Behaviour(BehaviourEvent<TSpec>),
     /// A new listening address has been established.
     NewListenAddr(Multiaddr),
+    /// We reached zero listening addresses.
+    ZeroListeners,
 }
 
 /// The configuration and state of the libp2p components for the beacon node.
@@ -283,10 +285,17 @@ impl<TSpec: EthSpec> Service<TSpec> {
                     debug!(self.log, "Listen address expired"; "multiaddr" => multiaddr.to_string())
                 }
                 SwarmEvent::ListenerClosed { addresses, reason } => {
-                    crit!(self.log, "Listener closed"; "addresses" => format!("{:?}", addresses), "reason" => format!("{:?}", reason))
+                    crit!(self.log, "Listener closed"; "addresses" => format!("{:?}", addresses), "reason" => format!("{:?}", reason));
+                    if Swarm::listeners(&self.swarm).count() == 0 {
+                        return Libp2pEvent::ZeroListeners;
+                    }
                 }
                 SwarmEvent::ListenerError { error } => {
-                    warn!(self.log, "Listener error"; "error" => format!("{:?}", error.to_string()))
+                    // this is non fatal, but we still check
+                    warn!(self.log, "Listener error"; "error" => format!("{:?}", error.to_string()));
+                    if Swarm::listeners(&self.swarm).count() == 0 {
+                        return Libp2pEvent::ZeroListeners;
+                    }
                 }
                 SwarmEvent::Dialing(peer_id) => {
                     debug!(self.log, "Dialing peer"; "peer_id" => peer_id.to_string());

beacon_node/eth2_libp2p/tests/common/mod.rs

@@ -94,8 +94,13 @@ pub async fn build_libp2p_instance(boot_nodes: Vec<Enr>, log: slog::Logger) -> L
     // launch libp2p service
 
     let (signal, exit) = exit_future::signal();
-    let executor =
+    let (shutdown_tx, _) = futures::channel::mpsc::channel(1);
-        environment::TaskExecutor::new(tokio::runtime::Handle::current(), exit, log.clone());
+    let executor = environment::TaskExecutor::new(
+        tokio::runtime::Handle::current(),
+        exit,
+        log.clone(),
+        shutdown_tx,
+    );
     Libp2pInstance(
         LibP2PService::new(executor, &config, EnrForkId::default(), &log)
             .await

beacon_node/network/src/service.rs

@@ -169,6 +169,7 @@ fn spawn_service<T: BeaconChainTypes>(
     mut service: NetworkService<T>,
 ) -> error::Result<()> {
     let mut exit_rx = executor.exit();
+    let mut shutdown_sender = executor.shutdown_sender();
 
     // spawn on the current executor
     executor.spawn_without_exit(async move {
@@ -376,6 +377,12 @@ fn spawn_service<T: BeaconChainTypes>(
                         Libp2pEvent::NewListenAddr(multiaddr) => {
                             service.network_globals.listen_multiaddrs.write().push(multiaddr);
                         }
+                        Libp2pEvent::ZeroListeners => {
+                            let _ = shutdown_sender.send("All listeners are closed. Unable to listen").await.map_err(|e| {
+                                warn!(service.log, "failed to send a shutdown signal"; "error" => e.to_string()
+                                )
+                            });
+                        }
                     }
                 }
             }

beacon_node/network/src/service/tests.rs

@@ -40,7 +40,13 @@ mod tests {
         let runtime = Runtime::new().unwrap();
 
         let (signal, exit) = exit_future::signal();
-        let executor = environment::TaskExecutor::new(runtime.handle().clone(), exit, log.clone());
+        let (shutdown_tx, _) = futures::channel::mpsc::channel(1);
+        let executor = environment::TaskExecutor::new(
+            runtime.handle().clone(),
+            exit,
+            log.clone(),
+            shutdown_tx,
+        );
 
         let mut config = NetworkConfig::default();
         config.libp2p_port = 21212;

lighthouse/environment/src/executor.rs

@@ -1,4 +1,5 @@
 use crate::metrics;
+use futures::channel::mpsc::Sender;
 use futures::prelude::*;
 use slog::{debug, trace};
 use tokio::runtime::Handle;
@@ -10,6 +11,12 @@ pub struct TaskExecutor {
     pub(crate) handle: Handle,
     /// The receiver exit future which on receiving shuts down the task
     pub(crate) exit: exit_future::Exit,
+    /// Sender given to tasks, so that if they encounter a state in which execution cannot
+    /// continue they can request that everything shuts down.
+    ///
+    /// The task must provide a reason for shutting down.
+    pub(crate) signal_tx: Sender<&'static str>,
+
     pub(crate) log: slog::Logger,
 }
 
@@ -18,8 +25,18 @@ impl TaskExecutor {
     ///
     /// Note: this function is mainly useful in tests. A `TaskExecutor` should be normally obtained from
     /// a [`RuntimeContext`](struct.RuntimeContext.html)
-    pub fn new(handle: Handle, exit: exit_future::Exit, log: slog::Logger) -> Self {
+    pub fn new(
-        Self { handle, exit, log }
+        handle: Handle,
+        exit: exit_future::Exit,
+        log: slog::Logger,
+        signal_tx: Sender<&'static str>,
+    ) -> Self {
+        Self {
+            handle,
+            exit,
+            signal_tx,
+            log,
+        }
     }
 
     /// Spawn a future on the tokio runtime wrapped in an `exit_future::Exit`. The task is canceled
@@ -121,6 +138,11 @@ impl TaskExecutor {
         self.exit.clone()
     }
 
+    /// Get a channel to request shutting down.
+    pub fn shutdown_sender(&self) -> Sender<&'static str> {
+        self.signal_tx.clone()
+    }
+
     /// Returns a reference to the logger.
     pub fn log(&self) -> &slog::Logger {
         &self.log

lighthouse/environment/src/lib.rs

@@ -9,7 +9,11 @@
 
 use eth2_config::Eth2Config;
 use eth2_testnet_config::Eth2TestnetConfig;
-use futures::channel::oneshot;
+use futures::channel::{
+    mpsc::{channel, Receiver, Sender},
+    oneshot,
+};
+use futures::{future, StreamExt};
 
 pub use executor::TaskExecutor;
 use slog::{info, o, Drain, Level, Logger};
@@ -260,10 +264,13 @@ impl<E: EthSpec> EnvironmentBuilder<E> {
     /// Consumes the builder, returning an `Environment`.
     pub fn build(self) -> Result<Environment<E>, String> {
         let (signal, exit) = exit_future::signal();
+        let (signal_tx, signal_rx) = channel(1);
         Ok(Environment {
             runtime: self
                 .runtime
                 .ok_or_else(|| "Cannot build environment without runtime".to_string())?,
+            signal_tx,
+            signal_rx: Some(signal_rx),
             signal: Some(signal),
             exit,
             log: self
@@ -295,6 +302,7 @@ impl<E: EthSpec> RuntimeContext<E> {
         Self {
             executor: TaskExecutor {
                 handle: self.executor.handle.clone(),
+                signal_tx: self.executor.signal_tx.clone(),
                 exit: self.executor.exit.clone(),
                 log: self.executor.log.new(o!("service" => service_name)),
             },
@@ -318,6 +326,10 @@ impl<E: EthSpec> RuntimeContext<E> {
 /// validator client, or to run tests that involve logging and async task execution.
 pub struct Environment<E: EthSpec> {
     runtime: Runtime,
+    /// Receiver side of an internal shutdown signal.
+    signal_rx: Option<Receiver<&'static str>>,
+    /// Sender to request shutting down.
+    signal_tx: Sender<&'static str>,
     signal: Option<exit_future::Signal>,
     exit: exit_future::Exit,
     log: Logger,
@@ -340,6 +352,7 @@ impl<E: EthSpec> Environment<E> {
         RuntimeContext {
             executor: TaskExecutor {
                 exit: self.exit.clone(),
+                signal_tx: self.signal_tx.clone(),
                 handle: self.runtime().handle().clone(),
                 log: self.log.clone(),
             },
@@ -353,6 +366,7 @@ impl<E: EthSpec> Environment<E> {
         RuntimeContext {
             executor: TaskExecutor {
                 exit: self.exit.clone(),
+                signal_tx: self.signal_tx.clone(),
                 handle: self.runtime().handle().clone(),
                 log: self.log.new(o!("service" => service_name)),
             },
@@ -361,8 +375,20 @@ impl<E: EthSpec> Environment<E> {
         }
     }
 
-    /// Block the current thread until Ctrl+C is received.
+    /// Block the current thread until a shutdown signal is received.
-    pub fn block_until_ctrl_c(&mut self) -> Result<(), String> {
+    ///
+    /// This can be either the user Ctrl-C'ing or a task requesting to shutdown.
+    pub fn block_until_shutdown_requested(&mut self) -> Result<(), String> {
+        // future of a task requesting to shutdown
+        let mut rx = self
+            .signal_rx
+            .take()
+            .ok_or("Inner shutdown already received")?;
+        let inner_shutdown =
+            async move { rx.next().await.ok_or("Internal shutdown channel exhausted") };
+        futures::pin_mut!(inner_shutdown);
+
+        // setup for handling a Ctrl-C
         let (ctrlc_send, ctrlc_oneshot) = oneshot::channel();
         let ctrlc_send_c = RefCell::new(Some(ctrlc_send));
         ctrlc::set_handler(move || {
@@ -372,10 +398,18 @@ impl<E: EthSpec> Environment<E> {
         })
         .map_err(|e| format!("Could not set ctrlc handler: {:?}", e))?;
 
-        // Block this thread until Crtl+C is pressed.
+        // Block this thread until a shutdown signal is received.
-        self.runtime()
+        match self
-            .block_on(ctrlc_oneshot)
+            .runtime()
-            .map_err(|e| format!("Ctrlc oneshot failed: {:?}", e))
+            .block_on(future::select(inner_shutdown, ctrlc_oneshot))
+        {
+            future::Either::Left((Ok(reason), _)) => {
+                info!(self.log, "Internal shutdown received"; "reason" => reason);
+                Ok(())
+            }
+            future::Either::Left((Err(e), _)) => Err(e.into()),
+            future::Either::Right((x, _)) => x.map_err(|e| format!("Ctrlc oneshot failed: {}", e)),
+        }
     }
 
     /// Shutdown the `tokio` runtime when all tasks are idle.

lighthouse/src/main.rs

@@ -300,8 +300,8 @@ fn run<E: EthSpec>(
         return Err("No subcommand supplied.".into());
     }
 
-    // Block this thread until Crtl+C is pressed.
+    // Block this thread until we get a ctrl-c or a task sends a shutdown signal.
-    environment.block_until_ctrl_c()?;
+    environment.block_until_shutdown_requested()?;
     info!(log, "Shutting down..");
 
     environment.fire_signal();

scripts/local_testnet/second_beacon_node.sh

@@ -16,5 +16,6 @@ exec lighthouse \
 	--testnet-dir $TESTNET_DIR \
 	--dummy-eth1 \
 	--http \
+    --port 9902 \
 	--http-port 6052 \
 	--boot-nodes $(cat $BEACON_DIR/beacon/network/enr.dat)