subetha_cxc/
message_transport.rs

1//! `MessageTransport` - byte-slice transport trait for the
2//! `BackgroundScheduler`, abstracting over `SharedRing` (MPMC) and
3//! `SharedDeque<PassSlot>` (SPMC work-stealing).
4//!
5//! The scheduler's wire format is a fixed-size payload (56 bytes
6//! per slot, the encoded `Pass` representation). Both MPMC ring and
7//! SPMC deque transports can carry this byte-slice payload; the
8//! caller picks based on the workload's topology.
9//!
10//! ## Picking a transport
11//!
12//! - `SharedRing` (MPMC): multiple producers + multiple consumers.
13//!   The canonical scheduler-submit-ring shape: any process
14//!   submits, the worker pops.
15//! - `SharedDeque<PassSlot>` (SPMC): single producer, many thieves.
16//!   The canonical scheduler-result-ring shape with a single
17//!   worker producing and many collectors draining.
18
19use std::sync::Arc;
20
21use subetha_core::{Marshal, MarshalError};
22
23use crate::shared_deque::{DequeError, SharedDeque};
24use crate::shared_ring::{RingError, SharedRing, PAYLOAD_BYTES};
25
26/// Errors a `MessageTransport` returns.
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub enum TransportError {
29    /// Transport is at capacity; producer must back off.
30    Full,
31    /// Transport is empty; consumer has nothing to take.
32    Empty,
33    /// Caller-supplied payload exceeds the wire-format slot size.
34    PayloadTooLarge,
35    /// Caller-supplied output buffer is shorter than the slot size.
36    OutBufferTooSmall,
37    /// Transport-specific error not covered by the categories above.
38    Other,
39}
40
41/// A byte-slice transport for fixed-size scheduler payloads. Both
42/// `SharedRing` and `SharedDeque<PassSlot>` impl this trait so the
43/// `BackgroundScheduler` can pick its underlying primitive at
44/// construction without changing its hot-loop code.
45pub trait MessageTransport: Send + Sync {
46    /// Push a payload of length `<= PAYLOAD_BYTES`. Returns
47    /// `Err(Full)` if the transport is at capacity.
48    fn try_push(&self, payload: &[u8]) -> Result<(), TransportError>;
49
50    /// Pop a payload into `out` (which must be `>= PAYLOAD_BYTES`
51    /// long). Returns the byte count written on success, or
52    /// `Err(Empty)` if there is nothing to take.
53    fn try_pop(&self, out: &mut [u8]) -> Result<usize, TransportError>;
54}
55
56/// 56-byte fixed-size payload type for the `SharedDeque<PassSlot>`
57/// transport path. Mirrors the byte layout that `SharedRing` uses
58/// for `BackgroundScheduler` so the same encoded Pass slot can ride
59/// either transport.
60#[repr(C, align(8))]
61#[derive(Clone, Copy, Debug, PartialEq, Eq)]
62pub struct PassSlot(pub [u8; PAYLOAD_BYTES]);
63
64impl Default for PassSlot {
65    fn default() -> Self {
66        Self([0u8; PAYLOAD_BYTES])
67    }
68}
69
70// SAFETY: `PassSlot` is `#[repr(C, align(8))]` over a single
71// `[u8; PAYLOAD_BYTES]` field. The bytes are position-independent
72// across address spaces; round-trip is a memcpy.
73unsafe impl Marshal for PassSlot {
74    const PAYLOAD_BYTES: usize = PAYLOAD_BYTES;
75
76    fn marshal(&self, dst: &mut [u8]) {
77        dst[..PAYLOAD_BYTES].copy_from_slice(&self.0);
78    }
79
80    fn unmarshal(src: &[u8]) -> Result<Self, MarshalError> {
81        if src.len() < PAYLOAD_BYTES {
82            return Err(MarshalError::ShortBuffer {
83                expected: PAYLOAD_BYTES,
84                got: src.len(),
85            });
86        }
87        let mut s = Self([0u8; PAYLOAD_BYTES]);
88        s.0.copy_from_slice(&src[..PAYLOAD_BYTES]);
89        Ok(s)
90    }
91}
92
93impl MessageTransport for SharedRing {
94    fn try_push(&self, payload: &[u8]) -> Result<(), TransportError> {
95        SharedRing::try_push(self, payload).map_err(map_ring_err)
96    }
97
98    fn try_pop(&self, out: &mut [u8]) -> Result<usize, TransportError> {
99        SharedRing::try_pop(self, out).map_err(map_ring_err)
100    }
101}
102
103impl MessageTransport for SharedDeque<PassSlot> {
104    fn try_push(&self, payload: &[u8]) -> Result<(), TransportError> {
105        if payload.len() > PAYLOAD_BYTES {
106            return Err(TransportError::PayloadTooLarge);
107        }
108        let mut slot = PassSlot([0u8; PAYLOAD_BYTES]);
109        slot.0[..payload.len()].copy_from_slice(payload);
110        self.push(&slot).map_err(map_deque_err)
111    }
112
113    fn try_pop(&self, out: &mut [u8]) -> Result<usize, TransportError> {
114        if out.len() < PAYLOAD_BYTES {
115            return Err(TransportError::OutBufferTooSmall);
116        }
117        // For SPMC Chase-Lev, the consumer-side primitive is `steal`,
118        // not `pop` (pop is the owner-side LIFO end). The
119        // BackgroundScheduler's worker is a thief in this topology.
120        match self.steal() {
121            Some(slot) => {
122                out[..PAYLOAD_BYTES].copy_from_slice(&slot.0);
123                Ok(PAYLOAD_BYTES)
124            }
125            None => Err(TransportError::Empty),
126        }
127    }
128}
129
130/// Blanket impl that lets `Arc<dyn MessageTransport>` delegate
131/// trait calls through the `Arc`.
132impl<T: MessageTransport + ?Sized> MessageTransport for Arc<T> {
133    fn try_push(&self, payload: &[u8]) -> Result<(), TransportError> {
134        (**self).try_push(payload)
135    }
136
137    fn try_pop(&self, out: &mut [u8]) -> Result<usize, TransportError> {
138        (**self).try_pop(out)
139    }
140}
141
142fn map_ring_err(e: RingError) -> TransportError {
143    match e {
144        RingError::Full => TransportError::Full,
145        RingError::Empty => TransportError::Empty,
146        RingError::PayloadTooLarge => TransportError::PayloadTooLarge,
147        _ => TransportError::Other,
148    }
149}
150
151fn map_deque_err(e: DequeError) -> TransportError {
152    match e {
153        DequeError::Full => TransportError::Full,
154        _ => TransportError::Other,
155    }
156}
157
158#[cfg(test)]
159mod tests {
160    use super::*;
161
162    fn tmp(name: &str) -> std::path::PathBuf {
163        let mut p = std::env::temp_dir();
164        let pid = std::process::id();
165        p.push(format!("subetha_transport_{name}_{pid}.bin"));
166        p
167    }
168
169    #[test]
170    fn shared_ring_satisfies_message_transport() {
171        let path = tmp("ring");
172        let ring = SharedRing::create(&path, 4).expect("create");
173        let mut payload = [0u8; PAYLOAD_BYTES];
174        payload[0] = 0xAB;
175        payload[1] = 0xCD;
176        ring.try_push(&payload).expect("push");
177
178        let mut out = [0u8; PAYLOAD_BYTES];
179        let n = (&ring as &dyn MessageTransport)
180            .try_pop(&mut out)
181            .expect("pop");
182        assert_eq!(n, PAYLOAD_BYTES);
183        assert_eq!(out, payload);
184        std::fs::remove_file(&path).ok();
185    }
186
187    #[test]
188    fn shared_deque_passslot_satisfies_message_transport() {
189        let path = tmp("deque");
190        let owner = SharedDeque::<PassSlot>::create(&path, 8).expect("create");
191        let thief = SharedDeque::<PassSlot>::open_as_thief(&path).expect("thief");
192
193        let mut payload = [0u8; PAYLOAD_BYTES];
194        payload[0] = 0x12;
195        payload[10] = 0x34;
196        (&owner as &dyn MessageTransport)
197            .try_push(&payload)
198            .expect("push");
199
200        // Steal-side path requires the thief handle.
201        let mut out = [0u8; PAYLOAD_BYTES];
202        let n = (&thief as &dyn MessageTransport)
203            .try_pop(&mut out)
204            .expect("pop");
205        assert_eq!(n, PAYLOAD_BYTES);
206        assert_eq!(out, payload);
207        std::fs::remove_file(&path).ok();
208    }
209
210    #[test]
211    fn arc_dyn_transport_delegates() {
212        let path = tmp("arc_dyn");
213        let ring: Arc<dyn MessageTransport> =
214            Arc::new(SharedRing::create(&path, 4).expect("create"));
215        let mut payload = [0u8; PAYLOAD_BYTES];
216        payload[5] = 0xEF;
217        ring.try_push(&payload).expect("push");
218
219        let mut out = [0u8; PAYLOAD_BYTES];
220        let n = ring.try_pop(&mut out).expect("pop");
221        assert_eq!(n, PAYLOAD_BYTES);
222        assert_eq!(out, payload);
223        std::fs::remove_file(&path).ok();
224    }
225
226    #[test]
227    fn payload_too_large_rejected() {
228        let path = tmp("oversize");
229        let ring = SharedRing::create(&path, 4).expect("create");
230        let oversized = vec![0u8; PAYLOAD_BYTES + 1];
231        let err = (&ring as &dyn MessageTransport)
232            .try_push(&oversized)
233            .expect_err("oversize");
234        assert_eq!(err, TransportError::PayloadTooLarge);
235        std::fs::remove_file(&path).ok();
236    }
237
238    #[test]
239    fn empty_returns_err_empty() {
240        let path = tmp("empty");
241        let ring = SharedRing::create(&path, 4).expect("create");
242        let mut out = [0u8; PAYLOAD_BYTES];
243        let err = (&ring as &dyn MessageTransport)
244            .try_pop(&mut out)
245            .expect_err("empty");
246        assert_eq!(err, TransportError::Empty);
247        std::fs::remove_file(&path).ok();
248    }
249}
subetha_cxc/message_transport.rs

subetha_cxc/
message_transport.rs
