Struct Producer 

pub struct Producer<T> { /* private fields */ }

Producer handle. Move it to the producing thread; the type system enforces the SPSC invariant.

Implementations

impl<T: Copy> Producer<T>

pub fn try_enqueue_bulk(&mut self, values: &[T]) -> usize

Copy as many items from values into the ring as will fit right now. Returns the count transferred. Single Release on the tail at the end.

Examples found in repository

examples/sample_app.rs (line 126)

fn bulk_batch_ingest() {
    println!("\n== bulk: batch a NIC receive into the ring ==");
    let (mut tx, mut rx) = SpscRingBuffer::with_capacity::<Tick>(16);
    let batch: Vec<Tick> = (0..10)
        .map(|seq| Tick {
            seq,
            price_cents: 20_000 + seq as u32,
        })
        .collect();

    let pushed = tx.try_enqueue_bulk(&batch);
    println!(
        "  offered {} ticks, took {pushed} in one fenced call",
        batch.len()
    );
    assert_eq!(pushed, 10);

    let mut out = [Tick {
        seq: 0,
        price_cents: 0,
    }; 10];
    let drained = rx.try_dequeue_bulk(&mut out);
    println!("  drained {drained} in one fenced call");
    assert_eq!(drained, 10);
    assert_eq!(out.as_slice(), batch.as_slice(), "bulk preserves order");
}

examples/perf_features.rs (line 230)

