fastwheel

README

Go TimingWheel

English | 中文

English

TimingWheel is a high-performance, lock-free (Actor Model), hierarchical timing wheel implementation for Go. It is designed to manage millions of scheduled tasks efficiently.

Features

• High Performance:
  • Lock-Free: Uses Actor Model (single goroutine managing the wheel) to avoid lock contention.
  • Zero Allocation: Zero memory allocation in hot paths by reusing the Job interface.
  • Sharding: ShardedWheel distributes tasks across multiple wheels to support high concurrency (millions of tasks) and reduce bottleneck.
• Reliable:
  • Hierarchical: Multi-level wheel (e.g., ms -> sec -> min) to handle large time ranges efficiently.
  • Overflow Protection: Handles long delays gracefully.
• Easy to Use:
  • Supports both Closures (AddJobFunc) and Interfaces (Schedule).
  • Supports periodic tasks (ScheduleFunc).
  • Supports Timer.Stop() to cancel tasks.

Installation

go get github.com/tophgg/fast-timingwheel

Usage

1. Simple Usage

Create a standard timing wheel.

package main

import (
	"fmt"
	"time"
	"github.com/tophgg/fast-timingwheel"
)

func main() {
	// precision: 10ms, slots: 20, levels: 3
	// Level 1: range [0, 200ms)    (tick: 10ms, slots: 20)
	// Level 2: range [0, 4s)       (tick: 200ms, slots: 20)
	// Level 3: range [0, 80s)      (tick: 4s, slots: 20)
	tw := fastwheel.New(10*time.Millisecond, 20, 3)
	tw.Start()
	defer tw.Stop()

	// Task 1 (Level 1): 50ms (< 200ms)
	tw.AddJobFunc(50*time.Millisecond, func() {
		fmt.Println("Task 1 (50ms) done")
	})

	// Task 2 (Level 2): 1s (> 200ms, < 4s)
	tw.AddJobFunc(1*time.Second, func() {
		fmt.Println("Task 2 (1s) done")
	})

	// Task 3 (Level 3): 5s (> 4s)
	tw.AddJobFunc(5*time.Second, func() {
		fmt.Println("Task 3 (5s) done")
	})

	// Wait for all tasks to complete
	time.Sleep(6 * time.Second)
}

2. Sharded Timing Wheel (High Concurrency)

Use ShardedWheel to reduce lock contention when you have massive concurrent add/cancel operations.

// 8 shards, 10ms precision
stw := fastwheel.NewSharded(8, 10*time.Millisecond, 20, 3)
stw.Start()
defer stw.Stop()

// Simulate concurrent task addition
// ShardedWheel distributes tasks to different shards based on the hash of ID or Key,
// reducing lock contention and improving concurrency performance.
for i := 0; i < 10; i++ {
	go func(id int) {
		// Add job by integer ID (automatically routed to the corresponding shard)
		stw.AddJobFuncByInt(int64(id), 100*time.Millisecond, func() {
			fmt.Printf("User %d timeout\n", id)
		})
	}(i)
}

// Add job by string key
stw.AddJobFuncByString("session-123", 200*time.Millisecond, func() {
    fmt.Println("Session 123 timeout")
})

// Wait for tasks to complete
time.Sleep(1 * time.Second)

3. Periodic Tasks

Schedule a task to run repeatedly.

// Initialize timing wheel
tw := fastwheel.New(10*time.Millisecond, 20, 3)
tw.Start()
defer tw.Stop()

// Schedule a periodic task (runs every 1 second)
timer := tw.ScheduleFunc(1*time.Second, func() {
    fmt.Println("Tick...")
})

// Stop after 5 seconds
time.Sleep(5 * time.Second)
timer.Stop()

Benchmarks

Run benchmarks using:

go test -bench=. -benchmem

Performance Comparison

Comparison with standard library time.AfterFunc and fast-timingwheel (this package).

Workload	Metric	NativeTimers	TimingWheel	Difference
1K timers (100ms)	Time/op	102 ms	101 ms	-1% faster
	Mem/op	604 KB	140 KB	-77% less
	Allocs/op	2965	3025	+2% more
10K timers (100ms)	Time/op	116 ms	111 ms	-4% faster
	Mem/op	4.9 MB	1.4 MB	-71% less
	Allocs/op	26 K	30 K	+15% more
100K timers (1s)	Time/op	1.12 s	1.08 s	-3% faster
	Mem/op	38.7 MB	18.8 MB	-51% less
	Allocs/op	235 K	300 K	+27% more

Memory Usage (Steady State)

Count	Native Timer	TimingWheel	Difference
100K Timers	38 MB	6.4 MB	-83% less
1M Timers	398 MB	64 MB	-83% less

Note: Benchmarks run on Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz (macOS).

