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/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 330fb031..cc250676 100644 --- a/rodbus/src/client/channel.rs +++ b/rodbus/src/client/channel.rs @@ -5,6 +5,7 @@ use crate::client::requests::read_bits::ReadBits; use crate::client::requests::read_registers::ReadRegisters; use crate::client::requests::write_multiple::{MultipleWriteRequest, WriteMultiple}; use crate::client::requests::write_single::SingleWrite; +use crate::common::cancellation::{ShutdownHandle, ShutdownSignal}; use crate::error::*; use crate::types::{AddressRange, BitIterator, Indexed, RegisterIterator, UnitId}; use crate::DecodeLevel; @@ -14,6 +15,7 @@ use crate::DecodeLevel; #[derive(Debug, Clone)] pub struct Channel { pub(crate) tx: tokio::sync::mpsc::Sender, + pub(crate) shutdown: ShutdownHandle, } /// A client channel task that has been created but not yet spawned. @@ -27,6 +29,7 @@ pub struct Channel { /// caller is free to wrap [`run`](ClientTask::run) with their own instrumentation. pub struct ClientTask { inner: ClientTaskInner, + shutdown: ShutdownSignal, } enum ClientTaskInner { @@ -36,29 +39,37 @@ enum ClientTaskInner { } impl ClientTask { - pub(crate) fn tcp(task: crate::tcp::client::TcpChannelTask) -> Self { + pub(crate) fn tcp(task: crate::tcp::client::TcpChannelTask, shutdown: ShutdownSignal) -> Self { Self { inner: ClientTaskInner::Tcp(task), + shutdown, } } #[cfg(feature = "serial")] - pub(crate) fn serial(task: crate::serial::client::SerialChannelTask) -> Self { + pub(crate) fn serial( + task: crate::serial::client::SerialChannelTask, + shutdown: ShutdownSignal, + ) -> Self { Self { inner: ClientTaskInner::Serial(task), + shutdown, } } /// Run the channel task until every [`Channel`] handle is dropped or [`Channel::shutdown`] is /// called. pub async fn run(self) { - match self.inner { + let ClientTask { inner, shutdown } = self; + match inner { ClientTaskInner::Tcp(mut task) => { - task.run().await; + shutdown.run_until_cancelled(task.run()).await; + task.shutdown().await; } #[cfg(feature = "serial")] ClientTaskInner::Serial(mut task) => { - task.run().await; + shutdown.run_until_cancelled(task.run()).await; + task.shutdown().await; } } } @@ -124,6 +135,7 @@ impl Channel { listener: Option>>, ) -> (Self, ClientTask) { let (tx, rx) = tokio::sync::mpsc::channel(max_queued_requests); + let (shutdown, signal) = crate::common::cancellation::pair(); let task = crate::serial::client::SerialChannelTask::new( path, serial_settings, @@ -132,7 +144,7 @@ impl Channel { decode, listener.unwrap_or_else(|| crate::client::NullListener::create()), ); - (Channel { tx }, ClientTask::serial(task)) + (Channel { tx, shutdown }, ClientTask::serial(task, signal)) } /// Enable communications @@ -149,10 +161,15 @@ impl Channel { /// Begin shutting down the channel task, even if one or more [`Channel`] handles are still alive /// - /// The task completes when it processes the command, which may be after this returns - pub async fn shutdown(&self) -> Result<(), Shutdown> { - self.tx.send(Command::Shutdown).await?; - Ok(()) + /// Active work is cancelled at its next suspension point. A request already on the wire is + /// abandoned, and every queued request fails with [`RequestError::Shutdown`]. A write that was + /// already transmitted may still be applied by the server, so its outcome is indeterminate. + /// + /// The task reports the terminal state to its listener before completing, so a + /// [`Listener`](crate::client::Listener) that never completes that update keeps the task alive. + /// Calling this more than once, or after the task has already terminated, has no effect. + pub fn shutdown(&self) { + self.shutdown.cancel(); } /// Read coils from the server 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/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..1fd3c9c3 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) => { @@ -168,7 +172,6 @@ impl ClientLoop { Ok(()) } Command::Request(mut request) => self.run_one_request(io, &mut request).await, - Command::Shutdown => Err(SessionError::Shutdown), } } @@ -337,7 +340,6 @@ impl ClientLoop { Err(StateChange::Disable) } } - Command::Shutdown => Err(StateChange::Shutdown), } } @@ -401,7 +403,10 @@ mod tests { let mut phys = PhysLayer::new_mock(mock); client_loop.run(&mut phys).await }); - let channel = Channel { tx }; + let channel = Channel { + tx, + shutdown: crate::common::cancellation::pair().0, + }; (channel, join_handle, io_handle) } @@ -432,83 +437,6 @@ mod tests { assert_eq!(task.await.unwrap(), SessionError::Shutdown); } - #[tokio::test] - async fn task_completes_when_shutdown_requested_with_a_handle_still_alive() { - let (channel, task, _io) = spawn_client_loop(); - - channel.shutdown().await.unwrap(); - assert_eq!(task.await.unwrap(), SessionError::Shutdown); - - // the handle outlived the task it terminated, and now reports that it is gone - assert_eq!(channel.shutdown().await, Err(Shutdown)); - let res = channel - .read_coils( - RequestParam::new(UnitId::new(1), Duration::from_secs(1)), - AddressRange::try_from(7, 2).unwrap(), - ) - .await; - assert_eq!(res, Err(RequestError::Shutdown)); - } - - #[tokio::test] - async fn transaction_in_flight_completes_before_requested_shutdown() { - let (channel, task, mut io) = spawn_client_loop(); - let other_handle = channel.clone(); - - let range = AddressRange::try_from(7, 2).unwrap(); - let request = get_framed_adu(FunctionCode::ReadCoils, &range); - let response = get_framed_adu( - FunctionCode::ReadCoils, - &BitWriter::new(ReadBitsRange { inner: range }, |idx| match idx { - 7 => Ok(true), - 8 => Ok(false), - _ => Err(ExceptionCode::IllegalDataAddress), - }), - ); - - let coils = tokio::spawn(async move { - channel - .read_coils( - RequestParam::new(UnitId::new(1), Duration::from_secs(1)), - range, - ) - .await - }); - - // the request is on the wire, so the loop is inside a transaction - assert_eq!(io.next_event().await, Event::Write(request)); - - // shutdown queues behind that transaction rather than interrupting it - other_handle.shutdown().await.unwrap(); - io.read(&response); - - assert_eq!( - coils.await.unwrap().unwrap(), - vec![Indexed::new(7, true), Indexed::new(8, false)] - ); - assert_eq!(task.await.unwrap(), SessionError::Shutdown); - } - - #[tokio::test] - async fn shutdown_ends_the_task_while_it_is_failing_requests() { - // fail_requests() is the path taken while disconnected or waiting to retry, which reads - // the queue separately from the session loop and so needs its own handling - let (tx, rx) = tokio::sync::mpsc::channel(16); - let mut client_loop = ClientLoop::new( - rx.into(), - FrameWriter::tcp(), - FramedReader::tcp(), - DecodeLevel::nothing(), - None, - ); - let channel = Channel { tx }; - - let task = tokio::spawn(async move { client_loop.fail_requests().await }); - channel.shutdown().await.unwrap(); - - assert_eq!(task.await.unwrap(), StateChange::Shutdown); - } - #[tokio::test] async fn returns_io_error_when_write_fails() { let (channel, _task, mut io) = spawn_client_loop(); diff --git a/rodbus/src/common/cancellation.rs b/rodbus/src/common/cancellation.rs new file mode 100644 index 00000000..c3d619e9 --- /dev/null +++ b/rodbus/src/common/cancellation.rs @@ -0,0 +1,110 @@ +use std::future::Future; + +use tokio_util::sync::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)) +} + +/// 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.0.cancel(); + } +} + +/// 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), + } + } +} + +#[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 (handle, signal) = pair(); + handle.cancel(); + + assert_eq!(signal.run_until_cancelled(PanicOnPoll).await, None); + } + + #[tokio::test] + async fn cancellation_drops_a_pending_operation() { + let (handle, signal) = pair(); + let dropped = Arc::new(AtomicBool::new(false)); + let flag = DropFlag(dropped.clone()); + + let task = tokio::spawn(async move { + signal + .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 b8453e8e..e439a640 100644 --- a/rodbus/src/serial/client.rs +++ b/rodbus/src/serial/client.rs @@ -40,14 +40,9 @@ impl SerialChannelTask { } } + /// 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; - let ret = self.run_inner().await; - self.listener.update(PortState::Shutdown).get().await; - ret - } - - 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 { @@ -64,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) => { @@ -98,3 +100,57 @@ 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, signal) = crate::common::cancellation::pair(); + 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(); + + 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 5d8cc81a..d441c854 100644 --- a/rodbus/src/serial/server.rs +++ b/rodbus/src/serial/server.rs @@ -50,3 +50,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(); + } +} diff --git a/rodbus/src/server/mod.rs b/rodbus/src/server/mod.rs index 68319728..9a0b662a 100644 --- a/rodbus/src/server/mod.rs +++ b/rodbus/src/server/mod.rs @@ -2,6 +2,7 @@ use std::net::SocketAddr; use tracing::Instrument; +use crate::common::cancellation::{ShutdownHandle, ShutdownSignal}; use crate::decode::DecodeLevel; use crate::server::task::ServerCommand; use crate::tcp::server::{ServerTask as TcpServerTask, TcpServerConnectionHandler}; @@ -29,24 +30,26 @@ 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 is shutdown when the handle is dropped or +/// [`ServerHandle::shutdown`] is called. #[derive(Debug)] pub struct ServerHandle { tx: tokio::sync::mpsc::Sender, + shutdown: ShutdownHandle, } /// 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 +/// [`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. pub struct ServerTask { inner: ServerTaskInner, + shutdown: ShutdownSignal, } enum ServerTaskInner { @@ -59,38 +62,45 @@ 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 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 { - 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 { - /// 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 } + fn new(tx: tokio::sync::mpsc::Sender, shutdown: ShutdownHandle) -> Self { + ServerHandle { tx, shutdown } } /// Change the decoding level for future sessions and all active sessions @@ -99,12 +109,14 @@ impl ServerHandle { Ok(()) } - /// Begin shutting down the server task, even if one or more [`ServerHandle`]s are still alive + /// Begin shutting down the server task, even while the [`ServerHandle`] is 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(ServerCommand::Shutdown).await?; - Ok(()) + /// 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) { + self.shutdown.cancel(); } } @@ -156,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, @@ -163,9 +176,13 @@ pub fn create_tcp_server_task( TcpServerConnectionHandler::Tcp, filter, decode, + signal.clone(), ); - (ServerHandle::new(tx), ServerTask::tcp(task, rx)) + ( + ServerHandle::new(tx, shutdown), + ServerTask::tcp(task, rx, signal), + ) } /// Spawns a RTU server task onto the runtime. @@ -215,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, @@ -231,7 +249,10 @@ pub fn create_rtu_server_task( session, }; - (ServerHandle::new(tx), 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 @@ -413,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, @@ -420,7 +442,11 @@ fn create_tls_server_task_impl( TcpServerConnectionHandler::Tls(tls_config, auth_handler), filter, decode, + signal.clone(), ); - (ServerHandle::new(tx), ServerTask::tcp(task, rx)) + ( + ServerHandle::new(tx, shutdown), + ServerTask::tcp(task, rx, signal), + ) } diff --git a/rodbus/src/server/task.rs b/rodbus/src/server/task.rs index 644fa6a1..513a5986 100644 --- a/rodbus/src/server/task.rs +++ b/rodbus/src/server/task.rs @@ -16,11 +16,9 @@ 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), - /// 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..21348b2d 100644 --- a/rodbus/src/tcp/client.rs +++ b/rodbus/src/tcp/client.rs @@ -44,6 +44,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(), @@ -52,7 +53,7 @@ pub(crate) fn create_tcp_channel( options, listener, ); - (Channel { tx }, ClientTask::tcp(task)) + (Channel { tx, shutdown }, ClientTask::tcp(task, signal)) } pub(crate) enum TcpTaskConnectionHandler { @@ -109,15 +110,9 @@ impl TcpChannelTask { } } - // runs until it is shut down + /// 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; - let ret = self.run_inner().await; - self.listener.update(ClientState::Shutdown).get().await; - ret - } - - async fn run_inner(&mut self) -> Shutdown { loop { if let Err(Shutdown) = self.client_loop.wait_for_enabled().await { return Shutdown; @@ -133,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() => { @@ -209,9 +211,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; + + /// 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 +237,40 @@ 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 { + 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 +283,172 @@ mod tests { ); let task = tokio::spawn(task.run()); - channel.shutdown().await.unwrap(); - task.await.unwrap(); + assert_eq!(states.recv().await, Some(ClientState::Disabled)); + 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 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(); + + 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 request).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_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(), + Box::new(BlockingShutdownListener { + tx, + release: Some(release_rx), + }), + 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(); + loop { + match states.recv().await { + Some(ClientState::Shutdown) => break, + Some(_) => {} + None => panic!("client task ended without reporting shutdown"), + } + } + + // requests fail before the terminal listener is allowed to complete + for handle in queued { + assert_eq!( + tokio::time::timeout(TEST_TIMEOUT, handle) + .await + .expect("a queued request was left waiting") + .unwrap(), + Err(RequestError::Shutdown) + ); + } + assert!(!task.is_finished()); + release_tx.send(()).unwrap(); + + 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)); + } } diff --git a/rodbus/src/tcp/server.rs b/rodbus/src/tcp/server.rs index 79154588..f09e9459 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::ShutdownSignal; use crate::common::frame::{FrameWriter, FramedReader}; use crate::common::phys::PhysLayer; use crate::decode::DecodeLevel; @@ -106,6 +107,8 @@ pub(crate) struct ServerTask { decode: DecodeLevel, tx: tokio::sync::mpsc::Sender, rx: tokio::sync::mpsc::Receiver, + /// sessions are spawned, so each gets a clone to observe cancellation directly + shutdown: ShutdownSignal, } impl ServerTask @@ -119,6 +122,7 @@ where connection_handler: TcpServerConnectionHandler, filter: AddressFilter, decode: DecodeLevel, + shutdown: ShutdownSignal, ) -> 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() { @@ -157,16 +160,8 @@ where 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"); - return; // shutdown signal - } + None => return, // the handle was dropped } } shutdown = self.rx.recv() => { @@ -211,17 +206,19 @@ where let connection_handler = self.connection_handler.clone(); let handler_map = self.handlers.clone(); let decode_level = self.decode; + let shutdown = self.shutdown.clone(); let session = async move { - run_session( - socket, - addr, - connection_handler, - decode_level, - handler_map, - rx, - ) - .await; + 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; @@ -267,17 +264,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 +292,55 @@ 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(); + + handle.shutdown(); + tokio::time::timeout(TEST_TIMEOUT, task) + .await + .unwrap() + .unwrap(); - // the handle outlived the task it terminated, and now reports that it is gone - assert_eq!(handle.shutdown().await, Err(Shutdown)); + // 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; 9]; + tokio::time::timeout(TEST_TIMEOUT, client.read_exact(&mut response)) + .await + .expect("the server never answered, so the session was not established") + .unwrap(); + + 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..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,7 +53,7 @@ pub(crate) fn create_tls_channel( options, listener, ); - (Channel { tx }, ClientTask::tcp(task)) + (Channel { tx, shutdown }, ClientTask::tcp(task, signal)) } impl TlsClientConfig {