@@ -388,8 +388,18 @@ func goparkunlock(lock *mutex, reason waitReason, traceEv byte, traceskip int) {
388388}
389389
390390func goready (gp * g , traceskip int ) {
391+ callergp := getg ()
392+ now := nanotime ()
393+ needWakep := readyUpdateAndPredictNeedWakep (callergp , now )
394+ systemstack (func () {
395+ ready (gp , traceskip , true , now , needWakep )
396+ })
397+ }
398+
399+ func goreadyNoWakep (gp * g , traceskip int ) {
400+ now := nanotime ()
391401 systemstack (func () {
392- ready (gp , traceskip , true )
402+ ready (gp , traceskip , true , now , false /* needWakep */ )
393403 })
394404}
395405
@@ -858,8 +868,42 @@ func fastrandinit() {
858868 getRandomData (s )
859869}
860870
871+ func parkUpdateNeedWakep (gp * g ) {
872+ gp .firstReady = 0
873+ if gp .lastReady != 0 {
874+ gp .runnextSwitchHistory = nanotime ()- gp .lastReady <= runnextSwitchNS
875+ gp .lastReady = 0
876+ }
877+ }
878+
879+ // Returns true if gp, calling goready() or newproc() at now, is predicted
880+ // *not* to park within the next runnextSwitchNS, such that a spinning P is
881+ // needed to run the new G.
882+ // Incorrectly returning true (causing a spinning P to be uselessly woken)
883+ // wastes cycles but is harmless.
884+ // Incorrectly returning false is handled by sysmon (in retake()).
885+ func readyUpdateAndPredictNeedWakep (gp * g , now int64 ) bool {
886+ pp := gp .m .p .ptr ()
887+
888+ // Update predictor state.
889+ gp .lastReady = now
890+ if now - gp .firstReady > runnextSwitchNS {
891+ if gp .firstReady != 0 {
892+ gp .runnextSwitchHistory = false
893+ }
894+ gp .firstReady = now
895+ }
896+
897+ // If the caller's runqueue is non-empty, predict that we need wakep();
898+ // even if gp parks, there's no guarantee that following Gs will.
899+ if ! runqempty (pp ) {
900+ return true
901+ }
902+ return ! gp .runnextSwitchHistory
903+ }
904+
861905// Mark gp ready to run.
862- func ready (gp * g , traceskip int , next bool ) {
906+ func ready (gp * g , traceskip int , next bool , now int64 , needWakep bool ) {
863907 if trace .enabled {
864908 traceGoUnpark (gp , traceskip )
865909 }
@@ -875,8 +919,10 @@ func ready(gp *g, traceskip int, next bool) {
875919
876920 // status is Gwaiting or Gscanwaiting, make Grunnable and put on runq
877921 casgstatus (gp , _Gwaiting , _Grunnable )
878- runqput (mp .p .ptr (), gp , next )
879- wakep ()
922+ runqput (mp .p .ptr (), gp , next , now )
923+ if needWakep {
924+ wakep ()
925+ }
880926 releasem (mp )
881927}
882928
@@ -2715,7 +2761,7 @@ top:
27152761 // Wake up the finalizer G.
27162762 if fingStatus .Load ()& (fingWait | fingWake ) == fingWait | fingWake {
27172763 if gp := wakefing (); gp != nil {
2718- ready (gp , 0 , true )
2764+ ready (gp , 0 , true , now , true )
27192765 }
27202766 }
27212767 if * cgo_yield != nil {
@@ -3027,7 +3073,7 @@ func pollWork() bool {
30273073func stealWork (now int64 ) (gp * g , inheritTime bool , rnow , pollUntil int64 , newWork bool ) {
30283074 pp := getg ().m .p .ptr ()
30293075
3030- ranTimer := false
3076+ ranTimerOrRunnextPending := false
30313077
30323078 const stealTries = 4
30333079 for i := 0 ; i < stealTries ; i ++ {
@@ -3072,16 +3118,21 @@ func stealWork(now int64) (gp *g, inheritTime bool, rnow, pollUntil int64, newWo
30723118 // stolen G's. So check now if there
30733119 // is a local G to run.
30743120 if gp , inheritTime := runqget (pp ); gp != nil {
3075- return gp , inheritTime , now , pollUntil , ranTimer
3121+ return gp , inheritTime , now , pollUntil , ranTimerOrRunnextPending
30763122 }
3077- ranTimer = true
3123+ ranTimerOrRunnextPending = true
30783124 }
30793125 }
30803126
30813127 // Don't bother to attempt to steal if p2 is idle.
30823128 if ! idlepMask .read (enum .position ()) {
3083- if gp := runqsteal (pp , p2 , stealTimersOrRunNextG ); gp != nil {
3084- return gp , false , now , pollUntil , ranTimer
3129+ tnow , gp , runnextPending := runqsteal (pp , p2 , stealTimersOrRunNextG , now )
3130+ now = tnow
3131+ if runnextPending {
3132+ ranTimerOrRunnextPending = true
3133+ }
3134+ if gp != nil {
3135+ return gp , false , now , pollUntil , ranTimerOrRunnextPending
30853136 }
30863137 }
30873138 }
@@ -3090,7 +3141,7 @@ func stealWork(now int64) (gp *g, inheritTime bool, rnow, pollUntil int64, newWo
30903141 // No goroutines found to steal. Regardless, running a timer may have
30913142 // made some goroutine ready that we missed. Indicate the next timer to
30923143 // wait for.
3093- return nil , false , now , pollUntil , ranTimer
3144+ return nil , false , now , pollUntil , ranTimerOrRunnextPending
30943145}
30953146
30963147// Check all Ps for a runnable G to steal.
@@ -3474,6 +3525,8 @@ func parkunlock_c(gp *g, lock unsafe.Pointer) bool {
34743525
34753526// park continuation on g0.
34763527func park_m (gp * g ) {
3528+ parkUpdateNeedWakep (gp )
3529+
34773530 mp := getg ().m
34783531
34793532 if trace .enabled {
@@ -3597,7 +3650,7 @@ func goyield_m(gp *g) {
35973650 pp := gp .m .p .ptr ()
35983651 casgstatus (gp , _Grunning , _Grunnable )
35993652 dropg ()
3600- runqput (pp , gp , false )
3653+ runqput (pp , gp , false , 0 )
36013654 schedule ()
36023655}
36033656
@@ -4232,11 +4285,13 @@ func newproc(fn *funcval) {
42324285 pc := getcallerpc ()
42334286 systemstack (func () {
42344287 newg := newproc1 (fn , gp , pc )
4288+ now := nanotime ()
4289+ needWakep := readyUpdateAndPredictNeedWakep (gp , now )
42354290
42364291 pp := getg ().m .p .ptr ()
4237- runqput (pp , newg , true )
4292+ runqput (pp , newg , true , now )
42384293
4239- if mainStarted {
4294+ if mainStarted && needWakep {
42404295 wakep ()
42414296 }
42424297 })
@@ -5389,6 +5444,7 @@ func sysmon() {
53895444 }
53905445 // retake P's blocked in syscalls
53915446 // and preempt long running G's
5447+ // and start P's if we mispredicted that wakep() was unnecessary
53925448 if retake (now ) != 0 {
53935449 idle = 0
53945450 } else {
@@ -5424,6 +5480,7 @@ const forcePreemptNS = 10 * 1000 * 1000 // 10ms
54245480
54255481func retake (now int64 ) uint32 {
54265482 n := 0
5483+ needWakep := false
54275484 // Prevent allp slice changes. This lock will be completely
54285485 // uncontended unless we're already stopping the world.
54295486 lock (& allpLock )
@@ -5437,6 +5494,9 @@ func retake(now int64) uint32 {
54375494 // allp but not yet created new Ps.
54385495 continue
54395496 }
5497+ if runqstealable (pp , now ) && sched .nmspinning .Load () == 0 {
5498+ needWakep = true
5499+ }
54405500 pd := & pp .sysmontick
54415501 s := pp .status
54425502 sysretake := false
@@ -5488,6 +5548,10 @@ func retake(now int64) uint32 {
54885548 }
54895549 }
54905550 unlock (& allpLock )
5551+ if needWakep {
5552+ wakep ()
5553+ n ++
5554+ }
54915555 return uint32 (n )
54925556}
54935557
@@ -5766,7 +5830,7 @@ func globrunqget(pp *p, max int32) *g {
57665830 n --
57675831 for ; n > 0 ; n -- {
57685832 gp1 := sched .runq .pop ()
5769- runqput (pp , gp1 , false )
5833+ runqput (pp , gp1 , false , 0 )
57705834 }
57715835 return gp
57725836}
@@ -5933,6 +5997,25 @@ func runqempty(pp *p) bool {
59335997 }
59345998}
59355999
6000+ // runqstealable is like !runqempty, but returns false if pp has a G in
6001+ // p.runnext that can't be stolen yet.
6002+ func runqstealable (pp * p , now int64 ) bool {
6003+ for {
6004+ head := atomic .Load (& pp .runqhead )
6005+ tail := atomic .Load (& pp .runqtail )
6006+ runnext := atomic .Loaduintptr ((* uintptr )(unsafe .Pointer (& pp .runnext )))
6007+ if tail == atomic .Load (& pp .runqtail ) {
6008+ if head != tail {
6009+ return true
6010+ }
6011+ if runnext == 0 {
6012+ return false
6013+ }
6014+ return ((guintptr )(runnext )).ptr ().runnextSince + runnextSwitchNS < now
6015+ }
6016+ }
6017+ }
6018+
59366019// To shake out latent assumptions about scheduling order,
59376020// we introduce some randomness into scheduling decisions
59386021// when running with the race detector.
@@ -5949,12 +6032,16 @@ const randomizeScheduler = raceenabled
59496032// If next is true, runqput puts g in the pp.runnext slot.
59506033// If the run queue is full, runnext puts g on the global queue.
59516034// Executed only by the owner P.
5952- func runqput (pp * p , gp * g , next bool ) {
6035+ func runqput (pp * p , gp * g , next bool , now int64 ) {
59536036 if randomizeScheduler && next && fastrandn (2 ) == 0 {
59546037 next = false
59556038 }
59566039
59576040 if next {
6041+ if now == 0 {
6042+ now = nanotime ()
6043+ }
6044+ gp .runnextSince = now
59586045 retryNext:
59596046 oldnext := pp .runnext
59606047 if ! pp .runnext .cas (oldnext , guintptr (unsafe .Pointer (gp ))) {
@@ -6123,11 +6210,22 @@ retry:
61236210 return
61246211}
61256212
6213+ // Don't steal p.runnext if it's been made runnable within the last
6214+ // runnextSwitchNS and the P is running.
6215+ // The important use case here is when the G running on the P ready()s another G
6216+ // and then almost immediately blocks.
6217+ // Giving the P a chance to schedule runnext avoids thrashing Gs between
6218+ // different Ps.
6219+ // On most platforms, sleep timeout granularity is coarser than
6220+ // runnextSwitchNS, so sleeping will result in significant overshoot; instead,
6221+ // stealWork() instructs findRunnable() to spin-wait.
6222+ const runnextSwitchNS = 5e3
6223+
61266224// Grabs a batch of goroutines from pp's runnable queue into batch.
61276225// Batch is a ring buffer starting at batchHead.
61286226// Returns number of grabbed goroutines.
61296227// Can be executed by any P.
6130- func runqgrab (pp * p , batch * [256 ]guintptr , batchHead uint32 , stealRunNextG bool ) uint32 {
6228+ func runqgrab (pp * p , batch * [256 ]guintptr , batchHead uint32 , stealRunNextG bool , now int64 ) ( rnow int64 , n uint32 , runnextPending bool ) {
61316229 for {
61326230 h := atomic .LoadAcq (& pp .runqhead ) // load-acquire, synchronize with other consumers
61336231 t := atomic .LoadAcq (& pp .runqtail ) // load-acquire, synchronize with the producer
@@ -6138,33 +6236,22 @@ func runqgrab(pp *p, batch *[256]guintptr, batchHead uint32, stealRunNextG bool)
61386236 // Try to steal from pp.runnext.
61396237 if next := pp .runnext ; next != 0 {
61406238 if pp .status == _Prunning {
6141- // Sleep to ensure that pp isn't about to run the g
6142- // we are about to steal.
6143- // The important use case here is when the g running
6144- // on pp ready()s another g and then almost
6145- // immediately blocks. Instead of stealing runnext
6146- // in this window, back off to give pp a chance to
6147- // schedule runnext. This will avoid thrashing gs
6148- // between different Ps.
6149- // A sync chan send/recv takes ~50ns as of time of
6150- // writing, so 3us gives ~50x overshoot.
6151- if GOOS != "windows" && GOOS != "openbsd" && GOOS != "netbsd" {
6152- usleep (3 )
6153- } else {
6154- // On some platforms system timer granularity is
6155- // 1-15ms, which is way too much for this
6156- // optimization. So just yield.
6157- osyield ()
6239+ // Enforce runnextSwitchNS.
6240+ if now == 0 {
6241+ now = nanotime ()
6242+ }
6243+ if now <= next .ptr ().runnextSince + runnextSwitchNS {
6244+ return now , 0 , true
61586245 }
61596246 }
61606247 if ! pp .runnext .cas (next , 0 ) {
61616248 continue
61626249 }
61636250 batch [batchHead % uint32 (len (batch ))] = next
6164- return 1
6251+ return now , 1 , false
61656252 }
61666253 }
6167- return 0
6254+ return now , 0 , false
61686255 }
61696256 if n > uint32 (len (pp .runq )/ 2 ) { // read inconsistent h and t
61706257 continue
@@ -6174,31 +6261,32 @@ func runqgrab(pp *p, batch *[256]guintptr, batchHead uint32, stealRunNextG bool)
61746261 batch [(batchHead + i )% uint32 (len (batch ))] = g
61756262 }
61766263 if atomic .CasRel (& pp .runqhead , h , h + n ) { // cas-release, commits consume
6177- return n
6264+ return now , n , false
61786265 }
61796266 }
61806267}
61816268
61826269// Steal half of elements from local runnable queue of p2
61836270// and put onto local runnable queue of p.
61846271// Returns one of the stolen elements (or nil if failed).
6185- func runqsteal (pp , p2 * p , stealRunNextG bool ) * g {
6272+ func runqsteal (pp , p2 * p , stealRunNextG bool , now int64 ) ( rnow int64 , gp * g , runnextPending bool ) {
61866273 t := pp .runqtail
6187- n := runqgrab (p2 , & pp .runq , t , stealRunNextG )
6274+ tnow , n , runnextPending := runqgrab (p2 , & pp .runq , t , stealRunNextG , now )
6275+ now = tnow
61886276 if n == 0 {
6189- return nil
6277+ return now , nil , runnextPending
61906278 }
61916279 n --
6192- gp : =pp .runq [(t + n )% uint32 (len (pp .runq ))].ptr ()
6280+ gp = pp .runq [(t + n )% uint32 (len (pp .runq ))].ptr ()
61936281 if n == 0 {
6194- return gp
6282+ return now , gp , runnextPending
61956283 }
61966284 h := atomic .LoadAcq (& pp .runqhead ) // load-acquire, synchronize with consumers
61976285 if t - h + n >= uint32 (len (pp .runq )) {
61986286 throw ("runqsteal: runq overflow" )
61996287 }
62006288 atomic .StoreRel (& pp .runqtail , t + n ) // store-release, makes the item available for consumption
6201- return gp
6289+ return now , gp , runnextPending
62026290}
62036291
62046292// A gQueue is a dequeue of Gs linked through g.schedlink. A G can only
0 commit comments