mirror of
https://github.com/tinygo-org/tinygo.git
synced 2026-08-21 12:59:04 +00:00
internal/task (threads): save stack bounds instead of scanning under a lock
In order to scan stacks, the GC preempts all other threads and has them scan their own stack. This is somewhat expensive since all of these threads have to fight over a single lock. Instead, save the stack bounds and let the GC thread perform the scan. This also fixes a few other bugs I ran into: 1. The GC starts scanning before the world stops. This can cause it to miss some objects (and mistakenly free them) if memory is modified while stopping. 2. The GC does not wait for threads to resume. This can cause notifications to be misinterpreted due to signal nesting if the GC is re-run before all threads wake.
This commit is contained in:
@@ -23,16 +23,15 @@ type state struct {
|
|||||||
// is needed to be able to scan the stack.
|
// is needed to be able to scan the stack.
|
||||||
stackTop uintptr
|
stackTop uintptr
|
||||||
|
|
||||||
|
// Lowest address of the stack.
|
||||||
|
// This is populated when the thread is stopped by the GC.
|
||||||
|
stackBottom uintptr
|
||||||
|
|
||||||
// Next task in the activeTasks queue.
|
// Next task in the activeTasks queue.
|
||||||
QueueNext *Task
|
QueueNext *Task
|
||||||
|
|
||||||
// Semaphore to pause/resume the thread atomically.
|
// Semaphore to pause/resume the thread atomically.
|
||||||
pauseSem Semaphore
|
pauseSem Semaphore
|
||||||
|
|
||||||
// Semaphore used for stack scanning.
|
|
||||||
// We can't reuse pauseSem here since the thread might have been paused for
|
|
||||||
// other reasons (for example, because it was waiting on a channel).
|
|
||||||
gcSem Semaphore
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Goroutine counter, starting at 0 for the main goroutine.
|
// Goroutine counter, starting at 0 for the main goroutine.
|
||||||
@@ -96,6 +95,9 @@ func (t *Task) Resume() {
|
|||||||
t.state.pauseSem.Post()
|
t.state.pauseSem.Post()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// otherGoroutines is the total number of live goroutines minus one.
|
||||||
|
var otherGoroutines uint32
|
||||||
|
|
||||||
// Start a new OS thread.
|
// Start a new OS thread.
|
||||||
func start(fn uintptr, args unsafe.Pointer, stackSize uintptr) {
|
func start(fn uintptr, args unsafe.Pointer, stackSize uintptr) {
|
||||||
t := &Task{}
|
t := &Task{}
|
||||||
@@ -115,6 +117,7 @@ func start(fn uintptr, args unsafe.Pointer, stackSize uintptr) {
|
|||||||
}
|
}
|
||||||
t.state.QueueNext = activeTasks
|
t.state.QueueNext = activeTasks
|
||||||
activeTasks = t
|
activeTasks = t
|
||||||
|
otherGoroutines++
|
||||||
activeTaskLock.Unlock()
|
activeTaskLock.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -135,6 +138,7 @@ func taskExited(t *Task) {
|
|||||||
break
|
break
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
otherGoroutines--
|
||||||
activeTaskLock.Unlock()
|
activeTaskLock.Unlock()
|
||||||
|
|
||||||
// Sanity check.
|
// Sanity check.
|
||||||
@@ -143,9 +147,42 @@ func taskExited(t *Task) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Futex to wait on until all tasks have finished scanning the stack.
|
// scanWaitGroup is used to wait on until all threads have finished the current state transition.
|
||||||
// This is basically a sync.WaitGroup.
|
var scanWaitGroup waitGroup
|
||||||
var scanDoneFutex Futex
|
|
||||||
|
type waitGroup struct {
|
||||||
|
f Futex
|
||||||
|
}
|
||||||
|
|
||||||
|
func initWaitGroup(n uint32) waitGroup {
|
||||||
|
var wg waitGroup
|
||||||
|
wg.f.Store(n)
|
||||||
|
return wg
|
||||||
|
}
|
||||||
|
|
||||||
|
func (wg *waitGroup) done() {
|
||||||
|
if wg.f.Add(^uint32(0)) == 0 {
|
||||||
|
wg.f.WakeAll()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (wg *waitGroup) wait() {
|
||||||
|
for {
|
||||||
|
val := wg.f.Load()
|
||||||
|
if val == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
wg.f.Wait(val)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// gcState is used to track and notify threads when the GC is stopping/resuming.
|
||||||
|
var gcState Futex
|
||||||
|
|
||||||
|
const (
|
||||||
|
gcStateResumed = iota
|
||||||
|
gcStateStopped
|
||||||
|
)
|
||||||
|
|
||||||
// GC scan phase. Because we need to stop the world while scanning, this kinda
|
// GC scan phase. Because we need to stop the world while scanning, this kinda
|
||||||
// needs to be done in the tasks package.
|
// needs to be done in the tasks package.
|
||||||
@@ -155,65 +192,71 @@ var scanDoneFutex Futex
|
|||||||
func GCStopWorldAndScan() {
|
func GCStopWorldAndScan() {
|
||||||
current := Current()
|
current := Current()
|
||||||
|
|
||||||
// Don't allow new goroutines to be started while pausing/resuming threads
|
// NOTE: This does not need to be atomic.
|
||||||
// in the stop-the-world phase.
|
if gcState.Load() == gcStateResumed {
|
||||||
activeTaskLock.Lock()
|
// Don't allow new goroutines to be started while pausing/resuming threads
|
||||||
|
// in the stop-the-world phase.
|
||||||
|
activeTaskLock.Lock()
|
||||||
|
|
||||||
// Pause all other threads.
|
// Wait for threads to finish resuming.
|
||||||
numOtherThreads := uint32(0)
|
scanWaitGroup.wait()
|
||||||
for t := activeTasks; t != nil; t = t.state.QueueNext {
|
|
||||||
if t != current {
|
// Change the gc state to stopped.
|
||||||
numOtherThreads++
|
// NOTE: This does not need to be atomic.
|
||||||
tinygo_task_send_gc_signal(t.state.thread)
|
gcState.Store(gcStateStopped)
|
||||||
|
|
||||||
|
// Set the number of threads to wait for.
|
||||||
|
scanWaitGroup = initWaitGroup(otherGoroutines)
|
||||||
|
|
||||||
|
// Pause all other threads.
|
||||||
|
for t := activeTasks; t != nil; t = t.state.QueueNext {
|
||||||
|
if t != current {
|
||||||
|
tinygo_task_send_gc_signal(t.state.thread)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Wait for the threads to finish stopping.
|
||||||
|
scanWaitGroup.wait()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Store the number of threads to wait for in the futex.
|
// Scan other thread stacks.
|
||||||
// This is the equivalent of doing an initial wg.Add(numOtherThreads).
|
for t := activeTasks; t != nil; t = t.state.QueueNext {
|
||||||
scanDoneFutex.Store(numOtherThreads)
|
if t != current {
|
||||||
|
markRoots(t.state.stackBottom, t.state.stackTop)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Scan the current stack, and all current registers.
|
// Scan the current stack, and all current registers.
|
||||||
scanCurrentStack()
|
scanCurrentStack()
|
||||||
|
|
||||||
// Wake each paused thread for the first time so it will scan the stack.
|
|
||||||
for t := activeTasks; t != nil; t = t.state.QueueNext {
|
|
||||||
if t != current {
|
|
||||||
t.state.gcSem.Post()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Wait until all threads have finished scanning their stack.
|
|
||||||
// This is the equivalent of wg.Wait()
|
|
||||||
for {
|
|
||||||
val := scanDoneFutex.Load()
|
|
||||||
if val == 0 {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
scanDoneFutex.Wait(val)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Scan all globals (implemented in the runtime).
|
// Scan all globals (implemented in the runtime).
|
||||||
gcScanGlobals()
|
gcScanGlobals()
|
||||||
}
|
}
|
||||||
|
|
||||||
// After the GC is done scanning, resume all other threads.
|
// After the GC is done scanning, resume all other threads.
|
||||||
//
|
|
||||||
// This must only be called after a GCStopWorldAndScan call.
|
|
||||||
func GCResumeWorld() {
|
func GCResumeWorld() {
|
||||||
current := Current()
|
// NOTE: This does not need to be atomic.
|
||||||
|
if gcState.Load() == gcStateResumed {
|
||||||
// Wake each paused thread for the second time, so they will resume normal
|
// This is already resumed.
|
||||||
// operation.
|
return
|
||||||
for t := activeTasks; t != nil; t = t.state.QueueNext {
|
|
||||||
if t != current {
|
|
||||||
t.state.gcSem.Post()
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Set the wait group to track resume progress.
|
||||||
|
scanWaitGroup = initWaitGroup(otherGoroutines)
|
||||||
|
|
||||||
|
// Set the state to resumed.
|
||||||
|
gcState.Store(gcStateResumed)
|
||||||
|
|
||||||
|
// Wake all of the stopped threads.
|
||||||
|
gcState.WakeAll()
|
||||||
|
|
||||||
// Allow goroutines to start and exit again.
|
// Allow goroutines to start and exit again.
|
||||||
activeTaskLock.Unlock()
|
activeTaskLock.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
//go:linkname markRoots runtime.markRoots
|
||||||
|
func markRoots(start, end uintptr)
|
||||||
|
|
||||||
// Scan globals, implemented in the runtime package.
|
// Scan globals, implemented in the runtime package.
|
||||||
func gcScanGlobals()
|
func gcScanGlobals()
|
||||||
|
|
||||||
@@ -221,34 +264,27 @@ var stackScanLock PMutex
|
|||||||
|
|
||||||
//export tinygo_task_gc_pause
|
//export tinygo_task_gc_pause
|
||||||
func tingyo_task_gc_pause(sig int32) {
|
func tingyo_task_gc_pause(sig int32) {
|
||||||
// Wait until we get the signal to start scanning the stack.
|
// Write the entrty stack pointer to the state.
|
||||||
Current().state.gcSem.Wait()
|
Current().state.stackBottom = uintptr(stacksave())
|
||||||
|
|
||||||
// Scan the thread stack.
|
// Notify the GC that we are stopped.
|
||||||
// Only scan a single thread stack at a time, because the GC marking phase
|
scanWaitGroup.done()
|
||||||
// doesn't support parallelism.
|
|
||||||
// TODO: it may be possible to call markRoots directly (without saving
|
|
||||||
// registers) since we are in a signal handler that already saved a bunch of
|
|
||||||
// registers. This is an optimization left for a future time.
|
|
||||||
stackScanLock.Lock()
|
|
||||||
scanCurrentStack()
|
|
||||||
stackScanLock.Unlock()
|
|
||||||
|
|
||||||
// Equivalent of wg.Done(): subtract one from the futex and if the result is
|
// Wait for the GC to resume.
|
||||||
// 0 (meaning we were the last in the waitgroup), wake the waiting thread.
|
for gcState.Load() == gcStateStopped {
|
||||||
n := uint32(1)
|
gcState.Wait(gcStateStopped)
|
||||||
if scanDoneFutex.Add(-n) == 0 {
|
|
||||||
scanDoneFutex.Wake()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Wait until we get the signal we can resume normally (after the mark phase
|
// Notify the GC that we have resumed.
|
||||||
// has finished).
|
scanWaitGroup.done()
|
||||||
Current().state.gcSem.Wait()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
//go:export tinygo_scanCurrentStack
|
//go:export tinygo_scanCurrentStack
|
||||||
func scanCurrentStack()
|
func scanCurrentStack()
|
||||||
|
|
||||||
|
//go:linkname stacksave runtime.stacksave
|
||||||
|
func stacksave() unsafe.Pointer
|
||||||
|
|
||||||
// Return the highest address of the current stack.
|
// Return the highest address of the current stack.
|
||||||
func StackTop() uintptr {
|
func StackTop() uintptr {
|
||||||
return Current().state.stackTop
|
return Current().state.stackTop
|
||||||
|
|||||||
+2
-24
@@ -31,10 +31,6 @@ var zeroSizedAlloc uint8
|
|||||||
|
|
||||||
var gcLock task.PMutex
|
var gcLock task.PMutex
|
||||||
|
|
||||||
// Normally false, set to true during a GC scan when all other threads get
|
|
||||||
// paused.
|
|
||||||
var needsResumeWorld bool
|
|
||||||
|
|
||||||
func initHeap() {
|
func initHeap() {
|
||||||
libgc_init()
|
libgc_init()
|
||||||
|
|
||||||
@@ -48,20 +44,8 @@ func gcInit()
|
|||||||
|
|
||||||
//export tinygo_runtime_bdwgc_callback
|
//export tinygo_runtime_bdwgc_callback
|
||||||
func gcCallback() {
|
func gcCallback() {
|
||||||
if hasParallelism && needsResumeWorld {
|
|
||||||
// Should never happen, check for it anyway.
|
|
||||||
runtimePanic("gc: world already stopped")
|
|
||||||
}
|
|
||||||
|
|
||||||
// Mark globals and all stacks, and stop the world if we're using threading.
|
// Mark globals and all stacks, and stop the world if we're using threading.
|
||||||
gcMarkReachable()
|
gcMarkReachable()
|
||||||
|
|
||||||
// If we use a scheduler with parallelism (the threads scheduler for
|
|
||||||
// example), we need to call gcResumeWorld() after scanning has finished.
|
|
||||||
if hasParallelism {
|
|
||||||
// Note that we need to resume the world after finishing the GC call.
|
|
||||||
needsResumeWorld = true
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func markRoots(start, end uintptr) {
|
func markRoots(start, end uintptr) {
|
||||||
@@ -87,7 +71,6 @@ func alloc(size uintptr, layout unsafe.Pointer) unsafe.Pointer {
|
|||||||
}
|
}
|
||||||
|
|
||||||
gcLock.Lock()
|
gcLock.Lock()
|
||||||
needsResumeWorld = false
|
|
||||||
var ptr unsafe.Pointer
|
var ptr unsafe.Pointer
|
||||||
if layout == gclayout.NoPtrs.AsPtr() {
|
if layout == gclayout.NoPtrs.AsPtr() {
|
||||||
// This object is entirely pointer free, for example make([]int, ...).
|
// This object is entirely pointer free, for example make([]int, ...).
|
||||||
@@ -104,9 +87,7 @@ func alloc(size uintptr, layout unsafe.Pointer) unsafe.Pointer {
|
|||||||
// Memory returned from libgc_malloc has already been zeroed, so nothing
|
// Memory returned from libgc_malloc has already been zeroed, so nothing
|
||||||
// to do here.
|
// to do here.
|
||||||
}
|
}
|
||||||
if needsResumeWorld {
|
gcResumeWorld()
|
||||||
gcResumeWorld()
|
|
||||||
}
|
|
||||||
gcLock.Unlock()
|
gcLock.Unlock()
|
||||||
if ptr == nil {
|
if ptr == nil {
|
||||||
runtimePanic("gc: out of memory")
|
runtimePanic("gc: out of memory")
|
||||||
@@ -121,11 +102,8 @@ func free(ptr unsafe.Pointer) {
|
|||||||
|
|
||||||
func GC() {
|
func GC() {
|
||||||
gcLock.Lock()
|
gcLock.Lock()
|
||||||
needsResumeWorld = false
|
|
||||||
libgc_gcollect()
|
libgc_gcollect()
|
||||||
if needsResumeWorld {
|
gcResumeWorld()
|
||||||
gcResumeWorld()
|
|
||||||
}
|
|
||||||
gcLock.Unlock()
|
gcLock.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user