Struct MpscQueue 

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

Multi-producer single-consumer linked queue.

Cloneable handles are not provided; share via Arc<MpscQueue<T>>. Consumer methods (try_pop) require &mut self to encode single-consumer invariant at the type level.

Implementations

impl<T> MpscQueue<T>

pub fn new() -> Self

Examples found in repository

examples/perf_features.rs (line 188)

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

examples/sample_app.rs (line 51)

fn base_order_fan_in() {
    println!("== base: order-entry gateways fan in to one matching engine ==");
    let q: Arc<MpscQueue<u64>> = Arc::new(MpscQueue::new());

    let gateways: Vec<_> = (0..GATEWAYS)
        .map(|g| {
            let q = Arc::clone(&q);
            thread::spawn(move || {
                for seq in 0..ORDERS_PER_GATEWAY {
                    q.push(order_id(g, seq));
                }
            })
        })
        .collect();
    for h in gateways {
        h.join().unwrap();
    }

    // Every gateway handle is joined, so the queue is uniquely owned again and
    // the single consumer can take it back out (try_pop needs &mut self).
    let mut q = Arc::into_inner(q).expect("all gateway handles dropped");

    let total = GATEWAYS * ORDERS_PER_GATEWAY;
    println!("  inbox depth before the match loop starts: {}", q.len());
    let first = *q.peek().expect("the inbox is not empty");
    println!(
        "  head of book: gateway {} seq {}",
        first >> 32,
        first & 0xffff_ffff
    );

    let mut per_gateway = [0usize; GATEWAYS];
    let mut last_seq = [None::<u64>; GATEWAYS];
    let mut matched = 0usize;
    loop {
        match q.try_pop() {
            PopResult::Some(order) => {
                let g = (order >> 32) as usize;
                let seq = order & 0xffff_ffff;
                if let Some(prev) = last_seq[g] {
                    assert!(seq > prev, "orders from one gateway stay in FIFO order");
                }
                last_seq[g] = Some(seq);
                per_gateway[g] += 1;
                matched += 1;
            }
            PopResult::Inconsistent => continue,
            PopResult::Empty => break,
        }
    }

    println!("  {GATEWAYS} gateways x {ORDERS_PER_GATEWAY} orders -> matched {matched}");
    println!("  per-gateway tally: {per_gateway:?}");
    assert_eq!(matched, total, "no order dropped, none duplicated");
    for count in per_gateway {
        assert_eq!(count, ORDERS_PER_GATEWAY, "every gateway fully drained");
    }

    // Kill-switch: a venue disconnect voids everything still queued rather than
    // matching it against a book that has moved on.
    for seq in 0..32 {
        q.push(order_id(0, seq));
    }
    let voided = q.clear();
    println!(
        "  kill switch voided {voided} queued orders, inbox empty: {}",
        q.is_empty()
    );
    assert_eq!(voided, 32);
    assert!(q.is_empty());
}

pub fn push(&self, value: T)

Multi-producer push. Wait-free for the producer once the node is allocated.

Examples found in repository

examples/perf_features.rs (line 190)

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

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

