runtime: add new modtimer function

This adds a new field to P, adjustTimers, that tells the P that one of
its existing timers was modified to be earlier, and that it therefore
needs to resort them.

Updates #27707

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

diff --git a/src/runtime/proc.go b/src/runtime/proc.go
index 09ef784a0491fc1778083f37fc6b23c046d8a272..edff454491f614ef1fb98aa6281b5248255cb0f6 100644 (file)
--- a/src/runtime/proc.go
+++ b/src/runtime/proc.go
@@ -4122,6 +4122,7 @@ func (pp *p) destroy() {
                // The world is stopped so we don't need to hold timersLock.
                moveTimers(plocal, pp.timers)
                pp.timers = nil
+               pp.adjustTimers = 0
        }
        // If there's a background worker, make it runnable and put
        // it on the global queue so it can clean itself up.

diff --git a/src/runtime/runtime2.go b/src/runtime/runtime2.go
index f44cd2fb1436c316107a5cb7688ac2eb6568d4f7..b57ae75bafa40b4d0edf9587b5d9248e1dafbde8 100644 (file)
--- a/src/runtime/runtime2.go
+++ b/src/runtime/runtime2.go
@@ -607,6 +607,12 @@ type p struct {
        // Must hold timersLock to access.
        timers []*timer
 
+       // Number of timerModifiedEarlier timers on P's heap.
+       // This should only be modified while holding timersLock,
+       // or while the timer status is in a transient state
+       // such as timerModifying.
+       adjustTimers uint32
+
        pad cpu.CacheLinePad
 }
 

diff --git a/src/runtime/time.go b/src/runtime/time.go
index f2dd40e6b4ae3a698aeced66555fc59dc36c69ca..54cbb0b6e469a4f23e53e2abcaff8ad1dfc529a6 100644 (file)
--- a/src/runtime/time.go
+++ b/src/runtime/time.go
@@ -131,6 +131,16 @@ type timersBucket struct {
 //   timerRunning    -> wait until status changes
 //   timerMoving     -> wait until status changes
 //   timerModifying  -> panic: concurrent deltimer/modtimer calls
+// modtimer:
+//   timerWaiting    -> timerModifying -> timerModifiedXX
+//   timerModifiedXX -> timerModifying -> timerModifiedYY
+//   timerNoStatus   -> timerWaiting
+//   timerRemoved    -> timerWaiting
+//   timerRunning    -> wait until status changes
+//   timerMoving     -> wait until status changes
+//   timerRemoving   -> wait until status changes
+//   timerDeleted    -> panic: concurrent modtimer/deltimer calls
+//   timerModifying  -> panic: concurrent modtimer calls
 
 // Values for the timer status field.
 const (
@@ -270,6 +280,11 @@ func addtimer(t *timer) {
        }
        t.status = timerWaiting
 
+       addInitializedTimer(t)
+}
+
+// addInitializedTimer adds an initialized timer to the current P.
+func addInitializedTimer(t *timer) {
        when := t.when
 
        pp := getg().m.p.ptr()
@@ -363,7 +378,9 @@ func deltimer(t *timer) bool {
                                return true
                        }
                case timerModifiedEarlier:
+                       tpp := t.pp.ptr()
                        if atomic.Cas(&t.status, s, timerModifying) {
+                               atomic.Xadd(&tpp.adjustTimers, -1)
                                if !atomic.Cas(&t.status, timerModifying, timerDeleted) {
                                        badTimer()
                                }
@@ -438,12 +455,94 @@ func (tb *timersBucket) deltimerLocked(t *timer) (removed, ok bool) {
        return true, ok
 }
 
+// modtimer modifies an existing timer.
+// This is called by the netpoll code.
 func modtimer(t *timer, when, period int64, f func(interface{}, uintptr), arg interface{}, seq uintptr) {
        if oldTimers {
                modtimerOld(t, when, period, f, arg, seq)
                return
        }
-       throw("new modtimer not yet implemented")
+
+       if when < 0 {
+               when = maxWhen
+       }
+
+       status := uint32(timerNoStatus)
+       wasRemoved := false
+loop:
+       for {
+               switch status = atomic.Load(&t.status); status {
+               case timerWaiting, timerModifiedEarlier, timerModifiedLater:
+                       if atomic.Cas(&t.status, status, timerModifying) {
+                               break loop
+                       }
+               case timerNoStatus, timerRemoved:
+                       // Timer was already run and t is no longer in a heap.
+                       // Act like addtimer.
+                       wasRemoved = true
+                       atomic.Store(&t.status, timerWaiting)
+                       break loop
+               case timerRunning, timerRemoving, timerMoving:
+                       // The timer is being run or moved, by a different P.
+                       // Wait for it to complete.
+                       osyield()
+               case timerDeleted:
+                       // Simultaneous calls to modtimer and deltimer.
+                       badTimer()
+               case timerModifying:
+                       // Multiple simultaneous calls to modtimer.
+                       badTimer()
+               default:
+                       badTimer()
+               }
+       }
+
+       t.period = period
+       t.f = f
+       t.arg = arg
+       t.seq = seq
+
+       if wasRemoved {
+               t.when = when
+               addInitializedTimer(t)
+       } else {
+               // The timer is in some other P's heap, so we can't change
+               // the when field. If we did, the other P's heap would
+               // be out of order. So we put the new when value in the
+               // nextwhen field, and let the other P set the when field
+               // when it is prepared to resort the heap.
+               t.nextwhen = when
+
+               newStatus := uint32(timerModifiedLater)
+               if when < t.when {
+                       newStatus = timerModifiedEarlier
+               }
+
+               // Update the adjustTimers field.  Subtract one if we
+               // are removing a timerModifiedEarlier, add one if we
+               // are adding a timerModifiedEarlier.
+               tpp := t.pp.ptr()
+               adjust := int32(0)
+               if status == timerModifiedEarlier {
+                       adjust--
+               }
+               if newStatus == timerModifiedEarlier {
+                       adjust++
+               }
+               if adjust != 0 {
+                       atomic.Xadd(&tpp.adjustTimers, adjust)
+               }
+
+               // Set the new status of the timer.
+               if !atomic.Cas(&t.status, timerModifying, newStatus) {
+                       badTimer()
+               }
+
+               // If the new status is earlier, wake up the poller.
+               if newStatus == timerModifiedEarlier {
+                       wakeNetPoller(when)
+               }
+       }
 }
 
 func modtimerOld(t *timer, when, period int64, f func(interface{}, uintptr), arg interface{}, seq uintptr) {