diff --git a/src/control_subscriber.rs b/src/control_subscriber.rs index b54d623..c9a5d08 100644 --- a/src/control_subscriber.rs +++ b/src/control_subscriber.rs @@ -117,7 +117,7 @@ impl ControlSubscriberConfig { pub async fn from_settings( settings: &Settings, plc_writer_factory: WriterFactory, - serial_cache: &crate::modbus_client::SerialActorCache, + actor_cache: &crate::modbus_client::ActorCache, ) -> Result { settings.modbus.as_ref().context( "ControlSubscriberConfig::from_settings called without a PLC in the mapping", @@ -143,7 +143,7 @@ impl ControlSubscriberConfig { } // Additional PLC or slave — build the factory from the connection block. let factory = - crate::modbus_client::build_source_writer(&device.connection, serial_cache) + crate::modbus_client::build_source_writer(&device.connection, actor_cache) .await .with_context(|| format!("build writer for device '{}'", device.device_id))?; writer_factories diff --git a/src/main.rs b/src/main.rs index 801009e..a367959 100644 --- a/src/main.rs +++ b/src/main.rs @@ -155,10 +155,10 @@ async fn main() -> Result<()> { // If there isn't, the telemetry/control/plc_status tasks use pending() // (idle). // --------------------------------------------------------------------------- - let serial_cache = modbus_client::SerialActorCache::new(); + let actor_cache = modbus_client::ActorCache::new(); let (reader_factory_opt, writer_factory_opt) = if let Some(modbus) = &settings.modbus { - let (rf, wf) = modbus_client::build_io(modbus, &serial_cache) + let (rf, wf) = modbus_client::build_io(modbus, &actor_cache) .await .context("build modbus I/O")?; (Some(rf), Some(wf)) @@ -170,10 +170,10 @@ async fn main() -> Result<()> { let (plc_status_tx, plc_status_rx) = tokio::sync::watch::channel(false); let tc = - edge_client::telemetry::TelemetryConfig::from_settings(&settings, rf, &serial_cache) + edge_client::telemetry::TelemetryConfig::from_settings(&settings, rf, &actor_cache) .await .context("build telemetry config")?; - let cc = ControlSubscriberConfig::from_settings(&settings, wf, &serial_cache) + let cc = ControlSubscriberConfig::from_settings(&settings, wf, &actor_cache) .await .context("build control subscriber config")?; diff --git a/src/modbus_actor.rs b/src/modbus_actor.rs index 7346ac7..46502ae 100644 --- a/src/modbus_actor.rs +++ b/src/modbus_actor.rs @@ -1,8 +1,11 @@ -//! Actor task for RTU mode (RTU-over-TCP & RTU-Serial). +//! Actor task for every Modbus transport (TCP, RTU-over-TCP, RTU-Serial). //! -//! Modbus RTU = half-duplex, one physical port. It can't be opened twice -//! (serial port: OS lock; RTU-over-TCP: the gateway converter usually doesn't -//! handle multi-client well). So we use the actor pattern: +//! A single connection that serializes all requests. Needed because the physical +//! link can't be opened twice safely: serial port = OS exclusive lock; RTU-over-TCP +//! = the gateway converter usually mishandles multi-client; plain TCP = many small +//! PLCs / Modbus-TCP gateways accept only ONE concurrent socket, so a separate +//! reader + writer socket to the same host:port would make the second one fail +//! (write returns EINPROGRESS / os error 115). So we use the actor pattern: //! //! - **One task** that owns the connection (serial port or TcpStream). //! - **An mpsc channel** for inbound requests from telemetry + control + bridge. @@ -38,7 +41,7 @@ use async_trait::async_trait; use std::time::Duration; use tokio::sync::{mpsc, oneshot}; use tokio_modbus::client::Context as ModbusContext; -use tokio_modbus::prelude::{Reader, Slave, SlaveContext, Writer, rtu}; +use tokio_modbus::prelude::{Reader, Slave, SlaveContext, Writer, rtu, tcp}; use tracing::{debug, info, warn}; use crate::modbus_client::{ReadConn, WriteConn}; @@ -133,22 +136,12 @@ impl ActorHandle { /// Requests that arrive before the connect finishes are queued and executed once /// the connection is up. /// -/// An error here means the transport is fundamentally invalid (e.g. an -/// unsupported variant). Network errors don't surface — the actor handles them -/// internally. +/// Returns `Result` for call-site ergonomics, but currently never fails: every +/// transport is supported and network errors don't surface here — the actor +/// handles connect/reconnect internally. pub async fn spawn(transport: Transport, unit_id: u8, timings: RtuTimings) -> Result { - // Sanity check at startup so misconfig is detected quickly. The real connect - // happens inside the actor loop (for automatic retry-on-fail). - match &transport { - Transport::RtuOverTcp { .. } | Transport::RtuSerial { .. } => {} - Transport::Tcp { .. } => { - return Err(anyhow!( - "modbus_actor::spawn called with Transport::Tcp; this is a bug — \ - TCP should use ModbusReader/Writer direct" - )); - } - } - + // The real connect happens inside the actor loop (for automatic + // retry-on-fail); nothing to validate up front — all transports are handled. let (tx, rx) = mpsc::channel(INBOX_CAP); tokio::spawn(run_actor(transport, timings, rx)); Ok(ActorHandle { @@ -535,9 +528,17 @@ async fn connect(transport: &Transport) -> Result { tokio::time::sleep(Duration::from_millis(200)).await; Ok(rtu::attach_slave(stream, DUMMY_SLAVE)) } - Transport::Tcp { .. } => Err(anyhow!( - "actor connect should not be called with Tcp transport — bug in factory" - )), + Transport::Tcp { host, port } => { + let addr = format!("{host}:{port}"); + let stream = tokio::net::TcpStream::connect(&addr) + .await + .with_context(|| format!("tcp connect {addr}"))?; + // Standard Modbus TCP framing (MBAP). One shared connection, + // serialized by this actor, so a single-session PLC/gateway isn't hit + // by two concurrent sockets (reader + writer). The slave/unit is set + // per request via `ctx.set_slave()`. + Ok(tcp::attach_slave(stream, DUMMY_SLAVE)) + } } } @@ -571,22 +572,18 @@ fn map_data_bits(d: SerialDataBits) -> tokio_serial::DataBits { mod tests { use super::*; - /// Sanity: spawn() explicitly rejects the Tcp transport. Defensive — - /// without this, a factory bug could slip through into an actor task that - /// immediately crashes in connect(). Better to fail in spawn() at startup. + /// spawn() accepts every transport, including plain Modbus TCP. The actor + /// task connects (and reconnects) in the background; spawn itself returns Ok + /// without touching the network. #[tokio::test] - async fn spawn_rejects_tcp_transport() { + async fn spawn_accepts_tcp_transport() { let t = Transport::Tcp { host: "127.0.0.1".into(), port: 502, }; let res = spawn(t, 1, RtuTimings::default()).await; - assert!(res.is_err(), "spawn must reject Tcp transport"); - let msg = res.unwrap_err().to_string(); - assert!( - msg.contains("Tcp"), - "error message should mention Tcp, got: {msg}" - ); + assert!(res.is_ok(), "spawn must accept Tcp transport"); + // Drop the handle → channel closes → the background actor exits. } /// Reply type assertion: if the actor reply variant mismatches (a bug in diff --git a/src/modbus_client.rs b/src/modbus_client.rs index 3624309..0464601 100644 --- a/src/modbus_client.rs +++ b/src/modbus_client.rs @@ -1,11 +1,19 @@ //! Modbus client abstraction: `ReadConn`/`WriteConn` traits + a factory for the //! 3 transports (TCP / RTU-over-TCP / RTU-Serial). //! -//! The TCP path uses `ModbusReader` + `ModbusWriter` as two separate -//! connections (a 2-client pattern). RTU half-duplex cannot be emulated with two -//! separate connections to the same physical port → it needs a single -//! connection that serializes requests. Solution: an actor task that owns the -//! connection while Reader/Writer tasks send requests over an mpsc channel. +//! **Every transport goes through a single actor connection** ([`crate::modbus_actor`]). +//! One task owns the link; reader + writer + bridge send requests over an mpsc +//! channel and the actor serializes them. This is required for RTU (half-duplex, +//! one physical port) and also for plain Modbus TCP, because many small PLCs / +//! Modbus-TCP gateways accept only ONE concurrent socket — a separate reader and +//! writer socket to the same host:port makes the second one fail (write returns +//! EINPROGRESS / os error 115). +//! +//! Actors are shared per endpoint via [`ActorCache`], keyed by connection +//! identity (`tcp:host:port`, `rtu-tcp:host:port`, `serial:path`). So a reader +//! and a writer to the **same** host:port share one connection, while a +//! split-port setup (e.g. `:502` read / `:503` write) keeps two independent +//! connections — one per key. //! //! Two layers: //! @@ -14,35 +22,17 @@ //! which transport is behind it. //! //! 2. **Factories** [`ReaderFactory`] + [`WriterFactory`] = built once at -//! startup ([`build_io`]). The `.connect()` method returns a -//! `Box` or `Box`. Per transport: -//! - **TCP**: each `.connect()` opens a new socket (preserves the 2-client -//! pattern). -//! - **RTU**: the factory holds an actor handle; `.connect()` clones the -//! handle (cheap). Reconnecting the physical link is handled inside the -//! actor. -//! -//! TCP lifecycle: -//! ```text -//! loop { -//! let reader = factory.connect().await?; // open TCP -//! loop { reader.read_holdings(...) } // until error -//! // drop reader → socket close → outer loop reconnects -//! } -//! ``` -//! -//! The RTU lifecycle is identical from the consumer's point of view — but -//! `.connect()` just clones the handle, and a read error = a per-request error -//! (the actor restarts in the background without the caller needing to know). +//! startup ([`build_io`]). Each holds an actor handle; `.connect()` just +//! clones the handle (cheap), returning a `Box` / +//! `Box`. Reconnecting the physical link is handled inside the +//! actor, so a read/write error is a per-request error the caller can retry +//! without managing sockets itself. -use anyhow::{Context as _, Result, anyhow}; +use anyhow::{Context as _, Result}; use async_trait::async_trait; use std::collections::HashMap; -use std::net::SocketAddr; use std::sync::Arc; use tokio::sync::Mutex; -use tokio_modbus::client::Context; -use tokio_modbus::prelude::{Reader, Slave, Writer, tcp}; use tracing::debug; use crate::modbus_actor::ActorHandle; @@ -91,328 +81,118 @@ pub trait WriteConn: Send { async fn write_multiple_registers(&mut self, address: u16, values: Vec) -> Result<()>; } -// =========================================================================== -// TCP: ModbusReader + ModbusWriter (preserve existing 2-client pattern) -// =========================================================================== - -/// Stateful Modbus TCP reader. Connect-per-session: on disconnect/error, drop -/// and recreate it via [`ModbusReader::connect`]. -pub struct ModbusReader { - ctx: Context, -} - -impl ModbusReader { - /// Connect to the PLC via Modbus TCP with a given unit ID. - pub async fn connect(addr: SocketAddr, unit_id: u8) -> Result { - let ctx = tcp::connect_slave(addr, Slave(unit_id)) - .await - .with_context(|| format!("modbus tcp connect {addr} (unit={unit_id})"))?; - debug!(%addr, unit_id, "modbus tcp reader connected"); - Ok(Self { ctx }) - } -} - -#[async_trait] -impl ReadConn for ModbusReader { - async fn read_holdings(&mut self, count: u16) -> Result> { - if count == 0 { - return Ok(Vec::new()); - } - // tokio-modbus 0.16+ separates a transport error (outer Result) from a - // Modbus protocol exception (inner Result, e.g. ILLEGAL_ADDRESS from the - // PLC). Both = fatal in the poll cycle, the handler reconnects. - let res = self - .ctx - .read_holding_registers(0, count) - .await - .context("modbus read_holding_registers transport error")?; - let regs = res.map_err(|e| anyhow!("modbus protocol exception (holdings): {e:?}"))?; - Ok(regs) - } - - async fn read_coils(&mut self, count: u16) -> Result> { - if count == 0 { - return Ok(Vec::new()); - } - let res = self - .ctx - .read_coils(0, count) - .await - .context("modbus read_coils transport error")?; - let coils = res.map_err(|e| anyhow!("modbus protocol exception (coils): {e:?}"))?; - Ok(coils) - } - - async fn read_input_registers(&mut self, address: u16, count: u16) -> Result> { - if count == 0 { - return Ok(Vec::new()); - } - let res = self - .ctx - .read_input_registers(address, count) - .await - .context("modbus read_input_registers transport error")?; - let regs = - res.map_err(|e| anyhow!("modbus protocol exception (input_registers): {e:?}"))?; - Ok(regs) - } - - async fn read_discrete_inputs(&mut self, address: u16, count: u16) -> Result> { - if count == 0 { - return Ok(Vec::new()); - } - let res = self - .ctx - .read_discrete_inputs(address, count) - .await - .context("modbus read_discrete_inputs transport error")?; - let bits = - res.map_err(|e| anyhow!("modbus protocol exception (discrete_inputs): {e:?}"))?; - Ok(bits) - } -} - -/// Stateful Modbus TCP writer. A connection separate from [`ModbusReader`] — -/// the 2-client pattern so telemetry reads aren't disturbed by control writes. -/// For RTU this lifecycle collapses to a single actor (see -/// [`crate::modbus_actor`]). -pub struct ModbusWriter { - ctx: Context, -} - -impl ModbusWriter { - pub async fn connect(addr: SocketAddr, unit_id: u8) -> Result { - let ctx = tcp::connect_slave(addr, Slave(unit_id)) - .await - .with_context(|| format!("modbus tcp connect (writer) {addr} (unit={unit_id})"))?; - debug!(%addr, unit_id, "modbus tcp writer connected"); - Ok(Self { ctx }) - } -} - -#[async_trait] -impl WriteConn for ModbusWriter { - async fn write_single_register(&mut self, address: u16, value: u16) -> Result<()> { - let res = self - .ctx - .write_single_register(address, value) - .await - .with_context(|| { - format!("modbus write_single_register(addr={address}) transport error") - })?; - res.map_err(|e| { - anyhow!("modbus protocol exception (write register {address}={value}): {e:?}") - })?; - debug!(address, value, "modbus register written"); - Ok(()) - } - - async fn write_single_coil(&mut self, address: u16, value: bool) -> Result<()> { - let res = self - .ctx - .write_single_coil(address, value) - .await - .with_context(|| format!("modbus write_single_coil(addr={address}) transport error"))?; - res.map_err(|e| anyhow!("modbus protocol exception (write coil {address}): {e:?}"))?; - debug!(address, value, "modbus coil written"); - Ok(()) - } - - async fn write_multiple_registers(&mut self, address: u16, values: Vec) -> Result<()> { - let count = values.len(); - let res = self - .ctx - .write_multiple_registers(address, &values) - .await - .with_context(|| { - format!( - "modbus write_multiple_registers(addr={address}, count={count}) transport error" - ) - })?; - res.map_err(|e| { - anyhow!("modbus protocol exception (write multi {address}, {count} regs): {e:?}") - })?; - debug!(address, count, "modbus multi registers written"); - Ok(()) - } -} - // =========================================================================== // Factory layer: ReaderFactory / WriterFactory // =========================================================================== -/// Factory for creating a `ReadConn` session. For TCP, each `connect()` opens a -/// new socket. For RTU, it returns a clone of the actor handle (shared with the -/// `WriterFactory` on the same transport). -#[derive(Clone)] -pub struct ReaderFactory(Backend); - -/// Factory for creating a `WriteConn` session. +/// Factory for creating a `ReadConn` session. Holds an actor handle; `connect()` +/// clones it (cheap). The actor owns the single shared connection. #[derive(Clone)] -pub struct WriterFactory(Backend); +pub struct ReaderFactory(ActorHandle); +/// Factory for creating a `WriteConn` session. Holds an actor handle; `connect()` +/// clones it (cheap). Shares the same actor as the [`ReaderFactory`] when both +/// target the same endpoint (see [`ActorCache`]). #[derive(Clone)] -enum Backend { - /// TCP: descriptor only, the connection happens on each `.connect()` call. - Tcp { addr: SocketAddr, unit_id: u8 }, - /// RTU (over-TCP or serial): a handle to the actor task. Cheap to clone. - Actor(ActorHandle), -} +pub struct WriterFactory(ActorHandle); impl ReaderFactory { pub async fn connect(&self) -> Result> { - match &self.0 { - Backend::Tcp { addr, unit_id } => { - let r = ModbusReader::connect(*addr, *unit_id).await?; - Ok(Box::new(r)) - } - Backend::Actor(h) => Ok(Box::new(h.clone())), - } - } - - /// Derive a new factory for a different slave/unit ID (e.g. PLC unit 1 - /// → chiller unit 2). For TCP: clone the descriptor with the new unit_id - /// (each `.connect()` opens a separate TCP socket with the MBAP unit field - /// = this unit_id). For RTU/Actor: clone the handle with the new slave_id - /// (sharing the same actor task). Cheap in both cases. - pub fn with_unit_id(&self, unit_id: u8) -> Self { - let backend = match &self.0 { - Backend::Tcp { addr, .. } => Backend::Tcp { - addr: *addr, - unit_id, - }, - Backend::Actor(h) => Backend::Actor(h.with_slave_id(unit_id)), - }; - Self(backend) - } - - /// Create a new TCP factory to a different host:port — for a source device - /// that has its own IP (a second PLC, etc.). Independent of the main - /// transport. - pub fn for_tcp(addr: std::net::SocketAddr, unit_id: u8) -> Self { - Self(Backend::Tcp { addr, unit_id }) + Ok(Box::new(self.0.clone())) } } impl WriterFactory { pub async fn connect(&self) -> Result> { - match &self.0 { - Backend::Tcp { addr, unit_id } => { - let w = ModbusWriter::connect(*addr, *unit_id).await?; - Ok(Box::new(w)) - } - Backend::Actor(h) => Ok(Box::new(h.clone())), - } - } - - /// See [`ReaderFactory::with_unit_id`]. - pub fn with_unit_id(&self, unit_id: u8) -> Self { - let backend = match &self.0 { - Backend::Tcp { addr, .. } => Backend::Tcp { - addr: *addr, - unit_id, - }, - Backend::Actor(h) => Backend::Actor(h.with_slave_id(unit_id)), - }; - Self(backend) - } - - /// See [`ReaderFactory::for_tcp`]. - pub fn for_tcp(addr: std::net::SocketAddr, unit_id: u8) -> Self { - Self(Backend::Tcp { addr, unit_id }) + Ok(Box::new(self.0.clone())) } } // =========================================================================== -// RTU-serial actor cache — shared across reader + writer + slaves on the same port +// Actor cache — one actor per endpoint, shared across reader + writer + slaves // =========================================================================== -/// A cache of RTU-serial actors keyed by the serial port `path`. +/// A cache of Modbus actors keyed by connection identity (`tcp:host:port`, +/// `rtu-tcp:host:port`, `serial:path`). /// -/// **Why it's needed:** `tokio_serial::SerialStream::open` opens the serial port -/// with `TIOCEXCL` (Linux). A second open to the same path is immediately -/// rejected by the OS with `"Unable to acquire exclusive lock on serial port"`. -/// Edge-client -/// has several code paths that each need Modbus access to a slave on a serial -/// port: (a) `build_io` for the main PLC if its transport is RTU-serial, -/// (b) `build_source_reader` per non-PLC slave for telemetry, (c) -/// `build_source_writer` per non-PLC slave for control. Without a cache, -/// (a) + (b) + (c) each `spawn` their own actor trying to open the same port → -/// only one wins the exclusive lock, the rest are stuck in a retry loop forever. +/// **Why it's needed:** several code paths each need Modbus access to the same +/// endpoint — (a) `build_io` for the main PLC, (b) `build_source_reader` per +/// device for telemetry, (c) `build_source_writer` per device for control. +/// Without a cache each would spawn its own actor and open its own connection. +/// That breaks two ways: +/// - **Serial**: `tokio_serial::SerialStream::open` takes `TIOCEXCL` (Linux) — +/// a second open to the same path is rejected ("Unable to acquire exclusive +/// lock on serial port"). +/// - **TCP / RTU-over-TCP**: many PLCs / gateways accept only ONE concurrent +/// socket — a second connection to the same host:port makes writes fail with +/// EINPROGRESS (os error 115). /// -/// This cache ensures one `path` = one actor task = one `open()` call. -/// Different slave-ids on the same RS-485 bus are handled via -/// [`ActorHandle::with_slave_id`] — a cheap sender clone + changing the ID per -/// request. The sequential FIFO of the actor's mpsc channel already guarantees -/// no two Modbus frames overlap on the bus (see the `modbus_actor` module docs). +/// The cache guarantees one endpoint = one actor task = one connection. Reader +/// and writer to the same endpoint share it; different endpoints (e.g. a +/// split-port `:502` read / `:503` write setup) get independent actors. Multiple +/// slave-ids on one RS-485 bus are handled via [`ActorHandle::with_slave_id`] — +/// a cheap sender clone + per-request ID. The actor's FIFO mpsc channel +/// serializes frames so none overlap (see the `modbus_actor` module docs). /// -/// Cheap to clone (internal `Arc`). Usage pattern: create one instance at -/// startup in `main`, then pass `&SerialActorCache` to every call site that -/// needs RTU-serial access. +/// Cheap to clone (internal `Arc`). Create one instance at startup in `main` and +/// pass `&ActorCache` to every call site that needs Modbus access. #[derive(Default, Clone)] -pub struct SerialActorCache { +pub struct ActorCache { inner: Arc>>, } -impl SerialActorCache { +/// Cache key for an endpoint. Reader + writer to the same endpoint collapse to +/// one key (= one shared actor); a different host:port or path stays separate. +fn actor_cache_key(transport: &Transport) -> String { + match transport { + Transport::Tcp { host, port } => format!("tcp:{host}:{port}"), + Transport::RtuOverTcp { host, port } => format!("rtu-tcp:{host}:{port}"), + Transport::RtuSerial { path, .. } => format!("serial:{path}"), + } +} + +impl ActorCache { pub fn new() -> Self { Self::default() } - /// Look up or spawn an actor for `Transport::RtuSerial`. Cache key = `path`. - /// A repeated call with the same path reuses the actor; only the slave_id is - /// updated in the returned handle. + /// Look up or spawn the actor for `transport`. A repeated call with the same + /// endpoint reuses the actor; only the slave_id is updated in the returned + /// handle. /// - /// `timings` is used only on the first spawn for that path. Subsequent calls - /// for the same path reuse the existing actor regardless of the new - /// `timings`. All slaves currently use the fixed default from - /// `device_conn_to_transport`, so there is no inconsistency to warn about. + /// `timings` is used only on the first spawn for that endpoint. Subsequent + /// calls reuse the existing actor regardless of the new `timings` — all + /// callers currently pass the fixed default from `device_conn_to_transport`. /// - /// Error: `Err` only if `transport` is not `RtuSerial` (a programmer bug, - /// not a runtime fault) or if `modbus_actor::spawn` rejects the transport. - /// Serial I/O itself is not attempted here — `spawn` returns quickly and the - /// actor task connects in the background with retry-on-fail. - async fn get_or_spawn_serial( + /// I/O is not attempted here — `spawn` returns quickly and the actor task + /// connects in the background with retry-on-fail. + async fn get_or_spawn( &self, transport: Transport, slave_id: u8, timings: crate::modbus_actor::RtuTimings, ) -> Result { - let path = match &transport { - Transport::RtuSerial { path, .. } => path.clone(), - other => { - return Err(anyhow!( - "SerialActorCache::get_or_spawn_serial called with a non-serial transport \ - ({}) — programmer bug", - other.display() - )); - } - }; + let key = actor_cache_key(&transport); let mut lock = self.inner.lock().await; - if let Some(existing) = lock.get(&path) { - debug!( - path = %path, - slave_id, - "RTU serial actor reused from cache" - ); + if let Some(existing) = lock.get(&key) { + debug!(key = %key, slave_id, "modbus actor reused from cache"); return Ok(existing.with_slave_id(slave_id)); } debug!( - path = %path, + key = %key, slave_id, transport = %transport.display(), - "RTU serial actor: cache miss, spawning new actor" + "modbus actor: cache miss, spawning new actor" ); let handle = crate::modbus_actor::spawn(transport, slave_id, timings) .await - .context("spawn RTU serial actor (via SerialActorCache)")?; - lock.insert(path, handle.clone()); + .context("spawn modbus actor (via ActorCache)")?; + lock.insert(key, handle.clone()); Ok(handle) } - /// Number of RTU-serial actor entries in the cache. Used by tests + can be - /// used for observability in the startup log (number of physical serial - /// ports in use). + /// Number of distinct endpoints (actor tasks) in the cache. Used by tests + + /// can be used for observability in the startup log. #[allow(dead_code)] pub(crate) async fn len(&self) -> usize { self.inner.lock().await.len() @@ -423,134 +203,71 @@ impl SerialActorCache { // Main builder // =========================================================================== -/// Build a factory pair from [`ModbusSettings`]. For RTU mode, this -/// **spawns an actor task** — the RTU process keeps running in the background -/// until both factories drop (all handles drop). For TCP, no I/O happens here, -/// it just packages the descriptor. -/// -/// `serial_cache` is only used if `settings.transport` is `RtuSerial`; it is -/// registered so `build_source_reader` / `build_source_writer` for slaves on the -/// same port can reuse the actor. +/// Build a factory pair from [`ModbusSettings`] for the main PLC. Gets (or +/// spawns) the shared actor for this endpoint via `cache`, so the reader and +/// writer share one connection, and any source device on the same endpoint +/// reuses it too. No I/O happens here — the actor connects in the background. pub async fn build_io( settings: &crate::settings::ModbusSettings, - serial_cache: &SerialActorCache, + cache: &ActorCache, ) -> Result<(ReaderFactory, WriterFactory)> { - match &settings.transport { - Transport::Tcp { host, port } => { - let addr: SocketAddr = format!("{host}:{port}") - .parse() - .map_err(|e| anyhow!("TCP address invalid: {e}"))?; - let backend = Backend::Tcp { - addr, - unit_id: settings.unit_id, - }; - Ok((ReaderFactory(backend.clone()), WriterFactory(backend))) - } - Transport::RtuOverTcp { .. } => { - // RTU-over-TCP is not cached via SerialActorCache: it's not a - // serial port, so the exclusive lock is irrelevant. Spawn one actor - // + share it via a cloned handle (reader & writer use the same - // handle). RTU-serial slaves added via `build_source_reader` / - // `_writer` still go through the cache — different transport, - // different actor. - let handle = crate::modbus_actor::spawn( - settings.transport.clone(), - settings.unit_id, - settings.rtu_timings, - ) - .await?; - Ok(( - ReaderFactory(Backend::Actor(handle.clone())), - WriterFactory(Backend::Actor(handle)), - )) - } - Transport::RtuSerial { .. } => { - let handle = serial_cache - .get_or_spawn_serial( - settings.transport.clone(), - settings.unit_id, - settings.rtu_timings, - ) - .await - .context("build_io: spawn RTU serial actor for the main PLC")?; - Ok(( - ReaderFactory(Backend::Actor(handle.clone())), - WriterFactory(Backend::Actor(handle)), - )) - } - } + let handle = cache + .get_or_spawn( + settings.transport.clone(), + settings.unit_id, + settings.rtu_timings, + ) + .await + .context("build_io: spawn modbus actor for the main PLC")?; + Ok((ReaderFactory(handle.clone()), WriterFactory(handle))) } -/// Build a ReaderFactory for one slave device. Used by telemetry polling. +/// Build a ReaderFactory for one source device. Used by telemetry polling. /// -/// 2 cases: -/// - TCP slave → a TCP factory to the slave host (independent of the PLC). -/// - RTU-serial slave → look up the actor in `serial_cache` (keyed by path). -/// On a cache hit (e.g. the main PLC uses the same serial port, or another -/// slave was registered earlier, or this slave's writer counterpart is already -/// registered), reuse the handle with `with_slave_id`. On a miss, spawn a new -/// actor and cache it so other call sites (the same slave's writer, or other -/// slaves on the same RS-485 bus) reuse it too. +/// Gets (or spawns) the shared actor for the device's endpoint via `cache`, so +/// it's shared with the main PLC / other devices on the same endpoint and with +/// this device's writer counterpart. A weigher (SerialAscii) has no Modbus +/// reader → error. pub(crate) async fn build_source_reader( conn: &crate::shared::DeviceConnection, - serial_cache: &SerialActorCache, + cache: &ActorCache, ) -> anyhow::Result { use crate::shared::DeviceConnection; - match conn { - DeviceConnection::Tcp { host, port, .. } => { - let addr: std::net::SocketAddr = format!("{host}:{port}") - .parse() - .map_err(|e| anyhow::anyhow!("slave tcp addr invalid: {e}"))?; - Ok(ReaderFactory::for_tcp(addr, conn.unit_id())) - } - DeviceConnection::RtuSerial { .. } => { - let (transport, timings) = crate::settings::device_conn_to_transport(conn) - .context("slave device_conn_to_transport (reader)")?; - let handle = serial_cache - .get_or_spawn_serial(transport, conn.unit_id(), timings) - .await - .context("get_or_spawn RTU serial actor for slave reader")?; - Ok(ReaderFactory(Backend::Actor(handle))) - } - DeviceConnection::SerialAscii { .. } => { - anyhow::bail!( - "build_source_reader called for SerialAscii — a weigher has no Modbus reader" - ) - } + if let DeviceConnection::SerialAscii { .. } = conn { + anyhow::bail!( + "build_source_reader called for SerialAscii — a weigher has no Modbus reader" + ); } + let (transport, timings) = crate::settings::device_conn_to_transport(conn) + .context("slave device_conn_to_transport (reader)")?; + let handle = cache + .get_or_spawn(transport, conn.unit_id(), timings) + .await + .context("get_or_spawn modbus actor for source reader")?; + Ok(ReaderFactory(handle)) } -/// Used by control_subscriber. -/// Logic identical to [`build_source_reader`] — look up the actor in -/// `serial_cache` so it's shared with the reader counterpart (if the same slave -/// also has a monitoring entry) and with other slaves on the same port. +/// Build a WriterFactory for one source device. Used by control_subscriber. +/// Logic identical to [`build_source_reader`] — shares the actor with the reader +/// counterpart (if the device also has a monitoring entry) and with other +/// devices on the same endpoint. pub(crate) async fn build_source_writer( conn: &crate::shared::DeviceConnection, - serial_cache: &SerialActorCache, + cache: &ActorCache, ) -> anyhow::Result { use crate::shared::DeviceConnection; - match conn { - DeviceConnection::Tcp { host, port, .. } => { - let addr: std::net::SocketAddr = format!("{host}:{port}") - .parse() - .map_err(|e| anyhow::anyhow!("slave tcp addr invalid: {e}"))?; - Ok(WriterFactory::for_tcp(addr, conn.unit_id())) - } - DeviceConnection::RtuSerial { .. } => { - let (transport, timings) = crate::settings::device_conn_to_transport(conn) - .context("slave device_conn_to_transport (writer)")?; - let handle = serial_cache - .get_or_spawn_serial(transport, conn.unit_id(), timings) - .await - .context("get_or_spawn RTU serial actor for slave writer")?; - Ok(WriterFactory(Backend::Actor(handle))) - } - DeviceConnection::SerialAscii { .. } => { - anyhow::bail!( - "build_source_writer called for SerialAscii — a weigher has no Modbus writer" - ) - } + if let DeviceConnection::SerialAscii { .. } = conn { + anyhow::bail!( + "build_source_writer called for SerialAscii — a weigher has no Modbus writer" + ); } + let (transport, timings) = crate::settings::device_conn_to_transport(conn) + .context("slave device_conn_to_transport (writer)")?; + let handle = cache + .get_or_spawn(transport, conn.unit_id(), timings) + .await + .context("get_or_spawn modbus actor for source writer")?; + Ok(WriterFactory(handle)) } // =========================================================================== @@ -573,23 +290,30 @@ mod tests { } } + fn tcp_transport(host: &str, port: u16) -> Transport { + Transport::Tcp { + host: host.to_string(), + port, + } + } + /// The cache returns the same actor handle (identical channel sender) for /// two calls with the same serial path. The slave_id may differ; the actor /// task stays single. This is the key invariant that prevents the "Unable to /// acquire exclusive lock on serial port" bug — with the cache, one path = /// one `tokio_serial::SerialStream::open`. #[tokio::test] - async fn serial_cache_reuses_actor_for_same_path() { - let cache = SerialActorCache::new(); + async fn cache_reuses_actor_for_same_serial_path() { + let cache = ActorCache::new(); let transport = serial_transport("/tmp/fake-ttyUSB-cache-test-a"); let timings = RtuTimings::default(); let reader_handle = cache - .get_or_spawn_serial(transport.clone(), 1, timings) + .get_or_spawn(transport.clone(), 1, timings) .await .expect("first call should spawn"); let writer_handle = cache - .get_or_spawn_serial(transport.clone(), 2, timings) + .get_or_spawn(transport.clone(), 2, timings) .await .expect("second call should reuse"); @@ -604,12 +328,12 @@ mod tests { /// different physical ports into one actor — that would send frames onto the /// wrong bus. #[tokio::test] - async fn serial_cache_separates_actors_per_path() { - let cache = SerialActorCache::new(); + async fn cache_separates_actors_per_serial_path() { + let cache = ActorCache::new(); let timings = RtuTimings::default(); let h_a = cache - .get_or_spawn_serial( + .get_or_spawn( serial_transport("/tmp/fake-ttyUSB-cache-test-b"), 1, timings, @@ -617,7 +341,7 @@ mod tests { .await .expect("spawn A"); let h_b = cache - .get_or_spawn_serial( + .get_or_spawn( serial_transport("/tmp/fake-ttyUSB-cache-test-c"), 1, timings, @@ -632,21 +356,50 @@ mod tests { assert_eq!(cache.len().await, 2); } - /// Calling the cache with a non-serial transport = programmer bug, it must - /// return Err (not panic, not silently spawn the wrong thing). + /// TCP reader + writer to the same host:port share ONE actor (= one socket). + /// This is the fix for single-session PLCs: a separate reader + writer socket + /// would make the writer fail with EINPROGRESS (os error 115). #[tokio::test] - async fn serial_cache_rejects_non_serial_transport() { - let cache = SerialActorCache::new(); - let result = cache - .get_or_spawn_serial( - Transport::Tcp { - host: "127.0.0.1".to_string(), - port: 502, - }, - 1, - RtuTimings::default(), - ) - .await; - assert!(result.is_err(), "TCP transport must be rejected"); + async fn cache_shares_actor_for_same_tcp_endpoint() { + let cache = ActorCache::new(); + let timings = RtuTimings::default(); + + let reader = cache + .get_or_spawn(tcp_transport("192.168.10.10", 502), 1, timings) + .await + .expect("reader spawn"); + let writer = cache + .get_or_spawn(tcp_transport("192.168.10.10", 502), 1, timings) + .await + .expect("writer reuse"); + + assert!( + reader.same_channel(&writer), + "reader & writer on the same host:port must share one actor/connection" + ); + assert_eq!(cache.len().await, 1); + } + + /// Split-port setup (`:502` read / `:503` write, same host) keeps TWO + /// independent actors — matches PLCs that expose a separate write port. + #[tokio::test] + async fn cache_separates_actors_per_tcp_port() { + let cache = ActorCache::new(); + let timings = RtuTimings::default(); + + let read = cache + .get_or_spawn(tcp_transport("192.168.10.10", 502), 1, timings) + .await + .expect("read spawn"); + let write = cache + .get_or_spawn(tcp_transport("192.168.10.10", 503), 1, timings) + .await + .expect("write spawn"); + + assert!( + !read.same_channel(&write), + "different ports on the same host must have separate actors" + ); + assert_eq!(cache.len().await, 2); } } diff --git a/src/telemetry.rs b/src/telemetry.rs index acd4e1c..200a44c 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -35,9 +35,9 @@ const RECONNECT_BACKOFF: Duration = Duration::from_secs(5); /// The PLC and source devices are polled separately: /// - PLC: `entries` + `holding_count` + `coil_count`, polled via /// `reader_factory` (unit_id from the mapping's primary PLC). -/// - Source (chiller, etc.): each entry in `source_sessions`, polled via a -/// `ReaderFactory` already `.with_unit_id(source.unit_id)`. An error on one -/// source does not affect the PLC poll loop or the other sources. +/// - Source (chiller, etc.): each entry in `source_sessions`, polled via its own +/// `ReaderFactory` (built with the device's unit_id). An error on one source +/// does not affect the PLC poll loop or the other sources. #[derive(Clone)] pub struct TelemetryConfig { pub reader_factory: ReaderFactory, @@ -64,7 +64,7 @@ pub struct SourcePollSession { /// This device's machine id `{location}/{name}` — used for the per-machine /// plc_status topic (`{base}/{machine_id}/plc/status`). pub machine_id: String, - /// ReaderFactory with this source's unit_id (derived via `with_unit_id`). + /// ReaderFactory for this source (built with the device's unit_id). pub reader_factory: ReaderFactory, /// Holding + coil entries — bulk read from address 0. pub entries: Vec, @@ -82,7 +82,7 @@ impl TelemetryConfig { pub async fn from_settings( settings: &Settings, reader_factory: ReaderFactory, - serial_cache: &crate::modbus_client::SerialActorCache, + actor_cache: &crate::modbus_client::ActorCache, ) -> Result { let modbus = settings .modbus @@ -139,7 +139,7 @@ impl TelemetryConfig { continue; } let source_reader_factory = - crate::modbus_client::build_source_reader(&device.connection, serial_cache) + crate::modbus_client::build_source_reader(&device.connection, actor_cache) .await .with_context(|| format!("build source reader for slave '{}'", device_id))?; let input_entries: Vec = source_entries