runtime: don't let P's timer heap get clogged with deleted timers

Whenever more than 1/4 of the timers on a P's heap are deleted,
remove them from the heap.

Change-Id: Iff63ed3d04e6f33ffc5c834f77f645c52c007e52
Reviewed-on: https://go-review.googlesource.com/c/go/+/214299
Run-TryBot: Ian Lance Taylor <iant@golang.org>
TryBot-Result: Gobot Gobot <gobot@golang.org>
Reviewed-by: Michael Knyszek <mknyszek@google.com>

src/runtime/proc.go

@@ -2643,6 +2643,13 @@ func checkTimers(pp *p, now int64) (rnow, pollUntil int64, ran bool) {
 		}
 	}
 
+	// If this is the local P, and there are a lot of deleted timers,
+	// clear them out. We only do this for the local P to reduce
+	// lock contention on timersLock.
+	if pp == getg().m.p.ptr() && int(atomic.Load(&pp.deletedTimers)) > len(pp.timers)/4 {
+		clearDeletedTimers(pp)
+	}
+
 	unlock(&pp.timersLock)
 
 	return rnow, pollUntil, ran
@@ -4087,6 +4094,7 @@ func (pp *p) destroy() {
 		moveTimers(plocal, pp.timers)
 		pp.timers = nil
 		pp.adjustTimers = 0
+		pp.deletedTimers = 0
 		unlock(&pp.timersLock)
 		unlock(&plocal.timersLock)
 	}

src/runtime/runtime2.go

