Struct BatchMpscQueue 

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

Batch-draining wrapper around the base MpscQueue.

Implementations

impl<T> BatchMpscQueue<T>

pub fn new() -> Self

Examples found in repository

examples/perf_features.rs (line 361)

        fn filled_batch(n: usize) -> BatchMpscQueue<u64> {
            let q = BatchMpscQueue::new();
            for i in 0..n {
                q.push(i as u64);
            }
            q
        }

examples/sample_app.rs (line 251)

fn batch_drain_per_tick() {
    use subms_mpsc_queue::BatchMpscQueue;

    println!("\n== batch: publish and drain a whole tick in one pass ==");
    const TICK: usize = 256;
    const BURST: usize = 50;
    let mut q: BatchMpscQueue<u64> = BatchMpscQueue::new();
    let total = 1_000usize;

    // A gateway that decodes a wire frame already holds a run of orders. One
    // head swap publishes the whole run instead of BURST of them.
    let mut published = 0usize;
    while published < total {
        let base = published;
        published += q.push_batch((base..base + BURST).map(|seq| order_id(0, seq)));
    }
    println!("  {published} orders published in {} swaps", total / BURST);
    assert_eq!(published, total);

    let mut buf: Vec<Option<u64>> = (0..TICK).map(|_| None).collect();
    let mut ticks = 0usize;
    let mut matched = 0usize;
    loop {
        let n = q.try_dequeue_batch(&mut buf);
        if n == 0 {
            break;
        }
        ticks += 1;
        for slot in buf.iter_mut().take(n) {
            let _ = slot.take();
            matched += 1;
        }
    }
    println!("  drained {matched} orders across {ticks} ticks of up to {TICK}");
    assert_eq!(matched, total, "every queued order is drained");
    assert_eq!(
        ticks,
        total.div_ceil(TICK),
        "each tick drains a full buffer until the tail"
    );

    // The callback form skips the buffer entirely when the match loop's work
    // is per-order anyway. Here it accumulates notional.
    q.push_batch((0..64).map(|seq| order_id(1, seq)));
    let mut notional = 0u64;
    let handled = q.drain(TICK, |order| notional += order & 0xffff_ffff);
    println!("  drain callback handled {handled} orders, notional {notional}");
    assert_eq!(handled, 64);
    assert_eq!(notional, (0..64u64).sum::<u64>());
    assert!(q.is_empty());
}

pub fn push(&self, value: T)

Same as the base MpscQueue::push.

Examples found in repository

examples/perf_features.rs (line 363)

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 push_batch<I: IntoIterator<Item = T>>(&self, values: I) -> usize

Publish a whole run with one head swap. The producer-side mirror of Self::try_dequeue_batch: N items cost one atomic exchange rather than N. Returns the number published.

Examples found in repository

examples/sample_app.rs (line 259)

fn batch_drain_per_tick() {
    use subms_mpsc_queue::BatchMpscQueue;

    println!("\n== batch: publish and drain a whole tick in one pass ==");
    const TICK: usize = 256;
    const BURST: usize = 50;
    let mut q: BatchMpscQueue<u64> = BatchMpscQueue::new();
    let total = 1_000usize;

    // A gateway that decodes a wire frame already holds a run of orders. One
    // head swap publishes the whole run instead of BURST of them.
    let mut published = 0usize;
    while published < total {
        let base = published;
        published += q.push_batch((base..base + BURST).map(|seq| order_id(0, seq)));
    }
    println!("  {published} orders published in {} swaps", total / BURST);
    assert_eq!(published, total);

    let mut buf: Vec<Option<u64>> = (0..TICK).map(|_| None).collect();
    let mut ticks = 0usize;
    let mut matched = 0usize;
    loop {
        let n = q.try_dequeue_batch(&mut buf);
        if n == 0 {
            break;
        }
        ticks += 1;
        for slot in buf.iter_mut().take(n) {
            let _ = slot.take();
            matched += 1;
        }
    }
    println!("  drained {matched} orders across {ticks} ticks of up to {TICK}");
    assert_eq!(matched, total, "every queued order is drained");
    assert_eq!(
        ticks,
        total.div_ceil(TICK),
        "each tick drains a full buffer until the tail"
    );

    // The callback form skips the buffer entirely when the match loop's work
    // is per-order anyway. Here it accumulates notional.
    q.push_batch((0..64).map(|seq| order_id(1, seq)));
    let mut notional = 0u64;
    let handled = q.drain(TICK, |order| notional += order & 0xffff_ffff);
    println!("  drain callback handled {handled} orders, notional {notional}");
    assert_eq!(handled, 64);
    assert_eq!(notional, (0..64u64).sum::<u64>());
    assert!(q.is_empty());
}

