// 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 方法来运行那些应该运行，但没有在运行的计时器。
到此这篇关于GoLang中的timer定时器实现原理分析的文章就介绍到这了,更多相关Go timer定时器内容请搜索aitechtogether.com以前的文章或继续浏览下面的相关文章希望大家以后多多支持aitechtogether.com！