@@ -650,6 +650,10 @@ type p struct {
 	// such as timerModifying.
 	adjustTimers uint32
 
+	// Number of timerDeleted times in P's heap.
+	// Modified using atomic instructions.
+	deletedTimers uint32
+
 	// Race context used while executing timer functions.
 	timerRaceCtx uintptr
 

src/runtime/time.go

@@ -169,6 +169,10 @@ const (
 // maxWhen is the maximum value for timer's when field.
 const maxWhen = 1<<63 - 1
 
+// verifyTimers can be set to true to add debugging checks that the
+// timer heaps are valid.
+const verifyTimers = false
+
 // Package time APIs.
 // Godoc uses the comments in package time, not these.
 
@@ -295,7 +299,9 @@ func deltimer(t *timer) bool {
 	for {
 		switch s := atomic.Load(&t.status); s {
 		case timerWaiting, timerModifiedLater:
+			tpp := t.pp.ptr()
 			if atomic.Cas(&t.status, s, timerDeleted) {
+				atomic.Xadd(&tpp.deletedTimers, 1)
 				// Timer was not yet run.
 				return true
 			}
@@ -306,6 +312,7 @@ func deltimer(t *timer) bool {
 				if !atomic.Cas(&t.status, timerModifying, timerDeleted) {
 					badTimer()
 				}
+				atomic.Xadd(&tpp.deletedTimers, 1)
 				// Timer was not yet run.
 				return true
 			}
@@ -486,6 +493,7 @@ func resettimer(t *timer, when int64) {
 				return
 			}
 		case timerDeleted:
+			tpp := t.pp.ptr()
 			if atomic.Cas(&t.status, s, timerModifying) {
 				t.nextwhen = when
 				newStatus := uint32(timerModifiedLater)
@@ -496,6 +504,7 @@ func resettimer(t *timer, when int64) {
 				if !atomic.Cas(&t.status, timerModifying, newStatus) {
 					badTimer()
 				}
+				atomic.Xadd(&tpp.deletedTimers, -1)
 				if newStatus == timerModifiedEarlier {
 					wakeNetPoller(when)
 				}
@@ -543,6 +552,7 @@ func cleantimers(pp *p) bool {
 			if !atomic.Cas(&t.status, timerRemoving, timerRemoved) {
 				return false
 			}
+			atomic.Xadd(&pp.deletedTimers, -1)
 		case timerModifiedEarlier, timerModifiedLater:
 			if !atomic.Cas(&t.status, s, timerMoving) {
 				continue
@@ -631,9 +641,13 @@ func adjusttimers(pp *p) {
 		return
 	}
 	if atomic.Load(&pp.adjustTimers) == 0 {
+		if verifyTimers {
+			verifyTimerHeap(pp.timers)
+		}
 		return
 	}
 	var moved []*timer
+loop:
 	for i := 0; i < len(pp.timers); i++ {
 		t := pp.timers[i]
 		if t.pp.ptr() != pp {
@@ -648,6 +662,7 @@ func adjusttimers(pp *p) {
 				if !atomic.Cas(&t.status, timerRemoving, timerRemoved) {
 					badTimer()
 				}
+				atomic.Xadd(&pp.deletedTimers, -1)
 				// Look at this heap position again.
 				i--
 			}
@@ -665,8 +680,7 @@ func adjusttimers(pp *p) {
 				moved = append(moved, t)
 				if s == timerModifiedEarlier {
 					if n := atomic.Xadd(&pp.adjustTimers, -1); int32(n) <= 0 {
-						addAdjustedTimers(pp, moved)
+						break loop
-						return
 					}
 				}
 				// Look at this heap position again.
@@ -688,6 +702,10 @@ func adjusttimers(pp *p) {
 	if len(moved) > 0 {
 		addAdjustedTimers(pp, moved)
 	}
+
+	if verifyTimers {
+		verifyTimerHeap(pp.timers)
+	}
 }
 
 // addAdjustedTimers adds any timers we adjusted in adjusttimers
@@ -762,6 +780,7 @@ func runtimer(pp *p, now int64) int64 {
 			if !atomic.Cas(&t.status, timerRemoving, timerRemoved) {
 				badTimer()
 			}
+			atomic.Xadd(&pp.deletedTimers, -1)
 			if len(pp.timers) == 0 {
 				return -1
 			}
@@ -859,6 +878,107 @@ func runOneTimer(pp *p, t *timer, now int64) {
 	}
 }
 
+// clearDeletedTimers removes all deleted timers from the P's timer heap.
+// This is used to avoid clogging up the heap if the program
+// starts a lot of long-running timers and then stops them.
+// For example, this can happen via context.WithTimeout.
+//
+// This is the only function that walks through the entire timer heap,
+// other than moveTimers which only runs when the world is stopped.
+//
+// The caller must have locked the timers for pp.
+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:
+				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:
+				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:
+				// Loop until modification complete.
+				osyield()
+			case timerNoStatus, timerRemoved:
+				// We should not see these status values in a timer heap.
+				badTimer()
+			case timerRunning, timerRemoving, timerMoving:
+				// Some other P thinks it owns this timer,
+				// which should not happen.
+				badTimer()
+			default:
+				badTimer()
+			}
+		}
+	}
+
+	// Set remaining slots in timers slice to nil,
+	// so that the timer values can be garbage collected.
+	for i := to; i < len(timers); i++ {
+		timers[i] = nil
+	}
+
+	timers = timers[:to]
+	if verifyTimers {
+		verifyTimerHeap(timers)
+	}
+	pp.timers = timers
+	atomic.Xadd(&pp.deletedTimers, -cdel)
+	atomic.Xadd(&pp.adjustTimers, -cearlier)
+}
+
+// verifyTimerHeap verifies that the timer heap is in a valid state.
+// This is only for debugging, and is only called if verifyTimers is true.
+// The caller must have locked the timers.
+func verifyTimerHeap(timers []*timer) {
+	for i, t := range timers {
+		if i == 0 {
+			// First timer has no parent.
+			continue
+		}
+
+		// The heap is 4-ary. See siftupTimer and siftdownTimer.
+		p := (i - 1) / 4
+		if t.when < timers[p].when {
+			print("bad timer heap at ", i, ": ", p, ": ", timers[p].when, ", ", i, ": ", t.when, "\n")
+			throw("bad timer heap")
+		}
+	}
+}
+
 func timejump() *p {
 	if faketime == 0 {
 		return nil