examples/perf_features.rs (line 412)

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 try_dequeue_batch(&mut self, out: &mut [Option<T>]) -> usize

Drain up to out.len() items into out. Returns the count.

Stops early on dangling-tail or empty. Caller can spin / back off and re-call.

Examples found in repository

examples/sample_app.rs (line 268)

fn batch_drain_per_tick() {
    use subms_mpsc_queue::BatchMpscQueue;

    println!("\n== batch: publish and drain a whole tick in one pass ==");
    const TICK: usize = 256;
    const BURST: usize = 50;
    let mut q: BatchMpscQueue<u64> = BatchMpscQueue::new();
    let total = 1_000usize;

    // A gateway that decodes a wire frame already holds a run of orders. One
    // head swap publishes the whole run instead of BURST of them.
    let mut published = 0usize;
    while published < total {
        let base = published;
        published += q.push_batch((base..base + BURST).map(|seq| order_id(0, seq)));
    }
    println!("  {published} orders published in {} swaps", total / BURST);
    assert_eq!(published, total);

    let mut buf: Vec<Option<u64>> = (0..TICK).map(|_| None).collect();
    let mut ticks = 0usize;
    let mut matched = 0usize;
    loop {
        let n = q.try_dequeue_batch(&mut buf);
        if n == 0 {
            break;
        }
        ticks += 1;
        for slot in buf.iter_mut().take(n) {
            let _ = slot.take();
            matched += 1;
        }
    }
    println!("  drained {matched} orders across {ticks} ticks of up to {TICK}");
    assert_eq!(matched, total, "every queued order is drained");
    assert_eq!(
        ticks,
        total.div_ceil(TICK),
        "each tick drains a full buffer until the tail"
    );

    // The callback form skips the buffer entirely when the match loop's work
    // is per-order anyway. Here it accumulates notional.
    q.push_batch((0..64).map(|seq| order_id(1, seq)));
    let mut notional = 0u64;
    let handled = q.drain(TICK, |order| notional += order & 0xffff_ffff);
    println!("  drain callback handled {handled} orders, notional {notional}");
    assert_eq!(handled, 64);
    assert_eq!(notional, (0..64u64).sum::<u64>());
    assert!(q.is_empty());
}

examples/perf_features.rs (line 378)

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 drain<F: FnMut(T)>(&mut self, limit: usize, f: F) -> usize

Drain up to limit items straight into f, with no intermediate buffer. The callback form of JCTools’ drain(Consumer, limit), and the one to reach for when the consumer’s work is per-item anyway.

Stops early on empty or dangling-tail, exactly as Self::try_dequeue_batch does. Returns the count handed to f.

Examples found in repository

examples/sample_app.rs (line 290)

