runtime: wake channel waiters after releasing locks

[rdon-key]

Schedule blocked channel senders and receivers only after releasing the channel and select locks.

With the cores scheduler, scheduleTask may allow the woken task to resume immediately on another core. Previously, channel operations could make a waiter runnable while the waking goroutine still held the same channel lock and, for select operations, chanSelectLock.

This PR changes trySend and tryRecv to return the task that should be woken. Their callers schedule that task only after releasing all relevant locks.

chanClose now collects blocked waiters, marks the channel closed, releases the channel lock, and then wakes the collected tasks.

Previous ordering

update channel state
schedule waiter
release channel/select locks

A waiter could immediately resume on another core and access the same channel before the locks were released.

New ordering

update channel state
release channel/select locks
schedule waiter

Testing

Tested on an RP2040 Pico with:

TinyGo 0.42.0-dev-0922e3e9
-scheduler=cores

Blocking multi-state select

• Reproducer:

c01_multistate_blocking_select_wake_order.go

Details

package main

import (
        "sync/atomic"
        "time"
)

const (
        testDuration = 3 * time.Minute
        reportEvery  = 10 * time.Second
)

var sent uint32
var received uint32

func fullChannel() chan uint32 {
        ch := make(chan uint32, 1)
        ch <- 0
        return ch
}

func main() {
        data := make(chan uint32)

        // Every blocker is permanently full, so these send cases can never proceed.
        b00 := fullChannel()
        b01 := fullChannel()
        b02 := fullChannel()
        b03 := fullChannel()
        b04 := fullChannel()
        b05 := fullChannel()
        b06 := fullChannel()
        b07 := fullChannel()
        b08 := fullChannel()
        b09 := fullChannel()
        b10 := fullChannel()
        b11 := fullChannel()
        b12 := fullChannel()
        b13 := fullChannel()
        b14 := fullChannel()
        b15 := fullChannel()

        // A single receiver immediately returns to the same channel after each wake.
        go func() {
                for {
                        <-data
                        atomic.AddUint32(&received, 1)
                }
        }()

        start := time.Now()

        // If sent/received stop changing while reports continue, the channel pair is
        // stuck rather than the whole runtime.
        go func() {
                for {
                        time.Sleep(reportEvery)
                        print(
                                "\nREPORT ms=", time.Since(start).Milliseconds(),
                                " sent=", atomic.LoadUint32(&sent),
                                " recv=", atomic.LoadUint32(&received),
                                "\n",
                        )
                }
        }()

        // Give the receiver time to enter the channel wait queue.
        time.Sleep(100 * time.Millisecond)

        deadline := start.Add(testDuration)

        for time.Now().Before(deadline) {
                value := atomic.LoadUint32(&sent) + 1

                // Deliberately blocking and multi-state: #5500 cannot apply.
                //
                // Keep data last. The current runtime unlocks channels in state order,
                // maximizing the interval between waking the receiver and unlocking data.
                select {
                case b00 <- value:
                case b01 <- value:
                case b02 <- value:
                case b03 <- value:
                case b04 <- value:
                case b05 <- value:
                case b06 <- value:
                case b07 <- value:
                case b08 <- value:
                case b09 <- value:
                case b10 <- value:
                case b11 <- value:
                case b12 <- value:
                case b13 <- value:
                case b14 <- value:
                case b15 <- value:
                case data <- value:
                }

                n := atomic.AddUint32(&sent, 1)
                if n%1000 == 0 {
                        print(".")
                }
        }

        // Let the receiver finish the final handoff before printing the result.
        for i := 0; i < 1000 && atomic.LoadUint32(&received) != atomic.LoadUint32(&sent); i++ {
                time.Sleep(time.Millisecond)
        }

        print(
                "\nDONE ms=", time.Since(start).Milliseconds(),
                " sent=", atomic.LoadUint32(&sent),
                " recv=", atomic.LoadUint32(&received),
                "\n",
        )
}

This test uses a blocking 17-case select and therefore does not use the non-blocking single-state fast path from #5500.

Unmodified dev:

3/3 runs froze
all freezes occurred within 40 seconds

With this change:

5/5 runs completed
3 minutes per run
approximately 3,000,000 channel handoffs per run
sent == received

