runtime/wasm: add bounded Web worker scheduling (based on #116)

.github/workflows/llgo.yml

@@ -484,6 +484,12 @@ jobs:
             test "$output" = "wasm timers ok"
           }
 
+          run_wasm_workers() {
+            local output
+            output=$(node -e "import('$1').then(module => module.default()).catch(error => { console.error(error); process.exit(1); });" 2>&1)
+            grep -Fxq "wasm workers ok" <<<"$output"
+          }
+
           run_wasi_timers() {
             local output
             wasm-tools validate --features all "$1"
@@ -508,6 +514,34 @@ jobs:
           run_wasm_timers "$RUNNER_TEMP/wasm-timers.mjs"
           GOOS=wasip1 GOARCH=wasm llgo build -o "$RUNNER_TEMP/wasm-timers-wasip1.wasm" ./internal/build/testdata/wasm-timers
           run_wasi_timers "$RUNNER_TEMP/wasm-timers-wasip1.wasm"
+          LLGO_WASM_WORKERS=2 GOOS=js GOARCH=wasm llgo build \
+            -o "$RUNNER_TEMP/wasm-workers-go.mjs" ./internal/build/testdata/wasm-workers
+          run_wasm_workers "$RUNNER_TEMP/wasm-workers-go.mjs"
+          LLGO_WASM_WORKERS=2 llgo build -target wasm \
+            -o "$RUNNER_TEMP/wasm-workers.mjs" ./internal/build/testdata/wasm-workers
+          run_wasm_workers "$RUNNER_TEMP/wasm-workers.mjs"
+          cp ./internal/build/testdata/wasm-workers/browser.html "$RUNNER_TEMP/browser.html"
+          node ./internal/build/testdata/wasm-workers/server.mjs "$RUNNER_TEMP" 8123 &
+          browser_server=$!
+          trap 'kill "$browser_server" 2>/dev/null || true' EXIT
+          for attempt in {1..50}; do
+            if curl -fsS http://127.0.0.1:8123/ >/dev/null; then
+              break
+            fi
+            sleep 0.1
+          done
+          curl -fsS http://127.0.0.1:8123/ >/dev/null
+          browser="$(command -v google-chrome || command -v google-chrome-stable || command -v chromium || true)"
+          test -n "$browser"
+          for module in wasm-timers.mjs wasm-workers.mjs wasm-workers-go.mjs; do
+            html=$("$browser" --headless=new --no-sandbox --disable-gpu \
+              --disable-dev-shm-usage --virtual-time-budget=15000 --dump-dom \
+              "http://127.0.0.1:8123/browser.html?module=$module")
+            grep -Fq 'data-result="pass"' <<<"$html"
+          done
+          kill "$browser_server"
+          wait "$browser_server" || true
+          trap - EXIT
           GOOS=js GOARCH=wasm llgo build -o "$RUNNER_TEMP/wasm-gc-go.mjs" ./internal/build/testdata/wasm-gc
           node --input-type=module -e "import Module from '$RUNNER_TEMP/wasm-gc-go.mjs'; await Module();"
           llgo build -target wasm -o "$RUNNER_TEMP/wasm-gc.mjs" ./internal/build/testdata/wasm-gc

benchmark/wasm_workers/main.go

@@ -0,0 +1,86 @@
+package main
+
+import (
+	"sync"
+	"time"
+)
+
+const (
+	lifecycleIterations = 10_000
+	channelIterations   = 100_000
+	cpuIterations       = 20_000_000
+)
+
+var cpuResult uint64
+
+func main() {
+	report("WasmGoroutineLifecycle", lifecycleIterations, benchmarkLifecycle())
+	report("WasmChannelRoundTrip", channelIterations, benchmarkChannel())
+	cpuResult = cpuWork(0)
+	report("WasmOneCPUJob", 1, benchmarkCPUJobs(1))
+	report("WasmTwoCPUJobs", 2, benchmarkCPUJobs(2))
+	if cpuResult == 0 {
+		panic("unexpected CPU benchmark result")
+	}
+}
+
+func benchmarkLifecycle() time.Duration {
+	done := make(chan struct{}, 1)
+	start := time.Now()
+	for range lifecycleIterations {
+		go func() {
+			done <- struct{}{}
+		}()
+		<-done
+	}
+	return time.Since(start)
+}
+
+func benchmarkChannel() time.Duration {
+	request := make(chan struct{})
+	response := make(chan struct{})
+	go func() {
+		for range channelIterations {
+			<-request
+			response <- struct{}{}
+		}
+	}()
+
+	start := time.Now()
+	for range channelIterations {
+		request <- struct{}{}
+		<-response
+	}
+	return time.Since(start)
+}
+
+func benchmarkCPUJobs(jobs int) time.Duration {
+	var wg sync.WaitGroup
+	results := make([]uint64, jobs)
+	wg.Add(jobs)
+	start := time.Now()
+	for i := range results {
+		go func() {
+			results[i] = cpuWork(uint64(i + 1))
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+	cpuResult = 0
+	for _, result := range results {
+		cpuResult ^= result
+	}
+	return time.Since(start)
+}
+
+//go:noinline
+func cpuWork(value uint64) uint64 {
+	for i := uint64(0); i < cpuIterations; i++ {
+		value = value*1664525 + 1013904223 + i
+	}
+	return value
+}
+
+func report(name string, iterations int, elapsed time.Duration) {
+	println("Benchmark"+name+"-1", iterations, elapsed.Nanoseconds()/int64(iterations), "ns/op")
+}

internal/build/build.go

@@ -60,6 +60,7 @@ import (
 	"github.com/goplus/llgo/internal/pclnmap"
 	"github.com/goplus/llgo/internal/pclnpost"
 	"github.com/goplus/llgo/internal/typepatch"
+	"github.com/goplus/llgo/internal/wasmworkers"
 	"github.com/goplus/llgo/ssa/abi"
 	xenv "github.com/goplus/llgo/xtool/env"
 	gllvm "github.com/xgo-dev/llvm"
@@ -386,7 +387,11 @@ func Build(inv Invocation) ([]Package, error) {
 	if conf.Target != "" && export.GOARCH != "" {
 		conf.Goarch = export.GOARCH
 	}
-	wasmGC, err := configureWasmGC(conf, &export)
+	wasmWorkers, err := configureWasmWorkers(conf, &export)
+	if err != nil {
+		return nil, err
+	}
+	wasmGC, err := configureWasmGC(conf, &export, wasmWorkers.Enabled())
 	if err != nil {
 		return nil, err
 	}
@@ -459,7 +464,7 @@ func Build(inv Invocation) ([]Package, error) {
 	prog.EnableGoGlobalDCE(conf.goGlobalDCEEnabled())
 	prog.EnableDeadcodeDrop(conf.deadcodeDropEnabled())
 	prog.EnableGCRoots(wasmGC)
-	prog.EnableCooperativeSafepoints(wasmGC)
+	prog.EnableCooperativeSafepoints(wasmGC || wasmWorkers.Enabled())
 	if conf.PthreadStackSize > 0 {
 		prog.SetPthreadStackSize(uint64(conf.PthreadStackSize))
 	}
@@ -785,7 +790,37 @@ func defaultBuildTags(goarch, target string) string {
 	return tags
 }
 
-func configureWasmGC(conf *Config, export *crosscompile.Export) (bool, error) {
+func configureWasmWorkers(conf *Config, export *crosscompile.Export) (wasmworkers.Config, error) {
+	config, err := wasmworkers.Parse(os.Getenv(llgoWasmWorkers))
+	if err != nil {
+		return config, err
+	}
+	if err := config.ValidateTarget(conf.Goos, conf.Goarch); err != nil {
+		return config, err
+	}
+	if !config.Enabled() {
+		return config, nil
+	}
+	workers := strconv.Itoa(config.Count)
+	poolSize := strconv.Itoa(config.Count)
+	preJS := wasmworkers.PreJSPath(env.LLGoROOT())
+	if _, err := os.Stat(preJS); err != nil {
+		return config, fmt.Errorf("locate WebAssembly worker host shim: %w", err)
+	}
+	export.BuildTags = append(export.BuildTags, "llgo.wasm_workers")
+	export.CCFLAGS = append(export.CCFLAGS, "-pthread", "-DLLGO_WASM_WORKERS="+workers)
+	export.LDFLAGS = append(export.LDFLAGS,
+		"--pre-js", preJS,
+		"-pthread",
+		"-sPTHREAD_POOL_SIZE="+poolSize,
+		"-sPROXY_TO_PTHREAD=1",
+		"-sEXIT_RUNTIME=1",
+	)
+	export.WasmRuntime.RunMainTask = true
+	return config, nil
+}
+
+func configureWasmGC(conf *Config, export *crosscompile.Export, wasmWorkers bool) (bool, error) {
 	explicit := hasBuildTag(conf.Tags, "llgo_wasm_gc")
 	if conf.Goarch != "wasm" {
 		if explicit {
@@ -795,6 +830,12 @@ func configureWasmGC(conf *Config, export *crosscompile.Export) (bool, error) {
 	}
 	switch conf.Goos {
 	case "js":
+		if wasmWorkers {
+			if explicit {
+				return false, errors.New("llgo_wasm_gc does not yet support multiple WebAssembly workers")
+			}
+			return false, nil
+		}
 		if !slices.Contains(export.LDFLAGS, "-sMALLOC=none") {
 			export.LDFLAGS = append(export.LDFLAGS, "-sMALLOC=none")
 		}
@@ -2419,6 +2460,7 @@ const llgoTrace = "LLGO_TRACE"
 const llgoOptimize = "LLGO_OPTIMIZE"
 const llgoWasmRuntime = "LLGO_WASM_RUNTIME"
 const llgoWasiThreads = "LLGO_WASI_THREADS"
+const llgoWasmWorkers = "LLGO_WASM_WORKERS"
 const llgoStdioNobuf = "LLGO_STDIO_NOBUF"
 const llgoFullRpath = "LLGO_FULL_RPATH"
 const llgoBuildCache = "LLGO_BUILD_CACHE"

internal/build/build_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/goplus/llgo/internal/meta"
 	"github.com/goplus/llgo/internal/mockable"
 	"github.com/goplus/llgo/internal/packages"
+	"github.com/goplus/llgo/internal/wasmworkers"
 	llssa "github.com/goplus/llgo/ssa"
 	"github.com/xgo-dev/llvm"
 )
@@ -344,7 +345,7 @@ func TestConfigureWasmGC(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			export := crosscompile.Export{}
-			enabled, err := configureWasmGC(&test.conf, &export)
+			enabled, err := configureWasmGC(&test.conf, &export, false)
 			if (err != nil) != test.err {
 				t.Fatalf("configureWasmGC error = %v, want error %v", err, test.err)
 			}
@@ -364,15 +365,15 @@ func TestConfigureWasmGC(t *testing.T) {
 func TestConfigureWasmGCRejectsWASIThreads(t *testing.T) {
 	t.Setenv("LLGO_WASI_THREADS", "1")
 	conf := Config{Goos: "wasip1", Goarch: "wasm", Tags: "llgo_wasm_gc"}
-	if _, err := configureWasmGC(&conf, &crosscompile.Export{}); err == nil {
+	if _, err := configureWasmGC(&conf, &crosscompile.Export{}, false); err == nil {
 		t.Fatal("expected llgo_wasm_gc with WASI threads to fail")
 	}
 }
 
 func TestConfigureWasmGCLeavesWASIThreadsDisabled(t *testing.T) {
 	t.Setenv("LLGO_WASI_THREADS", "1")
 	conf := Config{Goos: "wasip1", Goarch: "wasm"}
-	enabled, err := configureWasmGC(&conf, &crosscompile.Export{})
+	enabled, err := configureWasmGC(&conf, &crosscompile.Export{}, false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -381,6 +382,66 @@ func TestConfigureWasmGCLeavesWASIThreadsDisabled(t *testing.T) {
 	}
 }
 
+func TestConfigureWasmWorkers(t *testing.T) {
+	t.Setenv(llgoWasmWorkers, "4")
+	conf := Config{Goos: "js", Goarch: "wasm"}
+	export := crosscompile.Export{}
+	config, err := configureWasmWorkers(&conf, &export)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if config.Count != 4 || !config.Enabled() {
+		t.Fatalf("worker config = %+v, want four enabled workers", config)
+	}
+	for _, flag := range []string{"-pthread", "-DLLGO_WASM_WORKERS=4"} {
+		if !slices.Contains(export.CCFLAGS, flag) {
+			t.Fatalf("CCFLAGS do not contain %q: %v", flag, export.CCFLAGS)
+		}
+	}
+	for _, flag := range []string{"-pthread", "-sPTHREAD_POOL_SIZE=4", "-sPROXY_TO_PTHREAD=1", "-sEXIT_RUNTIME=1"} {
+		if !slices.Contains(export.LDFLAGS, flag) {
+			t.Fatalf("LDFLAGS do not contain %q: %v", flag, export.LDFLAGS)
+		}
+	}
+	preJS := wasmworkers.PreJSPath(env.LLGoROOT())
+	if i := slices.Index(export.LDFLAGS, "--pre-js"); i < 0 || i+1 == len(export.LDFLAGS) || export.LDFLAGS[i+1] != preJS {
+		t.Fatalf("LDFLAGS do not select worker host shim %q: %v", preJS, export.LDFLAGS)
+	}
+	if !slices.Contains(export.BuildTags, "llgo.wasm_workers") {
+		t.Fatalf("BuildTags do not select the worker runtime: %v", export.BuildTags)
+	}
+	if !export.WasmRuntime.RunMainTask {
+		t.Fatal("worker runtime does not run main as a schedulable task")
+	}
+}
+
+func TestConfigureWasmWorkersDefaultIsInert(t *testing.T) {
+	conf := Config{Goos: "linux", Goarch: "amd64"}
+	export := crosscompile.Export{}
+	config, err := configureWasmWorkers(&conf, &export)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if config.Enabled() || len(export.CCFLAGS) != 0 || len(export.LDFLAGS) != 0 || len(export.BuildTags) != 0 {
+		t.Fatalf("default worker config changed native build: config=%+v export=%+v", config, export)
+	}
+}
+
+func TestConfigureWasmWorkersRejectsUnsupportedTarget(t *testing.T) {
+	t.Setenv(llgoWasmWorkers, "2")
+	conf := Config{Goos: "wasip1", Goarch: "wasm"}
+	if _, err := configureWasmWorkers(&conf, &crosscompile.Export{}); err == nil {
+		t.Fatal("WASI worker configuration succeeded")
+	}
+}
+
+func TestConfigureWasmWorkersRejectsCurrentGC(t *testing.T) {
+	conf := Config{Goos: "js", Goarch: "wasm", Tags: "llgo_wasm_gc"}
+	if _, err := configureWasmGC(&conf, &crosscompile.Export{}, true); err == nil {
+		t.Fatal("multi-worker wasm GC configuration succeeded before M2")
+	}
+}
+
 func TestWasmRuntimeAvoidsNativeHostDependencies(t *testing.T) {
 	runtimeDir := filepath.Join(env.LLGoRuntimeDir(), "internal", "lib", "runtime")
 	for _, goos := range []string{"js", "wasip1"} {

internal/build/collect.go

@@ -86,6 +86,7 @@ func (c *context) collectEnvInputs(m *manifestBuilder) {
 		llgoOptimize,
 		llgoWasmRuntime,
 		llgoWasiThreads,
+		llgoWasmWorkers,
 		llgoStdioNobuf,
 		llgoFullRpath,
 	}

internal/build/main_module.go

@@ -128,7 +128,7 @@ func genMainModule(ctx *context, rtPkgPath string, pkg *packages.Package, cfg *g
 	}
 
 	var wasmRunMain llssa.Function
-	if ctx.crossCompile.WasmPostLink.Asyncify {
+	if ctx.crossCompile.WasmPostLink.Asyncify || ctx.crossCompile.WasmRuntime.RunMainTask {
 		defineWasmMainTask(mainPkg, mainInit, mainMain)
 		wasmRunMain = declareNoArgFunc(mainPkg, rtPkgPath+".RunWasmMain")
 	}
@@ -143,7 +143,7 @@ func genMainModule(ctx *context, rtPkgPath string, pkg *packages.Package, cfg *g
 		abiInit:     abiInit,
 	})
 
-	if needStart(ctx) {
+	if needStart(ctx) && !ctx.crossCompile.WasmRuntime.RunMainTask {
 		defineStart(mainPkg, entryFn, argvValueType)
 	}
 
@@ -248,7 +248,8 @@ type entryFunctions struct {
 func defineEntryFunction(ctx *context, pkg llssa.Package, argcVar, argvVar llssa.Global, argvType llssa.Type, fns entryFunctions) llssa.Function {
 	prog := pkg.Prog
 	entryName := "main"
-	if !needStart(ctx) && isWasmTarget(ctx.buildConf.Goos) {
+	if isWasmTarget(ctx.buildConf.Goos) &&
+		(!needStart(ctx) || ctx.crossCompile.WasmRuntime.RunMainTask) {
 		entryName = "__main_argc_argv"
 	}
 	sig := newEntrySignature(argvType.RawType())

internal/build/main_module_test.go

@@ -99,6 +99,36 @@ func TestGenMainModuleWASIAsyncifyEntry(t *testing.T) {
 	}
 }
 
+func TestGenMainModuleWasmWorkerEntry(t *testing.T) {
+	llvm.InitializeAllTargets()
+	t.Setenv(llgoStdioNobuf, "")
+	ctx := &context{
+		prog: llssa.NewProgram(nil),
+		buildConf: &Config{
+			BuildMode: BuildModeExe,
+			Goos:      "js",
+			Goarch:    "wasm",
+		},
+		crossCompile: crosscompile.Export{
+			WasmRuntime: crosscompile.WasmRuntime{RunMainTask: true},
+		},
+	}
+	pkg := &packages.Package{PkgPath: "example.com/foo", ExportFile: "foo.a"}
+	ir := genMainModule(ctx, llssa.PkgRuntime, pkg, &genConfig{}).LPkg.String()
+	if !strings.Contains(ir, "define hidden i32 @__main_argc_argv(") {
+		t.Fatalf("worker module missing Emscripten host entry:\n%s", ir)
+	}
+	if !strings.Contains(ir, `call void @"github.com/goplus/llgo/runtime/internal/runtime.RunWasmMain"()`) {
+		t.Fatalf("worker host entry does not start the runtime main task:\n%s", ir)
+	}
+	if strings.Contains(ir, "define i32 @main(") {
+		t.Fatalf("worker module should let Emscripten provide main:\n%s", ir)
+	}
+	if strings.Contains(ir, "define weak void @_start()") {
+		t.Fatalf("worker module should let Emscripten provide _start:\n%s", ir)
+	}
+}
+
 func TestGenMainModuleLibrary(t *testing.T) {
 	llvm.InitializeAllTargets()
 	t.Setenv(llgoStdioNobuf, "")