dope_sys/io/socket/record/
stream.rs

1use std::mem;
2
3use crate::{
4    Errno,
5    io::socket::record::{Fault, Record, Size, Transport},
6};
7
8const PAYLOAD: usize = mem::size_of::<u32>();
9const PROBE: usize = PAYLOAD + 1;
10
11/// Stream frames retain one length byte followed by at most 255 payload bytes.
12/// Invalid lengths are drained before reporting their bounded observed extent.
13pub struct Stream;
14
15impl Transport for Stream {
16    fn send(
17        value: u32,
18        mut send: impl for<'a> FnMut(&'a [u8]) -> Result<&'a [u8], Errno>,
19    ) -> Result<(), Fault> {
20        let [a, b, c, d] = value.to_ne_bytes();
21        let bytes = [PAYLOAD as u8, a, b, c, d];
22        let mut pending = bytes.as_slice();
23        while !pending.is_empty() {
24            let sent = match send(pending) {
25                Ok(sent) if !sent.is_empty() => sent.len(),
26                Ok(_) => return Err(size((bytes.len() - pending.len()).saturating_sub(1))),
27                Err(error) if error.is_intr() => continue,
28                Err(error) => return Err(Fault::Syscall(error)),
29            };
30            pending = pending.get(sent..).ok_or_else(|| size(sent))?;
31        }
32        Ok(())
33    }
34
35    fn recv(mut recv: impl for<'a> FnMut(&'a mut [u8]) -> Result<&'a mut [u8], Errno>) -> Record {
36        match receive(&mut recv) {
37            Ok(record) => record,
38            Err(error) => Record::Failed(error),
39        }
40    }
41}
42
43fn receive(
44    recv: &mut impl for<'a> FnMut(&'a mut [u8]) -> Result<&'a mut [u8], Errno>,
45) -> Result<Record, Fault> {
46    let mut header = [0];
47    if fill(&mut header, recv)? == 0 {
48        return Ok(Record::Closed);
49    }
50    let length = usize::from(header[0]);
51    if length == PAYLOAD {
52        let mut bytes = [0; PAYLOAD];
53        let received = fill(&mut bytes, recv)?;
54        return if received == PAYLOAD {
55            Ok(Record::Value(u32::from_ne_bytes(bytes)))
56        } else {
57            Err(size(received))
58        };
59    }
60    let mut bytes = [0; PROBE];
61    let mut pending = length;
62    while pending != 0 {
63        let extent = pending.min(bytes.len());
64        let received = fill(&mut bytes[..extent], recv)?;
65        pending -= received;
66        if received != extent {
67            break;
68        }
69    }
70    Err(size(length - pending))
71}
72
73/// Retains the unfilled suffix across short completions and interrupted calls.
74fn fill(
75    mut output: &mut [u8],
76    recv: &mut impl for<'a> FnMut(&'a mut [u8]) -> Result<&'a mut [u8], Errno>,
77) -> Result<usize, Fault> {
78    let capacity = output.len();
79    while !output.is_empty() {
80        let received = match recv(&mut *output) {
81            Ok(received) if !received.is_empty() => received.len(),
82            Ok(_) => break,
83            Err(error) if error.is_intr() => continue,
84            Err(error) => return Err(Fault::Syscall(error)),
85        };
86        output = output.get_mut(received..).ok_or_else(|| size(received))?;
87    }
88    Ok(capacity - output.len())
89}
90
91fn size(actual: usize) -> Fault {
92    Fault::Record(Size {
93        expected: PAYLOAD,
94        actual: actual.min(PROBE),
95    })
96}
dope_sys/io/socket/record/stream.rs

dope_sys/io/socket/record/
stream.rs
