GoLang中的timer定时器实现原理分析
raoxiaoya 人气:0// NewTimer creates a new Timer that will send // the current time on its channel after at least duration d. func NewTimer(d Duration) *Timer { c := make(chan Time, 1) t := &Timer{ C: c, r: runtimeTimer{ when: when(d), f: sendTime, arg: c, }, } startTimer(&t.r) return t } // The Timer type represents a single event. // When the Timer expires, the current time will be sent on C, // unless the Timer was created by AfterFunc. // A Timer must be created with NewTimer or AfterFunc. type Timer struct { C <-chan Time r runtimeTimer }
func NewTicker(d Duration) *Ticker { if d <= 0 { panic(errors.New("non-positive interval for NewTicker")) } // Give the channel a 1-element time buffer. // If the client falls behind while reading, we drop ticks // on the floor until the client catches up. c := make(chan Time, 1) t := &Ticker{ C: c, r: runtimeTimer{ when: when(d), period: int64(d), f: sendTime, arg: c, }, } startTimer(&t.r) return t } type Ticker struct { C <-chan Time // The channel on which the ticks are delivered. r runtimeTimer }
ticker 跟 timer 的初始化过程差不多,但是 ticker 比 timer 多了一个 period 参数,意为间隔的意思。
// Interface to timers implemented in package runtime. // Must be in sync with ../runtime/time.go:/^type timer type runtimeTimer struct { pp uintptr when int64 //触发时间 period int64 //执行周期性任务的时间间隔 f func(any, uintptr) // 执行的回调函数,NOTE: must not be closure arg any //执行任务的参数 seq uintptr //回调函数的参数,该参数仅在 netpoll 的应用场景下使用 nextwhen int64 //如果是周期性任务,下次执行任务时间 status uint32 //状态 } // sendTime does a non-blocking send of the current time on c. func sendTime(c any, seq uintptr) { select { case c.(chan Time) <- Now(): default: } }
sendTime 采用非阻塞的形式,意为,不管是否存在接收方,此定时器一旦到时间了就要触发掉。
// runtime/runtime2.go type p struct { ..... // The when field of the first entry on the timer heap. // This is updated using atomic functions. // This is 0 if the timer heap is empty. // 堆顶元素什么时候执行 timer0When uint64 // The earliest known nextwhen field of a timer with // timerModifiedEarlier status. Because the timer may have been // modified again, there need not be any timer with this value. // This is updated using atomic functions. // This is 0 if there are no timerModifiedEarlier timers. // 如果有timer修改为更早执行时间了,将会将执行时间更新到更早时间 timerModifiedEarliest uint64 // Lock for timers. We normally access the timers while running // on this P, but the scheduler can also do it from a different P. // 操作timer的互斥锁 timersLock mutex // Actions to take at some time. This is used to implement the // standard library's time package. // Must hold timersLock to access. //该p 上的所有timer,必须加锁去操作这个字段,因为不同的p 操作这个字段会有竞争关系 timers []*timer // Number of timers in P's heap. // Modified using atomic instructions. //p 堆上所有的timer数 numTimers uint32 // Number of timerDeleted timers in P's heap. // Modified using atomic instructions. //被标记为删除的timer,要么是我们调用stop,要么是timer 自己触发后过期导致的删除 deletedTimers uint32 } // runtime/time.go type timer struct { // If this timer is on a heap, which P's heap it is on. // puintptr rather than *p to match uintptr in the versions // of this struct defined in other packages. pp puintptr // Timer wakes up at when, and then at when+period, ... (period > 0 only) // each time calling f(arg, now) in the timer goroutine, so f must be // a well-behaved function and not block. // // when must be positive on an active timer. when int64 period int64 f func(any, uintptr) arg any seq uintptr // What to set the when field to in timerModifiedXX status. nextwhen int64 // The status field holds one of the values below. status uint32 } // startTimer adds t to the timer heap. //go:linkname startTimer time.startTimer func startTimer(t *timer) { if raceenabled { racerelease(unsafe.Pointer(t)) } addtimer(t) } // stopTimer stops a timer. // It reports whether t was stopped before being run. //go:linkname stopTimer time.stopTimer func stopTimer(t *timer) bool { return deltimer(t) } // addtimer adds a timer to the current P. // This should only be called with a newly created timer. // That avoids the risk of changing the when field of a timer in some P's heap, // which could cause the heap to become unsorted. func addtimer(t *timer) { // when must be positive. A negative value will cause runtimer to // overflow during its delta calculation and never expire other runtime // timers. Zero will cause checkTimers to fail to notice the timer. if t.when <= 0 { throw("timer when must be positive") } if t.period < 0 { throw("timer period must be non-negative") } if t.status != timerNoStatus { throw("addtimer called with initialized timer") } t.status = timerWaiting when := t.when // Disable preemption while using pp to avoid changing another P's heap. // 如果M在此之后被别的P抢占了,那么后续操作的就是别的P上的timers,这是不允许的 mp := acquirem() pp := getg().m.p.ptr() lock(&pp.timersLock) cleantimers(pp) // 清理掉已经过期的timer,以提高添加和删除timer的效率。 doaddtimer(pp, t) // 执行添加操作 unlock(&pp.timersLock) // 调用 wakeNetPoller 方法,唤醒网络轮询器,检查计时器被唤醒的时间(when)是 // 否在当前轮询预期运行的时间(pollerPollUntil)内,若是唤醒。 // 有的定时器是伴随着网络轮训器的,比如设置的 i/o timeout // This can have a spurious wakeup but should never miss a wakeup // 宁愿出现错误的唤醒,也不能漏掉一个唤醒 wakeNetPoller(when) releasem(mp) } // 将0位置的timer与下面的子节点比较,如果比子节点大则下移。子节点i*4 + 1,i*4 + 2,i*4 + 3,i*4 + 4 siftdownTimer(pp.timers, 0) // 将i位置的timer与上面的父节点比较,如果比父节点小则上移。父节点是(i - 1) / 4 siftupTimer(pp.timers, i)
timer
存储在P
中的 timers []*timer
成员属性上。timers看起来是一个切片,但是它是按照runtimeTimer.when
这个数值排序的小顶堆四叉树,触发时间越早越排在前面。
整体来讲就是父节点一定比其子节点小,子节点之间没有任何关系和大小的要求。
关于acquirem
和releasem
//go:nosplit func acquirem() *m { _g_ := getg() _g_.m.locks++ return _g_.m } //go:nosplit func releasem(mp *m) { _g_ := getg() mp.locks-- if mp.locks == 0 && _g_.preempt { // restore the preemption request in case we've cleared it in newstack _g_.stackguard0 = stackPreempt } }
acquirem函数获取当前M,并禁止M被抢占,因为M被抢占时的判断如下
//C:\Go\src\runtime\preempt.go +287 func canPreemptM(mp *m) bool { return mp.locks == 0 && mp.mallocing == 0 && mp.preemptoff == "" && mp.p.ptr().status == _Prunning }
- 运行时没有禁止抢占(
m.locks == 0
) - 运行时没有在执行内存分配(
m.mallocing == 0
) - 运行时没有关闭抢占机制(
m.preemptoff == ""
) - M 与 P 绑定且没有进入系统调用(
p.status == _Prunning
)
timers的触发
// runtime/proc.go func checkTimers(pp *p, now int64) (rnow, pollUntil int64, ran bool) // runtime/time.go func runtimer(pp *p, now int64) int64 func runOneTimer(pp *p, t *timer, now int64)
runtime/time.go
文件中提供了checkTimers/runtimer/runOneTimer
三个方法。checkTimers方法中,如果当前p的timers长度不为0,就不断地调用runtimers。runtimes会根据堆顶的timer的状态判断其能否执行,如果可以执行就调用runOneTimer实际执行。
触发定时器的途径有两个
- 通过调度器在调度时进行计时器的触发,findrunnable, schedule, stealWork。
- 通过系统监控检查并触发计时器(到期未执行),sysmon。
调度器的触发一共分两种情况,一种是在调度循环的时候调用 checkTimers 方法进行计时器的触发。另外一种是当前处理器 P 没有可执行的 Timer,且没有可执行的 G。那么按照调度模型,就会去窃取其他计时器和 G。
即使是通过每次调度器调度和窃取的时候触发,但毕竟是具有一定的随机和不确定性,因此系统监控触发依然是一个兜底保障,在 Go 语言中 runtime.sysmon 方法承担了这一个责任,存在触发计时器的逻辑,在每次进行系统监控时,都会在流程上调用 timeSleepUntil 方法去获取下一个计时器应触发的时间,以及保存该计时器已打开的计时器堆的 P。
在获取完毕后会马上检查当前是否存在 GC,若是正在 STW 则获取调度互斥锁。若发现下一个计时器的触发时间已经过去,则重新调用 timeSleepUntil 获取下一个计时器的时间和相应 P 的地址。检查 sched.sysmonlock 所花费的时间是否超过 50μs。若是,则有可能前面所获取的下一个计时器触发时间已过期,因此重新调用 timeSleepUntil 方法再次获取。如果发现超过 10ms 的时间没有进行 netpoll 网络轮询,则主动调用 netpoll 方法触发轮询。同时如果存在不可抢占的处理器 P,则调用 startm 方法来运行那些应该运行,但没有在运行的计时器。
加载全部内容