runtime: fix ticker not stopping when Stop races with its callback (#5487)

* runtime: fix ticker not stopping when Stop races with its callback

On the threads and cores schedulers, timer callbacks run concurrently
with user goroutines.

If Stop (or Reset) was called in the window after the node was popped
but before its callback re-added it, removeTimer would not find the
timer in the queue, so it would be re-added to timer anyway.

Track timers whose callback is currently running in a firing list, and
have removeTimer mark a firing timer as stopped so its callback does not
re-add it.

Also make the timers test drain robust: allow a possible in-flight tick
delivered concurrently with Stop to settle before draining the channel.

Signed-off-by: deadprogram <ron@hybridgroup.com>

* runtime: fix Stop/Reset semantics when racing a firing timer callback

Address review feedback on the ticker Stop-race fix. On the threads and
cores schedulers a timer callback runs concurrently with user goroutines,
which left several problems:

- removeTimer reported a firing timer as successfully removed, so
  Stop/Reset could return true even though the callback had already
  started (wrong semantics, notably for AfterFunc). firingTimerStop now
  returns a bool and removeTimer no longer hands the still-firing node
  back to resetTimer.

- The periodic advance (when += period) ran in timerCallback outside the
  scheduler timer lock. Move it into each scheduler's reAddTimer, under
  the lock and after the stopped check, so a concurrent Reset can't have
  its freshly-queued deadline corrupted.

- resetTimer now sets when/period after removeTimer for the same reason.

Add testdata/timer_stop_reset_race.go and TestTimerStopResetRace, which
reproduce the stop-while-firing and reset-while-firing races via the
runtime timer linkname hooks.

Signed-off-by: deadprogram <ron@hybridgroup.com>

* runtime: fix timer Stop/Reset race with firing periodic timers

Signed-off-by: deadprogram <ron@hybridgroup.com>

---------

Signed-off-by: deadprogram <ron@hybridgroup.com>

testdata/timer_stop_reset_race.go

@@ -0,0 +1,83 @@
+package main
+
+import (
+	"time"
+	"unsafe"
+	_ "unsafe"
+)
+
+func main() {
+	testStopWhileFiring()
+	testResetWhileFiring()
+	println("timer stop/reset race tests done")
+}
+
+func testStopWhileFiring() {
+	started := make(chan struct{})
+	release := make(chan struct{})
+
+	tim := newTimerWithCallback(time.Millisecond, 0, func(any, uintptr, int64) {
+		close(started)
+		<-release
+	})
+
+	<-started
+	if stopTimer(tim) {
+		println("fail: Stop returned true for a firing timer")
+	} else {
+		println("Stop returned false for firing timer")
+	}
+	close(release)
+}
+
+func testResetWhileFiring() {
+	started := make(chan struct{})
+	release := make(chan struct{})
+	firedAgain := make(chan struct{}, 1)
+	first := true
+
+	tim := newTimerWithCallback(time.Millisecond, 5*time.Second, func(any, uintptr, int64) {
+		if first {
+			first = false
+			close(started)
+			<-release
+			return
+		}
+		select {
+		case firedAgain <- struct{}{}:
+		default:
+		}
+	})
+
+	<-started
+	resetPeriodicTimer(tim, 20*time.Millisecond, 5*time.Second)
+	close(release)
+	time.Sleep(200 * time.Millisecond)
+	select {
+	case <-firedAgain:
+		println("reset firing timer used reset deadline")
+	default:
+		println("fail: reset firing timer missed reset deadline")
+	}
+	stopTimer(tim)
+}
+
+func newTimerWithCallback(delay, period time.Duration, f func(any, uintptr, int64)) *time.Timer {
+	return newTimer(runtimeNano()+int64(delay), int64(period), f, nil, nil)
+}
+
+func resetPeriodicTimer(tim *time.Timer, delay, period time.Duration) bool {
+	return resetTimer(tim, runtimeNano()+int64(delay), int64(period))
+}
+
+//go:linkname newTimer time.newTimer
+func newTimer(when, period int64, f func(any, uintptr, int64), arg any, cp unsafe.Pointer) *time.Timer
+
+//go:linkname stopTimer time.stopTimer
+func stopTimer(tim *time.Timer) bool
+
+//go:linkname resetTimer time.resetTimer
+func resetTimer(tim *time.Timer, when, period int64) bool
+
+//go:linkname runtimeNano time.runtimeNano
+func runtimeNano() int64

testdata/timer_stop_reset_race.txt

@@ -0,0 +1,3 @@
+Stop returned false for firing timer
+reset firing timer used reset deadline
+timer stop/reset race tests done

testdata/timers.go

@@ -21,7 +21,11 @@ func main() {
 	<-ticker.C
 	println("waited on ticker at 1000ms")
 	ticker.Stop()
-	// Drain any tick that was already queued before Stop took effect.
+	// Ticker.Stop does not drain already-buffered ticks from the channel, and
+	// on a slow CI runner a final tick may be delivered concurrently right as
+	// Stop is called. Give any such in-flight tick a chance to arrive, then
+	// drain the channel before checking that no further ticks are sent.
+	time.Sleep(time.Millisecond * 100)
 	select {
 	case <-ticker.C:
 	default: