Struct MpmcQueue 

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

Bounded MPMC ring queue. Capacity is rounded up to the next power of two (minimum 2).

Implementations

impl<T> MpmcQueue<T>

pub fn new(capacity: usize) -> Self

Examples found in repository

examples/perf_features.rs (line 317)

        fn ring(n: usize) -> MpmcQueue<u64> {
            let q = MpmcQueue::new(n);
            for i in 0..n / 2 {
                let _ = q.try_enqueue(i as u64);
            }
            q
        }

examples/sample_app.rs (line 133)

fn mpmc_sharded_match() {
    use std::sync::atomic::{AtomicUsize, Ordering};

    use subms_mpsc_queue::MpmcQueue;

    println!("\n== mpmc: shard the match loop across several consumers ==");
    let shards = 3usize;
    let ring: Arc<MpmcQueue<u64>> = Arc::new(MpmcQueue::new(1_024));
    let total = GATEWAYS * ORDERS_PER_GATEWAY;

    let gateways: Vec<_> = (0..GATEWAYS)
        .map(|g| {
            let ring = Arc::clone(&ring);
            thread::spawn(move || {
                for seq in 0..ORDERS_PER_GATEWAY {
                    let mut order = order_id(g, seq);
                    while let Err(rejected) = ring.try_enqueue(order) {
                        order = rejected;
                        std::hint::spin_loop();
                    }
                }
            })
        })
        .collect();

    let matched = Arc::new(AtomicUsize::new(0));
    let consumers: Vec<_> = (0..shards)
        .map(|_| {
            let ring = Arc::clone(&ring);
            let matched = Arc::clone(&matched);
            thread::spawn(move || {
                let mut local = 0usize;
                loop {
                    if ring.try_dequeue().is_some() {
                        local += 1;
                        matched.fetch_add(1, Ordering::Relaxed);
                    } else if matched.load(Ordering::Relaxed) >= total {
                        break;
                    } else {
                        std::hint::spin_loop();
                    }
                }
                local
            })
        })
        .collect();

    for h in gateways {
        h.join().unwrap();
    }
    let drained: usize = consumers.into_iter().map(|c| c.join().unwrap()).sum();
    // cas_retries() is the contention read-out, deliberately not printed: it is
    // a property of how the OS scheduled these threads on this run.
    println!(
        "  {shards} shards drained {drained} orders, ring empty: {}",
        ring.is_empty()
    );
    assert_eq!(
        drained, total,
        "shards together drain every order exactly once"
    );
    assert_eq!(
        ring.producer_index(),
        ring.consumer_index(),
        "every claimed slot was consumed"
    );
}

pub fn capacity(&self) -> usize

pub fn producer_index(&self) -> usize

Monotonic count of slots ever claimed by producers.

Examples found in repository

examples/sample_app.rs (line 188)

fn mpmc_sharded_match() {
    use std::sync::atomic::{AtomicUsize, Ordering};

    use subms_mpsc_queue::MpmcQueue;

    println!("\n== mpmc: shard the match loop across several consumers ==");
    let shards = 3usize;
    let ring: Arc<MpmcQueue<u64>> = Arc::new(MpmcQueue::new(1_024));
    let total = GATEWAYS * ORDERS_PER_GATEWAY;

    let gateways: Vec<_> = (0..GATEWAYS)
        .map(|g| {
            let ring = Arc::clone(&ring);
            thread::spawn(move || {
                for seq in 0..ORDERS_PER_GATEWAY {
                    let mut order = order_id(g, seq);
                    while let Err(rejected) = ring.try_enqueue(order) {
                        order = rejected;
                        std::hint::spin_loop();
                    }
                }
            })
        })
        .collect();

    let matched = Arc::new(AtomicUsize::new(0));
    let consumers: Vec<_> = (0..shards)
        .map(|_| {
            let ring = Arc::clone(&ring);
            let matched = Arc::clone(&matched);
            thread::spawn(move || {
                let mut local = 0usize;
                loop {
                    if ring.try_dequeue().is_some() {
                        local += 1;
                        matched.fetch_add(1, Ordering::Relaxed);
                    } else if matched.load(Ordering::Relaxed) >= total {
                        break;
                    } else {
                        std::hint::spin_loop();
                    }
                }
                local
            })
        })
        .collect();

    for h in gateways {
        h.join().unwrap();
    }
    let drained: usize = consumers.into_iter().map(|c| c.join().unwrap()).sum();
    // cas_retries() is the contention read-out, deliberately not printed: it is
    // a property of how the OS scheduled these threads on this run.
    println!(
        "  {shards} shards drained {drained} orders, ring empty: {}",
        ring.is_empty()
    );
    assert_eq!(
        drained, total,
        "shards together drain every order exactly once"
    );
    assert_eq!(
        ring.producer_index(),
        ring.consumer_index(),
        "every claimed slot was consumed"
    );
}

pub fn consumer_index(&self) -> usize

Monotonic count of slots ever claimed by consumers.

Examples found in repository

examples/sample_app.rs (line 189)

fn mpmc_sharded_match() {
    use std::sync::atomic::{AtomicUsize, Ordering};

    use subms_mpsc_queue::MpmcQueue;

    println!("\n== mpmc: shard the match loop across several consumers ==");
    let shards = 3usize;
    let ring: Arc<MpmcQueue<u64>> = Arc::new(MpmcQueue::new(1_024));
    let total = GATEWAYS * ORDERS_PER_GATEWAY;

    let gateways: Vec<_> = (0..GATEWAYS)
        .map(|g| {
            let ring = Arc::clone(&ring);
            thread::spawn(move || {
                for seq in 0..ORDERS_PER_GATEWAY {
                    let mut order = order_id(g, seq);
                    while let Err(rejected) = ring.try_enqueue(order) {
                        order = rejected;
                        std::hint::spin_loop();
                    }
                }
            })
        })
        .collect();

    let matched = Arc::new(AtomicUsize::new(0));
    let consumers: Vec<_> = (0..shards)
        .map(|_| {
            let ring = Arc::clone(&ring);
            let matched = Arc::clone(&matched);
            thread::spawn(move || {
                let mut local = 0usize;
                loop {
                    if ring.try_dequeue().is_some() {
                        local += 1;
                        matched.fetch_add(1, Ordering::Relaxed);
                    } else if matched.load(Ordering::Relaxed) >= total {
                        break;
                    } else {
                        std::hint::spin_loop();
                    }
                }
                local
            })
        })
        .collect();

    for h in gateways {
        h.join().unwrap();
    }
    let drained: usize = consumers.into_iter().map(|c| c.join().unwrap()).sum();
    // cas_retries() is the contention read-out, deliberately not printed: it is
    // a property of how the OS scheduled these threads on this run.
    println!(
        "  {shards} shards drained {drained} orders, ring empty: {}",
        ring.is_empty()
    );
    assert_eq!(
        drained, total,
        "shards together drain every order exactly once"
    );
    assert_eq!(
        ring.producer_index(),
        ring.consumer_index(),
        "every claimed slot was consumed"
    );
}

pub fn cas_retries(&self) -> u64

Total CAS retries (both producers losing tail-CAS and consumers losing head-CAS). Useful for diagnosing contention; ignored by the hot path otherwise.

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

Multi-producer enqueue. Returns Err(value) if the ring is full.

Examples found in repository

