层级时间轮 ｜ skynet 定时器

static void
systime(uint32_t *sec, uint32_t *cs) {
#if !defined(__APPLE__) || defined(AVAILABLE_MAC_OS_X_VERSION_10_12_AND_LATER)
	struct timespec ti;
	clock_gettime(CLOCK_REALTIME, &ti); //获取真实时间, 绝对时间
	*sec = (uint32_t)ti.tv_sec; //单位s 当前时间戳
	*cs = (uint32_t)(ti.tv_nsec / 10000000);//单位10ms， 当前时间戳
#else
	struct timeval tv;
	gettimeofday(&tv, NULL);
	*sec = tv.tv_sec;
	*cs = tv.tv_usec / 10000;
#endif
}

static uint64_t
gettime() {
	uint64_t t;
#if !defined(__APPLE__) || defined(AVAILABLE_MAC_OS_X_VERSION_10_12_AND_LATER)
	struct timespec ti;
	clock_gettime(CLOCK_MONOTONIC, &ti); //获取系统启动时间， 相对时间
	t = (uint64_t)ti.tv_sec * 100; //单位cs， 系统启动到现在的时间
	t += ti.tv_nsec / 10000000; //单位cs
#else
	struct timeval tv;
	gettimeofday(&tv, NULL);
	t = (uint64_t)tv.tv_sec * 100;
	t += tv.tv_usec / 10000;
#endif
	return t;
}

static void 
timer_update(struct timer *T) {
	SPIN_LOCK(T);

	// try to dispatch timeout 0 (rare condition)
	timer_execute(T);

	// shift time first, and then dispatch timer message
	timer_shift(T);

	timer_execute(T);

	SPIN_UNLOCK(T);
}

static inline void
timer_execute(struct timer *T) {
	int idx = T->time & TIME_NEAR_MASK;
	
	while (T->near[idx].head.next) {
		struct timer_node *current = link_clear(&T->near[idx]);
		SPIN_UNLOCK(T);
		// dispatch_list don't need lock T
		dispatch_list(current);
		SPIN_LOCK(T);
	}
}

static void
timer_shift(struct timer *T) {
	int mask = TIME_NEAR;
	uint32_t ct = ++T->time;
	if (ct == 0) {
		move_list(T, 3, 0);
	} else {
		uint32_t time = ct >> TIME_NEAR_SHIFT;
		int i=0;

		while ((ct & (mask-1))==0) {
			int idx=time & TIME_LEVEL_MASK;
			if (idx!=0) {
				move_list(T, i, idx);
				break;				
			}
			mask <<= TIME_LEVEL_SHIFT;
			time >>= TIME_LEVEL_SHIFT;
			++i;
		}
	}
} 

