WIP map every goroutine to a new OS thread

Author: aykevl

compileopts/config.go

@@ -340,6 +340,7 @@ func (c *Config) CFlags(libclang bool) []string {
 			"-nostdlibinc",
 			"-isystem", filepath.Join(path, "include"),
 			"-isystem", filepath.Join(root, "lib", "musl", "arch", arch),
+			"-isystem", filepath.Join(root, "lib", "musl", "arch", "generic"),
 			"-isystem", filepath.Join(root, "lib", "musl", "include"),
 		)
 	case "wasi-libc":

compileopts/options.go

@@ -10,7 +10,7 @@ import (
 var (
 	validBuildModeOptions     = []string{"default", "c-shared"}
 	validGCOptions            = []string{"none", "leaking", "conservative", "custom", "precise"}
-	validSchedulerOptions     = []string{"none", "tasks", "asyncify"}
+	validSchedulerOptions     = []string{"none", "tasks", "asyncify", "threads"}
 	validSerialOptions        = []string{"none", "uart", "usb", "rtt"}
 	validPrintSizeOptions     = []string{"none", "short", "full"}
 	validPanicStrategyOptions = []string{"print", "trap"}

compileopts/options_test.go

@@ -10,7 +10,7 @@ import (
 func TestVerifyOptions(t *testing.T) {
 
 	expectedGCError := errors.New(`invalid gc option 'incorrect': valid values are none, leaking, conservative, custom, precise`)
-	expectedSchedulerError := errors.New(`invalid scheduler option 'incorrect': valid values are none, tasks, asyncify`)
+	expectedSchedulerError := errors.New(`invalid scheduler option 'incorrect': valid values are none, tasks, asyncify, threads`)
 	expectedPrintSizeError := errors.New(`invalid size option 'incorrect': valid values are none, short, full`)
 	expectedPanicStrategyError := errors.New(`invalid panic option 'incorrect': valid values are print, trap`)
 

compileopts/target.go

@@ -247,7 +247,6 @@ func defaultTarget(options *Options) (*TargetSpec, error) {
 		GOARCH:           options.GOARCH,
 		BuildTags:        []string{options.GOOS, options.GOARCH},
 		GC:               "precise",
-		Scheduler:        "tasks",
 		Linker:           "cc",
 		DefaultStackSize: 1024 * 64, // 64kB
 		GDB:              []string{"gdb"},
@@ -378,6 +377,7 @@ func defaultTarget(options *Options) (*TargetSpec, error) {
 			platformVersion = "11.0.0" // first macosx platform with arm64 support
 		}
 		llvmvendor = "apple"
+		spec.Scheduler = "tasks"
 		spec.Linker = "ld.lld"
 		spec.Libc = "darwin-libSystem"
 		// Use macosx* instead of darwin, otherwise darwin/arm64 will refer to
@@ -394,6 +394,7 @@ func defaultTarget(options *Options) (*TargetSpec, error) {
 			"src/runtime/runtime_unix.c",
 			"src/runtime/signal.c")
 	case "linux":
+		spec.Scheduler = "threads"
 		spec.Linker = "ld.lld"
 		spec.RTLib = "compiler-rt"
 		spec.Libc = "musl"
@@ -414,9 +415,11 @@ func defaultTarget(options *Options) (*TargetSpec, error) {
 		}
 		spec.ExtraFiles = append(spec.ExtraFiles,
 			"src/internal/task/futex_linux.c",
+			"src/internal/task/task_threads.c",
 			"src/runtime/runtime_unix.c",
 			"src/runtime/signal.c")
 	case "windows":
+		spec.Scheduler = "tasks"
 		spec.Linker = "ld.lld"
 		spec.Libc = "mingw-w64"
 		// Note: using a medium code model, low image base and no ASLR

src/internal/task/futex_linux.c

@@ -5,6 +5,7 @@
 
 #include <stdint.h>
 #include <sys/syscall.h>
+#include <time.h>
 #include <unistd.h>
 
 #define FUTEX_WAIT    0
@@ -15,6 +16,13 @@ void tinygo_futex_wait(uint32_t *addr, uint32_t cmp) {
     syscall(SYS_futex, addr, FUTEX_WAIT|FUTEX_PRIVATE, cmp, NULL, NULL, 0);
 }
 
+void tinygo_futex_wait_timeout(uint32_t *addr, uint32_t cmp, uint64_t timeout) {
+    struct timespec ts = {0};
+    ts.tv_sec = timeout / 1000000000;
+    ts.tv_nsec = timeout % 1000000000;
+    syscall(SYS_futex, addr, FUTEX_WAIT|FUTEX_PRIVATE, cmp, &ts, NULL, 0);
+}
+
 void tinygo_futex_wake(uint32_t *addr, uint32_t num) {
     syscall(SYS_futex, addr, FUTEX_WAKE|FUTEX_PRIVATE, num, NULL, NULL, 0);
 }

src/internal/task/futex_linux.go

@@ -31,6 +31,11 @@ func (f *Futex) Wait(cmp uint32) bool {
 	return false
 }
 
+// Like Wait, but times out after the number of nanoseconds in timeout.
+func (f *Futex) WaitUntil(cmp uint32, timeout uint64) {
+	tinygo_futex_wait_timeout((*uint32)(unsafe.Pointer(&f.Uint32)), cmp, timeout)
+}
+
 // Wake a single waiter.
 func (f *Futex) Wake() {
 	tinygo_futex_wake((*uint32)(unsafe.Pointer(&f.Uint32)), 1)
@@ -45,5 +50,8 @@ func (f *Futex) WakeAll() {
 //export tinygo_futex_wait
 func tinygo_futex_wait(addr *uint32, cmp uint32)
 
+//export tinygo_futex_wait_timeout
+func tinygo_futex_wait_timeout(addr *uint32, cmp uint32, timeout uint64)
+
 //export tinygo_futex_wake
 func tinygo_futex_wake(addr *uint32, num uint32)

src/internal/task/linux.go

@@ -0,0 +1,9 @@
+//go:build linux && !baremetal
+
+package task
+
+import "unsafe"
+
+// Musl uses a pointer (or unsigned long for C++) so unsafe.Pointer should be
+// fine.
+type threadID unsafe.Pointer

src/internal/task/semaphore.go

@@ -0,0 +1,32 @@
+package task
+
+// Barebones semaphore implementation.
+// The main limitation is that if there are multiple waiters, a single Post()
+// call won't do anything. Only when Post() has been called to awaken all
+// waiters will the waiters proceed.
+// This limitation is not a problem when there will only be a single waiter.
+type Semaphore struct {
+	futex Futex
+}
+
+// Post (unlock) the semaphore, incrementing the value in the semaphore.
+func (s *Semaphore) Post() {
+	newValue := s.futex.Add(1)
+	if newValue == 0 {
+		s.futex.WakeAll()
+	}
+}
+
+// Wait (lock) the semaphore, decrementing the value in the semaphore.
+func (s *Semaphore) Wait() {
+	delta := int32(-1)
+	value := s.futex.Add(uint32(delta))
+	for {
+		if int32(value) >= 0 {
+			// Semaphore unlocked!
+			return
+		}
+		s.futex.Wait(value)
+		value = s.futex.Load()
+	}
+}

src/internal/task/task.go

@@ -26,6 +26,23 @@ type Task struct {
 	DeferFrame unsafe.Pointer
 }
 
+// DataUint32 returns the Data field as a uint32. The value is only valid after
+// setting it through SetDataUint32 or by storing to it using DataAtomicUint32.
+func (t *Task) DataUint32() uint32 {
+	return *(*uint32)(unsafe.Pointer(&t.Data))
+}
+
+// SetDataUint32 updates the uint32 portion of the Data field (which could be
+// the first 4 or last 4 bytes depending on the architecture endianness).
+func (t *Task) SetDataUint32(value uint32) {
+	*(*uint32)(unsafe.Pointer(&t.Data)) = value
+}
+
+// DataAtomicUint32 returns the Data field as an atomic-if-needed Uint32 value.
+func (t *Task) DataAtomicUint32() *Uint32 {
+	return (*Uint32)(unsafe.Pointer(&t.Data))
+}
+
 // getGoroutineStackSize is a compiler intrinsic that returns the stack size for
 // the given function and falls back to the default stack size. It is replaced
 // with a load from a special section just before codegen.

src/internal/task/task_threads.c

@@ -0,0 +1,109 @@
+//go:build none
+
+#define _GNU_SOURCE
+#include <pthread.h>
+#include <semaphore.h>
+#include <signal.h>
+#include <stdint.h>
+#include <stdio.h>
+
+// BDWGC also uses SIGRTMIN+6 on Linux, which seems like a reasonable choice.
+#ifdef __linux__
+#define taskPauseSignal (SIGRTMIN + 6)
+#endif
+
+// Pointer to the current task.Task structure.
+// Ideally the entire task.Task structure would be a thread-local variable but
+// this also works.
+static __thread void *current_task;
+
+struct state_pass {
+    void *(*start)(void*);
+    void *args;
+    void *task;
+    sem_t startlock;
+};
+
+// Handle the GC pause in Go.
+void tinygo_task_gc_pause(int sig);
+
+// Initialize the main thread.
+void tinygo_task_init(void *mainTask, pthread_t *thread, void *context) {
+    // Make sure the current task pointer is set correctly for the main
+    // goroutine as well.
+    current_task = mainTask;
+
+    // Store the thread ID of the main thread.
+    *thread = pthread_self();
+
+    // Register the "GC pause" signal for the entire process.
+    // Using pthread_kill, we can still send the signal to a specific thread.
+    struct sigaction act = { 0 };
+    act.sa_flags = SA_SIGINFO;
+    act.sa_handler = &tinygo_task_gc_pause;
+    sigaction(taskPauseSignal, &act, NULL);
+}
+
+void tinygo_task_exited(void*);
+
+// Helper to start a goroutine while also storing the 'task' structure.
+static void* start_wrapper(void *arg) {
+    struct state_pass *state = arg;
+    void *(*start)(void*) = state->start;
+    void *args = state->args;
+    current_task = state->task;
+
+    // Notify the caller that the thread has successfully started and
+    // initialized.
+    sem_post(&state->startlock);
+
+    // Run the goroutine function.
+    start(args);
+
+    // Notify the Go side this thread will exit.
+    tinygo_task_exited(current_task);
+
+    return NULL;
+};
+
+// Start a new goroutine in an OS thread.
+int tinygo_task_start(uintptr_t fn, void *args, void *task, pthread_t *thread, void *context) {
+    // Sanity check. Should get optimized away.
+    if (sizeof(pthread_t) != sizeof(void*)) {
+        __builtin_trap();
+    }
+
+    struct state_pass state = {
+        .start     = (void*)fn,
+        .args      = args,
+        .task      = task,
+    };
+    sem_init(&state.startlock, 0, 0);
+    int result = pthread_create(thread, NULL, &start_wrapper, &state);
+
+    // Wait until the thread has been crated and read all state_pass variables.
+    sem_wait(&state.startlock);
+
+    return result;
+}
+
+// Return the current task (for task.Current()).
+void* tinygo_task_current(void) {
+    return current_task;
+}
+
+// Obtain the highest address of the stack.
+uintptr_t tinygo_task_stacktop(void) {
+    pthread_attr_t attr;
+    pthread_getattr_np(pthread_self(), &attr);
+    void *stackbase;
+    size_t stacksize;
+    pthread_attr_getstack(&attr, &stackbase, &stacksize);
+    pthread_attr_destroy(&attr);
+    return (uintptr_t)stackbase + (uintptr_t)stacksize;
+}
+
+// Send a signal to cause the task to pause for the GC mark phase.
+void tinygo_task_send_gc_signal(pthread_t thread) {
+    pthread_kill(thread, taskPauseSignal);
+}