Go实现层级时间轮

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

type TimingWheel struct {
    // 时间跨度,单位是毫秒
    tick      int64 // in milliseconds
    // 时间轮个数
    wheelSize int64
    // 总跨度
    interval    int64 // in milliseconds
    // 当前指针指向时间
    currentTime int64 // in milliseconds
    // 时间格列表
    buckets     []*bucket
    // 延迟队列
    queue       *delayqueue.DelayQueue
    // 上级的时间轮引用
    overflowWheel unsafe.Pointer // type: *TimingWheel

    exitC     chan struct{}
    waitGroup waitGroupWrapper
}

1
2
3
4
5
6
7
8

type bucket struct {
    // 任务的过期时间
    expiration int64

    mu     sync.Mutex
    // 相同过期时间的任务队列
    timers *list.List
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

type Timer struct {
  // 到期时间
    expiration int64 // in milliseconds
  // 要被执行的具体任务
    task       func()
    // Timer所在bucket的指针
    b unsafe.Pointer // type: *bucket
    // bucket列表中对应的元素
    element *list.Element
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

func main() {
    tw := timingwheel.NewTimingWheel(time.Second, 10)
    tw.Start()
}

func NewTimingWheel(tick time.Duration, wheelSize int64) *TimingWheel {
  // 将传入的tick转化成毫秒
    tickMs := int64(tick / time.Millisecond)
  // 如果小于零，那么panic
    if tickMs <= 0 {
        panic(errors.New("tick must be greater than or equal to 1ms"))
    }
    // 设置开始时间
    startMs := timeToMs(time.Now().UTC())
    // 初始化TimingWheel
    return newTimingWheel(
        tickMs,
        wheelSize,
        startMs,
        delayqueue.New(int(wheelSize)),
    )
}

func newTimingWheel(tickMs int64, wheelSize int64, startMs int64, queue *delayqueue.DelayQueue) *TimingWheel {
  // 初始化buckets的大小
    buckets := make([]*bucket, wheelSize)
    for i := range buckets {
        buckets[i] = newBucket()
    }
  // 实例化TimingWheel
    return &TimingWheel{
        tick:        tickMs,
        wheelSize:   wheelSize,
    // currentTime必须是tickMs的倍数，所以这里使用truncate进行修剪
        currentTime: truncate(startMs, tickMs),
        interval:    tickMs * wheelSize,
        buckets:     buckets,
        queue:       queue,
        exitC:       make(chan struct{}),
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

func (tw *TimingWheel) Start() {
    // Poll会执行一个无限循环，将到期的元素放入到queue的C管道中
    tw.waitGroup.Wrap(func() {
        tw.queue.Poll(tw.exitC, func() int64 {
            return timeToMs(time.Now().UTC())
        })
    })
    // 开启无限循环获取queue中C的数据
    tw.waitGroup.Wrap(func() {
        for {
            select {
            // 从队列里面出来的数据都是到期的bucket
            case elem := <-tw.queue.C:
                b := elem.(*bucket)
                // 时间轮会将当前时间 currentTime 往前移动到 bucket的到期时间
                tw.advanceClock(b.Expiration())
                // 取出bucket队列的数据，并调用addOrRun方法执行
                b.Flush(tw.addOrRun)
            case <-tw.exitC:
                return
            }
        }
    })
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

func (tw *TimingWheel) advanceClock(expiration int64) {
    currentTime := atomic.LoadInt64(&tw.currentTime)
    // 过期时间大于等于（当前时间+tick）
    if expiration >= currentTime+tw.tick {
        // 将currentTime设置为expiration，从而推进currentTime
        currentTime = truncate(expiration, tw.tick)
        atomic.StoreInt64(&tw.currentTime, currentTime)

        // Try to advance the clock of the overflow wheel if present
        // 如果有上层时间轮，那么递归调用上层时间轮的引用
        overflowWheel := atomic.LoadPointer(&tw.overflowWheel)
        if overflowWheel != nil {
            (*TimingWheel)(overflowWheel).advanceClock(currentTime)
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

func (b *bucket) Flush(reinsert func(*Timer)) {
    var ts []*Timer

    b.mu.Lock()
    // 循环获取bucket队列节点
    for e := b.timers.Front(); e != nil; {
        next := e.Next()

        t := e.Value.(*Timer)
        // 将头节点移除bucket队列
        b.remove(t)
        ts = append(ts, t)

        e = next
    }
    b.mu.Unlock()

    b.SetExpiration(-1) // TODO: Improve the coordination with b.Add()

    for _, t := range ts {
        reinsert(t)
    }
}

1
2
3
4
5
6
7

func (tw *TimingWheel) addOrRun(t *Timer) {
    // 如果已经过期，那么直接执行
    if !tw.add(t) {
        // 异步执行定时任务
        go t.task()
    }
}

1
2
3
4
5
6
7
8
9

func main() {
    tw := timingwheel.NewTimingWheel(time.Second, 10)
    tw.Start()
    // 添加任务
    tw.AfterFunc(time.Second*15, func() {
        fmt.Println("The timer fires")
        exitC <- time.Now().UTC()
    })
}

1
2
3
4
5
6
7
8

func (tw *TimingWheel) AfterFunc(d time.Duration, f func()) *Timer {
    t := &Timer{
        expiration: timeToMs(time.Now().UTC().Add(d)),
        task:       f,
    }
    tw.addOrRun(t)
    return t
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

func (tw *TimingWheel) add(t *Timer) bool {
    currentTime := atomic.LoadInt64(&tw.currentTime)
    // 已经过期
    if t.expiration < currentTime+tw.tick {
        // Already expired
        return false
    //  到期时间在第一层环内
    } else if t.expiration < currentTime+tw.interval {
        // Put it into its own bucket
        // 获取时间轮的位置
        virtualID := t.expiration / tw.tick
        b := tw.buckets[virtualID%tw.wheelSize]
        // 将任务放入到bucket队列中
        b.Add(t)
        // 如果是相同的时间，那么返回false，防止被多次插入到队列中
        if b.SetExpiration(virtualID * tw.tick) {
            // 将该bucket加入到延迟队列中
            tw.queue.Offer(b, b.Expiration())
        }

        return true
    } else {
        // Out of the interval. Put it into the overflow wheel
        // 如果放入的到期时间超过第一层时间轮，那么放到上一层中去
        overflowWheel := atomic.LoadPointer(&tw.overflowWheel)
        if overflowWheel == nil {
            atomic.CompareAndSwapPointer(
                &tw.overflowWheel,
                nil,
                // 需要注意的是，这里tick变成了interval
                unsafe.Pointer(newTimingWheel(
                    tw.interval,
                    tw.wheelSize,
                    currentTime,
                    tw.queue,
                )),
            )
            overflowWheel = atomic.LoadPointer(&tw.overflowWheel)
        }
        // 往上递归
        return (*TimingWheel)(overflowWheel).add(t)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

goos: darwin
goarch: amd64
pkg: gin-test/api/main
BenchmarkTimingWheel_StartStop/N-1m-12   4582120               254 ns/op              85 B/op          2 allocs/op
BenchmarkTimingWheel_StartStop/N-5m-12   3356630               427 ns/op              46 B/op          1 allocs/op
BenchmarkTimingWheel_StartStop/N-10m-12                  2474842               483 ns/op              60 B/op          1 allocs/op
BenchmarkStandardTimer_StartStop/N-1m-12                 6777975               179 ns/op              84 B/op          1 allocs/op
BenchmarkStandardTimer_StartStop/N-5m-12                 6431217               231 ns/op              85 B/op          1 allocs/op
BenchmarkStandardTimer_StartStop/N-10m-12                5374492               266 ns/op              83 B/op          1 allocs/op
PASS
ok      gin-test/api/main       60.414s

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

name                      old time/op  new time/op  delta
AfterFunc-12              1.57ms ± 1%  0.07ms ± 1%  -95.42%  (p=0.000 n=10+8)
After-12                  1.63ms ± 3%  0.11ms ± 1%  -93.54%  (p=0.000 n=9+10)
Stop-12                   78.3µs ± 3%  73.6µs ± 3%   -6.01%  (p=0.000 n=9+10)
SimultaneousAfterFunc-12   138µs ± 1%   111µs ± 1%  -19.57%  (p=0.000 n=10+9)
StartStop-12              28.7µs ± 1%  31.5µs ± 5%   +9.64%  (p=0.000 n=10+7)
Reset-12                  6.78µs ± 1%  4.24µs ± 7%  -37.45%  (p=0.000 n=9+10)
Sleep-12                   183µs ± 1%   125µs ± 1%  -31.67%  (p=0.000 n=10+9)
Ticker-12                 5.40ms ± 2%  0.03ms ± 1%  -99.43%  (p=0.000 n=10+10)
...