fn main() -> io::Result<()> {
    let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
        .join("..")
        .join(".subms")
        .join("features")
        .join("rust.json");
    let existing = std::fs::read_to_string(&path).unwrap_or_default();
    let mut manifest = SubMsFeatureManifest::load_str("rust", &existing);
    // Stamp the box these numbers came from. The bench runs wherever it is
    // invoked, so an unstamped manifest is indistinguishable from a fleet
    // capture; the renderer will not publish one it cannot attribute.
    let (source, instance) = SubMsP99Source::from_env();
    manifest.set_p99_source(source, instance.as_deref());

    // The baseline: the base wait-free push + pop round trip. Swept as well as
    // sampled, because whether ring capacity moves the BASE op is the context
    // every feature curve is read against.
    let base_sweep = sweep("base/push+pop", |cap| {
        let (mut tx, mut rx) = pair(cap);
        batched(ITEMS_PER_SAMPLE, |i| {
            let _ = tx.try_push(i as u64);
            rx.try_pop().unwrap_or(0)
        })
    });
    let base_p50 = base_sweep
        .iter()
        .find(|(n, _)| *n == CANON)
        .map_or(0, |(_, v)| *v);
    eprintln!("base push+pop p50 per {ITEMS_PER_SAMPLE}-item sample: {base_p50}ns");

    // ---------- bulk: one fence per BULK_BATCH items ----------
    #[cfg(feature = "bulk")]
    {
        // A sample moves ITEMS_PER_SAMPLE items either way; only the call width
        // differs. That is the comparison the feature exists to win, and it is
        // why the reps count is divided rather than the batch grown.
        let sw = sweep("bulk/enqueue+dequeue", |cap| {
            let (mut tx, mut rx) = pair(cap);
            let batch = [0u64; BULK_BATCH];
            let mut out = [0u64; BULK_BATCH];
            batched(ITEMS_PER_SAMPLE / BULK_BATCH, |_| {
                let n = tx.try_enqueue_bulk(&batch);
                (n + rx.try_dequeue_bulk(&mut out)) as u64
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let batch = [0u64; BULK_BATCH];
        let mut out = [0u64; BULK_BATCH];
        let (mut tx, mut rx) = pair(CANON);
        let enq = single(
            |_| tx.try_enqueue_bulk(&batch) as u64,
            |_| rx.try_dequeue_bulk(&mut out) as u64,
        );
        let (mut tx, mut rx) = pair(CANON);
        let deq = single(
            |_| rx.try_dequeue_bulk(&mut out) as u64,
            |_| tx.try_enqueue_bulk(&batch) as u64,
        );
        let mut p99 = BTreeMap::new();
        p99.insert("enqueue_bulk".to_string(), enq);
        p99.insert("dequeue_bulk".to_string(), deq);
        manifest.set_feature("bulk", cat, &p99, &reason);
    }

    // ---------- wait-strategies: blocking wrappers, fast path only ----------
    #[cfg(feature = "wait-strategies")]
    {
        use subms_spsc_ring_buffer::{
            BlockingSpscConsumer, BlockingSpscProducer, BusySpin, ParkStrategy,
        };
        // Swept on BusySpin, whose `signal()` is a no-op, so the curve is the
        // wrapper's own overhead over the base ring and nothing else. The ring
        // is never full or empty, so `wait()` is never called; the parked-wakeup
        // path is a scheduler latency, not a per-op cost, and is not measured.
        let sw = sweep("wait-strategies/push+pop", |cap| {
            let (tx, rx) = pair(cap);
            let mut p = BlockingSpscProducer::new(tx, BusySpin);
            let mut c = BlockingSpscConsumer::new(rx, BusySpin);
            batched(ITEMS_PER_SAMPLE, |i| {
                p.push(i as u64);
                c.pop()
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        // ParkStrategy on the SAME ready ring, printed rather than classified.
        // Its `signal()` fires on every successful op even when nothing is
        // parked, and it takes a lock to do it, so the strategy choice shows up
        // here on the fast path - not in `wait()`, which this never enters.
        let park_batched = best(|| {
            let (ps, cs) = ParkStrategy::pair();
            let (tx, rx) = pair(CANON);
            let mut p = BlockingSpscProducer::new(tx, ps);
            let mut c = BlockingSpscConsumer::new(rx, cs);
            batched(ITEMS_PER_SAMPLE, |i| {
                p.push(i as u64);
                c.pop()
            })
        });
        eprintln!(
            "wait-strategies park fast path at {CANON}: {park_batched}ns per \
             {ITEMS_PER_SAMPLE}-item sample (spin {}ns) - signal() takes a lock per op",
            sw.iter().find(|(n, _)| *n == CANON).map_or(0, |(_, v)| *v)
        );

        let (tx, rx) = pair(CANON);
        let mut p = BlockingSpscProducer::new(tx, BusySpin);
        let mut c = BlockingSpscConsumer::new(rx, BusySpin);
        let push_spin = single(
            |i| {
                p.push(i as u64);
                0
            },
            |_| c.pop(),
        );
        let (tx, rx) = pair(CANON);
        let mut p = BlockingSpscProducer::new(tx, BusySpin);
        let mut c = BlockingSpscConsumer::new(rx, BusySpin);
        let pop_spin = single(
            |_| c.pop(),
            |i| {
                p.push(i as u64);
                0
            },
        );

        let (ps, cs) = ParkStrategy::pair();
        let (tx, rx) = pair(CANON);
        let mut p = BlockingSpscProducer::new(tx, ps);
        let mut c = BlockingSpscConsumer::new(rx, cs);
        let push_park = single(
            |i| {
                p.push(i as u64);
                0
            },
            |_| c.pop(),
        );
        let (ps, cs) = ParkStrategy::pair();
        let (tx, rx) = pair(CANON);
        let mut p = BlockingSpscProducer::new(tx, ps);
        let mut c = BlockingSpscConsumer::new(rx, cs);
        let pop_park = single(
            |_| c.pop(),
            |i| {
                p.push(i as u64);
                0
            },
        );

        let mut p99 = BTreeMap::new();
        p99.insert("push_spin".to_string(), push_spin);
        p99.insert("pop_spin".to_string(), pop_spin);
        p99.insert("push_park".to_string(), push_park);
        p99.insert("pop_park".to_string(), pop_park);
        manifest.set_feature("wait-strategies", cat, &p99, &reason);
    }

    // ---------- mpsc-fan-in: N rings, one round-robin consumer ----------
    #[cfg(feature = "mpsc-fan-in")]
    {
        use subms_spsc_ring_buffer::{MpscFanIn, MpscFanInConsumer, MpscFanInProducer};
        /// Producer count is held FIXED across the sweep. Moving it would sweep
        /// the fan-in width, which sets the consumer's probe loop, not the ring.
        const PRODUCERS: usize = 4;

        fn fanin(cap: usize) -> (Vec<MpscFanInProducer<u64>>, MpscFanInConsumer<u64>) {
            let (mut ps, c) = MpscFanIn::with_capacity::<u64>(PRODUCERS, cap);
            for p in ps.iter_mut() {
                for i in 0..cap / 2 {
                    let _ = p.try_push(i as u64);
                }
            }
            (ps, c)
        }

        // Pushes round-robin and the consumer cursor advances one ring per pop,
        // so the two stay in step and every ring holds a constant half load.
        // With every ring non-empty the consumer's probe hits on its first try,
        // which is the steady-state shape; a starved fan-in probes all N and that
        // is a different measurement.
        let sw = sweep("mpsc-fan-in/push+pop", |cap| {
            let (mut ps, mut c) = fanin(cap);
            batched(ITEMS_PER_SAMPLE, |i| {
                let _ = ps[i % PRODUCERS].try_push(i as u64);
                c.try_pop().unwrap_or(0)
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let (mut ps, mut c) = fanin(CANON);
        let enq = single(
            |i| {
                let _ = ps[i % PRODUCERS].try_push(i as u64);
                0
            },
            |_| c.try_pop().unwrap_or(0),
        );
        let (mut ps, mut c) = fanin(CANON);
        let deq = single(
            |_| c.try_pop().unwrap_or(0),
            |i| {
                let _ = ps[i % PRODUCERS].try_push(i as u64);
                0
            },
        );
        let mut p99 = BTreeMap::new();
        p99.insert("fanin_enqueue".to_string(), enq);
        p99.insert("fanin_dequeue".to_string(), deq);
        manifest.set_feature("mpsc-fan-in", cat, &p99, &reason);
    }

    // ---------- mpmc-disruptor: CAS claim + sequence barrier ----------
    #[cfg(feature = "mpmc-disruptor")]
    {
        use subms_spsc_ring_buffer::{DisruptorConsumer, DisruptorProducer, MpmcDisruptor};
        /// One consumer, held fixed. `try_publish` scans every consumer cursor
        /// before claiming, so the consumer count - not the capacity - is what
        /// that loop is O(). Sweeping it would answer a different question.
        const CONSUMERS: usize = 1;

        fn disruptor(cap: usize) -> (DisruptorProducer<u64>, DisruptorConsumer<u64>) {
            let (p, mut cs) = MpmcDisruptor::with_consumers::<u64>(cap, CONSUMERS);
            let c = cs.remove(0);
            for i in 0..cap / 2 {
                let _ = p.try_publish(i as u64);
            }
            (p, c)
        }

        // Half a ring of published-but-unconsumed items keeps the producer clear
        // of the gating spin (it only fires within one capacity of the slowest
        // consumer) and the consumer clear of the unpublished early return.
        let sw = sweep("mpmc-disruptor/publish+consume", |cap| {
            let (p, mut c) = disruptor(cap);
            batched(ITEMS_PER_SAMPLE, |i| {
                let _ = p.try_publish(i as u64);
                c.try_consume().unwrap_or(0)
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let (p, mut c) = disruptor(CANON);
        let published = single(
            |i| {
                let _ = p.try_publish(i as u64);
                0
            },
            |_| c.try_consume().unwrap_or(0),
        );
        let (p, mut c) = disruptor(CANON);
        let consumed = single(
            |_| c.try_consume().unwrap_or(0),
            |i| {
                let _ = p.try_publish(i as u64);
                0
            },
        );
        let mut p99 = BTreeMap::new();
        p99.insert("publish".to_string(), published);
        p99.insert("consume".to_string(), consumed);
        manifest.set_feature("mpmc-disruptor", cat, &p99, &reason);
    }

    // ---------- metrics: counters on the push / pop path ----------
    #[cfg(feature = "metrics")]
    {
        use subms_spsc_ring_buffer::InstrumentedSpsc;

        // The wrapper adds a `fetch_add` per op plus, on the producer side, a
        // high-water-mark update. `local_depth` only ever increments - a producer
        // cannot observe pops - so that update takes its CAS on every push rather
        // than settling once the mark stops moving. The counter cost on the push
        // path is two read-modify-writes, not one.
        let sw = sweep("metrics/push+pop", |cap| {
            let (tx, rx) = pair(cap);
            let (mut tx, mut rx, _m) = InstrumentedSpsc::wrap(tx, rx);
            batched(ITEMS_PER_SAMPLE, |i| {
                let _ = tx.try_push(i as u64);
                rx.try_pop().unwrap_or(0)
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let (tx, rx) = pair(CANON);
        let (mut tx, mut rx, _m) = InstrumentedSpsc::wrap(tx, rx);
        let enq = single(
            |i| {
                let _ = tx.try_push(i as u64);
                0
            },
            |_| rx.try_pop().unwrap_or(0),
        );
        let (tx, rx) = pair(CANON);
        let (mut tx, mut rx, _m) = InstrumentedSpsc::wrap(tx, rx);
        let deq = single(
            |_| rx.try_pop().unwrap_or(0),
            |i| {
                let _ = tx.try_push(i as u64);
                0
            },
        );
        let mut p99 = BTreeMap::new();
        p99.insert("metrics_enqueue".to_string(), enq);
        p99.insert("metrics_dequeue".to_string(), deq);
        manifest.set_feature("metrics", cat, &p99, &reason);
    }

    std::fs::create_dir_all(path.parent().unwrap())?;
    std::fs::write(&path, manifest.to_json())?;
    io::stdout().write_all(manifest.to_json().as_bytes())?;
    Ok(())
}

impl<T> Producer<T>

pub fn try_push(&mut self, value: T) -> Result<(), T>

Push a value. Returns the input back as Err(value) if the buffer is full. Wait-free: at most one atomic load + one atomic store.

Examples found in repository

examples/perf_features.rs (line 184)

fn pair(cap: usize) -> (Producer<u64>, Consumer<u64>) {
    let (mut tx, rx) = SpscRingBuffer::with_capacity::<u64>(cap);
    for i in 0..cap / 2 {
        let _ = tx.try_push(i as u64);
    }
    (tx, rx)
}

fn main() -> io::Result<()> {
    let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
        .join("..")
        .join(".subms")
        .join("features")
        .join("rust.json");
    let existing = std::fs::read_to_string(&path).unwrap_or_default();
    let mut manifest = SubMsFeatureManifest::load_str("rust", &existing);
    // Stamp the box these numbers came from. The bench runs wherever it is
    // invoked, so an unstamped manifest is indistinguishable from a fleet
    // capture; the renderer will not publish one it cannot attribute.
    let (source, instance) = SubMsP99Source::from_env();
    manifest.set_p99_source(source, instance.as_deref());

    // The baseline: the base wait-free push + pop round trip. Swept as well as
    // sampled, because whether ring capacity moves the BASE op is the context
    // every feature curve is read against.
    let base_sweep = sweep("base/push+pop", |cap| {
        let (mut tx, mut rx) = pair(cap);
        batched(ITEMS_PER_SAMPLE, |i| {
            let _ = tx.try_push(i as u64);
            rx.try_pop().unwrap_or(0)
        })
    });
    let base_p50 = base_sweep
        .iter()
        .find(|(n, _)| *n == CANON)
        .map_or(0, |(_, v)| *v);
    eprintln!("base push+pop p50 per {ITEMS_PER_SAMPLE}-item sample: {base_p50}ns");

    // ---------- bulk: one fence per BULK_BATCH items ----------
    #[cfg(feature = "bulk")]
    {
        // A sample moves ITEMS_PER_SAMPLE items either way; only the call width
        // differs. That is the comparison the feature exists to win, and it is
        // why the reps count is divided rather than the batch grown.
        let sw = sweep("bulk/enqueue+dequeue", |cap| {
            let (mut tx, mut rx) = pair(cap);
            let batch = [0u64; BULK_BATCH];
            let mut out = [0u64; BULK_BATCH];
            batched(ITEMS_PER_SAMPLE / BULK_BATCH, |_| {
                let n = tx.try_enqueue_bulk(&batch);
                (n + rx.try_dequeue_bulk(&mut out)) as u64
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let batch = [0u64; BULK_BATCH];
        let mut out = [0u64; BULK_BATCH];
        let (mut tx, mut rx) = pair(CANON);
        let enq = single(
            |_| tx.try_enqueue_bulk(&batch) as u64,
            |_| rx.try_dequeue_bulk(&mut out) as u64,
        );
        let (mut tx, mut rx) = pair(CANON);
        let deq = single(
            |_| rx.try_dequeue_bulk(&mut out) as u64,
            |_| tx.try_enqueue_bulk(&batch) as u64,
        );
        let mut p99 = BTreeMap::new();
        p99.insert("enqueue_bulk".to_string(), enq);
        p99.insert("dequeue_bulk".to_string(), deq);
        manifest.set_feature("bulk", cat, &p99, &reason);
    }

    // ---------- wait-strategies: blocking wrappers, fast path only ----------
    #[cfg(feature = "wait-strategies")]
    {
        use subms_spsc_ring_buffer::{
            BlockingSpscConsumer, BlockingSpscProducer, BusySpin, ParkStrategy,
        };
        // Swept on BusySpin, whose `signal()` is a no-op, so the curve is the
        // wrapper's own overhead over the base ring and nothing else. The ring
        // is never full or empty, so `wait()` is never called; the parked-wakeup
        // path is a scheduler latency, not a per-op cost, and is not measured.
        let sw = sweep("wait-strategies/push+pop", |cap| {
            let (tx, rx) = pair(cap);
            let mut p = BlockingSpscProducer::new(tx, BusySpin);
            let mut c = BlockingSpscConsumer::new(rx, BusySpin);
            batched(ITEMS_PER_SAMPLE, |i| {
                p.push(i as u64);
                c.pop()
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        // ParkStrategy on the SAME ready ring, printed rather than classified.
        // Its `signal()` fires on every successful op even when nothing is
        // parked, and it takes a lock to do it, so the strategy choice shows up
        // here on the fast path - not in `wait()`, which this never enters.
        let park_batched = best(|| {
            let (ps, cs) = ParkStrategy::pair();
            let (tx, rx) = pair(CANON);
            let mut p = BlockingSpscProducer::new(tx, ps);
            let mut c = BlockingSpscConsumer::new(rx, cs);
            batched(ITEMS_PER_SAMPLE, |i| {
                p.push(i as u64);
                c.pop()
            })
        });
        eprintln!(
            "wait-strategies park fast path at {CANON}: {park_batched}ns per \
             {ITEMS_PER_SAMPLE}-item sample (spin {}ns) - signal() takes a lock per op",
            sw.iter().find(|(n, _)| *n == CANON).map_or(0, |(_, v)| *v)
        );

        let (tx, rx) = pair(CANON);
        let mut p = BlockingSpscProducer::new(tx, BusySpin);
        let mut c = BlockingSpscConsumer::new(rx, BusySpin);
        let push_spin = single(
            |i| {
                p.push(i as u64);
                0
            },
            |_| c.pop(),
        );
        let (tx, rx) = pair(CANON);
        let mut p = BlockingSpscProducer::new(tx, BusySpin);
        let mut c = BlockingSpscConsumer::new(rx, BusySpin);
        let pop_spin = single(
            |_| c.pop(),
            |i| {
                p.push(i as u64);
                0
            },
        );

        let (ps, cs) = ParkStrategy::pair();
        let (tx, rx) = pair(CANON);
        let mut p = BlockingSpscProducer::new(tx, ps);
        let mut c = BlockingSpscConsumer::new(rx, cs);
        let push_park = single(
            |i| {
                p.push(i as u64);
                0
            },
            |_| c.pop(),
        );
        let (ps, cs) = ParkStrategy::pair();
        let (tx, rx) = pair(CANON);
        let mut p = BlockingSpscProducer::new(tx, ps);
        let mut c = BlockingSpscConsumer::new(rx, cs);
        let pop_park = single(
            |_| c.pop(),
            |i| {
                p.push(i as u64);
                0
            },
        );

        let mut p99 = BTreeMap::new();
        p99.insert("push_spin".to_string(), push_spin);
        p99.insert("pop_spin".to_string(), pop_spin);
        p99.insert("push_park".to_string(), push_park);
        p99.insert("pop_park".to_string(), pop_park);
        manifest.set_feature("wait-strategies", cat, &p99, &reason);
    }

    // ---------- mpsc-fan-in: N rings, one round-robin consumer ----------
    #[cfg(feature = "mpsc-fan-in")]
    {
        use subms_spsc_ring_buffer::{MpscFanIn, MpscFanInConsumer, MpscFanInProducer};
        /// Producer count is held FIXED across the sweep. Moving it would sweep
        /// the fan-in width, which sets the consumer's probe loop, not the ring.
        const PRODUCERS: usize = 4;

        fn fanin(cap: usize) -> (Vec<MpscFanInProducer<u64>>, MpscFanInConsumer<u64>) {
            let (mut ps, c) = MpscFanIn::with_capacity::<u64>(PRODUCERS, cap);
            for p in ps.iter_mut() {
                for i in 0..cap / 2 {
                    let _ = p.try_push(i as u64);
                }
            }
            (ps, c)
        }

        // Pushes round-robin and the consumer cursor advances one ring per pop,
        // so the two stay in step and every ring holds a constant half load.
        // With every ring non-empty the consumer's probe hits on its first try,
        // which is the steady-state shape; a starved fan-in probes all N and that
        // is a different measurement.
        let sw = sweep("mpsc-fan-in/push+pop", |cap| {
            let (mut ps, mut c) = fanin(cap);
            batched(ITEMS_PER_SAMPLE, |i| {
                let _ = ps[i % PRODUCERS].try_push(i as u64);
                c.try_pop().unwrap_or(0)
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let (mut ps, mut c) = fanin(CANON);
        let enq = single(
            |i| {
                let _ = ps[i % PRODUCERS].try_push(i as u64);
                0
            },
            |_| c.try_pop().unwrap_or(0),
        );
        let (mut ps, mut c) = fanin(CANON);
        let deq = single(
            |_| c.try_pop().unwrap_or(0),
            |i| {
                let _ = ps[i % PRODUCERS].try_push(i as u64);
                0
            },
        );
        let mut p99 = BTreeMap::new();
        p99.insert("fanin_enqueue".to_string(), enq);
        p99.insert("fanin_dequeue".to_string(), deq);
        manifest.set_feature("mpsc-fan-in", cat, &p99, &reason);
    }

    // ---------- mpmc-disruptor: CAS claim + sequence barrier ----------
    #[cfg(feature = "mpmc-disruptor")]
    {
        use subms_spsc_ring_buffer::{DisruptorConsumer, DisruptorProducer, MpmcDisruptor};
        /// One consumer, held fixed. `try_publish` scans every consumer cursor
        /// before claiming, so the consumer count - not the capacity - is what
        /// that loop is O(). Sweeping it would answer a different question.
        const CONSUMERS: usize = 1;

        fn disruptor(cap: usize) -> (DisruptorProducer<u64>, DisruptorConsumer<u64>) {
            let (p, mut cs) = MpmcDisruptor::with_consumers::<u64>(cap, CONSUMERS);
            let c = cs.remove(0);
            for i in 0..cap / 2 {
                let _ = p.try_publish(i as u64);
            }
            (p, c)
        }

        // Half a ring of published-but-unconsumed items keeps the producer clear
        // of the gating spin (it only fires within one capacity of the slowest
        // consumer) and the consumer clear of the unpublished early return.
        let sw = sweep("mpmc-disruptor/publish+consume", |cap| {
            let (p, mut c) = disruptor(cap);
            batched(ITEMS_PER_SAMPLE, |i| {
                let _ = p.try_publish(i as u64);
                c.try_consume().unwrap_or(0)
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let (p, mut c) = disruptor(CANON);
        let published = single(
            |i| {
                let _ = p.try_publish(i as u64);
                0
            },
            |_| c.try_consume().unwrap_or(0),
        );
        let (p, mut c) = disruptor(CANON);
        let consumed = single(
            |_| c.try_consume().unwrap_or(0),
            |i| {
                let _ = p.try_publish(i as u64);
                0
            },
        );
        let mut p99 = BTreeMap::new();
        p99.insert("publish".to_string(), published);
        p99.insert("consume".to_string(), consumed);
        manifest.set_feature("mpmc-disruptor", cat, &p99, &reason);
    }

    // ---------- metrics: counters on the push / pop path ----------
    #[cfg(feature = "metrics")]
    {
        use subms_spsc_ring_buffer::InstrumentedSpsc;

        // The wrapper adds a `fetch_add` per op plus, on the producer side, a
        // high-water-mark update. `local_depth` only ever increments - a producer
        // cannot observe pops - so that update takes its CAS on every push rather
        // than settling once the mark stops moving. The counter cost on the push
        // path is two read-modify-writes, not one.
        let sw = sweep("metrics/push+pop", |cap| {
            let (tx, rx) = pair(cap);
            let (mut tx, mut rx, _m) = InstrumentedSpsc::wrap(tx, rx);
            batched(ITEMS_PER_SAMPLE, |i| {
                let _ = tx.try_push(i as u64);
                rx.try_pop().unwrap_or(0)
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let (tx, rx) = pair(CANON);
        let (mut tx, mut rx, _m) = InstrumentedSpsc::wrap(tx, rx);
        let enq = single(
            |i| {
                let _ = tx.try_push(i as u64);
                0
            },
            |_| rx.try_pop().unwrap_or(0),
        );
        let (tx, rx) = pair(CANON);
        let (mut tx, mut rx, _m) = InstrumentedSpsc::wrap(tx, rx);
        let deq = single(
            |_| rx.try_pop().unwrap_or(0),
            |i| {
                let _ = tx.try_push(i as u64);
                0
            },
        );
        let mut p99 = BTreeMap::new();
        p99.insert("metrics_enqueue".to_string(), enq);
        p99.insert("metrics_dequeue".to_string(), deq);
        manifest.set_feature("metrics", cat, &p99, &reason);
    }

    std::fs::create_dir_all(path.parent().unwrap())?;
    std::fs::write(&path, manifest.to_json())?;
    io::stdout().write_all(manifest.to_json().as_bytes())?;
    Ok(())
}

examples/demo.rs (line 9)

fn main() {
    let (mut tx, mut rx) = SpscRingBuffer::with_capacity::<u32>(16);

    let producer = thread::spawn(move || {
        for i in 0..10u32 {
            while tx.try_push(i).is_err() {}
        }
    });

    let consumer = thread::spawn(move || {
        let mut seen = Vec::new();
        while seen.len() < 10 {
            if let Some(v) = rx.try_pop() {
                seen.push(v);
            }
        }
        seen
    });

    producer.join().unwrap();
    let seen = consumer.join().unwrap();
    println!("consumed: {seen:?}");
}

examples/sample_app.rs (line 61)

fn base_feed_to_strategy() {
    println!("== base: feed-handler -> strategy handoff ==");

    // Drop-on-full is the caller's decision. Shown deterministically on a
    // small ring: capacity 4, six ticks offered, the last two are dropped.
    let (mut tx, mut rx) = SpscRingBuffer::with_capacity::<Tick>(4);
    let mut dropped = 0usize;
    for seq in 0..6u64 {
        let tick = Tick {
            seq,
            price_cents: 10_000 + seq as u32,
        };
        if tx.try_push(tick).is_err() {
            dropped += 1;
        }
    }
    println!("  cap-4 ring, 6 offered -> {dropped} dropped under backpressure");
    assert_eq!(
        dropped, 2,
        "two ticks past capacity are dropped, not blocked"
    );

    // Occupancy is what a queue-depth alarm reads, and peek lets the strategy
    // inspect the oldest tick before deciding to consume it.
    let oldest = rx.peek().expect("ring is full").seq;
    println!(
        "  depth {}/{} full={}, oldest queued seq {oldest}",
        rx.len(),
        rx.capacity(),
        tx.is_full()
    );
    println!("  dropped {} stale ticks on resync", rx.clear());

    // Steady state: a drained ring loses nothing and preserves feed order.
    let n = 50_000u64;
    let (mut tx, mut rx) = SpscRingBuffer::with_capacity::<Tick>(1024);
    let feed = thread::spawn(move || {
        for seq in 0..n {
            let tick = Tick {
                seq,
                price_cents: 10_000 + (seq % 500) as u32,
            };
            while tx.try_push(tick).is_err() {
                std::hint::spin_loop();
            }
        }
    });
    let strategy = thread::spawn(move || {
        let mut expected = 0u64;
        while expected < n {
            if let Some(tick) = rx.try_pop() {
                assert_eq!(tick.seq, expected, "ticks arrive in feed order");
                expected += 1;
            }
        }
        expected
    });
    feed.join().unwrap();
    let received = strategy.join().unwrap();
    println!("  streamed {received} ticks in order, zero loss when drained");
    assert_eq!(received, n);
}

pub fn capacity(&self) -> usize

Total slot count (power of two; not the requested capacity).

pub fn len(&self) -> usize

Items currently buffered. A snapshot: the consumer runs concurrently, so the true count can only be lower by the time the caller acts on it. Use it for occupancy alarms and sizing, never to decide whether a push will succeed - try_push already answers that without a race.

pub fn is_empty(&self) -> bool

True when no items are buffered. Snapshot semantics, as Producer::len.

pub fn is_full(&self) -> bool

True when the ring holds capacity items. Snapshot semantics.

Examples found in repository

examples/sample_app.rs (line 78)

fn base_feed_to_strategy() {
    println!("== base: feed-handler -> strategy handoff ==");

    // Drop-on-full is the caller's decision. Shown deterministically on a
    // small ring: capacity 4, six ticks offered, the last two are dropped.
    let (mut tx, mut rx) = SpscRingBuffer::with_capacity::<Tick>(4);
    let mut dropped = 0usize;
    for seq in 0..6u64 {
        let tick = Tick {
            seq,
            price_cents: 10_000 + seq as u32,
        };
        if tx.try_push(tick).is_err() {
            dropped += 1;
        }
    }
    println!("  cap-4 ring, 6 offered -> {dropped} dropped under backpressure");
    assert_eq!(
        dropped, 2,
        "two ticks past capacity are dropped, not blocked"
    );

    // Occupancy is what a queue-depth alarm reads, and peek lets the strategy
    // inspect the oldest tick before deciding to consume it.
    let oldest = rx.peek().expect("ring is full").seq;
    println!(
        "  depth {}/{} full={}, oldest queued seq {oldest}",
        rx.len(),
        rx.capacity(),
        tx.is_full()
    );
    println!("  dropped {} stale ticks on resync", rx.clear());

    // Steady state: a drained ring loses nothing and preserves feed order.
    let n = 50_000u64;
    let (mut tx, mut rx) = SpscRingBuffer::with_capacity::<Tick>(1024);
    let feed = thread::spawn(move || {
        for seq in 0..n {
            let tick = Tick {
                seq,
                price_cents: 10_000 + (seq % 500) as u32,
            };
            while tx.try_push(tick).is_err() {
                std::hint::spin_loop();
            }
        }
    });
    let strategy = thread::spawn(move || {
        let mut expected = 0u64;
        while expected < n {
            if let Some(tick) = rx.try_pop() {
                assert_eq!(tick.seq, expected, "ticks arrive in feed order");
                expected += 1;
            }
        }
        expected
    });
    feed.join().unwrap();
    let received = strategy.join().unwrap();
    println!("  streamed {received} ticks in order, zero loss when drained");
    assert_eq!(received, n);
}

Trait Implementations

impl<T: Send> Send for Producer<T>