subms_mpsc_queue/
recipe.rs

1//! `SubMsRecipe` impl. Behind the `harness` feature.
2
3use std::sync::Arc;
4use std::thread;
5
6use subms::{SubMsBenchParams, SubMsPerfHarness, SubMsRecipe, SubMsStageKind, SubMsTimer};
7
8use crate::{MpscQueue, PopResult};
9
10/// Stages: `offer`, `poll`. Four producer threads share one queue; a single
11/// consumer drains. Per-op timing is recorded on the thread that does the op.
12pub struct MpscQueueRecipe;
13
14impl SubMsRecipe for MpscQueueRecipe {
15    fn name(&self) -> &str {
16        "mpsc-queue"
17    }
18
19    fn run(&self, h: &mut SubMsPerfHarness, params: &SubMsBenchParams) {
20        let entries = params.entries;
21        let warmup = params.warmup;
22        let producers = 4usize;
23        let per_producer = entries / producers;
24
25        let q: Arc<MpscQueue<u64>> = Arc::new(MpscQueue::new());
26
27        // Warm-up: brief uncontended push/pop pair on one thread.
28        {
29            let q_ptr = Arc::as_ptr(&q) as *mut MpscQueue<u64>;
30            let q_mut = unsafe { &mut *q_ptr };
31            for i in 0..warmup as u64 {
32                q.push(i);
33                loop {
34                    match q_mut.try_pop() {
35                        PopResult::Some(_) => break,
36                        _ => std::hint::spin_loop(),
37                    }
38                }
39            }
40        }
41
42        // Producers each record their per-offer latencies into a private vec,
43        // then return it for the main thread to feed into the harness.
44        let mut producer_handles = Vec::with_capacity(producers);
45        for tid in 0..producers as u64 {
46            let q = q.clone();
47            producer_handles.push(thread::spawn(move || {
48                let mut samples = Vec::with_capacity(per_producer);
49                for i in 0..per_producer as u64 {
50                    let t0 = SubMsTimer::tick();
51                    q.push((tid << 32) | i);
52                    samples.push(t0.elapsed_ns());
53                }
54                samples
55            }));
56        }
57
58        let consumer_q = q.clone();
59        let total = producers * per_producer;
60        let consumer = thread::spawn(move || {
61            let q_ptr = Arc::as_ptr(&consumer_q) as *mut MpscQueue<u64>;
62            let q_mut = unsafe { &mut *q_ptr };
63            let mut samples = Vec::with_capacity(total);
64            let mut count = 0usize;
65            while count < total {
66                let t0 = SubMsTimer::tick();
67                match q_mut.try_pop() {
68                    PopResult::Some(_) => {
69                        samples.push(t0.elapsed_ns());
70                        count += 1;
71                    }
72                    _ => std::hint::spin_loop(),
73                }
74            }
75            samples
76        });
77
78        let s_offer = h.stage("offer", total).with_kind(SubMsStageKind::HotPath);
79        for handle in producer_handles {
80            for ns in handle.join().expect("producer joined") {
81                s_offer.record(ns);
82            }
83        }
84        let poll_samples = consumer.join().expect("consumer joined");
85        let s_poll = h
86            .stage("poll", poll_samples.len())
87            .with_kind(SubMsStageKind::HotPath);
88        for ns in poll_samples {
89            s_poll.record(ns);
90        }
91        h.add_meta("producers", &producers.to_string());
92    }
93}
subms_mpsc_queue/recipe.rs

subms_mpsc_queue/
recipe.rs
