数据结构与算法回顾（三）：时间轮

type TimingWheel struct {
	//...
	interval      time.Duration	// 时间轮 ticker 时间
	slots         []*list.List	// 模拟时间轮环形结构，加任务存储
	timers        *SafeMap		// 用于存储
	tickedPos     int			// 记录当前指针所在的位置
	numSlots      int		// 时间轮的槽位数量
	//...
}

func (tw *TimingWheel) onTick() {
	tw.tickedPos = (tw.tickedPos + 1) % tw.numSlots
	l := tw.slots[tw.tickedPos]
	tw.scanAndRunTasks(l)
}

func (tw *TimingWheel) getPositionAndCircle(d time.Duration) (pos, circle int) {
	steps := int(d / tw.interval)
	pos = (tw.tickedPos + steps) % tw.numSlots
	circle = (steps - 1) / tw.numSlots

	return
}

func (tw *TimingWheel) setTimerPosition(pos int, task *timingEntry) {
	if val, ok := tw.timers.Get(task.key); ok {
		timer := val.(*positionEntry)
		timer.item = task
		timer.pos = pos
	} else {
		// 保存位置 pos 和任务
		tw.timers.Set(task.key, &positionEntry{
			pos:  pos,
			item: task,
		})
	}
}

type timingEntry struct {
	baseEntry
	value   interface{}
	circle  int		// 记住这个字段：用以解决分层的问题
	diff    int
	removed bool
}

// A TimingWheel is a timing wheel object to schedule tasks.
type TimingWheel struct {
	interval      time.Duration  // 单个时间格时间间隔
	ticker        timex.Ticker	// 定时器，做时间推动，以 interval 为单位推进
	slots         []*list.List	// 时间轮
	timers        *SafeMap	// 存储 task{key, value} 的 map [执行 execute 所需要的参数]
	tickedPos     int	 // at previous virtual circle
	numSlots      int	// 初始化 slots num
	execute       Execute	 // 执行函数
	setChannel    chan timingEntry
	moveChannel   chan baseEntry
	removeChannel chan interface{}
	drainChannel  chan func(key, value interface{})
	stopChannel   chan lang.PlaceholderType
}

// 真正做初始化
func newTimingWheelWithClock(interval time.Duration, numSlots int, execute Execute, ticker timex.Ticker) (
    *TimingWheel, error) {
	//...
    // 初始化 slots 中用来存储任务的所有 slots
    tw.initSlots()
    // start ticker
    go tw.run()

    return tw, nil
}

func (tw *TimingWheel) run() {
	for {
		select {
		// 定时器 ticker ，时间推动
		case <-tw.ticker.Chan():
			tw.onTick()
		// 异步处理增加任务
		case task := <-tw.setChannel:
			tw.setTask(&task)
		case key := <-tw.removeChannel:
			tw.removeTask(key)
		case task := <-tw.moveChannel:
			tw.moveTask(task)
		case fn := <-tw.drainChannel:
			tw.drainAll(fn)
		case <-tw.stopChannel:
			tw.ticker.Stop()
			return
		}
	}
}

func (tw *TimingWheel) onTick() {
	tw.tickedPos = (tw.tickedPos + 1) % tw.numSlots
	l := tw.slots[tw.tickedPos]
	tw.scanAndRunTasks(l)
}

func (tw *TimingWheel) scanAndRunTasks(l *list.List) {
	var tasks []timingTask

	for e := l.Front(); e != nil; {
		task := e.Value.(*timingEntry)
		if task.removed {
			next := e.Next()
			l.Remove(e)
			e = next
			continue
		} else if task.circle > 0 {
			task.circle--
			e = e.Next()
			continue
		} else if task.diff > 0 {
			next := e.Next()
			l.Remove(e)
			// (tw.tickedPos+task.diff)%tw.numSlots
			// cannot be the same value of tw.tickedPos
			pos := (tw.tickedPos + task.diff) % tw.numSlots
			tw.slots[pos].PushBack(task)
			tw.setTimerPosition(pos, task)
			task.diff = 0
			e = next
			continue
		}

		tasks = append(tasks, timingTask{
			key:   task.key,
			value: task.value,
		})
		next := e.Next()
		l.Remove(e)
		tw.timers.Del(task.key)
		e = next
	}

	tw.runTasks(tasks)
}

// MoveTimer moves the task with the given key to the given delay.
func (tw *TimingWheel) MoveTimer(key interface{}, delay time.Duration) error {
	if delay <= 0 || key == nil {
		return ErrArgument
	}

	select {
	case tw.moveChannel <- baseEntry{
		delay: delay,
		key:   key,
	}:
		return nil
	case <-tw.stopChannel:
		return ErrClosed
	}
}

func (tw *TimingWheel) moveTask(task baseEntry) {
    // timers: 通过任务的 key 获取 pos 位置及 task
    val, ok := tw.timers.Get(task.key)
    if !ok {
        return
    }

    timer := val.(*positionEntry)
      // {delay < interval} => 延迟时间比一个时间格间隔还小，没有更小的刻度，说明任务应该立即执行
    if task.delay < tw.interval {
        threading.GoSafe(func() {
            tw.execute(timer.item.key, timer.item.value)
        })
        return
    }
    // 如果 > interval，则通过 延迟时间 delay 计算其出时间轮中的 new pos, circle
    pos, circle := tw.getPositionAndCircle(task.delay)
    if pos >= timer.pos {
        timer.item.circle = circle
        // 记录前后的移动 offset。为了后面过程重新入队
        timer.item.diff = pos - timer.pos
    } else if circle > 0 {
        // 转移到下一层，将 circle 转换为 diff 一部分
        circle--
        timer.item.circle = circle
        // 因为是一个数组，要加上 numSlots [也就是相当于要走到下一层]
        timer.item.diff = tw.numSlots + pos - timer.pos
    } else {
        // 如果 offset 提前了，此时 task 也还在第一层
        // 标记删除老的 task，并重新入队，等待被执行
        timer.item.removed = true
        newItem := &timingEntry{
            baseEntry: task,
            value:     timer.item.value,
        }
        tw.slots[pos].PushBack(newItem)
        tw.setTimerPosition(pos, newItem)
    }
}