fn batch_drain_per_tick() {
    use subms_mpsc_queue::BatchMpscQueue;

    println!("\n== batch: publish and drain a whole tick in one pass ==");
    const TICK: usize = 256;
    const BURST: usize = 50;
    let mut q: BatchMpscQueue<u64> = BatchMpscQueue::new();
    let total = 1_000usize;

    // A gateway that decodes a wire frame already holds a run of orders. One
    // head swap publishes the whole run instead of BURST of them.
    let mut published = 0usize;
    while published < total {
        let base = published;
        published += q.push_batch((base..base + BURST).map(|seq| order_id(0, seq)));
    }
    println!("  {published} orders published in {} swaps", total / BURST);
    assert_eq!(published, total);

    let mut buf: Vec<Option<u64>> = (0..TICK).map(|_| None).collect();
    let mut ticks = 0usize;
    let mut matched = 0usize;
    loop {
        let n = q.try_dequeue_batch(&mut buf);
        if n == 0 {
            break;
        }
        ticks += 1;
        for slot in buf.iter_mut().take(n) {
            let _ = slot.take();
            matched += 1;
        }
    }
    println!("  drained {matched} orders across {ticks} ticks of up to {TICK}");
    assert_eq!(matched, total, "every queued order is drained");
    assert_eq!(
        ticks,
        total.div_ceil(TICK),
        "each tick drains a full buffer until the tail"
    );

    // The callback form skips the buffer entirely when the match loop's work
    // is per-order anyway. Here it accumulates notional.
    q.push_batch((0..64).map(|seq| order_id(1, seq)));
    let mut notional = 0u64;
    let handled = q.drain(TICK, |order| notional += order & 0xffff_ffff);
    println!("  drain callback handled {handled} orders, notional {notional}");
    assert_eq!(handled, 64);
    assert_eq!(notional, (0..64u64).sum::<u64>());
    assert!(q.is_empty());
}

pub fn drain_into_vec(&mut self, out: &mut Vec<T>, cap: usize) -> usize

Convenience: drain into a Vec, returning the count drained. Pre-sizes the vec to cap before draining.

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

Borrow the next value without consuming it. See MpscQueue::peek.

pub fn is_empty(&mut self) -> bool

See MpscQueue::is_empty.

Examples found in repository

examples/sample_app.rs (line 294)

fn batch_drain_per_tick() {
    use subms_mpsc_queue::BatchMpscQueue;

    println!("\n== batch: publish and drain a whole tick in one pass ==");
    const TICK: usize = 256;
    const BURST: usize = 50;
    let mut q: BatchMpscQueue<u64> = BatchMpscQueue::new();
    let total = 1_000usize;

    // A gateway that decodes a wire frame already holds a run of orders. One
    // head swap publishes the whole run instead of BURST of them.
    let mut published = 0usize;
    while published < total {
        let base = published;
        published += q.push_batch((base..base + BURST).map(|seq| order_id(0, seq)));
    }
    println!("  {published} orders published in {} swaps", total / BURST);
    assert_eq!(published, total);

    let mut buf: Vec<Option<u64>> = (0..TICK).map(|_| None).collect();
    let mut ticks = 0usize;
    let mut matched = 0usize;
    loop {
        let n = q.try_dequeue_batch(&mut buf);
        if n == 0 {
            break;
        }
        ticks += 1;
        for slot in buf.iter_mut().take(n) {
            let _ = slot.take();
            matched += 1;
        }
    }
    println!("  drained {matched} orders across {ticks} ticks of up to {TICK}");
    assert_eq!(matched, total, "every queued order is drained");
    assert_eq!(
        ticks,
        total.div_ceil(TICK),
        "each tick drains a full buffer until the tail"
    );

    // The callback form skips the buffer entirely when the match loop's work
    // is per-order anyway. Here it accumulates notional.
    q.push_batch((0..64).map(|seq| order_id(1, seq)));
    let mut notional = 0u64;
    let handled = q.drain(TICK, |order| notional += order & 0xffff_ffff);
    println!("  drain callback handled {handled} orders, notional {notional}");
    assert_eq!(handled, 64);
    assert_eq!(notional, (0..64u64).sum::<u64>());
    assert!(q.is_empty());
}

pub fn len(&mut self) -> usize

See MpscQueue::len. O(n) in the backlog.

pub fn clear(&mut self) -> usize

See MpscQueue::clear.

Trait Implementations

impl<T> Default for BatchMpscQueue<T>

fn default() -> Self

Returns the “default value” for a type.