Go中定时器实现原理及源码解析

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

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)
...

const timersLen = 64

var timers [timersLen]struct {
	timersBucket
	// padding， 防止false sharing 
	pad [sys.CacheLineSize - unsafe.Sizeof(timersBucket{})%sys.CacheLineSize]byte
}
// 获取 P 对应的 Bucket
func (t *timer) assignBucket() *timersBucket {
	id := uint8(getg().m.p.ptr().id) % timersLen
	t.tb = &timers[id].timersBucket
	return t.tb
}

type timersBucket struct {
	lock         mutex
	gp           *g
	created      bool
	sleeping     bool
	rescheduling bool
	sleepUntil   int64
	waitnote     note
    // timer 列表
	t            []*timer
}

type p struct {
	... 
	// 互斥锁
	timersLock mutex
	// 存储计时器的最小四叉堆
	timers []*timer
	// 计时器数量
	numTimers uint32
	// 处于 timerModifiedEarlier 状态的计时器数量
	adjustTimers uint32
	// 处于 timerDeleted 状态的计时器数量
	deletedTimers uint32
	...
}

func main() {
	//初始化定时器
	t := time.NewTimer(2 * time.Second)
	//当前时间
	now := time.Now()
	fmt.Printf("Now time : %v.\\n", now)

	expire := <- t.C
	fmt.Printf("Expiration time: %v.\\n", expire)
}

func main() { 
	ch1 := make(chan int, 1)
	select {
	case e1 := <-ch1:
		//如果ch1通道成功读取数据，则执行该case处理语句
		fmt.Printf("1th case is selected. e1=%v",e1)
	case <- time.After(2 * time.Second):
		fmt.Println("Timed out")
	}
}

func main() {
	f := func(){
		fmt.Printf("Expiration time : %v.\\n", time.Now())
	}
	time.AfterFunc(1*time.Second, f)
 	time.Sleep(2 * time.Second)
}

type Timer struct {
    C <-chan Time
    r runtimeTimer
}

func NewTimer(d Duration) *Timer {
	// 初始化一个channel，用于返回
	c := make(chan Time, 1)
	t := &Timer{
		C: c,
		r: runtimeTimer{
			when: when(d),
			f:    sendTime,
			arg:  c,
		},
	}
	// 调用runtime.time的startTimer方法
	startTimer(&t.r)
	return t
}

func startTimer(*runtimeTimer)

sleep.go:94     0xd8ea09        e872c7faff              call $time.startTimer

func startTimer(t *timer) {
	...
	addtimer(t)
}

type timer struct {
	// 对应处理器P的指针
	pp puintptr 
	// 定时器被唤醒的时间
	when   int64
    // 唤醒的间隔时间
	period int64
	// 唤醒时被调用的函数
	f      func(interface{}, uintptr)
	// 被调用的函数的参数
	arg    interface{} 
	seq    uintptr
	// 处于timerModifiedXX状态时用于设置when字段 
	nextwhen int64 
	// 定时器的状态
	status uint32
}

const (
	// 初始化状态
	timerNoStatus = iota

	// 等待被调用
	// timer 已在 P 的列表中
	timerWaiting

	// 表示 timer 在运行中 
	timerRunning

	// timer 已被删除 
	timerDeleted

	// timer 正在被移除 
	timerRemoving

	// timer 已被移除，并停止运行 
	timerRemoved

	// timer 被修改了 
	timerModifying
 
  	// 被修改到了更早的时间 
	timerModifiedEarlier 
  
  	// 被修改到了更晚的时间
	timerModifiedLater
 
  // 已经被修改，并且正在被移动
	timerMoving
)

func addtimer(t *timer) {
	// 定时器被唤醒的时间的时间不能为负数
	if t.when < 0 {
		t.when = maxWhen
	}
	// 状态必须为初始化
	if t.status != timerNoStatus {
		throw("addtimer called with initialized timer")
	}
	// 设置为等待调度
	t.status = timerWaiting

	when := t.when
	// 获取当前 P
	pp := getg().m.p.ptr()
	lock(&pp.timersLock)
	// 清理 P 的 timer 列表头中的 timer
	cleantimers(pp)
	// 将 timer 加入到 P 的最小堆中
	doaddtimer(pp, t)
	unlock(&pp.timersLock)
	// 唤醒 netpoller 中休眠的线程
	wakeNetPoller(when)
}

