always return a listening task, even in tests

Author: jgallagher

bfd-async/src/dispatcher.rs

@@ -191,14 +191,12 @@ struct Listener {
     /// listening for and the channel on which we should send any incoming
     /// packets from that peer.
     sessions: SharedSessions,
-    listen_task: Option<JoinHandle<()>>,
+    listen_task: JoinHandle<()>,
 }
 
 impl Drop for Listener {
     fn drop(&mut self) {
-        if let Some(listen_task) = self.listen_task.take() {
-            listen_task.abort();
-        }
+        self.listen_task.abort();
     }
 }
 
@@ -223,12 +221,10 @@ impl Listener {
     }
 
     async fn shutdown(mut self) {
-        if let Some(listen_task) = self.listen_task.take() {
-            listen_task.abort();
-            // Discard the result; this can only fail if the listen task
-            // panicked, but we're shutting it down anyway.
-            let _: Result<_, _> = listen_task.await;
-        }
+        self.listen_task.abort();
+        // Discard the result; this can only fail if the listen task
+        // panicked, but we're shutting it down anyway.
+        let _: Result<_, _> = (&mut self.listen_task).await;
     }
 
     fn remove_peer(&self, peer: IpAddr) -> ListenerRemovePeerResult {
@@ -269,15 +265,13 @@ trait ListenerBackend: Send + Sync + 'static {
     /// Bind a socket at `listen_addr` and spawn a task that reads packets and
     /// dispatches them to the per-peer channels in `sessions`.
     ///
-    /// Returns the spawned task's handle, or `None` for test fakes that don't
-    /// bind a real socket. A `Listener` holding `None` is shut down/dropped as
-    /// a no-op.
+    /// Returns the spawned task's handle.
     fn spawn(
         &self,
         listen_addr: SocketAddr,
         sessions: SharedSessions,
         log: Logger,
-    ) -> Result<Option<JoinHandle<()>>, AddPeerError>;
+    ) -> Result<JoinHandle<()>, AddPeerError>;
 }
 
 /// Production [`ListenerBackend`]: binds a real UDP socket and spawns a
@@ -290,7 +284,7 @@ impl ListenerBackend for TokioUdpBinder {
         listen_addr: SocketAddr,
         sessions: SharedSessions,
         log: Logger,
-    ) -> Result<Option<JoinHandle<()>>, AddPeerError> {
+    ) -> Result<JoinHandle<()>, AddPeerError> {
         // This is pretty spicy: we're using std's `UdpSocket` so we can
         // _synchronously_ bind, even though this function is ultimately called
         // from an async context. The arguments for this are:
@@ -313,7 +307,7 @@ impl ListenerBackend for TokioUdpBinder {
             UdpSocket::from_std(socket).map_err(AddPeerError::StdToTokio)?;
 
         let listen_task = ListenerTask::new(socket, sessions, log);
-        Ok(Some(tokio::spawn(listen_task.run())))
+        Ok(tokio::spawn(listen_task.run()))
     }
 }
 

bfd-async/src/dispatcher/end_to_end.rs

@@ -77,15 +77,15 @@ impl ListenerBackend for PreBoundSocketBackend {
         _listen_addr: SocketAddr,
         sessions: SharedSessions,
         log: Logger,
-    ) -> Result<Option<JoinHandle<()>>, AddPeerError> {
+    ) -> Result<JoinHandle<()>, AddPeerError> {
         let socket = self
             .socket
             .lock()
             .unwrap()
             .take()
             .expect("PreBoundSocketBackend::spawn called more than once");
         let listen_task = ListenerTask::new(socket, sessions, log);
-        Ok(Some(tokio::spawn(listen_task.run())))
+        Ok(tokio::spawn(listen_task.run()))
     }
 }
 

bfd-async/src/dispatcher/proptests.rs

@@ -41,7 +41,7 @@ impl ListenerBackend for FakeBackend {
         listen_addr: SocketAddr,
         _sessions: SharedSessions,
         _log: Logger,
-    ) -> Result<Option<JoinHandle<()>>, AddPeerError> {
+    ) -> Result<JoinHandle<()>, AddPeerError> {
         if self.fail_addrs.contains(&listen_addr) {
             Err(AddPeerError::Bind {
                 addr: listen_addr,
@@ -51,10 +51,9 @@ impl ListenerBackend for FakeBackend {
                 ),
             })
         } else {
-            // No real socket and no task: the dispatcher only needs the
-            // bookkeeping, and a `Listener` holding `None` shuts down as a
-            // no-op.
-            Ok(None)
+            // No real socket and no meaningful task: the dispatcher only needs
+            // the bookkeeping.
+            Ok(tokio::spawn(async {}))
         }
     }
 }
@@ -211,7 +210,11 @@ fn run(ops: Vec<Op>, fail_addrs: HashSet<SocketAddr>) {
 
 #[proptest]
 fn state_machine(#[strategy(vec(any::<Op>(), 0..40))] ops: Vec<Op>) {
-    run(ops, HashSet::new());
+    // `run()` isn't itself async, but it does spawn (no-op) tokio tasks so
+    // needs to execute in the context of a runtime.
+    tokio::runtime::Runtime::new().unwrap().block_on(async {
+        run(ops, HashSet::new());
+    });
 }
 
 #[proptest]
@@ -220,5 +223,9 @@ fn state_machine_with_bind_failures(
     #[strategy(proptest::collection::hash_set(arb_listen_addr(), 0..=3))]
     fail_addrs: HashSet<SocketAddr>,
 ) {
-    run(ops, fail_addrs);
+    // `run()` isn't itself async, but it does spawn (no-op) tokio tasks so
+    // needs to execute in the context of a runtime.
+    tokio::runtime::Runtime::new().unwrap().block_on(async {
+        run(ops, fail_addrs);
+    });
 }

bfd-async/src/dispatcher/tests.rs

@@ -68,7 +68,7 @@ impl ListenerBackend for TestBackend {
         listen_addr: SocketAddr,
         sessions: super::SharedSessions,
         log: Logger,
-    ) -> Result<Option<tokio::task::JoinHandle<()>>, crate::AddPeerError> {
+    ) -> Result<tokio::task::JoinHandle<()>, crate::AddPeerError> {
         assert_eq!(listen_addr, LISTEN_ADDR);
 
         let mut bound_address = self.bound_address.lock().unwrap();
@@ -103,7 +103,7 @@ impl ListenerBackend for TestBackend {
             }
         });
 
-        Ok(Some(listen_task))
+        Ok(listen_task)
     }
 }