leodos_protocols/coding/
pipeline.rs

1//! Coding pipeline that composes randomizer, FEC, and framer
2//! into a single `CodingWrite` / `CodingRead`.
3
4use crate::coding::CodingRead;
5use crate::coding::CodingWrite;
6use crate::coding::Deframer;
7use crate::coding::FecDecoder;
8use crate::coding::FecEncoder;
9use crate::coding::Framer;
10use crate::coding::randomizer::Randomizer;
11use crate::physical::PhysicalRead;
12use crate::physical::PhysicalWrite;
13
14// ── Write pipeline ──────────────────────────────────────────
15
16/// Composes randomizer → FEC → framer → physical writer into a
17/// single [`CodingWrite`].
18pub struct CodingWriter<R, F, M, W, const BUF: usize> {
19    /// Randomizer applied to the transfer frame.
20    pub randomizer: R,
21    /// Forward error-correction encoder.
22    pub fec: F,
23    /// Framer (e.g. ASM, CLTU).
24    pub framer: M,
25    /// Physical layer writer.
26    pub writer: W,
27    buf_a: [u8; BUF],
28    buf_b: [u8; BUF],
29}
30
31impl<R, F, M, W, const BUF: usize> CodingWriter<R, F, M, W, BUF> {
32    /// Creates a new write pipeline.
33    pub fn new(randomizer: R, fec: F, framer: M, writer: W) -> Self {
34        Self {
35            randomizer,
36            fec,
37            framer,
38            writer,
39            buf_a: [0u8; BUF],
40            buf_b: [0u8; BUF],
41        }
42    }
43}
44
45/// Error from a coding write pipeline.
46#[derive(Debug)]
47pub enum CodingWriteError<F, M, W> {
48    /// FEC encoding failed.
49    Fec(F),
50    /// Framing failed.
51    Framer(M),
52    /// Physical writer failed.
53    Writer(W),
54    /// Internal buffer too small.
55    BufferTooSmall,
56}
57
58impl<F: core::fmt::Display, M: core::fmt::Display, W: core::fmt::Display> core::fmt::Display
59    for CodingWriteError<F, M, W>
60{
61    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
62        match self {
63            Self::Fec(e) => write!(f, "FEC encode: {e}"),
64            Self::Framer(e) => write!(f, "framer: {e}"),
65            Self::Writer(e) => write!(f, "writer: {e}"),
66            Self::BufferTooSmall => write!(f, "buffer too small"),
67        }
68    }
69}
70
71impl<F: core::error::Error, M: core::error::Error, W: core::error::Error> core::error::Error
72    for CodingWriteError<F, M, W>
73{
74}
75
76impl<R, F, M, W, const BUF: usize> CodingWrite for CodingWriter<R, F, M, W, BUF>
77where
78    R: Randomizer,
79    F: FecEncoder,
80    F::Error: core::error::Error,
81    M: Framer,
82    M::Error: core::error::Error,
83    W: PhysicalWrite,
84    W::Error: core::error::Error,
85{
86    type Error = CodingWriteError<F::Error, M::Error, W::Error>;
87
88    async fn write(&mut self, frame: &[u8]) -> Result<(), Self::Error> {
89        if frame.len() > BUF {
90            return Err(CodingWriteError::BufferTooSmall);
91        }
92
93        // 1. Copy frame → buf_a, randomize
94        self.buf_a[..frame.len()].copy_from_slice(frame);
95        self.randomizer.apply(&mut self.buf_a[..frame.len()]);
96
97        // 2. FEC encode: buf_a → buf_b
98        let fec_len = self
99            .fec
100            .encode(&self.buf_a[..frame.len()], &mut self.buf_b)
101            .map_err(CodingWriteError::Fec)?;
102
103        // 3. Frame: buf_b → buf_a
104        let framed_len = self
105            .framer
106            .frame(&self.buf_b[..fec_len], &mut self.buf_a)
107            .map_err(CodingWriteError::Framer)?;
108
109        // 4. Write to physical layer
110        self.writer
111            .write(&self.buf_a[..framed_len])
112            .await
113            .map_err(CodingWriteError::Writer)
114    }
115}
116
117// ── Read pipeline ───────────────────────────────────────────
118
119/// Composes physical reader → deframer → FEC → derandomizer into
120/// a single [`CodingRead`].
121pub struct CodingReader<R, D, F, P, const BUF: usize> {
122    /// Derandomizer (same as randomizer — XOR is self-inverse).
123    pub randomizer: R,
124    /// Deframer (e.g. ASM sync, CLTU decode).
125    pub deframer: D,
126    /// Forward error-correction decoder.
127    pub fec: F,
128    /// Physical layer reader.
129    pub reader: P,
130    buf_a: [u8; BUF],
131    buf_b: [u8; BUF],
132}
133
134impl<R, D, F, P, const BUF: usize> CodingReader<R, D, F, P, BUF> {
135    /// Creates a new read pipeline.
136    pub fn new(randomizer: R, deframer: D, fec: F, reader: P) -> Self {
137        Self {
138            randomizer,
139            deframer,
140            fec,
141            reader,
142            buf_a: [0u8; BUF],
143            buf_b: [0u8; BUF],
144        }
145    }
146}
147
148/// Error from a coding read pipeline.
149#[derive(Debug)]
150pub enum CodingReadError<D, F, P> {
151    /// Deframing failed.
152    Deframer(D),
153    /// FEC decoding failed.
154    Fec(F),
155    /// Physical reader failed.
156    Reader(P),
157    /// Internal buffer too small.
158    BufferTooSmall,
159}
160
161impl<D: core::fmt::Display, F: core::fmt::Display, P: core::fmt::Display> core::fmt::Display
162    for CodingReadError<D, F, P>
163{
164    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
165        match self {
166            Self::Deframer(e) => write!(f, "deframer: {e}"),
167            Self::Fec(e) => write!(f, "FEC decode: {e}"),
168            Self::Reader(e) => write!(f, "reader: {e}"),
169            Self::BufferTooSmall => write!(f, "buffer too small"),
170        }
171    }
172}
173
174impl<D: core::error::Error, F: core::error::Error, P: core::error::Error> core::error::Error
175    for CodingReadError<D, F, P>
176{
177}
178
179impl<R, D, F, P, const BUF: usize> CodingRead for CodingReader<R, D, F, P, BUF>
180where
181    R: Randomizer,
182    D: Deframer,
183    D::Error: core::error::Error,
184    F: FecDecoder,
185    F::Error: core::error::Error,
186    P: PhysicalRead,
187    P::Error: core::error::Error,
188{
189    type Error = CodingReadError<D::Error, F::Error, P::Error>;
190
191    async fn read(&mut self, buffer: &mut [u8]) -> Result<usize, Self::Error> {
192        // 1. Read raw bytes from physical layer
193        let raw_len = self
194            .reader
195            .read(&mut self.buf_a)
196            .await
197            .map_err(CodingReadError::Reader)?;
198
199        if raw_len == 0 {
200            return Ok(0);
201        }
202
203        // 2. Deframe: buf_a → buf_b
204        let deframed_len = self
205            .deframer
206            .deframe(&self.buf_a[..raw_len], &mut self.buf_b)
207            .map_err(CodingReadError::Deframer)?;
208
209        // 3. FEC decode: copy buf_b → buf_a, decode in-place
210        self.buf_a[..deframed_len].copy_from_slice(&self.buf_b[..deframed_len]);
211        let data_len = self
212            .fec
213            .decode(&mut self.buf_a[..deframed_len])
214            .map_err(CodingReadError::Fec)?;
215
216        // 4. Derandomize in-place
217        self.randomizer.apply(&mut self.buf_a[..data_len]);
218
219        // 5. Copy to caller's buffer
220        if buffer.len() < data_len {
221            return Err(CodingReadError::BufferTooSmall);
222        }
223        buffer[..data_len].copy_from_slice(&self.buf_a[..data_len]);
224        Ok(data_len)
225    }
226}
227
228#[cfg(test)]
229mod tests {
230    use super::*;
231    use crate::coding::{NoFec, NoFramer, NoRandomizer};
232
233    struct MemWriter {
234        data: [u8; 1024],
235        len: usize,
236    }
237
238    impl MemWriter {
239        fn new() -> Self {
240            Self {
241                data: [0u8; 1024],
242                len: 0,
243            }
244        }
245    }
246
247    #[derive(Debug)]
248    struct MemError;
249    impl core::fmt::Display for MemError {
250        fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
251            write!(f, "mem error")
252        }
253    }
254    impl core::error::Error for MemError {}
255
256    impl PhysicalWrite for MemWriter {
257        type Error = MemError;
258        async fn write(&mut self, data: &[u8]) -> Result<(), Self::Error> {
259            self.data[..data.len()].copy_from_slice(data);
260            self.len = data.len();
261            Ok(())
262        }
263    }
264
265    struct MemReader {
266        data: [u8; 1024],
267        len: usize,
268    }
269
270    impl MemReader {
271        fn new(data: &[u8]) -> Self {
272            let mut buf = [0u8; 1024];
273            buf[..data.len()].copy_from_slice(data);
274            Self {
275                data: buf,
276                len: data.len(),
277            }
278        }
279    }
280
281    impl PhysicalRead for MemReader {
282        type Error = MemError;
283        async fn read(&mut self, buffer: &mut [u8]) -> Result<usize, Self::Error> {
284            let len = self.len.min(buffer.len());
285            buffer[..len].copy_from_slice(&self.data[..len]);
286            Ok(len)
287        }
288    }
289
290    #[test]
291    fn no_op_pipeline_roundtrip() {
292        futures::executor::block_on(async {
293            let writer = MemWriter::new();
294            let mut write_pipe: CodingWriter<_, _, _, _, 1024> =
295                CodingWriter::new(NoRandomizer, NoFec, NoFramer, writer);
296
297            let original = b"Hello, pipeline!";
298            write_pipe.write(original).await.unwrap();
299
300            let written = &write_pipe.writer.data[..write_pipe.writer.len];
301            assert_eq!(written, original);
302
303            // Read pipeline: D=NoFramer (deframer), F=NoFec (decoder)
304            let reader = MemReader::new(written);
305            let mut read_pipe: CodingReader<_, _, _, _, 1024> =
306                CodingReader::new(NoRandomizer, NoFramer, NoFec, reader);
307
308            let mut buf = [0u8; 256];
309            let len = read_pipe.read(&mut buf).await.unwrap();
310            assert_eq!(&buf[..len], original);
311        });
312    }
313
314    #[test]
315    fn pipeline_with_randomizer() {
316        use crate::coding::randomizer::Tm255Randomizer;
317
318        futures::executor::block_on(async {
319            let writer = MemWriter::new();
320            let mut write_pipe: CodingWriter<_, _, _, _, 1024> =
321                CodingWriter::new(Tm255Randomizer::new(), NoFec, NoFramer, writer);
322
323            let original = b"Randomized data!";
324            write_pipe.write(original).await.unwrap();
325
326            let written = &write_pipe.writer.data[..write_pipe.writer.len];
327            assert_ne!(written, original);
328
329            let reader = MemReader::new(written);
330            let mut read_pipe: CodingReader<_, _, _, _, 1024> =
331                CodingReader::new(Tm255Randomizer::new(), NoFramer, NoFec, reader);
332
333            let mut buf = [0u8; 256];
334            let len = read_pipe.read(&mut buf).await.unwrap();
335            assert_eq!(&buf[..len], original);
336        });
337    }
338
339    #[test]
340    fn pipeline_with_asm_framing() {
341        use crate::coding::framing::cadu::{AsmDeframer, AsmFramer};
342
343        futures::executor::block_on(async {
344            let writer = MemWriter::new();
345            let mut write_pipe: CodingWriter<_, _, _, _, 1024> =
346                CodingWriter::new(NoRandomizer, NoFec, AsmFramer::tm(), writer);
347
348            let original = [0xAAu8; 32];
349            write_pipe.write(&original).await.unwrap();
350
351            let written_len = write_pipe.writer.len;
352            assert_eq!(written_len, 36);
353
354            let reader = MemReader::new(&write_pipe.writer.data[..written_len]);
355            let mut read_pipe: CodingReader<_, _, _, _, 1024> =
356                CodingReader::new(NoRandomizer, AsmDeframer::tm(32), NoFec, reader);
357
358            let mut buf = [0u8; 256];
359            let len = read_pipe.read(&mut buf).await.unwrap();
360            assert_eq!(&buf[..len], &original);
361        });
362    }
363}
leodos_protocols/coding/pipeline.rs

leodos_protocols/coding/
pipeline.rs
