Struct TimerWheel 

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

Implementations

impl<V> TimerWheel<V>

pub fn new(num_slots: usize) -> Self

num_slots rounded up to a power of two.

Examples found in repository

examples/perf_features.rs (line 216)

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

/// The baseline: base `schedule`, the O(1) per-op write every feature either
/// decorates or replaces. Re-measured immediately before EACH feature is
/// classified rather than once at the top. Measured once, it sits several
/// half-million-timer builds away from the feature it is compared against, and
/// on this host that gap moves it between 3000 and 4300 ns run to run - as large
/// as a real feature delta. `metrics`, whose entire cost is one u64 increment,
/// flipped between auxiliary and hot-path on that drift alone; measured
/// adjacent, both runs land on auxiliary.
fn base_p50() -> u64 {
    let mut scratch: TimerWheel<u32> = TimerWheel::new(SLOTS);
    let mut w = base_wheel(CANON);
    let m = keyed(
        BATCH,
        |i| {
            scratch.schedule(resident_delay(i), 0);
        },
        |i| {
            w.schedule(resident_delay(i), 0);
        },
    );
    eprintln!("base schedule: p50 {} p99 {} max {}", m.p50, m.p99, m.max);
    m.p50
}

examples/growth_main.rs (line 103)

fn main() -> ExitCode {
    let mut raw = String::new();
    if io::stdin().read_to_string(&mut raw).is_err() {
        eprintln!("growth_main: failed to read stdin");
        return ExitCode::FAILURE;
    }
    let mut map = BTreeMap::new();
    for line in raw.lines() {
        let line = line.trim();
        if line.is_empty() || line.starts_with('#') {
            continue;
        }
        if let Some((k, v)) = line.split_once('=') {
            map.insert(k.trim().to_string(), v.trim().to_string());
        }
    }
    let rounds = parse_usize(&map, "rounds", 50);
    let num_slots = parse_usize(&map, "num_slots", 256).max(4);
    let ops_per_round = parse_usize(&map, "ops_per_round", 20_000);

    let mut recipe = WheelChurn {
        wheel: TimerWheel::new(num_slots),
        num_slots,
        rounds,
        ops_per_round,
        seq: 0,
    };
    let report = grow(&mut recipe, "rust");

    if growth_to_json(&report, &mut io::stdout().lock()).is_err() {
        eprintln!("growth_main: failed to write json");
        return ExitCode::FAILURE;
    }
    ExitCode::SUCCESS
}

examples/sample_app.rs (line 63)