func cleantimers(pp *p) {
	gp := getg()
	for {
		// 调度器列表为空，直接返回
		if len(pp.timers) == 0 {
			return
		}
		// 如果当前 G 被抢占了，直接返回
		if gp.preemptStop {
			return
		}
		// 获取第一个 timer
		t := pp.timers[0]
		if t.pp.ptr() != pp {
			throw("cleantimers: bad p")
		}
		switch s := atomic.Load(&t.status); s {
		case timerDeleted:
			// 设置 timer 的状态
			if !atomic.Cas(&t.status, s, timerRemoving) {
				continue
			}
			// 删除第一个 timer
			dodeltimer0(pp)
			// 删除完毕后重置状态为 timerRemoved
			if !atomic.Cas(&t.status, timerRemoving, timerRemoved) {
				badTimer()
			}
			atomic.Xadd(&pp.deletedTimers, -1)
		// timer 被修改到了更早或更晚的时间
		case timerModifiedEarlier, timerModifiedLater:
			// 将 timer 状态设置为 timerMoving
			if !atomic.Cas(&t.status, s, timerMoving) {
				continue
			}
			// 重新设置 when 字段
			t.when = t.nextwhen
			// 在列表中删除后重新加入 
			dodeltimer0(pp)
			doaddtimer(pp, t)
			if s == timerModifiedEarlier {
				atomic.Xadd(&pp.adjustTimers, -1)
			}
			// 设置状态为 timerWaiting
			if !atomic.Cas(&t.status, timerMoving, timerWaiting) {
				badTimer()
			}
		default: 
			return
		}
	}
}
展开 

func doaddtimer(pp *p, t *timer) { 
	// Timers 依赖于 netpoller
	// 所以如果 netpoller 没有启动，需要启动一下
	if netpollInited == 0 {
		netpollGenericInit()
	}
	// 校验是否早已在 timer 列表中
	if t.pp != 0 {
		throw("doaddtimer: P already set in timer")
	}
	// 设置 timer 与 P 的关联
	t.pp.set(pp)
	i := len(pp.timers)
	// 将 timer 加入到 P 的 timer 列表中
	pp.timers = append(pp.timers, t)
	// 维护 timer 在 最小堆中的位置
	siftupTimer(pp.timers, i)
	// 如果 timer 是列表中头节点，需要设置一下 timer0When
	if t == pp.timers[0] {
		atomic.Store64(&pp.timer0When, uint64(t.when))
	}
	atomic.Xadd(&pp.numTimers, 1)
}

func wakeNetPoller(when int64) {
	if atomic.Load64(&sched.lastpoll) == 0 {  
		pollerPollUntil := int64(atomic.Load64(&sched.pollUntil))
		// 如果计时器的触发时间小于netpoller的下一次轮询时间
		if pollerPollUntil == 0 || pollerPollUntil > when {
			// 向netpollBreakWr里面写入数据，立即中断netpoll
			netpollBreak()
		}
	}
}

func netpollBreak() {
	if atomic.Cas(&netpollWakeSig, 0, 1) {
		for {
			var b byte
            // 向 netpollBreakWr 里面写入数据
			n := write(netpollBreakWr, unsafe.Pointer(&b), 1)
			if n == 1 {
				break
			}
			if n == -_EINTR {
				continue
			}
			if n == -_EAGAIN {
				return
			}
			println("runtime: netpollBreak write failed with", -n)
			throw("runtime: netpollBreak write failed")
		}
	}
}

