From 736afcb25bb1b9b31e82e9dd66c2567d516b6a67 Mon Sep 17 00:00:00 2001 From: Bernie Innocenti Date: Sun, 5 Jul 2026 23:26:46 +0900 Subject: [PATCH 1/2] serial: bidirectional TCP sink (mode = "tcp") Bridges Paula's serial port to a host TCP listener, like UAE's TCP: serial device (same default port 1234; override with [serial] listen). With an AUX: shell on the Amiga side this gives a remote AmigaDOS console. Co-Authored-By: Claude Fable 5 --- src/config.rs | 15 ++++- src/emulator.rs | 3 + src/serial.rs | 149 ++++++++++++++++++++++++++++++++++++++++++ src/video/launcher.rs | 8 ++- 4 files changed, 173 insertions(+), 2 deletions(-) diff --git a/src/config.rs b/src/config.rs index 0f5d7a7..4c71bca 100644 --- a/src/config.rs +++ b/src/config.rs @@ -241,6 +241,10 @@ pub enum SerialMode { /// Serial in/out is bridged to host MIDI endpoints. Requires a build /// with the `midi` feature; without it, resolving this mode is an error. Midi, + /// Serial in/out is bridged to a host TCP port, like UAE's `TCP:` + /// serial device. With an `AUX:` shell on the Amiga side, a connected + /// client gets a remote AmigaDOS console. + Tcp, } impl SerialMode { @@ -250,6 +254,7 @@ impl SerialMode { Self::Off => "off", Self::Stdout => "stdout", Self::Midi => "midi", + Self::Tcp => "tcp", } } } @@ -263,6 +268,9 @@ pub struct SerialConfig { pub mode: SerialMode, pub midi_out: Option, pub midi_in: Option, + /// TCP listen address for [`SerialMode::Tcp`]; `None` means the + /// default `127.0.0.1:1234` (the port UAE's `TCP:` serial uses). + pub listen: Option, } /// A configured hard-drive image: the host path plus an optional volume-name @@ -1208,6 +1216,9 @@ pub(crate) struct RawSerial { /// Host MIDI input endpoint name (substring match); MIDI mode only. #[serde(skip_serializing_if = "Option::is_none")] pub(crate) midi_in: Option, + /// TCP listen address; tcp mode only. Defaults to 127.0.0.1:1234. + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) listen: Option, } /// A drive image entry in `[ide]`/`[scsi]`. Accepts either a bare path string @@ -1775,6 +1786,7 @@ impl TryFrom for Config { }, midi_out: raw.serial.midi_out.clone(), midi_in: raw.serial.midi_in.clone(), + listen: raw.serial.listen.clone(), }; let ide = IdeConfig { @@ -1981,8 +1993,9 @@ pub(crate) fn parse_serial_mode(s: &str) -> Result { "off" | "none" => Ok(SerialMode::Off), "stdout" | "terminal" => Ok(SerialMode::Stdout), "midi" => Ok(SerialMode::Midi), + "tcp" => Ok(SerialMode::Tcp), _ => Err(anyhow!( - "unknown [serial] mode {:?}: expected \"off\", \"stdout\", or \"midi\"", + "unknown [serial] mode {:?}: expected \"off\", \"stdout\", \"midi\", or \"tcp\"", s )), } diff --git a/src/emulator.rs b/src/emulator.rs index 11d6a07..0e05cb5 100644 --- a/src/emulator.rs +++ b/src/emulator.rs @@ -1754,6 +1754,9 @@ fn build_serial_sink(cfg: &Config) -> Result> SerialMode::Midi => Err(anyhow!( "[serial] mode = \"midi\" needs a build with --features midi" )), + SerialMode::Tcp => Ok(Box::new(crate::serial::TcpSerialSink::listen( + cfg.serial.listen.as_deref().unwrap_or("127.0.0.1:1234"), + )?)), } } diff --git a/src/serial.rs b/src/serial.rs index bf37c76..1d78758 100644 --- a/src/serial.rs +++ b/src/serial.rs @@ -69,6 +69,155 @@ pub trait SerialSink: Send { fn flush(&mut self); } +/// Bidirectional TCP bridge, like the UAE "TCP:" serial device (port 1234 +/// there too): listens on a host port, and the connected client talks to +/// Paula's serial port -- with an +/// `AUX:` shell on the Amiga side, a full remote AmigaDOS console. One +/// client at a time; a new connection replaces a finished one. Output with +/// no client connected is dropped, like an unplugged serial cable. +/// +/// A background thread owns the accept loop and the read half (pushing +/// bytes into a channel), so Paula's idle fast path polls a channel probe, +/// never a socket syscall. +pub struct TcpSerialSink { + rx: std::sync::mpsc::Receiver, + /// Write half of the current client, shared with the acceptor thread + /// (which installs/clears it as clients come and go). + writer: std::sync::Arc>>, + /// Bytes queued in `rx`: the reader thread increments, `read_byte` + /// decrements. Lets `has_pending_input` (`&self`) probe the channel + /// without consuming from it. + buffered: std::sync::Arc, + /// The bound listen address (resolves port 0 to the real port). + local_addr: std::net::SocketAddr, +} + +impl TcpSerialSink { + pub fn listen(addr: &str) -> anyhow::Result { + let listener = std::net::TcpListener::bind(addr) + .map_err(|e| anyhow::anyhow!("[serial] tcp: binding {addr}: {e}"))?; + let local_addr = listener.local_addr()?; + log::info!( + "serial: listening on tcp://{local_addr} (connect with e.g. \"nc {}\" \ + or \"socat -,raw,echo=0 tcp:{}\")", + addr.replace(':', " "), + addr, + ); + let (tx, rx) = std::sync::mpsc::channel(); + let writer = std::sync::Arc::new(std::sync::Mutex::new(None::)); + let acceptor_writer = std::sync::Arc::clone(&writer); + let buffered = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let reader_buffered = std::sync::Arc::clone(&buffered); + std::thread::Builder::new() + .name("serial-tcp".into()) + .spawn(move || loop { + let Ok((stream, peer)) = listener.accept() else { + return; + }; + log::info!("serial: client connected from {peer}"); + let _ = stream.set_nodelay(true); + match stream.try_clone() { + Ok(w) => *acceptor_writer.lock().unwrap() = Some(w), + Err(e) => { + log::warn!("serial: cloning client stream: {e}"); + continue; + } + } + let mut buf = [0u8; 512]; + let mut stream = stream; + loop { + use std::io::Read; + match stream.read(&mut buf) { + Ok(0) | Err(_) => break, + Ok(n) => { + for &b in &buf[..n] { + if tx.send(b).is_err() { + return; + } + reader_buffered.fetch_add(1, std::sync::atomic::Ordering::Release); + } + } + } + } + log::info!("serial: client {peer} disconnected"); + *acceptor_writer.lock().unwrap() = None; + })?; + Ok(Self { + rx, + writer, + buffered, + local_addr, + }) + } + + /// The bound listen address (a port of 0 in the config resolves here). + #[cfg_attr(not(test), allow(dead_code))] + pub fn local_addr(&self) -> std::net::SocketAddr { + self.local_addr + } +} + +impl SerialSink for TcpSerialSink { + fn write_byte(&mut self, b: u8, _at_cck: u64) { + let mut guard = self.writer.lock().unwrap(); + if let Some(w) = guard.as_mut() { + if w.write_all(&[b]).is_err() { + *guard = None; + } + } + } + + fn read_byte(&mut self) -> Option { + let b = self.rx.try_recv().ok(); + if b.is_some() { + self.buffered + .fetch_sub(1, std::sync::atomic::Ordering::Release); + } + b + } + + fn has_pending_input(&self) -> bool { + self.buffered.load(std::sync::atomic::Ordering::Acquire) > 0 + } + + fn flush(&mut self) {} +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Read; + + #[test] + fn tcp_sink_round_trips_input_and_output() { + let mut sink = TcpSerialSink::listen("127.0.0.1:0").unwrap(); + let mut client = std::net::TcpStream::connect(sink.local_addr()).unwrap(); + client.write_all(b"ab").unwrap(); + // Input: wait for the reader thread to stage the bytes. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while !sink.has_pending_input() { + assert!(std::time::Instant::now() < deadline, "input never arrived"); + std::thread::yield_now(); + } + assert_eq!(sink.read_byte(), Some(b'a')); + while !sink.has_pending_input() { + assert!(std::time::Instant::now() < deadline, "second byte lost"); + std::thread::yield_now(); + } + assert_eq!(sink.read_byte(), Some(b'b')); + assert!(!sink.has_pending_input()); + + // Output: bytes written to the sink arrive at the client. + sink.write_byte(b'x', 0); + let mut got = [0u8; 1]; + client + .set_read_timeout(Some(std::time::Duration::from_secs(5))) + .unwrap(); + client.read_exact(&mut got).unwrap(); + assert_eq!(&got, b"x"); + } +} + /// Inert sink: discards output and never produces input. Placeholder used /// where a `Box` must exist before the host wires the real /// one (serde-skipped fields during save-state deserialization). diff --git a/src/video/launcher.rs b/src/video/launcher.rs index fe17e5d..3f123cb 100644 --- a/src/video/launcher.rs +++ b/src/video/launcher.rs @@ -498,7 +498,12 @@ const WARPS: [WarpSpeed; 5] = [ const JOYSTICK_MODES: [JoystickInputMode; 2] = [JoystickInputMode::Gamepad, JoystickInputMode::Keyboard]; #[cfg(feature = "midi")] -const SERIAL_MODES: [SerialMode; 3] = [SerialMode::Off, SerialMode::Stdout, SerialMode::Midi]; +const SERIAL_MODES: [SerialMode; 4] = [ + SerialMode::Off, + SerialMode::Stdout, + SerialMode::Midi, + SerialMode::Tcp, +]; /// Stereo-separation presets the picker steps through (percent), ascending so /// the right arrow steps up (wrapping 100 -> 0) and the left arrow steps down. @@ -1209,6 +1214,7 @@ impl MachineSetup { SerialMode::Off => "Off".to_string(), SerialMode::Stdout => "Stdout".to_string(), SerialMode::Midi => "MIDI".to_string(), + SerialMode::Tcp => "TCP".to_string(), }, #[cfg(feature = "midi")] F::MidiOut => self.midi_out.clone().unwrap_or_else(|| "None".to_string()), From 8a068c6596036a82e2c444efe5e029f815496c48 Mon Sep 17 00:00:00 2001 From: Bernie Innocenti Date: Mon, 6 Jul 2026 16:23:39 +0900 Subject: [PATCH 2/2] serial: address Copilot review of the TCP sink - buffered: AtomicIsize so a read that races ahead of the reader thread's fetch_add dips to a -1 debt instead of wrapping usize. - nc hint: format local_addr ip()/port() separately, IPv6-safe. - launcher: carry [serial] listen through MachineSetup so a save no longer drops the override; add a round-trip test. Co-Authored-By: Claude Fable 5 --- src/serial.rs | 90 +++++++++++++++++++++++-------------------- src/video/launcher.rs | 18 +++++++++ 2 files changed, 66 insertions(+), 42 deletions(-) diff --git a/src/serial.rs b/src/serial.rs index 1d78758..8af33c9 100644 --- a/src/serial.rs +++ b/src/serial.rs @@ -86,8 +86,11 @@ pub struct TcpSerialSink { writer: std::sync::Arc>>, /// Bytes queued in `rx`: the reader thread increments, `read_byte` /// decrements. Lets `has_pending_input` (`&self`) probe the channel - /// without consuming from it. - buffered: std::sync::Arc, + /// without consuming from it. Signed so a `read_byte` that consumes a + /// just-sent byte before the reader thread's matching `fetch_add` lands + /// dips to a transient -1 "debt" instead of wrapping to a stuck-huge + /// unsigned value. + buffered: std::sync::Arc, /// The bound listen address (resolves port 0 to the real port). local_addr: std::net::SocketAddr, } @@ -97,16 +100,19 @@ impl TcpSerialSink { let listener = std::net::TcpListener::bind(addr) .map_err(|e| anyhow::anyhow!("[serial] tcp: binding {addr}: {e}"))?; let local_addr = listener.local_addr()?; + // `nc` takes host and port as separate args; formatting the + // SocketAddr's ip()/port() keeps the hint correct for IPv6 (whose + // literal contains colons that a `:`->` ` replace would mangle). log::info!( - "serial: listening on tcp://{local_addr} (connect with e.g. \"nc {}\" \ - or \"socat -,raw,echo=0 tcp:{}\")", - addr.replace(':', " "), - addr, + "serial: listening on tcp://{local_addr} (connect with e.g. \"nc {} {}\" \ + or \"socat -,raw,echo=0 tcp:{local_addr}\")", + local_addr.ip(), + local_addr.port(), ); let (tx, rx) = std::sync::mpsc::channel(); let writer = std::sync::Arc::new(std::sync::Mutex::new(None::)); let acceptor_writer = std::sync::Arc::clone(&writer); - let buffered = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let buffered = std::sync::Arc::new(std::sync::atomic::AtomicIsize::new(0)); let reader_buffered = std::sync::Arc::clone(&buffered); std::thread::Builder::new() .name("serial-tcp".into()) @@ -183,41 +189,6 @@ impl SerialSink for TcpSerialSink { fn flush(&mut self) {} } -#[cfg(test)] -mod tests { - use super::*; - use std::io::Read; - - #[test] - fn tcp_sink_round_trips_input_and_output() { - let mut sink = TcpSerialSink::listen("127.0.0.1:0").unwrap(); - let mut client = std::net::TcpStream::connect(sink.local_addr()).unwrap(); - client.write_all(b"ab").unwrap(); - // Input: wait for the reader thread to stage the bytes. - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); - while !sink.has_pending_input() { - assert!(std::time::Instant::now() < deadline, "input never arrived"); - std::thread::yield_now(); - } - assert_eq!(sink.read_byte(), Some(b'a')); - while !sink.has_pending_input() { - assert!(std::time::Instant::now() < deadline, "second byte lost"); - std::thread::yield_now(); - } - assert_eq!(sink.read_byte(), Some(b'b')); - assert!(!sink.has_pending_input()); - - // Output: bytes written to the sink arrive at the client. - sink.write_byte(b'x', 0); - let mut got = [0u8; 1]; - client - .set_read_timeout(Some(std::time::Duration::from_secs(5))) - .unwrap(); - client.read_exact(&mut got).unwrap(); - assert_eq!(&got, b"x"); - } -} - /// Inert sink: discards output and never produces input. Placeholder used /// where a `Box` must exist before the host wires the real /// one (serde-skipped fields during save-state deserialization). @@ -267,3 +238,38 @@ impl SerialSink for StdoutSink { } } } + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Read; + + #[test] + fn tcp_sink_round_trips_input_and_output() { + let mut sink = TcpSerialSink::listen("127.0.0.1:0").unwrap(); + let mut client = std::net::TcpStream::connect(sink.local_addr()).unwrap(); + client.write_all(b"ab").unwrap(); + // Input: wait for the reader thread to stage the bytes. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while !sink.has_pending_input() { + assert!(std::time::Instant::now() < deadline, "input never arrived"); + std::thread::yield_now(); + } + assert_eq!(sink.read_byte(), Some(b'a')); + while !sink.has_pending_input() { + assert!(std::time::Instant::now() < deadline, "second byte lost"); + std::thread::yield_now(); + } + assert_eq!(sink.read_byte(), Some(b'b')); + assert!(!sink.has_pending_input()); + + // Output: bytes written to the sink arrive at the client. + sink.write_byte(b'x', 0); + let mut got = [0u8; 1]; + client + .set_read_timeout(Some(std::time::Duration::from_secs(5))) + .unwrap(); + client.read_exact(&mut got).unwrap(); + assert_eq!(&got, b"x"); + } +} diff --git a/src/video/launcher.rs b/src/video/launcher.rs index 3f123cb..e95ad12 100644 --- a/src/video/launcher.rs +++ b/src/video/launcher.rs @@ -565,6 +565,9 @@ pub struct MachineSetup { serial_mode: SerialMode, midi_out: Option, midi_in: Option, + /// TCP listen address for `mode = "tcp"`; carried so the override + /// round-trips through a launcher save even though no tab edits it. + serial_listen: Option, /// Host endpoints for the device pickers, read once when this setup is /// built so a fresh config screen sees currently-connected devices. #[cfg(feature = "midi")] @@ -658,6 +661,7 @@ impl MachineSetup { serial_mode: cfg.serial.mode, midi_out: cfg.serial.midi_out.clone(), midi_in: cfg.serial.midi_in.clone(), + serial_listen: cfg.serial.listen.clone(), // Left empty here so config construction stays side-effect free; the // config screen fills it via refresh_midi_endpoints on open. #[cfg(feature = "midi")] @@ -886,6 +890,7 @@ impl MachineSetup { } raw.serial.midi_out = self.midi_out.clone(); raw.serial.midi_in = self.midi_in.clone(); + raw.serial.listen = self.serial_listen.clone(); // The Audio output picker is one of default / a named device / Disabled. // A named device sets output_device; Disabled sets output_enabled=false // (the resolved default is true, so it is omitted otherwise). @@ -2013,6 +2018,19 @@ mod tests { assert_eq!(back.agnus, Some(AgnusRevision::Ocs)); } + #[test] + fn serial_tcp_listen_round_trips_through_raw() { + // A launcher save must not drop the [serial] listen override, which + // no tab edits (regression: it was absent from MachineSetup/to_raw). + let mut raw = RawConfig::default(); + raw.serial.mode = Some("tcp".into()); + raw.serial.listen = Some("0.0.0.0:2323".into()); + let setup = MachineSetup::from_raw(&raw).unwrap(); + assert_eq!(setup.serial_listen.as_deref(), Some("0.0.0.0:2323")); + let back = setup.to_raw(); + assert_eq!(back.serial.listen.as_deref(), Some("0.0.0.0:2323")); + } + #[test] fn joystick_input_mode_round_trips_through_raw() { let mut s = MachineSetup::default();