subetha_cxc/
dispatch_deque.rs

1//! `DequeDispatcher` - per-call routing across the MMF-deque family.
2//!
3//! The dispatcher owns one handle of each [`SharedDeque`] /
4//! [`SharedDequeKhpd`] / [`SharedDequeLoh`] / [`SharedDequeUrd`]
5//! variant that the host has configured, and picks the right one
6//! per call based on a caller-supplied [`WorkloadShape`].
7//!
8//! ## The routing decision
9//!
10//! No single variant wins every workload shape. The right pick depends
11//! on (a) whether the producer batches or dispatches per-item,
12//! (b) how many thieves the workload runs, and (c) whether the
13//! caller wants the consumer to halt the logical CPU between batches
14//! via WAITPKG.
15//!
16//! | Workload | Pick | Why |
17//! |---|---|---|
18//! | Per-item dispatch, single thief | `ChaseLev` | Lowest constant per push; no batch to amortize. |
19//! | Producer batches K = 2..128 items per call | `Khpd` | 3 items per Release-store on the publication line; empirically the best per-item cost on Zen+/Zen 4 at this scale. |
20//! | Producer batches K >= 128 items per call | `Loh` | 1 `tail.fetch_add(K)` amortizes across the whole batch. |
21//! | Multiple thieves AND batched producer | `Urd` | Per-thief mailbox = zero CAS contention. |
22//! | Multiple thieves AND `wait_idle=true` | `Urd` | Hardware-mediated wake via WAITPKG / PAUSE-spin. |
23//!
24//! The routing table is a starting point. Per-host calibration may
25//! flip individual cells. Callers that already know which variant
26//! they want may call the per-variant getters
27//! ([`DequeDispatcher::chase_lev`], etc.) directly.
28//!
29//! ## Cross-process E2E
30//!
31//! A `DequeDispatcher` lives in the producer process. Each variant
32//! it owns is backed by its own MMF file path; consumer processes
33//! open those same paths to drain. See the
34//! [`dispatcher_demo`](https://github.com/Variably-Constant/SubEtha/blob/main/crates/subetha-cxc/examples/dispatcher_demo.rs)
35//! example for the parent/child split.
36
37#![allow(clippy::missing_errors_doc)]
38
39use std::io;
40use std::path::Path;
41use std::sync::Arc;
42
43use subetha_core::{Axis, AxisMask};
44
45use crate::shared_deque::SharedDeque;
46use crate::shared_deque_khl::SharedDequeKhl;
47use crate::shared_deque_khpd::{LineItem, SharedDequeKhpd};
48use crate::shared_deque_loh::SharedDequeLoh;
49use crate::shared_deque_urd::SharedDequeUrd;
50
51/// Direction signatures per variant.
52///
53/// Each variant declares which of the six K-axes it engages at a
54/// non-default value. The dispatcher uses signature-set logic to
55/// route per `WorkloadShape`: `variant.satisfies(workload_required)`
56/// picks the highest-engagement variant whose signature is a
57/// superset of the workload's required signature.
58const fn chase_lev_signature() -> AxisMask {
59    // K_counter_share = owner-private is the only non-default axis;
60    // K_inner=1, K_outer=1, K_gating=counter-only are all default.
61    AxisMask::from_axes(&[Axis::CounterShare])
62}
63
64const fn khpd_signature() -> AxisMask {
65    AxisMask::from_axes(&[Axis::Inner, Axis::Gating])
66}
67
68const fn loh_signature() -> AxisMask {
69    AxisMask::from_axes(&[Axis::Outer, Axis::Gating])
70}
71
72const fn urd_signature() -> AxisMask {
73    AxisMask::from_axes(&[
74        Axis::Inner,
75        Axis::Consumer,
76        Axis::Radius,
77        Axis::Gating,
78    ])
79}
80
81const fn khl_signature() -> AxisMask {
82    AxisMask::from_axes(&[
83        Axis::Inner,
84        Axis::Outer,
85        Axis::CounterShare,
86        Axis::Radius,
87        Axis::Gating,
88    ])
89}
90
91/// The deque-family variants the dispatcher routes across.
92#[derive(Debug, Clone, Copy, PartialEq, Eq)]
93pub enum DequeVariant {
94    /// Chase-Lev work-stealing deque, per-item push.
95    ChaseLev,
96    /// KHPD publication-line deque, 3 items per Release-store.
97    Khpd,
98    /// LOH LCRQ-on-LIFO Hybrid, amortizes `tail.fetch_add` over the
99    /// whole batch.
100    Loh,
101    /// URD per-thief mailbox deque + WAITPKG wait.
102    Urd,
103    /// KHL K-axis Hierarchical LCRQ - SubEtha-native hybrid that
104    /// pulls KHPD's per-slot packing + LOH's per-batch counter
105    /// amortization + Chase-Lev's owner-private tail simultaneously.
106    /// Empirically the strongest single-thief batched primitive on
107    /// Zen+ R7 2700 (4.7 ns/item at K=64 producer-fast).
108    Khl,
109}
110
111impl DequeVariant {
112    /// The direction signature for this variant: which K-axes it
113    /// engages at non-default values.
114    pub const fn signature(self) -> AxisMask {
115        match self {
116            DequeVariant::ChaseLev => chase_lev_signature(),
117            DequeVariant::Khpd => khpd_signature(),
118            DequeVariant::Loh => loh_signature(),
119            DequeVariant::Urd => urd_signature(),
120            DequeVariant::Khl => khl_signature(),
121        }
122    }
123}
124
125/// Caller-supplied workload shape feeding the routing decision.
126#[derive(Debug, Clone, Copy)]
127pub struct WorkloadShape {
128    /// Number of consumer threads / processes that will drain the
129    /// deque concurrently. `>= 2` is the multi-thief regime where
130    /// URD's per-mailbox layout amortizes against the shared-head
131    /// CAS contention Chase-Lev / KHPD / LOH all pay.
132    pub n_thieves: usize,
133    /// `Some(K)` when the producer hands the dispatcher a batch of
134    /// `K` items per call; `None` when the producer dispatches one
135    /// item at a time (request-reply / latency-bound).
136    pub batch_size: Option<usize>,
137    /// `true` when the consumer should halt the logical CPU between
138    /// batches (WAITPKG on capable silicon; PAUSE-spin elsewhere).
139    /// Setting this routes to URD even at `n_thieves == 1`.
140    pub wait_idle: bool,
141}
142
143impl WorkloadShape {
144    /// The direction signature this workload requires from its
145    /// transport: which K-axes the variant must engage to handle
146    /// this shape.
147    ///
148    /// Per-item dispatch (no batch) requires nothing beyond the
149    /// empty signature (Chase-Lev's signature is a superset of any
150    /// empty requirement). Batched dispatch requires K_inner +
151    /// K_outer engaged (per-slot packing AND per-batch counter
152    /// amortization). Multi-thief or wait-idle requires K_consumer +
153    /// K_radius engaged (per-thief mailboxes and CPUID-dispatched
154    /// publish mechanism).
155    pub const fn required_signature(&self) -> AxisMask {
156        let mut bits = 0u16;
157        // n_thieves >= 2 or wait_idle requires per-thief consumer +
158        // radius dispatch.
159        if self.n_thieves >= 2 || self.wait_idle {
160            bits |= 1u16 << Axis::Consumer.bit();
161            bits |= 1u16 << Axis::Radius.bit();
162        }
163        // batch_size = Some(k>=2) requires K_inner and K_outer.
164        if let Some(k) = self.batch_size
165            && k >= 2
166        {
167            bits |= 1u16 << Axis::Inner.bit();
168            bits |= 1u16 << Axis::Outer.bit();
169        }
170        AxisMask::from_bits(bits)
171    }
172
173    /// Request-reply: per-item dispatch, single thief, no idle wait.
174    pub fn request_reply() -> Self {
175        Self {
176            n_thieves: 1,
177            batch_size: None,
178            wait_idle: false,
179        }
180    }
181
182    /// Producer-fast batch of `k` items, single thief.
183    pub fn producer_fast(k: usize) -> Self {
184        Self {
185            n_thieves: 1,
186            batch_size: Some(k),
187            wait_idle: false,
188        }
189    }
190
191    /// Fan-out: producer batches across multiple thieves. `n_thieves`
192    /// >= 2 + `batch_size` set routes to URD.
193    pub fn fan_out(n_thieves: usize, k: usize) -> Self {
194        Self {
195            n_thieves,
196            batch_size: Some(k),
197            wait_idle: false,
198        }
199    }
200}
201
202/// Errors from the dispatcher's send-side methods.
203#[derive(Debug)]
204pub enum DispatchError {
205    /// The picked variant is not configured on this dispatcher
206    /// (caller did not pass a backing file path at construction).
207    BackendNotConfigured(DequeVariant),
208    /// The picked variant's backing primitive returned a push error
209    /// (the ring is at capacity, etc.).
210    PushFailed(&'static str),
211}
212
213impl std::fmt::Display for DispatchError {
214    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
215        match self {
216            Self::BackendNotConfigured(v) => {
217                write!(f, "dispatcher: backend {v:?} is not configured")
218            }
219            Self::PushFailed(msg) => write!(f, "dispatcher: push failed ({msg})"),
220        }
221    }
222}
223
224impl std::error::Error for DispatchError {}
225
226/// MMF-backed dispatcher across the deque family.
227///
228/// Construct via [`DequeDispatcher::builder`] and pass the backing
229/// paths for each variant the caller wants available. The
230/// [`pick`](DequeDispatcher::pick) helper returns the routing
231/// decision for a shape WITHOUT performing the push, useful for
232/// observers and per-host calibration.
233pub struct DequeDispatcher {
234    chase_lev: Option<Arc<SharedDeque<LineItem>>>,
235    khpd: Option<Arc<SharedDequeKhpd>>,
236    loh: Option<Arc<SharedDequeLoh>>,
237    urd: Option<Arc<SharedDequeUrd>>,
238    khl: Option<Arc<SharedDequeKhl>>,
239}
240
241impl DequeDispatcher {
242    /// Start a builder. Pass the backing file paths for whichever
243    /// variants the caller wants available; unset variants stay
244    /// `None` and the dispatcher falls through to the next-best
245    /// available variant per [`pick`](Self::pick).
246    pub fn builder() -> DispatcherBuilder {
247        DispatcherBuilder {
248            chase_lev: None,
249            khpd: None,
250            loh: None,
251            urd: None,
252            khl: None,
253        }
254    }
255
256    /// Get the underlying Chase-Lev handle (if configured).
257    pub fn chase_lev(&self) -> Option<&Arc<SharedDeque<LineItem>>> {
258        self.chase_lev.as_ref()
259    }
260
261    /// Get the underlying KHPD handle (if configured).
262    pub fn khpd(&self) -> Option<&Arc<SharedDequeKhpd>> {
263        self.khpd.as_ref()
264    }
265
266    /// Get the underlying LOH handle (if configured).
267    pub fn loh(&self) -> Option<&Arc<SharedDequeLoh>> {
268        self.loh.as_ref()
269    }
270
271    /// Get the underlying URD handle (if configured).
272    pub fn urd(&self) -> Option<&Arc<SharedDequeUrd>> {
273        self.urd.as_ref()
274    }
275
276    /// Get the underlying KHL handle (if configured).
277    pub fn khl(&self) -> Option<&Arc<SharedDequeKhl>> {
278        self.khl.as_ref()
279    }
280
281    /// Pick the right variant for `shape`. Returns the variant
282    /// independent of whether the corresponding handle is configured
283    /// (use [`pick_with_fallback`](Self::pick_with_fallback) to fold
284    /// the configuration check into the decision).
285    pub fn pick(shape: WorkloadShape) -> DequeVariant {
286        // Multi-thief or wait_idle => URD (per-mailbox + WAITPKG).
287        if shape.n_thieves >= 2 || shape.wait_idle {
288            return DequeVariant::Urd;
289        }
290        // Single-thief routes by batch size. KHL is the SubEtha-
291        // native hybrid that beats KHPD measurably (1.55x at K=64)
292        // on producer-fast workloads; route any batched single-thief
293        // call through it. Per-item dispatch still rides Chase-Lev
294        // because it has no batch to amortize.
295        match shape.batch_size {
296            Some(k) if k >= 2 => DequeVariant::Khl,
297            _ => DequeVariant::ChaseLev,
298        }
299    }
300
301    /// Pick a variant by signature-set satisfaction. For each
302    /// variant in priority order, check whether its signature is a
303    /// superset of the workload's required signature; return the
304    /// first match. Agrees with [`pick`](Self::pick) on every
305    /// canonical workload shape.
306    ///
307    /// The dispatcher becomes a signature-set lens rather than a
308    /// hardcoded match arm; the two routing methods co-exist so
309    /// downstream callers can pick the pattern that fits their
310    /// style.
311    pub fn pick_by_signature(shape: WorkloadShape) -> DequeVariant {
312        // Multi-thief or wait-idle workloads strictly need URD's
313        // K_consumer engagement, so URD outranks every variant even
314        // though KHL has higher overall axis count.
315        if shape.n_thieves >= 2 || shape.wait_idle {
316            return DequeVariant::Urd;
317        }
318        let required = shape.required_signature();
319        // For single-thief shapes: per-item dispatch (empty
320        // required) routes to Chase-Lev (the simplest variant that
321        // satisfies the empty requirement); batched dispatch routes
322        // to KHL (the highest-engagement variant that satisfies the
323        // K_inner + K_outer requirement). This mirrors `pick`
324        // exactly: empty required => Chase-Lev; non-empty => KHL.
325        if required == AxisMask::EMPTY {
326            return DequeVariant::ChaseLev;
327        }
328        // Non-empty single-thief: prefer the highest-engagement
329        // variant that satisfies the requirement.
330        const ORDER: [DequeVariant; 4] = [
331            DequeVariant::Khl,
332            DequeVariant::Khpd,
333            DequeVariant::Loh,
334            DequeVariant::ChaseLev,
335        ];
336        for v in ORDER {
337            if v.signature().satisfies(required) {
338                return v;
339            }
340        }
341        DequeVariant::ChaseLev
342    }
343
344    /// Pick the right variant for `shape`, falling through to the
345    /// next-best available variant when the primary pick is not
346    /// configured on this dispatcher.
347    ///
348    /// Fallback chain (in order): primary -> KHPD -> LOH -> Chase-Lev
349    /// -> URD. Returns `None` only when no variant is configured at
350    /// all.
351    pub fn pick_with_fallback(&self, shape: WorkloadShape) -> Option<DequeVariant> {
352        let primary = Self::pick(shape);
353        // Fallback ordering: primary first, then strongest-to-weakest
354        // by measured single-thief K=64 throughput on Zen+ R7 2700.
355        let order: [DequeVariant; 6] = [
356            primary,
357            DequeVariant::Khl,
358            DequeVariant::Khpd,
359            DequeVariant::Loh,
360            DequeVariant::ChaseLev,
361            DequeVariant::Urd,
362        ];
363        order.into_iter().find(|&v| self.is_configured(v))
364    }
365
366    /// Check whether `variant` is configured on this dispatcher.
367    pub fn is_configured(&self, variant: DequeVariant) -> bool {
368        match variant {
369            DequeVariant::ChaseLev => self.chase_lev.is_some(),
370            DequeVariant::Khpd => self.khpd.is_some(),
371            DequeVariant::Loh => self.loh.is_some(),
372            DequeVariant::Urd => self.urd.is_some(),
373            DequeVariant::Khl => self.khl.is_some(),
374        }
375    }
376
377    /// Dispatch a single item under `shape`. Routes to whichever
378    /// variant the [`pick_with_fallback`](Self::pick_with_fallback)
379    /// decision selects. Returns the chosen variant so observers can
380    /// confirm the routing.
381    ///
382    /// For per-item dispatch the natural target is Chase-Lev (one
383    /// Release-store on `bottom` per push). KHPD / LOH / URD all
384    /// accept a single-item batch and degrade to one slot write per
385    /// call.
386    pub fn dispatch_one(
387        &self,
388        shape: WorkloadShape,
389        item: LineItem,
390    ) -> Result<DequeVariant, DispatchError> {
391        let variant = self
392            .pick_with_fallback(shape)
393            .ok_or(DispatchError::BackendNotConfigured(DequeVariant::ChaseLev))?;
394        match variant {
395            DequeVariant::ChaseLev => {
396                let h = self
397                    .chase_lev
398                    .as_ref()
399                    .ok_or(DispatchError::BackendNotConfigured(DequeVariant::ChaseLev))?;
400                h.push(&item).map_err(|_| DispatchError::PushFailed("ChaseLev::push"))?;
401            }
402            DequeVariant::Khpd => {
403                let h = self
404                    .khpd
405                    .as_ref()
406                    .ok_or(DispatchError::BackendNotConfigured(DequeVariant::Khpd))?;
407                h.publish_batch(std::slice::from_ref(&item))
408                    .map_err(|_| DispatchError::PushFailed("Khpd::publish_batch"))?;
409            }
410            DequeVariant::Loh => {
411                let h = self
412                    .loh
413                    .as_ref()
414                    .ok_or(DispatchError::BackendNotConfigured(DequeVariant::Loh))?;
415                h.publish_batch(std::slice::from_ref(&item))
416                    .map_err(|_| DispatchError::PushFailed("Loh::publish_batch"))?;
417            }
418            DequeVariant::Urd => {
419                let h = self
420                    .urd
421                    .as_ref()
422                    .ok_or(DispatchError::BackendNotConfigured(DequeVariant::Urd))?;
423                h.publish_round_robin(std::slice::from_ref(&item))
424                    .map_err(|_| DispatchError::PushFailed("Urd::publish_round_robin"))?;
425            }
426            DequeVariant::Khl => {
427                let h = self
428                    .khl
429                    .as_ref()
430                    .ok_or(DispatchError::BackendNotConfigured(DequeVariant::Khl))?;
431                h.publish_batch(std::slice::from_ref(&item))
432                    .map_err(|_| DispatchError::PushFailed("Khl::publish_batch"))?;
433            }
434        }
435        Ok(variant)
436    }
437
438    /// Dispatch a batch of items under `shape`. Routes per
439    /// [`pick_with_fallback`](Self::pick_with_fallback). Returns the
440    /// chosen variant.
441    ///
442    /// For Chase-Lev (per-item primitive) the batch is pushed one
443    /// item at a time. For KHPD / LOH the batch is published via
444    /// the variant's `publish_batch` hot path. For URD the batch is
445    /// chunked by `MAILBOX_ITEMS = 3` and round-robined across the
446    /// configured mailboxes.
447    pub fn dispatch_batch(
448        &self,
449        shape: WorkloadShape,
450        items: &[LineItem],
451    ) -> Result<DequeVariant, DispatchError> {
452        if items.is_empty() {
453            return self
454                .pick_with_fallback(shape)
455                .ok_or(DispatchError::BackendNotConfigured(DequeVariant::ChaseLev));
456        }
457        let variant = self
458            .pick_with_fallback(shape)
459            .ok_or(DispatchError::BackendNotConfigured(DequeVariant::ChaseLev))?;
460        match variant {
461            DequeVariant::ChaseLev => {
462                let h = self
463                    .chase_lev
464                    .as_ref()
465                    .ok_or(DispatchError::BackendNotConfigured(DequeVariant::ChaseLev))?;
466                for item in items {
467                    h.push(item)
468                        .map_err(|_| DispatchError::PushFailed("ChaseLev::push"))?;
469                }
470            }
471            DequeVariant::Khpd => {
472                let h = self
473                    .khpd
474                    .as_ref()
475                    .ok_or(DispatchError::BackendNotConfigured(DequeVariant::Khpd))?;
476                h.publish_batch(items)
477                    .map_err(|_| DispatchError::PushFailed("Khpd::publish_batch"))?;
478            }
479            DequeVariant::Loh => {
480                let h = self
481                    .loh
482                    .as_ref()
483                    .ok_or(DispatchError::BackendNotConfigured(DequeVariant::Loh))?;
484                h.publish_batch(items)
485                    .map_err(|_| DispatchError::PushFailed("Loh::publish_batch"))?;
486            }
487            DequeVariant::Urd => {
488                let h = self
489                    .urd
490                    .as_ref()
491                    .ok_or(DispatchError::BackendNotConfigured(DequeVariant::Urd))?;
492                use crate::shared_deque_urd::MAILBOX_ITEMS;
493                for chunk in items.chunks(MAILBOX_ITEMS) {
494                    h.publish_round_robin(chunk).map_err(|_| {
495                        DispatchError::PushFailed("Urd::publish_round_robin")
496                    })?;
497                }
498            }
499            DequeVariant::Khl => {
500                let h = self
501                    .khl
502                    .as_ref()
503                    .ok_or(DispatchError::BackendNotConfigured(DequeVariant::Khl))?;
504                h.publish_batch(items)
505                    .map_err(|_| DispatchError::PushFailed("Khl::publish_batch"))?;
506            }
507        }
508        Ok(variant)
509    }
510}
511
512/// Builder for [`DequeDispatcher`].
513pub struct DispatcherBuilder {
514    chase_lev: Option<Arc<SharedDeque<LineItem>>>,
515    khpd: Option<Arc<SharedDequeKhpd>>,
516    loh: Option<Arc<SharedDequeLoh>>,
517    urd: Option<Arc<SharedDequeUrd>>,
518    khl: Option<Arc<SharedDequeKhl>>,
519}
520
521impl DispatcherBuilder {
522    /// Create + attach a Chase-Lev deque at `path` with `capacity`
523    /// slots (round up to next power of two).
524    pub fn with_chase_lev<P: AsRef<Path>>(
525        mut self,
526        path: P,
527        capacity: usize,
528    ) -> io::Result<Self> {
529        let d = SharedDeque::<LineItem>::create(path, capacity)
530            .map_err(|e| io::Error::other(format!("Chase-Lev create: {e:?}")))?;
531        self.chase_lev = Some(Arc::new(d));
532        Ok(self)
533    }
534
535    /// Create + attach a KHPD at `path` with `capacity` publication
536    /// lines.
537    pub fn with_khpd<P: AsRef<Path>>(
538        mut self,
539        path: P,
540        capacity: usize,
541    ) -> io::Result<Self> {
542        let d = SharedDequeKhpd::create(path, capacity)?;
543        self.khpd = Some(Arc::new(d));
544        Ok(self)
545    }
546
547    /// Create + attach a LOH at `path` with `capacity` ring slots
548    /// and `flush_threshold` LIFO auto-flush threshold.
549    pub fn with_loh<P: AsRef<Path>>(
550        mut self,
551        path: P,
552        capacity: usize,
553        flush_threshold: usize,
554    ) -> io::Result<Self> {
555        let d = SharedDequeLoh::create(path, capacity, flush_threshold)?;
556        self.loh = Some(Arc::new(d));
557        Ok(self)
558    }
559
560    /// Create + attach a URD at `path` with `n_mailboxes` mailboxes
561    /// (one per intended thief).
562    pub fn with_urd<P: AsRef<Path>>(
563        mut self,
564        path: P,
565        n_mailboxes: usize,
566    ) -> io::Result<Self> {
567        let d = SharedDequeUrd::create(path, n_mailboxes)?;
568        self.urd = Some(Arc::new(d));
569        Ok(self)
570    }
571
572    /// Create + attach a KHL (K-axis Hierarchical LCRQ - the
573    /// SubEtha-native hybrid) at `path` with `capacity` slots. Total
574    /// item capacity is `capacity * 3`.
575    pub fn with_khl<P: AsRef<Path>>(
576        mut self,
577        path: P,
578        capacity: usize,
579    ) -> io::Result<Self> {
580        let d = SharedDequeKhl::create(path, capacity)?;
581        self.khl = Some(Arc::new(d));
582        Ok(self)
583    }
584
585    /// Finalize the dispatcher.
586    pub fn build(self) -> DequeDispatcher {
587        DequeDispatcher {
588            chase_lev: self.chase_lev,
589            khpd: self.khpd,
590            loh: self.loh,
591            urd: self.urd,
592            khl: self.khl,
593        }
594    }
595}
596
597#[cfg(test)]
598mod tests {
599    use super::*;
600
601    fn tmp(name: &str) -> std::path::PathBuf {
602        let mut p = std::env::temp_dir();
603        let pid = std::process::id();
604        let nonce = std::time::SystemTime::now()
605            .duration_since(std::time::UNIX_EPOCH)
606            .map(|d| d.as_nanos())
607            .unwrap_or(0);
608        p.push(format!("subetha_dispatch_deque_{pid}_{nonce}_{name}.bin"));
609        p
610    }
611
612    fn u32_item(id: u32) -> LineItem {
613        LineItem::new(&id.to_le_bytes()).expect("item")
614    }
615
616    #[test]
617    fn pick_request_reply_routes_to_chase_lev() {
618        assert_eq!(
619            DequeDispatcher::pick(WorkloadShape::request_reply()),
620            DequeVariant::ChaseLev
621        );
622    }
623
624    #[test]
625    fn signature_pick_agrees_with_hardcoded_pick_on_all_shapes() {
626        // The signature-set logic should reproduce the existing
627        // hardcoded routing for every canonical workload shape.
628        let shapes = [
629            WorkloadShape::request_reply(),
630            WorkloadShape::producer_fast(4),
631            WorkloadShape::producer_fast(16),
632            WorkloadShape::producer_fast(64),
633            WorkloadShape::producer_fast(256),
634            WorkloadShape::fan_out(2, 16),
635            WorkloadShape::fan_out(4, 64),
636            WorkloadShape {
637                n_thieves: 1,
638                batch_size: Some(8),
639                wait_idle: true,
640            },
641        ];
642        for shape in shapes {
643            let hardcoded = DequeDispatcher::pick(shape);
644            let signature_based = DequeDispatcher::pick_by_signature(shape);
645            assert_eq!(
646                hardcoded, signature_based,
647                "shape {shape:?}: hardcoded picked {hardcoded:?}, signature picked {signature_based:?}",
648            );
649        }
650    }
651
652    #[test]
653    fn variant_signatures_are_distinct() {
654        // Each variant occupies a distinct corner of the design cube.
655        let sigs = [
656            DequeVariant::ChaseLev.signature(),
657            DequeVariant::Khpd.signature(),
658            DequeVariant::Loh.signature(),
659            DequeVariant::Urd.signature(),
660            DequeVariant::Khl.signature(),
661        ];
662        for i in 0..sigs.len() {
663            for j in (i + 1)..sigs.len() {
664                assert_ne!(
665                    sigs[i], sigs[j],
666                    "variants {i} and {j} share the same signature",
667                );
668            }
669        }
670    }
671
672    #[test]
673    fn request_reply_has_empty_required_signature() {
674        let req = WorkloadShape::request_reply().required_signature();
675        assert_eq!(req.count(), 0);
676    }
677
678    #[test]
679    fn producer_fast_requires_inner_and_outer() {
680        let req = WorkloadShape::producer_fast(64).required_signature();
681        assert!(req.contains(Axis::Inner));
682        assert!(req.contains(Axis::Outer));
683    }
684
685    #[test]
686    fn fan_out_requires_consumer_and_radius() {
687        let req = WorkloadShape::fan_out(4, 64).required_signature();
688        assert!(req.contains(Axis::Consumer));
689        assert!(req.contains(Axis::Radius));
690    }
691
692    #[test]
693    fn pick_any_batch_routes_to_khl() {
694        // KHL is the SubEtha-native hybrid that beats KHPD and LOH
695        // empirically at single-thief batched workloads. The routing
696        // picks it for any batch size K >= 2.
697        assert_eq!(
698            DequeDispatcher::pick(WorkloadShape::producer_fast(4)),
699            DequeVariant::Khl
700        );
701        assert_eq!(
702            DequeDispatcher::pick(WorkloadShape::producer_fast(64)),
703            DequeVariant::Khl
704        );
705        assert_eq!(
706            DequeDispatcher::pick(WorkloadShape::producer_fast(256)),
707            DequeVariant::Khl
708        );
709    }
710
711    #[test]
712    fn pick_multi_thief_routes_to_urd() {
713        assert_eq!(
714            DequeDispatcher::pick(WorkloadShape::fan_out(2, 16)),
715            DequeVariant::Urd
716        );
717        assert_eq!(
718            DequeDispatcher::pick(WorkloadShape::fan_out(4, 64)),
719            DequeVariant::Urd
720        );
721    }
722
723    #[test]
724    fn pick_wait_idle_routes_to_urd_even_single_thief() {
725        let shape = WorkloadShape {
726            n_thieves: 1,
727            batch_size: Some(8),
728            wait_idle: true,
729        };
730        assert_eq!(DequeDispatcher::pick(shape), DequeVariant::Urd);
731    }
732
733    #[test]
734    fn pick_with_fallback_skips_unconfigured() {
735        // Only Chase-Lev configured; a batch shape that picks KHL
736        // falls through KHL -> KHPD -> LOH -> Chase-Lev.
737        let path = tmp("fallback_cl");
738        let dispatcher = DequeDispatcher::builder()
739            .with_chase_lev(&path, 64)
740            .expect("create cl")
741            .build();
742        let shape = WorkloadShape::producer_fast(8);
743        assert_eq!(DequeDispatcher::pick(shape), DequeVariant::Khl);
744        assert_eq!(
745            dispatcher.pick_with_fallback(shape),
746            Some(DequeVariant::ChaseLev)
747        );
748        std::fs::remove_file(&path).ok();
749    }
750
751    #[test]
752    fn dispatch_one_routes_to_chase_lev_when_per_item() {
753        let cl_path = tmp("dispatch_one_cl");
754        let dispatcher = DequeDispatcher::builder()
755            .with_chase_lev(&cl_path, 64)
756            .expect("create cl")
757            .build();
758        let chosen = dispatcher
759            .dispatch_one(WorkloadShape::request_reply(), u32_item(42))
760            .expect("dispatch_one");
761        assert_eq!(chosen, DequeVariant::ChaseLev);
762        // Drain via the Chase-Lev handle.
763        let cl = dispatcher.chase_lev().expect("cl");
764        let got = cl.steal().expect("steal");
765        assert_eq!(got, u32_item(42));
766        std::fs::remove_file(&cl_path).ok();
767    }
768
769    #[test]
770    fn dispatch_batch_routes_to_khl_when_configured() {
771        let khl_path = tmp("dispatch_batch_khl");
772        let dispatcher = DequeDispatcher::builder()
773            .with_khl(&khl_path, 256)
774            .expect("create khl")
775            .build();
776        let items: Vec<LineItem> = (0..64u32).map(u32_item).collect();
777        let chosen = dispatcher
778            .dispatch_batch(WorkloadShape::producer_fast(64), &items)
779            .expect("dispatch_batch");
780        assert_eq!(chosen, DequeVariant::Khl);
781        // 64 items = 22 slots (ceil(64/3)).
782        let khl = dispatcher.khl().expect("khl");
783        let (_, tail, _) = khl.snapshot_size();
784        assert_eq!(tail, 22);
785        std::fs::remove_file(&khl_path).ok();
786    }
787
788    #[test]
789    fn dispatch_batch_falls_through_to_khpd_when_khl_unconfigured() {
790        // No KHL configured; KHPD next in fallback chain.
791        let khpd_path = tmp("fallback_khpd");
792        let dispatcher = DequeDispatcher::builder()
793            .with_khpd(&khpd_path, 64)
794            .expect("create khpd")
795            .build();
796        let items: Vec<LineItem> = (0..6u32).map(u32_item).collect();
797        let chosen = dispatcher
798            .dispatch_batch(WorkloadShape::producer_fast(6), &items)
799            .expect("dispatch_batch");
800        assert_eq!(chosen, DequeVariant::Khpd);
801        let khpd = dispatcher.khpd().expect("khpd");
802        let (_, tail, _, _) = khpd.snapshot_size();
803        assert_eq!(tail, 2);
804        std::fs::remove_file(&khpd_path).ok();
805    }
806
807    #[test]
808    fn dispatch_batch_falls_through_to_loh_when_khl_khpd_unconfigured() {
809        // No KHL, no KHPD configured; LOH next in fallback chain.
810        let loh_path = tmp("fallback_loh");
811        let dispatcher = DequeDispatcher::builder()
812            .with_loh(&loh_path, 512, usize::MAX)
813            .expect("create loh")
814            .build();
815        let items: Vec<LineItem> = (0..200u32).map(u32_item).collect();
816        let chosen = dispatcher
817            .dispatch_batch(WorkloadShape::producer_fast(200), &items)
818            .expect("dispatch_batch");
819        assert_eq!(chosen, DequeVariant::Loh);
820        let loh = dispatcher.loh().expect("loh");
821        let (_, tail, _, _) = loh.snapshot_size();
822        assert_eq!(tail, 200);
823        std::fs::remove_file(&loh_path).ok();
824    }
825
826    #[test]
827    fn dispatch_batch_routes_to_urd_for_multi_thief() {
828        let urd_path = tmp("dispatch_batch_urd");
829        let dispatcher = DequeDispatcher::builder()
830            .with_urd(&urd_path, 2)
831            .expect("create urd")
832            .build();
833        let items: Vec<LineItem> = (0..6u32).map(u32_item).collect();
834        let chosen = dispatcher
835            .dispatch_batch(WorkloadShape::fan_out(2, 6), &items)
836            .expect("dispatch_batch");
837        assert_eq!(chosen, DequeVariant::Urd);
838        std::fs::remove_file(&urd_path).ok();
839    }
840
841    #[test]
842    fn full_dispatcher_round_trips_mixed_shapes() {
843        // Full dispatcher with Chase-Lev + KHL configured.
844        // Per-item shape routes to Chase-Lev; batch shape routes to
845        // KHL. Drain both sides and verify bit-exact recovery.
846        let cl_path = tmp("full_cl");
847        let khl_path = tmp("full_khl");
848        let dispatcher = DequeDispatcher::builder()
849            .with_chase_lev(&cl_path, 128)
850            .expect("create cl")
851            .with_khl(&khl_path, 64)
852            .expect("create khl")
853            .build();
854
855        // 5 per-item dispatches -> Chase-Lev.
856        for i in 0..5u32 {
857            let v = dispatcher
858                .dispatch_one(WorkloadShape::request_reply(), u32_item(i))
859                .expect("dispatch_one");
860            assert_eq!(v, DequeVariant::ChaseLev);
861        }
862        // 12-item batch -> KHL.
863        let batch: Vec<LineItem> = (100..112u32).map(u32_item).collect();
864        let v = dispatcher
865            .dispatch_batch(WorkloadShape::producer_fast(12), &batch)
866            .expect("dispatch_batch");
867        assert_eq!(v, DequeVariant::Khl);
868
869        // Drain Chase-Lev.
870        let cl = dispatcher.chase_lev().expect("cl");
871        let mut seen = Vec::new();
872        while let Some(x) = cl.steal() {
873            seen.push(x);
874        }
875        assert_eq!(seen.len(), 5);
876        for (i, item) in seen.iter().enumerate() {
877            assert_eq!(*item, u32_item(i as u32));
878        }
879
880        // Drain KHL.
881        let khl = dispatcher.khl().expect("khl");
882        let mut drained = Vec::new();
883        loop {
884            match khl.steal_slot() {
885                crate::shared_deque_khl::Steal::Success(r) => {
886                    for i in 0..r.n_items {
887                        drained.push(r.items[i]);
888                    }
889                }
890                crate::shared_deque_khl::Steal::Empty => break,
891                crate::shared_deque_khl::Steal::Retry => continue,
892            }
893        }
894        assert_eq!(drained.len(), 12);
895        for (i, item) in drained.iter().enumerate() {
896            assert_eq!(*item, u32_item(100 + i as u32));
897        }
898
899        std::fs::remove_file(&cl_path).ok();
900        std::fs::remove_file(&khl_path).ok();
901    }
902}
subetha_cxc/dispatch_deque.rs

subetha_cxc/
dispatch_deque.rs
