Struct ParkStrategy 

pub struct ParkStrategy { /* private fields */ }

Park the calling thread; the opposite-end wrapper unparks it on progress. Lowest CPU; adds wakeup latency.

Both ends share a Parker instance via Arc so the producer can wake the consumer and vice versa.

Implementations

impl ParkStrategy

pub fn pair() -> (Self, Self)

Build a matched (producer_strategy, consumer_strategy) pair that share a parking state.

Examples found in repository

examples/perf_features.rs (line 280)

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

    // The baseline: the base wait-free push + pop round trip. Swept as well as
    // sampled, because whether ring capacity moves the BASE op is the context
    // every feature curve is read against.
    let base_sweep = sweep("base/push+pop", |cap| {
        let (mut tx, mut rx) = pair(cap);
        batched(ITEMS_PER_SAMPLE, |i| {
            let _ = tx.try_push(i as u64);
            rx.try_pop().unwrap_or(0)
        })
    });
    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");

    // ---------- bulk: one fence per BULK_BATCH items ----------
    #[cfg(feature = "bulk")]
    {
        // A sample moves ITEMS_PER_SAMPLE items either way; only the call width
        // differs. That is the comparison the feature exists to win, and it is
        // why the reps count is divided rather than the batch grown.
        let sw = sweep("bulk/enqueue+dequeue", |cap| {
            let (mut tx, mut rx) = pair(cap);
            let batch = [0u64; BULK_BATCH];
            let mut out = [0u64; BULK_BATCH];
            batched(ITEMS_PER_SAMPLE / BULK_BATCH, |_| {
                let n = tx.try_enqueue_bulk(&batch);
                (n + rx.try_dequeue_bulk(&mut out)) as u64
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let batch = [0u64; BULK_BATCH];
        let mut out = [0u64; BULK_BATCH];
        let (mut tx, mut rx) = pair(CANON);
        let enq = single(
            |_| tx.try_enqueue_bulk(&batch) as u64,
            |_| rx.try_dequeue_bulk(&mut out) as u64,
        );
        let (mut tx, mut rx) = pair(CANON);
        let deq = single(
            |_| rx.try_dequeue_bulk(&mut out) as u64,
            |_| tx.try_enqueue_bulk(&batch) as u64,
        );
        let mut p99 = BTreeMap::new();
        p99.insert("enqueue_bulk".to_string(), enq);
        p99.insert("dequeue_bulk".to_string(), deq);
        manifest.set_feature("bulk", cat, &p99, &reason);
    }

    // ---------- wait-strategies: blocking wrappers, fast path only ----------
    #[cfg(feature = "wait-strategies")]
    {
        use subms_spsc_ring_buffer::{
            BlockingSpscConsumer, BlockingSpscProducer, BusySpin, ParkStrategy,
        };
        // Swept on BusySpin, whose `signal()` is a no-op, so the curve is the
        // wrapper's own overhead over the base ring and nothing else. The ring
        // is never full or empty, so `wait()` is never called; the parked-wakeup
        // path is a scheduler latency, not a per-op cost, and is not measured.
        let sw = sweep("wait-strategies/push+pop", |cap| {
            let (tx, rx) = pair(cap);
            let mut p = BlockingSpscProducer::new(tx, BusySpin);
            let mut c = BlockingSpscConsumer::new(rx, BusySpin);
            batched(ITEMS_PER_SAMPLE, |i| {
                p.push(i as u64);
                c.pop()
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        // ParkStrategy on the SAME ready ring, printed rather than classified.
        // Its `signal()` fires on every successful op even when nothing is
        // parked, and it takes a lock to do it, so the strategy choice shows up
        // here on the fast path - not in `wait()`, which this never enters.
        let park_batched = best(|| {
            let (ps, cs) = ParkStrategy::pair();
            let (tx, rx) = pair(CANON);
            let mut p = BlockingSpscProducer::new(tx, ps);
            let mut c = BlockingSpscConsumer::new(rx, cs);
            batched(ITEMS_PER_SAMPLE, |i| {
                p.push(i as u64);
                c.pop()
            })
        });
        eprintln!(
            "wait-strategies park fast path at {CANON}: {park_batched}ns per \
             {ITEMS_PER_SAMPLE}-item sample (spin {}ns) - signal() takes a lock per op",
            sw.iter().find(|(n, _)| *n == CANON).map_or(0, |(_, v)| *v)
        );

        let (tx, rx) = pair(CANON);
        let mut p = BlockingSpscProducer::new(tx, BusySpin);
        let mut c = BlockingSpscConsumer::new(rx, BusySpin);
        let push_spin = single(
            |i| {
                p.push(i as u64);
                0
            },
            |_| c.pop(),
        );
        let (tx, rx) = pair(CANON);
        let mut p = BlockingSpscProducer::new(tx, BusySpin);
        let mut c = BlockingSpscConsumer::new(rx, BusySpin);
        let pop_spin = single(
            |_| c.pop(),
            |i| {
                p.push(i as u64);
                0
            },
        );

        let (ps, cs) = ParkStrategy::pair();
        let (tx, rx) = pair(CANON);
        let mut p = BlockingSpscProducer::new(tx, ps);
        let mut c = BlockingSpscConsumer::new(rx, cs);
        let push_park = single(
            |i| {
                p.push(i as u64);
                0
            },
            |_| c.pop(),
        );
        let (ps, cs) = ParkStrategy::pair();
        let (tx, rx) = pair(CANON);
        let mut p = BlockingSpscProducer::new(tx, ps);
        let mut c = BlockingSpscConsumer::new(rx, cs);
        let pop_park = single(
            |_| c.pop(),
            |i| {
                p.push(i as u64);
                0
            },
        );

        let mut p99 = BTreeMap::new();
        p99.insert("push_spin".to_string(), push_spin);
        p99.insert("pop_spin".to_string(), pop_spin);
        p99.insert("push_park".to_string(), push_park);
        p99.insert("pop_park".to_string(), pop_park);
        manifest.set_feature("wait-strategies", cat, &p99, &reason);
    }

    // ---------- mpsc-fan-in: N rings, one round-robin consumer ----------
    #[cfg(feature = "mpsc-fan-in")]
    {
        use subms_spsc_ring_buffer::{MpscFanIn, MpscFanInConsumer, MpscFanInProducer};
        /// Producer count is held FIXED across the sweep. Moving it would sweep
        /// the fan-in width, which sets the consumer's probe loop, not the ring.
        const PRODUCERS: usize = 4;

        fn fanin(cap: usize) -> (Vec<MpscFanInProducer<u64>>, MpscFanInConsumer<u64>) {
            let (mut ps, c) = MpscFanIn::with_capacity::<u64>(PRODUCERS, cap);
            for p in ps.iter_mut() {
                for i in 0..cap / 2 {
                    let _ = p.try_push(i as u64);
                }
            }
            (ps, c)
        }

        // Pushes round-robin and the consumer cursor advances one ring per pop,
        // so the two stay in step and every ring holds a constant half load.
        // With every ring non-empty the consumer's probe hits on its first try,
        // which is the steady-state shape; a starved fan-in probes all N and that
        // is a different measurement.
        let sw = sweep("mpsc-fan-in/push+pop", |cap| {
            let (mut ps, mut c) = fanin(cap);
            batched(ITEMS_PER_SAMPLE, |i| {
                let _ = ps[i % PRODUCERS].try_push(i as u64);
                c.try_pop().unwrap_or(0)
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let (mut ps, mut c) = fanin(CANON);
        let enq = single(
            |i| {
                let _ = ps[i % PRODUCERS].try_push(i as u64);
                0
            },
            |_| c.try_pop().unwrap_or(0),
        );
        let (mut ps, mut c) = fanin(CANON);
        let deq = single(
            |_| c.try_pop().unwrap_or(0),
            |i| {
                let _ = ps[i % PRODUCERS].try_push(i as u64);
                0
            },
        );
        let mut p99 = BTreeMap::new();
        p99.insert("fanin_enqueue".to_string(), enq);
        p99.insert("fanin_dequeue".to_string(), deq);
        manifest.set_feature("mpsc-fan-in", cat, &p99, &reason);
    }

    // ---------- mpmc-disruptor: CAS claim + sequence barrier ----------
    #[cfg(feature = "mpmc-disruptor")]
    {
        use subms_spsc_ring_buffer::{DisruptorConsumer, DisruptorProducer, MpmcDisruptor};
        /// One consumer, held fixed. `try_publish` scans every consumer cursor
        /// before claiming, so the consumer count - not the capacity - is what
        /// that loop is O(). Sweeping it would answer a different question.
        const CONSUMERS: usize = 1;

        fn disruptor(cap: usize) -> (DisruptorProducer<u64>, DisruptorConsumer<u64>) {
            let (p, mut cs) = MpmcDisruptor::with_consumers::<u64>(cap, CONSUMERS);
            let c = cs.remove(0);
            for i in 0..cap / 2 {
                let _ = p.try_publish(i as u64);
            }
            (p, c)
        }

        // Half a ring of published-but-unconsumed items keeps the producer clear
        // of the gating spin (it only fires within one capacity of the slowest
        // consumer) and the consumer clear of the unpublished early return.
        let sw = sweep("mpmc-disruptor/publish+consume", |cap| {
            let (p, mut c) = disruptor(cap);
            batched(ITEMS_PER_SAMPLE, |i| {
                let _ = p.try_publish(i as u64);
                c.try_consume().unwrap_or(0)
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let (p, mut c) = disruptor(CANON);
        let published = single(
            |i| {
                let _ = p.try_publish(i as u64);
                0
            },
            |_| c.try_consume().unwrap_or(0),
        );
        let (p, mut c) = disruptor(CANON);
        let consumed = single(
            |_| c.try_consume().unwrap_or(0),
            |i| {
                let _ = p.try_publish(i as u64);
                0
            },
        );
        let mut p99 = BTreeMap::new();
        p99.insert("publish".to_string(), published);
        p99.insert("consume".to_string(), consumed);
        manifest.set_feature("mpmc-disruptor", cat, &p99, &reason);
    }

    // ---------- metrics: counters on the push / pop path ----------
    #[cfg(feature = "metrics")]
    {
        use subms_spsc_ring_buffer::InstrumentedSpsc;

        // The wrapper adds a `fetch_add` per op plus, on the producer side, a
        // high-water-mark update. `local_depth` only ever increments - a producer
        // cannot observe pops - so that update takes its CAS on every push rather
        // than settling once the mark stops moving. The counter cost on the push
        // path is two read-modify-writes, not one.
        let sw = sweep("metrics/push+pop", |cap| {
            let (tx, rx) = pair(cap);
            let (mut tx, mut rx, _m) = InstrumentedSpsc::wrap(tx, rx);
            batched(ITEMS_PER_SAMPLE, |i| {
                let _ = tx.try_push(i as u64);
                rx.try_pop().unwrap_or(0)
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50), None);

        let (tx, rx) = pair(CANON);
        let (mut tx, mut rx, _m) = InstrumentedSpsc::wrap(tx, rx);
        let enq = single(
            |i| {
                let _ = tx.try_push(i as u64);
                0
            },
            |_| rx.try_pop().unwrap_or(0),
        );
        let (tx, rx) = pair(CANON);
        let (mut tx, mut rx, _m) = InstrumentedSpsc::wrap(tx, rx);
        let deq = single(
            |_| rx.try_pop().unwrap_or(0),
            |i| {
                let _ = tx.try_push(i as u64);
                0
            },
        );
        let mut p99 = BTreeMap::new();
        p99.insert("metrics_enqueue".to_string(), enq);
        p99.insert("metrics_dequeue".to_string(), deq);
        manifest.set_feature("metrics", cat, &p99, &reason);
    }

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

Trait Implementations

impl WaitStrategy for ParkStrategy

fn wait(&mut self)

Wait briefly. Called between retries when the ring isn’t ready.

fn signal(&self)

Signal the waiter. Default is a no-op; only ParkStrategy overrides.