---

中文 (Chinese)

TimingWheel 是一个高性能、无锁（基于 Actor 模型）、分层的 Go 语言时间轮实现。它专为高效管理百万级定时任务而设计。

特性

• 高性能:
  • 无锁设计: 使用 Actor 模型（单 Goroutine 管理时间轮）避免锁竞争。
  • 零分配: 在热点路径上通过复用 Job 接口实现零内存分配。
  • 分片支持: ShardedTimeWheel 将任务分散到多个时间轮中，支持高并发（百万级任务）并减少瓶颈。
• 可靠性:
  • 分层结构: 多级时间轮（例如：毫秒 -> 秒 -> 分）高效处理大跨度时间。
  • 溢出保护: 优雅处理超长延迟任务。
• 易用性:
  • 支持闭包 (AddJobFunc) 和接口 (AddJob) 两种方式。
  • 支持周期性任务 (ScheduleFunc)。
  • 支持 Timer.Stop() 取消任务。

安装

go get github.com/tophgg/fast-timingwheel

使用指南

1. 基本用法

创建一个标准时间轮。

package main

import (
	"fmt"
	"time"
	"github.com/tophgg/fast-timingwheel"
)

func main() {
	// 精度: 10ms, 槽位: 20, 层级: 3
	// 层级 1: 范围 [0, 200ms)    (tick: 10ms, slots: 20)
	// 层级 2: 范围 [0, 4s)       (tick: 200ms, slots: 20)
	// 层级 3: 范围 [0, 80s)      (tick: 4s, slots: 20)
	tw := fastwheel.New(10*time.Millisecond, 20, 3)
	tw.Start()
	defer tw.Stop()

	// 任务 1 (层级 1): 50ms (< 200ms)
	tw.AddJobFunc(50*time.Millisecond, func() {
		fmt.Println("Task 1 (50ms) done")
	})

	// 任务 2 (层级 2): 1s (> 200ms, < 4s)
	tw.AddJobFunc(1*time.Second, func() {
		fmt.Println("Task 2 (1s) done")
	})

	// 任务 3 (层级 3): 5s (> 4s)
	tw.AddJobFunc(5*time.Second, func() {
		fmt.Println("Task 3 (5s) done")
	})

	// 等待所有任务完成
	time.Sleep(6 * time.Second)
}

2. 分片时间轮 (高并发)

当有大规模并发添加/取消操作时，使用 ShardedWheel 来减少锁竞争。

// 8 个分片, 10ms 精度
stw := fastwheel.NewSharded(8, 10*time.Millisecond, 20, 3)
stw.Start()
defer stw.Stop()

// 模拟并发添加任务
// ShardedWheel 将根据 ID 或 Key 的哈希值将任务分发到不同的分片，
// 从而减少锁竞争，提高并发性能。
for i := 0; i < 10; i++ {
	go func(id int) {
		// 通过整数 ID 添加任务 (自动路由到对应分片)
		stw.AddJobFuncByInt(int64(id), 100*time.Millisecond, func() {
			fmt.Printf("User %d timeout\n", id)
		})
	}(i)
}

// 通过字符串 Key 添加任务
stw.AddJobFuncByString("session-123", 200*time.Millisecond, func() {
    fmt.Println("Session 123 timeout")
})

// 等待任务完成
time.Sleep(1 * time.Second)

3. 周期性任务

调度一个重复运行的任务。

// 初始化时间轮
tw := fastwheel.New(10*time.Millisecond, 20, 3)
tw.Start()
defer tw.Stop()

// 调度周期性任务 (每 1 秒执行一次)
timer := tw.ScheduleFunc(1*time.Second, func() {
    fmt.Println("Tick...")
})

// 运行 5 秒后停止
time.Sleep(5 * time.Second)
timer.Stop()
fmt.Println("Timer stopped")

基准测试 (Benchmarks)

运行基准测试：

go test -bench=. -benchmem

License

Available under the Apache 2.0 License.

Documentation

Index

• type FuncJob
  • func (f FuncJob) Run()
• type Job
• type Level
• type ShardedWheel
  • func NewSharded(shardCount int, tick time.Duration, wheelSize, levelCount int) *ShardedWheel
  • func (sw *ShardedWheel) AddJobByInt(id int64, delay time.Duration, job Job) *Timer
  • func (sw *ShardedWheel) AddJobByString(key string, delay time.Duration, job Job) *Timer
  • func (sw *ShardedWheel) AddJobFuncByInt(id int64, delay time.Duration, f func()) *Timer
  • func (sw *ShardedWheel) AddJobFuncByString(key string, delay time.Duration, f func()) *Timer
  • func (sw *ShardedWheel) AddJobFuncRandom(delay time.Duration, f func()) *Timer
  • func (sw *ShardedWheel) AddJobRandom(delay time.Duration, job Job) *Timer
  • func (sw *ShardedWheel) Start()
  • func (sw *ShardedWheel) Stop()
