Struct DisruptorConsumer 

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

One consumer side. Each consumer sees every published item.

Implementations

impl<T: Clone> DisruptorConsumer<T>

pub fn try_consume(&mut self) -> Option<T>

Returns a clone of the next published item, or None if not ready. Clone because each consumer of a broadcast disruptor sees the same item; we cannot move out of the slot.

Examples found in repository

examples/sample_app.rs (line 252)

fn broadcast_to_strategy_and_risk() {
    use subms_spsc_ring_buffer::MpmcDisruptor;

    println!("\n== mpmc-disruptor: broadcast to strategy + risk ==");
    let n = 8u64;
    let (producer, mut consumers) = MpmcDisruptor::with_consumers::<Tick>(16, 2);
    let (strategy, rest) = consumers.split_at_mut(1);
    let strategy = &mut strategy[0];
    let risk = &mut rest[0];

    // Small and single-threaded so the tour self-verifies; the threaded
    // broadcast path is pinned in the tests.
    let mut published = 0u64;
    let mut strat_seen = Vec::new();
    let mut risk_seen = Vec::new();
    while published < n {
        while published < n
            && producer
                .try_publish(Tick {
                    seq: published,
                    price_cents: 50_000,
                })
                .is_ok()
        {
            published += 1;
        }
        while let Some(t) = strategy.try_consume() {
            strat_seen.push(t.seq);
        }
        while let Some(t) = risk.try_consume() {
            risk_seen.push(t.seq);
        }
    }
    println!(
        "  published {published}; strategy saw {}, risk saw {}",
        strat_seen.len(),
        risk_seen.len()
    );
    let expected: Vec<u64> = (0..n).collect();
    assert_eq!(strat_seen, expected, "strategy sees every tick");
    assert_eq!(risk_seen, expected, "risk monitor sees every tick too");
}

examples/perf_features.rs (line 426)

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(())
}

pub fn capacity(&self) -> usize

Trait Implementations

impl<T: Send> Send for DisruptorConsumer<T>