fn base_order_fan_in() {
    println!("== base: order-entry gateways fan in to one matching engine ==");
    let q: Arc<MpscQueue<u64>> = Arc::new(MpscQueue::new());

    let gateways: Vec<_> = (0..GATEWAYS)
        .map(|g| {
            let q = Arc::clone(&q);
            thread::spawn(move || {
                for seq in 0..ORDERS_PER_GATEWAY {
                    q.push(order_id(g, seq));
                }
            })
        })
        .collect();
    for h in gateways {
        h.join().unwrap();
    }

    // Every gateway handle is joined, so the queue is uniquely owned again and
    // the single consumer can take it back out (try_pop needs &mut self).
    let mut q = Arc::into_inner(q).expect("all gateway handles dropped");

    let total = GATEWAYS * ORDERS_PER_GATEWAY;
    println!("  inbox depth before the match loop starts: {}", q.len());
    let first = *q.peek().expect("the inbox is not empty");
    println!(
        "  head of book: gateway {} seq {}",
        first >> 32,
        first & 0xffff_ffff
    );

    let mut per_gateway = [0usize; GATEWAYS];
    let mut last_seq = [None::<u64>; GATEWAYS];
    let mut matched = 0usize;
    loop {
        match q.try_pop() {
            PopResult::Some(order) => {
                let g = (order >> 32) as usize;
                let seq = order & 0xffff_ffff;
                if let Some(prev) = last_seq[g] {
                    assert!(seq > prev, "orders from one gateway stay in FIFO order");
                }
                last_seq[g] = Some(seq);
                per_gateway[g] += 1;
                matched += 1;
            }
            PopResult::Inconsistent => continue,
            PopResult::Empty => break,
        }
    }

    println!("  {GATEWAYS} gateways x {ORDERS_PER_GATEWAY} orders -> matched {matched}");
    println!("  per-gateway tally: {per_gateway:?}");
    assert_eq!(matched, total, "no order dropped, none duplicated");
    for count in per_gateway {
        assert_eq!(count, ORDERS_PER_GATEWAY, "every gateway fully drained");
    }

    // Kill-switch: a venue disconnect voids everything still queued rather than
    // matching it against a book that has moved on.
    for seq in 0..32 {
        q.push(order_id(0, seq));
    }
    let voided = q.clear();
    println!(
        "  kill switch voided {voided} queued orders, inbox empty: {}",
        q.is_empty()
    );
    assert_eq!(voided, 32);
    assert!(q.is_empty());
}

pub fn push_batch<I: IntoIterator<Item = T>>(&self, values: I) -> usize

Publish a whole run of values with a single head swap.

The producer links the nodes together privately before publication, so the chain costs one swap and one release-store no matter how many items it carries. Returns the number published; an empty iterator touches no atomics at all.

Items keep their iteration order relative to each other, and the run is published atomically: a consumer either sees none of it or sees the whole chain reachable from the node it links onto.

pub fn try_pop(&mut self) -> PopResult<T>

Consume one entry. Returns PopResult::Inconsistent if a producer is mid-publish; callers should retry.

Single consumer only: requires &mut self.

Examples found in repository

examples/sample_app.rs (line 84)

fn base_order_fan_in() {
    println!("== base: order-entry gateways fan in to one matching engine ==");
    let q: Arc<MpscQueue<u64>> = Arc::new(MpscQueue::new());

    let gateways: Vec<_> = (0..GATEWAYS)
        .map(|g| {
            let q = Arc::clone(&q);
            thread::spawn(move || {
                for seq in 0..ORDERS_PER_GATEWAY {
                    q.push(order_id(g, seq));
                }
            })
        })
        .collect();
    for h in gateways {
        h.join().unwrap();
    }

    // Every gateway handle is joined, so the queue is uniquely owned again and
    // the single consumer can take it back out (try_pop needs &mut self).
    let mut q = Arc::into_inner(q).expect("all gateway handles dropped");

    let total = GATEWAYS * ORDERS_PER_GATEWAY;
    println!("  inbox depth before the match loop starts: {}", q.len());
    let first = *q.peek().expect("the inbox is not empty");
    println!(
        "  head of book: gateway {} seq {}",
        first >> 32,
        first & 0xffff_ffff
    );

    let mut per_gateway = [0usize; GATEWAYS];
    let mut last_seq = [None::<u64>; GATEWAYS];
    let mut matched = 0usize;
    loop {
        match q.try_pop() {
            PopResult::Some(order) => {
                let g = (order >> 32) as usize;
                let seq = order & 0xffff_ffff;
                if let Some(prev) = last_seq[g] {
                    assert!(seq > prev, "orders from one gateway stay in FIFO order");
                }
                last_seq[g] = Some(seq);
                per_gateway[g] += 1;
                matched += 1;
            }
            PopResult::Inconsistent => continue,
            PopResult::Empty => break,
        }
    }

    println!("  {GATEWAYS} gateways x {ORDERS_PER_GATEWAY} orders -> matched {matched}");
    println!("  per-gateway tally: {per_gateway:?}");
    assert_eq!(matched, total, "no order dropped, none duplicated");
    for count in per_gateway {
        assert_eq!(count, ORDERS_PER_GATEWAY, "every gateway fully drained");
    }

    // Kill-switch: a venue disconnect voids everything still queued rather than
    // matching it against a book that has moved on.
    for seq in 0..32 {
        q.push(order_id(0, seq));
    }
    let voided = q.clear();
    println!(
        "  kill switch voided {voided} queued orders, inbox empty: {}",
        q.is_empty()
    );
    assert_eq!(voided, 32);
    assert!(q.is_empty());
}

