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  	"unsafe"
     9  )
    10  
    11  // A synctestGroup is a group of goroutines started by synctest.Run.
    12  type synctestGroup struct {
    13  	mu      mutex
    14  	timers  timers
    15  	now     int64 // current fake time
    16  	root    *g    // caller of synctest.Run
    17  	waiter  *g    // caller of synctest.Wait
    18  	waiting bool  // true if a goroutine is calling synctest.Wait
    19  
    20  	// The group is active (not blocked) so long as running > 0 || active > 0.
    21  	//
    22  	// running is the number of goroutines which are not "durably blocked":
    23  	// Goroutines which are either running, runnable, or non-durably blocked
    24  	// (for example, blocked in a syscall).
    25  	//
    26  	// active is used to keep the group from becoming blocked,
    27  	// even if all goroutines in the group are blocked.
    28  	// For example, park_m can choose to immediately unpark a goroutine after parking it.
    29  	// It increments the active count to keep the group active until it has determined
    30  	// that the park operation has completed.
    31  	total   int // total goroutines
    32  	running int // non-blocked goroutines
    33  	active  int // other sources of activity
    34  }
    35  
    36  // changegstatus is called when the non-lock status of a g changes.
    37  // It is never called with a Gscanstatus.
    38  func (sg *synctestGroup) changegstatus(gp *g, oldval, newval uint32) {
    39  	// Determine whether this change in status affects the idleness of the group.
    40  	// If this isn't a goroutine starting, stopping, durably blocking,
    41  	// or waking up after durably blocking, then return immediately without
    42  	// locking sg.mu.
    43  	//
    44  	// For example, stack growth (newstack) will changegstatus
    45  	// from _Grunning to _Gcopystack. This is uninteresting to synctest,
    46  	// but if stack growth occurs while sg.mu is held, we must not recursively lock.
    47  	totalDelta := 0
    48  	wasRunning := true
    49  	switch oldval {
    50  	case _Gdead:
    51  		wasRunning = false
    52  		totalDelta++
    53  	case _Gwaiting:
    54  		if gp.waitreason.isIdleInSynctest() {
    55  			wasRunning = false
    56  		}
    57  	}
    58  	isRunning := true
    59  	switch newval {
    60  	case _Gdead:
    61  		isRunning = false
    62  		totalDelta--
    63  	case _Gwaiting:
    64  		if gp.waitreason.isIdleInSynctest() {
    65  			isRunning = false
    66  		}
    67  	}
    68  	// It's possible for wasRunning == isRunning while totalDelta != 0;
    69  	// for example, if a new goroutine is created in a non-running state.
    70  	if wasRunning == isRunning && totalDelta == 0 {
    71  		return
    72  	}
    73  
    74  	lock(&sg.mu)
    75  	sg.total += totalDelta
    76  	if wasRunning != isRunning {
    77  		if isRunning {
    78  			sg.running++
    79  		} else {
    80  			sg.running--
    81  			if raceenabled && newval != _Gdead {
    82  				// Record that this goroutine parking happens before
    83  				// any subsequent Wait.
    84  				racereleasemergeg(gp, sg.raceaddr())
    85  			}
    86  		}
    87  	}
    88  	if sg.total < 0 {
    89  		fatal("total < 0")
    90  	}
    91  	if sg.running < 0 {
    92  		fatal("running < 0")
    93  	}
    94  	wake := sg.maybeWakeLocked()
    95  	unlock(&sg.mu)
    96  	if wake != nil {
    97  		goready(wake, 0)
    98  	}
    99  }
   100  
   101  // incActive increments the active-count for the group.
   102  // A group does not become durably blocked while the active-count is non-zero.
   103  func (sg *synctestGroup) incActive() {
   104  	lock(&sg.mu)
   105  	sg.active++
   106  	unlock(&sg.mu)
   107  }
   108  
   109  // decActive decrements the active-count for the group.
   110  func (sg *synctestGroup) decActive() {
   111  	lock(&sg.mu)
   112  	sg.active--
   113  	if sg.active < 0 {
   114  		throw("active < 0")
   115  	}
   116  	wake := sg.maybeWakeLocked()
   117  	unlock(&sg.mu)
   118  	if wake != nil {
   119  		goready(wake, 0)
   120  	}
   121  }
   122  
   123  // maybeWakeLocked returns a g to wake if the group is durably blocked.
   124  func (sg *synctestGroup) maybeWakeLocked() *g {
   125  	if sg.running > 0 || sg.active > 0 {
   126  		return nil
   127  	}
   128  	// Increment the group active count, since we've determined to wake something.
   129  	// The woken goroutine will decrement the count.
   130  	// We can't just call goready and let it increment sg.running,
   131  	// since we can't call goready with sg.mu held.
   132  	//
   133  	// Incrementing the active count here is only necessary if something has gone wrong,
   134  	// and a goroutine that we considered durably blocked wakes up unexpectedly.
   135  	// Two wakes happening at the same time leads to very confusing failure modes,
   136  	// so we take steps to avoid it happening.
   137  	sg.active++
   138  	next := sg.timers.wakeTime()
   139  	if next > 0 && next <= sg.now {
   140  		// A timer is scheduled to fire. Wake the root goroutine to handle it.
   141  		return sg.root
   142  	}
   143  	if gp := sg.waiter; gp != nil {
   144  		// A goroutine is blocked in Wait. Wake it.
   145  		return gp
   146  	}
   147  	// All goroutines in the group are durably blocked, and nothing has called Wait.
   148  	// Wake the root goroutine.
   149  	return sg.root
   150  }
   151  
   152  func (sg *synctestGroup) raceaddr() unsafe.Pointer {
   153  	// Address used to record happens-before relationships created by the group.
   154  	//
   155  	// Wait creates a happens-before relationship between itself and
   156  	// the blocking operations which caused other goroutines in the group to park.
   157  	return unsafe.Pointer(sg)
   158  }
   159  
   160  //go:linkname synctestRun internal/synctest.Run
   161  func synctestRun(f func()) {
   162  	if debug.asynctimerchan.Load() != 0 {
   163  		panic("synctest.Run not supported with asynctimerchan!=0")
   164  	}
   165  
   166  	gp := getg()
   167  	if gp.syncGroup != nil {
   168  		panic("synctest.Run called from within a synctest bubble")
   169  	}
   170  	gp.syncGroup = &synctestGroup{
   171  		total:   1,
   172  		running: 1,
   173  		root:    gp,
   174  	}
   175  	const synctestBaseTime = 946684800000000000 // midnight UTC 2000-01-01
   176  	gp.syncGroup.now = synctestBaseTime
   177  	gp.syncGroup.timers.syncGroup = gp.syncGroup
   178  	lockInit(&gp.syncGroup.mu, lockRankSynctest)
   179  	lockInit(&gp.syncGroup.timers.mu, lockRankTimers)
   180  	defer func() {
   181  		gp.syncGroup = nil
   182  	}()
   183  
   184  	fv := *(**funcval)(unsafe.Pointer(&f))
   185  	newproc(fv)
   186  
   187  	sg := gp.syncGroup
   188  	lock(&sg.mu)
   189  	sg.active++
   190  	for {
   191  		unlock(&sg.mu)
   192  		systemstack(func() {
   193  			// Clear gp.m.curg while running timers,
   194  			// so timer goroutines inherit their child race context from g0.
   195  			curg := gp.m.curg
   196  			gp.m.curg = nil
   197  			gp.syncGroup.timers.check(gp.syncGroup.now)
   198  			gp.m.curg = curg
   199  		})
   200  		gopark(synctestidle_c, nil, waitReasonSynctestRun, traceBlockSynctest, 0)
   201  		lock(&sg.mu)
   202  		if sg.active < 0 {
   203  			throw("active < 0")
   204  		}
   205  		next := sg.timers.wakeTime()
   206  		if next == 0 {
   207  			break
   208  		}
   209  		if next < sg.now {
   210  			throw("time went backwards")
   211  		}
   212  		sg.now = next
   213  	}
   214  
   215  	total := sg.total
   216  	unlock(&sg.mu)
   217  	if raceenabled {
   218  		// Establish a happens-before relationship between bubbled goroutines exiting
   219  		// and Run returning.
   220  		raceacquireg(gp, gp.syncGroup.raceaddr())
   221  	}
   222  	if total != 1 {
   223  		panic("deadlock: all goroutines in bubble are blocked")
   224  	}
   225  	if gp.timer != nil && gp.timer.isFake {
   226  		// Verify that we haven't marked this goroutine's sleep timer as fake.
   227  		// This could happen if something in Run were to call timeSleep.
   228  		throw("synctest root goroutine has a fake timer")
   229  	}
   230  }
   231  
   232  func synctestidle_c(gp *g, _ unsafe.Pointer) bool {
   233  	lock(&gp.syncGroup.mu)
   234  	canIdle := true
   235  	if gp.syncGroup.running == 0 && gp.syncGroup.active == 1 {
   236  		// All goroutines in the group have blocked or exited.
   237  		canIdle = false
   238  	} else {
   239  		gp.syncGroup.active--
   240  	}
   241  	unlock(&gp.syncGroup.mu)
   242  	return canIdle
   243  }
   244  
   245  //go:linkname synctestWait internal/synctest.Wait
   246  func synctestWait() {
   247  	gp := getg()
   248  	if gp.syncGroup == nil {
   249  		panic("goroutine is not in a bubble")
   250  	}
   251  	lock(&gp.syncGroup.mu)
   252  	// We use a syncGroup.waiting bool to detect simultaneous calls to Wait rather than
   253  	// checking to see if syncGroup.waiter is non-nil. This avoids a race between unlocking
   254  	// syncGroup.mu and setting syncGroup.waiter while parking.
   255  	if gp.syncGroup.waiting {
   256  		unlock(&gp.syncGroup.mu)
   257  		panic("wait already in progress")
   258  	}
   259  	gp.syncGroup.waiting = true
   260  	unlock(&gp.syncGroup.mu)
   261  	gopark(synctestwait_c, nil, waitReasonSynctestWait, traceBlockSynctest, 0)
   262  
   263  	lock(&gp.syncGroup.mu)
   264  	gp.syncGroup.active--
   265  	if gp.syncGroup.active < 0 {
   266  		throw("active < 0")
   267  	}
   268  	gp.syncGroup.waiter = nil
   269  	gp.syncGroup.waiting = false
   270  	unlock(&gp.syncGroup.mu)
   271  
   272  	// Establish a happens-before relationship on the activity of the now-blocked
   273  	// goroutines in the group.
   274  	if raceenabled {
   275  		raceacquireg(gp, gp.syncGroup.raceaddr())
   276  	}
   277  }
   278  
   279  func synctestwait_c(gp *g, _ unsafe.Pointer) bool {
   280  	lock(&gp.syncGroup.mu)
   281  	if gp.syncGroup.running == 0 && gp.syncGroup.active == 0 {
   282  		// This shouldn't be possible, since gopark increments active during unlockf.
   283  		throw("running == 0 && active == 0")
   284  	}
   285  	gp.syncGroup.waiter = gp
   286  	unlock(&gp.syncGroup.mu)
   287  	return true
   288  }
   289  
   290  //go:linkname synctest_acquire internal/synctest.acquire
   291  func synctest_acquire() any {
   292  	if sg := getg().syncGroup; sg != nil {
   293  		sg.incActive()
   294  		return sg
   295  	}
   296  	return nil
   297  }
   298  
   299  //go:linkname synctest_release internal/synctest.release
   300  func synctest_release(sg any) {
   301  	sg.(*synctestGroup).decActive()
   302  }
   303  
   304  //go:linkname synctest_inBubble internal/synctest.inBubble
   305  func synctest_inBubble(sg any, f func()) {
   306  	gp := getg()
   307  	if gp.syncGroup != nil {
   308  		panic("goroutine is already bubbled")
   309  	}
   310  	gp.syncGroup = sg.(*synctestGroup)
   311  	defer func() {
   312  		gp.syncGroup = nil
   313  	}()
   314  	f()
   315  }
   316
View as plain text