Struct CronScheduler 

pub struct CronScheduler { /* private fields */ }

Recurring scheduler glued to a base TimerWheel. After each fire the scheduler computes the next deadline from the cron schedule and re-arms.

Implementations

impl CronScheduler

pub fn new(schedule: CronSchedule, now_epoch: u64) -> Self

Examples found in repository

examples/sample_app.rs (line 248)

fn cron_risk_snapshot() {
    use subms_timer_wheel::{CronSchedule, CronScheduler};
    println!("\n== cron: mark-to-market every 5 minutes ==");
    let schedule = CronSchedule::parse("*/5 * * * *").expect("valid cron");

    // 2024-01-01 00:00:01 UTC. Next */5 boundary is 00:05:00.
    let start = 1_704_067_201;
    let mut scheduler = CronScheduler::new(schedule, start);

    let first = scheduler.next_fire(start).expect("a next fire exists");
    scheduler.record_fire(first);
    let second = scheduler.next_fire(first).expect("a next fire exists");
    println!("  first snapshot at epoch {first}, next at {second}");

    assert_eq!(
        first, 1_704_067_500,
        "first fire lands on the 5-minute grid"
    );
    assert_eq!(second, first + 300, "re-arms exactly 5 minutes later");
}

examples/perf_features.rs (line 521)

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 schedule(&self) -> &CronSchedule

pub fn next_fire(&self, now_epoch: u64) -> Option<u64>

Epoch-second the schedule will next fire, given the current epoch second. Returns None if no firing within the schedule’s look-ahead horizon.

Examples found in repository

examples/sample_app.rs (line 250)

fn cron_risk_snapshot() {
    use subms_timer_wheel::{CronSchedule, CronScheduler};
    println!("\n== cron: mark-to-market every 5 minutes ==");
    let schedule = CronSchedule::parse("*/5 * * * *").expect("valid cron");

    // 2024-01-01 00:00:01 UTC. Next */5 boundary is 00:05:00.
    let start = 1_704_067_201;
    let mut scheduler = CronScheduler::new(schedule, start);

    let first = scheduler.next_fire(start).expect("a next fire exists");
    scheduler.record_fire(first);
    let second = scheduler.next_fire(first).expect("a next fire exists");
    println!("  first snapshot at epoch {first}, next at {second}");

    assert_eq!(
        first, 1_704_067_500,
        "first fire lands on the 5-minute grid"
    );
    assert_eq!(second, first + 300, "re-arms exactly 5 minutes later");
}

examples/perf_features.rs (line 529)

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 record_fire(&mut self, epoch: u64)

Mark a fire at epoch consumed; next call to next_fire will look past it.

Examples found in repository

examples/sample_app.rs (line 251)

fn cron_risk_snapshot() {
    use subms_timer_wheel::{CronSchedule, CronScheduler};
    println!("\n== cron: mark-to-market every 5 minutes ==");
    let schedule = CronSchedule::parse("*/5 * * * *").expect("valid cron");

    // 2024-01-01 00:00:01 UTC. Next */5 boundary is 00:05:00.
    let start = 1_704_067_201;
    let mut scheduler = CronScheduler::new(schedule, start);

    let first = scheduler.next_fire(start).expect("a next fire exists");
    scheduler.record_fire(first);
    let second = scheduler.next_fire(first).expect("a next fire exists");
    println!("  first snapshot at epoch {first}, next at {second}");

    assert_eq!(
        first, 1_704_067_500,
        "first fire lands on the 5-minute grid"
    );
    assert_eq!(second, first + 300, "re-arms exactly 5 minutes later");
}

examples/perf_features.rs (line 530)

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