pamoja_hal/
port.rs

1//! One serial port, shared by a program and every driver on it.
2//!
3//! A UART carries bytes one character at a time and knows nothing of messages: a framing such
4//! as SLIP or COBS from `pamoja-serial`, or the silences of Modbus RTU, is what marks where one
5//! ends. [`SerialPort`] is a handle to one port that clones cheaply, as
6//! [`I2cBus`](crate::bus::I2cBus) is to one bus, so a program, a thread that reads, and a
7//! protocol client can all hold the same port. Behind the handle is one of five things:
8//!
9//! - The kernel's serial device, opened raw with [`SerialPort::open`]: `/dev/serial0` for a
10//!   Raspberry Pi's own UART, `/dev/ttyUSB0` or `/dev/ttyACM0` for a USB adapter. Only Linux
11//!   has one here.
12//! - A line looped back on itself, [`SerialPort::looped`]: every byte written is read back,
13//!   as with TX wired to RX.
14//! - One end of a null-modem pair, [`SerialPort::pair`]: what one end writes, the other reads.
15//! - A simulated device, [`SerialPort::simulated`], that takes every write and answers with
16//!   bytes of its own: anything that implements [`Peer`], such as a Modbus server.
17//! - A script, [`SerialPort::scripted`], of the writes a driver is expected to make and the
18//!   bytes the far end sends back.
19//!
20//! A read waits for bytes up to a timeout. On the kernel's device the process waits. On every
21//! other port nothing waits, since nothing else is coming, and the time a read would have
22//! waited is added to [`SerialPort::waited_micros`], so a test of a device that never answers
23//! runs at once and still says how long a real one would have taken.
24//!
25//! # Examples
26//!
27//! ```
28//! use std::time::Duration;
29//!
30//! use pamoja_hal::port::{SerialPort, Settings};
31//!
32//! // Two ends of one line: a gateway and the node it listens to.
33//! let (gateway, node) = SerialPort::pair(Settings::new(115_200));
34//! node.write(b"t=21.5")?;
35//!
36//! let mut buffer = [0u8; 16];
37//! let got = gateway.read(&mut buffer, Duration::from_millis(100))?;
38//! assert_eq!(&buffer[..got], b"t=21.5");
39//!
40//! // Nothing more is coming: the read gives up after its timeout, which is counted, not slept.
41//! assert_eq!(gateway.read(&mut buffer, Duration::from_millis(100))?, 0);
42//! assert_eq!(gateway.waited_micros(), 100_000);
43//! # Ok::<(), pamoja_hal::port::PortError>(())
44//! ```
45
46use std::collections::VecDeque;
47use std::fmt;
48use std::path::{Path, PathBuf};
49use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
50use std::time::Duration;
51
52/// The parity bit each character carries.
53#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
54pub enum Parity {
55    /// No parity bit.
56    None,
57    /// A bit that makes the count of ones even, which Modbus RTU asks for by default.
58    Even,
59    /// A bit that makes the count of ones odd.
60    Odd,
61}
62
63/// How many stop bits end each character.
64#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
65pub enum StopBits {
66    /// One stop bit.
67    One,
68    /// Two stop bits.
69    Two,
70}
71
72/// A port's speed and character format: eight data bits, with the parity and stop bits given.
73///
74/// # Examples
75///
76/// ```
77/// use pamoja_hal::port::{Parity, Settings};
78///
79/// // A Modbus RTU meter at 9600 baud, with the even parity the protocol defaults to: one
80/// // stop bit, and eleven bits a character.
81/// let meter = Settings::new(9_600).with_parity(Parity::Even);
82/// assert_eq!(meter.to_string(), "9600 8E1");
83/// assert_eq!(meter.bits_per_character(), 11);
84/// assert_eq!(meter.character_nanos(), 1_145_834);
85/// ```
86#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
87pub struct Settings {
88    /// The speed, in bits a second.
89    pub baud: u32,
90    /// The parity bit each character carries.
91    pub parity: Parity,
92    /// How many stop bits end each character.
93    pub stop_bits: StopBits,
94}
95
96impl Settings {
97    /// Eight data bits, no parity, and one stop bit, the format most devices start in.
98    ///
99    /// # Arguments
100    ///
101    /// * `baud` - the speed, in bits a second.
102    ///
103    /// # Returns
104    ///
105    /// The settings.
106    pub const fn new(baud: u32) -> Settings {
107        Settings {
108            baud,
109            parity: Parity::None,
110            stop_bits: StopBits::One,
111        }
112    }
113
114    /// Sets the parity bit.
115    ///
116    /// # Arguments
117    ///
118    /// * `parity` - the parity each character carries.
119    ///
120    /// # Returns
121    ///
122    /// The settings.
123    pub const fn with_parity(mut self, parity: Parity) -> Settings {
124        self.parity = parity;
125        self
126    }
127
128    /// Sets the stop bits.
129    ///
130    /// # Arguments
131    ///
132    /// * `stop_bits` - how many stop bits end each character.
133    ///
134    /// # Returns
135    ///
136    /// The settings.
137    pub const fn with_stop_bits(mut self, stop_bits: StopBits) -> Settings {
138        self.stop_bits = stop_bits;
139        self
140    }
141
142    /// Returns the bits one character takes on the wire: a start bit, eight data bits, the
143    /// parity bit if there is one, and the stop bits.
144    pub const fn bits_per_character(&self) -> u32 {
145        let parity = match self.parity {
146            Parity::None => 0,
147            Parity::Even | Parity::Odd => 1,
148        };
149        let stop = match self.stop_bits {
150            StopBits::One => 1,
151            StopBits::Two => 2,
152        };
153        1 + 8 + parity + stop
154    }
155
156    /// Returns how long one character takes on the wire, in nanoseconds, rounded up.
157    pub const fn character_nanos(&self) -> u64 {
158        let baud = if self.baud == 0 { 1 } else { self.baud as u64 };
159        (self.bits_per_character() as u64 * 1_000_000_000).div_ceil(baud)
160    }
161
162    /// Returns how long a run of bytes takes on the wire, sent back to back, in microseconds,
163    /// rounded up.
164    ///
165    /// # Arguments
166    ///
167    /// * `bytes` - how many bytes.
168    ///
169    /// # Returns
170    ///
171    /// The time on the wire.
172    pub const fn transfer_micros(&self, bytes: usize) -> u64 {
173        (self.character_nanos() * bytes as u64).div_ceil(1_000)
174    }
175}
176
177impl fmt::Display for Settings {
178    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
179        let parity = match self.parity {
180            Parity::None => 'N',
181            Parity::Even => 'E',
182            Parity::Odd => 'O',
183        };
184        let stop = match self.stop_bits {
185            StopBits::One => 1,
186            StopBits::Two => 2,
187        };
188        write!(f, "{} 8{parity}{stop}", self.baud)
189    }
190}
191
192/// What is on the other end of a [`SerialPort`].
193#[derive(Clone, Copy, Debug, PartialEq, Eq)]
194pub enum PortKind {
195    /// The kernel's serial device, with a real line on the other end.
196    Device,
197    /// The port's own output, looped back to its input.
198    Looped,
199    /// The other end of a null-modem pair.
200    Paired,
201    /// A simulated device that answers each write.
202    Simulated,
203    /// A script of the writes a driver is expected to make.
204    Scripted,
205}
206
207/// A device at the far end of a simulated serial line.
208///
209/// Every write to a [`SerialPort::simulated`] port reaches [`Peer::receive`], and whatever it
210/// returns is what the device sends back, waiting for the next read. A program that wants to
211/// look at the device afterward shares it: `Arc<Mutex<P>>` is a peer whenever `P` is.
212///
213/// # Examples
214///
215/// ```
216/// use std::sync::{Arc, Mutex};
217/// use std::time::Duration;
218///
219/// use pamoja_hal::port::{Peer, SerialPort, Settings};
220///
221/// /// A device that answers every write with the same bytes, upper-cased.
222/// #[derive(Default)]
223/// struct Shouter {
224///     heard: usize,
225/// }
226///
227/// impl Peer for Shouter {
228///     fn receive(&mut self, bytes: &[u8]) -> Vec<u8> {
229///         self.heard += bytes.len();
230///         bytes.to_ascii_uppercase()
231///     }
232/// }
233///
234/// let device = Arc::new(Mutex::new(Shouter::default()));
235/// let port = SerialPort::simulated(Settings::new(9_600), Arc::clone(&device));
236/// port.write(b"ping")?;
237///
238/// let mut reply = [0u8; 8];
239/// let got = port.read(&mut reply, Duration::from_millis(50))?;
240/// assert_eq!(&reply[..got], b"PING");
241/// assert_eq!(device.lock().unwrap().heard, 4);
242/// # Ok::<(), pamoja_hal::port::PortError>(())
243/// ```
244pub trait Peer: Send {
245    /// Takes the bytes the port wrote and returns the bytes the device sends back.
246    ///
247    /// # Arguments
248    ///
249    /// * `bytes` - one write's bytes.
250    ///
251    /// # Returns
252    ///
253    /// The device's answer, empty when it says nothing.
254    fn receive(&mut self, bytes: &[u8]) -> Vec<u8>;
255}
256
257impl<P: Peer> Peer for Arc<Mutex<P>> {
258    fn receive(&mut self, bytes: &[u8]) -> Vec<u8> {
259        self.lock()
260            .unwrap_or_else(PoisonError::into_inner)
261            .receive(bytes)
262    }
263}
264
265/// One step of a [`SerialPort::scripted`] port.
266#[derive(Clone, Debug, PartialEq, Eq)]
267pub enum PortStep {
268    /// Bytes the program is expected to write, in one call.
269    Write(Vec<u8>),
270    /// Bytes the far end sends, readable once every step before this one has happened.
271    Read(Vec<u8>),
272}
273
274impl PortStep {
275    /// A write the program is expected to make.
276    ///
277    /// # Arguments
278    ///
279    /// * `bytes` - the bytes of the one write.
280    ///
281    /// # Returns
282    ///
283    /// The step.
284    pub fn write(bytes: impl Into<Vec<u8>>) -> PortStep {
285        PortStep::Write(bytes.into())
286    }
287
288    /// Bytes the far end sends.
289    ///
290    /// # Arguments
291    ///
292    /// * `bytes` - what arrives.
293    ///
294    /// # Returns
295    ///
296    /// The step.
297    pub fn read(bytes: impl Into<Vec<u8>>) -> PortStep {
298        PortStep::Read(bytes.into())
299    }
300}
301
302/// Why a serial device could not be opened.
303#[derive(Clone, Debug, PartialEq, Eq)]
304pub enum OpenError {
305    /// This platform has no serial device this crate opens, or this build left out the
306    /// `linux` feature that opens one.
307    Unsupported,
308    /// The device file could not be opened or set up as a raw byte pipe.
309    Device {
310        /// The device file.
311        path: PathBuf,
312        /// Why, in the kernel's words.
313        reason: String,
314    },
315}
316
317impl fmt::Display for OpenError {
318    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
319        match self {
320            OpenError::Unsupported => f.write_str(
321                "a serial device is opened through the kernel's terminal interface, which only Linux has here",
322            ),
323            OpenError::Device { path, reason } => write!(f, "{}: {reason}", path.display()),
324        }
325    }
326}
327
328impl std::error::Error for OpenError {}
329
330/// Why a read or a write on a [`SerialPort`] failed.
331#[derive(Clone, Debug, PartialEq, Eq)]
332pub enum PortError {
333    /// A scripted port was written something other than what its script expected next.
334    Unexpected {
335        /// The write the script expected, or none when the script had no writes left.
336        expected: Option<Vec<u8>>,
337        /// What was written.
338        written: Vec<u8>,
339    },
340    /// The kernel's device failed.
341    Device {
342        /// The device file and the kernel's reason.
343        reason: String,
344    },
345}
346
347impl fmt::Display for PortError {
348    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
349        match self {
350            PortError::Unexpected {
351                expected: Some(expected),
352                written,
353            } => write!(
354                f,
355                "the script expected {} to be written, not {}",
356                hex(expected),
357                hex(written)
358            ),
359            PortError::Unexpected {
360                expected: None,
361                written,
362            } => write!(
363                f,
364                "the script expected no more writes, but {} was written",
365                hex(written)
366            ),
367            PortError::Device { reason } => f.write_str(reason),
368        }
369    }
370}
371
372impl std::error::Error for PortError {}
373
374fn hex(bytes: &[u8]) -> String {
375    let mut text = String::with_capacity(bytes.len() * 3);
376    for (index, byte) in bytes.iter().enumerate() {
377        if index > 0 {
378            text.push(' ');
379        }
380        text.push_str(&format!("{byte:02x}"));
381    }
382    text
383}
384
385/// One serial port, reached by every clone of this handle.
386///
387/// # Examples
388///
389/// ```
390/// use std::time::Duration;
391///
392/// use pamoja_hal::port::{PortKind, SerialPort, Settings};
393///
394/// // TX wired to RX: what goes out comes straight back.
395/// let port = SerialPort::looped(Settings::new(115_200));
396/// port.write(b"PING\r\n")?;
397///
398/// let mut echo = [0u8; 6];
399/// assert_eq!(port.read(&mut echo, Duration::from_millis(10))?, 6);
400/// assert_eq!(&echo, b"PING\r\n");
401/// assert_eq!(port.kind(), PortKind::Looped);
402/// assert_eq!((port.written(), port.received()), (6, 6));
403/// # Ok::<(), pamoja_hal::port::PortError>(())
404/// ```
405#[derive(Clone)]
406pub struct SerialPort {
407    shared: Arc<Mutex<Shared>>,
408}
409
410struct Shared {
411    backend: Backend,
412    settings: Settings,
413    written: usize,
414    received: usize,
415    waited_ns: u64,
416}
417
418type Line = Arc<Mutex<VecDeque<u8>>>;
419
420enum Backend {
421    #[cfg(all(feature = "linux", target_os = "linux"))]
422    Device {
423        path: PathBuf,
424        file: Arc<std::fs::File>,
425    },
426    Looped(VecDeque<u8>),
427    Paired {
428        inbox: Line,
429        outbox: Line,
430    },
431    Simulated {
432        peer: Box<dyn Peer>,
433        inbox: VecDeque<u8>,
434    },
435    Scripted {
436        steps: VecDeque<PortStep>,
437        inbox: VecDeque<u8>,
438    },
439}
440
441impl Backend {
442    fn kind(&self) -> PortKind {
443        match self {
444            #[cfg(all(feature = "linux", target_os = "linux"))]
445            Backend::Device { .. } => PortKind::Device,
446            Backend::Looped(_) => PortKind::Looped,
447            Backend::Paired { .. } => PortKind::Paired,
448            Backend::Simulated { .. } => PortKind::Simulated,
449            Backend::Scripted { .. } => PortKind::Scripted,
450        }
451    }
452}
453
454fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
455    mutex.lock().unwrap_or_else(PoisonError::into_inner)
456}
457
458fn take(queue: &mut VecDeque<u8>, buffer: &mut [u8]) -> usize {
459    let count = buffer.len().min(queue.len());
460    for (slot, byte) in buffer.iter_mut().zip(queue.drain(..count)) {
461        *slot = byte;
462    }
463    count
464}
465
466fn nanos(duration: Duration) -> u64 {
467    u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX)
468}
469
470/// Moves the reads a script has reached into the inbox.
471fn release(steps: &mut VecDeque<PortStep>, inbox: &mut VecDeque<u8>) {
472    while let Some(PortStep::Read(bytes)) = steps.front() {
473        inbox.extend(bytes);
474        steps.pop_front();
475    }
476}
477
478impl SerialPort {
479    fn with(backend: Backend, settings: Settings) -> SerialPort {
480        SerialPort {
481            shared: Arc::new(Mutex::new(Shared {
482                backend,
483                settings,
484                written: 0,
485                received: 0,
486                waited_ns: 0,
487            })),
488        }
489    }
490
491    /// Opens the kernel's serial device, raw: no echo, no line editing, no translation, just
492    /// bytes at the given speed and character format.
493    ///
494    /// Whatever the device received before it was opened is dropped.
495    ///
496    /// # Arguments
497    ///
498    /// * `path` - the device file: `/dev/serial0` for a Raspberry Pi's own UART,
499    ///   `/dev/ttyUSB0` or `/dev/ttyACM0` for a USB adapter.
500    /// * `settings` - the speed, one of the standard rates from 1200 to 921600, and the
501    ///   character format.
502    ///
503    /// # Returns
504    ///
505    /// The port, with the real line on the other end.
506    ///
507    /// # Errors
508    ///
509    /// [`OpenError::Unsupported`] anywhere but Linux, or in a build without the `linux`
510    /// feature, and [`OpenError::Device`] when the file cannot be opened, the rate is not a
511    /// standard one, or the terminal settings cannot be applied.
512    ///
513    /// # Examples
514    ///
515    /// ```no_run
516    /// use pamoja_hal::port::{SerialPort, Settings};
517    ///
518    /// let port = SerialPort::open("/dev/serial0", Settings::new(115_200))?;
519    /// # Ok::<(), pamoja_hal::port::OpenError>(())
520    /// ```
521    pub fn open(path: impl AsRef<Path>, settings: Settings) -> Result<SerialPort, OpenError> {
522        #[cfg(all(feature = "linux", target_os = "linux"))]
523        {
524            let path = path.as_ref().to_path_buf();
525            let file = device::open(&path, settings).map_err(|error| OpenError::Device {
526                path: path.clone(),
527                reason: error.to_string(),
528            })?;
529            Ok(SerialPort::with(
530                Backend::Device {
531                    path,
532                    file: Arc::new(file),
533                },
534                settings,
535            ))
536        }
537        #[cfg(not(all(feature = "linux", target_os = "linux")))]
538        {
539            let _ = (path, settings);
540            Err(OpenError::Unsupported)
541        }
542    }
543
544    /// A line looped back on itself: every byte written is waiting to be read, as with TX
545    /// wired to RX.
546    ///
547    /// # Arguments
548    ///
549    /// * `settings` - the speed and character format the line runs at.
550    ///
551    /// # Returns
552    ///
553    /// The port.
554    pub fn looped(settings: Settings) -> SerialPort {
555        SerialPort::with(Backend::Looped(VecDeque::new()), settings)
556    }
557
558    /// The two ends of a null-modem pair: what one end writes, the other reads.
559    ///
560    /// # Arguments
561    ///
562    /// * `settings` - the speed and character format both ends run at.
563    ///
564    /// # Returns
565    ///
566    /// The two ends.
567    pub fn pair(settings: Settings) -> (SerialPort, SerialPort) {
568        let one: Line = Arc::default();
569        let other: Line = Arc::default();
570        (
571            SerialPort::with(
572                Backend::Paired {
573                    inbox: Arc::clone(&one),
574                    outbox: Arc::clone(&other),
575                },
576                settings,
577            ),
578            SerialPort::with(
579                Backend::Paired {
580                    inbox: other,
581                    outbox: one,
582                },
583                settings,
584            ),
585        )
586    }
587
588    /// A simulated device on the other end of the line, answering each write.
589    ///
590    /// # Arguments
591    ///
592    /// * `settings` - the speed and character format the line runs at.
593    /// * `peer` - the device.
594    ///
595    /// # Returns
596    ///
597    /// The port.
598    pub fn simulated(settings: Settings, peer: impl Peer + 'static) -> SerialPort {
599        SerialPort::with(
600            Backend::Simulated {
601                peer: Box::new(peer),
602                inbox: VecDeque::new(),
603            },
604            settings,
605        )
606    }
607
608    /// A port that plays a script: each write has to be the one the script expects next, and
609    /// the bytes the far end sends become readable as the script reaches them.
610    ///
611    /// # Arguments
612    ///
613    /// * `settings` - the speed and character format the line runs at.
614    /// * `steps` - the writes and reads, in order. Reads at the start are readable at once.
615    ///
616    /// # Returns
617    ///
618    /// The port.
619    ///
620    /// # Examples
621    ///
622    /// ```
623    /// use std::time::Duration;
624    ///
625    /// use pamoja_hal::port::{PortStep, SerialPort, Settings};
626    ///
627    /// let port = SerialPort::scripted(
628    ///     Settings::new(9_600),
629    ///     [PortStep::write(*b"?"), PortStep::read(*b"42")],
630    /// );
631    /// port.write(b"?")?;
632    /// let mut answer = [0u8; 2];
633    /// port.read(&mut answer, Duration::from_millis(10))?;
634    /// assert_eq!(&answer, b"42");
635    /// assert_eq!(port.remaining(), Some(0));
636    /// assert!(port.write(b"?").is_err(), "the script is done");
637    /// # Ok::<(), pamoja_hal::port::PortError>(())
638    /// ```
639    pub fn scripted(settings: Settings, steps: impl IntoIterator<Item = PortStep>) -> SerialPort {
640        let mut steps: VecDeque<PortStep> = steps.into_iter().collect();
641        let mut inbox = VecDeque::new();
642        release(&mut steps, &mut inbox);
643        SerialPort::with(Backend::Scripted { steps, inbox }, settings)
644    }
645
646    /// Returns the speed and character format the port runs at.
647    pub fn settings(&self) -> Settings {
648        lock(&self.shared).settings
649    }
650
651    /// Returns what is on the other end.
652    pub fn kind(&self) -> PortKind {
653        lock(&self.shared).backend.kind()
654    }
655
656    /// Writes bytes, and on the kernel's device waits until they have left the UART.
657    ///
658    /// # Arguments
659    ///
660    /// * `bytes` - the bytes, in order.
661    ///
662    /// # Errors
663    ///
664    /// [`PortError::Unexpected`] when a scripted port is written something its script does
665    /// not expect next, and [`PortError::Device`] when the kernel's device fails.
666    pub fn write(&self, bytes: &[u8]) -> Result<(), PortError> {
667        let mut shared = lock(&self.shared);
668        match &mut shared.backend {
669            #[cfg(all(feature = "linux", target_os = "linux"))]
670            Backend::Device { path, file } => {
671                let (path, file) = (path.clone(), Arc::clone(file));
672                drop(shared);
673                device::write(&file, bytes).map_err(|error| PortError::Device {
674                    reason: format!("{}: {error}", path.display()),
675                })?;
676                shared = lock(&self.shared);
677            }
678            Backend::Looped(inbox) => inbox.extend(bytes),
679            Backend::Paired { outbox, .. } => lock(outbox).extend(bytes),
680            Backend::Simulated { peer, inbox } => inbox.extend(peer.receive(bytes)),
681            Backend::Scripted { steps, inbox } => match steps.front() {
682                Some(PortStep::Write(expected)) if expected == bytes => {
683                    steps.pop_front();
684                    release(steps, inbox);
685                }
686                next => {
687                    let expected = match next {
688                        Some(PortStep::Write(expected)) => Some(expected.clone()),
689                        _ => None,
690                    };
691                    return Err(PortError::Unexpected {
692                        expected,
693                        written: bytes.to_vec(),
694                    });
695                }
696            },
697        }
698        shared.written += bytes.len();
699        Ok(())
700    }
701
702    /// Reads what has arrived, waiting up to `timeout` for the first byte when nothing has.
703    ///
704    /// On any port but the kernel's device the read does not wait: when nothing has arrived it
705    /// returns at once, and the timeout is added to [`SerialPort::waited_micros`].
706    ///
707    /// # Arguments
708    ///
709    /// * `buffer` - where the bytes go; at most its length is read.
710    /// * `timeout` - how long to wait for the first byte.
711    ///
712    /// # Returns
713    ///
714    /// How many bytes were read, zero when the timeout passed with nothing.
715    ///
716    /// # Errors
717    ///
718    /// [`PortError::Device`] when the kernel's device fails.
719    pub fn read(&self, buffer: &mut [u8], timeout: Duration) -> Result<usize, PortError> {
720        let mut shared = lock(&self.shared);
721        let count = match &mut shared.backend {
722            #[cfg(all(feature = "linux", target_os = "linux"))]
723            Backend::Device { path, file } => {
724                let (path, file) = (path.clone(), Arc::clone(file));
725                drop(shared);
726                let count =
727                    device::read(&file, buffer, timeout).map_err(|error| PortError::Device {
728                        reason: format!("{}: {error}", path.display()),
729                    })?;
730                shared = lock(&self.shared);
731                count
732            }
733            Backend::Looped(inbox)
734            | Backend::Simulated { inbox, .. }
735            | Backend::Scripted { inbox, .. } => take(inbox, buffer),
736            Backend::Paired { inbox, .. } => take(&mut lock(inbox), buffer),
737        };
738        if count == 0 && !buffer.is_empty() {
739            shared.waited_ns = shared.waited_ns.saturating_add(nanos(timeout));
740        }
741        shared.received += count;
742        Ok(count)
743    }
744
745    /// Drops whatever has arrived and not been read, as a client does before it sends a
746    /// request so a stale reply cannot be taken for the new one.
747    ///
748    /// # Errors
749    ///
750    /// [`PortError::Device`] when the kernel's device fails.
751    pub fn discard_input(&self) -> Result<(), PortError> {
752        let mut shared = lock(&self.shared);
753        match &mut shared.backend {
754            #[cfg(all(feature = "linux", target_os = "linux"))]
755            Backend::Device { path, file } => {
756                device::discard(file).map_err(|error| PortError::Device {
757                    reason: format!("{}: {error}", path.display()),
758                })?;
759            }
760            Backend::Looped(inbox)
761            | Backend::Simulated { inbox, .. }
762            | Backend::Scripted { inbox, .. } => inbox.clear(),
763            Backend::Paired { inbox, .. } => lock(inbox).clear(),
764        }
765        Ok(())
766    }
767
768    /// Waits, as a protocol does to leave the line silent between frames: the process sleeps
769    /// on the kernel's device, and anywhere else the wait is only counted.
770    ///
771    /// # Arguments
772    ///
773    /// * `duration` - how long.
774    pub fn wait(&self, duration: Duration) {
775        let sleeps = {
776            let mut shared = lock(&self.shared);
777            shared.waited_ns = shared.waited_ns.saturating_add(nanos(duration));
778            shared.backend.kind() == PortKind::Device
779        };
780        if sleeps {
781            std::thread::sleep(duration);
782        }
783    }
784
785    /// Returns how many bytes have been written through this port and every clone of it.
786    pub fn written(&self) -> usize {
787        lock(&self.shared).written
788    }
789
790    /// Returns how many bytes have been read through this port and every clone of it.
791    pub fn received(&self) -> usize {
792        lock(&self.shared).received
793    }
794
795    /// Returns how long reads have waited without an answer, and [`SerialPort::wait`] has
796    /// waited, in microseconds, whether or not the process slept through it.
797    pub fn waited_micros(&self) -> u64 {
798        lock(&self.shared).waited_ns / 1_000
799    }
800
801    /// Returns how many steps a script has left, or `None` when the port is not scripted.
802    pub fn remaining(&self) -> Option<usize> {
803        match &lock(&self.shared).backend {
804            Backend::Scripted { steps, .. } => Some(steps.len()),
805            _ => None,
806        }
807    }
808}
809
810impl fmt::Debug for SerialPort {
811    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
812        let shared = lock(&self.shared);
813        f.debug_struct("SerialPort")
814            .field("kind", &shared.backend.kind())
815            .field("settings", &shared.settings)
816            .field("written", &shared.written)
817            .field("received", &shared.received)
818            .finish_non_exhaustive()
819    }
820}
821
822/// The kernel's terminal interface, set up as a raw byte pipe.
823#[cfg(all(feature = "linux", target_os = "linux"))]
824mod device {
825    use std::fs::{File, OpenOptions};
826    use std::io::{self, Read, Write};
827    use std::os::fd::AsFd;
828    use std::os::unix::fs::OpenOptionsExt;
829    use std::path::Path;
830    use std::time::{Duration, Instant};
831
832    use nix::errno::Errno;
833    use nix::poll::{poll, PollFd, PollFlags, PollTimeout};
834    use nix::sys::termios::{
835        self, BaudRate, ControlFlags, FlushArg, InputFlags, SetArg, SpecialCharacterIndices,
836    };
837
838    use super::{Parity, Settings, StopBits};
839
840    pub(super) fn open(path: &Path, settings: Settings) -> io::Result<File> {
841        let file = OpenOptions::new()
842            .read(true)
843            .write(true)
844            .custom_flags(nix::libc::O_NOCTTY)
845            .open(path)?;
846        let speed = match settings.baud {
847            1_200 => BaudRate::B1200,
848            2_400 => BaudRate::B2400,
849            4_800 => BaudRate::B4800,
850            9_600 => BaudRate::B9600,
851            19_200 => BaudRate::B19200,
852            38_400 => BaudRate::B38400,
853            57_600 => BaudRate::B57600,
854            115_200 => BaudRate::B115200,
855            230_400 => BaudRate::B230400,
856            460_800 => BaudRate::B460800,
857            921_600 => BaudRate::B921600,
858            _ => {
859                return Err(io::Error::new(
860                    io::ErrorKind::InvalidInput,
861                    "not a standard serial rate",
862                ))
863            }
864        };
865        let mut tty = termios::tcgetattr(&file)?;
866        termios::cfmakeraw(&mut tty);
867        termios::cfsetispeed(&mut tty, speed)?;
868        termios::cfsetospeed(&mut tty, speed)?;
869        tty.control_flags |= ControlFlags::CLOCAL | ControlFlags::CREAD;
870        tty.control_flags &= !(ControlFlags::PARENB
871            | ControlFlags::PARODD
872            | ControlFlags::CSTOPB
873            | ControlFlags::CRTSCTS
874            | ControlFlags::CSIZE);
875        tty.control_flags |= ControlFlags::CS8;
876        match settings.parity {
877            Parity::None => {}
878            Parity::Even => tty.control_flags |= ControlFlags::PARENB,
879            Parity::Odd => tty.control_flags |= ControlFlags::PARENB | ControlFlags::PARODD,
880        }
881        if settings.parity != Parity::None {
882            tty.input_flags |= InputFlags::INPCK | InputFlags::IGNPAR;
883        }
884        if settings.stop_bits == StopBits::Two {
885            tty.control_flags |= ControlFlags::CSTOPB;
886        }
887        tty.control_chars[SpecialCharacterIndices::VMIN as usize] = 0;
888        tty.control_chars[SpecialCharacterIndices::VTIME as usize] = 0;
889        termios::tcsetattr(&file, SetArg::TCSANOW, &tty)?;
890        termios::tcflush(&file, FlushArg::TCIOFLUSH)?;
891        Ok(file)
892    }
893
894    pub(super) fn write(file: &File, bytes: &[u8]) -> io::Result<()> {
895        let mut writer = file;
896        writer.write_all(bytes)?;
897        termios::tcdrain(file)?;
898        Ok(())
899    }
900
901    pub(super) fn read(file: &File, buffer: &mut [u8], timeout: Duration) -> io::Result<usize> {
902        if buffer.is_empty() {
903            return Ok(0);
904        }
905        let deadline = Instant::now() + timeout;
906        loop {
907            let left = deadline.saturating_duration_since(Instant::now());
908            let millis = i32::try_from(left.as_micros().div_ceil(1_000)).unwrap_or(i32::MAX);
909            let wait = PollTimeout::try_from(millis).unwrap_or(PollTimeout::MAX);
910            let mut fds = [PollFd::new(file.as_fd(), PollFlags::POLLIN)];
911            match poll(&mut fds, wait) {
912                Ok(0) => return Ok(0),
913                Ok(_) => {
914                    let mut reader = file;
915                    return reader.read(buffer);
916                }
917                Err(Errno::EINTR) => continue,
918                Err(error) => return Err(error.into()),
919            }
920        }
921    }
922
923    pub(super) fn discard(file: &File) -> io::Result<()> {
924        termios::tcflush(file, FlushArg::TCIFLUSH)?;
925        Ok(())
926    }
927}
928
929#[cfg(test)]
930mod tests {
931    use super::*;
932
933    #[test]
934    fn a_character_is_its_start_data_parity_and_stop_bits() {
935        let eight_n_one = Settings::new(115_200);
936        assert_eq!(eight_n_one.bits_per_character(), 10);
937        assert_eq!(eight_n_one.character_nanos(), 86_806, "10 bits at 115200");
938        assert_eq!(eight_n_one.to_string(), "115200 8N1");
939
940        // Modbus over Serial Line V1.02, 2.5.1: eleven bits a character, even parity by
941        // default, and two stop bits in place of a parity bit.
942        let modbus = Settings::new(9_600).with_parity(Parity::Even);
943        assert_eq!(modbus.bits_per_character(), 11);
944        let no_parity = Settings::new(9_600).with_stop_bits(StopBits::Two);
945        assert_eq!(no_parity.bits_per_character(), 11);
946        assert_eq!(no_parity.to_string(), "9600 8N2");
947        assert_eq!(
948            Settings::new(19_200).with_parity(Parity::Odd).to_string(),
949            "19200 8O1"
950        );
951
952        // 3.5 characters at 9600 8E1 is the 4.01 ms silence that ends an RTU frame.
953        assert_eq!((modbus.character_nanos() * 7).div_ceil(2), 4_010_419);
954        assert_eq!(modbus.transfer_micros(8), 9_167, "an eight-byte request");
955    }
956
957    #[test]
958    fn a_looped_line_reads_back_what_it_wrote() {
959        let port = SerialPort::looped(Settings::new(9_600));
960        port.write(b"abc").unwrap();
961        let mut buffer = [0u8; 2];
962        assert_eq!(port.read(&mut buffer, Duration::from_millis(5)).unwrap(), 2);
963        assert_eq!(&buffer, b"ab");
964        assert_eq!(port.read(&mut buffer, Duration::from_millis(5)).unwrap(), 1);
965        assert_eq!(buffer[0], b'c');
966        assert_eq!(port.read(&mut buffer, Duration::from_millis(5)).unwrap(), 0);
967        assert_eq!(port.waited_micros(), 5_000, "only the empty read waited");
968        assert_eq!((port.written(), port.received()), (3, 3));
969    }
970
971    #[test]
972    fn each_end_of_a_pair_reads_what_the_other_wrote() {
973        let (left, right) = SerialPort::pair(Settings::new(115_200));
974        left.write(b"ping").unwrap();
975        right.write(b"pong").unwrap();
976        let mut buffer = [0u8; 8];
977        let got = right.read(&mut buffer, Duration::ZERO).unwrap();
978        assert_eq!(&buffer[..got], b"ping");
979        let got = left.read(&mut buffer, Duration::ZERO).unwrap();
980        assert_eq!(&buffer[..got], b"pong");
981        assert_eq!(left.kind(), PortKind::Paired);
982
983        let reader = right.clone();
984        left.write(b"!").unwrap();
985        assert_eq!(reader.read(&mut buffer, Duration::ZERO).unwrap(), 1);
986        assert_eq!(right.received(), 5, "a clone reaches the same end");
987    }
988
989    struct Counter(usize);
990
991    impl Peer for Counter {
992        fn receive(&mut self, bytes: &[u8]) -> Vec<u8> {
993            self.0 += 1;
994            vec![bytes.len() as u8, self.0 as u8]
995        }
996    }
997
998    #[test]
999    fn a_simulated_device_answers_each_write_and_stays_reachable() {
1000        let device = Arc::new(Mutex::new(Counter(0)));
1001        let port = SerialPort::simulated(Settings::new(9_600), Arc::clone(&device));
1002        port.write(b"one").unwrap();
1003        port.write(b"three").unwrap();
1004        let mut buffer = [0u8; 8];
1005        let got = port.read(&mut buffer, Duration::from_millis(1)).unwrap();
1006        assert_eq!(&buffer[..got], &[3, 1, 5, 2]);
1007        assert_eq!(device.lock().unwrap().0, 2);
1008        assert_eq!(port.kind(), PortKind::Simulated);
1009    }
1010
1011    #[test]
1012    fn stale_input_is_discarded() {
1013        let port = SerialPort::looped(Settings::new(9_600));
1014        port.write(b"stale").unwrap();
1015        port.discard_input().unwrap();
1016        let mut buffer = [0u8; 8];
1017        assert_eq!(port.read(&mut buffer, Duration::from_millis(2)).unwrap(), 0);
1018        assert_eq!(port.received(), 0);
1019    }
1020
1021    #[test]
1022    fn a_script_checks_each_write_and_releases_what_follows() {
1023        let port = SerialPort::scripted(
1024            Settings::new(9_600),
1025            [
1026                PortStep::read(*b"hello"),
1027                PortStep::write(*b"?"),
1028                PortStep::read(*b"4"),
1029                PortStep::read(*b"2"),
1030                PortStep::write(*b"bye"),
1031            ],
1032        );
1033        let mut buffer = [0u8; 8];
1034        let got = port.read(&mut buffer, Duration::ZERO).unwrap();
1035        assert_eq!(
1036            &buffer[..got],
1037            b"hello",
1038            "reads at the start are there at once"
1039        );
1040        assert_eq!(
1041            port.write(b"!"),
1042            Err(PortError::Unexpected {
1043                expected: Some(b"?".to_vec()),
1044                written: b"!".to_vec(),
1045            })
1046        );
1047        port.write(b"?").unwrap();
1048        let got = port.read(&mut buffer, Duration::ZERO).unwrap();
1049        assert_eq!(&buffer[..got], b"42");
1050        assert_eq!(port.remaining(), Some(1));
1051        port.write(b"bye").unwrap();
1052        let error = port.write(b"again").unwrap_err();
1053        assert_eq!(
1054            error.to_string(),
1055            "the script expected no more writes, but 61 67 61 69 6e was written"
1056        );
1057        assert_eq!(port.written(), 4, "a refused write is not counted");
1058    }
1059
1060    #[test]
1061    fn a_wait_is_counted_and_not_slept_off_a_real_device() {
1062        let port = SerialPort::looped(Settings::new(9_600));
1063        let started = std::time::Instant::now();
1064        port.wait(Duration::from_secs(3));
1065        assert!(started.elapsed() < Duration::from_secs(1));
1066        assert_eq!(port.waited_micros(), 3_000_000);
1067    }
1068
1069    #[cfg(all(feature = "linux", target_os = "linux"))]
1070    #[test]
1071    fn a_pseudo_terminal_opens_raw_and_carries_bytes_both_ways() {
1072        use nix::fcntl::OFlag;
1073        use nix::pty::{grantpt, posix_openpt, ptsname_r, unlockpt};
1074        use std::io::{Read, Write};
1075        use std::time::Instant;
1076
1077        let mut far_end = posix_openpt(OFlag::O_RDWR | OFlag::O_NOCTTY).unwrap();
1078        grantpt(&far_end).unwrap();
1079        unlockpt(&far_end).unwrap();
1080        let path = ptsname_r(&far_end).unwrap();
1081        let port = SerialPort::open(&path, Settings::new(9_600).with_parity(Parity::Even)).unwrap();
1082        assert_eq!(port.kind(), PortKind::Device);
1083
1084        // A NUL, the XON and XOFF flow control bytes, and a carriage return: a terminal left
1085        // cooked would swallow or translate every one of them.
1086        let awkward = [0x00, 0x11, 0x13, 0x0D];
1087        far_end.write_all(&awkward).unwrap();
1088        let mut buffer = [0u8; 8];
1089        let mut got = 0;
1090        while got < awkward.len() {
1091            let count = port
1092                .read(&mut buffer[got..], Duration::from_millis(500))
1093                .unwrap();
1094            assert!(count > 0, "the pseudo-terminal went quiet");
1095            got += count;
1096        }
1097        assert_eq!(&buffer[..got], &awkward);
1098
1099        port.write(b"ok").unwrap();
1100        let mut back = [0u8; 2];
1101        far_end.read_exact(&mut back).unwrap();
1102        assert_eq!(&back, b"ok");
1103
1104        let started = Instant::now();
1105        assert_eq!(
1106            port.read(&mut buffer, Duration::from_millis(40)).unwrap(),
1107            0
1108        );
1109        assert!(
1110            started.elapsed() >= Duration::from_millis(40),
1111            "a real read waits"
1112        );
1113        assert_eq!(port.waited_micros(), 40_000);
1114
1115        far_end.write_all(b"stale").unwrap();
1116        std::thread::sleep(Duration::from_millis(20));
1117        port.discard_input().unwrap();
1118        assert_eq!(
1119            port.read(&mut buffer, Duration::from_millis(20)).unwrap(),
1120            0
1121        );
1122
1123        assert!(matches!(
1124            SerialPort::open(&path, Settings::new(12_345)),
1125            Err(OpenError::Device { .. })
1126        ));
1127    }
1128
1129    #[cfg(not(all(feature = "linux", target_os = "linux")))]
1130    #[test]
1131    fn only_linux_opens_a_device() {
1132        assert_eq!(
1133            SerialPort::open("/dev/serial0", Settings::new(115_200)).unwrap_err(),
1134            OpenError::Unsupported
1135        );
1136    }
1137}
pamoja_hal/port.rs

pamoja_hal/
port.rs
