From 7bc21ab4faa79c743b1c350e7f7e19d72f9f7b1a Mon Sep 17 00:00:00 2001 From: mdmzfzl Date: Fri, 28 Aug 2026 16:43:18 -0700 Subject: [PATCH 1/7] feat: make shutdown() immediate and synchronous Shutdown was delivered as a `Command::Shutdown` / `ServerCommand::Shutdown` on the same bounded mpsc that carries requests, so it was processed strictly FIFO behind whatever was already queued. Each queued request runs to completion, so the delay before the task ended was the *sum* of the pending response timeouts, not one of them. Worse, `shutdown()` was an `async fn` that awaited `send()` on a bounded channel, so when the queue was full -- exactly when a caller most wants out -- the shutdown request itself blocked waiting for a slot. Signal it out of band with a `CancellationToken` instead, selected against the task's inner loop. The losing future is dropped, which unwinds the task at whichever await point it is parked on: a socket write, a read awaiting a response, or a retry backoff sleep. `Promise::drop` already fails requests with `RequestError::Shutdown`, so both the abandoned transaction and everything left in the queue report the right error with no extra plumbing. Because the signal no longer needs backpressure, and cancelling an already cancelled or already dead task is a no-op rather than a failure, both methods become `pub fn shutdown(&self)` -- callable from a `Drop` impl, a signal handler, or any non-async context. Notes on the design: * The selects are `biased` with cancellation first. The inner loop is an infinite future, so an unbiased select would let it run whatever work was already ready before noticing the shutdown, roughly half the time. * The client's select sits inside `run()` around `run_inner()` rather than wrapping `run()`, so a cancelled task still reports its terminal `ClientState::Shutdown` / `PortState::Shutdown` to the listener. * TCP sessions are spawned and never joined, so each receives its own clone of the token. The server task returning only drops their command sender, which a session parked mid-write would not observe. * Dropping every handle keeps its previous behaviour of winding down at the next queue poll. Only `shutdown()` is immediate, which leaves callers both a graceful and an abrupt option. The trade-off is that a write already on the wire is abandoned, so the server may still apply it and the caller cannot know. That is inherent to any immediate shutdown and is documented on both methods. `ServerHandle::new()` gains a `CancellationToken` parameter. It is public but documented as existing only for the C bindings, and nothing in the tree calls it. `Channel::shutdown()` and `ServerHandle::shutdown()` landed after the 1.6.0-M2 publish, so no released signature changes. --- Cargo.lock | 1 + allowed.json | 15 +++ rodbus/Cargo.toml | 1 + rodbus/src/client/channel.rs | 32 ++++-- rodbus/src/client/message.rs | 2 - rodbus/src/client/task.rs | 84 +-------------- rodbus/src/serial/client.rs | 18 +++- rodbus/src/serial/server.rs | 14 +++ rodbus/src/server/mod.rs | 43 +++++--- rodbus/src/server/task.rs | 63 ++--------- rodbus/src/tcp/client.rs | 196 ++++++++++++++++++++++++++++++++++- rodbus/src/tcp/server.rs | 124 +++++++++++++++++----- rodbus/src/tcp/tls/client.rs | 4 +- 13 files changed, 399 insertions(+), 198 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e8859225..fec67c9a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -980,6 +980,7 @@ dependencies = [ "tokio-rustls", "tokio-serial", "tokio-test", + "tokio-util", "tracing", ] diff --git a/allowed.json b/allowed.json index 5a07b03b..56136aa3 100644 --- a/allowed.json +++ b/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", diff --git a/rodbus/Cargo.toml b/rodbus/Cargo.toml index 7ca4f465..f6a90437 100644 --- a/rodbus/Cargo.toml +++ b/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 diff --git a/rodbus/src/client/channel.rs b/rodbus/src/client/channel.rs index 330fb031..2e087631 100644 --- a/rodbus/src/client/channel.rs +++ b/rodbus/src/client/channel.rs @@ -8,20 +8,24 @@ use crate::client::requests::write_single::SingleWrite; use crate::error::*; use crate::types::{AddressRange, BitIterator, Indexed, RegisterIterator, UnitId}; use crate::DecodeLevel; +use tokio_util::sync::CancellationToken; -/// Async channel used to make requests. The associated task is shutdown when every handle is -/// dropped or [`Channel::shutdown`] is called. +/// Async channel used to make requests. +/// +/// The associated task terminates when every handle is dropped, or immediately when +/// [`Channel::shutdown`] is called. #[derive(Debug, Clone)] pub struct Channel { pub(crate) tx: tokio::sync::mpsc::Sender, + pub(crate) shutdown: CancellationToken, } /// A client channel task that has been created but not yet spawned. /// /// This is returned, alongside its [`Channel`] handle, by the `create_*_client_task` functions. /// Drive it to completion by awaiting [`ClientTask::run`], typically from within -/// [`tokio::spawn`]. The task completes when every associated [`Channel`] handle is dropped or -/// [`Channel::shutdown`] is called. +/// [`tokio::spawn`]. The task completes when every associated [`Channel`] handle is dropped, or +/// immediately when [`Channel::shutdown`] is called. /// /// Unlike the `spawn_*_client_task` functions, no tracing span is attached to the task, so the /// caller is free to wrap [`run`](ClientTask::run) with their own instrumentation. @@ -49,7 +53,7 @@ impl ClientTask { } } - /// Run the channel task until every [`Channel`] handle is dropped or [`Channel::shutdown`] is + /// Run the channel task until every [`Channel`] handle is dropped, or [`Channel::shutdown`] is /// called. pub async fn run(self) { match self.inner { @@ -124,6 +128,7 @@ impl Channel { listener: Option>>, ) -> (Self, ClientTask) { let (tx, rx) = tokio::sync::mpsc::channel(max_queued_requests); + let shutdown = CancellationToken::new(); let task = crate::serial::client::SerialChannelTask::new( path, serial_settings, @@ -131,8 +136,9 @@ impl Channel { retry, decode, listener.unwrap_or_else(|| crate::client::NullListener::create()), + shutdown.clone(), ); - (Channel { tx }, ClientTask::serial(task)) + (Channel { tx, shutdown }, ClientTask::serial(task)) } /// Enable communications @@ -147,12 +153,16 @@ impl Channel { Ok(()) } - /// Begin shutting down the channel task, even if one or more [`Channel`] handles are still alive + /// Shut down the channel task immediately, even if other [`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(()) + /// Unlike dropping every handle, this does not wait for queued work. The task is cancelled at + /// its next suspension point, so a request already on the wire is abandoned rather than awaited + /// 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. + /// + /// 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 diff --git a/rodbus/src/client/message.rs b/rodbus/src/client/message.rs index aa099658..1fe86276 100644 --- a/rodbus/src/client/message.rs +++ b/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 { diff --git a/rodbus/src/client/task.rs b/rodbus/src/client/task.rs index b06413c4..f8b3a47b 100644 --- a/rodbus/src/client/task.rs +++ b/rodbus/src/client/task.rs @@ -168,7 +168,6 @@ impl ClientLoop { Ok(()) } Command::Request(mut request) => self.run_one_request(io, &mut request).await, - Command::Shutdown => Err(SessionError::Shutdown), } } @@ -337,7 +336,6 @@ impl ClientLoop { Err(StateChange::Disable) } } - Command::Shutdown => Err(StateChange::Shutdown), } } @@ -401,7 +399,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: tokio_util::sync::CancellationToken::new(), + }; (channel, join_handle, io_handle) } @@ -432,83 +433,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(); diff --git a/rodbus/src/serial/client.rs b/rodbus/src/serial/client.rs index b8453e8e..d705c9e6 100644 --- a/rodbus/src/serial/client.rs +++ b/rodbus/src/serial/client.rs @@ -7,6 +7,7 @@ use crate::client::task::{ClientLoop, SessionError, StateChange}; use crate::client::{Listener, PortState, RetryStrategy}; use crate::common::frame::{FrameWriter, FramedReader}; use crate::error::Shutdown; +use tokio_util::sync::CancellationToken; pub(crate) struct SerialChannelTask { path: String, @@ -14,9 +15,11 @@ pub(crate) struct SerialChannelTask { retry: Box, client_loop: ClientLoop, listener: Box>, + shutdown: CancellationToken, } impl SerialChannelTask { + #[allow(clippy::too_many_arguments)] pub(crate) fn new( path: &str, serial_settings: SerialSettings, @@ -24,6 +27,7 @@ impl SerialChannelTask { retry: Box, decode: DecodeLevel, listener: Box>, + shutdown: CancellationToken, ) -> Self { Self { path: path.to_string(), @@ -37,14 +41,24 @@ impl SerialChannelTask { None, ), listener, + shutdown, } } pub(crate) async fn run(&mut self) -> Shutdown { self.listener.update(PortState::Disabled).get().await; - let ret = self.run_inner().await; + // the clone releases the borrow of self that run_inner() needs mutably + let shutdown = self.shutdown.clone(); + tokio::select! { + // biased so that a cancelled task cannot be given another poll of run_inner(), which + // would let it finish work that is already ready before noticing the shutdown + biased; + _ = shutdown.cancelled() => {} + _ = self.run_inner() => {} + } + // outside the select so that a cancelled task still reports its terminal state self.listener.update(PortState::Shutdown).get().await; - ret + Shutdown } async fn run_inner(&mut self) -> Shutdown { diff --git a/rodbus/src/serial/server.rs b/rodbus/src/serial/server.rs index 5d8cc81a..6e5b2833 100644 --- a/rodbus/src/serial/server.rs +++ b/rodbus/src/serial/server.rs @@ -2,6 +2,7 @@ use crate::common::phys::PhysLayer; use crate::server::task::SessionTask; use crate::server::RequestHandler; use crate::{RequestError, RetryStrategy, SerialSettings, Shutdown}; +use tokio_util::sync::CancellationToken; pub(crate) struct RtuServerTask where @@ -11,6 +12,7 @@ where pub(crate) retry: Box, pub(crate) settings: SerialSettings, pub(crate) session: SessionTask, + pub(crate) shutdown: CancellationToken, } impl RtuServerTask @@ -18,6 +20,18 @@ where T: RequestHandler, { pub(crate) async fn run(&mut self) -> Shutdown { + // the clone releases the borrow of self that run_inner() needs mutably + let shutdown = self.shutdown.clone(); + tokio::select! { + // biased so that a cancelled task cannot be given another poll of run_inner(), which + // would let it finish work that is already ready before noticing the shutdown + biased; + _ = shutdown.cancelled() => Shutdown, + res = self.run_inner() => res, + } + } + + async fn run_inner(&mut self) -> Shutdown { loop { match crate::serial::open(&self.port, self.settings) { Ok(serial) => { diff --git a/rodbus/src/server/mod.rs b/rodbus/src/server/mod.rs index 68319728..0046d9bc 100644 --- a/rodbus/src/server/mod.rs +++ b/rodbus/src/server/mod.rs @@ -5,6 +5,7 @@ use tracing::Instrument; use crate::decode::DecodeLevel; use crate::server::task::ServerCommand; use crate::tcp::server::{ServerTask as TcpServerTask, TcpServerConnectionHandler}; +use tokio_util::sync::CancellationToken; /// server handling mod address_filter; @@ -29,19 +30,22 @@ pub use crate::tcp::tls::server::TlsServerConfig; #[cfg(feature = "enable-tls")] pub use crate::tcp::tls::*; -/// Handle to the server async task. The associated task is shutdown when every handle is dropped -/// or [`ServerHandle::shutdown`] is called. +/// Handle to the server async task. +/// +/// The associated task terminates when the handle is dropped, or immediately when +/// [`ServerHandle::shutdown`] is called. #[derive(Debug)] pub struct ServerHandle { tx: tokio::sync::mpsc::Sender, + shutdown: CancellationToken, } /// A server task that has been created but not yet spawned. /// /// This is returned, alongside its [`ServerHandle`], by the `create_*_server_task` functions. /// Drive it to completion by awaiting [`ServerTask::run`], typically from within -/// [`tokio::spawn`]. The task completes when every associated [`ServerHandle`] is dropped or -/// [`ServerHandle::shutdown`] is called. +/// [`tokio::spawn`]. The task completes when the associated [`ServerHandle`] is dropped, or +/// immediately when [`ServerHandle::shutdown`] is called. /// /// Unlike the `spawn_*_server_task` functions, no tracing span is attached to the task, so the /// caller is free to wrap [`run`](ServerTask::run) with their own instrumentation. @@ -72,7 +76,7 @@ impl ServerTask { } } - /// Run the server task until every [`ServerHandle`] is dropped or + /// Run the server task until the associated [`ServerHandle`] is dropped, or /// [`ServerHandle::shutdown`] is called. pub async fn run(self) { match self.inner { @@ -89,8 +93,8 @@ impl ServerHandle { /// Construct a [ServerHandle] from its fields /// /// This function is only required for the C bindings - pub fn new(tx: tokio::sync::mpsc::Sender) -> Self { - ServerHandle { tx } + pub fn new(tx: tokio::sync::mpsc::Sender, shutdown: CancellationToken) -> Self { + ServerHandle { tx, shutdown } } /// Change the decoding level for future sessions and all active sessions @@ -99,12 +103,15 @@ impl ServerHandle { Ok(()) } - /// Begin shutting down the server task, even if one or more [`ServerHandle`]s are still alive + /// Shut down the server task and every active session immediately /// - /// The task completes when it processes the command, which may be after this returns - pub async fn shutdown(&self) -> Result<(), Shutdown> { - self.tx.send(ServerCommand::Shutdown).await?; - Ok(()) + /// Unlike dropping the handle, this does not wait for sessions to reach a quiescent point. The + /// listener stops accepting and each session is cancelled at its next suspension point, so a + /// response that is partially written is abandoned. + /// + /// Calling this more than once, or after the task has already terminated, has no effect. + pub fn shutdown(&self) { + self.shutdown.cancel(); } } @@ -156,6 +163,7 @@ pub fn create_tcp_server_task( decode: DecodeLevel, ) -> (ServerHandle, ServerTask) { let (tx, rx) = tokio::sync::mpsc::channel(SERVER_COMMAND_CHANNEL_CAPACITY); + let shutdown = CancellationToken::new(); let task = TcpServerTask::new( max_sessions, listener, @@ -163,9 +171,10 @@ pub fn create_tcp_server_task( TcpServerConnectionHandler::Tcp, filter, decode, + shutdown.clone(), ); - (ServerHandle::new(tx), ServerTask::tcp(task, rx)) + (ServerHandle::new(tx, shutdown), ServerTask::tcp(task, rx)) } /// Spawns a RTU server task onto the runtime. @@ -215,6 +224,7 @@ pub fn create_rtu_server_task( decode: DecodeLevel, ) -> (ServerHandle, ServerTask) { let (tx, rx) = tokio::sync::mpsc::channel(SERVER_COMMAND_CHANNEL_CAPACITY); + let shutdown = CancellationToken::new(); let session = crate::server::task::SessionTask::new( handlers, crate::server::task::AuthorizationType::None, @@ -229,9 +239,10 @@ pub fn create_rtu_server_task( retry, settings, session, + shutdown: shutdown.clone(), }; - (ServerHandle::new(tx), ServerTask::rtu(rtu)) + (ServerHandle::new(tx, shutdown), ServerTask::rtu(rtu)) } /// Spawns a "raw" TLS server task onto the runtime. This TLS server does NOT require that @@ -413,6 +424,7 @@ fn create_tls_server_task_impl( decode: DecodeLevel, ) -> (ServerHandle, ServerTask) { let (tx, rx) = tokio::sync::mpsc::channel(SERVER_COMMAND_CHANNEL_CAPACITY); + let shutdown = CancellationToken::new(); let task = TcpServerTask::new( max_sessions, listener, @@ -420,7 +432,8 @@ fn create_tls_server_task_impl( TcpServerConnectionHandler::Tls(tls_config, auth_handler), filter, decode, + shutdown.clone(), ); - (ServerHandle::new(tx), ServerTask::tcp(task, rx)) + (ServerHandle::new(tx, shutdown), ServerTask::tcp(task, rx)) } diff --git a/rodbus/src/server/task.rs b/rodbus/src/server/task.rs index 644fa6a1..d5c1c39b 100644 --- a/rodbus/src/server/task.rs +++ b/rodbus/src/server/task.rs @@ -19,8 +19,6 @@ use std::sync::Arc; pub enum ServerCommand { /// Change the decoding level dynamically ChangeDecoding(DecodeLevel), - /// Shut down the server task - Shutdown, } pub(crate) struct SessionTask @@ -110,11 +108,7 @@ where loop { match self.commands.recv().await { None => return Shutdown, - Some(command) => { - if self.apply_command(command).is_err() { - return Shutdown; - } - } + Some(command) => self.apply_command(command), } } } @@ -128,23 +122,18 @@ where cmd = self.commands.recv() => { match cmd { None => Err(crate::error::RequestError::Shutdown), - Some(command) => match self.apply_command(command) { - Ok(()) => Ok(()), - Err(Shutdown) => Err(crate::error::RequestError::Shutdown), - }, + Some(command) => { + self.apply_command(command); + Ok(()) + } } } } } - /// Apply a command, returning `Err(Shutdown)` if the session should complete - fn apply_command(&mut self, command: ServerCommand) -> Result<(), Shutdown> { + fn apply_command(&mut self, command: ServerCommand) { match command { - ServerCommand::ChangeDecoding(level) => { - self.decode = level; - Ok(()) - } - ServerCommand::Shutdown => Err(Shutdown), + ServerCommand::ChangeDecoding(level) => self.decode = level, } } @@ -292,41 +281,3 @@ impl AuthorizationType { } } } - -#[cfg(test)] -mod tests { - use super::*; - use crate::server::ServerHandle; - - struct DefaultHandler; - impl RequestHandler for DefaultHandler {} - - #[tokio::test] - async fn session_ends_when_shutdown_requested_with_the_handle_still_alive() { - // the session reads commands straight from the handle on the RTU path - let (tx, rx) = tokio::sync::mpsc::channel(8); - let (mock, _io) = sfio_tokio_mock_io::mock(); - let mut session = SessionTask::new( - ServerHandlerMap::single(UnitId::new(1), DefaultHandler.wrap()), - AuthorizationType::None, - FrameWriter::tcp(), - FramedReader::tcp(), - rx, - DecodeLevel::nothing(), - ); - let handle = ServerHandle::new(tx); - - // the mock never yields a frame, so the loop is parked on the command queue - let task = tokio::spawn(async move { - let mut phys = PhysLayer::new_mock(mock); - session.run(&mut phys).await - }); - - handle.shutdown().await.unwrap(); - - assert_eq!(task.await.unwrap(), RequestError::Shutdown); - - // the handle outlived the session it terminated, and now reports that it is gone - assert_eq!(handle.shutdown().await, Err(Shutdown)); - } -} diff --git a/rodbus/src/tcp/client.rs b/rodbus/src/tcp/client.rs index 23fb46ab..59bfcbc6 100644 --- a/rodbus/src/tcp/client.rs +++ b/rodbus/src/tcp/client.rs @@ -11,6 +11,7 @@ use crate::retry::RetryStrategy; use crate::{ChannelLoggingMode, ClientOptions}; use tokio::net::TcpStream; +use tokio_util::sync::CancellationToken; macro_rules! log_channel_event { ($channel_logging:expr, $($arg:tt)*) => { @@ -44,6 +45,7 @@ pub(crate) fn create_tcp_channel( options: ClientOptions, ) -> (Channel, ClientTask) { let (tx, rx) = tokio::sync::mpsc::channel(options.max_queued_requests); + let shutdown = CancellationToken::new(); let task = TcpChannelTask::new( host, rx.into(), @@ -51,8 +53,9 @@ pub(crate) fn create_tcp_channel( connect_retry, options, listener, + shutdown.clone(), ); - (Channel { tx }, ClientTask::tcp(task)) + (Channel { tx, shutdown }, ClientTask::tcp(task)) } pub(crate) enum TcpTaskConnectionHandler { @@ -82,6 +85,7 @@ pub(crate) struct TcpChannelTask { client_loop: ClientLoop, listener: Box>, channel_logging: ChannelLoggingMode, + shutdown: CancellationToken, } impl TcpChannelTask { @@ -92,6 +96,7 @@ impl TcpChannelTask { connect_retry: Box, options: ClientOptions, listener: Box>, + shutdown: CancellationToken, ) -> Self { Self { host, @@ -106,15 +111,25 @@ impl TcpChannelTask { ), listener, channel_logging: options.channel_logging, + shutdown, } } // runs until it is shut down pub(crate) async fn run(&mut self) -> Shutdown { self.listener.update(ClientState::Disabled).get().await; - let ret = self.run_inner().await; + // the clone releases the borrow of self that run_inner() needs mutably + let shutdown = self.shutdown.clone(); + tokio::select! { + // biased so that a cancelled task cannot be given another poll of run_inner(), which + // would let it finish work that is already ready before noticing the shutdown + biased; + _ = shutdown.cancelled() => {} + _ = self.run_inner() => {} + } + // outside the select so that a cancelled task still reports its terminal state self.listener.update(ClientState::Shutdown).get().await; - ret + Shutdown } async fn run_inner(&mut self) -> Shutdown { @@ -209,9 +224,20 @@ impl TcpChannelTask { #[cfg(test)] mod tests { use super::*; + use crate::client::RequestParam; use crate::maybe_async::MaybeAsync; use crate::retry::default_retry_strategy; + use crate::{AddressRange, RequestError, RetryStrategy, UnitId}; use std::net::Ipv4Addr; + use std::time::Duration; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + /// A request timeout long enough that any test which observes a request completing before it + /// elapses has necessarily observed cancellation rather than a timeout + const NEVER: Duration = Duration::from_secs(3600); + + /// Bounds how long a test will hang if shutdown stops being immediate + const TEST_TIMEOUT: Duration = Duration::from_secs(5); struct StateRecorder { tx: tokio::sync::mpsc::UnboundedSender, @@ -224,6 +250,20 @@ mod tests { } } + struct NeverRetry; + + impl RetryStrategy for NeverRetry { + fn reset(&mut self) {} + + fn after_failed_connect(&mut self) -> Duration { + NEVER + } + + fn after_disconnect(&mut self) -> Duration { + NEVER + } + } + #[tokio::test] async fn reports_shutdown_state_when_shutdown_requested() { let (tx, mut states) = tokio::sync::mpsc::unbounded_channel(); @@ -236,11 +276,157 @@ mod tests { ); let task = tokio::spawn(task.run()); - channel.shutdown().await.unwrap(); - task.await.unwrap(); + channel.shutdown(); + tokio::time::timeout(TEST_TIMEOUT, task) + .await + .unwrap() + .unwrap(); // the task announced its own termination on the way out assert_eq!(states.recv().await, Some(ClientState::Disabled)); assert_eq!(states.recv().await, Some(ClientState::Shutdown)); } + + #[tokio::test] + async fn shutdown_abandons_the_request_in_flight() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let (channel, task) = create_tcp_channel( + HostAddr::ip(addr.ip(), addr.port()), + default_retry_strategy(), + crate::client::NullListener::create(), + ClientOptions::default(), + ); + let task = tokio::spawn(task.run()); + channel.enable().await.unwrap(); + + // accept the connection and read the request, but never answer it + let (mut socket, _) = listener.accept().await.unwrap(); + let requester = channel.clone(); + let coils = tokio::spawn(async move { + requester + .read_coils( + RequestParam::new(UnitId::new(1), NEVER), + AddressRange::try_from(7, 2).unwrap(), + ) + .await + }); + let mut buffer = [0u8; 32]; + let read = socket.read(&mut buffer).await.unwrap(); + assert!(read > 0, "the request never reached the wire"); + + // the loop is now parked awaiting a response that will never arrive + channel.shutdown(); + + assert_eq!( + tokio::time::timeout(TEST_TIMEOUT, coils) + .await + .expect("shutdown did not abandon the in-flight request") + .unwrap(), + Err(RequestError::Shutdown) + ); + tokio::time::timeout(TEST_TIMEOUT, task) + .await + .unwrap() + .unwrap(); + + // the peer sees the connection go away rather than a lingering half-open socket + assert_eq!(socket.read(&mut buffer).await.unwrap(), 0); + } + + #[tokio::test] + async fn shutdown_interrupts_the_wait_before_reconnecting() { + // bind and immediately drop, so connecting fails and the task enters its retry delay + let addr = { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + listener.local_addr().unwrap() + }; + + let (tx, mut states) = tokio::sync::mpsc::unbounded_channel(); + let (channel, task) = create_tcp_channel( + HostAddr::ip(addr.ip(), addr.port()), + Box::new(NeverRetry), + Box::new(StateRecorder { tx }), + ClientOptions::default(), + ); + let task = tokio::spawn(task.run()); + channel.enable().await.unwrap(); + + // wait until the task is actually sleeping on the retry delay + loop { + match states.recv().await.unwrap() { + ClientState::WaitAfterFailedConnect(delay) => { + assert_eq!(delay, NEVER); + break; + } + _ => continue, + } + } + + channel.shutdown(); + + tokio::time::timeout(TEST_TIMEOUT, task) + .await + .expect("shutdown did not interrupt the retry delay") + .unwrap(); + } + + #[tokio::test] + async fn queued_requests_fail_when_shutdown_is_requested() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let (channel, task) = create_tcp_channel( + HostAddr::ip(addr.ip(), addr.port()), + default_retry_strategy(), + crate::client::NullListener::create(), + ClientOptions::default(), + ); + let task = tokio::spawn(task.run()); + channel.enable().await.unwrap(); + + let (mut socket, _) = listener.accept().await.unwrap(); + + // one request occupies the loop, the rest sit in the queue behind it + let queued: Vec<_> = (0..4) + .map(|_| { + let requester = channel.clone(); + tokio::spawn(async move { + requester + .read_coils( + RequestParam::new(UnitId::new(1), NEVER), + AddressRange::try_from(7, 2).unwrap(), + ) + .await + }) + }) + .collect(); + + let mut buffer = [0u8; 32]; + assert!(socket.read(&mut buffer).await.unwrap() > 0); + + channel.shutdown(); + + // every request fails, in flight or queued, without waiting out a single timeout + for handle in queued { + assert_eq!( + tokio::time::timeout(TEST_TIMEOUT, handle) + .await + .expect("a queued request was left waiting") + .unwrap(), + Err(RequestError::Shutdown) + ); + } + tokio::time::timeout(TEST_TIMEOUT, task) + .await + .unwrap() + .unwrap(); + + // the handle outlives the task it terminated + channel.shutdown(); + assert_eq!(channel.enable().await, Err(crate::error::Shutdown)); + + drop(socket.write(&[]).await); + } } diff --git a/rodbus/src/tcp/server.rs b/rodbus/src/tcp/server.rs index 79154588..bb613bab 100644 --- a/rodbus/src/tcp/server.rs +++ b/rodbus/src/tcp/server.rs @@ -11,6 +11,7 @@ use crate::server::task::{AuthorizationType, ServerCommand}; use crate::server::AddressFilter; use std::net::SocketAddr; use tokio::net::TcpListener; +use tokio_util::sync::CancellationToken; #[cfg(feature = "enable-tls")] use crate::server::AuthorizationHandler; @@ -106,12 +107,14 @@ pub(crate) struct ServerTask { decode: DecodeLevel, tx: tokio::sync::mpsc::Sender, rx: tokio::sync::mpsc::Receiver, + shutdown: CancellationToken, } impl ServerTask where T: RequestHandler, { + #[allow(clippy::too_many_arguments)] pub(crate) fn new( max_sessions: usize, listener: TcpListener, @@ -119,6 +122,7 @@ where connection_handler: TcpServerConnectionHandler, filter: AddressFilter, decode: DecodeLevel, + shutdown: CancellationToken, ) -> Self { let (tx, rx) = tokio::sync::mpsc::channel(8); @@ -131,6 +135,7 @@ where decode, tx, rx, + shutdown, } } @@ -141,8 +146,6 @@ where tracing::info!("changed decoding level to {:?}", level); self.decode = level; } - // handled by the caller, which returns instead of forwarding it to the sessions - ServerCommand::Shutdown => return, } for sender in self.tracker.sessions.values_mut() { @@ -153,15 +156,24 @@ where } pub(crate) async fn run(&mut self, mut commands: tokio::sync::mpsc::Receiver) { + // the clone releases the borrow of self that run_inner() needs mutably + let shutdown = self.shutdown.clone(); + tokio::select! { + // biased so that a cancelled task cannot be given another poll of run_inner(), which + // would let it accept another connection before noticing the shutdown + biased; + _ = shutdown.cancelled() => { + tracing::info!("server shutdown requested"); + } + _ = self.run_inner(&mut commands) => {} + } + } + + async fn run_inner(&mut self, commands: &mut tokio::sync::mpsc::Receiver) { loop { tokio::select! { command = commands.recv() => { match command { - // dropping the tracker ends every session, just as dropping the handle does - Some(ServerCommand::Shutdown) => { - tracing::info!("server shutdown requested"); - return; - } Some(command) => self.apply_command(command).await, None => { tracing::info!("server shutdown"); @@ -211,17 +223,23 @@ where let connection_handler = self.connection_handler.clone(); let handler_map = self.handlers.clone(); let decode_level = self.decode; + // sessions are spawned and never joined, so they need the signal directly: the server task + // returning only drops their command sender, which they would not notice mid-write + let shutdown = self.shutdown.clone(); let session = async move { - run_session( - socket, - addr, - connection_handler, - decode_level, - handler_map, - rx, - ) - .await; + tokio::select! { + biased; + _ = shutdown.cancelled() => {} + _ = run_session( + socket, + addr, + connection_handler, + decode_level, + handler_map, + rx, + ) => {} + } // no matter what happens, we send the id back to the server let _ = notify_close.send(SessionClose(id)).await; @@ -267,17 +285,27 @@ async fn run_session( #[cfg(test)] mod tests { use super::*; - use crate::error::Shutdown; use crate::server::create_tcp_server_task; use crate::UnitId; + use std::time::Duration; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + /// Bounds how long a test will hang if shutdown stops being immediate + const TEST_TIMEOUT: Duration = Duration::from_secs(5); struct DefaultHandler; impl RequestHandler for DefaultHandler {} - #[tokio::test] - async fn task_ends_when_shutdown_requested_with_the_handle_still_alive() { - // bound but never connected to: shutdown is a queued command, not something on the wire - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + fn spawn_server() -> ( + crate::server::ServerHandle, + tokio::task::JoinHandle<()>, + SocketAddr, + ) { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let addr = listener.local_addr().unwrap(); + let listener = TcpListener::from_std(listener).unwrap(); + let (handle, task) = create_tcp_server_task( 1, listener, @@ -285,12 +313,56 @@ mod tests { AddressFilter::Any, DecodeLevel::nothing(), ); - let task = tokio::spawn(task.run()); - handle.shutdown().await.unwrap(); - task.await.unwrap(); + (handle, tokio::spawn(task.run()), addr) + } + + #[tokio::test] + async fn task_ends_when_shutdown_requested_with_the_handle_still_alive() { + let (handle, task, _addr) = spawn_server(); - // the handle outlived the task it terminated, and now reports that it is gone - assert_eq!(handle.shutdown().await, Err(Shutdown)); + handle.shutdown(); + tokio::time::timeout(TEST_TIMEOUT, task) + .await + .unwrap() + .unwrap(); + + // the handle outlived the task it terminated + handle.shutdown(); + } + + #[tokio::test] + async fn shutdown_terminates_an_established_session() { + let (handle, task, addr) = spawn_server(); + + let mut client = tokio::net::TcpStream::connect(addr).await.unwrap(); + // exchanging a frame proves the session task is running before we shut it down; otherwise + // the connection could still be sitting in the accept queue and the EOF below would only + // show that the listener closed. The default handler answers with an exception, which is + // all we need here -- the reply's contents are irrelevant. + let read_coils = [0u8, 1, 0, 0, 0, 6, 1, 1, 0, 7, 0, 2]; + client.write_all(&read_coils).await.unwrap(); + let mut response = [0u8; 32]; + let replied = tokio::time::timeout(TEST_TIMEOUT, client.read(&mut response)) + .await + .expect("the server never answered, so the session was not established") + .unwrap(); + assert!(replied > 0); + + handle.shutdown(); + + // the session drops its socket, which the peer observes as EOF + let mut buffer = [0u8; 8]; + assert_eq!( + tokio::time::timeout(TEST_TIMEOUT, client.read(&mut buffer)) + .await + .expect("shutdown did not terminate the session") + .unwrap(), + 0 + ); + tokio::time::timeout(TEST_TIMEOUT, task) + .await + .unwrap() + .unwrap(); } } diff --git a/rodbus/src/tcp/tls/client.rs b/rodbus/src/tcp/tls/client.rs index ba12fbc6..96e7814b 100644 --- a/rodbus/src/tcp/tls/client.rs +++ b/rodbus/src/tcp/tls/client.rs @@ -44,6 +44,7 @@ pub(crate) fn create_tls_channel( listener: Box>, ) -> (Channel, ClientTask) { let (tx, rx) = tokio::sync::mpsc::channel(options.max_queued_requests); + let shutdown = tokio_util::sync::CancellationToken::new(); let task = TcpChannelTask::new( host, rx.into(), @@ -51,8 +52,9 @@ pub(crate) fn create_tls_channel( connect_retry, options, listener, + shutdown.clone(), ); - (Channel { tx }, ClientTask::tcp(task)) + (Channel { tx, shutdown }, ClientTask::tcp(task)) } impl TlsClientConfig { From e00841b3b581ad7e6980f61d06966946cdb6447d Mon Sep 17 00:00:00 2001 From: mdmzfzl Date: Tue, 1 Sep 2026 13:14:25 -0700 Subject: [PATCH 2/7] refactor: simplify immediate shutdown --- rodbus/src/channel.rs | 5 ++ rodbus/src/client/channel.rs | 31 +++++------ rodbus/src/client/task.rs | 6 ++- rodbus/src/common/cancellation.rs | 89 +++++++++++++++++++++++++++++++ rodbus/src/common/mod.rs | 1 + rodbus/src/serial/client.rs | 83 ++++++++++++++++++++++------ rodbus/src/serial/server.rs | 35 ++++++++---- rodbus/src/server/mod.rs | 42 +++++---------- rodbus/src/server/task.rs | 2 +- rodbus/src/tcp/client.rs | 87 +++++++++++++++++++++--------- rodbus/src/tcp/server.rs | 45 +++++++--------- rodbus/src/tcp/tls/client.rs | 3 +- 12 files changed, 306 insertions(+), 123 deletions(-) create mode 100644 rodbus/src/common/cancellation.rs diff --git a/rodbus/src/channel.rs b/rodbus/src/channel.rs index 806aa9cf..db6c4b3c 100644 --- a/rodbus/src/channel.rs +++ b/rodbus/src/channel.rs @@ -14,4 +14,9 @@ impl Receiver { pub(crate) async fn recv(&mut self) -> Result { 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() {} + } } diff --git a/rodbus/src/client/channel.rs b/rodbus/src/client/channel.rs index 2e087631..9cc5d6d0 100644 --- a/rodbus/src/client/channel.rs +++ b/rodbus/src/client/channel.rs @@ -5,27 +5,25 @@ 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::TaskCancellation; use crate::error::*; use crate::types::{AddressRange, BitIterator, Indexed, RegisterIterator, UnitId}; use crate::DecodeLevel; -use tokio_util::sync::CancellationToken; -/// Async channel used to make requests. -/// -/// The associated task terminates when every handle is dropped, or immediately when -/// [`Channel::shutdown`] is called. +/// Async channel used to make requests. The associated task is shutdown when every handle is +/// dropped or [`Channel::shutdown`] is called. #[derive(Debug, Clone)] pub struct Channel { pub(crate) tx: tokio::sync::mpsc::Sender, - pub(crate) shutdown: CancellationToken, + pub(crate) shutdown: TaskCancellation, } /// A client channel task that has been created but not yet spawned. /// /// This is returned, alongside its [`Channel`] handle, by the `create_*_client_task` functions. /// Drive it to completion by awaiting [`ClientTask::run`], typically from within -/// [`tokio::spawn`]. The task completes when every associated [`Channel`] handle is dropped, or -/// immediately when [`Channel::shutdown`] is called. +/// [`tokio::spawn`]. The task completes when every associated [`Channel`] handle is dropped or +/// [`Channel::shutdown`] is called. /// /// Unlike the `spawn_*_client_task` functions, no tracing span is attached to the task, so the /// caller is free to wrap [`run`](ClientTask::run) with their own instrumentation. @@ -53,7 +51,7 @@ impl ClientTask { } } - /// Run the channel task until every [`Channel`] handle is dropped, or [`Channel::shutdown`] is + /// Run the channel task until every [`Channel`] handle is dropped or [`Channel::shutdown`] is /// called. pub async fn run(self) { match self.inner { @@ -128,7 +126,6 @@ impl Channel { listener: Option>>, ) -> (Self, ClientTask) { let (tx, rx) = tokio::sync::mpsc::channel(max_queued_requests); - let shutdown = CancellationToken::new(); let task = crate::serial::client::SerialChannelTask::new( path, serial_settings, @@ -136,8 +133,8 @@ impl Channel { retry, decode, listener.unwrap_or_else(|| crate::client::NullListener::create()), - shutdown.clone(), ); + let shutdown = task.cancellation(); (Channel { tx, shutdown }, ClientTask::serial(task)) } @@ -153,14 +150,14 @@ impl Channel { Ok(()) } - /// Shut down the channel task immediately, even if other [`Channel`] handles are still alive + /// Begin shutting down the channel task, even if one or more [`Channel`] handles are still alive /// - /// Unlike dropping every handle, this does not wait for queued work. The task is cancelled at - /// its next suspension point, so a request already on the wire is abandoned rather than awaited - /// 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. + /// 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. /// - /// Calling this more than once, or after the task has already terminated, has no effect. + /// The task reports the terminal state to its listener before completing. Calling this more than + /// once, or after the task has already terminated, has no effect. pub fn shutdown(&self) { self.shutdown.cancel(); } diff --git a/rodbus/src/client/task.rs b/rodbus/src/client/task.rs index f8b3a47b..87a9e4de 100644 --- a/rodbus/src/client/task.rs +++ b/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) => { @@ -401,7 +405,7 @@ mod tests { }); let channel = Channel { tx, - shutdown: tokio_util::sync::CancellationToken::new(), + shutdown: crate::common::cancellation::TaskCancellation::default(), }; (channel, join_handle, io_handle) } diff --git a/rodbus/src/common/cancellation.rs b/rodbus/src/common/cancellation.rs new file mode 100644 index 00000000..a897e6c1 --- /dev/null +++ b/rodbus/src/common/cancellation.rs @@ -0,0 +1,89 @@ +use std::future::Future; + +use tokio_util::sync::CancellationToken; + +/// Cancellation signal shared by a public handle and its background task. +#[derive(Clone, Debug, Default)] +pub(crate) struct TaskCancellation { + token: CancellationToken, +} + +impl TaskCancellation { + pub(crate) fn cancel(&self) { + self.token.cancel(); + } + + /// Run a future until it completes or cancellation is requested. + /// + /// Cancellation takes priority. When it wins, the operation is dropped before this returns. + pub(crate) async fn run_until_cancelled(&self, operation: F) -> Option + where + F: Future, + { + tokio::select! { + biased; + _ = self.token.cancelled() => None, + result = operation => Some(result), + } + } +} + +#[cfg(test)] +mod tests { + use std::future::Future; + use std::pin::Pin; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + use std::task::{Context, Poll}; + + use super::*; + + struct PanicOnPoll; + + impl Future for PanicOnPoll { + type Output = (); + + fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll { + panic!("operation was polled after cancellation") + } + } + + struct DropFlag(Arc); + + impl Drop for DropFlag { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + + #[tokio::test] + async fn cancellation_wins_without_polling_the_operation() { + let cancellation = TaskCancellation::default(); + cancellation.cancel(); + + assert_eq!(cancellation.run_until_cancelled(PanicOnPoll).await, None); + } + + #[tokio::test] + async fn cancellation_drops_a_pending_operation() { + let cancellation = TaskCancellation::default(); + let handle = cancellation.clone(); + let dropped = Arc::new(AtomicBool::new(false)); + let flag = DropFlag(dropped.clone()); + + let task = tokio::spawn(async move { + cancellation + .run_until_cancelled(async move { + let _flag = flag; + std::future::pending::<()>().await; + }) + .await + }); + + tokio::task::yield_now().await; + handle.cancel(); + + assert_eq!(task.await.unwrap(), None); + assert!(dropped.load(Ordering::SeqCst)); + } +} diff --git a/rodbus/src/common/mod.rs b/rodbus/src/common/mod.rs index 145bf918..e6301e63 100644 --- a/rodbus/src/common/mod.rs +++ b/rodbus/src/common/mod.rs @@ -3,6 +3,7 @@ pub(crate) mod traits; pub(crate) mod bits; pub(crate) mod buffer; +pub(crate) mod cancellation; pub(crate) mod frame; mod parse; pub(crate) mod phys; diff --git a/rodbus/src/serial/client.rs b/rodbus/src/serial/client.rs index d705c9e6..84b45feb 100644 --- a/rodbus/src/serial/client.rs +++ b/rodbus/src/serial/client.rs @@ -5,9 +5,9 @@ use crate::serial::SerialSettings; use crate::client::message::Command; use crate::client::task::{ClientLoop, SessionError, StateChange}; use crate::client::{Listener, PortState, RetryStrategy}; +use crate::common::cancellation::TaskCancellation; use crate::common::frame::{FrameWriter, FramedReader}; use crate::error::Shutdown; -use tokio_util::sync::CancellationToken; pub(crate) struct SerialChannelTask { path: String, @@ -15,11 +15,10 @@ pub(crate) struct SerialChannelTask { retry: Box, client_loop: ClientLoop, listener: Box>, - shutdown: CancellationToken, + shutdown: TaskCancellation, } impl SerialChannelTask { - #[allow(clippy::too_many_arguments)] pub(crate) fn new( path: &str, serial_settings: SerialSettings, @@ -27,7 +26,6 @@ impl SerialChannelTask { retry: Box, decode: DecodeLevel, listener: Box>, - shutdown: CancellationToken, ) -> Self { Self { path: path.to_string(), @@ -41,26 +39,30 @@ impl SerialChannelTask { None, ), listener, - shutdown, + shutdown: TaskCancellation::default(), } } + pub(crate) fn cancellation(&self) -> TaskCancellation { + self.shutdown.clone() + } + pub(crate) async fn run(&mut self) -> Shutdown { - self.listener.update(PortState::Disabled).get().await; - // the clone releases the borrow of self that run_inner() needs mutably let shutdown = self.shutdown.clone(); - tokio::select! { - // biased so that a cancelled task cannot be given another poll of run_inner(), which - // would let it finish work that is already ready before noticing the shutdown - biased; - _ = shutdown.cancelled() => {} - _ = self.run_inner() => {} - } - // outside the select so that a cancelled task still reports its terminal state + shutdown + .run_until_cancelled(self.run_until_shutdown()) + .await; + + self.client_loop.shutdown().await; self.listener.update(PortState::Shutdown).get().await; Shutdown } + async fn run_until_shutdown(&mut self) -> Shutdown { + self.listener.update(PortState::Disabled).get().await; + self.run_inner().await + } + async fn run_inner(&mut self) -> Shutdown { loop { // wait for the channel to be enabled @@ -112,3 +114,54 @@ impl SerialChannelTask { } } } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use crate::maybe_async::MaybeAsync; + use crate::retry::default_retry_strategy; + + use super::*; + + const TEST_TIMEOUT: Duration = Duration::from_secs(5); + + struct BlockingDisabledListener { + states: tokio::sync::mpsc::UnboundedSender, + } + + impl Listener for BlockingDisabledListener { + fn update(&mut self, state: PortState) -> MaybeAsync<()> { + self.states.send(state).unwrap(); + match state { + PortState::Disabled => MaybeAsync::asynchronous(std::future::pending()), + _ => MaybeAsync::ready(()), + } + } + } + + #[tokio::test] + async fn shutdown_interrupts_a_pending_listener_notification() { + let (_tx, rx) = tokio::sync::mpsc::channel(1); + let (states, mut state_rx) = tokio::sync::mpsc::unbounded_channel(); + let mut task = SerialChannelTask::new( + "unused", + SerialSettings::default(), + rx.into(), + default_retry_strategy(), + DecodeLevel::nothing(), + Box::new(BlockingDisabledListener { states }), + ); + let cancellation = task.cancellation(); + let task = tokio::spawn(async move { task.run().await }); + + assert_eq!(state_rx.recv().await, Some(PortState::Disabled)); + cancellation.cancel(); + + tokio::time::timeout(TEST_TIMEOUT, task) + .await + .expect("shutdown did not interrupt the listener") + .unwrap(); + assert_eq!(state_rx.recv().await, Some(PortState::Shutdown)); + } +} diff --git a/rodbus/src/serial/server.rs b/rodbus/src/serial/server.rs index 6e5b2833..38c0fa3f 100644 --- a/rodbus/src/serial/server.rs +++ b/rodbus/src/serial/server.rs @@ -1,8 +1,8 @@ +use crate::common::cancellation::TaskCancellation; use crate::common::phys::PhysLayer; use crate::server::task::SessionTask; use crate::server::RequestHandler; use crate::{RequestError, RetryStrategy, SerialSettings, Shutdown}; -use tokio_util::sync::CancellationToken; pub(crate) struct RtuServerTask where @@ -12,23 +12,38 @@ where pub(crate) retry: Box, pub(crate) settings: SerialSettings, pub(crate) session: SessionTask, - pub(crate) shutdown: CancellationToken, + shutdown: TaskCancellation, } impl RtuServerTask where T: RequestHandler, { + pub(crate) fn new( + port: String, + retry: Box, + settings: SerialSettings, + session: SessionTask, + ) -> Self { + Self { + port, + retry, + settings, + session, + shutdown: TaskCancellation::default(), + } + } + + pub(crate) fn cancellation(&self) -> TaskCancellation { + self.shutdown.clone() + } + pub(crate) async fn run(&mut self) -> Shutdown { - // the clone releases the borrow of self that run_inner() needs mutably let shutdown = self.shutdown.clone(); - tokio::select! { - // biased so that a cancelled task cannot be given another poll of run_inner(), which - // would let it finish work that is already ready before noticing the shutdown - biased; - _ = shutdown.cancelled() => Shutdown, - res = self.run_inner() => res, - } + shutdown + .run_until_cancelled(self.run_inner()) + .await + .unwrap_or(Shutdown) } async fn run_inner(&mut self) -> Shutdown { diff --git a/rodbus/src/server/mod.rs b/rodbus/src/server/mod.rs index 0046d9bc..0ca33f47 100644 --- a/rodbus/src/server/mod.rs +++ b/rodbus/src/server/mod.rs @@ -2,10 +2,10 @@ use std::net::SocketAddr; use tracing::Instrument; +use crate::common::cancellation::TaskCancellation; use crate::decode::DecodeLevel; use crate::server::task::ServerCommand; use crate::tcp::server::{ServerTask as TcpServerTask, TcpServerConnectionHandler}; -use tokio_util::sync::CancellationToken; /// server handling mod address_filter; @@ -30,22 +30,20 @@ pub use crate::tcp::tls::server::TlsServerConfig; #[cfg(feature = "enable-tls")] pub use crate::tcp::tls::*; -/// Handle to the server async task. -/// -/// The associated task terminates when the handle is dropped, or immediately when +/// Handle to the server async task. The associated task is shutdown when the handle is dropped or /// [`ServerHandle::shutdown`] is called. #[derive(Debug)] pub struct ServerHandle { tx: tokio::sync::mpsc::Sender, - shutdown: CancellationToken, + shutdown: TaskCancellation, } /// A server task that has been created but not yet spawned. /// /// This is returned, alongside its [`ServerHandle`], by the `create_*_server_task` functions. /// Drive it to completion by awaiting [`ServerTask::run`], typically from within -/// [`tokio::spawn`]. The task completes when the associated [`ServerHandle`] is dropped, or -/// immediately when [`ServerHandle::shutdown`] is called. +/// [`tokio::spawn`]. The task completes when the associated [`ServerHandle`] is dropped or +/// [`ServerHandle::shutdown`] is called. /// /// Unlike the `spawn_*_server_task` functions, no tracing span is attached to the task, so the /// caller is free to wrap [`run`](ServerTask::run) with their own instrumentation. @@ -76,7 +74,7 @@ impl ServerTask { } } - /// Run the server task until the associated [`ServerHandle`] is dropped, or + /// Run the server task until the associated [`ServerHandle`] is dropped or /// [`ServerHandle::shutdown`] is called. pub async fn run(self) { match self.inner { @@ -90,10 +88,7 @@ impl ServerTask { } impl ServerHandle { - /// Construct a [ServerHandle] from its fields - /// - /// This function is only required for the C bindings - pub fn new(tx: tokio::sync::mpsc::Sender, shutdown: CancellationToken) -> Self { + fn new(tx: tokio::sync::mpsc::Sender, shutdown: TaskCancellation) -> Self { ServerHandle { tx, shutdown } } @@ -103,11 +98,10 @@ impl ServerHandle { Ok(()) } - /// Shut down the server task and every active session immediately + /// Begin shutting down the server task, even while the [`ServerHandle`] is still alive /// - /// Unlike dropping the handle, this does not wait for sessions to reach a quiescent point. The - /// listener stops accepting and each session is cancelled at its next suspension point, so a - /// response that is partially written is abandoned. + /// The listener stops accepting connections, and each active session is cancelled at its next + /// suspension point. A response that is partially written is abandoned. /// /// Calling this more than once, or after the task has already terminated, has no effect. pub fn shutdown(&self) { @@ -163,7 +157,6 @@ pub fn create_tcp_server_task( decode: DecodeLevel, ) -> (ServerHandle, ServerTask) { let (tx, rx) = tokio::sync::mpsc::channel(SERVER_COMMAND_CHANNEL_CAPACITY); - let shutdown = CancellationToken::new(); let task = TcpServerTask::new( max_sessions, listener, @@ -171,8 +164,8 @@ pub fn create_tcp_server_task( TcpServerConnectionHandler::Tcp, filter, decode, - shutdown.clone(), ); + let shutdown = task.cancellation(); (ServerHandle::new(tx, shutdown), ServerTask::tcp(task, rx)) } @@ -224,7 +217,6 @@ pub fn create_rtu_server_task( decode: DecodeLevel, ) -> (ServerHandle, ServerTask) { let (tx, rx) = tokio::sync::mpsc::channel(SERVER_COMMAND_CHANNEL_CAPACITY); - let shutdown = CancellationToken::new(); let session = crate::server::task::SessionTask::new( handlers, crate::server::task::AuthorizationType::None, @@ -234,13 +226,8 @@ pub fn create_rtu_server_task( decode, ); - let rtu = crate::serial::server::RtuServerTask { - port: path.to_string(), - retry, - settings, - session, - shutdown: shutdown.clone(), - }; + let rtu = crate::serial::server::RtuServerTask::new(path.to_string(), retry, settings, session); + let shutdown = rtu.cancellation(); (ServerHandle::new(tx, shutdown), ServerTask::rtu(rtu)) } @@ -424,7 +411,6 @@ fn create_tls_server_task_impl( decode: DecodeLevel, ) -> (ServerHandle, ServerTask) { let (tx, rx) = tokio::sync::mpsc::channel(SERVER_COMMAND_CHANNEL_CAPACITY); - let shutdown = CancellationToken::new(); let task = TcpServerTask::new( max_sessions, listener, @@ -432,8 +418,8 @@ fn create_tls_server_task_impl( TcpServerConnectionHandler::Tls(tls_config, auth_handler), filter, decode, - shutdown.clone(), ); + let shutdown = task.cancellation(); (ServerHandle::new(tx, shutdown), ServerTask::tcp(task, rx)) } diff --git a/rodbus/src/server/task.rs b/rodbus/src/server/task.rs index d5c1c39b..513a5986 100644 --- a/rodbus/src/server/task.rs +++ b/rodbus/src/server/task.rs @@ -16,7 +16,7 @@ use std::sync::Arc; /// Commands that can be sent to a running server task #[derive(Copy, Clone)] -pub enum ServerCommand { +pub(crate) enum ServerCommand { /// Change the decoding level dynamically ChangeDecoding(DecodeLevel), } diff --git a/rodbus/src/tcp/client.rs b/rodbus/src/tcp/client.rs index 59bfcbc6..1ab96bbb 100644 --- a/rodbus/src/tcp/client.rs +++ b/rodbus/src/tcp/client.rs @@ -1,6 +1,7 @@ use tracing::Instrument; use crate::client::{Channel, ClientState, ClientTask, HostAddr, Listener}; +use crate::common::cancellation::TaskCancellation; use crate::common::phys::PhysLayer; use crate::client::message::Command; @@ -11,7 +12,6 @@ use crate::retry::RetryStrategy; use crate::{ChannelLoggingMode, ClientOptions}; use tokio::net::TcpStream; -use tokio_util::sync::CancellationToken; macro_rules! log_channel_event { ($channel_logging:expr, $($arg:tt)*) => { @@ -45,7 +45,6 @@ pub(crate) fn create_tcp_channel( options: ClientOptions, ) -> (Channel, ClientTask) { let (tx, rx) = tokio::sync::mpsc::channel(options.max_queued_requests); - let shutdown = CancellationToken::new(); let task = TcpChannelTask::new( host, rx.into(), @@ -53,8 +52,8 @@ pub(crate) fn create_tcp_channel( connect_retry, options, listener, - shutdown.clone(), ); + let shutdown = task.cancellation(); (Channel { tx, shutdown }, ClientTask::tcp(task)) } @@ -85,7 +84,7 @@ pub(crate) struct TcpChannelTask { client_loop: ClientLoop, listener: Box>, channel_logging: ChannelLoggingMode, - shutdown: CancellationToken, + shutdown: TaskCancellation, } impl TcpChannelTask { @@ -96,7 +95,6 @@ impl TcpChannelTask { connect_retry: Box, options: ClientOptions, listener: Box>, - shutdown: CancellationToken, ) -> Self { Self { host, @@ -111,27 +109,31 @@ impl TcpChannelTask { ), listener, channel_logging: options.channel_logging, - shutdown, + shutdown: TaskCancellation::default(), } } + pub(crate) fn cancellation(&self) -> TaskCancellation { + self.shutdown.clone() + } + // runs until it is shut down pub(crate) async fn run(&mut self) -> Shutdown { - self.listener.update(ClientState::Disabled).get().await; - // the clone releases the borrow of self that run_inner() needs mutably let shutdown = self.shutdown.clone(); - tokio::select! { - // biased so that a cancelled task cannot be given another poll of run_inner(), which - // would let it finish work that is already ready before noticing the shutdown - biased; - _ = shutdown.cancelled() => {} - _ = self.run_inner() => {} - } - // outside the select so that a cancelled task still reports its terminal state + shutdown + .run_until_cancelled(self.run_until_shutdown()) + .await; + + self.client_loop.shutdown().await; self.listener.update(ClientState::Shutdown).get().await; Shutdown } + async fn run_until_shutdown(&mut self) -> Shutdown { + self.listener.update(ClientState::Disabled).get().await; + self.run_inner().await + } + async fn run_inner(&mut self) -> Shutdown { loop { if let Err(Shutdown) = self.client_loop.wait_for_enabled().await { @@ -250,6 +252,26 @@ mod tests { } } + struct BlockingShutdownListener { + tx: tokio::sync::mpsc::UnboundedSender, + release: Option>, + } + + impl Listener for BlockingShutdownListener { + fn update(&mut self, value: ClientState) -> MaybeAsync<()> { + let _ = self.tx.send(value); + match value { + ClientState::Shutdown => { + let release = self.release.take().unwrap(); + MaybeAsync::asynchronous(async move { + let _ = release.await; + }) + } + _ => MaybeAsync::ready(()), + } + } + } + struct NeverRetry; impl RetryStrategy for NeverRetry { @@ -276,6 +298,7 @@ mod tests { ); let task = tokio::spawn(task.run()); + assert_eq!(states.recv().await, Some(ClientState::Disabled)); channel.shutdown(); tokio::time::timeout(TEST_TIMEOUT, task) .await @@ -283,7 +306,6 @@ mod tests { .unwrap(); // the task announced its own termination on the way out - assert_eq!(states.recv().await, Some(ClientState::Disabled)); assert_eq!(states.recv().await, Some(ClientState::Shutdown)); } @@ -312,9 +334,11 @@ mod tests { ) .await }); - let mut buffer = [0u8; 32]; - let read = socket.read(&mut buffer).await.unwrap(); - assert!(read > 0, "the request never reached the wire"); + let mut request = [0u8; 12]; + socket + .read_exact(&mut request) + .await + .expect("the request never reached the wire"); // the loop is now parked awaiting a response that will never arrive channel.shutdown(); @@ -332,7 +356,7 @@ mod tests { .unwrap(); // the peer sees the connection go away rather than a lingering half-open socket - assert_eq!(socket.read(&mut buffer).await.unwrap(), 0); + assert_eq!(socket.read(&mut request).await.unwrap(), 0); } #[tokio::test] @@ -373,14 +397,19 @@ mod tests { } #[tokio::test] - async fn queued_requests_fail_when_shutdown_is_requested() { + async fn queued_requests_fail_before_shutdown_listener_completes() { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); + let (tx, mut states) = tokio::sync::mpsc::unbounded_channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); let (channel, task) = create_tcp_channel( HostAddr::ip(addr.ip(), addr.port()), default_retry_strategy(), - crate::client::NullListener::create(), + Box::new(BlockingShutdownListener { + tx, + release: Some(release_rx), + }), ClientOptions::default(), ); let task = tokio::spawn(task.run()); @@ -407,8 +436,15 @@ mod tests { assert!(socket.read(&mut buffer).await.unwrap() > 0); channel.shutdown(); + loop { + match states.recv().await { + Some(ClientState::Shutdown) => break, + Some(_) => {} + None => panic!("client task ended without reporting shutdown"), + } + } - // every request fails, in flight or queued, without waiting out a single timeout + // requests fail before the terminal listener is allowed to complete for handle in queued { assert_eq!( tokio::time::timeout(TEST_TIMEOUT, handle) @@ -418,6 +454,9 @@ mod tests { Err(RequestError::Shutdown) ); } + assert!(!task.is_finished()); + release_tx.send(()).unwrap(); + tokio::time::timeout(TEST_TIMEOUT, task) .await .unwrap() diff --git a/rodbus/src/tcp/server.rs b/rodbus/src/tcp/server.rs index bb613bab..7f2fa224 100644 --- a/rodbus/src/tcp/server.rs +++ b/rodbus/src/tcp/server.rs @@ -2,6 +2,7 @@ use std::collections::BTreeMap; use tracing::Instrument; +use crate::common::cancellation::TaskCancellation; use crate::common::frame::{FrameWriter, FramedReader}; use crate::common::phys::PhysLayer; use crate::decode::DecodeLevel; @@ -11,7 +12,6 @@ use crate::server::task::{AuthorizationType, ServerCommand}; use crate::server::AddressFilter; use std::net::SocketAddr; use tokio::net::TcpListener; -use tokio_util::sync::CancellationToken; #[cfg(feature = "enable-tls")] use crate::server::AuthorizationHandler; @@ -107,14 +107,13 @@ pub(crate) struct ServerTask { decode: DecodeLevel, tx: tokio::sync::mpsc::Sender, rx: tokio::sync::mpsc::Receiver, - shutdown: CancellationToken, + shutdown: TaskCancellation, } impl ServerTask where T: RequestHandler, { - #[allow(clippy::too_many_arguments)] pub(crate) fn new( max_sessions: usize, listener: TcpListener, @@ -122,7 +121,6 @@ where connection_handler: TcpServerConnectionHandler, filter: AddressFilter, decode: DecodeLevel, - shutdown: CancellationToken, ) -> Self { let (tx, rx) = tokio::sync::mpsc::channel(8); @@ -135,10 +133,14 @@ where decode, tx, rx, - shutdown, + shutdown: TaskCancellation::default(), } } + pub(crate) fn cancellation(&self) -> TaskCancellation { + self.shutdown.clone() + } + async fn apply_command(&mut self, command: ServerCommand) { // first, change it locally so that it is applied to new sessions match command { @@ -156,16 +158,13 @@ where } pub(crate) async fn run(&mut self, mut commands: tokio::sync::mpsc::Receiver) { - // the clone releases the borrow of self that run_inner() needs mutably let shutdown = self.shutdown.clone(); - tokio::select! { - // biased so that a cancelled task cannot be given another poll of run_inner(), which - // would let it accept another connection before noticing the shutdown - biased; - _ = shutdown.cancelled() => { - tracing::info!("server shutdown requested"); - } - _ = self.run_inner(&mut commands) => {} + if shutdown + .run_until_cancelled(self.run_inner(&mut commands)) + .await + .is_none() + { + tracing::info!("server shutdown requested"); } } @@ -223,23 +222,20 @@ where let connection_handler = self.connection_handler.clone(); let handler_map = self.handlers.clone(); let decode_level = self.decode; - // sessions are spawned and never joined, so they need the signal directly: the server task - // returning only drops their command sender, which they would not notice mid-write + // sessions are spawned, so they need to observe cancellation directly let shutdown = self.shutdown.clone(); let session = async move { - tokio::select! { - biased; - _ = shutdown.cancelled() => {} - _ = run_session( + shutdown + .run_until_cancelled(run_session( socket, addr, connection_handler, decode_level, handler_map, rx, - ) => {} - } + )) + .await; // no matter what happens, we send the id back to the server let _ = notify_close.send(SessionClose(id)).await; @@ -342,12 +338,11 @@ mod tests { // all we need here -- the reply's contents are irrelevant. let read_coils = [0u8, 1, 0, 0, 0, 6, 1, 1, 0, 7, 0, 2]; client.write_all(&read_coils).await.unwrap(); - let mut response = [0u8; 32]; - let replied = tokio::time::timeout(TEST_TIMEOUT, client.read(&mut response)) + let mut response = [0u8; 9]; + tokio::time::timeout(TEST_TIMEOUT, client.read_exact(&mut response)) .await .expect("the server never answered, so the session was not established") .unwrap(); - assert!(replied > 0); handle.shutdown(); diff --git a/rodbus/src/tcp/tls/client.rs b/rodbus/src/tcp/tls/client.rs index 96e7814b..daa4d29c 100644 --- a/rodbus/src/tcp/tls/client.rs +++ b/rodbus/src/tcp/tls/client.rs @@ -44,7 +44,6 @@ pub(crate) fn create_tls_channel( listener: Box>, ) -> (Channel, ClientTask) { let (tx, rx) = tokio::sync::mpsc::channel(options.max_queued_requests); - let shutdown = tokio_util::sync::CancellationToken::new(); let task = TcpChannelTask::new( host, rx.into(), @@ -52,8 +51,8 @@ pub(crate) fn create_tls_channel( connect_retry, options, listener, - shutdown.clone(), ); + let shutdown = task.cancellation(); (Channel { tx, shutdown }, ClientTask::tcp(task)) } From 8643a6c7ec92d204dec2d82bb921efd73d44db63 Mon Sep 17 00:00:00 2001 From: mdmzfzl Date: Tue, 1 Sep 2026 16:21:37 -0700 Subject: [PATCH 3/7] docs: note that listeners must complete, add RTU cancellation test --- rodbus/src/client/channel.rs | 5 ++- rodbus/src/client/listener.rs | 5 +++ rodbus/src/common/cancellation.rs | 4 ++ rodbus/src/serial/server.rs | 65 +++++++++++++++++++++++++++++++ 4 files changed, 77 insertions(+), 2 deletions(-) diff --git a/rodbus/src/client/channel.rs b/rodbus/src/client/channel.rs index 9cc5d6d0..477cefa4 100644 --- a/rodbus/src/client/channel.rs +++ b/rodbus/src/client/channel.rs @@ -156,8 +156,9 @@ impl Channel { /// 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. Calling this more than - /// once, or after the task has already terminated, has no effect. + /// 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(); } diff --git a/rodbus/src/client/listener.rs b/rodbus/src/client/listener.rs index d3f5b7b3..02f2d1be 100644 --- a/rodbus/src/client/listener.rs +++ b/rodbus/src/client/listener.rs @@ -3,6 +3,11 @@ use crate::MaybeAsync; /// Generic listener type that can be invoked multiple times pub trait Listener: 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(()) } diff --git a/rodbus/src/common/cancellation.rs b/rodbus/src/common/cancellation.rs index a897e6c1..40825347 100644 --- a/rodbus/src/common/cancellation.rs +++ b/rodbus/src/common/cancellation.rs @@ -16,6 +16,10 @@ impl TaskCancellation { /// Run a future until it completes or cancellation is requested. /// /// Cancellation takes priority. When it wins, the operation is dropped before this returns. + /// + /// `CancellationToken` has a method of the same name, deliberately not used here: it polls the + /// operation first, so a tie goes to the operation rather than to cancellation, and it was only + /// added in tokio-util 0.7.12 whereas this crate depends on `0.7`. pub(crate) async fn run_until_cancelled(&self, operation: F) -> Option where F: Future, diff --git a/rodbus/src/serial/server.rs b/rodbus/src/serial/server.rs index 38c0fa3f..75ecc426 100644 --- a/rodbus/src/serial/server.rs +++ b/rodbus/src/serial/server.rs @@ -79,3 +79,68 @@ where } } } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use crate::server::{create_rtu_server_task, ServerHandlerMap}; + use crate::{DecodeLevel, UnitId}; + + use super::*; + + /// Bounds how long this test hangs if shutdown stops being immediate + const TEST_TIMEOUT: Duration = Duration::from_secs(5); + + /// A retry delay long enough that the task completing proves cancellation rather than the + /// delay simply elapsing + const NEVER: Duration = Duration::from_secs(3600); + + struct DefaultHandler; + impl RequestHandler for DefaultHandler {} + + /// Reports when the task gives up on opening the port, which is the point just before it parks + /// on the retry delay + struct SignalingRetry { + failed_connect: tokio::sync::mpsc::UnboundedSender<()>, + } + + impl RetryStrategy for SignalingRetry { + fn reset(&mut self) {} + + fn after_failed_connect(&mut self) -> Duration { + let _ = self.failed_connect.send(()); + NEVER + } + + fn after_disconnect(&mut self) -> Duration { + NEVER + } + } + + #[tokio::test] + async fn shutdown_interrupts_the_wait_before_reopening_the_port() { + let (failed_connect, mut failures) = tokio::sync::mpsc::unbounded_channel(); + + // a path that cannot be opened, so the task fails and falls into its retry delay + let (handle, task) = create_rtu_server_task( + "/dev/rodbus-does-not-exist", + SerialSettings::default(), + Box::new(SignalingRetry { failed_connect }), + ServerHandlerMap::single(UnitId::new(1), DefaultHandler.wrap()), + DecodeLevel::nothing(), + ); + let task = tokio::spawn(task.run()); + + // wait until opening the port has actually failed, otherwise the task might be cancelled + // before it ever reaches the delay and the test would prove nothing + failures.recv().await.unwrap(); + + handle.shutdown(); + + tokio::time::timeout(TEST_TIMEOUT, task) + .await + .expect("shutdown did not interrupt the retry delay") + .unwrap(); + } +} From cbfad685c4219cdbc7a99edcca946635c6d39679 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Thu, 3 Sep 2026 09:46:23 -0700 Subject: [PATCH 4/7] refactor(client): replace TaskCancellation with a ShutdownHandle/ShutdownSignal pair Create the cancellation pair next to the mpsc pair in the create_*_channel functions and thread the signal through ClientTask, so the client task structs no longer own a token or expose a getter for it. The handle can only cancel; the signal can only observe. The server still uses TaskCancellation until it is migrated. --- rodbus/src/client/channel.rs | 22 ++++++---- rodbus/src/client/task.rs | 2 +- rodbus/src/common/cancellation.rs | 68 +++++++++++++++++++++++++------ rodbus/src/serial/client.rs | 15 ++----- rodbus/src/tcp/client.rs | 15 ++----- rodbus/src/tcp/tls/client.rs | 4 +- 6 files changed, 80 insertions(+), 46 deletions(-) diff --git a/rodbus/src/client/channel.rs b/rodbus/src/client/channel.rs index 477cefa4..0b4193bc 100644 --- a/rodbus/src/client/channel.rs +++ b/rodbus/src/client/channel.rs @@ -5,7 +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::TaskCancellation; +use crate::common::cancellation::{ShutdownHandle, ShutdownSignal}; use crate::error::*; use crate::types::{AddressRange, BitIterator, Indexed, RegisterIterator, UnitId}; use crate::DecodeLevel; @@ -15,7 +15,7 @@ use crate::DecodeLevel; #[derive(Debug, Clone)] pub struct Channel { pub(crate) tx: tokio::sync::mpsc::Sender, - pub(crate) shutdown: TaskCancellation, + pub(crate) shutdown: ShutdownHandle, } /// A client channel task that has been created but not yet spawned. @@ -29,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 { @@ -38,16 +39,21 @@ 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, } } @@ -56,11 +62,11 @@ impl ClientTask { pub async fn run(self) { match self.inner { ClientTaskInner::Tcp(mut task) => { - task.run().await; + task.run(self.shutdown).await; } #[cfg(feature = "serial")] ClientTaskInner::Serial(mut task) => { - task.run().await; + task.run(self.shutdown).await; } } } @@ -126,6 +132,7 @@ impl Channel { listener: Option>>, ) -> (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, @@ -134,8 +141,7 @@ impl Channel { decode, listener.unwrap_or_else(|| crate::client::NullListener::create()), ); - let shutdown = task.cancellation(); - (Channel { tx, shutdown }, ClientTask::serial(task)) + (Channel { tx, shutdown }, ClientTask::serial(task, signal)) } /// Enable communications diff --git a/rodbus/src/client/task.rs b/rodbus/src/client/task.rs index 87a9e4de..1fd3c9c3 100644 --- a/rodbus/src/client/task.rs +++ b/rodbus/src/client/task.rs @@ -405,7 +405,7 @@ mod tests { }); let channel = Channel { tx, - shutdown: crate::common::cancellation::TaskCancellation::default(), + shutdown: crate::common::cancellation::pair().0, }; (channel, join_handle, io_handle) } diff --git a/rodbus/src/common/cancellation.rs b/rodbus/src/common/cancellation.rs index 40825347..b27ae2a0 100644 --- a/rodbus/src/common/cancellation.rs +++ b/rodbus/src/common/cancellation.rs @@ -2,24 +2,67 @@ use std::future::Future; use tokio_util::sync::CancellationToken; -/// Cancellation signal shared by a public handle and its background task. -#[derive(Clone, Debug, Default)] -pub(crate) struct TaskCancellation { - token: CancellationToken, +/// Create a linked pair of shutdown handle and signal, in the spirit of an mpsc channel. +/// +/// The [`ShutdownHandle`] goes to the public handle and can only request shutdown. The +/// [`ShutdownSignal`] goes to the background task and can only observe that request. Neither side +/// can do the other's job, so the direction of control is fixed by the types. +pub(crate) fn pair() -> (ShutdownHandle, ShutdownSignal) { + let token = CancellationToken::new(); + (ShutdownHandle(token.clone()), ShutdownSignal(token)) } -impl TaskCancellation { +/// The requesting half of a shutdown [`pair`]. Cloneable so that public handles can be cloned. +#[derive(Clone, Debug)] +pub(crate) struct ShutdownHandle(CancellationToken); + +impl ShutdownHandle { + /// Request shutdown. Idempotent. pub(crate) fn cancel(&self) { - self.token.cancel(); + self.0.cancel(); } +} - /// Run a future until it completes or cancellation is requested. +/// The observing half of a shutdown [`pair`]. +/// +/// Cloneable so that a task can fan the signal out to sub-tasks it spawns, but a clone can still +/// only observe cancellation, never request it. +#[derive(Clone, Debug)] +pub(crate) struct ShutdownSignal(CancellationToken); + +impl ShutdownSignal { + /// Run a future until it completes or shutdown is requested. /// /// Cancellation takes priority. When it wins, the operation is dropped before this returns. /// /// `CancellationToken` has a method of the same name, deliberately not used here: it polls the /// operation first, so a tie goes to the operation rather than to cancellation, and it was only /// added in tokio-util 0.7.12 whereas this crate depends on `0.7`. + pub(crate) async fn run_until_cancelled(&self, operation: F) -> Option + where + F: Future, + { + tokio::select! { + biased; + _ = self.0.cancelled() => None, + result = operation => Some(result), + } + } +} + +/// Cancellation signal shared by a public handle and its background task. +/// +/// Superseded by [`pair`]; still used by the server side until it is migrated. +#[derive(Clone, Debug, Default)] +pub(crate) struct TaskCancellation { + token: CancellationToken, +} + +impl TaskCancellation { + pub(crate) fn cancel(&self) { + self.token.cancel(); + } + pub(crate) async fn run_until_cancelled(&self, operation: F) -> Option where F: Future, @@ -62,21 +105,20 @@ mod tests { #[tokio::test] async fn cancellation_wins_without_polling_the_operation() { - let cancellation = TaskCancellation::default(); - cancellation.cancel(); + let (handle, signal) = pair(); + handle.cancel(); - assert_eq!(cancellation.run_until_cancelled(PanicOnPoll).await, None); + assert_eq!(signal.run_until_cancelled(PanicOnPoll).await, None); } #[tokio::test] async fn cancellation_drops_a_pending_operation() { - let cancellation = TaskCancellation::default(); - let handle = cancellation.clone(); + let (handle, signal) = pair(); let dropped = Arc::new(AtomicBool::new(false)); let flag = DropFlag(dropped.clone()); let task = tokio::spawn(async move { - cancellation + signal .run_until_cancelled(async move { let _flag = flag; std::future::pending::<()>().await; diff --git a/rodbus/src/serial/client.rs b/rodbus/src/serial/client.rs index 84b45feb..be10a869 100644 --- a/rodbus/src/serial/client.rs +++ b/rodbus/src/serial/client.rs @@ -5,7 +5,7 @@ use crate::serial::SerialSettings; use crate::client::message::Command; use crate::client::task::{ClientLoop, SessionError, StateChange}; use crate::client::{Listener, PortState, RetryStrategy}; -use crate::common::cancellation::TaskCancellation; +use crate::common::cancellation::ShutdownSignal; use crate::common::frame::{FrameWriter, FramedReader}; use crate::error::Shutdown; @@ -15,7 +15,6 @@ pub(crate) struct SerialChannelTask { retry: Box, client_loop: ClientLoop, listener: Box>, - shutdown: TaskCancellation, } impl SerialChannelTask { @@ -39,16 +38,10 @@ impl SerialChannelTask { None, ), listener, - shutdown: TaskCancellation::default(), } } - pub(crate) fn cancellation(&self) -> TaskCancellation { - self.shutdown.clone() - } - - pub(crate) async fn run(&mut self) -> Shutdown { - let shutdown = self.shutdown.clone(); + pub(crate) async fn run(&mut self, shutdown: ShutdownSignal) -> Shutdown { shutdown .run_until_cancelled(self.run_until_shutdown()) .await; @@ -152,8 +145,8 @@ mod tests { DecodeLevel::nothing(), Box::new(BlockingDisabledListener { states }), ); - let cancellation = task.cancellation(); - let task = tokio::spawn(async move { task.run().await }); + let (cancellation, signal) = crate::common::cancellation::pair(); + let task = tokio::spawn(async move { task.run(signal).await }); assert_eq!(state_rx.recv().await, Some(PortState::Disabled)); cancellation.cancel(); diff --git a/rodbus/src/tcp/client.rs b/rodbus/src/tcp/client.rs index 1ab96bbb..a9da7624 100644 --- a/rodbus/src/tcp/client.rs +++ b/rodbus/src/tcp/client.rs @@ -1,7 +1,7 @@ use tracing::Instrument; use crate::client::{Channel, ClientState, ClientTask, HostAddr, Listener}; -use crate::common::cancellation::TaskCancellation; +use crate::common::cancellation::ShutdownSignal; use crate::common::phys::PhysLayer; use crate::client::message::Command; @@ -45,6 +45,7 @@ pub(crate) fn create_tcp_channel( options: ClientOptions, ) -> (Channel, ClientTask) { let (tx, rx) = tokio::sync::mpsc::channel(options.max_queued_requests); + let (shutdown, signal) = crate::common::cancellation::pair(); let task = TcpChannelTask::new( host, rx.into(), @@ -53,8 +54,7 @@ pub(crate) fn create_tcp_channel( options, listener, ); - let shutdown = task.cancellation(); - (Channel { tx, shutdown }, ClientTask::tcp(task)) + (Channel { tx, shutdown }, ClientTask::tcp(task, signal)) } pub(crate) enum TcpTaskConnectionHandler { @@ -84,7 +84,6 @@ pub(crate) struct TcpChannelTask { client_loop: ClientLoop, listener: Box>, channel_logging: ChannelLoggingMode, - shutdown: TaskCancellation, } impl TcpChannelTask { @@ -109,17 +108,11 @@ impl TcpChannelTask { ), listener, channel_logging: options.channel_logging, - shutdown: TaskCancellation::default(), } } - pub(crate) fn cancellation(&self) -> TaskCancellation { - self.shutdown.clone() - } - // runs until it is shut down - pub(crate) async fn run(&mut self) -> Shutdown { - let shutdown = self.shutdown.clone(); + pub(crate) async fn run(&mut self, shutdown: ShutdownSignal) -> Shutdown { shutdown .run_until_cancelled(self.run_until_shutdown()) .await; diff --git a/rodbus/src/tcp/tls/client.rs b/rodbus/src/tcp/tls/client.rs index daa4d29c..7db1cf3b 100644 --- a/rodbus/src/tcp/tls/client.rs +++ b/rodbus/src/tcp/tls/client.rs @@ -44,6 +44,7 @@ pub(crate) fn create_tls_channel( listener: Box>, ) -> (Channel, ClientTask) { let (tx, rx) = tokio::sync::mpsc::channel(options.max_queued_requests); + let (shutdown, signal) = crate::common::cancellation::pair(); let task = TcpChannelTask::new( host, rx.into(), @@ -52,8 +53,7 @@ pub(crate) fn create_tls_channel( options, listener, ); - let shutdown = task.cancellation(); - (Channel { tx, shutdown }, ClientTask::tcp(task)) + (Channel { tx, shutdown }, ClientTask::tcp(task, signal)) } impl TlsClientConfig { From 0a630ad4b6b26c38bd091b879620ae7b43d26f2e Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Thu, 3 Sep 2026 09:59:30 -0700 Subject: [PATCH 5/7] refactor(server): move to the ShutdownHandle/ShutdownSignal pair and delete TaskCancellation Create the pair next to the command mpsc in the create_*_server_task functions. ServerTask::run applies cancellation once for both transports, so the TCP and RTU tasks lose their run/run_inner split. The TCP task keeps a signal, passed into new(), solely to clone into each spawned session. RtuServerTask goes back to the struct literal now that it has no token to initialize. Log a single "server shutdown" line regardless of whether the handle was dropped or shutdown() was called. --- rodbus/src/common/cancellation.rs | 25 -------------- rodbus/src/serial/server.rs | 29 ---------------- rodbus/src/server/mod.rs | 57 +++++++++++++++++++++++-------- rodbus/src/tcp/server.rs | 29 ++++------------ 4 files changed, 48 insertions(+), 92 deletions(-) diff --git a/rodbus/src/common/cancellation.rs b/rodbus/src/common/cancellation.rs index b27ae2a0..c3d619e9 100644 --- a/rodbus/src/common/cancellation.rs +++ b/rodbus/src/common/cancellation.rs @@ -50,31 +50,6 @@ impl ShutdownSignal { } } -/// Cancellation signal shared by a public handle and its background task. -/// -/// Superseded by [`pair`]; still used by the server side until it is migrated. -#[derive(Clone, Debug, Default)] -pub(crate) struct TaskCancellation { - token: CancellationToken, -} - -impl TaskCancellation { - pub(crate) fn cancel(&self) { - self.token.cancel(); - } - - pub(crate) async fn run_until_cancelled(&self, operation: F) -> Option - where - F: Future, - { - tokio::select! { - biased; - _ = self.token.cancelled() => None, - result = operation => Some(result), - } - } -} - #[cfg(test)] mod tests { use std::future::Future; diff --git a/rodbus/src/serial/server.rs b/rodbus/src/serial/server.rs index 75ecc426..d441c854 100644 --- a/rodbus/src/serial/server.rs +++ b/rodbus/src/serial/server.rs @@ -1,4 +1,3 @@ -use crate::common::cancellation::TaskCancellation; use crate::common::phys::PhysLayer; use crate::server::task::SessionTask; use crate::server::RequestHandler; @@ -12,41 +11,13 @@ where pub(crate) retry: Box, pub(crate) settings: SerialSettings, pub(crate) session: SessionTask, - shutdown: TaskCancellation, } impl RtuServerTask where T: RequestHandler, { - pub(crate) fn new( - port: String, - retry: Box, - settings: SerialSettings, - session: SessionTask, - ) -> Self { - Self { - port, - retry, - settings, - session, - shutdown: TaskCancellation::default(), - } - } - - pub(crate) fn cancellation(&self) -> TaskCancellation { - self.shutdown.clone() - } - pub(crate) async fn run(&mut self) -> Shutdown { - let shutdown = self.shutdown.clone(); - shutdown - .run_until_cancelled(self.run_inner()) - .await - .unwrap_or(Shutdown) - } - - async fn run_inner(&mut self) -> Shutdown { loop { match crate::serial::open(&self.port, self.settings) { Ok(serial) => { diff --git a/rodbus/src/server/mod.rs b/rodbus/src/server/mod.rs index 0ca33f47..9a0b662a 100644 --- a/rodbus/src/server/mod.rs +++ b/rodbus/src/server/mod.rs @@ -2,7 +2,7 @@ use std::net::SocketAddr; use tracing::Instrument; -use crate::common::cancellation::TaskCancellation; +use crate::common::cancellation::{ShutdownHandle, ShutdownSignal}; use crate::decode::DecodeLevel; use crate::server::task::ServerCommand; use crate::tcp::server::{ServerTask as TcpServerTask, TcpServerConnectionHandler}; @@ -35,7 +35,7 @@ pub use crate::tcp::tls::*; #[derive(Debug)] pub struct ServerHandle { tx: tokio::sync::mpsc::Sender, - shutdown: TaskCancellation, + shutdown: ShutdownHandle, } /// A server task that has been created but not yet spawned. @@ -49,6 +49,7 @@ pub struct ServerHandle { /// caller is free to wrap [`run`](ServerTask::run) with their own instrumentation. pub struct ServerTask { inner: ServerTaskInner, + shutdown: ShutdownSignal, } enum ServerTaskInner { @@ -61,34 +62,44 @@ enum ServerTaskInner { } impl ServerTask { - fn tcp(task: TcpServerTask, commands: tokio::sync::mpsc::Receiver) -> Self { + fn tcp( + task: TcpServerTask, + commands: tokio::sync::mpsc::Receiver, + shutdown: ShutdownSignal, + ) -> Self { Self { inner: ServerTaskInner::Tcp(Box::new(task), commands), + shutdown, } } #[cfg(feature = "serial")] - fn rtu(task: crate::serial::server::RtuServerTask) -> Self { + fn rtu(task: crate::serial::server::RtuServerTask, shutdown: ShutdownSignal) -> Self { Self { inner: ServerTaskInner::Rtu(Box::new(task)), + shutdown, } } /// Run the server task until the associated [`ServerHandle`] is dropped or /// [`ServerHandle::shutdown`] is called. pub async fn run(self) { - match self.inner { - ServerTaskInner::Tcp(mut task, commands) => task.run(commands).await, + let ServerTask { inner, shutdown } = self; + match inner { + ServerTaskInner::Tcp(mut task, commands) => { + shutdown.run_until_cancelled(task.run(commands)).await; + } #[cfg(feature = "serial")] ServerTaskInner::Rtu(mut task) => { - task.run().await; + shutdown.run_until_cancelled(task.run()).await; } } + tracing::info!("server shutdown"); } } impl ServerHandle { - fn new(tx: tokio::sync::mpsc::Sender, shutdown: TaskCancellation) -> Self { + fn new(tx: tokio::sync::mpsc::Sender, shutdown: ShutdownHandle) -> Self { ServerHandle { tx, shutdown } } @@ -157,6 +168,7 @@ pub fn create_tcp_server_task( decode: DecodeLevel, ) -> (ServerHandle, ServerTask) { let (tx, rx) = tokio::sync::mpsc::channel(SERVER_COMMAND_CHANNEL_CAPACITY); + let (shutdown, signal) = crate::common::cancellation::pair(); let task = TcpServerTask::new( max_sessions, listener, @@ -164,10 +176,13 @@ pub fn create_tcp_server_task( TcpServerConnectionHandler::Tcp, filter, decode, + signal.clone(), ); - let shutdown = task.cancellation(); - (ServerHandle::new(tx, shutdown), ServerTask::tcp(task, rx)) + ( + ServerHandle::new(tx, shutdown), + ServerTask::tcp(task, rx, signal), + ) } /// Spawns a RTU server task onto the runtime. @@ -217,6 +232,7 @@ pub fn create_rtu_server_task( decode: DecodeLevel, ) -> (ServerHandle, ServerTask) { let (tx, rx) = tokio::sync::mpsc::channel(SERVER_COMMAND_CHANNEL_CAPACITY); + let (shutdown, signal) = crate::common::cancellation::pair(); let session = crate::server::task::SessionTask::new( handlers, crate::server::task::AuthorizationType::None, @@ -226,10 +242,17 @@ pub fn create_rtu_server_task( decode, ); - let rtu = crate::serial::server::RtuServerTask::new(path.to_string(), retry, settings, session); - let shutdown = rtu.cancellation(); + let rtu = crate::serial::server::RtuServerTask { + port: path.to_string(), + retry, + settings, + session, + }; - (ServerHandle::new(tx, shutdown), ServerTask::rtu(rtu)) + ( + ServerHandle::new(tx, shutdown), + ServerTask::rtu(rtu, signal), + ) } /// Spawns a "raw" TLS server task onto the runtime. This TLS server does NOT require that @@ -411,6 +434,7 @@ fn create_tls_server_task_impl( decode: DecodeLevel, ) -> (ServerHandle, ServerTask) { let (tx, rx) = tokio::sync::mpsc::channel(SERVER_COMMAND_CHANNEL_CAPACITY); + let (shutdown, signal) = crate::common::cancellation::pair(); let task = TcpServerTask::new( max_sessions, listener, @@ -418,8 +442,11 @@ fn create_tls_server_task_impl( TcpServerConnectionHandler::Tls(tls_config, auth_handler), filter, decode, + signal.clone(), ); - let shutdown = task.cancellation(); - (ServerHandle::new(tx, shutdown), ServerTask::tcp(task, rx)) + ( + ServerHandle::new(tx, shutdown), + ServerTask::tcp(task, rx, signal), + ) } diff --git a/rodbus/src/tcp/server.rs b/rodbus/src/tcp/server.rs index 7f2fa224..f09e9459 100644 --- a/rodbus/src/tcp/server.rs +++ b/rodbus/src/tcp/server.rs @@ -2,7 +2,7 @@ use std::collections::BTreeMap; use tracing::Instrument; -use crate::common::cancellation::TaskCancellation; +use crate::common::cancellation::ShutdownSignal; use crate::common::frame::{FrameWriter, FramedReader}; use crate::common::phys::PhysLayer; use crate::decode::DecodeLevel; @@ -107,7 +107,8 @@ pub(crate) struct ServerTask { decode: DecodeLevel, tx: tokio::sync::mpsc::Sender, rx: tokio::sync::mpsc::Receiver, - shutdown: TaskCancellation, + /// sessions are spawned, so each gets a clone to observe cancellation directly + shutdown: ShutdownSignal, } impl ServerTask @@ -121,6 +122,7 @@ where connection_handler: TcpServerConnectionHandler, filter: AddressFilter, decode: DecodeLevel, + shutdown: ShutdownSignal, ) -> Self { let (tx, rx) = tokio::sync::mpsc::channel(8); @@ -133,14 +135,10 @@ where decode, tx, rx, - shutdown: TaskCancellation::default(), + shutdown, } } - pub(crate) fn cancellation(&self) -> TaskCancellation { - self.shutdown.clone() - } - async fn apply_command(&mut self, command: ServerCommand) { // first, change it locally so that it is applied to new sessions match command { @@ -158,26 +156,12 @@ where } pub(crate) async fn run(&mut self, mut commands: tokio::sync::mpsc::Receiver) { - let shutdown = self.shutdown.clone(); - if shutdown - .run_until_cancelled(self.run_inner(&mut commands)) - .await - .is_none() - { - tracing::info!("server shutdown requested"); - } - } - - async fn run_inner(&mut self, commands: &mut tokio::sync::mpsc::Receiver) { loop { tokio::select! { command = commands.recv() => { match command { Some(command) => self.apply_command(command).await, - None => { - tracing::info!("server shutdown"); - return; // shutdown signal - } + None => return, // the handle was dropped } } shutdown = self.rx.recv() => { @@ -222,7 +206,6 @@ where let connection_handler = self.connection_handler.clone(); let handler_map = self.handlers.clone(); let decode_level = self.decode; - // sessions are spawned, so they need to observe cancellation directly let shutdown = self.shutdown.clone(); let session = async move { From 373cb0ac9bcdc0e4897a1aded404946965c616e5 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Thu, 3 Sep 2026 10:02:40 -0700 Subject: [PATCH 6/7] refactor(client): apply cancellation once in ClientTask::run Mirror the server: ClientTask::run wraps the transport's run() in run_until_cancelled and then calls its shutdown() to drain the queue and report the terminal state. The TCP and serial tasks lose their run/run_until_shutdown/run_inner stack and never see the signal. --- rodbus/src/client/channel.rs | 9 ++++++--- rodbus/src/serial/client.rs | 30 +++++++++++++----------------- rodbus/src/tcp/client.rs | 26 +++++++++----------------- 3 files changed, 28 insertions(+), 37 deletions(-) diff --git a/rodbus/src/client/channel.rs b/rodbus/src/client/channel.rs index 0b4193bc..cc250676 100644 --- a/rodbus/src/client/channel.rs +++ b/rodbus/src/client/channel.rs @@ -60,13 +60,16 @@ impl ClientTask { /// 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(self.shutdown).await; + shutdown.run_until_cancelled(task.run()).await; + task.shutdown().await; } #[cfg(feature = "serial")] ClientTaskInner::Serial(mut task) => { - task.run(self.shutdown).await; + shutdown.run_until_cancelled(task.run()).await; + task.shutdown().await; } } } diff --git a/rodbus/src/serial/client.rs b/rodbus/src/serial/client.rs index be10a869..e439a640 100644 --- a/rodbus/src/serial/client.rs +++ b/rodbus/src/serial/client.rs @@ -5,7 +5,6 @@ use crate::serial::SerialSettings; use crate::client::message::Command; use crate::client::task::{ClientLoop, SessionError, StateChange}; use crate::client::{Listener, PortState, RetryStrategy}; -use crate::common::cancellation::ShutdownSignal; use crate::common::frame::{FrameWriter, FramedReader}; use crate::error::Shutdown; @@ -41,22 +40,9 @@ impl SerialChannelTask { } } - pub(crate) async fn run(&mut self, shutdown: ShutdownSignal) -> Shutdown { - shutdown - .run_until_cancelled(self.run_until_shutdown()) - .await; - - self.client_loop.shutdown().await; - self.listener.update(PortState::Shutdown).get().await; - Shutdown - } - - async fn run_until_shutdown(&mut self) -> Shutdown { + /// Run until every handle is dropped. Cancellation is applied by the caller. + pub(crate) async fn run(&mut self) -> Shutdown { self.listener.update(PortState::Disabled).get().await; - self.run_inner().await - } - - async fn run_inner(&mut self) -> Shutdown { loop { // wait for the channel to be enabled if let Err(Shutdown) = self.client_loop.wait_for_enabled().await { @@ -73,6 +59,13 @@ impl SerialChannelTask { } } + /// Fail everything still queued and report the terminal state. Called after `run` completes or + /// is cancelled. + pub(crate) async fn shutdown(&mut self) { + self.client_loop.shutdown().await; + self.listener.update(PortState::Shutdown).get().await; + } + pub(crate) async fn try_open_and_run(&mut self) -> Result<(), StateChange> { match crate::serial::open(self.path.as_str(), self.serial_settings) { Err(err) => { @@ -146,7 +139,10 @@ mod tests { Box::new(BlockingDisabledListener { states }), ); let (cancellation, signal) = crate::common::cancellation::pair(); - let task = tokio::spawn(async move { task.run(signal).await }); + let task = tokio::spawn(async move { + signal.run_until_cancelled(task.run()).await; + task.shutdown().await + }); assert_eq!(state_rx.recv().await, Some(PortState::Disabled)); cancellation.cancel(); diff --git a/rodbus/src/tcp/client.rs b/rodbus/src/tcp/client.rs index a9da7624..f0cd613e 100644 --- a/rodbus/src/tcp/client.rs +++ b/rodbus/src/tcp/client.rs @@ -1,7 +1,6 @@ use tracing::Instrument; use crate::client::{Channel, ClientState, ClientTask, HostAddr, Listener}; -use crate::common::cancellation::ShutdownSignal; use crate::common::phys::PhysLayer; use crate::client::message::Command; @@ -111,23 +110,9 @@ impl TcpChannelTask { } } - // runs until it is shut down - pub(crate) async fn run(&mut self, shutdown: ShutdownSignal) -> Shutdown { - shutdown - .run_until_cancelled(self.run_until_shutdown()) - .await; - - self.client_loop.shutdown().await; - self.listener.update(ClientState::Shutdown).get().await; - Shutdown - } - - async fn run_until_shutdown(&mut self) -> Shutdown { + /// Run until every handle is dropped. Cancellation is applied by the caller. + pub(crate) async fn run(&mut self) -> Shutdown { self.listener.update(ClientState::Disabled).get().await; - self.run_inner().await - } - - async fn run_inner(&mut self) -> Shutdown { loop { if let Err(Shutdown) = self.client_loop.wait_for_enabled().await { return Shutdown; @@ -143,6 +128,13 @@ impl TcpChannelTask { } } + /// Fail everything still queued and report the terminal state. Called after `run` completes or + /// is cancelled. + pub(crate) async fn shutdown(&mut self) { + self.client_loop.shutdown().await; + self.listener.update(ClientState::Shutdown).get().await; + } + async fn connect(&mut self) -> Result, StateChange> { tokio::select! { res = self.host.connect() => { From 5d1fd0701ed5a28649ca8ce4943f72e2eebd6546 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Thu, 3 Sep 2026 10:08:32 -0700 Subject: [PATCH 7/7] test(client): drop a no-op socket write at the end of the queued-requests test The socket already lives to the end of the test, so the zero-byte write kept nothing alive. Remove it and the import it alone used. --- rodbus/src/tcp/client.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/rodbus/src/tcp/client.rs b/rodbus/src/tcp/client.rs index f0cd613e..21348b2d 100644 --- a/rodbus/src/tcp/client.rs +++ b/rodbus/src/tcp/client.rs @@ -217,7 +217,7 @@ mod tests { use crate::{AddressRange, RequestError, RetryStrategy, UnitId}; use std::net::Ipv4Addr; use std::time::Duration; - use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::io::AsyncReadExt; /// A request timeout long enough that any test which observes a request completing before it /// elapses has necessarily observed cancellation rather than a timeout @@ -450,7 +450,5 @@ mod tests { // the handle outlives the task it terminated channel.shutdown(); assert_eq!(channel.enable().await, Err(crate::error::Shutdown)); - - drop(socket.write(&[]).await); } }