• type Timer
  • func (t *Timer) GetExpiration() int64
  • func (t *Timer) Stop()
• type Wheel
  • func New(tick time.Duration, wheelSize int, levelCount int) *Wheel
  • func (w *Wheel) AddJob(delay time.Duration, job Job) *Timer
  • func (w *Wheel) AddJobFunc(delay time.Duration, f func()) *Timer
  • func (w *Wheel) RemoveJob(t *Timer)
  • func (w *Wheel) Schedule(interval time.Duration, job Job) *Timer
  • func (w *Wheel) ScheduleFunc(interval time.Duration, f func()) *Timer
  • func (w *Wheel) Start()
  • func (w *Wheel) Stop()

Examples

• New
• NewSharded

Types

type FuncJob

type FuncJob func()

func (FuncJob) Run

func (f FuncJob) Run()

type Job

type Job interface {
	Run()
}

type Level

type Level struct {
	// contains filtered or unexported fields
}

type ShardedWheel

type ShardedWheel struct {
	// contains filtered or unexported fields
}

func NewSharded

func NewSharded(shardCount int, tick time.Duration, wheelSize, levelCount int) *ShardedWheel

Example

package main

import (
	"time"

	fastwheel "github.com/tophgg/fast-timingwheel"
)

func main() {
	stw := fastwheel.NewSharded(8, 10*time.Millisecond, 20, 3)
	stw.Start()
	defer stw.Stop()

	userIds := []int64{1001, 1002, 1003}

	for _, uid := range userIds {
		stw.AddJobFuncByInt(uid, 50*time.Millisecond, func() {
			// fmt.Printf("User %d timeout\n", uid)
		})
	}

	time.Sleep(100 * time.Millisecond)
}

func (*ShardedWheel) AddJobByInt

func (sw *ShardedWheel) AddJobByInt(id int64, delay time.Duration, job Job) *Timer

func (*ShardedWheel) AddJobByString

func (sw *ShardedWheel) AddJobByString(key string, delay time.Duration, job Job) *Timer

func (*ShardedWheel) AddJobFuncByInt

func (sw *ShardedWheel) AddJobFuncByInt(id int64, delay time.Duration, f func()) *Timer

func (*ShardedWheel) AddJobFuncByString

func (sw *ShardedWheel) AddJobFuncByString(key string, delay time.Duration, f func()) *Timer

func (*ShardedWheel) AddJobFuncRandom

func (sw *ShardedWheel) AddJobFuncRandom(delay time.Duration, f func()) *Timer

func (*ShardedWheel) AddJobRandom

func (sw *ShardedWheel) AddJobRandom(delay time.Duration, job Job) *Timer

func (*ShardedWheel) Start

func (sw *ShardedWheel) Start()

func (*ShardedWheel) Stop

func (sw *ShardedWheel) Stop()

type Timer

type Timer struct {
	// contains filtered or unexported fields
}

func (*Timer) GetExpiration

func (t *Timer) GetExpiration() int64

func (*Timer) Stop

func (t *Timer) Stop()

type Wheel

type Wheel struct {
	// contains filtered or unexported fields
}

func New

func New(tick time.Duration, wheelSize int, levelCount int) *Wheel

Example

package main

import (
	"fmt"
	"time"

	fastwheel "github.com/tophgg/fast-timingwheel"
)

func main() {
	tw := fastwheel.New(10*time.Millisecond, 20, 3)
	tw.Start()
	defer tw.Stop()

	c := make(chan string)
	tw.AddJobFunc(50*time.Millisecond, func() {
		c <- "Hello World"
	})

	fmt.Println(<-c)

	timer := tw.ScheduleFunc(100*time.Millisecond, func() {
		fmt.Println("Tick")
	})

	time.Sleep(390 * time.Millisecond)

	timer.Stop()

}

Output:
Hello World
Tick
Tick
Tick

func (*Wheel) AddJob

func (w *Wheel) AddJob(delay time.Duration, job Job) *Timer

func (*Wheel) AddJobFunc

func (w *Wheel) AddJobFunc(delay time.Duration, f func()) *Timer

func (*Wheel) RemoveJob

func (w *Wheel) RemoveJob(t *Timer)

func (*Wheel) Schedule

func (w *Wheel) Schedule(interval time.Duration, job Job) *Timer

func (*Wheel) ScheduleFunc

func (w *Wheel) ScheduleFunc(interval time.Duration, f func()) *Timer

func (*Wheel) Start

func (w *Wheel) Start()

func (*Wheel) Stop

func (w *Wheel) Stop()