diff --git a/.github/workflows/llgo.yml b/.github/workflows/llgo.yml index dadff854d0..0e673b3d60 100644 --- a/.github/workflows/llgo.yml +++ b/.github/workflows/llgo.yml @@ -481,7 +481,7 @@ jobs: GOOS=js GOARCH=wasm llgo build -o "$RUNNER_TEMP/runtime-js.wasm" ./internal/build/testdata/wasm-runtime GOOS=wasip1 GOARCH=wasm llgo build -o "$RUNNER_TEMP/runtime-wasip1.wasm" ./internal/build/testdata/wasm-runtime LLGO_WASI_THREADS=1 GOOS=wasip1 GOARCH=wasm llgo build \ - -o "$RUNNER_TEMP/runtime-wasip1-threads.wasm" ./internal/build/testdata/wasm-runtime + -o "$RUNNER_TEMP/runtime-wasip1-threads.wasm" ./internal/build/testdata/wasm-blocking test "$(wasmtime run -W exceptions=y "$RUNNER_TEMP/runtime-wasip1.wasm" 2>&1)" = "wasip1" GOOS=js GOARCH=wasm llgo build -o "$RUNNER_TEMP/wasm-scheduler-go.mjs" ./internal/build/testdata/wasm-scheduler run_wasm_scheduler "$RUNNER_TEMP/wasm-scheduler-go.mjs" diff --git a/internal/build/source_patch_test.go b/internal/build/source_patch_test.go index 84d214f759..9aa0b9c1b4 100644 --- a/internal/build/source_patch_test.go +++ b/internal/build/source_patch_test.go @@ -245,6 +245,39 @@ func boolToUint8(bool) uint8 } } +func TestBuildSourcePatchOverlayForGo124HashTrieMap(t *testing.T) { + goroot := t.TempDir() + syncDir := filepath.Join(goroot, "src", "internal", "sync") + hashTrieMap := filepath.Join(syncDir, "hashtriemap.go") + mustWriteFile(t, hashTrieMap, `package sync + +type HashTrieMap[K comparable, V any] struct{} + +func (ht *HashTrieMap[K, V]) CompareAndSwap(key K, old, new V) bool { + return false +} +`) + + overlay, _, err := buildSourcePatchOverlayForGOROOT(nil, env.LLGoRuntimeDir(), goroot, sourcePatchBuildContext{ + goos: runtime.GOOS, + goarch: runtime.GOARCH, + goversion: "go1.24.2", + }) + if err != nil { + t.Fatal(err) + } + + patch := filepath.Join(syncDir, "z_llgo_patch_hashtriemap.go") + if src, ok := overlay[patch]; !ok { + t.Fatalf("missing source patch file %s", patch) + } else if !strings.Contains(string(src), "type HashTrieMap") { + t.Fatalf("source patch file %s does not contain HashTrieMap replacement", patch) + } + if stdSrc := string(overlay[hashTrieMap]); strings.Contains(stdSrc, "type HashTrieMap") { + t.Fatalf("stub overlay for internal/sync still contains HashTrieMap: %s", stdSrc) + } +} + func TestGo126PayloadsUseSourcePatchInsteadOfAltPkg(t *testing.T) { for _, pkgPath := range []string{"internal/sync", "crypto/internal/constanttime"} { if !llruntime.HasSourcePatchPkg(pkgPath) { diff --git a/internal/build/testdata/wasm-blocking/main.go b/internal/build/testdata/wasm-blocking/main.go new file mode 100644 index 0000000000..0a8a7fae5d --- /dev/null +++ b/internal/build/testdata/wasm-blocking/main.go @@ -0,0 +1,63 @@ +package main + +import ( + "sync" + "sync/atomic" +) + +func main() { + values := make(chan int) + var wg sync.WaitGroup + var mu sync.Mutex + wg.Add(1) + go func() { + mu.Lock() + values <- 42 + mu.Unlock() + wg.Done() + }() + if value := <-values; value != 42 { + panic("channel value mismatch") + } + wg.Wait() + + var rw sync.RWMutex + rw.RLock() + rw.RUnlock() + rw.Lock() + rw.Unlock() + + cond := sync.NewCond(&mu) + started := make(chan struct{}) + ready := false + wg.Add(1) + go func() { + mu.Lock() + close(started) + for !ready { + cond.Wait() + } + mu.Unlock() + wg.Done() + }() + <-started + mu.Lock() + ready = true + cond.Signal() + mu.Unlock() + wg.Wait() + + var once sync.Once + once.Do(func() {}) + var pool sync.Pool + pool.Put("value") + if pool.Get() != "value" { + panic("sync.Pool value mismatch") + } + var atomicValue atomic.Value + atomicValue.Store("value") + if atomicValue.Load() != "value" { + panic("atomic.Value mismatch") + } + println("wasm blocking primitives ok") +} diff --git a/internal/build/testdata/wasm-scheduler/main.go b/internal/build/testdata/wasm-scheduler/main.go index 1fbc440c45..a39ee5a94b 100644 --- a/internal/build/testdata/wasm-scheduler/main.go +++ b/internal/build/testdata/wasm-scheduler/main.go @@ -2,6 +2,8 @@ package main import ( "runtime" + "sync" + "sync/atomic" "unsafe" ) @@ -137,9 +139,252 @@ func main() { panic("not all goroutines ran") } testGoroutineLifecycle() + testBlockingPrimitives() println("wasm scheduler ok") } +func testBlockingPrimitives() { + testChannelsAndSelect() + testWaitGroup() + testMutexes() + testCond() + testSyncHelpers() + testSyncMap() +} + +func testChannelsAndSelect() { + values := make(chan int) + ack := make(chan struct{}) + go func() { + values <- 41 + close(ack) + }() + if value := <-values; value != 41 { + panic("unexpected channel value") + } + <-ack + + buffered := make(chan int, 2) + buffered <- 1 + buffered <- 2 + if len(buffered) != 2 || cap(buffered) != 2 { + panic("buffered channel size mismatch") + } + if <-buffered != 1 || <-buffered != 2 { + panic("buffered channel order mismatch") + } + + left := make(chan int) + right := make(chan int) + go func() { + right <- 42 + }() + select { + case <-left: + panic("select chose a blocked channel") + case value := <-right: + if value != 42 { + panic("unexpected select value") + } + } + + selected := make(chan int) + go func() { + select { + case value := <-left: + selected <- value + case value := <-right: + selected <- value + } + }() + runtime.Gosched() + left <- 43 + if value := <-selected; value != 43 { + panic("blocked select chose the wrong channel") + } + + select { + case <-right: + panic("non-blocking select chose a blocked channel") + default: + } + + close(right) + if value, ok := <-right; value != 0 || ok { + panic("closed channel receive mismatch") + } +} + +func testWaitGroup() { + var wg sync.WaitGroup + count := 0 + wg.Add(2) + go func() { + count++ + wg.Done() + }() + go func() { + runtime.Gosched() + count++ + wg.Done() + }() + wg.Wait() + if count != 2 { + panic("WaitGroup returned too early") + } + + wg.Add(1) + go wg.Done() + wg.Wait() +} + +func testMutexes() { + var mu sync.Mutex + started := make(chan struct{}) + finished := make(chan struct{}) + value := 0 + mu.Lock() + go func() { + close(started) + mu.Lock() + value = 1 + mu.Unlock() + close(finished) + }() + <-started + mu.Unlock() + <-finished + if value != 1 { + panic("Mutex waiter did not run") + } + if !mu.TryLock() { + panic("Mutex.TryLock failed") + } + mu.Unlock() + + var rw sync.RWMutex + rw.RLock() + finished = make(chan struct{}) + go func() { + rw.Lock() + value = 2 + rw.Unlock() + close(finished) + }() + runtime.Gosched() + rw.RUnlock() + <-finished + if value != 2 { + panic("RWMutex waiter did not run") + } +} + +func testCond() { + var mu sync.Mutex + cond := sync.NewCond(&mu) + arrived := make(chan struct{}, 2) + done := make(chan struct{}, 2) + ready := false + for i := 0; i < 2; i++ { + go func() { + mu.Lock() + arrived <- struct{}{} + for !ready { + cond.Wait() + } + mu.Unlock() + done <- struct{}{} + }() + } + <-arrived + <-arrived + + mu.Lock() + ready = true + cond.Signal() + mu.Unlock() + <-done + select { + case <-done: + panic("Cond.Signal woke more than one waiter") + default: + } + + mu.Lock() + cond.Broadcast() + mu.Unlock() + <-done +} + +func testSyncHelpers() { + var once sync.Once + var wg sync.WaitGroup + count := 0 + wg.Add(2) + for i := 0; i < 2; i++ { + go func() { + once.Do(func() { + count++ + }) + wg.Done() + }() + } + wg.Wait() + if count != 1 { + panic("sync.Once ran more than once") + } + + var pool sync.Pool + pool.Put("pooled") + if value := pool.Get(); value != "pooled" { + panic("sync.Pool value mismatch") + } + + var value atomic.Value + value.Store("before") + if old := value.Swap("after"); old != "before" || value.Load() != "after" { + panic("atomic.Value swap mismatch") + } +} + +func testSyncMap() { + var m sync.Map + if _, loaded := m.LoadOrStore("key", 1); loaded { + panic("sync.Map unexpectedly loaded a missing key") + } + if value, loaded := m.Load("key"); !loaded || value != 1 { + panic("sync.Map load mismatch") + } + if !m.CompareAndSwap("key", 1, 2) { + panic("sync.Map compare-and-swap failed") + } + if previous, loaded := m.Swap("key", 3); !loaded || previous != 2 { + panic("sync.Map swap mismatch") + } + count := 0 + m.Range(func(key, value any) bool { + if key != "key" || value != 3 { + panic("sync.Map range mismatch") + } + count++ + return true + }) + if count != 1 { + panic("sync.Map range count mismatch") + } + if !m.CompareAndDelete("key", 3) { + panic("sync.Map compare-and-delete failed") + } + if _, loaded := m.LoadAndDelete("key"); loaded { + panic("sync.Map retained a deleted key") + } + m.Store("clear", 4) + m.Clear() + if _, loaded := m.Load("clear"); loaded { + panic("sync.Map clear failed") + } +} + func testGoroutineLifecycle() { const count = 5000 for i := 1; i <= count; i++ { diff --git a/runtime/_patch/internal/sync/hashtriemap.go b/runtime/_patch/internal/sync/hashtriemap.go index c6a2f12918..22ba1e736e 100644 --- a/runtime/_patch/internal/sync/hashtriemap.go +++ b/runtime/_patch/internal/sync/hashtriemap.go @@ -2,7 +2,7 @@ // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. -//go:build go1.26 +//go:build go1.24 //llgo:skipall package sync @@ -108,15 +108,17 @@ func (ht *HashTrieMap[K, V]) Swap(key K, new V) (previous V, loaded bool) { } func (ht *HashTrieMap[K, V]) CompareAndSwap(key K, old, new V) bool { + var swapped bool + ht.compareAndSwap(&swapped, key, old, new) + return swapped +} + +func (ht *HashTrieMap[K, V]) compareAndSwap(swapped *bool, key K, old, new V) { ht.mu.Lock() defer ht.mu.Unlock() - if i := ht.findIndex(key); i < 0 { - return false - } else if !hashTrieValueEqual(ht.m[i].value, old) { - return false - } else { + if i := ht.findIndex(key); i >= 0 && hashTrieValueEqual(ht.m[i].value, old) { ht.m[i].value = new - return true + *swapped = true } } @@ -135,15 +137,17 @@ func (ht *HashTrieMap[K, V]) Delete(key K) { } func (ht *HashTrieMap[K, V]) CompareAndDelete(key K, old V) bool { + var deleted bool + ht.compareAndDelete(&deleted, key, old) + return deleted +} + +func (ht *HashTrieMap[K, V]) compareAndDelete(deleted *bool, key K, old V) { ht.mu.Lock() defer ht.mu.Unlock() - if i := ht.findIndex(key); i < 0 { - return false - } else if !hashTrieValueEqual(ht.m[i].value, old) { - return false - } else { + if i := ht.findIndex(key); i >= 0 && hashTrieValueEqual(ht.m[i].value, old) { ht.deleteIndex(i) - return true + *deleted = true } } @@ -169,14 +173,18 @@ func (ht *HashTrieMap[K, V]) Clear() { } func (ht *HashTrieMap[K, V]) snapshot() []hashTrieEntry[K, V] { + var entries []hashTrieEntry[K, V] + ht.snapshotInto(&entries) + return entries +} + +func (ht *HashTrieMap[K, V]) snapshotInto(entries *[]hashTrieEntry[K, V]) { ht.mu.Lock() defer ht.mu.Unlock() - if len(ht.m) == 0 { - return nil + if len(ht.m) != 0 { + *entries = make([]hashTrieEntry[K, V], len(ht.m)) + copy(*entries, ht.m) } - entries := make([]hashTrieEntry[K, V], len(ht.m)) - copy(entries, ht.m) - return entries } func hashTrieValueEqual[V any](a, b V) bool { diff --git a/runtime/_patch/internal/sync/mutex.go b/runtime/_patch/internal/sync/mutex.go index 0cca2f2a56..4e00a88cd1 100644 --- a/runtime/_patch/internal/sync/mutex.go +++ b/runtime/_patch/internal/sync/mutex.go @@ -2,7 +2,7 @@ // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. -//go:build go1.26 +//go:build go1.24 package sync diff --git a/runtime/_patch/internal/sync/runtime.go b/runtime/_patch/internal/sync/runtime.go index f2e0a85c39..8abbfdd714 100644 --- a/runtime/_patch/internal/sync/runtime.go +++ b/runtime/_patch/internal/sync/runtime.go @@ -2,7 +2,7 @@ // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. -//go:build go1.26 +//go:build go1.24 package sync diff --git a/runtime/internal/clite/pthread/sync/sync.go b/runtime/internal/clite/pthread/sync/sync.go index 688c88303a..9074c5d5cb 100644 --- a/runtime/internal/clite/pthread/sync/sync.go +++ b/runtime/internal/clite/pthread/sync/sync.go @@ -80,10 +80,10 @@ type MutexAttr struct { } // llgo:link (*MutexAttr).Init C.pthread_mutexattr_init -func (a *MutexAttr) Init(attr *MutexAttr) c.Int { return 0 } +func (a *MutexAttr) Init() c.Int { return 0 } // llgo:link (*MutexAttr).Destroy C.pthread_mutexattr_destroy -func (a *MutexAttr) Destroy() {} +func (a *MutexAttr) Destroy() c.Int { return 0 } // llgo:link (*MutexAttr).SetType C.pthread_mutexattr_settype func (a *MutexAttr) SetType(typ MutexType) c.Int { return 0 } @@ -142,10 +142,10 @@ type RWLockAttr struct { } // llgo:link (*RWLockAttr).Init C.pthread_rwlockattr_init -func (a *RWLockAttr) Init(attr *RWLockAttr) c.Int { return 0 } +func (a *RWLockAttr) Init() c.Int { return 0 } // llgo:link (*RWLockAttr).Destroy C.pthread_rwlockattr_destroy -func (a *RWLockAttr) Destroy() {} +func (a *RWLockAttr) Destroy() c.Int { return 0 } // llgo:link (*RWLockAttr).SetPShared C.pthread_rwlockattr_setpshared func (a *RWLockAttr) SetPShared(pshared c.Int) c.Int { return 0 } @@ -222,10 +222,10 @@ type CondAttr struct { } // llgo:link (*CondAttr).Init C.pthread_condattr_init -func (a *CondAttr) Init(attr *CondAttr) c.Int { return 0 } +func (a *CondAttr) Init() c.Int { return 0 } // llgo:link (*CondAttr).Destroy C.pthread_condattr_destroy -func (a *CondAttr) Destroy() {} +func (a *CondAttr) Destroy() c.Int { return 0 } // // llgo:link (*CondAttr).SetClock C.pthread_condattr_setclock // func (a *CondAttr) SetClock(clock time.ClockidT) c.Int { return 0 } diff --git a/runtime/internal/clite/pthread/sync/sync_test.go b/runtime/internal/clite/pthread/sync/sync_test.go new file mode 100644 index 0000000000..697508244e --- /dev/null +++ b/runtime/internal/clite/pthread/sync/sync_test.go @@ -0,0 +1,25 @@ +package sync + +import "testing" + +func TestAttrMethods(t *testing.T) { + for name, initDestroy := range map[string]func() (int32, int32){ + "mutex": func() (int32, int32) { + var attr MutexAttr + return int32(attr.Init()), int32(attr.Destroy()) + }, + "rwlock": func() (int32, int32) { + var attr RWLockAttr + return int32(attr.Init()), int32(attr.Destroy()) + }, + "cond": func() (int32, int32) { + var attr CondAttr + return int32(attr.Init()), int32(attr.Destroy()) + }, + } { + initResult, destroyResult := initDestroy() + if initResult != 0 || destroyResult != 0 { + t.Errorf("%s attribute lifecycle returned (%d, %d)", name, initResult, destroyResult) + } + } +} diff --git a/runtime/internal/lib/runtime/sema_llgo.go b/runtime/internal/lib/runtime/sema_llgo.go index f7ab6c3434..0bb1fb2667 100644 --- a/runtime/internal/lib/runtime/sema_llgo.go +++ b/runtime/internal/lib/runtime/sema_llgo.go @@ -1,4 +1,4 @@ -//go:build darwin || linux +//go:build darwin || linux || (llgo && wasip1 && wasm && llgo.wasi_threads) package runtime @@ -99,8 +99,7 @@ func sync_runtime_SemacquireRWMutex(addr *uint32, _ bool, _ int) { semaAcquire(addr) } -//go:linkname sync_runtime_SemacquireWaitGroup sync.runtime_SemacquireWaitGroup -func sync_runtime_SemacquireWaitGroup(addr *uint32, _ bool) { +func syncWaitGroupAcquire(addr *uint32) { semaAcquire(addr) } diff --git a/runtime/internal/lib/runtime/sema_waitgroup_go124_llgo.go b/runtime/internal/lib/runtime/sema_waitgroup_go124_llgo.go new file mode 100644 index 0000000000..6765f7e863 --- /dev/null +++ b/runtime/internal/lib/runtime/sema_waitgroup_go124_llgo.go @@ -0,0 +1,10 @@ +//go:build (darwin || linux || (llgo && wasm)) && !go1.25 + +package runtime + +import _ "unsafe" + +//go:linkname sync_runtime_SemacquireWaitGroup sync.runtime_SemacquireWaitGroup +func sync_runtime_SemacquireWaitGroup(addr *uint32) { + syncWaitGroupAcquire(addr) +} diff --git a/runtime/internal/lib/runtime/sema_waitgroup_go125_llgo.go b/runtime/internal/lib/runtime/sema_waitgroup_go125_llgo.go new file mode 100644 index 0000000000..d019ea83b0 --- /dev/null +++ b/runtime/internal/lib/runtime/sema_waitgroup_go125_llgo.go @@ -0,0 +1,10 @@ +//go:build (darwin || linux || (llgo && wasm)) && go1.25 + +package runtime + +import _ "unsafe" + +//go:linkname sync_runtime_SemacquireWaitGroup sync.runtime_SemacquireWaitGroup +func sync_runtime_SemacquireWaitGroup(addr *uint32, _ bool) { + syncWaitGroupAcquire(addr) +} diff --git a/runtime/internal/lib/runtime/sema_wasm_llgo.go b/runtime/internal/lib/runtime/sema_wasm_llgo.go new file mode 100644 index 0000000000..3370596563 --- /dev/null +++ b/runtime/internal/lib/runtime/sema_wasm_llgo.go @@ -0,0 +1,315 @@ +//go:build llgo && wasm && !(wasip1 && llgo.wasi_threads) + +package runtime + +import ( + "unsafe" + + latomic "github.com/goplus/llgo/runtime/internal/lib/sync/atomic" + llruntime "github.com/goplus/llgo/runtime/internal/runtime" +) + +type wasmWaiter struct { + next *wasmWaiter + waiter llruntime.SchedulerWaiter + ticket uint32 +} + +type wasmWaitQueue struct { + head *wasmWaiter + tail *wasmWaiter +} + +func (q *wasmWaitQueue) push(w *wasmWaiter, lifo bool) { + if lifo { + w.next = q.head + q.head = w + if q.tail == nil { + q.tail = w + } + return + } + if q.tail == nil { + q.head = w + } else { + q.tail.next = w + } + q.tail = w +} + +func (q *wasmWaitQueue) pop() *wasmWaiter { + w := q.head + if w == nil { + return nil + } + q.head = w.next + if q.head == nil { + q.tail = nil + } + w.next = nil + return w +} + +func (q *wasmWaitQueue) removeTicket(ticket uint32) *wasmWaiter { + var prev *wasmWaiter + for w := q.head; w != nil; w = w.next { + if w.ticket == ticket { + if prev == nil { + q.head = w.next + } else { + prev.next = w.next + } + if q.tail == w { + q.tail = prev + } + w.next = nil + return w + } + prev = w + } + return nil +} + +var semaQueues map[uintptr]*wasmWaitQueue + +func semaQueue(addr *uint32) *wasmWaitQueue { + if semaQueues == nil { + semaQueues = make(map[uintptr]*wasmWaitQueue) + } + key := uintptr(unsafe.Pointer(addr)) + q := semaQueues[key] + if q == nil { + q = new(wasmWaitQueue) + semaQueues[key] = q + } + return q +} + +func semaAcquire(addr *uint32, lifo bool) { + value := latomic.LoadUint32(addr) + if value != 0 && latomic.CompareAndSwapUint32(addr, value, value-1) { + return + } + w := &wasmWaiter{waiter: llruntime.CurrentSchedulerWaiter()} + semaQueue(addr).push(w, lifo) + w.waiter.Park() +} + +func semaRelease(addr *uint32, handoff bool) { + key := uintptr(unsafe.Pointer(addr)) + if q := semaQueues[key]; q != nil { + if w := q.pop(); w != nil { + if q.head == nil { + delete(semaQueues, key) + } + w.waiter.Ready() + if handoff { + llruntime.Gosched() + } + return + } + } + latomic.AddUint32(addr, 1) +} + +//go:linkname sync_runtime_Semacquire sync.runtime_Semacquire +func sync_runtime_Semacquire(addr *uint32) { + semaAcquire(addr, false) +} + +//go:linkname poll_runtime_Semacquire internal/poll.runtime_Semacquire +func poll_runtime_Semacquire(addr *uint32) { + semaAcquire(addr, false) +} + +//go:linkname sync_runtime_Semrelease sync.runtime_Semrelease +func sync_runtime_Semrelease(addr *uint32, handoff bool, _ int) { + semaRelease(addr, handoff) +} + +//go:linkname sync_runtime_SemacquireRWMutexR sync.runtime_SemacquireRWMutexR +func sync_runtime_SemacquireRWMutexR(addr *uint32, lifo bool, _ int) { + semaAcquire(addr, lifo) +} + +//go:linkname sync_runtime_SemacquireRWMutex sync.runtime_SemacquireRWMutex +func sync_runtime_SemacquireRWMutex(addr *uint32, lifo bool, _ int) { + semaAcquire(addr, lifo) +} + +func syncWaitGroupAcquire(addr *uint32) { + semaAcquire(addr, false) +} + +func runtime_SemacquireMutex(addr *uint32, lifo bool, _ int) { + semaAcquire(addr, lifo) +} + +//go:linkname sync_runtime_SemacquireMutex sync.runtime_SemacquireMutex +func sync_runtime_SemacquireMutex(addr *uint32, lifo bool, skipframes int) { + runtime_SemacquireMutex(addr, lifo, skipframes) +} + +func runtime_Semrelease(addr *uint32, handoff bool, _ int) { + semaRelease(addr, handoff) +} + +//go:linkname poll_runtime_Semrelease internal/poll.runtime_Semrelease +func poll_runtime_Semrelease(addr *uint32) { + semaRelease(addr, false) +} + +func runtime_canSpin(int) bool { return false } +func runtime_doSpin() {} +func runtime_nanotime() int64 { return runtimeNano() } + +//go:linkname sync_runtime_canSpin sync.runtime_canSpin +func sync_runtime_canSpin(i int) bool { return runtime_canSpin(i) } + +//go:linkname sync_runtime_doSpin sync.runtime_doSpin +func sync_runtime_doSpin() { runtime_doSpin() } + +//go:linkname sync_runtime_nanotime sync.runtime_nanotime +func sync_runtime_nanotime() int64 { return runtime_nanotime() } + +//go:linkname internal_sync_runtime_canSpin internal/sync.runtime_canSpin +func internal_sync_runtime_canSpin(i int) bool { return runtime_canSpin(i) } + +//go:linkname internal_sync_runtime_doSpin internal/sync.runtime_doSpin +func internal_sync_runtime_doSpin() { runtime_doSpin() } + +//go:linkname internal_sync_runtime_nanotime internal/sync.runtime_nanotime +func internal_sync_runtime_nanotime() int64 { return runtime_nanotime() } + +//go:linkname internal_sync_runtime_SemacquireMutex internal/sync.runtime_SemacquireMutex +func internal_sync_runtime_SemacquireMutex(addr *uint32, lifo bool, skipframes int) { + runtime_SemacquireMutex(addr, lifo, skipframes) +} + +//go:linkname internal_sync_runtime_Semrelease internal/sync.runtime_Semrelease +func internal_sync_runtime_Semrelease(addr *uint32, handoff bool, skipframes int) { + runtime_Semrelease(addr, handoff, skipframes) +} + +//go:linkname internal_sync_throw internal/sync.throw +func internal_sync_throw(s string) { throw(s) } + +//go:linkname internal_sync_fatal internal/sync.fatal +func internal_sync_fatal(s string) { fatal(s) } + +type notifyList struct { + wait uint32 + notify uint32 + lock uintptr + head unsafe.Pointer + tail unsafe.Pointer +} + +var notifyQueues map[uintptr]*wasmWaitQueue + +func notifyQueue(l *notifyList) *wasmWaitQueue { + if notifyQueues == nil { + notifyQueues = make(map[uintptr]*wasmWaitQueue) + } + key := uintptr(unsafe.Pointer(l)) + q := notifyQueues[key] + if q == nil { + q = new(wasmWaitQueue) + notifyQueues[key] = q + } + return q +} + +func ticketLess(a, b uint32) bool { + return int32(a-b) < 0 +} + +//go:linkname sync_runtime_notifyListAdd sync.runtime_notifyListAdd +func sync_runtime_notifyListAdd(l *notifyList) uint32 { + return latomic.AddUint32(&l.wait, 1) - 1 +} + +//go:linkname sync_runtime_notifyListWait sync.runtime_notifyListWait +func sync_runtime_notifyListWait(l *notifyList, ticket uint32) { + if ticketLess(ticket, latomic.LoadUint32(&l.notify)) { + return + } + w := &wasmWaiter{ + waiter: llruntime.CurrentSchedulerWaiter(), + ticket: ticket, + } + notifyQueue(l).push(w, false) + w.waiter.Park() +} + +//go:linkname sync_runtime_notifyListNotifyAll sync.runtime_notifyListNotifyAll +func sync_runtime_notifyListNotifyAll(l *notifyList) { + wait := latomic.LoadUint32(&l.wait) + if latomic.LoadUint32(&l.notify) == wait { + return + } + latomic.StoreUint32(&l.notify, wait) + key := uintptr(unsafe.Pointer(l)) + q := notifyQueues[key] + if q == nil { + return + } + delete(notifyQueues, key) + for { + w := q.pop() + if w == nil { + return + } + w.waiter.Ready() + } +} + +//go:linkname sync_runtime_notifyListNotifyOne sync.runtime_notifyListNotifyOne +func sync_runtime_notifyListNotifyOne(l *notifyList) { + notify := latomic.LoadUint32(&l.notify) + if notify == latomic.LoadUint32(&l.wait) { + return + } + latomic.StoreUint32(&l.notify, notify+1) + key := uintptr(unsafe.Pointer(l)) + q := notifyQueues[key] + if q == nil { + return + } + if w := q.removeTicket(notify); w != nil { + if q.head == nil { + delete(notifyQueues, key) + } + w.waiter.Ready() + } +} + +//go:linkname sync_runtime_notifyListCheck sync.runtime_notifyListCheck +func sync_runtime_notifyListCheck(size uintptr) { + if size != unsafe.Sizeof(notifyList{}) { + panic("sync.notifyList size mismatch") + } +} + +var poolCleanup func() + +//go:linkname sync_runtime_registerPoolCleanup sync.runtime_registerPoolCleanup +func sync_runtime_registerPoolCleanup(cleanup func()) { + poolCleanup = cleanup +} + +//go:linkname sync_runtime_procPin sync.runtime_procPin +func sync_runtime_procPin() int { + return 0 +} + +//go:linkname sync_runtime_procUnpin sync.runtime_procUnpin +func sync_runtime_procUnpin() {} + +//go:linkname atomic_runtime_procPin sync/atomic.runtime_procPin +func atomic_runtime_procPin() int { + return 0 +} + +//go:linkname atomic_runtime_procUnpin sync/atomic.runtime_procUnpin +func atomic_runtime_procUnpin() {} diff --git a/runtime/internal/lib/runtime/sync_runtime_llgo.go b/runtime/internal/lib/runtime/sync_runtime_llgo.go index 800ccab4bb..2b4e245912 100644 --- a/runtime/internal/lib/runtime/sync_runtime_llgo.go +++ b/runtime/internal/lib/runtime/sync_runtime_llgo.go @@ -1,4 +1,4 @@ -//go:build darwin || linux +//go:build darwin || linux || (llgo && wasip1 && wasm && llgo.wasi_threads) package runtime diff --git a/runtime/internal/lib/runtime/synctest_llgo.go b/runtime/internal/lib/runtime/synctest_llgo.go index df08c497f4..277686154a 100644 --- a/runtime/internal/lib/runtime/synctest_llgo.go +++ b/runtime/internal/lib/runtime/synctest_llgo.go @@ -1,4 +1,4 @@ -//go:build darwin || linux +//go:build darwin || linux || (llgo && wasm) package runtime diff --git a/runtime/internal/runtime/chan_sync_pthread.go b/runtime/internal/runtime/chan_sync_pthread.go new file mode 100644 index 0000000000..c37740d341 --- /dev/null +++ b/runtime/internal/runtime/chan_sync_pthread.go @@ -0,0 +1,61 @@ +//go:build !llgo || !wasm || (wasip1 && llgo.wasi_threads) + +package runtime + +import "github.com/goplus/llgo/runtime/internal/clite/pthread/sync" + +type chanMutex struct { + mutex sync.Mutex +} + +func (m *chanMutex) init() { + m.mutex.Init(nil) +} + +func (m *chanMutex) Lock() { + m.mutex.Lock() +} + +func (m *chanMutex) Unlock() { + m.mutex.Unlock() +} + +type chanSignal struct { + mutex sync.Mutex + cond sync.Cond +} + +func (s *chanSignal) init() { + s.mutex.Init(nil) + s.cond.Init(nil) +} + +func (s *chanSignal) lock() { + s.mutex.Lock() +} + +func (s *chanSignal) unlock() { + s.mutex.Unlock() +} + +func (s *chanSignal) park() { + s.cond.Wait(&s.mutex) +} + +func (s *chanSignal) ready() { + s.cond.Signal() +} + +func (s *chanSignal) destroy() { + s.cond.Destroy() + s.mutex.Destroy() +} + +func chanBlockForever() { + var signal chanSignal + signal.init() + signal.lock() + for { + signal.park() + } +} diff --git a/runtime/internal/runtime/chan_sync_wasm.go b/runtime/internal/runtime/chan_sync_wasm.go new file mode 100644 index 0000000000..d3877770b3 --- /dev/null +++ b/runtime/internal/runtime/chan_sync_wasm.go @@ -0,0 +1,35 @@ +//go:build llgo && wasm && !(wasip1 && llgo.wasi_threads) + +package runtime + +type chanMutex struct{} + +func (*chanMutex) init() {} +func (*chanMutex) Lock() {} +func (*chanMutex) Unlock() {} + +type chanSignal struct { + waiter SchedulerWaiter +} + +func (s *chanSignal) init() { + s.waiter = CurrentSchedulerWaiter() +} + +func (*chanSignal) lock() {} +func (*chanSignal) unlock() {} + +func (s *chanSignal) park() { + s.waiter.Park() +} + +func (s *chanSignal) ready() { + s.waiter.Ready() +} + +func (*chanSignal) destroy() {} + +func chanBlockForever() { + CurrentSchedulerWaiter().Park() + fatal("runtime: permanently parked goroutine was resumed") +} diff --git a/runtime/internal/runtime/scheduler_waiter_wasm.go b/runtime/internal/runtime/scheduler_waiter_wasm.go new file mode 100644 index 0000000000..81f22d2434 --- /dev/null +++ b/runtime/internal/runtime/scheduler_waiter_wasm.go @@ -0,0 +1,32 @@ +//go:build llgo && wasm && !(wasip1 && llgo.wasi_threads) + +package runtime + +// SchedulerWaiter is an opaque handle used by runtime primitives that park +// the current G without exposing scheduler-owned state. +type SchedulerWaiter struct { + gp *g +} + +// CurrentSchedulerWaiter returns a handle for the current G. +func CurrentSchedulerWaiter() SchedulerWaiter { + return SchedulerWaiter{gp: getg()} +} + +// Park suspends the waiter until Ready makes it runnable. +func (w SchedulerWaiter) Park() { + if w.gp == nil || getg() != w.gp { + fatal("runtime: invalid WebAssembly scheduler waiter") + return + } + gopark() +} + +// Ready makes a previously parked waiter runnable. +func (w SchedulerWaiter) Ready() { + if w.gp == nil { + fatal("runtime: ready of invalid WebAssembly scheduler waiter") + return + } + goready(w.gp) +} diff --git a/runtime/internal/runtime/z_chan.go b/runtime/internal/runtime/z_chan.go index eeac62eebf..45b46f840e 100644 --- a/runtime/internal/runtime/z_chan.go +++ b/runtime/internal/runtime/z_chan.go @@ -20,14 +20,13 @@ import ( "unsafe" c "github.com/goplus/llgo/runtime/internal/clite" - "github.com/goplus/llgo/runtime/internal/clite/pthread/sync" "github.com/goplus/llgo/runtime/internal/runtime/math" ) // ----------------------------------------------------------------------------- type Chan struct { - mutex sync.Mutex + mutex chanMutex qcount int dataqsiz int @@ -59,16 +58,14 @@ type chanWaiter struct { queued bool status waitStatus - mutex sync.Mutex - cond sync.Cond + signal chanSignal sel *selectState caseIndex int } type selectState struct { - mutex sync.Mutex - cond sync.Cond + signal chanSignal status waitStatus chosen int @@ -150,7 +147,7 @@ func NewChan(eltSize, cap int) *Chan { if cap > 0 { ret.buf = AllocU(mem) } - ret.mutex.Init(nil) + ret.mutex.init() return ret } @@ -201,9 +198,8 @@ func newChanWaiter(ch *Chan, elem unsafe.Pointer, eltSize int, send bool) *chanW w.elem = elem w.size = eltSize w.send = send - w.mutex.Init(nil) - w.cond.Init(nil) - w.mutex.Lock() + w.signal.init() + w.signal.lock() return w } @@ -214,12 +210,12 @@ func newSelectState() *selectState { } c.Memset(unsafe.Pointer(state), 0, unsafe.Sizeof(selectState{})) state.chosen = -1 - state.mutex.Init(nil) - state.cond.Init(nil) + state.signal.init() return state } func freeSelectState(state *selectState) { + state.signal.destroy() c.Free(unsafe.Pointer(state)) } @@ -248,37 +244,36 @@ func freeSelectWaiters(w *chanWaiter) { func (w *chanWaiter) wait() { for !w.status.done() { - w.cond.Wait(&w.mutex) + w.signal.park() } - w.mutex.Unlock() - w.cond.Destroy() - w.mutex.Destroy() + w.signal.unlock() + w.signal.destroy() } func (w *chanWaiter) finish(status waitStatus) { if w.sel != nil { - w.sel.mutex.Lock() + w.sel.signal.lock() w.sel.status = status - w.sel.mutex.Unlock() - w.sel.cond.Signal() + w.sel.signal.unlock() + w.sel.signal.ready() return } - w.mutex.Lock() + w.signal.lock() w.status = status - w.mutex.Unlock() - w.cond.Signal() + w.signal.unlock() + w.signal.ready() } func claimWaiter(w *chanWaiter) bool { if w.sel != nil { - w.sel.mutex.Lock() + w.sel.signal.lock() if w.sel.status != waitPending { - w.sel.mutex.Unlock() + w.sel.signal.unlock() return false } w.sel.status = waitClaimed w.sel.chosen = w.caseIndex - w.sel.mutex.Unlock() + w.sel.signal.unlock() return true } return true @@ -497,14 +492,7 @@ func ChanClose(p *Chan) { } func blockForever() { - var mutex sync.Mutex - var cond sync.Cond - mutex.Init(nil) - cond.Init(nil) - mutex.Lock() - for { - cond.Wait(&mutex) - } + chanBlockForever() } // ----------------------------------------------------------------------------- @@ -688,20 +676,18 @@ func Select(ops ...ChanOp) (isel int, recvOK bool) { } unlockSelectChannels(&chans) - state.mutex.Lock() + state.signal.lock() for !state.status.done() { - state.cond.Wait(&state.mutex) + state.signal.park() } isel = state.chosen status := state.status recvOK = status.recvOK() - state.mutex.Unlock() + state.signal.unlock() for w := waiters; w != nil; w = w.all { cleanupSelectWaiter(w) } - state.cond.Destroy() - state.mutex.Destroy() freeSelectState(state) freeSelectWaiters(waiters) if status.panicOnWake() {