Struct MpmcDisruptor 

pub struct MpmcDisruptor;

Builder + constructor. Use MpmcDisruptor::with_consumers.

Implementations

impl MpmcDisruptor

pub fn with_consumers<T: Send + 'static>( requested_capacity: usize, consumer_count: usize, ) -> (DisruptorProducer<T>, Vec<DisruptorConsumer<T>>)

Build a disruptor with consumer_count consumers and at least requested_capacity slots (rounded up to power of two, floor 2). All consumers see every published item (broadcast); for work-stealing semantics use MpscFanIn with a single consumer per producer instead.

Examples found in repository

examples/perf_features.rs (line 411)

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

examples/sample_app.rs (line 231)

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");
}