static void
add_node(struct timer *T,struct timer_node *node) {
    // 逾期时间是相对于 T->time的
	uint32_t time=node->expire;
	uint32_t current_time=T->time;
	
    // 高24位一致，表示在一个near周期内。
	if ((time|TIME_NEAR_MASK)==(current_time|TIME_NEAR_MASK)) {
		link(&T->near[time&TIME_NEAR_MASK],node);
	} else {
		int i;
		uint32_t mask=TIME_NEAR << TIME_LEVEL_SHIFT;
		for (i=0;i<3;i++) {
			if ((time|(mask-1))==(current_time|(mask-1))) {
				break;
			}
			mask <<= TIME_LEVEL_SHIFT;
		}

		link(&T->t[i][((time>>(TIME_NEAR_SHIFT + i*TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)],node);	
	}
}

package main

import (
	"container/list"
	"sync"
	"sync/atomic"
	"time"
)

//skynet的时间轮 + 协程池
const (
	TimeNearShift  = 8
	TimeNear       = 1 << TimeNearShift
	TimeLevelShift = 6
	TimeLevel      = 1 << TimeLevelShift
	TimeNearMask   = TimeNear - 1
	TimeLevelMask  = TimeLevel - 1

	//协程池 大小
	WorkerPoolSize = 10
	MaxTaskPerWorker = 20
)

type bucket struct {
	expiration int32
	timers *list.List

	mu sync.Mutex
}

func newBucket() *bucket {
	return &bucket{
		expiration: -1,
		timers:     list.New(),
		mu: sync.Mutex{},
	}
}

func (b*bucket) Add(t *timer)  {
	b.mu.Lock()
	defer b.mu.Unlock()

	b.timers.PushBack(t)
}

func (b*bucket) Flush(reinsert func(t *timer))  {
	b.mu.Lock()
	defer b.mu.Unlock()

	for e := b.timers.Front(); e != nil; {
		next := e.Next()
		reinsert(e.Value.(*timer))

		b.timers.Remove(e)
		e = next
	}
}

type timer struct {
	expiration	 uint32
	f func()
}

var TimingWheel *TimeWheel

func init()  {
	TimingWheel = NewTimeWheel()
	TimingWheel.Start()
}

type TimeWheel struct {
	tick time.Duration
	ticker *time.Ticker
	near 			[TimeNear]*bucket
	t				[4][TimeLevel]*bucket
	time 			uint32

	WorkPool *WorkPool
	exit chan struct{}
	exitFlag uint32
}

func NewTimeWheel() *TimeWheel {
	tw := &TimeWheel{
		tick:     10*time.Millisecond,
		time:     0,
		WorkPool: NewWorkPool(WorkerPoolSize, MaxTaskPerWorker),
		exit:     make(chan struct{}),
		exitFlag: 0,
	}
	for i :=0; i < TimeNear; i++ {
		tw.near[i] = newBucket()
	}

	for i :=0; i < 4; i++ {
		for j :=0; j < TimeLevel; j++ {
			tw.t[i][j] = newBucket()
		}
	}
	return tw
}

func (tw *TimeWheel) add(t *timer) bool {
	time := t.expiration
	currentTime := atomic.LoadUint32(&tw.time)
	if time <= currentTime {
		return false
	}

	if (time | TimeNearMask) == (currentTime | TimeNearMask) {
		tw.near[time&TimeNearMask].Add(t)
	}else {
		i := 0
		mask := TimeNear << TimeNearShift
		for i=0; i < 3; i ++ {
			if (time | uint32(mask - 1)) == (currentTime | uint32(mask - 1)) {
				break
			}
			mask <<= TimeLevelShift
		}

		tw.t[i][((time>>(TimeNearShift + i*TimeLevelShift)) & TimeLevelMask)].Add(t)
	}
	return true
}

func (tw *TimeWheel) addOrRun(t *timer)  {
	if !tw.add(t) {
		workerID := int64(t.expiration) % tw.WorkPool.WorkerPoolSize
		//将请求消息发送给任务队列
		tw.WorkPool.TaskQueue[workerID] <- t.f
	}
}

func (tw *TimeWheel) moveList(level, idx int)  {
	current := tw.t[level][idx]
	current.Flush(tw.addOrRun)
}

func (tw *TimeWheel) shift()  {
	mask := TimeNear
	ct := atomic.AddUint32(&tw.time, 1)
	if ct == 0 {
		tw.moveList(3, 0)
	}else {
		time := ct >> TimeNearShift

		i := 0
		for (ct & uint32(mask-1)) == 0{
			idx := time & TimeLevelMask
			if idx != 0 {
				tw.moveList(i, int(idx))
				break
			}

			mask <<= TimeLevelShift
			time >>= TimeLevelShift
			i++
		}
	}
}

func (tw *TimeWheel) execute()  {
	idx := tw.time & TimeNearMask
	tw.near[idx].Flush(tw.addOrRun)
}

func (tw *TimeWheel) update()  {
	tw.execute()
	tw.shift()
	tw.execute()
}

func (tw *TimeWheel) Start()  {
	tw.ticker = time.NewTicker(tw.tick)
	tw.WorkPool.StartWorkerPool()

	go func() {
		for  {
			select {
			case <- tw.ticker.C:
				tw.update()
			case <- tw.exit:
				return
			}
		}
	}()
}

func (tw *TimeWheel) Stop()  {
	flag := atomic.LoadUint32(&tw.exitFlag)
	if flag != 0 {
		return
	}

	atomic.StoreUint32(&tw.exitFlag, 1)
	close(tw.exit)
}

func (tw *TimeWheel) afterFunc(expiration time.Duration, f func()) {
	time := atomic.LoadUint32(&tw.time)
	tw.addOrRun(&timer{
		expiration: uint32(expiration / tw.tick) + time,
		f:          f,
	})
}

func TimeOut(expire time.Duration, f func()) {
	TimingWheel.afterFunc(expire, f)
}

func StopTimer()  {
	TimingWheel.Stop()
}

package main

type Job func()

type WorkPool struct {
	WorkerPoolSize int64
	MaxTaskPerWorker int64
	TaskQueue []chan Job
}

func NewWorkPool(poolSize, maxTaskSize int64) *WorkPool {
	return &WorkPool{
		WorkerPoolSize: poolSize,
		MaxTaskPerWorker: maxTaskSize,
		TaskQueue: make([]chan Job, poolSize),
	}
}

//StartOneWorker 启动一个Worker工作流程
func (wp *WorkPool) StartOneWorker(workerID int, taskQueue chan Job) {
	//不断的等待队列中的消息
	for {
		select {
		//有消息则取出队列的Request，并执行绑定的业务方法
		case job := <-taskQueue:
			_ = workerID
			job()
		}
	}
}

func (wp *WorkPool) StartWorkerPool() {
	//遍历需要启动worker的数量，依此启动
	for i := 0; i < int(wp.WorkerPoolSize); i++ {
		//一个worker被启动
		//给当前worker对应的任务队列开辟空间
		wp.TaskQueue[i] = make(chan Job,  wp.MaxTaskPerWorker)
		//启动当前Worker，阻塞的等待对应的任务队列是否有消息传递进来
		go wp.StartOneWorker(i, wp.TaskQueue[i])
	}
}

package main

import (
	"fmt"
	"testing"
	"time"
)

func PRINT()  {
	fmt.Println("12312312312")
}

func TestTimeWheel_Start(t *testing.T) {
	TimeOut(1 * time.Second, func() {
		fmt.Println("12312313123")
	})
	select{}
}