diff --git a/Cargo.lock b/Cargo.lock index 83465d8..3f036fb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2074,7 +2074,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 daa2528..cf536f8 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; @@ -21,7 +21,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 @@ -61,7 +61,7 @@ pub struct ZigguratServer { /// The radio transport owns the serial port 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>>, + phy: AsyncMutex>>, stack: Mutex>>>, started: AtomicBool, notification_tx: broadcast::Sender, @@ -76,7 +76,7 @@ impl ZigguratServer { Self { serial, - phy: Mutex::new(None), + phy: AsyncMutex::new(None), stack: Mutex::new(None), started: AtomicBool::new(false), notification_tx, @@ -140,20 +140,15 @@ impl ZigguratServer { } /// 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(); + 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); @@ -551,7 +546,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()))?; @@ -561,7 +556,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 @@ -573,7 +568,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()))?; @@ -593,7 +588,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( @@ -700,7 +695,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 { @@ -773,7 +768,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 @@ -807,7 +802,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 @@ -841,6 +836,27 @@ pub struct SerialConfig { flow_control: FlowControlMode, } +/// Connects to the RCP per `serial.device`: a `tcp://host:port` 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. +async fn open_transport(serial: &SerialConfig) -> std::io::Result> { + if let Some(addr) = serial.device.strip_prefix("tcp://") { + let stream = TcpStream::connect(addr).await?; + // 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 { @@ -860,15 +876,16 @@ 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 #[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..5a5e82d 100644 --- a/crates/ziggurat-spinel/src/client.rs +++ b/crates/ziggurat-spinel/src/client.rs @@ -5,18 +5,22 @@ 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 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 serial link answers in milliseconds; anything beyond this is a failure. const TIMEOUT: Duration = Duration::from_secs(2); @@ -227,17 +231,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 +251,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 {