func modtimer(t *timer, when, period int64, f func(interface{}, uintptr), arg interface{}, seq uintptr) bool {
	if when < 0 {
		when = maxWhen
	} 
	status := uint32(timerNoStatus)
	wasRemoved := false
	var pending bool
	var mp *m
loop:
	for {
		// 修改 timer 状态
		switch status = atomic.Load(&t.status); status {
		...
	} 
	t.period = period
	t.f = f
	t.arg = arg
	t.seq = seq
	// 如果 timer 已被删除，那么需要重新添加到 timer 列表中
	if wasRemoved {
		t.when = when
		pp := getg().m.p.ptr()
		lock(&pp.timersLock)
		doaddtimer(pp, t)
		unlock(&pp.timersLock)
		if !atomic.Cas(&t.status, timerModifying, timerWaiting) {
			badTimer()
		}
		releasem(mp)
		wakeNetPoller(when)
	} else {
		
		t.nextwhen = when

		newStatus := uint32(timerModifiedLater)
		// 如果修改后的时间小于修改前的时间，将状态设置为 timerModifiedEarlier
		if when < t.when {
			newStatus = timerModifiedEarlier
		} 
		... 
		if !atomic.Cas(&t.status, timerModifying, newStatus) {
			badTimer()
		}
		releasem(mp)

		// 如果修改时间提前，那么触发 netpoll 中断
		if newStatus == timerModifiedEarlier {
			wakeNetPoller(when)
		}
	}

	return pending
}
展开 

func runtimer(pp *p, now int64) int64 {
	for {
		// 获取最小堆的第一个元素
		t := pp.timers[0]
		if t.pp.ptr() != pp {
			throw("runtimer: bad p")
		}
		// 获取 timer 状态
		switch s := atomic.Load(&t.status); s {
		// timerWaiting
		case timerWaiting:
			// 还没到时间，返回下次执行时间
			if t.when > now {
				// Not ready to run.
				return t.when
			}
			// 修改状态为 timerRunning
			if !atomic.Cas(&t.status, s, timerRunning) {
				continue
			}
			// 运行该 timer
			runOneTimer(pp, t, now)
			return 0
		// timerDeleted
		case timerDeleted:
			if !atomic.Cas(&t.status, s, timerRemoving) {
				continue
			}
			// 删除最小堆的第一个 timer
			dodeltimer0(pp)
			if !atomic.Cas(&t.status, timerRemoving, timerRemoved) {
				badTimer()
			}
			atomic.Xadd(&pp.deletedTimers, -1)
			if len(pp.timers) == 0 {
				return -1
			}
		// 需要重新移动位置的 timer
		case timerModifiedEarlier, timerModifiedLater:
			if !atomic.Cas(&t.status, s, timerMoving) {
				continue
			}
			t.when = t.nextwhen
			// 删除最小堆的第一个 timer
			dodeltimer0(pp)
			// 将该 timer 重新添加到最小堆
			doaddtimer(pp, t)
			if s == timerModifiedEarlier {
				atomic.Xadd(&pp.adjustTimers, -1)
			}
			if !atomic.Cas(&t.status, timerMoving, timerWaiting) {
				badTimer()
			}

		case timerModifying: 
			osyield()

		case timerNoStatus, timerRemoved: 
			badTimer()
		case timerRunning, timerRemoving, timerMoving: 
			badTimer()
		default:
			badTimer()
		}
	}
}
展开 

func runOneTimer(pp *p, t *timer, now int64) {
	... 
	// 需要被执行的函数
	f := t.f
	// 被执行函数的参数
	arg := t.arg
	seq := t.seq
	// 表示该 timer 为 ticker，需要再次触发
	if t.period > 0 { 
		// 放入堆中并调整触发时间
		delta := t.when - now
		t.when += t.period * (1 + -delta/t.period)
		siftdownTimer(pp.timers, 0)
		if !atomic.Cas(&t.status, timerRunning, timerWaiting) {
			badTimer()
		}
		updateTimer0When(pp)
	// 一次性 timer
	} else {
		// 删除该 timer.
		dodeltimer0(pp)
		if !atomic.Cas(&t.status, timerRunning, timerNoStatus) {
			badTimer()
		}
	}  
	unlock(&pp.timersLock)
	// 运行该函数
	f(arg, seq)
	lock(&pp.timersLock)
	...
}

func schedule() {
	_g_ := getg()
	...
top:
	pp := _g_.m.p.ptr()
	...
	// 检查是否有可执行 timer 并执行
	checkTimers(pp, 0) 
	var gp *g
	...
	if gp == nil {
		gp, inheritTime = findrunnable() // blocks until work is available
	}
	... 
	execute(gp, inheritTime)
}

func checkTimers(pp *p, now int64) (rnow, pollUntil int64, ran bool) { 
	// 如果没有需要调整的 timer
	if atomic.Load(&pp.adjustTimers) == 0 {
		// 获取 timer0 的执行时间 
		next := int64(atomic.Load64(&pp.timer0When))
		if next == 0 {
			return now, 0, false
		}
		if now == 0 {
			now = nanotime()
		}
		// 下次执行大于当前时间，
		if now < next { 
			// 需要删除的 timer 个数小于 timer列表个数的4分之1，直接返回
			if pp != getg().m.p.ptr() || int(atomic.Load(&pp.deletedTimers)) <= int(atomic.Load(&pp.numTimers)/4) {
				return now, next, false
			}
		}
	} 
	lock(&pp.timersLock)
	// 进行调整 timer
	adjusttimers(pp) 
	rnow = now
	if len(pp.timers) > 0 {
		if rnow == 0 {
			rnow = nanotime()
		}
		for len(pp.timers) > 0 { 
			// 查找堆中是否存在需要执行的 timer
			if tw := runtimer(pp, rnow); tw != 0 {
				if tw > 0 {
					pollUntil = tw
				}
				break
			}
			ran = true
		}
	}

	// 如果需要删除的 timer 超过了 timer 列表数量的四分之一，那么清理需要删除的 timer
	if pp == getg().m.p.ptr() && int(atomic.Load(&pp.deletedTimers)) > len(pp.timers)/4 {
		clearDeletedTimers(pp)
	}

	unlock(&pp.timersLock)

	return rnow, pollUntil, ran
}
展开 

func findrunnable() (gp *g, inheritTime bool) {
	_g_ := getg()
top:
	_p_ := _g_.m.p.ptr()
	...
	// 检查 P 中可执行的 timer
	now, pollUntil, _ := checkTimers(_p_, 0)
	... 
    // 如果 netpoll 已被初始化，并且 Waiters 大于零，并且 lastpoll 不为0
	if netpollinited() && atomic.Load(&netpollWaiters) > 0 && atomic.Load64(&sched.lastpoll) != 0 {
		// 尝试从netpoller获取Glist
		if list := netpoll(0); !list.empty() { // 无阻塞
			gp := list.pop()
			//将其余队列放入 P 的可运行G队列
			injectglist(&list)
			casgstatus(gp, _Gwaiting, _Grunnable)
			if trace.enabled {
				traceGoUnpark(gp, 0)
			}
			return gp, false
		}
	}
	...
	// 开始窃取
	for i := 0; i < 4; i++ {
		for enum := stealOrder.start(fastrand()); !enum.done(); enum.next() {
			if sched.gcwaiting != 0 {
				goto top
			}
			// 如果 i>2 表示如果其他 P 运行队列中没有 G ，将要从其他队列的 runnext 中获取
			stealRunNextG := i > 2 // first look for ready queues with more than 1 g
			// 随机获取一个 P
			p2 := allp[enum.position()]
			if _p_ == p2 {
				continue
			}
			// 从其他 P 的运行队列中获取一般的 G 到当前队列中
			if gp := runqsteal(_p_, p2, stealRunNextG); gp != nil {
				return gp, false
			}

			// 如果运行队列中没有 G，那么从 timers 中获取可执行的 timer
			if i > 2 || (i > 1 && shouldStealTimers(p2)) {
				// ran 为 true 表示有执行过 timer
				tnow, w, ran := checkTimers(p2, now)
				now = tnow
				if w != 0 && (pollUntil == 0 || w < pollUntil) {
					pollUntil = w
				}
				if ran {
					// 因为已经运行过 timer 了，说不定已经有准备就绪的 G 了
					// 再次检查本地队列尝试获取 G
					if gp, inheritTime := runqget(_p_); gp != nil {
						return gp, inheritTime
					}
					ranTimer = true
				}
			}
		}
	} 

	if ranTimer {
		// 执行完一个 timer 后可能存在已经就绪的 G
		goto top
	}

stop:  
	...
	delta := int64(-1)
	if pollUntil != 0 {
		// checkTimers ensures that polluntil > now.
		delta = pollUntil - now
	}
	...
	// poll network
	// 休眠前再次检查 poll 网络
	if netpollinited() && (atomic.Load(&netpollWaiters) > 0 || pollUntil != 0) && atomic.Xchg64(&sched.lastpoll, 0) != 0 {
		...
		list := netpoll(delta) // 阻塞调用
		lock(&sched.lock)
		_p_ = pidleget()
		unlock(&sched.lock)
		if _p_ == nil {
			injectglist(&list)
		} else {
			acquirep(_p_)
			if !list.empty() {
				gp := list.pop()
				injectglist(&list)
				casgstatus(gp, _Gwaiting, _Grunnable)
				if trace.enabled {
					traceGoUnpark(gp, 0)
				}
				return gp, false
			}
			 
			goto top
		}
	} else if pollUntil != 0 && netpollinited() {
		pollerPollUntil := int64(atomic.Load64(&sched.pollUntil))
		if pollerPollUntil == 0 || pollerPollUntil > pollUntil {
			netpollBreak()
		}
	}
	// 休眠当前 M
	stopm()
	goto top
}
展开 

func sysmon() {
	...
	for {
		...
		now := nanotime()
		// 返回下次需要调度 timer 到期时间
		next, _ := timeSleepUntil()
		...  
		// 如果超过 10ms 没有 poll，则 poll 一下网络
		lastpoll := int64(atomic.Load64(&sched.lastpoll))
		if netpollinited() && lastpoll != 0 && lastpoll+10*1000*1000 < now {
			atomic.Cas64(&sched.lastpoll, uint64(lastpoll), uint64(now))
			list := netpoll(0) // 非阻塞，返回 G 列表
			// G 列表不为空
			if !list.empty() { 
				incidlelocked(-1)
				// 将获取到的 G 列表插入到空闲的 P 中或全局列表中
				injectglist(&list)
				incidlelocked(1)
			}
		}
		// 如果有 timer 到期
		if next < now {
			// 启动新的 M 处理 timer
			startm(nil, false)
		}
		...
	}
}

func TestNetpollBreak(t *testing.T) {
	if runtime.GOMAXPROCS(0) == 1 {
		t.Skip("skipping: GOMAXPROCS=1")
	}
	// 初始化 netpoll
	runtime.NetpollGenericInit()

	start := time.Now()
	c := make(chan bool, 2)
	go func() {
		c <- true
		// netpoll 等待时间
		runtime.Netpoll(10 * time.Second.Nanoseconds())
		c <- true
	}()
	<-c 
loop:
	for {
		runtime.Usleep(100)
		// 中断netpoll 等待
		runtime.NetpollBreak()
		runtime.NetpollBreak()
		select {
		case <-c:
			break loop
		default:
		}
	}
	if dur := time.Since(start); dur > 5*time.Second {
		t.Errorf("netpollBreak did not interrupt netpoll: slept for: %v", dur)
	}
}

func netpoll(delay int64) gList {
	if epfd == -1 {
		return gList{}
	}
	var waitms int32
	// 因为传入delay单位是纳秒，下面将纳秒转换成毫秒
	if delay < 0 {
		waitms = -1
	} else if delay == 0 {
		waitms = 0
	} else if delay < 1e6 {
		waitms = 1
	} else if delay < 1e15 {
		waitms = int32(delay / 1e6)
	} else { 
		waitms = 1e9
	}
	var events [128]epollevent
retry:
	// 等待文件描述符转换成可读或者可写
	n := epollwait(epfd, &events[0], int32(len(events)), waitms)
	// 返回负值，那么重新调用epollwait进行等待
	if n < 0 {
		...
		goto retry
	}
	var toRun gList
	for i := int32(0); i < n; i++ {
		ev := &events[i]
		if ev.events == 0 {
			continue
		}
		// 如果是 NetpollBreak 中断的，那么执行 continue 跳过
		if *(**uintptr)(unsafe.Pointer(&ev.data)) == &netpollBreakRd {
			if ev.events != _EPOLLIN {
				println("runtime: netpoll: break fd ready for", ev.events)
				throw("runtime: netpoll: break fd ready for something unexpected")
			}
			if delay != 0 { 
				var tmp [16]byte
				read(int32(netpollBreakRd), noescape(unsafe.Pointer(&tmp[0])), int32(len(tmp)))
				atomic.Store(&netpollWakeSig, 0)
			}
			continue
		} 
		...
	}
	return toRun
}
展开 

func findrunnable() (gp *g, inheritTime bool) {
	...
	delta := int64(-1)
	if pollUntil != 0 {
		// checkTimers ensures that polluntil > now.
		delta = pollUntil - now
	}
	...
	// poll network
	// 休眠前再次检查 poll 网络
	if netpollinited() && (atomic.Load(&netpollWaiters) > 0 || pollUntil != 0) && atomic.Xchg64(&sched.lastpoll, 0) != 0 {
		...
		// 阻塞调用
		list := netpoll(delta) 
		
	}  
	...
	// 休眠当前 M
	stopm()
	goto top
}