synctest.go

     1  // Copyright 2024 The Go Authors. All rights reserved.
     2  // Use of this source code is governed by a BSD-style
     3  // license that can be found in the LICENSE file.
     4  
     5  package runtime
     6  
     7  import (
     8  	"internal/runtime/atomic"
     9  	"internal/runtime/sys"
    10  	"unsafe"
    11  )
    12  
    13  // A synctestBubble is a set of goroutines started by synctest.Run.
    14  type synctestBubble struct {
    15  	mu      mutex
    16  	timers  timers
    17  	id      uint64 // unique id
    18  	now     int64  // current fake time
    19  	root    *g     // caller of synctest.Run
    20  	waiter  *g     // caller of synctest.Wait
    21  	main    *g     // goroutine started by synctest.Run
    22  	waiting bool   // true if a goroutine is calling synctest.Wait
    23  	done    bool   // true if main has exited
    24  
    25  	// The bubble is active (not blocked) so long as running > 0 || active > 0.
    26  	//
    27  	// running is the number of goroutines which are not "durably blocked":
    28  	// Goroutines which are either running, runnable, or non-durably blocked
    29  	// (for example, blocked in a syscall).
    30  	//
    31  	// active is used to keep the bubble from becoming blocked,
    32  	// even if all goroutines in the bubble are blocked.
    33  	// For example, park_m can choose to immediately unpark a goroutine after parking it.
    34  	// It increments the active count to keep the bubble active until it has determined
    35  	// that the park operation has completed.
    36  	total   int // total goroutines
    37  	running int // non-blocked goroutines
    38  	active  int // other sources of activity
    39  }
    40  
    41  // changegstatus is called when the non-lock status of a g changes.
    42  // It is never called with a Gscanstatus.
    43  func (bubble *synctestBubble) changegstatus(gp *g, oldval, newval uint32) {
    44  	// Determine whether this change in status affects the idleness of the bubble.
    45  	// If this isn't a goroutine starting, stopping, durably blocking,
    46  	// or waking up after durably blocking, then return immediately without
    47  	// locking bubble.mu.
    48  	//
    49  	// For example, stack growth (newstack) will changegstatus
    50  	// from _Grunning to _Gcopystack. This is uninteresting to synctest,
    51  	// but if stack growth occurs while bubble.mu is held, we must not recursively lock.
    52  	totalDelta := 0
    53  	wasRunning := true
    54  	switch oldval {
    55  	case _Gdead:
    56  		wasRunning = false
    57  		totalDelta++
    58  	case _Gwaiting:
    59  		if gp.waitreason.isIdleInSynctest() {
    60  			wasRunning = false
    61  		}
    62  	}
    63  	isRunning := true
    64  	switch newval {
    65  	case _Gdead:
    66  		isRunning = false
    67  		totalDelta--
    68  		if gp == bubble.main {
    69  			bubble.done = true
    70  		}
    71  	case _Gwaiting:
    72  		if gp.waitreason.isIdleInSynctest() {
    73  			isRunning = false
    74  		}
    75  	}
    76  	// It's possible for wasRunning == isRunning while totalDelta != 0;
    77  	// for example, if a new goroutine is created in a non-running state.
    78  	if wasRunning == isRunning && totalDelta == 0 {
    79  		return
    80  	}
    81  
    82  	lock(&bubble.mu)
    83  	bubble.total += totalDelta
    84  	if wasRunning != isRunning {
    85  		if isRunning {
    86  			bubble.running++
    87  		} else {
    88  			bubble.running--
    89  			if raceenabled && newval != _Gdead {
    90  				// Record that this goroutine parking happens before
    91  				// any subsequent Wait.
    92  				racereleasemergeg(gp, bubble.raceaddr())
    93  			}
    94  		}
    95  	}
    96  	if bubble.total < 0 {
    97  		fatal("total < 0")
    98  	}
    99  	if bubble.running < 0 {
   100  		fatal("running < 0")
   101  	}
   102  	wake := bubble.maybeWakeLocked()
   103  	unlock(&bubble.mu)
   104  	if wake != nil {
   105  		goready(wake, 0)
   106  	}
   107  }
   108  
   109  // incActive increments the active-count for the bubble.
   110  // A bubble does not become durably blocked while the active-count is non-zero.
   111  func (bubble *synctestBubble) incActive() {
   112  	lock(&bubble.mu)
   113  	bubble.active++
   114  	unlock(&bubble.mu)
   115  }
   116  
   117  // decActive decrements the active-count for the bubble.
   118  func (bubble *synctestBubble) decActive() {
   119  	lock(&bubble.mu)
   120  	bubble.active--
   121  	if bubble.active < 0 {
   122  		throw("active < 0")
   123  	}
   124  	wake := bubble.maybeWakeLocked()
   125  	unlock(&bubble.mu)
   126  	if wake != nil {
   127  		goready(wake, 0)
   128  	}
   129  }
   130  
   131  // maybeWakeLocked returns a g to wake if the bubble is durably blocked.
   132  func (bubble *synctestBubble) maybeWakeLocked() *g {
   133  	if bubble.running > 0 || bubble.active > 0 {
   134  		return nil
   135  	}
   136  	// Increment the bubble active count, since we've determined to wake something.
   137  	// The woken goroutine will decrement the count.
   138  	// We can't just call goready and let it increment bubble.running,
   139  	// since we can't call goready with bubble.mu held.
   140  	//
   141  	// Incrementing the active count here is only necessary if something has gone wrong,
   142  	// and a goroutine that we considered durably blocked wakes up unexpectedly.
   143  	// Two wakes happening at the same time leads to very confusing failure modes,
   144  	// so we take steps to avoid it happening.
   145  	bubble.active++
   146  	next := bubble.timers.wakeTime()
   147  	if next > 0 && next <= bubble.now {
   148  		// A timer is scheduled to fire. Wake the root goroutine to handle it.
   149  		return bubble.root
   150  	}
   151  	if gp := bubble.waiter; gp != nil {
   152  		// A goroutine is blocked in Wait. Wake it.
   153  		return gp
   154  	}
   155  	// All goroutines in the bubble are durably blocked, and nothing has called Wait.
   156  	// Wake the root goroutine.
   157  	return bubble.root
   158  }
   159  
   160  func (bubble *synctestBubble) raceaddr() unsafe.Pointer {
   161  	// Address used to record happens-before relationships created by the bubble.
   162  	//
   163  	// Wait creates a happens-before relationship between itself and
   164  	// the blocking operations which caused other goroutines in the bubble to park.
   165  	return unsafe.Pointer(bubble)
   166  }
   167  
   168  var bubbleGen atomic.Uint64 // bubble ID counter
   169  
   170  //go:linkname synctestRun internal/synctest.Run
   171  func synctestRun(f func()) {
   172  	if debug.asynctimerchan.Load() != 0 {
   173  		panic("synctest.Run not supported with asynctimerchan!=0")
   174  	}
   175  
   176  	gp := getg()
   177  	if gp.bubble != nil {
   178  		panic("synctest.Run called from within a synctest bubble")
   179  	}
   180  	bubble := &synctestBubble{
   181  		id:      bubbleGen.Add(1),
   182  		total:   1,
   183  		running: 1,
   184  		root:    gp,
   185  	}
   186  	const synctestBaseTime = 946684800000000000 // midnight UTC 2000-01-01
   187  	bubble.now = synctestBaseTime
   188  	lockInit(&bubble.mu, lockRankSynctest)
   189  	lockInit(&bubble.timers.mu, lockRankTimers)
   190  
   191  	gp.bubble = bubble
   192  	defer func() {
   193  		gp.bubble = nil
   194  	}()
   195  
   196  	// This is newproc, but also records the new g in bubble.main.
   197  	pc := sys.GetCallerPC()
   198  	systemstack(func() {
   199  		fv := *(**funcval)(unsafe.Pointer(&f))
   200  		bubble.main = newproc1(fv, gp, pc, false, waitReasonZero)
   201  		pp := getg().m.p.ptr()
   202  		runqput(pp, bubble.main, true)
   203  		wakep()
   204  	})
   205  
   206  	lock(&bubble.mu)
   207  	bubble.active++
   208  	for {
   209  		unlock(&bubble.mu)
   210  		systemstack(func() {
   211  			// Clear gp.m.curg while running timers,
   212  			// so timer goroutines inherit their child race context from g0.
   213  			curg := gp.m.curg
   214  			gp.m.curg = nil
   215  			gp.bubble.timers.check(bubble.now, bubble)
   216  			gp.m.curg = curg
   217  		})
   218  		gopark(synctestidle_c, nil, waitReasonSynctestRun, traceBlockSynctest, 0)
   219  		lock(&bubble.mu)
   220  		if bubble.active < 0 {
   221  			throw("active < 0")
   222  		}
   223  		next := bubble.timers.wakeTime()
   224  		if next == 0 {
   225  			break
   226  		}
   227  		if next < bubble.now {
   228  			throw("time went backwards")
   229  		}
   230  		if bubble.done {
   231  			// Time stops once the bubble's main goroutine has exited.
   232  			break
   233  		}
   234  		bubble.now = next
   235  	}
   236  
   237  	total := bubble.total
   238  	unlock(&bubble.mu)
   239  	if raceenabled {
   240  		// Establish a happens-before relationship between bubbled goroutines exiting
   241  		// and Run returning.
   242  		raceacquireg(gp, gp.bubble.raceaddr())
   243  	}
   244  	if total != 1 {
   245  		var reason string
   246  		if bubble.done {
   247  			reason = "deadlock: main bubble goroutine has exited but blocked goroutines remain"
   248  		} else {
   249  			reason = "deadlock: all goroutines in bubble are blocked"
   250  		}
   251  		panic(synctestDeadlockError{reason: reason, bubble: bubble})
   252  	}
   253  	if gp.timer != nil && gp.timer.isFake {
   254  		// Verify that we haven't marked this goroutine's sleep timer as fake.
   255  		// This could happen if something in Run were to call timeSleep.
   256  		throw("synctest root goroutine has a fake timer")
   257  	}
   258  }
   259  
   260  type synctestDeadlockError struct {
   261  	reason string
   262  	bubble *synctestBubble
   263  }
   264  
   265  func (e synctestDeadlockError) Error() string {
   266  	return e.reason
   267  }
   268  
   269  func synctestidle_c(gp *g, _ unsafe.Pointer) bool {
   270  	lock(&gp.bubble.mu)
   271  	canIdle := true
   272  	if gp.bubble.running == 0 && gp.bubble.active == 1 {
   273  		// All goroutines in the bubble have blocked or exited.
   274  		canIdle = false
   275  	} else {
   276  		gp.bubble.active--
   277  	}
   278  	unlock(&gp.bubble.mu)
   279  	return canIdle
   280  }
   281  
   282  //go:linkname synctestWait internal/synctest.Wait
   283  func synctestWait() {
   284  	gp := getg()
   285  	if gp.bubble == nil {
   286  		panic("goroutine is not in a bubble")
   287  	}
   288  	lock(&gp.bubble.mu)
   289  	// We use a bubble.waiting bool to detect simultaneous calls to Wait rather than
   290  	// checking to see if bubble.waiter is non-nil. This avoids a race between unlocking
   291  	// bubble.mu and setting bubble.waiter while parking.
   292  	if gp.bubble.waiting {
   293  		unlock(&gp.bubble.mu)
   294  		panic("wait already in progress")
   295  	}
   296  	gp.bubble.waiting = true
   297  	unlock(&gp.bubble.mu)
   298  	gopark(synctestwait_c, nil, waitReasonSynctestWait, traceBlockSynctest, 0)
   299  
   300  	lock(&gp.bubble.mu)
   301  	gp.bubble.active--
   302  	if gp.bubble.active < 0 {
   303  		throw("active < 0")
   304  	}
   305  	gp.bubble.waiter = nil
   306  	gp.bubble.waiting = false
   307  	unlock(&gp.bubble.mu)
   308  
   309  	// Establish a happens-before relationship on the activity of the now-blocked
   310  	// goroutines in the bubble.
   311  	if raceenabled {
   312  		raceacquireg(gp, gp.bubble.raceaddr())
   313  	}
   314  }
   315  
   316  func synctestwait_c(gp *g, _ unsafe.Pointer) bool {
   317  	lock(&gp.bubble.mu)
   318  	if gp.bubble.running == 0 && gp.bubble.active == 0 {
   319  		// This shouldn't be possible, since gopark increments active during unlockf.
   320  		throw("running == 0 && active == 0")
   321  	}
   322  	gp.bubble.waiter = gp
   323  	unlock(&gp.bubble.mu)
   324  	return true
   325  }
   326  
   327  //go:linkname synctest_isInBubble internal/synctest.IsInBubble
   328  func synctest_isInBubble() bool {
   329  	return getg().bubble != nil
   330  }
   331  
   332  //go:linkname synctest_acquire internal/synctest.acquire
   333  func synctest_acquire() any {
   334  	if bubble := getg().bubble; bubble != nil {
   335  		bubble.incActive()
   336  		return bubble
   337  	}
   338  	return nil
   339  }
   340  
   341  //go:linkname synctest_release internal/synctest.release
   342  func synctest_release(bubble any) {
   343  	bubble.(*synctestBubble).decActive()
   344  }
   345  
   346  //go:linkname synctest_inBubble internal/synctest.inBubble
   347  func synctest_inBubble(bubble any, f func()) {
   348  	gp := getg()
   349  	if gp.bubble != nil {
   350  		panic("goroutine is already bubbled")
   351  	}
   352  	gp.bubble = bubble.(*synctestBubble)
   353  	defer func() {
   354  		gp.bubble = nil
   355  	}()
   356  	f()
   357  }
   358  
   359  // specialBubble is a special used to associate objects with bubbles.
   360  type specialBubble struct {
   361  	_        sys.NotInHeap
   362  	special  special
   363  	bubbleid uint64
   364  }
   365  
   366  // Keep these in sync with internal/synctest.
   367  const (
   368  	bubbleAssocUnbubbled     = iota // not associated with any bubble
   369  	bubbleAssocCurrentBubble        // associated with the current bubble
   370  	bubbleAssocOtherBubble          // associated with a different bubble
   371  )
   372  
   373  // getOrSetBubbleSpecial checks the special record for p's bubble membership.
   374  //
   375  // If add is true and p is not associated with any bubble,
   376  // it adds a special record for p associating it with bubbleid.
   377  //
   378  // It returns ok==true if p is associated with bubbleid
   379  // (including if a new association was added),
   380  // and ok==false if not.
   381  func getOrSetBubbleSpecial(p unsafe.Pointer, bubbleid uint64, add bool) (assoc int) {
   382  	span := spanOfHeap(uintptr(p))
   383  	if span == nil {
   384  		// This is probably a package var.
   385  		// We can't attach a special to it, so always consider it unbubbled.
   386  		return bubbleAssocUnbubbled
   387  	}
   388  
   389  	// Ensure that the span is swept.
   390  	// Sweeping accesses the specials list w/o locks, so we have
   391  	// to synchronize with it. And it's just much safer.
   392  	mp := acquirem()
   393  	span.ensureSwept()
   394  
   395  	offset := uintptr(p) - span.base()
   396  
   397  	lock(&span.speciallock)
   398  
   399  	// Find splice point, check for existing record.
   400  	iter, exists := span.specialFindSplicePoint(offset, _KindSpecialBubble)
   401  	if exists {
   402  		// p is already associated with a bubble.
   403  		// Return true iff it's the same bubble.
   404  		s := (*specialBubble)((unsafe.Pointer)(*iter))
   405  		if s.bubbleid == bubbleid {
   406  			assoc = bubbleAssocCurrentBubble
   407  		} else {
   408  			assoc = bubbleAssocOtherBubble
   409  		}
   410  	} else if add {
   411  		// p is not associated with a bubble,
   412  		// and we've been asked to add an association.
   413  		s := (*specialBubble)(mheap_.specialBubbleAlloc.alloc())
   414  		s.bubbleid = bubbleid
   415  		s.special.kind = _KindSpecialBubble
   416  		s.special.offset = offset
   417  		s.special.next = *iter
   418  		*iter = (*special)(unsafe.Pointer(s))
   419  		spanHasSpecials(span)
   420  		assoc = bubbleAssocCurrentBubble
   421  	} else {
   422  		// p is not associated with a bubble.
   423  		assoc = bubbleAssocUnbubbled
   424  	}
   425  
   426  	unlock(&span.speciallock)
   427  	releasem(mp)
   428  
   429  	return assoc
   430  }
   431  
   432  // synctest_associate associates p with the current bubble.
   433  // It returns false if p is already associated with a different bubble.
   434  //
   435  //go:linkname synctest_associate internal/synctest.associate
   436  func synctest_associate(p unsafe.Pointer) int {
   437  	return getOrSetBubbleSpecial(p, getg().bubble.id, true)
   438  }
   439  
   440  // synctest_disassociate disassociates p from its bubble.
   441  //
   442  //go:linkname synctest_disassociate internal/synctest.disassociate
   443  func synctest_disassociate(p unsafe.Pointer) {
   444  	removespecial(p, _KindSpecialBubble)
   445  }
   446  
   447  // synctest_isAssociated reports whether p is associated with the current bubble.
   448  //
   449  //go:linkname synctest_isAssociated internal/synctest.isAssociated
   450  func synctest_isAssociated(p unsafe.Pointer) bool {
   451  	return getOrSetBubbleSpecial(p, getg().bubble.id, false) == bubbleAssocCurrentBubble
   452  }
   453
View as plain text