fn tif_supervisor() {
    println!("== base: order time-in-force supervisor ==");

    let tape = [
        (0usize, Event::Rest("ORD-A", 3)),
        (0, Event::Rest("ORD-B", 5)),
        (0, Event::Rest("ORD-C", 9)),
        (0, Event::Rest("ORD-D", 12)),
        (2, Event::Fill("ORD-B")),
        (4, Event::Amend("ORD-C", 6)),
    ];

    let mut expiries: TimerWheel<&'static str> = TimerWheel::new(256);
    let mut timer_of: Vec<(&'static str, u64)> = Vec::new();
    let lookup = |map: &Vec<(&'static str, u64)>, ord: &str| {
        map.iter().find(|(o, _)| *o == ord).map(|(_, t)| *t)
    };

    let session_secs = 11;
    for second in 0..=session_secs {
        for (at, ev) in tape.iter() {
            if *at != second {
                continue;
            }
            match ev {
                Event::Rest(ord, tif) => {
                    let id = expiries.schedule(*tif, ord);
                    timer_of.push((ord, id));
                    println!("  t={second}s rest {ord} tif={tif}s");
                }
                Event::Fill(ord) => {
                    let id = lookup(&timer_of, ord).expect("a resting order");
                    expiries.cancel(id);
                    println!("  t={second}s fill {ord} -> expiry cancelled");
                }
                Event::Amend(ord, tif) => {
                    let id = lookup(&timer_of, ord).expect("a resting order");
                    expiries.reschedule(id, *tif);
                    println!("  t={second}s amend {ord} tif -> {tif}s from now");
                }
            }
        }
        if second == session_secs {
            break;
        }
        for ord in expiries.tick() {
            println!("  t={}s expire {ord}", second + 1);
        }
    }

    let unfilled = expiries.drain();
    println!(
        "  session close: {} orders still resting {:?}",
        unfilled.len(),
        unfilled
    );
    println!("  pending after drain: {}", expiries.pending());

    assert_eq!(
        unfilled,
        vec!["ORD-D"],
        "only the 12s TIF outlives the session"
    );
    assert_eq!(expiries.pending(), 0);
}

pub fn num_slots(&self) -> usize

pub fn max_delay(&self) -> u64

Largest delay the wheel can represent: a timer can sit out at most i32::MAX revolutions of N slots. Held at the signed bound rather than u32::MAX so the Java port refuses exactly the same delays.

pub fn pending(&self) -> usize

Number of live (scheduled, not yet fired or cancelled) timers. A correct wheel returns this to 0 once every scheduled timer has fired; a leak would let it climb without bound.

Examples found in repository

examples/growth_main.rs (line 59)

    fn memory_bytes(&mut self) -> u64 {
        self.wheel.pending() as u64 * ENTRY_BYTES
    }
    fn live_bytes(&mut self) -> u64 {
        // The genuinely-in-flight timers are the live set; a correct wheel holds
        // exactly those, so resident == live.
        self.wheel.pending() as u64 * ENTRY_BYTES
    }
    fn structures(&mut self) -> Vec<(String, u64)> {
        vec![("pending".to_string(), self.wheel.pending() as u64)]
    }

examples/sample_app.rs (line 107)

fn tif_supervisor() {
    println!("== base: order time-in-force supervisor ==");

    let tape = [
        (0usize, Event::Rest("ORD-A", 3)),
        (0, Event::Rest("ORD-B", 5)),
        (0, Event::Rest("ORD-C", 9)),
        (0, Event::Rest("ORD-D", 12)),
        (2, Event::Fill("ORD-B")),
        (4, Event::Amend("ORD-C", 6)),
    ];

    let mut expiries: TimerWheel<&'static str> = TimerWheel::new(256);
    let mut timer_of: Vec<(&'static str, u64)> = Vec::new();
    let lookup = |map: &Vec<(&'static str, u64)>, ord: &str| {
        map.iter().find(|(o, _)| *o == ord).map(|(_, t)| *t)
    };

    let session_secs = 11;
    for second in 0..=session_secs {
        for (at, ev) in tape.iter() {
            if *at != second {
                continue;
            }
            match ev {
                Event::Rest(ord, tif) => {
                    let id = expiries.schedule(*tif, ord);
                    timer_of.push((ord, id));
                    println!("  t={second}s rest {ord} tif={tif}s");
                }
                Event::Fill(ord) => {
                    let id = lookup(&timer_of, ord).expect("a resting order");
                    expiries.cancel(id);
                    println!("  t={second}s fill {ord} -> expiry cancelled");
                }
                Event::Amend(ord, tif) => {
                    let id = lookup(&timer_of, ord).expect("a resting order");
                    expiries.reschedule(id, *tif);
                    println!("  t={second}s amend {ord} tif -> {tif}s from now");
                }
            }
        }
        if second == session_secs {
            break;
        }
        for ord in expiries.tick() {
            println!("  t={}s expire {ord}", second + 1);
        }
    }

    let unfilled = expiries.drain();
    println!(
        "  session close: {} orders still resting {:?}",
        unfilled.len(),
        unfilled
    );
    println!("  pending after drain: {}", expiries.pending());

    assert_eq!(
        unfilled,
        vec!["ORD-D"],
        "only the 12s TIF outlives the session"
    );
    assert_eq!(expiries.pending(), 0);
}

pub fn is_empty(&self) -> bool

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

Entries physically held in one bucket, including cancelled ones not yet swept. Reading the spread across buckets is how you catch a workload whose delays all collide on one slot.

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

Schedule value to fire in delay_ticks. Returns an id for cancel.

A delay of 0 fires on the next tick, matching Netty’s treatment of a deadline already in the past. A delay past Self::max_delay is clamped; use Self::try_schedule to have it refused instead.

Examples found in repository

examples/perf_features.rs (line 218)

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

/// The baseline: base `schedule`, the O(1) per-op write every feature either
/// decorates or replaces. Re-measured immediately before EACH feature is
/// classified rather than once at the top. Measured once, it sits several
/// half-million-timer builds away from the feature it is compared against, and
/// on this host that gap moves it between 3000 and 4300 ns run to run - as large
/// as a real feature delta. `metrics`, whose entire cost is one u64 increment,
/// flipped between auxiliary and hot-path on that drift alone; measured
/// adjacent, both runs land on auxiliary.
fn base_p50() -> u64 {
    let mut scratch: TimerWheel<u32> = TimerWheel::new(SLOTS);
    let mut w = base_wheel(CANON);
    let m = keyed(
        BATCH,
        |i| {
            scratch.schedule(resident_delay(i), 0);
        },
        |i| {
            w.schedule(resident_delay(i), 0);
        },
    );
    eprintln!("base schedule: p50 {} p99 {} max {}", m.p50, m.p99, m.max);
    m.p50
}

examples/growth_main.rs (line 54)

    fn op(&mut self, _round: usize, i: usize) {
        // Schedule one timer somewhere in the next rotation, then advance the hand
        // one tick (firing anything now due). Over many ops the in-flight set
        // reaches a steady size; a leaking tick would let it grow without bound.
        let delay = (i % (self.num_slots - 1)) + 1;
        self.wheel.schedule(delay, self.seq);
        self.seq += 1;
        let _ = self.wheel.tick();
    }

examples/sample_app.rs (line 77)

fn tif_supervisor() {
    println!("== base: order time-in-force supervisor ==");

    let tape = [
        (0usize, Event::Rest("ORD-A", 3)),
        (0, Event::Rest("ORD-B", 5)),
        (0, Event::Rest("ORD-C", 9)),
        (0, Event::Rest("ORD-D", 12)),
        (2, Event::Fill("ORD-B")),
        (4, Event::Amend("ORD-C", 6)),
    ];

    let mut expiries: TimerWheel<&'static str> = TimerWheel::new(256);
    let mut timer_of: Vec<(&'static str, u64)> = Vec::new();
    let lookup = |map: &Vec<(&'static str, u64)>, ord: &str| {
        map.iter().find(|(o, _)| *o == ord).map(|(_, t)| *t)
    };

    let session_secs = 11;
    for second in 0..=session_secs {
        for (at, ev) in tape.iter() {
            if *at != second {
                continue;
            }
            match ev {
                Event::Rest(ord, tif) => {
                    let id = expiries.schedule(*tif, ord);
                    timer_of.push((ord, id));
                    println!("  t={second}s rest {ord} tif={tif}s");
                }
                Event::Fill(ord) => {
                    let id = lookup(&timer_of, ord).expect("a resting order");
                    expiries.cancel(id);
                    println!("  t={second}s fill {ord} -> expiry cancelled");
                }
                Event::Amend(ord, tif) => {
                    let id = lookup(&timer_of, ord).expect("a resting order");
                    expiries.reschedule(id, *tif);
                    println!("  t={second}s amend {ord} tif -> {tif}s from now");
                }
            }
        }
        if second == session_secs {
            break;
        }
        for ord in expiries.tick() {
            println!("  t={}s expire {ord}", second + 1);
        }
    }

    let unfilled = expiries.drain();
    println!(
        "  session close: {} orders still resting {:?}",
        unfilled.len(),
        unfilled
    );
    println!("  pending after drain: {}", expiries.pending());

    assert_eq!(
        unfilled,
        vec!["ORD-D"],
        "only the 12s TIF outlives the session"
    );
    assert_eq!(expiries.pending(), 0);
}

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

Schedule value, refusing a delay the wheel cannot represent.

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

Mark a scheduled timer cancelled. Returns true if it was pending.

Examples found in repository

examples/sample_app.rs (line 83)

fn tif_supervisor() {
    println!("== base: order time-in-force supervisor ==");

    let tape = [
        (0usize, Event::Rest("ORD-A", 3)),
        (0, Event::Rest("ORD-B", 5)),
        (0, Event::Rest("ORD-C", 9)),
        (0, Event::Rest("ORD-D", 12)),
        (2, Event::Fill("ORD-B")),
        (4, Event::Amend("ORD-C", 6)),
    ];

    let mut expiries: TimerWheel<&'static str> = TimerWheel::new(256);
    let mut timer_of: Vec<(&'static str, u64)> = Vec::new();
    let lookup = |map: &Vec<(&'static str, u64)>, ord: &str| {
        map.iter().find(|(o, _)| *o == ord).map(|(_, t)| *t)
    };

    let session_secs = 11;
    for second in 0..=session_secs {
        for (at, ev) in tape.iter() {
            if *at != second {
                continue;
            }
            match ev {
                Event::Rest(ord, tif) => {
                    let id = expiries.schedule(*tif, ord);
                    timer_of.push((ord, id));
                    println!("  t={second}s rest {ord} tif={tif}s");
                }
                Event::Fill(ord) => {
                    let id = lookup(&timer_of, ord).expect("a resting order");
                    expiries.cancel(id);
                    println!("  t={second}s fill {ord} -> expiry cancelled");
                }
                Event::Amend(ord, tif) => {
                    let id = lookup(&timer_of, ord).expect("a resting order");
                    expiries.reschedule(id, *tif);
                    println!("  t={second}s amend {ord} tif -> {tif}s from now");
                }
            }
        }
        if second == session_secs {
            break;
        }
        for ord in expiries.tick() {
            println!("  t={}s expire {ord}", second + 1);
        }
    }

    let unfilled = expiries.drain();
    println!(
        "  session close: {} orders still resting {:?}",
        unfilled.len(),
        unfilled
    );
    println!("  pending after drain: {}", expiries.pending());

    assert_eq!(
        unfilled,
        vec!["ORD-D"],
        "only the 12s TIF outlives the session"
    );
    assert_eq!(expiries.pending(), 0);
}

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

Move a pending timer to a new delay, keeping its id. Returns false if the id is not pending (already fired, already cancelled, unknown).

Unlike cancel this removes the entry eagerly - leaving a flagged entry behind would let one id sit in two buckets at once.

Examples found in repository

examples/sample_app.rs (line 88)

fn tif_supervisor() {
    println!("== base: order time-in-force supervisor ==");

    let tape = [
        (0usize, Event::Rest("ORD-A", 3)),
        (0, Event::Rest("ORD-B", 5)),
        (0, Event::Rest("ORD-C", 9)),
        (0, Event::Rest("ORD-D", 12)),
        (2, Event::Fill("ORD-B")),
        (4, Event::Amend("ORD-C", 6)),
    ];

    let mut expiries: TimerWheel<&'static str> = TimerWheel::new(256);
    let mut timer_of: Vec<(&'static str, u64)> = Vec::new();
    let lookup = |map: &Vec<(&'static str, u64)>, ord: &str| {
        map.iter().find(|(o, _)| *o == ord).map(|(_, t)| *t)
    };

    let session_secs = 11;
    for second in 0..=session_secs {
        for (at, ev) in tape.iter() {
            if *at != second {
                continue;
            }
            match ev {
                Event::Rest(ord, tif) => {
                    let id = expiries.schedule(*tif, ord);
                    timer_of.push((ord, id));
                    println!("  t={second}s rest {ord} tif={tif}s");
                }
                Event::Fill(ord) => {
                    let id = lookup(&timer_of, ord).expect("a resting order");
                    expiries.cancel(id);
                    println!("  t={second}s fill {ord} -> expiry cancelled");
                }
                Event::Amend(ord, tif) => {
                    let id = lookup(&timer_of, ord).expect("a resting order");
                    expiries.reschedule(id, *tif);
                    println!("  t={second}s amend {ord} tif -> {tif}s from now");
                }
            }
        }
        if second == session_secs {
            break;
        }
        for ord in expiries.tick() {
            println!("  t={}s expire {ord}", second + 1);
        }
    }

    let unfilled = expiries.drain();
    println!(
        "  session close: {} orders still resting {:?}",
        unfilled.len(),
        unfilled
    );
    println!("  pending after drain: {}", expiries.pending());

    assert_eq!(
        unfilled,
        vec!["ORD-D"],
        "only the 12s TIF outlives the session"
    );
    assert_eq!(expiries.pending(), 0);
}

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

Advance the hand one tick. Returns the values of all timers that fired (rounds was 0 and not cancelled). Cancelled timers are dropped silently. Other timers have their rounds decremented.

Examples found in repository

examples/growth_main.rs (line 56)

    fn op(&mut self, _round: usize, i: usize) {
        // Schedule one timer somewhere in the next rotation, then advance the hand
        // one tick (firing anything now due). Over many ops the in-flight set
        // reaches a steady size; a leaking tick would let it grow without bound.
        let delay = (i % (self.num_slots - 1)) + 1;
        self.wheel.schedule(delay, self.seq);
        self.seq += 1;
        let _ = self.wheel.tick();
    }

examples/sample_app.rs (line 96)

fn tif_supervisor() {
    println!("== base: order time-in-force supervisor ==");

    let tape = [
        (0usize, Event::Rest("ORD-A", 3)),
        (0, Event::Rest("ORD-B", 5)),
        (0, Event::Rest("ORD-C", 9)),
        (0, Event::Rest("ORD-D", 12)),
        (2, Event::Fill("ORD-B")),
        (4, Event::Amend("ORD-C", 6)),
    ];

    let mut expiries: TimerWheel<&'static str> = TimerWheel::new(256);
    let mut timer_of: Vec<(&'static str, u64)> = Vec::new();
    let lookup = |map: &Vec<(&'static str, u64)>, ord: &str| {
        map.iter().find(|(o, _)| *o == ord).map(|(_, t)| *t)
    };

    let session_secs = 11;
    for second in 0..=session_secs {
        for (at, ev) in tape.iter() {
            if *at != second {
                continue;
            }
            match ev {
                Event::Rest(ord, tif) => {
                    let id = expiries.schedule(*tif, ord);
                    timer_of.push((ord, id));
                    println!("  t={second}s rest {ord} tif={tif}s");
                }
                Event::Fill(ord) => {
                    let id = lookup(&timer_of, ord).expect("a resting order");
                    expiries.cancel(id);
                    println!("  t={second}s fill {ord} -> expiry cancelled");
                }
                Event::Amend(ord, tif) => {
                    let id = lookup(&timer_of, ord).expect("a resting order");
                    expiries.reschedule(id, *tif);
                    println!("  t={second}s amend {ord} tif -> {tif}s from now");
                }
            }
        }
        if second == session_secs {
            break;
        }
        for ord in expiries.tick() {
            println!("  t={}s expire {ord}", second + 1);
        }
    }

    let unfilled = expiries.drain();
    println!(
        "  session close: {} orders still resting {:?}",
        unfilled.len(),
        unfilled
    );
    println!("  pending after drain: {}", expiries.pending());

    assert_eq!(
        unfilled,
        vec!["ORD-D"],
        "only the 12s TIF outlives the session"
    );
    assert_eq!(expiries.pending(), 0);
}

examples/perf_features.rs (line 268)

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(&mut self, ticks: usize) -> Vec<V>

Advance ticks ticks and return everything that fired across them, in tick order. A ticker thread that woke late catches up here rather than firing a whole revolution’s timers on one bucket.

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

Remove every pending timer and return its value. The hand stays where it is. This is the shutdown path: Netty’s HashedWheelTimer::stop hands back the timeouts it never got to run, and so does this.

Examples found in repository

examples/sample_app.rs (line 101)

fn tif_supervisor() {
    println!("== base: order time-in-force supervisor ==");

    let tape = [
        (0usize, Event::Rest("ORD-A", 3)),
        (0, Event::Rest("ORD-B", 5)),
        (0, Event::Rest("ORD-C", 9)),
        (0, Event::Rest("ORD-D", 12)),
        (2, Event::Fill("ORD-B")),
        (4, Event::Amend("ORD-C", 6)),
    ];

    let mut expiries: TimerWheel<&'static str> = TimerWheel::new(256);
    let mut timer_of: Vec<(&'static str, u64)> = Vec::new();
    let lookup = |map: &Vec<(&'static str, u64)>, ord: &str| {
        map.iter().find(|(o, _)| *o == ord).map(|(_, t)| *t)
    };

    let session_secs = 11;
    for second in 0..=session_secs {
        for (at, ev) in tape.iter() {
            if *at != second {
                continue;
            }
            match ev {
                Event::Rest(ord, tif) => {
                    let id = expiries.schedule(*tif, ord);
                    timer_of.push((ord, id));
                    println!("  t={second}s rest {ord} tif={tif}s");
                }
                Event::Fill(ord) => {
                    let id = lookup(&timer_of, ord).expect("a resting order");
                    expiries.cancel(id);
                    println!("  t={second}s fill {ord} -> expiry cancelled");
                }
                Event::Amend(ord, tif) => {
                    let id = lookup(&timer_of, ord).expect("a resting order");
                    expiries.reschedule(id, *tif);
                    println!("  t={second}s amend {ord} tif -> {tif}s from now");
                }
            }
        }
        if second == session_secs {
            break;
        }
        for ord in expiries.tick() {
            println!("  t={}s expire {ord}", second + 1);
        }
    }

    let unfilled = expiries.drain();
    println!(
        "  session close: {} orders still resting {:?}",
        unfilled.len(),
        unfilled
    );
    println!("  pending after drain: {}", expiries.pending());

    assert_eq!(
        unfilled,
        vec!["ORD-D"],
        "only the 12s TIF outlives the session"
    );
    assert_eq!(expiries.pending(), 0);
}

pub fn clear(&mut self)

Drop every pending timer and reset the hand. Ids already handed out are never reused, so a late cancel on a cleared timer returns false rather than hitting an unrelated timer.