examples/perf_features.rs (line 319)

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

    // Optional, off by default: pin this thread to one core for the whole run.
    // On a heterogeneous laptop the scheduler moves the bench between core
    // clusters and every measurement lands in one of two clock states 1.31x
    // apart - a spread three times wider than the deltas being classified, and
    // large enough on its own to flip a feature between auxiliary and hot-path.
    // Pinned, the same sweep repeats to within 1%. Left OFF by default because a
    // fleet box isolates cores outside the process, and pinning from in here
    // would override that placement with a core the orchestrator did not choose.
    #[cfg(feature = "affinity")]
    if let Some(core) = std::env::var("SUBMS_PIN").ok().and_then(|v| v.parse().ok()) {
        let _ = subms_mpsc_queue::set_affinity(&[core]);
    }

    // Burn before the first measurement, not just before each one. Every
    // `batched` call warms itself, but the FIRST measurement in the process pays
    // a ramp the per-measurement warm sits inside rather than absorbs, and the
    // sweep runs smallest-first: without this the base curve read 71800 / 45300 /
    // 46500 ns, a 1.6x fall with size that is the process settling, not the
    // queue.
    {
        let mut q = filled(CANON);
        let start = std::time::Instant::now();
        while (start.elapsed().as_nanos() as u64) < BURN_NANOS {
            for i in 0..ITEMS_PER_SAMPLE {
                q.push(i as u64);
                black_box(q.try_pop());
            }
        }
    }

    // The baseline: the base queue's push + try_pop round trip. Swept as well as
    // sampled, because whether queue depth moves the BASE op is the context
    // every feature curve is read against.
    let base_sweep = sweep("base/push+pop", |n| {
        let mut q = filled(n);
        batched(ITEMS_PER_SAMPLE, |i| {
            q.push(i as u64);
            black_box(q.try_pop());
        })
    });
    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");

    // ---------- bounded: fixed-capacity ring, backpressure on enqueue ----------
    #[cfg(feature = "bounded")]
    {
        use subms_mpsc_queue::BoundedMpscQueue;
        // Half full at every sweep point. Filled to a FIXED element count
        // instead, the big rings would sit 98% empty and the enqueue would be
        // measuring the fill fraction rather than the footprint.
        fn ring(n: usize) -> BoundedMpscQueue<u64> {
            let q = BoundedMpscQueue::new(n);
            for i in 0..n / 2 {
                let _ = q.try_enqueue(i as u64);
            }
            q
        }
        let sw = sweep("bounded/enqueue+dequeue", |n| {
            let mut q = ring(n);
            batched(ITEMS_PER_SAMPLE, |i| {
                let _ = q.try_enqueue(i as u64);
                black_box(q.try_dequeue());
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let mut q = ring(CANON);
        let mut p99 = BTreeMap::new();
        p99.insert(
            "enqueue".to_string(),
            single(|st, i| {
                st.time(|| {
                    let _ = q.try_enqueue(i as u64);
                });
                black_box(q.try_dequeue());
            }),
        );
        p99.insert(
            "dequeue".to_string(),
            single(|st, i| {
                let _ = q.try_enqueue(i as u64);
                st.time(|| black_box(q.try_dequeue()));
            }),
        );
        // The reject path, which is the reason the feature exists. The ring is
        // filled to capacity once, OUTSIDE the timed region; every timed call
        // then takes the full branch and hands the value back to the caller.
        let full: BoundedMpscQueue<u64> = BoundedMpscQueue::new(CANON);
        while full.try_enqueue(0).is_ok() {}
        p99.insert(
            "enqueue_full".to_string(),
            single(|st, i| {
                st.time(|| {
                    let _ = full.try_enqueue(i as u64);
                });
            }),
        );
        manifest.set_feature("bounded", cat, &p99, &reason);
    }

    // ---------- mpmc: bounded ring, sequence CAS on both ends ----------
    #[cfg(feature = "mpmc")]
    {
        use subms_mpsc_queue::MpmcQueue;
        fn ring(n: usize) -> MpmcQueue<u64> {
            let q = MpmcQueue::new(n);
            for i in 0..n / 2 {
                let _ = q.try_enqueue(i as u64);
            }
            q
        }
        // Uncontended, so every CAS succeeds first try. That is the figure the
        // category is about: what the multi-consumer claim costs a queue that is
        // NOT contended, which is the state a well-sized pipeline runs in.
        let sw = sweep("mpmc/enqueue+dequeue", |n| {
            let q = ring(n);
            batched(ITEMS_PER_SAMPLE, |i| {
                let _ = q.try_enqueue(i as u64);
                black_box(q.try_dequeue());
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let q = ring(CANON);
        let mut p99 = BTreeMap::new();
        p99.insert(
            "enqueue".to_string(),
            single(|st, i| {
                st.time(|| {
                    let _ = q.try_enqueue(i as u64);
                });
                black_box(q.try_dequeue());
            }),
        );
        p99.insert(
            "dequeue".to_string(),
            single(|st, i| {
                let _ = q.try_enqueue(i as u64);
                st.time(|| black_box(q.try_dequeue()));
            }),
        );
        manifest.set_feature("mpmc", cat, &p99, &reason);
    }

    // ---------- batch: drain up to BATCH items behind one acquire fence ----------
    #[cfg(feature = "batch")]
    {
        use subms_mpsc_queue::BatchMpscQueue;
        fn filled_batch(n: usize) -> BatchMpscQueue<u64> {
            let q = BatchMpscQueue::new();
            for i in 0..n {
                q.push(i as u64);
            }
            q
        }
        // A sample moves ITEMS_PER_SAMPLE items either way; only the call width
        // differs. That is why the reps count is divided rather than the batch
        // grown - growing it would sweep the batch size, and the number would
        // stop being comparable to the base round trip.
        let sw = sweep("batch/push+dequeue_batch", |n| {
            let mut q = filled_batch(n);
            let mut buf: Vec<Option<u64>> = (0..BATCH).map(|_| None).collect();
            batched(ITEMS_PER_SAMPLE / BATCH, |i| {
                for j in 0..BATCH {
                    q.push((i + j) as u64);
                }
                black_box(q.try_dequeue_batch(&mut buf));
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let mut q = filled_batch(CANON);
        let mut buf: Vec<Option<u64>> = (0..BATCH).map(|_| None).collect();
        let mut p99 = BTreeMap::new();
        // The refill is outside the timed region: timing it would put a BATCH of
        // pushes inside the drain's number and the stage would stop being a
        // drain figure at all.
        p99.insert(
            "dequeue_batch".to_string(),
            single(|st, i| {
                st.time(|| black_box(q.try_dequeue_batch(&mut buf)));
                for j in 0..BATCH {
                    q.push((i + j) as u64);
                }
            }),
        );
        p99.insert(
            "enqueue".to_string(),
            single(|st, i| {
                st.time(|| q.push(i as u64));
                let _ = q.try_dequeue_batch(&mut buf[..1]);
            }),
        );
        // The producer mirror: BATCH items published behind one head swap. The
        // drain that puts the queue back is outside the timed region for the
        // same reason the refill is above.
        p99.insert(
            "enqueue_batch".to_string(),
            single(|st, i| {
                let base = i as u64;
                st.time(|| black_box(q.push_batch(base..base + BATCH as u64)));
                let _ = q.try_dequeue_batch(&mut buf);
            }),
        );
        manifest.set_feature("batch", cat, &p99, &reason);
    }

    // ---------- metrics: relaxed atomic counters around each op ----------
    #[cfg(feature = "metrics")]
    {
        use subms_mpsc_queue::MetricsMpscQueue;
        fn filled_metrics(n: usize) -> MetricsMpscQueue<u64> {
            let q = MetricsMpscQueue::new();
            for i in 0..n {
                q.push(i as u64);
            }
            q
        }
        let sw = sweep("metrics/push+pop", |n| {
            let mut q = filled_metrics(n);
            batched(ITEMS_PER_SAMPLE, |i| {
                q.push(i as u64);
                black_box(q.try_pop());
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let mut q = filled_metrics(CANON);
        let mut p99 = BTreeMap::new();
        p99.insert(
            "enqueue".to_string(),
            single(|st, i| {
                st.time(|| q.push(i as u64));
                black_box(q.try_pop());
            }),
        );
        p99.insert(
            "dequeue".to_string(),
            single(|st, i| {
                q.push(i as u64);
                st.time(|| black_box(q.try_pop()));
            }),
        );
        p99.insert(
            "snapshot".to_string(),
            single(|st, _| {
                st.time(|| black_box(q.snapshot()));
            }),
        );
        manifest.set_feature("metrics", cat, &p99, &reason);
    }

    // ---------- affinity: pin the calling thread, once, at startup ----------
    // Runs LAST because measuring it pins THIS process to core 0, and every
    // number taken afterwards would be a number taken on one core.
    #[cfg(feature = "affinity")]
    {
        use subms_mpsc_queue::set_affinity;
        // Swept over the same axis to show what it is: a call that touches no
        // queue state and cannot move with queue size. PINNED auxiliary rather
        // than left to the base-delta test, which would see a syscall costing
        // more than an enqueue and call it hot-path. It is not on the hot path at
        // any price - `set_affinity` is called once per thread at startup and
        // appears in neither `push` nor `try_pop`. The two ports are not even
        // measuring the same thing: Rust issues a real `SetThreadAffinityMask` /
        // `sched_setaffinity`, while the Java sibling validates its argument and
        // returns UNSUPPORTED because the stock JDK has no pinning API. The
        // per-call figure, not the sample, is the interpretable one and it is in
        // `p99ByStage`.
        let sw = sweep("affinity/set_affinity", |_| {
            batched(ITEMS_PER_SAMPLE, |_| {
                let _ = set_affinity(&[0]);
            })
        });
        let (cat, reason) = classify_feature(
            &sw,
            Some(base_p50),
            Some(subms::SubMsFeatureCategory::Auxiliary),
        );

        let mut p99 = BTreeMap::new();
        p99.insert(
            "set_affinity".to_string(),
            single(|st, _| {
                st.time(|| {
                    let _ = set_affinity(&[0]);
                });
            }),
        );
        manifest.set_feature("affinity", cat, &p99, &reason);

        let cores: Vec<usize> = (0..std::thread::available_parallelism()
            .map_or(1, std::num::NonZeroUsize::get))
            .collect();
        let _ = set_affinity(&cores);
    }

    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/sample_app.rs (line 142)

fn mpmc_sharded_match() {
    use std::sync::atomic::{AtomicUsize, Ordering};

    use subms_mpsc_queue::MpmcQueue;

    println!("\n== mpmc: shard the match loop across several consumers ==");
    let shards = 3usize;
    let ring: Arc<MpmcQueue<u64>> = Arc::new(MpmcQueue::new(1_024));
    let total = GATEWAYS * ORDERS_PER_GATEWAY;

    let gateways: Vec<_> = (0..GATEWAYS)
        .map(|g| {
            let ring = Arc::clone(&ring);
            thread::spawn(move || {
                for seq in 0..ORDERS_PER_GATEWAY {
                    let mut order = order_id(g, seq);
                    while let Err(rejected) = ring.try_enqueue(order) {
                        order = rejected;
                        std::hint::spin_loop();
                    }
                }
            })
        })
        .collect();

    let matched = Arc::new(AtomicUsize::new(0));
    let consumers: Vec<_> = (0..shards)
        .map(|_| {
            let ring = Arc::clone(&ring);
            let matched = Arc::clone(&matched);
            thread::spawn(move || {
                let mut local = 0usize;
                loop {
                    if ring.try_dequeue().is_some() {
                        local += 1;
                        matched.fetch_add(1, Ordering::Relaxed);
                    } else if matched.load(Ordering::Relaxed) >= total {
                        break;
                    } else {
                        std::hint::spin_loop();
                    }
                }
                local
            })
        })
        .collect();

    for h in gateways {
        h.join().unwrap();
    }
    let drained: usize = consumers.into_iter().map(|c| c.join().unwrap()).sum();
    // cas_retries() is the contention read-out, deliberately not printed: it is
    // a property of how the OS scheduled these threads on this run.
    println!(
        "  {shards} shards drained {drained} orders, ring empty: {}",
        ring.is_empty()
    );
    assert_eq!(
        drained, total,
        "shards together drain every order exactly once"
    );
    assert_eq!(
        ring.producer_index(),
        ring.consumer_index(),
        "every claimed slot was consumed"
    );
}

pub fn try_dequeue(&self) -> Option<T>

Multi-consumer dequeue. Returns None if the ring is empty.

Examples found in repository

examples/sample_app.rs (line 159)

fn mpmc_sharded_match() {
    use std::sync::atomic::{AtomicUsize, Ordering};

    use subms_mpsc_queue::MpmcQueue;

    println!("\n== mpmc: shard the match loop across several consumers ==");
    let shards = 3usize;
    let ring: Arc<MpmcQueue<u64>> = Arc::new(MpmcQueue::new(1_024));
    let total = GATEWAYS * ORDERS_PER_GATEWAY;

    let gateways: Vec<_> = (0..GATEWAYS)
        .map(|g| {
            let ring = Arc::clone(&ring);
            thread::spawn(move || {
                for seq in 0..ORDERS_PER_GATEWAY {
                    let mut order = order_id(g, seq);
                    while let Err(rejected) = ring.try_enqueue(order) {
                        order = rejected;
                        std::hint::spin_loop();
                    }
                }
            })
        })
        .collect();

    let matched = Arc::new(AtomicUsize::new(0));
    let consumers: Vec<_> = (0..shards)
        .map(|_| {
            let ring = Arc::clone(&ring);
            let matched = Arc::clone(&matched);
            thread::spawn(move || {
                let mut local = 0usize;
                loop {
                    if ring.try_dequeue().is_some() {
                        local += 1;
                        matched.fetch_add(1, Ordering::Relaxed);
                    } else if matched.load(Ordering::Relaxed) >= total {
                        break;
                    } else {
                        std::hint::spin_loop();
                    }
                }
                local
            })
        })
        .collect();

    for h in gateways {
        h.join().unwrap();
    }
    let drained: usize = consumers.into_iter().map(|c| c.join().unwrap()).sum();
    // cas_retries() is the contention read-out, deliberately not printed: it is
    // a property of how the OS scheduled these threads on this run.
    println!(
        "  {shards} shards drained {drained} orders, ring empty: {}",
        ring.is_empty()
    );
    assert_eq!(
        drained, total,
        "shards together drain every order exactly once"
    );
    assert_eq!(
        ring.producer_index(),
        ring.consumer_index(),
        "every claimed slot was consumed"
    );
}

examples/perf_features.rs (line 330)

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

    // Optional, off by default: pin this thread to one core for the whole run.
    // On a heterogeneous laptop the scheduler moves the bench between core
    // clusters and every measurement lands in one of two clock states 1.31x
    // apart - a spread three times wider than the deltas being classified, and
    // large enough on its own to flip a feature between auxiliary and hot-path.
    // Pinned, the same sweep repeats to within 1%. Left OFF by default because a
    // fleet box isolates cores outside the process, and pinning from in here
    // would override that placement with a core the orchestrator did not choose.
    #[cfg(feature = "affinity")]
    if let Some(core) = std::env::var("SUBMS_PIN").ok().and_then(|v| v.parse().ok()) {
        let _ = subms_mpsc_queue::set_affinity(&[core]);
    }

    // Burn before the first measurement, not just before each one. Every
    // `batched` call warms itself, but the FIRST measurement in the process pays
    // a ramp the per-measurement warm sits inside rather than absorbs, and the
    // sweep runs smallest-first: without this the base curve read 71800 / 45300 /
    // 46500 ns, a 1.6x fall with size that is the process settling, not the
    // queue.
    {
        let mut q = filled(CANON);
        let start = std::time::Instant::now();
        while (start.elapsed().as_nanos() as u64) < BURN_NANOS {
            for i in 0..ITEMS_PER_SAMPLE {
                q.push(i as u64);
                black_box(q.try_pop());
            }
        }
    }

    // The baseline: the base queue's push + try_pop round trip. Swept as well as
    // sampled, because whether queue depth moves the BASE op is the context
    // every feature curve is read against.
    let base_sweep = sweep("base/push+pop", |n| {
        let mut q = filled(n);
        batched(ITEMS_PER_SAMPLE, |i| {
            q.push(i as u64);
            black_box(q.try_pop());
        })
    });
    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");

    // ---------- bounded: fixed-capacity ring, backpressure on enqueue ----------
    #[cfg(feature = "bounded")]
    {
        use subms_mpsc_queue::BoundedMpscQueue;
        // Half full at every sweep point. Filled to a FIXED element count
        // instead, the big rings would sit 98% empty and the enqueue would be
        // measuring the fill fraction rather than the footprint.
        fn ring(n: usize) -> BoundedMpscQueue<u64> {
            let q = BoundedMpscQueue::new(n);
            for i in 0..n / 2 {
                let _ = q.try_enqueue(i as u64);
            }
            q
        }
        let sw = sweep("bounded/enqueue+dequeue", |n| {
            let mut q = ring(n);
            batched(ITEMS_PER_SAMPLE, |i| {
                let _ = q.try_enqueue(i as u64);
                black_box(q.try_dequeue());
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let mut q = ring(CANON);
        let mut p99 = BTreeMap::new();
        p99.insert(
            "enqueue".to_string(),
            single(|st, i| {
                st.time(|| {
                    let _ = q.try_enqueue(i as u64);
                });
                black_box(q.try_dequeue());
            }),
        );
        p99.insert(
            "dequeue".to_string(),
            single(|st, i| {
                let _ = q.try_enqueue(i as u64);
                st.time(|| black_box(q.try_dequeue()));
            }),
        );
        // The reject path, which is the reason the feature exists. The ring is
        // filled to capacity once, OUTSIDE the timed region; every timed call
        // then takes the full branch and hands the value back to the caller.
        let full: BoundedMpscQueue<u64> = BoundedMpscQueue::new(CANON);
        while full.try_enqueue(0).is_ok() {}
        p99.insert(
            "enqueue_full".to_string(),
            single(|st, i| {
                st.time(|| {
                    let _ = full.try_enqueue(i as u64);
                });
            }),
        );
        manifest.set_feature("bounded", cat, &p99, &reason);
    }

    // ---------- mpmc: bounded ring, sequence CAS on both ends ----------
    #[cfg(feature = "mpmc")]
    {
        use subms_mpsc_queue::MpmcQueue;
        fn ring(n: usize) -> MpmcQueue<u64> {
            let q = MpmcQueue::new(n);
            for i in 0..n / 2 {
                let _ = q.try_enqueue(i as u64);
            }
            q
        }
        // Uncontended, so every CAS succeeds first try. That is the figure the
        // category is about: what the multi-consumer claim costs a queue that is
        // NOT contended, which is the state a well-sized pipeline runs in.
        let sw = sweep("mpmc/enqueue+dequeue", |n| {
            let q = ring(n);
            batched(ITEMS_PER_SAMPLE, |i| {
                let _ = q.try_enqueue(i as u64);
                black_box(q.try_dequeue());
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let q = ring(CANON);
        let mut p99 = BTreeMap::new();
        p99.insert(
            "enqueue".to_string(),
            single(|st, i| {
                st.time(|| {
                    let _ = q.try_enqueue(i as u64);
                });
                black_box(q.try_dequeue());
            }),
        );
        p99.insert(
            "dequeue".to_string(),
            single(|st, i| {
                let _ = q.try_enqueue(i as u64);
                st.time(|| black_box(q.try_dequeue()));
            }),
        );
        manifest.set_feature("mpmc", cat, &p99, &reason);
    }

    // ---------- batch: drain up to BATCH items behind one acquire fence ----------
    #[cfg(feature = "batch")]
    {
        use subms_mpsc_queue::BatchMpscQueue;
        fn filled_batch(n: usize) -> BatchMpscQueue<u64> {
            let q = BatchMpscQueue::new();
            for i in 0..n {
                q.push(i as u64);
            }
            q
        }
        // A sample moves ITEMS_PER_SAMPLE items either way; only the call width
        // differs. That is why the reps count is divided rather than the batch
        // grown - growing it would sweep the batch size, and the number would
        // stop being comparable to the base round trip.
        let sw = sweep("batch/push+dequeue_batch", |n| {
            let mut q = filled_batch(n);
            let mut buf: Vec<Option<u64>> = (0..BATCH).map(|_| None).collect();
            batched(ITEMS_PER_SAMPLE / BATCH, |i| {
                for j in 0..BATCH {
                    q.push((i + j) as u64);
                }
                black_box(q.try_dequeue_batch(&mut buf));
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let mut q = filled_batch(CANON);
        let mut buf: Vec<Option<u64>> = (0..BATCH).map(|_| None).collect();
        let mut p99 = BTreeMap::new();
        // The refill is outside the timed region: timing it would put a BATCH of
        // pushes inside the drain's number and the stage would stop being a
        // drain figure at all.
        p99.insert(
            "dequeue_batch".to_string(),
            single(|st, i| {
                st.time(|| black_box(q.try_dequeue_batch(&mut buf)));
                for j in 0..BATCH {
                    q.push((i + j) as u64);
                }
            }),
        );
        p99.insert(
            "enqueue".to_string(),
            single(|st, i| {
                st.time(|| q.push(i as u64));
                let _ = q.try_dequeue_batch(&mut buf[..1]);
            }),
        );
        // The producer mirror: BATCH items published behind one head swap. The
        // drain that puts the queue back is outside the timed region for the
        // same reason the refill is above.
        p99.insert(
            "enqueue_batch".to_string(),
            single(|st, i| {
                let base = i as u64;
                st.time(|| black_box(q.push_batch(base..base + BATCH as u64)));
                let _ = q.try_dequeue_batch(&mut buf);
            }),
        );
        manifest.set_feature("batch", cat, &p99, &reason);
    }

    // ---------- metrics: relaxed atomic counters around each op ----------
    #[cfg(feature = "metrics")]
    {
        use subms_mpsc_queue::MetricsMpscQueue;
        fn filled_metrics(n: usize) -> MetricsMpscQueue<u64> {
            let q = MetricsMpscQueue::new();
            for i in 0..n {
                q.push(i as u64);
            }
            q
        }
        let sw = sweep("metrics/push+pop", |n| {
            let mut q = filled_metrics(n);
            batched(ITEMS_PER_SAMPLE, |i| {
                q.push(i as u64);
                black_box(q.try_pop());
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let mut q = filled_metrics(CANON);
        let mut p99 = BTreeMap::new();
        p99.insert(
            "enqueue".to_string(),
            single(|st, i| {
                st.time(|| q.push(i as u64));
                black_box(q.try_pop());
            }),
        );
        p99.insert(
            "dequeue".to_string(),
            single(|st, i| {
                q.push(i as u64);
                st.time(|| black_box(q.try_pop()));
            }),
        );
        p99.insert(
            "snapshot".to_string(),
            single(|st, _| {
                st.time(|| black_box(q.snapshot()));
            }),
        );
        manifest.set_feature("metrics", cat, &p99, &reason);
    }

    // ---------- affinity: pin the calling thread, once, at startup ----------
    // Runs LAST because measuring it pins THIS process to core 0, and every
    // number taken afterwards would be a number taken on one core.
    #[cfg(feature = "affinity")]
    {
        use subms_mpsc_queue::set_affinity;
        // Swept over the same axis to show what it is: a call that touches no
        // queue state and cannot move with queue size. PINNED auxiliary rather
        // than left to the base-delta test, which would see a syscall costing
        // more than an enqueue and call it hot-path. It is not on the hot path at
        // any price - `set_affinity` is called once per thread at startup and
        // appears in neither `push` nor `try_pop`. The two ports are not even
        // measuring the same thing: Rust issues a real `SetThreadAffinityMask` /
        // `sched_setaffinity`, while the Java sibling validates its argument and
        // returns UNSUPPORTED because the stock JDK has no pinning API. The
        // per-call figure, not the sample, is the interpretable one and it is in
        // `p99ByStage`.
        let sw = sweep("affinity/set_affinity", |_| {
            batched(ITEMS_PER_SAMPLE, |_| {
                let _ = set_affinity(&[0]);
            })
        });
        let (cat, reason) = classify_feature(
            &sw,
            Some(base_p50),
            Some(subms::SubMsFeatureCategory::Auxiliary),
        );

        let mut p99 = BTreeMap::new();
        p99.insert(
            "set_affinity".to_string(),
            single(|st, _| {
                st.time(|| {
                    let _ = set_affinity(&[0]);
                });
            }),
        );
        manifest.set_feature("affinity", cat, &p99, &reason);

        let cores: Vec<usize> = (0..std::thread::available_parallelism()
            .map_or(1, std::num::NonZeroUsize::get))
            .collect();
        let _ = set_affinity(&cores);
    }

    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 clear(&self) -> usize

Drop everything currently readable and return the count. Any consumer may call it, and other consumers keep draining alongside, so the count is this caller’s share rather than the queue’s total.

pub fn len(&self) -> usize

Approximate length.

pub fn is_empty(&self) -> bool

Examples found in repository

examples/sample_app.rs (line 181)

fn mpmc_sharded_match() {
    use std::sync::atomic::{AtomicUsize, Ordering};

    use subms_mpsc_queue::MpmcQueue;

    println!("\n== mpmc: shard the match loop across several consumers ==");
    let shards = 3usize;
    let ring: Arc<MpmcQueue<u64>> = Arc::new(MpmcQueue::new(1_024));
    let total = GATEWAYS * ORDERS_PER_GATEWAY;

    let gateways: Vec<_> = (0..GATEWAYS)
        .map(|g| {
            let ring = Arc::clone(&ring);
            thread::spawn(move || {
                for seq in 0..ORDERS_PER_GATEWAY {
                    let mut order = order_id(g, seq);
                    while let Err(rejected) = ring.try_enqueue(order) {
                        order = rejected;
                        std::hint::spin_loop();
                    }
                }
            })
        })
        .collect();

    let matched = Arc::new(AtomicUsize::new(0));
    let consumers: Vec<_> = (0..shards)
        .map(|_| {
            let ring = Arc::clone(&ring);
            let matched = Arc::clone(&matched);
            thread::spawn(move || {
                let mut local = 0usize;
                loop {
                    if ring.try_dequeue().is_some() {
                        local += 1;
                        matched.fetch_add(1, Ordering::Relaxed);
                    } else if matched.load(Ordering::Relaxed) >= total {
                        break;
                    } else {
                        std::hint::spin_loop();
                    }
                }
                local
            })
        })
        .collect();

    for h in gateways {
        h.join().unwrap();
    }
    let drained: usize = consumers.into_iter().map(|c| c.join().unwrap()).sum();
    // cas_retries() is the contention read-out, deliberately not printed: it is
    // a property of how the OS scheduled these threads on this run.
    println!(
        "  {shards} shards drained {drained} orders, ring empty: {}",
        ring.is_empty()
    );
    assert_eq!(
        drained, total,
        "shards together drain every order exactly once"
    );
    assert_eq!(
        ring.producer_index(),
        ring.consumer_index(),
        "every claimed slot was consumed"
    );
}

pub fn is_full(&self) -> bool

Best-effort fullness. Stale the instant any consumer drains a slot.

Trait Implementations

impl<T> Drop for MpmcQueue<T>

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.

impl<T: Send> Send for MpmcQueue<T>

impl<T: Send> Sync for MpmcQueue<T>