subms_spsc_ring_buffer/features/
mpsc_fan_in.rs

1//! N-producer single-consumer fan-in over N independent SPSC rings.
2//!
3//! Each producer pushes into its own SPSC ring (so it stays wait-free against
4//! its own counter), and the consumer round-robins across all rings. The
5//! consumer never blocks any producer; producers never contend with each
6//! other.
7//!
8//! Memory cost: `N * capacity` slots. Round-robin is fair under steady-state
9//! load; under skewed load the consumer spends extra `try_pop` calls on
10//! quiet producers but never starves a busy one.
11
12use crate::{Consumer, Producer, SpscRingBuffer};
13
14/// Builder + handle factory for an N-producer fan-in. After construction,
15/// move each `MpscFanInProducer` to its producing thread and the single
16/// `MpscFanInConsumer` to its consuming thread.
17pub struct MpscFanIn;
18
19impl MpscFanIn {
20    /// Build `producer_count` SPSC rings of capacity `per_ring_capacity` and
21    /// return matched producer / consumer handles.
22    pub fn with_capacity<T: Send + 'static>(
23        producer_count: usize,
24        per_ring_capacity: usize,
25    ) -> (Vec<MpscFanInProducer<T>>, MpscFanInConsumer<T>) {
26        assert!(producer_count >= 1, "need at least one producer");
27        let mut producers = Vec::with_capacity(producer_count);
28        let mut consumers = Vec::with_capacity(producer_count);
29        for _ in 0..producer_count {
30            let (p, c) = SpscRingBuffer::with_capacity::<T>(per_ring_capacity);
31            producers.push(MpscFanInProducer { inner: p });
32            consumers.push(c);
33        }
34        (
35            producers,
36            MpscFanInConsumer {
37                rings: consumers,
38                cursor: 0,
39            },
40        )
41    }
42}
43
44/// One producer side of an `MpscFanIn`. Wait-free against its own ring;
45/// independent of any other producer.
46pub struct MpscFanInProducer<T> {
47    inner: Producer<T>,
48}
49
50impl<T> MpscFanInProducer<T> {
51    /// Push a value into this producer's ring. Returns `Err(value)` if full.
52    pub fn try_push(&mut self, value: T) -> Result<(), T> {
53        self.inner.try_push(value)
54    }
55
56    pub fn capacity(&self) -> usize {
57        self.inner.capacity()
58    }
59}
60
61/// The single consumer side. Round-robins across the producer rings.
62pub struct MpscFanInConsumer<T> {
63    rings: Vec<Consumer<T>>,
64    /// Next ring to probe; advances every successful pop so producers don't
65    /// starve. Wraps modulo `rings.len()`.
66    cursor: usize,
67}
68
69impl<T> MpscFanInConsumer<T> {
70    /// Try one full round of probes across all producer rings. Returns the
71    /// first value found, advancing the cursor past the producing ring so
72    /// the next call starts at a fresh point.
73    pub fn try_pop(&mut self) -> Option<T> {
74        let n = self.rings.len();
75        for offset in 0..n {
76            let idx = (self.cursor + offset) % n;
77            if let Some(v) = self.rings[idx].try_pop() {
78                self.cursor = (idx + 1) % n;
79                return Some(v);
80            }
81        }
82        None
83    }
84
85    /// Producer count.
86    pub fn producer_count(&self) -> usize {
87        self.rings.len()
88    }
89}
90
91#[cfg(test)]
92#[path = "mpsc_fan_in_tests.rs"]
93mod tests;
subms_spsc_ring_buffer/features/mpsc_fan_in.rs

subms_spsc_ring_buffer/features/
mpsc_fan_in.rs
