|
我们看下 checkTimers 做了什么
func checkTimers(pp *p, now int64) (rnow, pollUntil int64, ran bool) {
if atomic.Load(&pp.adjustTimers) == 0 {// 如果没有需要可调整的，则直接返回最先执行 timer 的时间
next := int64(atomic.Load64(&pp.timer0When))
if next == 0 {
return now, 0, false
}
if now == 0 {
now = nanotime()
}
if now < next { // 表示还没有到执行时间
if pp != getg().m.p.ptr() || int(atomic.Load(&pp.deletedTimers)) <= int(atomic.Load(&pp.numTimers)/4) {
//且要删除的 Timer数量小于 Timer总数的1/4
return now, next, false
}
}
}
lock(&pp.timersLock)
adjusttimers(pp)// 可以看下面的源码解析，当前 p 上的所有 timers 的状态，该删除的删了，该调整的调整
rnow = now
if len(pp.timers) > 0 {
if rnow == 0 {
rnow = nanotime()
}
for len(pp.timers) > 0 {
if tw := runtimer(pp, rnow); tw != 0 { // 通过 runtimer（可以看上面的源码解析） 开始调用
if tw > 0 {
pollUntil = tw
}
break
}
ran = true
}
}
// 如果可删除的 Timers 大于 Timer总数量的1/4，则进行删除（因为上面执行了 runtimer)
if pp == getg().m.p.ptr() && int(atomic.Load(&pp.deletedTimers)) > len(pp.timers)/4 {
clearDeletedTimers(pp)
}
unlock(&pp.timersLock)
return rnow, pollUntil, ran
}
adjusttimers
func adjusttimers(pp *p) {
if len(pp.timers) == 0 {
return
}
if atomic.Load(&pp.adjustTimers) == 0 { // 如果需要调整的 Timer 为 0，则直接返回
...
return
}
var moved []*timer
loop:
for i := 0; i < len(pp.timers); i++ {
t := pp.timers[i]
if t.pp.ptr() != pp {
throw("adjusttimers: bad p")
}
switch s := atomic.Load(&t.status); s {
case timerDeleted: // 这里就是将部分需要删除的 Timer 给清理掉
if atomic.Cas(&t.status, s, timerRemoving) {
dodeltimer(pp, i)
if !atomic.Cas(&t.status, timerRemoving, timerRemoved) {
badTimer()
}
atomic.Xadd(&pp.deletedTimers, -1)
i--
}
case timerModifiedEarlier, timerModifiedLater:
// 把需要调整 Timer 放到 moved 中，然后删除当前堆的数据进行堆调整，后续将 moved 通过 addAdjustedTimers 添加
if atomic.Cas(&t.status, s, timerMoving) {
t.when = t.nextwhen
dodeltimer(pp, i)
moved = append(moved, t)
if s == timerModifiedEarlier {
if n := atomic.Xadd(&pp.adjustTimers, -1); int32(n) <= 0 {
break loop
}
}
i--
}
case timerNoStatus, timerRunning, timerRemoving, timerRemoved, timerMoving:
badTimer()
case timerWaiting:
case timerModifying:
osyield()
i--
default:
badTimer()
}
}
if len(moved) > 0 {
addAdjustedTimers(pp, moved) // 这里就是将需要调整的 timer 重新添加进来
}
...
}
addAdjustedTimers
func addAdjustedTimers(pp *p, moved []*timer) {
for _, t := range moved {
doaddtimer(pp, t)// 上文有源码解析
if !atomic.Cas(&t.status, timerMoving, timerWaiting) {
badTimer()
}
}
}
clearDeletedTimers
func clearDeletedTimers(pp *p) {
cdel := int32(0)
cearlier := int32(0)
to := 0
changedHeap := false
timers := pp.timers
nextTimer:
for _, t := range timers {
for {
switch s := atomic.Load(&t.status); s {
case timerWaiting:
if changedHeap {
timers[to] = t
siftupTimer(timers, to)
}
to++
continue nextTimer
case timerModifiedEarlier, timerModifiedLater: // 将 timer 状态调整成 timeWaiting，将其放至其正确的执行时间位置
if atomic.Cas(&t.status, s, timerMoving) {
t.when = t.nextwhen
timers[to] = t
siftupTimer(timers, to)
to++
changedHeap = true
if !atomic.Cas(&t.status, timerMoving, timerWaiting) {
badTimer()
}
if s == timerModifiedEarlier {
cearlier++
}
continue nextTimer
}
case timerDeleted: // 将 timerDeleted 转变成 timerRemoved，然后从 timers 堆中删掉（在当前函数后面可以看出）
if atomic.Cas(&t.status, s, timerRemoving) {
t.pp = 0
cdel++
if !atomic.Cas(&t.status, timerRemoving, timerRemoved) {
badTimer()
}
changedHeap = true
continue nextTimer
}
case timerModifying:
osyield()
case timerNoStatus, timerRemoved:
badTimer()
case timerRunning, timerRemoving, timerMoving:
badTimer()
default:
badTimer()
}
}
}
// 在这里对于剩余的空间 设置为 nil 操作（垃圾回收方便）
for i := to; i < len(timers); i++ {
timers[i] = nil
}
atomic.Xadd(&pp.deletedTimers, -cdel)
atomic.Xadd(&pp.numTimers, -cdel)
atomic.Xadd(&pp.adjustTimers, -cearlier)
// 在这里进行一次大清理
timers = timers[:to]
pp.timers = timers
updateTimer0When(pp)
...
}
大致执行的情况我们看好了，那我们接下来看 Stop() 的源码部分
| 
发表于 2023-2-8 15:04