Channel close

• Reproducer:

d02_close_receiver_reentry_stress.go

Details

package main

import (
        "sync/atomic"
        "time"
)

const (
        startDelay   = 2 * time.Second
        rounds       = 1000
        workers      = 8
        reentries    = 128
        settleDelay  = 2 * time.Millisecond
        roundTimeout = 2 * time.Second
)

var completedRounds uint32

func hammerClosedChannel(ch chan uint32) bool {
        for i := 0; i < reentries; i++ {
                value, ok := <-ch
                if ok || value != 0 {
                        return false
                }
        }
        return true
}

func receiverWorker(index int, jobs <-chan chan uint32, ready chan<- struct{}, done chan<- bool) {
        var never chan uint32

        for ch := range jobs {
                ready <- struct{}{}

                var value uint32
                var ok bool

                if index&1 == 0 {
                        value, ok = <-ch
                } else {
                        select {
                        case value, ok = <-ch:
                        case value, ok = <-never:
                        }
                }

                done <- !ok && value == 0 && hammerClosedChannel(ch)
        }
}

func waitReady(ready <-chan struct{}, timeout time.Duration) bool {
        timer := time.NewTimer(timeout)
        defer timer.Stop()

        for i := 0; i < workers; i++ {
                select {
                case <-ready:
                case <-timer.C:
                        return false
                }
        }
        return true
}

func waitDone(done <-chan bool, timeout time.Duration) (completed int, bad int, timedOut bool) {
        timer := time.NewTimer(timeout)
        defer timer.Stop()

        for completed < workers {
                select {
                case ok := <-done:
                        completed++
                        if !ok {
                                bad++
                        }
                case <-timer.C:
                        return completed, bad, true
                }
        }
        return completed, bad, false
}

func main() {
        time.Sleep(startDelay)
        println("starting chanClose receiver wake stress")

        jobs := make([]chan chan uint32, workers)
        ready := make(chan struct{}, workers)
        done := make(chan bool, workers)

        for i := 0; i < workers; i++ {
                jobs[i] = make(chan chan uint32)
                go receiverWorker(i, jobs[i], ready, done)
        }

        start := time.Now()

        go func() {
                for {
                        time.Sleep(time.Second)
                        print(
                                "\nREPORT ms=", time.Since(start).Milliseconds(),
                                " rounds=", atomic.LoadUint32(&completedRounds),
                                "\n",
                        )
                }
        }()

        for round := 1; round <= rounds; round++ {
                ch := make(chan uint32)

                for i := 0; i < workers; i++ {
                        jobs[i] <- ch
                }

                if !waitReady(ready, roundTimeout) {
                        print("\nFAIL ready timeout round=", round, "\n")
                        return
                }

                time.Sleep(settleDelay)
                close(ch)

                completed, bad, timedOut := waitDone(done, roundTimeout)
                if timedOut {
                        print(
                                "\nFAIL wake timeout round=", round,
                                " completed=", completed,
                                " bad=", bad,
                                "\n",
                        )
                        return
                }
                if bad != 0 {
                        print(
                                "\nFAIL result round=", round,
                                " bad=", bad,
                                "\n",
                        )
                        return
                }

                atomic.StoreUint32(&completedRounds, uint32(round))
                if round%50 == 0 {
                        print(".")
                }
        }

        print(
                "\nDONE ms=", time.Since(start).Milliseconds(),
                " rounds=", atomic.LoadUint32(&completedRounds),
                " receiver_wakes=", rounds*workers,
                " reentries=", rounds*workers*reentries,
                "\n",
        )
}

This test repeatedly wakes blocked receivers by closing their channel. The receivers immediately access the same closed channel again after being woken.

Unmodified dev:

3/3 runs froze
all runs stopped between rounds 50 and 99

With this change:

3/3 runs completed
1,000 rounds per run
8,000 receiver wakeups per run
1,024,000 closed-channel reentries per run

Relation to #5500

#5500 adds a fast path for non-blocking single-state selects.

The c01 reproducer uses a blocking multi-state select, so that fast path cannot apply. This PR addresses the waiter wake-up ordering in the common channel runtime path and is independent of the optimization in #5500.