examples/perf_features.rs (line 234)

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 peek(&mut self) -> Option<&T>

Borrow the next value without consuming it.

Returns None both when the queue is drained and when a producer is mid-publish, matching JCTools’ relaxedPeek: the caller that needs to tell the two apart calls Self::is_empty.

Consumer-side only.

Examples found in repository

examples/sample_app.rs (line 73)

fn base_order_fan_in() {
    println!("== base: order-entry gateways fan in to one matching engine ==");
    let q: Arc<MpscQueue<u64>> = Arc::new(MpscQueue::new());

    let gateways: Vec<_> = (0..GATEWAYS)
        .map(|g| {
            let q = Arc::clone(&q);
            thread::spawn(move || {
                for seq in 0..ORDERS_PER_GATEWAY {
                    q.push(order_id(g, seq));
                }
            })
        })
        .collect();
    for h in gateways {
        h.join().unwrap();
    }

    // Every gateway handle is joined, so the queue is uniquely owned again and
    // the single consumer can take it back out (try_pop needs &mut self).
    let mut q = Arc::into_inner(q).expect("all gateway handles dropped");

    let total = GATEWAYS * ORDERS_PER_GATEWAY;
    println!("  inbox depth before the match loop starts: {}", q.len());
    let first = *q.peek().expect("the inbox is not empty");
    println!(
        "  head of book: gateway {} seq {}",
        first >> 32,
        first & 0xffff_ffff
    );

    let mut per_gateway = [0usize; GATEWAYS];
    let mut last_seq = [None::<u64>; GATEWAYS];
    let mut matched = 0usize;
    loop {
        match q.try_pop() {
            PopResult::Some(order) => {
                let g = (order >> 32) as usize;
                let seq = order & 0xffff_ffff;
                if let Some(prev) = last_seq[g] {
                    assert!(seq > prev, "orders from one gateway stay in FIFO order");
                }
                last_seq[g] = Some(seq);
                per_gateway[g] += 1;
                matched += 1;
            }
            PopResult::Inconsistent => continue,
            PopResult::Empty => break,
        }
    }

    println!("  {GATEWAYS} gateways x {ORDERS_PER_GATEWAY} orders -> matched {matched}");
    println!("  per-gateway tally: {per_gateway:?}");
    assert_eq!(matched, total, "no order dropped, none duplicated");
    for count in per_gateway {
        assert_eq!(count, ORDERS_PER_GATEWAY, "every gateway fully drained");
    }

    // Kill-switch: a venue disconnect voids everything still queued rather than
    // matching it against a book that has moved on.
    for seq in 0..32 {
        q.push(order_id(0, seq));
    }
    let voided = q.clear();
    println!(
        "  kill switch voided {voided} queued orders, inbox empty: {}",
        q.is_empty()
    );
    assert_eq!(voided, 32);
    assert!(q.is_empty());
}

pub fn is_empty(&mut self) -> bool

True only when the queue is genuinely drained.

A producer inside the dangling-tail window reads as non-empty, because its item is already committed to the chain even though the link is not written yet.

Consumer-side only.

Examples found in repository

examples/sample_app.rs (line 115)

