Struct ConcurrentTimerWheel 

pub struct ConcurrentTimerWheel<V> { /* private fields */ }

Implementations

impl<V> ConcurrentTimerWheel<V>

pub fn new(num_slots: usize) -> Self

Examples found in repository

examples/perf_features.rs (line 370)

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

    // Diagnostic, not a feature: the base wheel's own tick. A single-level
    // wheel decrements the rounds counter of every entry in the bucket it
    // walks, fired or not, so its tick is O(resident/slots) - the cost the
    // hierarchical feature exists to remove. Printed so the feature curves
    // below have something to be read against.
    sweep("base/tick", |n| {
        drain(BATCH, base_wheel(n), |w| {
            let _ = w.tick();
        })
    });

    // ---------- hierarchical: cascade across three 64-slot wheels ----------
    #[cfg(feature = "hierarchical")]
    {
        use subms_timer_wheel::HierarchicalTimerWheel;

        fn hier(n: usize) -> HierarchicalTimerWheel<u32> {
            let mut w: HierarchicalTimerWheel<u32> = HierarchicalTimerWheel::new();
            load(&mut w, n, |w, d| {
                w.schedule(d as u64, 0);
            });
            w
        }

        // Swept on `tick`, the op the feature transforms. The cascade is the
        // expensive path and it fires on 1 tick in 64 (level 1) and 1 in 4096
        // (level 2), so the measured window has to be long enough to contain
        // both: 4096 timed ticks contains 64 level-1 cascades and one level-2.
        //
        // The curve is flat, and that is the correct reading rather than a
        // hidden cost: a cascade moves the entries in ONE coarse bucket, which
        // holds the timers due in the next 64 (level 1) or 4096 (level 2)
        // ticks. With the due rate held fixed that bucket's size is fixed too,
        // so resident timers further out cost the tick nothing. This is exactly
        // what the level structure buys - the base wheel's own tick, printed
        // above, walks resident/slots entries on EVERY tick.
        let sw = sweep("hierarchical/tick", |n| {
            drain(BATCH, hier(n), |w| {
                let _ = w.tick();
            })
        });

        // `cancel` is the O(resident) op the feature introduces. It has no
        // id->slot index (the base wheel's index would need patching on every
        // cascade) so it sweeps all 192 buckets and every entry in them.
        // Cancelling a MISS walks all of them and is non-destructive, which is
        // what makes it safe to repeat against one input.
        sweep("hierarchical/cancel-miss", |n| {
            bulk(hier(n), |w| {
                let _ = w.cancel(u64::MAX);
            })
        });

        // PINNED structural on the strength of `cancel`, not of the swept op.
        // From the source, `HierarchicalTimerWheel::cancel` iterates
        // LEVELS * SLOTS buckets and every entry in each until it matches, so
        // it is O(resident) - measured 29x over a 16x sweep, 1.0 ms p99 at
        // 524288 resident. The base wheel does not have that op shape: it keeps
        // an id->slot map and cancels in O(bucket). Classifying the feature
        // hot-path off a flat `tick` would tell a reader every op it introduces
        // is safe per-operation, and one of them lands on the millisecond line
        // at half a million timers.
        let (cat, reason) = classify_feature(
            &sw,
            Some(base_p50()),
            Some(subms::SubMsFeatureCategory::Structural),
        );

        let mut p99 = BTreeMap::new();
        p99.insert(
            "tick".to_string(),
            drain(1, hier(CANON), |w| {
                let _ = w.tick();
            })
            .p99,
        );
        p99.insert(
            "schedule".to_string(),
            {
                let mut scratch: HierarchicalTimerWheel<u32> = HierarchicalTimerWheel::new();
                let mut w = hier(CANON);
                keyed(
                    1,
                    |i| {
                        scratch.schedule(resident_delay(i) as u64, 0);
                    },
                    |i| {
                        w.schedule(resident_delay(i) as u64, 0);
                    },
                )
            }
            .p99,
        );
        p99.insert(
            "cancel".to_string(),
            bulk(hier(CANON), |w| {
                let _ = w.cancel(u64::MAX);
            })
            .p99,
        );
        manifest.set_feature("hierarchical", cat, &p99, &reason);
    }

    // ---------- concurrent: short-mutex wrapper ----------
    #[cfg(feature = "concurrent")]
    {
        use subms_timer_wheel::ConcurrentTimerWheel;

        fn conc(n: usize) -> ConcurrentTimerWheel<u32> {
            let mut w: ConcurrentTimerWheel<u32> = ConcurrentTimerWheel::new(SLOTS);
            load(&mut w, n, |w, d| {
                w.schedule(d, 0);
            });
            w
        }

        // Swept on `schedule` and measured single-threaded. The feature adds a
        // lock acquire and release to every op; running it contended would
        // measure the contention instead of the indirection, and the thread
        // count would then be a second thing varying across the sweep.
        let sw = sweep("concurrent/schedule", |n| {
            let scratch: ConcurrentTimerWheel<u32> = ConcurrentTimerWheel::new(SLOTS);
            let w = conc(n);
            keyed(
                BATCH,
                |i| {
                    scratch.schedule(resident_delay(i), 0);
                },
                |i| {
                    w.schedule(resident_delay(i), 0);
                },
            )
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50()), None);

        let mut p99 = BTreeMap::new();
        p99.insert(
            "schedule".to_string(),
            {
                let scratch: ConcurrentTimerWheel<u32> = ConcurrentTimerWheel::new(SLOTS);
                let w = conc(CANON);
                keyed(
                    1,
                    |i| {
                        scratch.schedule(resident_delay(i), 0);
                    },
                    |i| {
                        w.schedule(resident_delay(i), 0);
                    },
                )
            }
            .p99,
        );
        p99.insert(
            "tick".to_string(),
            drain(1, conc(CANON), |w| {
                let _ = w.tick();
            })
            .p99,
        );
        manifest.set_feature("concurrent", cat, &p99, &reason);
    }

    // ---------- deadline-scheduler: absolute deadlines over an injected clock ----------
    #[cfg(feature = "deadline-scheduler")]
    {
        use std::cell::Cell;
        use std::rc::Rc;
        use std::time::Duration;
        use subms_timer_wheel::{Clock, DeadlineScheduler};

        /// Time only moves when the bench moves it. A free-running clock makes
        /// `poll` tick however many ticks the host happened to take, which is
        /// neither repeatable nor comparable across sweep points; a frozen one
        /// makes `poll` a no-op and publishes an empty drain as the cost.
        struct StepClock {
            now: Cell<u64>,
            step: Cell<u64>,
        }
        struct Shared(Rc<StepClock>);
        impl Clock for Shared {
            fn now_nanos(&self) -> u64 {
                self.0.now.set(self.0.now.get() + self.0.step.get());
                self.0.now.get()
            }
        }

        fn sched(n: usize) -> (DeadlineScheduler<u32, Shared>, Rc<StepClock>) {
            let clock = Rc::new(StepClock {
                now: Cell::new(0),
                step: Cell::new(0),
            });
            let mut s: DeadlineScheduler<u32, Shared> = DeadlineScheduler::new(
                SLOTS,
                Shared(Rc::clone(&clock)),
                Duration::from_nanos(TICK_NS),
            );
            load(&mut s, n, |s, d| {
                s.schedule_at(d as u64 * TICK_NS, 0);
            });
            (s, clock)
        }

        // Swept on `poll`, the op the layer introduces. With the clock stepped
        // exactly one tick per call, a poll is one wheel tick plus the deadline
        // arithmetic, so the sweep reads the drain the layer is driving.
        let sw = sweep("deadline-scheduler/poll", |n| {
            let (s, clock) = sched(n);
            clock.step.set(TICK_NS);
            drain(BATCH, s, |s| {
                let _ = s.poll();
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50()), None);

        let mut p99 = BTreeMap::new();
        p99.insert(
            "schedule_at".to_string(),
            {
                let (mut scratch, _sc) = sched(0);
                let (mut s, _c) = sched(CANON);
                keyed(
                    1,
                    |i| {
                        scratch.schedule_at(resident_delay(i) as u64 * TICK_NS, 0);
                    },
                    |i| {
                        s.schedule_at(resident_delay(i) as u64 * TICK_NS, 0);
                    },
                )
            }
            .p99,
        );
        p99.insert(
            "poll".to_string(),
            {
                let (s, clock) = sched(CANON);
                clock.step.set(TICK_NS);
                drain(1, s, |s| {
                    let _ = s.poll();
                })
            }
            .p99,
        );
        manifest.set_feature("deadline-scheduler", cat, &p99, &reason);
    }

    // ---------- cron: 5-field expression parser + next-fire search ----------
    #[cfg(feature = "cron")]
    {
        use subms_timer_wheel::{CronSchedule, CronScheduler};
        const EXPR: &str = "*/5 * * * *";
        const EPOCH0: u64 = 1_704_067_200;

        // Swept on `next_fire`, the op the feature introduces. It searches
        // forward minute by minute from a rolling epoch and never touches a
        // wheel, so it is expected to read FLAT against resident timers - that
        // is the correct result for this feature, not a broken sweep.
        let sw = sweep("cron/next_fire", |_n| {
            let mut warm =
                CronScheduler::new(CronSchedule::parse(EXPR).expect("valid cron expr"), EPOCH0);
            let mut warm_epoch = EPOCH0;
            let mut cs =
                CronScheduler::new(CronSchedule::parse(EXPR).expect("valid cron expr"), EPOCH0);
            let mut epoch = EPOCH0;
            keyed(
                BATCH,
                |_| {
                    if let Some(n) = warm.next_fire(warm_epoch) {
                        warm.record_fire(n);
                        warm_epoch = n;
                    }
                },
                |_| {
                    let next = cs.next_fire(epoch);
                    if let Some(n) = next {
                        cs.record_fire(n);
                        epoch = n;
                    }
                },
            )
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50()), None);

        let mut p99 = BTreeMap::new();
        p99.insert(
            "parse".to_string(),
            keyed(
                1,
                |_| {
                    let _ = CronSchedule::parse(EXPR);
                },
                |_| {
                    let _ = CronSchedule::parse(EXPR);
                },
            )
            .p99,
        );
        p99.insert(
            "next_fire".to_string(),
            {
                let mut cs =
                    CronScheduler::new(CronSchedule::parse(EXPR).expect("valid cron expr"), EPOCH0);
                let mut epoch = EPOCH0;
                let mut warm =
                    CronScheduler::new(CronSchedule::parse(EXPR).expect("valid cron expr"), EPOCH0);
                let mut warm_epoch = EPOCH0;
                keyed(
                    1,
                    |_| {
                        if let Some(n) = warm.next_fire(warm_epoch) {
                            warm.record_fire(n);
                            warm_epoch = n;
                        }
                    },
                    |_| {
                        let next = cs.next_fire(epoch);
                        if let Some(n) = next {
                            cs.record_fire(n);
                            epoch = n;
                        }
                    },
                )
            }
            .p99,
        );
        manifest.set_feature("cron", cat, &p99, &reason);
    }

    // ---------- metrics: per-instance counters ----------
    #[cfg(feature = "metrics")]
    {
        use subms_timer_wheel::MeteredTimerWheel;

        fn metered(n: usize) -> MeteredTimerWheel<u32> {
            let mut w: MeteredTimerWheel<u32> = MeteredTimerWheel::new(SLOTS);
            load(&mut w, n, |w, d| {
                w.schedule(d, 0);
            });
            w
        }

        // Swept on `schedule`. The counters are the feature and they sit on the
        // per-op path; sweeping `tick` instead would measure the base wheel's
        // bucket walk and attribute it to a pair of u64 increments. The tick
        // number is still recorded below so it is visible.
        let sw = sweep("metrics/schedule", |n| {
            let mut scratch: MeteredTimerWheel<u32> = MeteredTimerWheel::new(SLOTS);
            let mut w = metered(n);
            keyed(
                BATCH,
                |i| {
                    scratch.schedule(resident_delay(i), 0);
                },
                |i| {
                    w.schedule(resident_delay(i), 0);
                },
            )
        });
        // PINNED auxiliary. From the source, `MeteredTimerWheel::schedule` is
        // one non-atomic increment of an owned u64 field followed by the base
        // call - no allocation, no branch, no lock. That is well under a
        // nanosecond against a ~55 ns schedule, and nothing on this host
        // resolves half a percent: the base op's own p50 spreads 3300-4400 ns
        // per 64-op sample across runs, and the feature crossed the classifier's
        // 10% band in both directions on four consecutive runs of unchanged
        // code. Pinning states that a human read the source instead of
        // publishing a coin toss as a measurement.
        let (cat, reason) = classify_feature(
            &sw,
            Some(base_p50()),
            Some(subms::SubMsFeatureCategory::Auxiliary),
        );

        let mut p99 = BTreeMap::new();
        p99.insert(
            "schedule".to_string(),
            {
                let mut scratch: MeteredTimerWheel<u32> = MeteredTimerWheel::new(SLOTS);
                let mut w = metered(CANON);
                keyed(
                    1,
                    |i| {
                        scratch.schedule(resident_delay(i), 0);
                    },
                    |i| {
                        w.schedule(resident_delay(i), 0);
                    },
                )
            }
            .p99,
        );
        p99.insert(
            "tick".to_string(),
            drain(1, metered(CANON), |w| {
                let _ = w.tick();
            })
            .p99,
        );
        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(())
}

examples/sample_app.rs (line 171)

fn concurrent_quote_timeouts() {
    use std::thread;
    use subms_timer_wheel::ConcurrentTimerWheel;
    println!("\n== concurrent: quote timeouts from many feeds ==");
    let wheel: ConcurrentTimerWheel<usize> = ConcurrentTimerWheel::new(256);

    let feeds = 4;
    let per_feed = 50;
    let mut handles = Vec::new();
    for feed in 0..feeds {
        let wheel = wheel.clone();
        handles.push(thread::spawn(move || {
            for i in 0..per_feed {
                wheel.schedule(1 + (i % 8), feed * 1000 + i);
            }
        }));
    }
    for h in handles {
        h.join().unwrap();
    }
    println!(
        "  {} quote timeouts armed across {feeds} feeds",
        wheel.pending()
    );

    let fired = wheel.advance(16).len();
    println!("  {feeds} feeds x {per_feed} quotes -> {fired} timeouts fired");
    assert_eq!(
        fired,
        feeds * per_feed,
        "every armed timeout fired exactly once"
    );
    assert!(wheel.is_empty());
}

pub fn num_slots(&self) -> usize

pub fn max_delay(&self) -> u64

pub fn pending(&self) -> usize

Examples found in repository

examples/sample_app.rs (line 189)

fn concurrent_quote_timeouts() {
    use std::thread;
    use subms_timer_wheel::ConcurrentTimerWheel;
    println!("\n== concurrent: quote timeouts from many feeds ==");
    let wheel: ConcurrentTimerWheel<usize> = ConcurrentTimerWheel::new(256);

    let feeds = 4;
    let per_feed = 50;
    let mut handles = Vec::new();
    for feed in 0..feeds {
        let wheel = wheel.clone();
        handles.push(thread::spawn(move || {
            for i in 0..per_feed {
                wheel.schedule(1 + (i % 8), feed * 1000 + i);
            }
        }));
    }
    for h in handles {
        h.join().unwrap();
    }
    println!(
        "  {} quote timeouts armed across {feeds} feeds",
        wheel.pending()
    );

    let fired = wheel.advance(16).len();
    println!("  {feeds} feeds x {per_feed} quotes -> {fired} timeouts fired");
    assert_eq!(
        fired,
        feeds * per_feed,
        "every armed timeout fired exactly once"
    );
    assert!(wheel.is_empty());
}

pub fn is_empty(&self) -> bool

Examples found in repository

examples/sample_app.rs (line 199)

fn concurrent_quote_timeouts() {
    use std::thread;
    use subms_timer_wheel::ConcurrentTimerWheel;
    println!("\n== concurrent: quote timeouts from many feeds ==");
    let wheel: ConcurrentTimerWheel<usize> = ConcurrentTimerWheel::new(256);

    let feeds = 4;
    let per_feed = 50;
    let mut handles = Vec::new();
    for feed in 0..feeds {
        let wheel = wheel.clone();
        handles.push(thread::spawn(move || {
            for i in 0..per_feed {
                wheel.schedule(1 + (i % 8), feed * 1000 + i);
            }
        }));
    }
    for h in handles {
        h.join().unwrap();
    }
    println!(
        "  {} quote timeouts armed across {feeds} feeds",
        wheel.pending()
    );

    let fired = wheel.advance(16).len();
    println!("  {feeds} feeds x {per_feed} quotes -> {fired} timeouts fired");
    assert_eq!(
        fired,
        feeds * per_feed,
        "every armed timeout fired exactly once"
    );
    assert!(wheel.is_empty());
}

pub fn slot_len(&self, slot: usize) -> usize

pub fn schedule(&self, delay_ticks: usize, value: V) -> u64

Examples found in repository

examples/perf_features.rs (line 372)

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

    // Diagnostic, not a feature: the base wheel's own tick. A single-level
    // wheel decrements the rounds counter of every entry in the bucket it
    // walks, fired or not, so its tick is O(resident/slots) - the cost the
    // hierarchical feature exists to remove. Printed so the feature curves
    // below have something to be read against.
    sweep("base/tick", |n| {
        drain(BATCH, base_wheel(n), |w| {
            let _ = w.tick();
        })
    });

    // ---------- hierarchical: cascade across three 64-slot wheels ----------
    #[cfg(feature = "hierarchical")]
    {
        use subms_timer_wheel::HierarchicalTimerWheel;

        fn hier(n: usize) -> HierarchicalTimerWheel<u32> {
            let mut w: HierarchicalTimerWheel<u32> = HierarchicalTimerWheel::new();
            load(&mut w, n, |w, d| {
                w.schedule(d as u64, 0);
            });
            w
        }

        // Swept on `tick`, the op the feature transforms. The cascade is the
        // expensive path and it fires on 1 tick in 64 (level 1) and 1 in 4096
        // (level 2), so the measured window has to be long enough to contain
        // both: 4096 timed ticks contains 64 level-1 cascades and one level-2.
        //
        // The curve is flat, and that is the correct reading rather than a
        // hidden cost: a cascade moves the entries in ONE coarse bucket, which
        // holds the timers due in the next 64 (level 1) or 4096 (level 2)
        // ticks. With the due rate held fixed that bucket's size is fixed too,
        // so resident timers further out cost the tick nothing. This is exactly
        // what the level structure buys - the base wheel's own tick, printed
        // above, walks resident/slots entries on EVERY tick.
        let sw = sweep("hierarchical/tick", |n| {
            drain(BATCH, hier(n), |w| {
                let _ = w.tick();
            })
        });

        // `cancel` is the O(resident) op the feature introduces. It has no
        // id->slot index (the base wheel's index would need patching on every
        // cascade) so it sweeps all 192 buckets and every entry in them.
        // Cancelling a MISS walks all of them and is non-destructive, which is
        // what makes it safe to repeat against one input.
        sweep("hierarchical/cancel-miss", |n| {
            bulk(hier(n), |w| {
                let _ = w.cancel(u64::MAX);
            })
        });

        // PINNED structural on the strength of `cancel`, not of the swept op.
        // From the source, `HierarchicalTimerWheel::cancel` iterates
        // LEVELS * SLOTS buckets and every entry in each until it matches, so
        // it is O(resident) - measured 29x over a 16x sweep, 1.0 ms p99 at
        // 524288 resident. The base wheel does not have that op shape: it keeps
        // an id->slot map and cancels in O(bucket). Classifying the feature
        // hot-path off a flat `tick` would tell a reader every op it introduces
        // is safe per-operation, and one of them lands on the millisecond line
        // at half a million timers.
        let (cat, reason) = classify_feature(
            &sw,
            Some(base_p50()),
            Some(subms::SubMsFeatureCategory::Structural),
        );

        let mut p99 = BTreeMap::new();
        p99.insert(
            "tick".to_string(),
            drain(1, hier(CANON), |w| {
                let _ = w.tick();
            })
            .p99,
        );
        p99.insert(
            "schedule".to_string(),
            {
                let mut scratch: HierarchicalTimerWheel<u32> = HierarchicalTimerWheel::new();
                let mut w = hier(CANON);
                keyed(
                    1,
                    |i| {
                        scratch.schedule(resident_delay(i) as u64, 0);
                    },
                    |i| {
                        w.schedule(resident_delay(i) as u64, 0);
                    },
                )
            }
            .p99,
        );
        p99.insert(
            "cancel".to_string(),
            bulk(hier(CANON), |w| {
                let _ = w.cancel(u64::MAX);
            })
            .p99,
        );
        manifest.set_feature("hierarchical", cat, &p99, &reason);
    }

    // ---------- concurrent: short-mutex wrapper ----------
    #[cfg(feature = "concurrent")]
    {
        use subms_timer_wheel::ConcurrentTimerWheel;

        fn conc(n: usize) -> ConcurrentTimerWheel<u32> {
            let mut w: ConcurrentTimerWheel<u32> = ConcurrentTimerWheel::new(SLOTS);
            load(&mut w, n, |w, d| {
                w.schedule(d, 0);
            });
            w
        }

        // Swept on `schedule` and measured single-threaded. The feature adds a
        // lock acquire and release to every op; running it contended would
        // measure the contention instead of the indirection, and the thread
        // count would then be a second thing varying across the sweep.
        let sw = sweep("concurrent/schedule", |n| {
            let scratch: ConcurrentTimerWheel<u32> = ConcurrentTimerWheel::new(SLOTS);
            let w = conc(n);
            keyed(
                BATCH,
                |i| {
                    scratch.schedule(resident_delay(i), 0);
                },
                |i| {
                    w.schedule(resident_delay(i), 0);
                },
            )
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50()), None);

        let mut p99 = BTreeMap::new();
        p99.insert(
            "schedule".to_string(),
            {
                let scratch: ConcurrentTimerWheel<u32> = ConcurrentTimerWheel::new(SLOTS);
                let w = conc(CANON);
                keyed(
                    1,
                    |i| {
                        scratch.schedule(resident_delay(i), 0);
                    },
                    |i| {
                        w.schedule(resident_delay(i), 0);
                    },
                )
            }
            .p99,
        );
        p99.insert(
            "tick".to_string(),
            drain(1, conc(CANON), |w| {
                let _ = w.tick();
            })
            .p99,
        );
        manifest.set_feature("concurrent", cat, &p99, &reason);
    }

    // ---------- deadline-scheduler: absolute deadlines over an injected clock ----------
    #[cfg(feature = "deadline-scheduler")]
    {
        use std::cell::Cell;
        use std::rc::Rc;
        use std::time::Duration;
        use subms_timer_wheel::{Clock, DeadlineScheduler};

        /// Time only moves when the bench moves it. A free-running clock makes
        /// `poll` tick however many ticks the host happened to take, which is
        /// neither repeatable nor comparable across sweep points; a frozen one
        /// makes `poll` a no-op and publishes an empty drain as the cost.
        struct StepClock {
            now: Cell<u64>,
            step: Cell<u64>,
        }
        struct Shared(Rc<StepClock>);
        impl Clock for Shared {
            fn now_nanos(&self) -> u64 {
                self.0.now.set(self.0.now.get() + self.0.step.get());
                self.0.now.get()
            }
        }

        fn sched(n: usize) -> (DeadlineScheduler<u32, Shared>, Rc<StepClock>) {
            let clock = Rc::new(StepClock {
                now: Cell::new(0),
                step: Cell::new(0),
            });
            let mut s: DeadlineScheduler<u32, Shared> = DeadlineScheduler::new(
                SLOTS,
                Shared(Rc::clone(&clock)),
                Duration::from_nanos(TICK_NS),
            );
            load(&mut s, n, |s, d| {
                s.schedule_at(d as u64 * TICK_NS, 0);
            });
            (s, clock)
        }

        // Swept on `poll`, the op the layer introduces. With the clock stepped
        // exactly one tick per call, a poll is one wheel tick plus the deadline
        // arithmetic, so the sweep reads the drain the layer is driving.
        let sw = sweep("deadline-scheduler/poll", |n| {
            let (s, clock) = sched(n);
            clock.step.set(TICK_NS);
            drain(BATCH, s, |s| {
                let _ = s.poll();
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50()), None);

        let mut p99 = BTreeMap::new();
        p99.insert(
            "schedule_at".to_string(),
            {
                let (mut scratch, _sc) = sched(0);
                let (mut s, _c) = sched(CANON);
                keyed(
                    1,
                    |i| {
                        scratch.schedule_at(resident_delay(i) as u64 * TICK_NS, 0);
                    },
                    |i| {
                        s.schedule_at(resident_delay(i) as u64 * TICK_NS, 0);
                    },
                )
            }
            .p99,
        );
        p99.insert(
            "poll".to_string(),
            {
                let (s, clock) = sched(CANON);
                clock.step.set(TICK_NS);
                drain(1, s, |s| {
                    let _ = s.poll();
                })
            }
            .p99,
        );
        manifest.set_feature("deadline-scheduler", cat, &p99, &reason);
    }

    // ---------- cron: 5-field expression parser + next-fire search ----------
    #[cfg(feature = "cron")]
    {
        use subms_timer_wheel::{CronSchedule, CronScheduler};
        const EXPR: &str = "*/5 * * * *";
        const EPOCH0: u64 = 1_704_067_200;

        // Swept on `next_fire`, the op the feature introduces. It searches
        // forward minute by minute from a rolling epoch and never touches a
        // wheel, so it is expected to read FLAT against resident timers - that
        // is the correct result for this feature, not a broken sweep.
        let sw = sweep("cron/next_fire", |_n| {
            let mut warm =
                CronScheduler::new(CronSchedule::parse(EXPR).expect("valid cron expr"), EPOCH0);
            let mut warm_epoch = EPOCH0;
            let mut cs =
                CronScheduler::new(CronSchedule::parse(EXPR).expect("valid cron expr"), EPOCH0);
            let mut epoch = EPOCH0;
            keyed(
                BATCH,
                |_| {
                    if let Some(n) = warm.next_fire(warm_epoch) {
                        warm.record_fire(n);
                        warm_epoch = n;
                    }
                },
                |_| {
                    let next = cs.next_fire(epoch);
                    if let Some(n) = next {
                        cs.record_fire(n);
                        epoch = n;
                    }
                },
            )
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50()), None);

        let mut p99 = BTreeMap::new();
        p99.insert(
            "parse".to_string(),
            keyed(
                1,
                |_| {
                    let _ = CronSchedule::parse(EXPR);
                },
                |_| {
                    let _ = CronSchedule::parse(EXPR);
                },
            )
            .p99,
        );
        p99.insert(
            "next_fire".to_string(),
            {
                let mut cs =
                    CronScheduler::new(CronSchedule::parse(EXPR).expect("valid cron expr"), EPOCH0);
                let mut epoch = EPOCH0;
                let mut warm =
                    CronScheduler::new(CronSchedule::parse(EXPR).expect("valid cron expr"), EPOCH0);
                let mut warm_epoch = EPOCH0;
                keyed(
                    1,
                    |_| {
                        if let Some(n) = warm.next_fire(warm_epoch) {
                            warm.record_fire(n);
                            warm_epoch = n;
                        }
                    },
                    |_| {
                        let next = cs.next_fire(epoch);
                        if let Some(n) = next {
                            cs.record_fire(n);
                            epoch = n;
                        }
                    },
                )
            }
            .p99,
        );
        manifest.set_feature("cron", cat, &p99, &reason);
    }

    // ---------- metrics: per-instance counters ----------
    #[cfg(feature = "metrics")]
    {
        use subms_timer_wheel::MeteredTimerWheel;

        fn metered(n: usize) -> MeteredTimerWheel<u32> {
            let mut w: MeteredTimerWheel<u32> = MeteredTimerWheel::new(SLOTS);
            load(&mut w, n, |w, d| {
                w.schedule(d, 0);
            });
            w
        }

        // Swept on `schedule`. The counters are the feature and they sit on the
        // per-op path; sweeping `tick` instead would measure the base wheel's
        // bucket walk and attribute it to a pair of u64 increments. The tick
        // number is still recorded below so it is visible.
        let sw = sweep("metrics/schedule", |n| {
            let mut scratch: MeteredTimerWheel<u32> = MeteredTimerWheel::new(SLOTS);
            let mut w = metered(n);
            keyed(
                BATCH,
                |i| {
                    scratch.schedule(resident_delay(i), 0);
                },
                |i| {
                    w.schedule(resident_delay(i), 0);
                },
            )
        });
        // PINNED auxiliary. From the source, `MeteredTimerWheel::schedule` is
        // one non-atomic increment of an owned u64 field followed by the base
        // call - no allocation, no branch, no lock. That is well under a
        // nanosecond against a ~55 ns schedule, and nothing on this host
        // resolves half a percent: the base op's own p50 spreads 3300-4400 ns
        // per 64-op sample across runs, and the feature crossed the classifier's
        // 10% band in both directions on four consecutive runs of unchanged
        // code. Pinning states that a human read the source instead of
        // publishing a coin toss as a measurement.
        let (cat, reason) = classify_feature(
            &sw,
            Some(base_p50()),
            Some(subms::SubMsFeatureCategory::Auxiliary),
        );

        let mut p99 = BTreeMap::new();
        p99.insert(
            "schedule".to_string(),
            {
                let mut scratch: MeteredTimerWheel<u32> = MeteredTimerWheel::new(SLOTS);
                let mut w = metered(CANON);
                keyed(
                    1,
                    |i| {
                        scratch.schedule(resident_delay(i), 0);
                    },
                    |i| {
                        w.schedule(resident_delay(i), 0);
                    },
                )
            }
            .p99,
        );
        p99.insert(
            "tick".to_string(),
            drain(1, metered(CANON), |w| {
                let _ = w.tick();
            })
            .p99,
        );
        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(())
}

examples/sample_app.rs (line 180)

fn concurrent_quote_timeouts() {
    use std::thread;
    use subms_timer_wheel::ConcurrentTimerWheel;
    println!("\n== concurrent: quote timeouts from many feeds ==");
    let wheel: ConcurrentTimerWheel<usize> = ConcurrentTimerWheel::new(256);

    let feeds = 4;
    let per_feed = 50;
    let mut handles = Vec::new();
    for feed in 0..feeds {
        let wheel = wheel.clone();
        handles.push(thread::spawn(move || {
            for i in 0..per_feed {
                wheel.schedule(1 + (i % 8), feed * 1000 + i);
            }
        }));
    }
    for h in handles {
        h.join().unwrap();
    }
    println!(
        "  {} quote timeouts armed across {feeds} feeds",
        wheel.pending()
    );

    let fired = wheel.advance(16).len();
    println!("  {feeds} feeds x {per_feed} quotes -> {fired} timeouts fired");
    assert_eq!(
        fired,
        feeds * per_feed,
        "every armed timeout fired exactly once"
    );
    assert!(wheel.is_empty());
}

pub fn try_schedule( &self, delay_ticks: usize, value: V, ) -> Result<u64, TimerError>

pub fn cancel(&self, id: u64) -> bool

pub fn reschedule(&self, id: u64, delay_ticks: usize) -> bool

pub fn tick(&self) -> Vec<V>

Advance one tick. Returns the fired values; the mutex is released before the caller dispatches them.

Examples found in repository

examples/perf_features.rs (line 417)

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

    // Diagnostic, not a feature: the base wheel's own tick. A single-level
    // wheel decrements the rounds counter of every entry in the bucket it
    // walks, fired or not, so its tick is O(resident/slots) - the cost the
    // hierarchical feature exists to remove. Printed so the feature curves
    // below have something to be read against.
    sweep("base/tick", |n| {
        drain(BATCH, base_wheel(n), |w| {
            let _ = w.tick();
        })
    });

    // ---------- hierarchical: cascade across three 64-slot wheels ----------
    #[cfg(feature = "hierarchical")]
    {
        use subms_timer_wheel::HierarchicalTimerWheel;

        fn hier(n: usize) -> HierarchicalTimerWheel<u32> {
            let mut w: HierarchicalTimerWheel<u32> = HierarchicalTimerWheel::new();
            load(&mut w, n, |w, d| {
                w.schedule(d as u64, 0);
            });
            w
        }

        // Swept on `tick`, the op the feature transforms. The cascade is the
        // expensive path and it fires on 1 tick in 64 (level 1) and 1 in 4096
        // (level 2), so the measured window has to be long enough to contain
        // both: 4096 timed ticks contains 64 level-1 cascades and one level-2.
        //
        // The curve is flat, and that is the correct reading rather than a
        // hidden cost: a cascade moves the entries in ONE coarse bucket, which
        // holds the timers due in the next 64 (level 1) or 4096 (level 2)
        // ticks. With the due rate held fixed that bucket's size is fixed too,
        // so resident timers further out cost the tick nothing. This is exactly
        // what the level structure buys - the base wheel's own tick, printed
        // above, walks resident/slots entries on EVERY tick.
        let sw = sweep("hierarchical/tick", |n| {
            drain(BATCH, hier(n), |w| {
                let _ = w.tick();
            })
        });

        // `cancel` is the O(resident) op the feature introduces. It has no
        // id->slot index (the base wheel's index would need patching on every
        // cascade) so it sweeps all 192 buckets and every entry in them.
        // Cancelling a MISS walks all of them and is non-destructive, which is
        // what makes it safe to repeat against one input.
        sweep("hierarchical/cancel-miss", |n| {
            bulk(hier(n), |w| {
                let _ = w.cancel(u64::MAX);
            })
        });

        // PINNED structural on the strength of `cancel`, not of the swept op.
        // From the source, `HierarchicalTimerWheel::cancel` iterates
        // LEVELS * SLOTS buckets and every entry in each until it matches, so
        // it is O(resident) - measured 29x over a 16x sweep, 1.0 ms p99 at
        // 524288 resident. The base wheel does not have that op shape: it keeps
        // an id->slot map and cancels in O(bucket). Classifying the feature
        // hot-path off a flat `tick` would tell a reader every op it introduces
        // is safe per-operation, and one of them lands on the millisecond line
        // at half a million timers.
        let (cat, reason) = classify_feature(
            &sw,
            Some(base_p50()),
            Some(subms::SubMsFeatureCategory::Structural),
        );

        let mut p99 = BTreeMap::new();
        p99.insert(
            "tick".to_string(),
            drain(1, hier(CANON), |w| {
                let _ = w.tick();
            })
            .p99,
        );
        p99.insert(
            "schedule".to_string(),
            {
                let mut scratch: HierarchicalTimerWheel<u32> = HierarchicalTimerWheel::new();
                let mut w = hier(CANON);
                keyed(
                    1,
                    |i| {
                        scratch.schedule(resident_delay(i) as u64, 0);
                    },
                    |i| {
                        w.schedule(resident_delay(i) as u64, 0);
                    },
                )
            }
            .p99,
        );
        p99.insert(
            "cancel".to_string(),
            bulk(hier(CANON), |w| {
                let _ = w.cancel(u64::MAX);
            })
            .p99,
        );
        manifest.set_feature("hierarchical", cat, &p99, &reason);
    }

    // ---------- concurrent: short-mutex wrapper ----------
    #[cfg(feature = "concurrent")]
    {
        use subms_timer_wheel::ConcurrentTimerWheel;

        fn conc(n: usize) -> ConcurrentTimerWheel<u32> {
            let mut w: ConcurrentTimerWheel<u32> = ConcurrentTimerWheel::new(SLOTS);
            load(&mut w, n, |w, d| {
                w.schedule(d, 0);
            });
            w
        }

        // Swept on `schedule` and measured single-threaded. The feature adds a
        // lock acquire and release to every op; running it contended would
        // measure the contention instead of the indirection, and the thread
        // count would then be a second thing varying across the sweep.
        let sw = sweep("concurrent/schedule", |n| {
            let scratch: ConcurrentTimerWheel<u32> = ConcurrentTimerWheel::new(SLOTS);
            let w = conc(n);
            keyed(
                BATCH,
                |i| {
                    scratch.schedule(resident_delay(i), 0);
                },
                |i| {
                    w.schedule(resident_delay(i), 0);
                },
            )
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50()), None);

        let mut p99 = BTreeMap::new();
        p99.insert(
            "schedule".to_string(),
            {
                let scratch: ConcurrentTimerWheel<u32> = ConcurrentTimerWheel::new(SLOTS);
                let w = conc(CANON);
                keyed(
                    1,
                    |i| {
                        scratch.schedule(resident_delay(i), 0);
                    },
                    |i| {
                        w.schedule(resident_delay(i), 0);
                    },
                )
            }
            .p99,
        );
        p99.insert(
            "tick".to_string(),
            drain(1, conc(CANON), |w| {
                let _ = w.tick();
            })
            .p99,
        );
        manifest.set_feature("concurrent", cat, &p99, &reason);
    }

    // ---------- deadline-scheduler: absolute deadlines over an injected clock ----------
    #[cfg(feature = "deadline-scheduler")]
    {
        use std::cell::Cell;
        use std::rc::Rc;
        use std::time::Duration;
        use subms_timer_wheel::{Clock, DeadlineScheduler};

        /// Time only moves when the bench moves it. A free-running clock makes
        /// `poll` tick however many ticks the host happened to take, which is
        /// neither repeatable nor comparable across sweep points; a frozen one
        /// makes `poll` a no-op and publishes an empty drain as the cost.
        struct StepClock {
            now: Cell<u64>,
            step: Cell<u64>,
        }
        struct Shared(Rc<StepClock>);
        impl Clock for Shared {
            fn now_nanos(&self) -> u64 {
                self.0.now.set(self.0.now.get() + self.0.step.get());
                self.0.now.get()
            }
        }

        fn sched(n: usize) -> (DeadlineScheduler<u32, Shared>, Rc<StepClock>) {
            let clock = Rc::new(StepClock {
                now: Cell::new(0),
                step: Cell::new(0),
            });
            let mut s: DeadlineScheduler<u32, Shared> = DeadlineScheduler::new(
                SLOTS,
                Shared(Rc::clone(&clock)),
                Duration::from_nanos(TICK_NS),
            );
            load(&mut s, n, |s, d| {
                s.schedule_at(d as u64 * TICK_NS, 0);
            });
            (s, clock)
        }

        // Swept on `poll`, the op the layer introduces. With the clock stepped
        // exactly one tick per call, a poll is one wheel tick plus the deadline
        // arithmetic, so the sweep reads the drain the layer is driving.
        let sw = sweep("deadline-scheduler/poll", |n| {
            let (s, clock) = sched(n);
            clock.step.set(TICK_NS);
            drain(BATCH, s, |s| {
                let _ = s.poll();
            })
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50()), None);

        let mut p99 = BTreeMap::new();
        p99.insert(
            "schedule_at".to_string(),
            {
                let (mut scratch, _sc) = sched(0);
                let (mut s, _c) = sched(CANON);
                keyed(
                    1,
                    |i| {
                        scratch.schedule_at(resident_delay(i) as u64 * TICK_NS, 0);
                    },
                    |i| {
                        s.schedule_at(resident_delay(i) as u64 * TICK_NS, 0);
                    },
                )
            }
            .p99,
        );
        p99.insert(
            "poll".to_string(),
            {
                let (s, clock) = sched(CANON);
                clock.step.set(TICK_NS);
                drain(1, s, |s| {
                    let _ = s.poll();
                })
            }
            .p99,
        );
        manifest.set_feature("deadline-scheduler", cat, &p99, &reason);
    }

    // ---------- cron: 5-field expression parser + next-fire search ----------
    #[cfg(feature = "cron")]
    {
        use subms_timer_wheel::{CronSchedule, CronScheduler};
        const EXPR: &str = "*/5 * * * *";
        const EPOCH0: u64 = 1_704_067_200;

        // Swept on `next_fire`, the op the feature introduces. It searches
        // forward minute by minute from a rolling epoch and never touches a
        // wheel, so it is expected to read FLAT against resident timers - that
        // is the correct result for this feature, not a broken sweep.
        let sw = sweep("cron/next_fire", |_n| {
            let mut warm =
                CronScheduler::new(CronSchedule::parse(EXPR).expect("valid cron expr"), EPOCH0);
            let mut warm_epoch = EPOCH0;
            let mut cs =
                CronScheduler::new(CronSchedule::parse(EXPR).expect("valid cron expr"), EPOCH0);
            let mut epoch = EPOCH0;
            keyed(
                BATCH,
                |_| {
                    if let Some(n) = warm.next_fire(warm_epoch) {
                        warm.record_fire(n);
                        warm_epoch = n;
                    }
                },
                |_| {
                    let next = cs.next_fire(epoch);
                    if let Some(n) = next {
                        cs.record_fire(n);
                        epoch = n;
                    }
                },
            )
        });
        let (cat, reason) = classify_feature(&sw, Some(base_p50()), None);

        let mut p99 = BTreeMap::new();
        p99.insert(
            "parse".to_string(),
            keyed(
                1,
                |_| {
                    let _ = CronSchedule::parse(EXPR);
                },
                |_| {
                    let _ = CronSchedule::parse(EXPR);
                },
            )
            .p99,
        );
        p99.insert(
            "next_fire".to_string(),
            {
                let mut cs =
                    CronScheduler::new(CronSchedule::parse(EXPR).expect("valid cron expr"), EPOCH0);
                let mut epoch = EPOCH0;
                let mut warm =
                    CronScheduler::new(CronSchedule::parse(EXPR).expect("valid cron expr"), EPOCH0);
                let mut warm_epoch = EPOCH0;
                keyed(
                    1,
                    |_| {
                        if let Some(n) = warm.next_fire(warm_epoch) {
                            warm.record_fire(n);
                            warm_epoch = n;
                        }
                    },
                    |_| {
                        let next = cs.next_fire(epoch);
                        if let Some(n) = next {
                            cs.record_fire(n);
                            epoch = n;
                        }
                    },
                )
            }
            .p99,
        );
        manifest.set_feature("cron", cat, &p99, &reason);
    }

    // ---------- metrics: per-instance counters ----------
    #[cfg(feature = "metrics")]
    {
        use subms_timer_wheel::MeteredTimerWheel;

        fn metered(n: usize) -> MeteredTimerWheel<u32> {
            let mut w: MeteredTimerWheel<u32> = MeteredTimerWheel::new(SLOTS);
            load(&mut w, n, |w, d| {
                w.schedule(d, 0);
            });
            w
        }

        // Swept on `schedule`. The counters are the feature and they sit on the
        // per-op path; sweeping `tick` instead would measure the base wheel's
        // bucket walk and attribute it to a pair of u64 increments. The tick
        // number is still recorded below so it is visible.
        let sw = sweep("metrics/schedule", |n| {
            let mut scratch: MeteredTimerWheel<u32> = MeteredTimerWheel::new(SLOTS);
            let mut w = metered(n);
            keyed(
                BATCH,
                |i| {
                    scratch.schedule(resident_delay(i), 0);
                },
                |i| {
                    w.schedule(resident_delay(i), 0);
                },
            )
        });
        // PINNED auxiliary. From the source, `MeteredTimerWheel::schedule` is
        // one non-atomic increment of an owned u64 field followed by the base
        // call - no allocation, no branch, no lock. That is well under a
        // nanosecond against a ~55 ns schedule, and nothing on this host
        // resolves half a percent: the base op's own p50 spreads 3300-4400 ns
        // per 64-op sample across runs, and the feature crossed the classifier's
        // 10% band in both directions on four consecutive runs of unchanged
        // code. Pinning states that a human read the source instead of
        // publishing a coin toss as a measurement.
        let (cat, reason) = classify_feature(
            &sw,
            Some(base_p50()),
            Some(subms::SubMsFeatureCategory::Auxiliary),
        );

        let mut p99 = BTreeMap::new();
        p99.insert(
            "schedule".to_string(),
            {
                let mut scratch: MeteredTimerWheel<u32> = MeteredTimerWheel::new(SLOTS);
                let mut w = metered(CANON);
                keyed(
                    1,
                    |i| {
                        scratch.schedule(resident_delay(i), 0);
                    },
                    |i| {
                        w.schedule(resident_delay(i), 0);
                    },
                )
            }
            .p99,
        );
        p99.insert(
            "tick".to_string(),
            drain(1, metered(CANON), |w| {
                let _ = w.tick();
            })
            .p99,
        );
        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(())
}

pub fn advance(&self, ticks: usize) -> Vec<V>

Examples found in repository

examples/sample_app.rs (line 192)

fn concurrent_quote_timeouts() {
    use std::thread;
    use subms_timer_wheel::ConcurrentTimerWheel;
    println!("\n== concurrent: quote timeouts from many feeds ==");
    let wheel: ConcurrentTimerWheel<usize> = ConcurrentTimerWheel::new(256);

    let feeds = 4;
    let per_feed = 50;
    let mut handles = Vec::new();
    for feed in 0..feeds {
        let wheel = wheel.clone();
        handles.push(thread::spawn(move || {
            for i in 0..per_feed {
                wheel.schedule(1 + (i % 8), feed * 1000 + i);
            }
        }));
    }
    for h in handles {
        h.join().unwrap();
    }
    println!(
        "  {} quote timeouts armed across {feeds} feeds",
        wheel.pending()
    );

    let fired = wheel.advance(16).len();
    println!("  {feeds} feeds x {per_feed} quotes -> {fired} timeouts fired");
    assert_eq!(
        fired,
        feeds * per_feed,
        "every armed timeout fired exactly once"
    );
    assert!(wheel.is_empty());
}

pub fn drain(&self) -> Vec<V>

pub fn clear(&self)

Trait Implementations

impl<V> Clone for ConcurrentTimerWheel<V>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.