timerwheel

timerwheel provides delayed task scheduling with a hierarchical timer wheel. The scheduler owns time advancement and dispatches expired tasks to an executor, so long-running work does not block later timeout scheduling.

Install

[dependencies]
timerwheel = "0.1"

Enable the optional Tokio adapter with:

[dependencies]
timerwheel = { version = "0.1", features = ["tokio"] }

Quick Start

use std::sync::mpsc;
use std::time::Duration;

use timerwheel::Timer;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let timer = Timer::builder()
        .tick(Duration::from_millis(1))
        .bucket_count(512)
        .build()?;
    let (tx, rx) = mpsc::channel();

    let timeout = timer.schedule(Duration::from_millis(10), move || {
        tx.send("fired").expect("send succeeds");
    })?;

    println!("{}", rx.recv_timeout(Duration::from_secs(1))?);
    assert!(timeout.is_expired());
    timer.shutdown()?;
    Ok(())
}

Configuration

Timer::builder() configures the scheduler:

• tick: timer resolution.
• bucket_count: buckets per timing-wheel level.
• command_capacity: bounded schedule-command capacity.
• max_pending: maximum accepted pending timeouts.
• backpressure_policy: reject or wait when the command path is full.
• expired_task_policy: reject or retry when executor dispatch is saturated.
• metric_sink: observe timer metric snapshots after metric changes.

executor::Pool::builder() configures the worker executor:

• workers: fixed worker count.
• queue_capacity: bounded task queue capacity.
• reject_policy: reject or wait when the worker queue is full.
• metric_sink: observe executor metric snapshots after metric changes.

Prelude

Applications that use timer, executor, policy, and metric sink APIs together can import the prelude:

use timerwheel::prelude::*;

The prelude keeps timer and executor metric sinks distinct with aliases: TimerMetricSink, NoopTimerMetricSink, ExecutorMetricSink, and NoopExecutorMetricSink.

Executor Boundary

The scheduler advances time and dispatches expired tasks through executor::Executor::try_execute. The default pool and Tokio executor keep worker execution from blocking timer-wheel advancement. Custom executors must return from try_execute immediately.

use std::sync::{Arc, Mutex, mpsc};
use std::thread::{self, JoinHandle};

use timerwheel::executor::{BoxTask, Executor, ExecutorMetrics, RejectedTask};

struct ChannelExecutor {
    sender: Mutex<Option<mpsc::Sender<BoxTask>>>,
    worker: Mutex<Option<JoinHandle<()>>>,
}

impl ChannelExecutor {
    fn new() -> Arc<Self> {
        let (sender, receiver) = mpsc::channel::<BoxTask>();
        let worker = thread::spawn(move || {
            while let Ok(task) = receiver.recv() {
                task.run();
            }
        });

        Arc::new(Self {
            sender: Mutex::new(Some(sender)),
            worker: Mutex::new(Some(worker)),
        })
    }
}

impl Executor for ChannelExecutor {
    fn try_execute(&self, task: BoxTask) -> std::result::Result<(), RejectedTask> {
        let Some(sender) = self.sender.lock().expect("sender lock").as_ref().cloned() else {
            return Err(RejectedTask::new(timerwheel::Error::Closed, task));
        };
        sender
            .send(task)
            .map_err(|error| RejectedTask::new(timerwheel::Error::Closed, error.0))
    }

    fn shutdown(&self) -> timerwheel::Result<()> {
        self.sender.lock().expect("sender lock").take();
        if let Some(worker) = self.worker.lock().expect("worker lock").take() {
            let _ = worker.join();
        }
        Ok(())
    }

    fn metrics(&self) -> ExecutorMetrics {
        ExecutorMetrics::default()
    }
}

Cancellation

Timeout::cancel() is logical. A cancelled entry may stay in a bucket until the scheduler reaches that bucket, but it will not execute.

Metrics

Timer::metrics() and executor::Pool::metrics() return immutable snapshots. The counters cover accepted, expired, cancelled, rejected, panicked, pending, and queue-depth states. Builders can also attach metric sinks for push-style observation.

Shutdown

Timer::shutdown() stops accepting new schedules, wakes the scheduler, drains accepted commands, and shuts down timer-owned executors. Calling shutdown more than once is safe.

Optional Tokio Adapter

With the tokio feature:

use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let timer = timerwheel::tokio::Timer::builder().build()?;

    timer
        .schedule(Duration::from_millis(10), || {
            println!("fired");
        })?;

    timer.shutdown()?;
    Ok(())
}