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, _Gdeadextra:
    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, _Gdeadextra:
    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 && newval != _Gdeadextra {
    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  	gp := getg()
   173  	if gp.bubble != nil {
   174  		panic("synctest.Run called from within a synctest bubble")
   175  	}
   176  	bubble := &synctestBubble{
   177  		id:      bubbleGen.Add(1),
   178  		total:   1,
   179  		running: 1,
   180  		root:    gp,
   181  	}
   182  	const synctestBaseTime = 946684800000000000 // midnight UTC 2000-01-01
   183  	bubble.now = synctestBaseTime
   184  	lockInit(&bubble.mu, lockRankSynctest)
   185  	lockInit(&bubble.timers.mu, lockRankTimers)
   186  
   187  	gp.bubble = bubble
   188  	defer func() {
   189  		gp.bubble = nil
   190  	}()
   191  
   192  	// This is newproc, but also records the new g in bubble.main.
   193  	pc := sys.GetCallerPC()
   194  	systemstack(func() {
   195  		fv := *(**funcval)(unsafe.Pointer(&f))
   196  		bubble.main = newproc1(fv, gp, pc, false, waitReasonZero)
   197  		pp := getg().m.p.ptr()
   198  		runqput(pp, bubble.main, true)
   199  		wakep()
   200  	})
   201  
   202  	lock(&bubble.mu)
   203  	bubble.active++
   204  	for {
   205  		unlock(&bubble.mu)
   206  		systemstack(func() {
   207  			// Clear gp.m.curg while running timers,
   208  			// so timer goroutines inherit their child race context from g0.
   209  			curg := gp.m.curg
   210  			gp.m.curg = nil
   211  			gp.bubble.timers.check(bubble.now, bubble)
   212  			gp.m.curg = curg
   213  		})
   214  		gopark(synctestidle_c, nil, waitReasonSynctestRun, traceBlockSynctest, 0)
   215  		lock(&bubble.mu)
   216  		if bubble.active < 0 {
   217  			throw("active < 0")
   218  		}
   219  		next := bubble.timers.wakeTime()
   220  		if next == 0 {
   221  			break
   222  		}
   223  		if next < bubble.now {
   224  			throw("time went backwards")
   225  		}
   226  		if bubble.done {
   227  			// Time stops once the bubble's main goroutine has exited.
   228  			break
   229  		}
   230  		bubble.now = next
   231  	}
   232  
   233  	total := bubble.total
   234  	unlock(&bubble.mu)
   235  	if raceenabled {
   236  		// Establish a happens-before relationship between bubbled goroutines exiting
   237  		// and Run returning.
   238  		raceacquireg(gp, gp.bubble.raceaddr())
   239  	}
   240  	if total != 1 {
   241  		var reason string
   242  		if bubble.done {
   243  			reason = "deadlock: main bubble goroutine has exited but blocked goroutines remain"
   244  		} else {
   245  			reason = "deadlock: all goroutines in bubble are blocked"
   246  		}
   247  		panic(synctestDeadlockError{reason: reason, bubble: bubble})
   248  	}
   249  	if gp.timer != nil && gp.timer.isFake {
   250  		// Verify that we haven't marked this goroutine's sleep timer as fake.
   251  		// This could happen if something in Run were to call timeSleep.
   252  		throw("synctest root goroutine has a fake timer")
   253  	}
   254  }
   255  
   256  type synctestDeadlockError struct {
   257  	reason string
   258  	bubble *synctestBubble
   259  }
   260  
   261  var _ error = synctestDeadlockError{}
   262  
   263  func (e synctestDeadlockError) Error() string {
   264  	return e.reason
   265  }
   266  
   267  func synctestidle_c(gp *g, _ unsafe.Pointer) bool {
   268  	lock(&gp.bubble.mu)
   269  	canIdle := true
   270  	if gp.bubble.running == 0 && gp.bubble.active == 1 {
   271  		// All goroutines in the bubble have blocked or exited.
   272  		canIdle = false
   273  	} else {
   274  		gp.bubble.active--
   275  	}
   276  	unlock(&gp.bubble.mu)
   277  	return canIdle
   278  }
   279  
   280  //go:linkname synctestWait internal/synctest.Wait
   281  func synctestWait() {
   282  	gp := getg()
   283  	if gp.bubble == nil {
   284  		panic("goroutine is not in a bubble")
   285  	}
   286  	lock(&gp.bubble.mu)
   287  	// We use a bubble.waiting bool to detect simultaneous calls to Wait rather than
   288  	// checking to see if bubble.waiter is non-nil. This avoids a race between unlocking
   289  	// bubble.mu and setting bubble.waiter while parking.
   290  	if gp.bubble.waiting {
   291  		unlock(&gp.bubble.mu)
   292  		panic("wait already in progress")
   293  	}
   294  	gp.bubble.waiting = true
   295  	unlock(&gp.bubble.mu)
   296  	gopark(synctestwait_c, nil, waitReasonSynctestWait, traceBlockSynctest, 0)
   297  
   298  	lock(&gp.bubble.mu)
   299  	gp.bubble.active--
   300  	if gp.bubble.active < 0 {
   301  		throw("active < 0")
   302  	}
   303  	gp.bubble.waiter = nil
   304  	gp.bubble.waiting = false
   305  	unlock(&gp.bubble.mu)
   306  
   307  	// Establish a happens-before relationship on the activity of the now-blocked
   308  	// goroutines in the bubble.
   309  	if raceenabled {
   310  		raceacquireg(gp, gp.bubble.raceaddr())
   311  	}
   312  }
   313  
   314  func synctestwait_c(gp *g, _ unsafe.Pointer) bool {
   315  	lock(&gp.bubble.mu)
   316  	if gp.bubble.running == 0 && gp.bubble.active == 0 {
   317  		// This shouldn't be possible, since gopark increments active during unlockf.
   318  		throw("running == 0 && active == 0")
   319  	}
   320  	gp.bubble.waiter = gp
   321  	unlock(&gp.bubble.mu)
   322  	return true
   323  }
   324  
   325  //go:linkname synctest_isInBubble internal/synctest.IsInBubble
   326  func synctest_isInBubble() bool {
   327  	return getg().bubble != nil
   328  }
   329  
   330  //go:linkname synctest_acquire internal/synctest.acquire
   331  func synctest_acquire() any {
   332  	if bubble := getg().bubble; bubble != nil {
   333  		bubble.incActive()
   334  		return bubble
   335  	}
   336  	return nil
   337  }
   338  
   339  //go:linkname synctest_release internal/synctest.release
   340  func synctest_release(bubble any) {
   341  	bubble.(*synctestBubble).decActive()
   342  }
   343  
   344  //go:linkname synctest_inBubble internal/synctest.inBubble
   345  func synctest_inBubble(bubble any, f func()) {
   346  	gp := getg()
   347  	if gp.bubble != nil {
   348  		panic("goroutine is already bubbled")
   349  	}
   350  	gp.bubble = bubble.(*synctestBubble)
   351  	defer func() {
   352  		gp.bubble = nil
   353  	}()
   354  	f()
   355  }
   356  
   357  // specialBubble is a special used to associate objects with bubbles.
   358  type specialBubble struct {
   359  	_        sys.NotInHeap
   360  	special  special
   361  	bubbleid uint64
   362  }
   363  
   364  // Keep these in sync with internal/synctest.
   365  const (
   366  	bubbleAssocUnbubbled     = iota // not associated with any bubble
   367  	bubbleAssocCurrentBubble        // associated with the current bubble
   368  	bubbleAssocOtherBubble          // associated with a different bubble
   369  )
   370  
   371  // getOrSetBubbleSpecial checks the special record for p's bubble membership.
   372  //
   373  // If add is true and p is not associated with any bubble,
   374  // it adds a special record for p associating it with bubbleid.
   375  //
   376  // It returns ok==true if p is associated with bubbleid
   377  // (including if a new association was added),
   378  // and ok==false if not.
   379  func getOrSetBubbleSpecial(p unsafe.Pointer, bubbleid uint64, add bool) (assoc int) {
   380  	span := spanOfHeap(uintptr(p))
   381  	if span == nil {
   382  		// This is probably a package var.
   383  		// We can't attach a special to it, so always consider it unbubbled.
   384  		return bubbleAssocUnbubbled
   385  	}
   386  
   387  	// Ensure that the span is swept.
   388  	// Sweeping accesses the specials list w/o locks, so we have
   389  	// to synchronize with it. And it's just much safer.
   390  	mp := acquirem()
   391  	span.ensureSwept()
   392  
   393  	offset := uintptr(p) - span.base()
   394  
   395  	lock(&span.speciallock)
   396  
   397  	// Find splice point, check for existing record.
   398  	iter, exists := span.specialFindSplicePoint(offset, _KindSpecialBubble)
   399  	if exists {
   400  		// p is already associated with a bubble.
   401  		// Return true iff it's the same bubble.
   402  		s := (*specialBubble)((unsafe.Pointer)(*iter))
   403  		if s.bubbleid == bubbleid {
   404  			assoc = bubbleAssocCurrentBubble
   405  		} else {
   406  			assoc = bubbleAssocOtherBubble
   407  		}
   408  	} else if add {
   409  		// p is not associated with a bubble,
   410  		// and we've been asked to add an association.
   411  		lock(&mheap_.speciallock)
   412  		s := (*specialBubble)(mheap_.specialBubbleAlloc.alloc())
   413  		unlock(&mheap_.speciallock)
   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