diff --git a/Cargo.lock b/Cargo.lock index f12209c..c0f684f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2240,7 +2240,6 @@ dependencies = [ "rand 0.10.1", "thiserror", "tokio", - "tokio-serial", "tracing", "ziggurat-ieee-802154", ] diff --git a/crates/ziggurat-server/src/main.rs b/crates/ziggurat-server/src/main.rs index f365de4..28cc0cf 100644 --- a/crates/ziggurat-server/src/main.rs +++ b/crates/ziggurat-server/src/main.rs @@ -5,8 +5,8 @@ use std::time::Duration; use clap::{Parser, ValueEnum}; use futures_util::{SinkExt, StreamExt}; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; -use tokio::net::{TcpListener, UnixListener}; -use tokio::sync::{broadcast, mpsc}; +use tokio::net::{TcpListener, TcpStream, UnixListener}; +use tokio::sync::{Mutex as AsyncMutex, broadcast, mpsc}; use tokio::task::JoinHandle; use tokio_serial::{FlowControl, SerialPortBuilderExt}; use tokio_tungstenite::tungstenite::Message; @@ -22,7 +22,7 @@ use ziggurat_driver::ziggurat_ieee_802154::types::{Eui64, Nwk, PanId}; use ziggurat_phy::{RadioConfig, RadioPhy, Receiver}; use ziggurat_phy_spinel::SpinelPhy; use ziggurat_protocol::{self as proto}; -use ziggurat_spinel::client::SpinelClient; +use ziggurat_spinel::client::{RcpTransport, SpinelClient}; /// Outbound frames a connection can queue before it is considered too slow and /// disconnected. Received frames dominate the traffic; a client that cannot keep up @@ -59,10 +59,11 @@ fn radio_error(e: impl ToString) -> proto::Error { pub struct ZigguratServer { serial: SerialConfig, - /// The radio transport owns the serial port for the lifetime of the process: it is + /// The radio transport owns the RCP link for the lifetime of the process: it is /// opened lazily by the first command that needs it and never reopened, so stack - /// replacement cannot race a straggling port handle (`EBUSY`) - phy: Mutex>>, + /// replacement cannot race a straggling handle — a serial port re-open failing + /// with `EBUSY`, or a second concurrent connection to the same TCP endpoint + phy: AsyncMutex>>, stack: Mutex>>>, started: AtomicBool, notification_tx: broadcast::Sender, @@ -70,14 +71,14 @@ pub struct ZigguratServer { } impl ZigguratServer { - /// The serial port is not opened and the Zigbee stack is not created until a + /// The RCP transport is not opened and the Zigbee stack is not created until a /// client sends a command that needs them. pub fn new(serial: SerialConfig) -> Self { let (notification_tx, _) = broadcast::channel(NOTIFICATION_HUB_DEPTH); Self { serial, - phy: Mutex::new(None), + phy: AsyncMutex::new(None), stack: Mutex::new(None), started: AtomicBool::new(false), notification_tx, @@ -140,21 +141,16 @@ impl ZigguratServer { self.stack.lock().unwrap().clone() } - /// The process-lifetime radio transport, opening the serial port on first use. - fn phy(&self) -> Result, tokio_serial::Error> { - let mut phy = self.phy.lock().unwrap(); + /// The process-lifetime radio transport, opening the RCP link on first use. + async fn phy(&self) -> std::io::Result> { + let mut phy = self.phy.lock().await; if let Some(phy) = &*phy { return Ok(phy.clone()); } - // Without flow control the RCP's UART drops bytes under load, corrupting - // host->RCP frames ("Framing error" + command timeout) - let port = tokio_serial::new(&self.serial.device, self.serial.baudrate) - .flow_control(self.serial.flow_control.into()) - .open_native_async()?; - - let new_phy = Arc::new(SpinelPhy::new(Arc::new(SpinelClient::new(port)))); + let transport = open_transport(&self.serial).await?; + let new_phy = Arc::new(SpinelPhy::new(Arc::new(SpinelClient::new(transport)))); *phy = Some(new_phy.clone()); drop(phy); @@ -552,7 +548,7 @@ impl ZigguratServer { payload: proto::ResetPayload, ) -> Result { if payload.hard { - let phy = self.phy().map_err(radio_error)?; + let phy = self.phy().await.map_err(radio_error)?; phy.reset() .await .map_err(|e| proto::Error::new(proto::Status::RadioError, &e.to_string()))?; @@ -562,7 +558,7 @@ impl ZigguratServer { } async fn handle_get_hw_address(&self) -> Result { - let phy = self.phy().map_err(radio_error)?; + let phy = self.phy().await.map_err(radio_error)?; let ieee = phy .hw_address() .await @@ -574,7 +570,7 @@ impl ZigguratServer { async fn handle_shutdown(&self) -> Result { self.teardown_stack().await; - let phy = self.phy().map_err(radio_error)?; + let phy = self.phy().await.map_err(radio_error)?; phy.set_frame_pending_table(&[], &[]) .await .map_err(|e| proto::Error::new(proto::Status::RadioError, &e.to_string()))?; @@ -594,7 +590,7 @@ impl ZigguratServer { self.teardown_stack().await; tracing::info!("Initializing Zigbee stack with new settings..."); - let phy = self.phy().map_err(radio_error)?; + let phy = self.phy().await.map_err(radio_error)?; let aps_frame_counter = payload.state.aps_frame_counter; let stack = ZigbeeStack::new( @@ -701,7 +697,7 @@ impl ZigguratServer { ) -> Result { // An energy detect is a radio operation, not a network one: it drives the // radio directly and needs no configured stack. - let phy = self.phy().map_err(radio_error)?; + let phy = self.phy().await.map_err(radio_error)?; let duration = Duration::from_millis(u64::from(payload.duration_per_channel_ms)); for channel in payload.channels { @@ -774,7 +770,7 @@ impl ZigguratServer { payload: proto::ChannelPayload, outbound: &mpsc::Sender>, ) -> Result { - let phy = self.phy().map_err(radio_error)?; + let phy = self.phy().await.map_err(radio_error)?; phy.reconfigure(&capture_config(payload.channel)) .await @@ -808,7 +804,7 @@ impl ZigguratServer { &self, payload: proto::ChannelPayload, ) -> Result { - let phy = self.phy().map_err(radio_error)?; + let phy = self.phy().await.map_err(radio_error)?; phy.reconfigure(&capture_config(payload.channel)) .await @@ -842,6 +838,47 @@ pub struct SerialConfig { flow_control: FlowControlMode, } +/// A local network hop should establish in well under this; past it, something (a +/// blackholed host, a firewall silently dropping packets) is wrong. Without a bound, +/// `TcpStream::connect` rides out the kernel's full SYN-retry window (minutes, by +/// default) while `phy()`'s lock is held, wedging every other command behind it. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); + +/// Connects to the RCP per `serial.device`: a `tcp://host:port` (or `socket://host:port`, +/// accepted for parity with pyserial-style tools) address for a raw TCP socket — e.g. a +/// network-attached RCP or a `ser2net`-style serial-to-TCP bridge — or a path for a +/// local serial device. +/// +/// The TCP path is meant for a wired (Ethernet) link. Wi-Fi's latency variance is not +/// supported: a spike can spuriously trip `MAX_CONSECUTIVE_TIMEOUTS` and force a +/// reset-recovery against a perfectly healthy radio. +async fn open_transport(serial: &SerialConfig) -> std::io::Result> { + let tcp_addr = serial + .device + .strip_prefix("tcp://") + .or_else(|| serial.device.strip_prefix("socket://")); + + if let Some(addr) = tcp_addr { + let stream = tokio::time::timeout(CONNECT_TIMEOUT, TcpStream::connect(addr)) + .await + .map_err(|_| { + std::io::Error::new(std::io::ErrorKind::TimedOut, "RCP connect timed out") + })??; + // The Spinel control plane is latency-sensitive request/response traffic in + // small frames; Nagle's algorithm would needlessly delay them. + stream.set_nodelay(true)?; + return Ok(Box::new(stream)); + } + + // Without flow control the RCP's UART drops bytes under load, corrupting + // host->RCP frames ("Framing error" + command timeout) + let port = tokio_serial::new(&serial.device, serial.baudrate) + .flow_control(serial.flow_control.into()) + .open_native_async() + .map_err(std::io::Error::other)?; + Ok(Box::new(port)) +} + /// How the Zigbee API is exposed to clients. #[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] enum ApiMode { @@ -861,15 +898,18 @@ struct Args { #[arg(long, value_enum, default_value_t = ApiMode::Ws)] api: ApiMode, - /// Serial device of the 802.15.4 RCP + /// RCP transport: a serial device path, or `tcp://host:port` for a raw TCP socket. + /// The TCP form is for a wired (Ethernet) link; Wi-Fi is not recommended or + /// supported. #[arg(long)] device: String, - /// Serial baudrate + /// Serial baudrate; ignored for a `tcp://` device #[arg(long, default_value_t = 460_800)] baudrate: u32, - /// Serial flow control; the RCP UART drops bytes under load without it + /// Serial flow control; the RCP UART drops bytes under load without it. Ignored + /// for a `tcp://` device #[arg(long, value_enum, default_value_t = FlowControlMode::Hardware)] flow_control: FlowControlMode, diff --git a/crates/ziggurat-spinel/Cargo.toml b/crates/ziggurat-spinel/Cargo.toml index 3eee68e..564ab84 100644 --- a/crates/ziggurat-spinel/Cargo.toml +++ b/crates/ziggurat-spinel/Cargo.toml @@ -16,7 +16,6 @@ tracing = "0.1" num_enum = "0.7.3" thiserror = "2.0.12" tokio = { version = "1.43.0", features = ["rt", "time", "sync", "io-util"] } -tokio-serial = "5.4" [dev-dependencies] hex-literal = "1.1.0" diff --git a/crates/ziggurat-spinel/src/client.rs b/crates/ziggurat-spinel/src/client.rs index aad7689..4790d92 100644 --- a/crates/ziggurat-spinel/src/client.rs +++ b/crates/ziggurat-spinel/src/client.rs @@ -5,19 +5,26 @@ use crate::{ }; use std::string::String; use thiserror::Error; -use tokio_serial::SerialStream; use ziggurat_ieee_802154::FrameBytes; use ziggurat_ieee_802154::types::Eui64; use std::sync::Arc; use std::sync::Mutex; use std::sync::atomic::{AtomicU32, Ordering}; -use tokio::io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadHalf, WriteHalf}; use tokio::sync::Mutex as AsyncMutex; use tokio::sync::mpsc; use tokio::time::{Duration, timeout}; -/// A local serial link answers in milliseconds; anything beyond this is a failure. +/// A link to the RCP: a serial port or a TCP socket. Boxed so [`SpinelClient`] stays +/// transport-agnostic regardless of which one a caller opens. +pub trait RcpTransport: AsyncRead + AsyncWrite + Send + Unpin + 'static {} +impl RcpTransport for T {} + +/// A local RCP link — serial, or Ethernet to a network-attached RCP — answers in +/// milliseconds; anything beyond this is a failure. Wi-Fi is not supported: its +/// latency variance can spuriously trip `MAX_CONSECUTIVE_TIMEOUTS` below and force a +/// reset-recovery against a perfectly healthy radio. const TIMEOUT: Duration = Duration::from_secs(2); /// Consecutive command timeouts before the RCP is presumed wedged and the @@ -227,17 +234,15 @@ pub enum SpinelError { /// The writer half of the port plus its serialization scratch: frames go out one at a /// time, so two persistent buffers cover every TX without per-frame allocation. -#[derive(Debug)] struct SpinelWriter { - port: WriteHalf, + port: WriteHalf>, frame_scratch: Vec, hdlc_scratch: Vec, } -#[derive(Debug)] pub struct SpinelClient { /// The reader half of the port, owned by the task spawned in `spawn_reader`. - reader: Mutex>>, + reader: Mutex>>>, /// The writer half of the port. The mutex also serializes outbound HDLC writes so /// concurrent commands cannot interleave partial frames inside the byte stream. writer: AsyncMutex, @@ -249,7 +254,7 @@ pub struct SpinelClient { } impl SpinelClient { - pub fn new(port: SerialStream) -> Self { + pub fn new(port: Box) -> Self { let (reader, writer) = tokio::io::split(port); Self {