fn base_order_fan_in() {
    println!("== base: order-entry gateways fan in to one matching engine ==");
    let q: Arc<MpscQueue<u64>> = Arc::new(MpscQueue::new());

    let gateways: Vec<_> = (0..GATEWAYS)
        .map(|g| {
            let q = Arc::clone(&q);
            thread::spawn(move || {
                for seq in 0..ORDERS_PER_GATEWAY {
                    q.push(order_id(g, seq));
                }
            })
        })
        .collect();
    for h in gateways {
        h.join().unwrap();
    }

    // Every gateway handle is joined, so the queue is uniquely owned again and
    // the single consumer can take it back out (try_pop needs &mut self).
    let mut q = Arc::into_inner(q).expect("all gateway handles dropped");

    let total = GATEWAYS * ORDERS_PER_GATEWAY;
    println!("  inbox depth before the match loop starts: {}", q.len());
    let first = *q.peek().expect("the inbox is not empty");
    println!(
        "  head of book: gateway {} seq {}",
        first >> 32,
        first & 0xffff_ffff
    );

    let mut per_gateway = [0usize; GATEWAYS];
    let mut last_seq = [None::<u64>; GATEWAYS];
    let mut matched = 0usize;
    loop {
        match q.try_pop() {
            PopResult::Some(order) => {
                let g = (order >> 32) as usize;
                let seq = order & 0xffff_ffff;
                if let Some(prev) = last_seq[g] {
                    assert!(seq > prev, "orders from one gateway stay in FIFO order");
                }
                last_seq[g] = Some(seq);
                per_gateway[g] += 1;
                matched += 1;
            }
            PopResult::Inconsistent => continue,
            PopResult::Empty => break,
        }
    }

    println!("  {GATEWAYS} gateways x {ORDERS_PER_GATEWAY} orders -> matched {matched}");
    println!("  per-gateway tally: {per_gateway:?}");
    assert_eq!(matched, total, "no order dropped, none duplicated");
    for count in per_gateway {
        assert_eq!(count, ORDERS_PER_GATEWAY, "every gateway fully drained");
    }

    // Kill-switch: a venue disconnect voids everything still queued rather than
    // matching it against a book that has moved on.
    for seq in 0..32 {
        q.push(order_id(0, seq));
    }
    let voided = q.clear();
    println!(
        "  kill switch voided {voided} queued orders, inbox empty: {}",
        q.is_empty()
    );
    assert_eq!(voided, 32);
    assert!(q.is_empty());
}

pub fn len(&mut self) -> usize

Count the linked items. O(n) in the backlog, and an estimate under a live producer, which is the same contract JCTools’ size() carries on its linked queues. Use it for alarms and sizing, not on the hot path.

Consumer-side only.

Examples found in repository

examples/sample_app.rs (line 72)

fn base_order_fan_in() {
    println!("== base: order-entry gateways fan in to one matching engine ==");
    let q: Arc<MpscQueue<u64>> = Arc::new(MpscQueue::new());

    let gateways: Vec<_> = (0..GATEWAYS)
        .map(|g| {
            let q = Arc::clone(&q);
            thread::spawn(move || {
                for seq in 0..ORDERS_PER_GATEWAY {
                    q.push(order_id(g, seq));
                }
            })
        })
        .collect();
    for h in gateways {
        h.join().unwrap();
    }

    // Every gateway handle is joined, so the queue is uniquely owned again and
    // the single consumer can take it back out (try_pop needs &mut self).
    let mut q = Arc::into_inner(q).expect("all gateway handles dropped");

    let total = GATEWAYS * ORDERS_PER_GATEWAY;
    println!("  inbox depth before the match loop starts: {}", q.len());
    let first = *q.peek().expect("the inbox is not empty");
    println!(
        "  head of book: gateway {} seq {}",
        first >> 32,
        first & 0xffff_ffff
    );

    let mut per_gateway = [0usize; GATEWAYS];
    let mut last_seq = [None::<u64>; GATEWAYS];
    let mut matched = 0usize;
    loop {
        match q.try_pop() {
            PopResult::Some(order) => {
                let g = (order >> 32) as usize;
                let seq = order & 0xffff_ffff;
                if let Some(prev) = last_seq[g] {
                    assert!(seq > prev, "orders from one gateway stay in FIFO order");
                }
                last_seq[g] = Some(seq);
                per_gateway[g] += 1;
                matched += 1;
            }
            PopResult::Inconsistent => continue,
            PopResult::Empty => break,
        }
    }

    println!("  {GATEWAYS} gateways x {ORDERS_PER_GATEWAY} orders -> matched {matched}");
    println!("  per-gateway tally: {per_gateway:?}");
    assert_eq!(matched, total, "no order dropped, none duplicated");
    for count in per_gateway {
        assert_eq!(count, ORDERS_PER_GATEWAY, "every gateway fully drained");
    }

    // Kill-switch: a venue disconnect voids everything still queued rather than
    // matching it against a book that has moved on.
    for seq in 0..32 {
        q.push(order_id(0, seq));
    }
    let voided = q.clear();
    println!(
        "  kill switch voided {voided} queued orders, inbox empty: {}",
        q.is_empty()
    );
    assert_eq!(voided, 32);
    assert!(q.is_empty());
}

