// Copyright 2024 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package runtime
import (
"unsafe"
)
// A synctestGroup is a group of goroutines started by synctest.Run.
type synctestGroup struct {
mu mutex
timers timers
now int64 // current fake time
root *g // caller of synctest.Run
waiter *g // caller of synctest.Wait
waiting bool // true if a goroutine is calling synctest.Wait
// The group is active (not blocked) so long as running > 0 || active > 0.
//
// running is the number of goroutines which are not "durably blocked":
// Goroutines which are either running, runnable, or non-durably blocked
// (for example, blocked in a syscall).
//
// active is used to keep the group from becoming blocked,
// even if all goroutines in the group are blocked.
// For example, park_m can choose to immediately unpark a goroutine after parking it.
// It increments the active count to keep the group active until it has determined
// that the park operation has completed.
total int // total goroutines
running int // non-blocked goroutines
active int // other sources of activity
}
// changegstatus is called when the non-lock status of a g changes.
// It is never called with a Gscanstatus.
func (sg *synctestGroup) changegstatus(gp *g, oldval, newval uint32) {
// Determine whether this change in status affects the idleness of the group.
// If this isn't a goroutine starting, stopping, durably blocking,
// or waking up after durably blocking, then return immediately without
// locking sg.mu.
//
// For example, stack growth (newstack) will changegstatus
// from _Grunning to _Gcopystack. This is uninteresting to synctest,
// but if stack growth occurs while sg.mu is held, we must not recursively lock.
totalDelta := 0
wasRunning := true
switch oldval {
case _Gdead:
wasRunning = false
totalDelta++
case _Gwaiting:
if gp.waitreason.isIdleInSynctest() {
wasRunning = false
}
}
isRunning := true
switch newval {
case _Gdead:
isRunning = false
totalDelta--
case _Gwaiting:
if gp.waitreason.isIdleInSynctest() {
isRunning = false
}
}
// It's possible for wasRunning == isRunning while totalDelta != 0;
// for example, if a new goroutine is created in a non-running state.
if wasRunning == isRunning && totalDelta == 0 {
return
}
lock(&sg.mu)
sg.total += totalDelta
if wasRunning != isRunning {
if isRunning {
sg.running++
} else {
sg.running--
if raceenabled && newval != _Gdead {
racereleasemergeg(gp, sg.raceaddr())
}
}
}
if sg.total < 0 {
fatal("total < 0")
}
if sg.running < 0 {
fatal("running < 0")
}
wake := sg.maybeWakeLocked()
unlock(&sg.mu)
if wake != nil {
goready(wake, 0)
}
}
// incActive increments the active-count for the group.
// A group does not become durably blocked while the active-count is non-zero.
func (sg *synctestGroup) incActive() {
lock(&sg.mu)
sg.active++
unlock(&sg.mu)
}
// decActive decrements the active-count for the group.
func (sg *synctestGroup) decActive() {
lock(&sg.mu)
sg.active--
if sg.active < 0 {
throw("active < 0")
}
wake := sg.maybeWakeLocked()
unlock(&sg.mu)
if wake != nil {
goready(wake, 0)
}
}
// maybeWakeLocked returns a g to wake if the group is durably blocked.
func (sg *synctestGroup) maybeWakeLocked() *g {
if sg.running > 0 || sg.active > 0 {
return nil
}
// Increment the group active count, since we've determined to wake something.
// The woken goroutine will decrement the count.
// We can't just call goready and let it increment sg.running,
// since we can't call goready with sg.mu held.
//
// Incrementing the active count here is only necessary if something has gone wrong,
// and a goroutine that we considered durably blocked wakes up unexpectedly.
// Two wakes happening at the same time leads to very confusing failure modes,
// so we take steps to avoid it happening.
sg.active++
if gp := sg.waiter; gp != nil {
// A goroutine is blocked in Wait. Wake it.
return gp
}
// All goroutines in the group are durably blocked, and nothing has called Wait.
// Wake the root goroutine.
return sg.root
}
func (sg *synctestGroup) raceaddr() unsafe.Pointer {
// Address used to record happens-before relationships created by the group.
//
// Wait creates a happens-before relationship between itself and
// the blocking operations which caused other goroutines in the group to park.
return unsafe.Pointer(sg)
}
//go:linkname synctestRun internal/synctest.Run
func synctestRun(f func()) {
if debug.asynctimerchan.Load() != 0 {
panic("synctest.Run not supported with asynctimerchan!=0")
}
gp := getg()
if gp.syncGroup != nil {
panic("synctest.Run called from within a synctest bubble")
}
gp.syncGroup = &synctestGroup{
total: 1,
running: 1,
root: gp,
}
const synctestBaseTime = 946684800000000000 // midnight UTC 2000-01-01
gp.syncGroup.now = synctestBaseTime
gp.syncGroup.timers.syncGroup = gp.syncGroup
lockInit(&gp.syncGroup.mu, lockRankSynctest)
lockInit(&gp.syncGroup.timers.mu, lockRankTimers)
defer func() {
gp.syncGroup = nil
}()
fv := *(**funcval)(unsafe.Pointer(&f))
newproc(fv)
sg := gp.syncGroup
lock(&sg.mu)
sg.active++
for {
if raceenabled {
raceacquireg(gp, gp.syncGroup.raceaddr())
}
unlock(&sg.mu)
systemstack(func() {
gp.syncGroup.timers.check(gp.syncGroup.now)
})
gopark(synctestidle_c, nil, waitReasonSynctestRun, traceBlockSynctest, 0)
lock(&sg.mu)
if sg.active < 0 {
throw("active < 0")
}
next := sg.timers.wakeTime()
if next == 0 {
break
}
if next < sg.now {
throw("time went backwards")
}
sg.now = next
}
total := sg.total
unlock(&sg.mu)
if total != 1 {
panic("deadlock: all goroutines in bubble are blocked")
}
if gp.timer != nil && gp.timer.isFake {
// Verify that we haven't marked this goroutine's sleep timer as fake.
// This could happen if something in Run were to call timeSleep.
throw("synctest root goroutine has a fake timer")
}
}
func synctestidle_c(gp *g, _ unsafe.Pointer) bool {
lock(&gp.syncGroup.mu)
canIdle := true
if gp.syncGroup.running == 0 && gp.syncGroup.active == 1 {
// All goroutines in the group have blocked or exited.
canIdle = false
} else {
gp.syncGroup.active--
}
unlock(&gp.syncGroup.mu)
return canIdle
}
//go:linkname synctestWait internal/synctest.Wait
func synctestWait() {
gp := getg()
if gp.syncGroup == nil {
panic("goroutine is not in a bubble")
}
lock(&gp.syncGroup.mu)
// We use a syncGroup.waiting bool to detect simultaneous calls to Wait rather than
// checking to see if syncGroup.waiter is non-nil. This avoids a race between unlocking
// syncGroup.mu and setting syncGroup.waiter while parking.
if gp.syncGroup.waiting {
unlock(&gp.syncGroup.mu)
panic("wait already in progress")
}
gp.syncGroup.waiting = true
unlock(&gp.syncGroup.mu)
gopark(synctestwait_c, nil, waitReasonSynctestWait, traceBlockSynctest, 0)
lock(&gp.syncGroup.mu)
gp.syncGroup.active--
if gp.syncGroup.active < 0 {
throw("active < 0")
}
gp.syncGroup.waiter = nil
gp.syncGroup.waiting = false
unlock(&gp.syncGroup.mu)
// Establish a happens-before relationship on the activity of the now-blocked
// goroutines in the group.
if raceenabled {
raceacquireg(gp, gp.syncGroup.raceaddr())
}
}
func synctestwait_c(gp *g, _ unsafe.Pointer) bool {
lock(&gp.syncGroup.mu)
if gp.syncGroup.running == 0 && gp.syncGroup.active == 0 {
// This shouldn't be possible, since gopark increments active during unlockf.
throw("running == 0 && active == 0")
}
gp.syncGroup.waiter = gp
unlock(&gp.syncGroup.mu)
return true
}
//go:linkname synctest_acquire internal/synctest.acquire
func synctest_acquire() any {
if sg := getg().syncGroup; sg != nil {
sg.incActive()
return sg
}
return nil
}
//go:linkname synctest_release internal/synctest.release
func synctest_release(sg any) {
sg.(*synctestGroup).decActive()
}
//go:linkname synctest_inBubble internal/synctest.inBubble
func synctest_inBubble(sg any, f func()) {
gp := getg()
if gp.syncGroup != nil {
panic("goroutine is already bubbled")
}
gp.syncGroup = sg.(*synctestGroup)
defer func() {
gp.syncGroup = nil
}()
f()
}