feat: make shutdown() immediate rather than queued

allowed.json

@@ -1683,6 +1683,21 @@
         }
       ]
     },
+    "tokio-util": {
+      "id": "tokio-util",
+      "source": "crates.io",
+      "licenses": [
+        {
+          "MIT": {
+            "copyright": {
+              "Lines": [
+                "Copyright (c) Tokio Contributors"
+              ]
+            }
+          }
+        }
+      ]
+    },
     "tracing": {
       "id": "tracing",
       "source": "crates.io",

rodbus/Cargo.toml

@@ -21,6 +21,7 @@ workspace = true
 crc = { version = "3", optional = true }
 scursor = "0.5.0"
 tokio = { workspace = true, features = ["net", "sync", "io-util", "io-std", "time", "rt", "rt-multi-thread", "macros"] }
+tokio-util = { workspace = true }
 tracing = { workspace = true }
 
 # TLS dependencies

rodbus/src/channel.rs

@@ -14,4 +14,9 @@ impl<T> Receiver<T> {
     pub(crate) async fn recv(&mut self) -> Result<T, Shutdown> {
         self.0.recv().await.ok_or(Shutdown)
     }
+
+    pub(crate) async fn close_and_drain(&mut self) {
+        self.0.close();
+        while self.0.recv().await.is_some() {}
+    }
 }

rodbus/src/client/channel.rs

@@ -5,6 +5,7 @@ use crate::client::requests::read_bits::ReadBits;
 use crate::client::requests::read_registers::ReadRegisters;
 use crate::client::requests::write_multiple::{MultipleWriteRequest, WriteMultiple};
 use crate::client::requests::write_single::SingleWrite;
+use crate::common::cancellation::{ShutdownHandle, ShutdownSignal};
 use crate::error::*;
 use crate::types::{AddressRange, BitIterator, Indexed, RegisterIterator, UnitId};
 use crate::DecodeLevel;
@@ -14,6 +15,7 @@ use crate::DecodeLevel;
 #[derive(Debug, Clone)]
 pub struct Channel {
     pub(crate) tx: tokio::sync::mpsc::Sender<Command>,
+    pub(crate) shutdown: ShutdownHandle,
 }
 
 /// A client channel task that has been created but not yet spawned.
@@ -27,6 +29,7 @@ pub struct Channel {
 /// caller is free to wrap [`run`](ClientTask::run) with their own instrumentation.
 pub struct ClientTask {
     inner: ClientTaskInner,
+    shutdown: ShutdownSignal,
 }
 
 enum ClientTaskInner {
@@ -36,29 +39,37 @@ enum ClientTaskInner {
 }
 
 impl ClientTask {
-    pub(crate) fn tcp(task: crate::tcp::client::TcpChannelTask) -> Self {
+    pub(crate) fn tcp(task: crate::tcp::client::TcpChannelTask, shutdown: ShutdownSignal) -> Self {
         Self {
             inner: ClientTaskInner::Tcp(task),
+            shutdown,
         }
     }
 
     #[cfg(feature = "serial")]
-    pub(crate) fn serial(task: crate::serial::client::SerialChannelTask) -> Self {
+    pub(crate) fn serial(
+        task: crate::serial::client::SerialChannelTask,
+        shutdown: ShutdownSignal,
+    ) -> Self {
         Self {
             inner: ClientTaskInner::Serial(task),
+            shutdown,
         }
     }
 
     /// Run the channel task until every [`Channel`] handle is dropped or [`Channel::shutdown`] is
     /// called.
     pub async fn run(self) {
-        match self.inner {
+        let ClientTask { inner, shutdown } = self;
+        match inner {
             ClientTaskInner::Tcp(mut task) => {
-                task.run().await;
+                shutdown.run_until_cancelled(task.run()).await;
+                task.shutdown().await;
             }
             #[cfg(feature = "serial")]
             ClientTaskInner::Serial(mut task) => {
-                task.run().await;
+                shutdown.run_until_cancelled(task.run()).await;
+                task.shutdown().await;
             }
         }
     }
@@ -124,6 +135,7 @@ impl Channel {
         listener: Option<Box<dyn crate::client::Listener<crate::client::PortState>>>,
     ) -> (Self, ClientTask) {
         let (tx, rx) = tokio::sync::mpsc::channel(max_queued_requests);
+        let (shutdown, signal) = crate::common::cancellation::pair();
         let task = crate::serial::client::SerialChannelTask::new(
             path,
             serial_settings,
@@ -132,7 +144,7 @@ impl Channel {
             decode,
             listener.unwrap_or_else(|| crate::client::NullListener::create()),
         );
-        (Channel { tx }, ClientTask::serial(task))
+        (Channel { tx, shutdown }, ClientTask::serial(task, signal))
     }
 
     /// Enable communications
@@ -149,10 +161,15 @@ impl Channel {
 
     /// Begin shutting down the channel task, even if one or more [`Channel`] handles are still alive
     ///
-    /// The task completes when it processes the command, which may be after this returns
-    pub async fn shutdown(&self) -> Result<(), Shutdown> {
-        self.tx.send(Command::Shutdown).await?;
-        Ok(())
+    /// Active work is cancelled at its next suspension point. A request already on the wire is
+    /// abandoned, and every queued request fails with [`RequestError::Shutdown`]. A write that was
+    /// already transmitted may still be applied by the server, so its outcome is indeterminate.
+    ///
+    /// The task reports the terminal state to its listener before completing, so a
+    /// [`Listener`](crate::client::Listener) that never completes that update keeps the task alive.
+    /// Calling this more than once, or after the task has already terminated, has no effect.
+    pub fn shutdown(&self) {
+        self.shutdown.cancel();
     }
 
     /// Read coils from the server

rodbus/src/client/listener.rs

@@ -3,6 +3,11 @@ use crate::MaybeAsync;
 /// Generic listener type that can be invoked multiple times
 pub trait Listener<T>: Send {
     /// Inform the listener that the value has changed
+    ///
+    /// The task delivering the update awaits the returned [`MaybeAsync`], so an implementation
+    /// that never completes stalls that task. This includes the terminal update delivered while
+    /// shutting down, which the task awaits before it finishes: a listener that never completes
+    /// that update keeps the task alive indefinitely.
     fn update(&mut self, _value: T) -> MaybeAsync<()> {
         MaybeAsync::ready(())
     }

rodbus/src/client/message.rs

@@ -27,8 +27,6 @@ pub(crate) enum Command {
     Request(Request),
     /// Change a setting
     Setting(Setting),
-    /// Shut down the channel task
-    Shutdown,
 }
 
 pub(crate) struct Request {

rodbus/src/client/task.rs

@@ -158,6 +158,10 @@ impl ClientLoop {
         self.enabled
     }
 
+    pub(crate) async fn shutdown(&mut self) {
+        self.rx.close_and_drain().await;
+    }
+
     async fn run_cmd(&mut self, cmd: Command, io: &mut PhysLayer) -> Result<(), SessionError> {
         match cmd {
             Command::Setting(setting) => {
@@ -168,7 +172,6 @@ impl ClientLoop {
                 Ok(())
             }
             Command::Request(mut request) => self.run_one_request(io, &mut request).await,
-            Command::Shutdown => Err(SessionError::Shutdown),
         }
     }
 
@@ -337,7 +340,6 @@ impl ClientLoop {
                     Err(StateChange::Disable)
                 }
             }
-            Command::Shutdown => Err(StateChange::Shutdown),
         }
     }
 
@@ -401,7 +403,10 @@ mod tests {
             let mut phys = PhysLayer::new_mock(mock);
             client_loop.run(&mut phys).await
         });
-        let channel = Channel { tx };
+        let channel = Channel {
+            tx,
+            shutdown: crate::common::cancellation::pair().0,
+        };
         (channel, join_handle, io_handle)
     }
 
@@ -432,83 +437,6 @@ mod tests {
         assert_eq!(task.await.unwrap(), SessionError::Shutdown);
     }
 
-    #[tokio::test]
-    async fn task_completes_when_shutdown_requested_with_a_handle_still_alive() {
-        let (channel, task, _io) = spawn_client_loop();
-
-        channel.shutdown().await.unwrap();
-        assert_eq!(task.await.unwrap(), SessionError::Shutdown);
-
-        // the handle outlived the task it terminated, and now reports that it is gone
-        assert_eq!(channel.shutdown().await, Err(Shutdown));
-        let res = channel
-            .read_coils(
-                RequestParam::new(UnitId::new(1), Duration::from_secs(1)),
-                AddressRange::try_from(7, 2).unwrap(),
-            )
-            .await;
-        assert_eq!(res, Err(RequestError::Shutdown));
-    }
-
-    #[tokio::test]
-    async fn transaction_in_flight_completes_before_requested_shutdown() {
-        let (channel, task, mut io) = spawn_client_loop();
-        let other_handle = channel.clone();
-
-        let range = AddressRange::try_from(7, 2).unwrap();
-        let request = get_framed_adu(FunctionCode::ReadCoils, &range);
-        let response = get_framed_adu(
-            FunctionCode::ReadCoils,
-            &BitWriter::new(ReadBitsRange { inner: range }, |idx| match idx {
-                7 => Ok(true),
-                8 => Ok(false),
-                _ => Err(ExceptionCode::IllegalDataAddress),
-            }),
-        );
-
-        let coils = tokio::spawn(async move {
-            channel
-                .read_coils(
-                    RequestParam::new(UnitId::new(1), Duration::from_secs(1)),
-                    range,
-                )
-                .await
-        });
-
-        // the request is on the wire, so the loop is inside a transaction
-        assert_eq!(io.next_event().await, Event::Write(request));
-
-        // shutdown queues behind that transaction rather than interrupting it
-        other_handle.shutdown().await.unwrap();
-        io.read(&response);
-
-        assert_eq!(
-            coils.await.unwrap().unwrap(),
-            vec![Indexed::new(7, true), Indexed::new(8, false)]
-        );
-        assert_eq!(task.await.unwrap(), SessionError::Shutdown);
-    }
-
-    #[tokio::test]
-    async fn shutdown_ends_the_task_while_it_is_failing_requests() {
-        // fail_requests() is the path taken while disconnected or waiting to retry, which reads
-        // the queue separately from the session loop and so needs its own handling
-        let (tx, rx) = tokio::sync::mpsc::channel(16);
-        let mut client_loop = ClientLoop::new(
-            rx.into(),
-            FrameWriter::tcp(),
-            FramedReader::tcp(),
-            DecodeLevel::nothing(),
-            None,
-        );
-        let channel = Channel { tx };
-
-        let task = tokio::spawn(async move { client_loop.fail_requests().await });
-        channel.shutdown().await.unwrap();
-
-        assert_eq!(task.await.unwrap(), StateChange::Shutdown);
-    }
-
     #[tokio::test]
     async fn returns_io_error_when_write_fails() {
         let (channel, _task, mut io) = spawn_client_loop();