pub fn clear(&mut self) -> usize

Drop everything the consumer can currently reach and return the count.

Best-effort: producers keep publishing throughout, so this is not a barrier and the queue is not guaranteed empty on return. It stops at the first dangling-tail window rather than spinning through it.

Consumer-side only.

Examples found in repository

examples/sample_app.rs (line 112)

fn base_order_fan_in() {
    println!("== base: order-entry gateways fan in to one matching engine ==");
    let q: Arc<MpscQueue<u64>> = Arc::new(MpscQueue::new());

    let gateways: Vec<_> = (0..GATEWAYS)
        .map(|g| {
            let q = Arc::clone(&q);
            thread::spawn(move || {
                for seq in 0..ORDERS_PER_GATEWAY {
                    q.push(order_id(g, seq));
                }
            })
        })
        .collect();
    for h in gateways {
        h.join().unwrap();
    }

    // Every gateway handle is joined, so the queue is uniquely owned again and
    // the single consumer can take it back out (try_pop needs &mut self).
    let mut q = Arc::into_inner(q).expect("all gateway handles dropped");

    let total = GATEWAYS * ORDERS_PER_GATEWAY;
    println!("  inbox depth before the match loop starts: {}", q.len());
    let first = *q.peek().expect("the inbox is not empty");
    println!(
        "  head of book: gateway {} seq {}",
        first >> 32,
        first & 0xffff_ffff
    );

    let mut per_gateway = [0usize; GATEWAYS];
    let mut last_seq = [None::<u64>; GATEWAYS];
    let mut matched = 0usize;
    loop {
        match q.try_pop() {
            PopResult::Some(order) => {
                let g = (order >> 32) as usize;
                let seq = order & 0xffff_ffff;
                if let Some(prev) = last_seq[g] {
                    assert!(seq > prev, "orders from one gateway stay in FIFO order");
                }
                last_seq[g] = Some(seq);
                per_gateway[g] += 1;
                matched += 1;
            }
            PopResult::Inconsistent => continue,
            PopResult::Empty => break,
        }
    }

    println!("  {GATEWAYS} gateways x {ORDERS_PER_GATEWAY} orders -> matched {matched}");
    println!("  per-gateway tally: {per_gateway:?}");
    assert_eq!(matched, total, "no order dropped, none duplicated");
    for count in per_gateway {
        assert_eq!(count, ORDERS_PER_GATEWAY, "every gateway fully drained");
    }

    // Kill-switch: a venue disconnect voids everything still queued rather than
    // matching it against a book that has moved on.
    for seq in 0..32 {
        q.push(order_id(0, seq));
    }
    let voided = q.clear();
    println!(
        "  kill switch voided {voided} queued orders, inbox empty: {}",
        q.is_empty()
    );
    assert_eq!(voided, 32);
    assert!(q.is_empty());
}

Trait Implementations

impl<T> Default for MpscQueue<T>

fn default() -> Self

Returns the “default value” for a type.

impl<T> Drop for MpscQueue<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 MpscQueue<T>

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