Support os.Worker on Windows

bnoordhuis · bnoordhuis · commit a74a6fa1b3e5 · 2025-03-30T14:00:13.000+02:00
Add thread create/join utility functions and use those to implement
os.Worker on the operating system from Redmond.
diff --git a/cutils.c b/cutils.c
@@ -1267,6 +1267,36 @@ int js_cond_timedwait(js_cond_t *cond, js_mutex_t *mutex, uint64_t timeout) {
     return -1;
 }
 
+int js_thread_create(js_thread_t *thrd, void (*start)(void *), void *arg,
+                     int flags)
+{
+    HANDLE h, cp;
+
+    *thrd = INVALID_HANDLE_VALUE;
+    if (flags & ~JS_THREAD_CREATE_DETACHED)
+        return -1;
+    cp = GetCurrentProcess();
+    h = (HANDLE)_beginthread(start, /*stacksize*/2<<20, arg);
+    if (!h)
+        return -1;
+    // _endthread() automatically closes the handle but we want to wait on
+    // it so make a copy. Race-y for very short-lived threads. Can be solved
+    // by switching to _beginthreadex(CREATE_SUSPENDED) but means changing
+    // |start| from __cdecl to __stdcall.
+    if (!(flags & JS_THREAD_CREATE_DETACHED))
+        if (!DuplicateHandle(cp, h, cp, thrd, 0, false, DUPLICATE_SAME_ACCESS))
+            return -1;
+    return 0;
+}
+
+int js_thread_join(js_thread_t thrd)
+{
+    if (WaitForSingleObject(thrd, INFINITE))
+        return -1;
+    CloseHandle(thrd);
+    return 0;
+}
+
 #else /* !defined(_WIN32) */
 
 void js_once(js_once_t *guard, void (*callback)(void)) {
@@ -1409,6 +1439,41 @@ int js_cond_timedwait(js_cond_t *cond, js_mutex_t *mutex, uint64_t timeout) {
 
 #endif
 
+int js_thread_create(js_thread_t *thrd, void (*start)(void *), void *arg,
+                     int flags)
+{
+    union {
+        void (*x)(void *);
+        void *(*f)(void *);
+    } u = {start};
+    pthread_attr_t attr;
+    int ret;
+
+    if (flags & ~JS_THREAD_CREATE_DETACHED)
+        return -1;
+    if (pthread_attr_init(&attr))
+        return -1;
+    ret = -1;
+    if (pthread_attr_setstacksize(&attr, 2<<20))
+        goto fail;
+    if (flags & JS_THREAD_CREATE_DETACHED)
+        if (pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED))
+            goto fail;
+    if (pthread_create(thrd, &attr, u.f, arg))
+        goto fail;
+    ret = 0;
+fail:
+    pthread_attr_destroy(&attr);
+    return ret;
+}
+
+int js_thread_join(js_thread_t thrd)
+{
+    if (pthread_join(thrd, NULL))
+        return -1;
+    return 0;
+}
+
 #endif /* !defined(EMSCRIPTEN) && !defined(__wasi__) */
 
 #ifdef __GNUC__
diff --git a/cutils.h b/cutils.h
@@ -570,11 +570,13 @@ static inline size_t js__malloc_usable_size(const void *ptr)
 typedef INIT_ONCE js_once_t;
 typedef CRITICAL_SECTION js_mutex_t;
 typedef CONDITION_VARIABLE js_cond_t;
+typedef HANDLE js_thread_t;
 #else
 #define JS_ONCE_INIT PTHREAD_ONCE_INIT
 typedef pthread_once_t js_once_t;
 typedef pthread_mutex_t js_mutex_t;
 typedef pthread_cond_t js_cond_t;
+typedef pthread_t js_thread_t;
 #endif
 
 void js_once(js_once_t *guard, void (*callback)(void));
@@ -591,6 +593,15 @@ void js_cond_broadcast(js_cond_t *cond);
 void js_cond_wait(js_cond_t *cond, js_mutex_t *mutex);
 int js_cond_timedwait(js_cond_t *cond, js_mutex_t *mutex, uint64_t timeout);
 
+enum {
+    JS_THREAD_CREATE_DETACHED = 1,
+};
+
+// creates threads with 2 MB stacks (glibc default)
+int js_thread_create(js_thread_t *thrd, void (*start)(void *), void *arg,
+                     int flags);
+int js_thread_join(js_thread_t thrd);
+
 #endif /* !defined(EMSCRIPTEN) && !defined(__wasi__) */
 
 #ifdef __cplusplus
diff --git a/quickjs-libc.c b/quickjs-libc.c
@@ -80,13 +80,12 @@ extern char **environ;
 
 #endif /* _WIN32 */
 
-#if !defined(_WIN32) && !defined(__wasi__)
+#if !defined(__wasi__)
 /* enable the os.Worker API. IT relies on POSIX threads */
 #define USE_WORKER
 #endif
 
 #ifdef USE_WORKER
-#include <pthread.h>
 #include "quickjs-c-atomics.h"
 #endif
 
@@ -143,7 +142,7 @@ typedef struct {
 typedef struct {
     int ref_count;
 #ifdef USE_WORKER
-    pthread_mutex_t mutex;
+    js_mutex_t mutex;
 #endif
     struct list_head msg_queue; /* list of JSWorkerMessage.link */
     int read_fd;
@@ -2461,7 +2460,7 @@ static int handle_posted_message(JSRuntime *rt, JSContext *ctx,
     JSWorkerMessage *msg;
     JSValue obj, data_obj, func, retval;
 
-    pthread_mutex_lock(&ps->mutex);
+    js_mutex_lock(&ps->mutex);
     if (!list_empty(&ps->msg_queue)) {
         el = ps->msg_queue.next;
         msg = list_entry(el, JSWorkerMessage, link);
@@ -2481,7 +2480,7 @@ static int handle_posted_message(JSRuntime *rt, JSContext *ctx,
             }
         }
 
-        pthread_mutex_unlock(&ps->mutex);
+        js_mutex_unlock(&ps->mutex);
 
         data_obj = JS_ReadObject(ctx, msg->data, msg->data_len,
                                  JS_READ_OBJ_SAB | JS_READ_OBJ_REFERENCE);
@@ -2511,7 +2510,7 @@ static int handle_posted_message(JSRuntime *rt, JSContext *ctx,
         }
         ret = 1;
     } else {
-        pthread_mutex_unlock(&ps->mutex);
+        js_mutex_unlock(&ps->mutex);
         ret = 0;
     }
     return ret;
@@ -3499,7 +3498,7 @@ static JSWorkerMessagePipe *js_new_message_pipe(void)
     }
     ps->ref_count = 1;
     init_list_head(&ps->msg_queue);
-    pthread_mutex_init(&ps->mutex, NULL);
+    js_mutex_init(&ps->mutex);
     ps->read_fd = pipe_fds[0];
     ps->write_fd = pipe_fds[1];
     return ps;
@@ -3539,7 +3538,7 @@ static void js_free_message_pipe(JSWorkerMessagePipe *ps)
             msg = list_entry(el, JSWorkerMessage, link);
             js_free_message(msg);
         }
-        pthread_mutex_destroy(&ps->mutex);
+        js_mutex_destroy(&ps->mutex);
         close(ps->read_fd);
         close(ps->write_fd);
         free(ps);
@@ -3573,7 +3572,7 @@ static JSClassDef js_worker_class = {
     .finalizer = js_worker_finalizer,
 };
 
-static void *worker_func(void *opaque)
+static void worker_func(void *opaque)
 {
     WorkerFuncArgs *args = opaque;
     JSRuntime *rt;
@@ -3620,7 +3619,6 @@ static void *worker_func(void *opaque)
     js_std_free_handlers(rt);
     JS_FreeContext(ctx);
     JS_FreeRuntime(rt);
-    return NULL;
 }
 
 static JSValue js_worker_ctor_internal(JSContext *ctx, JSValueConst new_target,
@@ -3662,8 +3660,7 @@ static JSValue js_worker_ctor(JSContext *ctx, JSValueConst new_target,
 {
     JSRuntime *rt = JS_GetRuntime(ctx);
     WorkerFuncArgs *args = NULL;
-    pthread_t tid;
-    pthread_attr_t attr;
+    js_thread_t thr;
     JSValue obj = JS_UNDEFINED;
     int ret;
     const char *filename = NULL, *basename;
@@ -3710,14 +3707,7 @@ static JSValue js_worker_ctor(JSContext *ctx, JSValueConst new_target,
     if (JS_IsException(obj))
         goto fail;
 
-    pthread_attr_init(&attr);
-    /* no join at the end */
-    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
-    // musl libc gives threads 80 kb stacks, much smaller than
-    // JS_DEFAULT_STACK_SIZE (1 MB)
-    pthread_attr_setstacksize(&attr, 2 << 20); // 2 MB, glibc default
-    ret = pthread_create(&tid, &attr, worker_func, args);
-    pthread_attr_destroy(&attr);
+    ret = js_thread_create(&thr, worker_func, args, JS_THREAD_CREATE_DETACHED);
     if (ret != 0) {
         JS_ThrowTypeError(ctx, "could not create worker");
         goto fail;
@@ -3792,7 +3782,7 @@ static JSValue js_worker_postMessage(JSContext *ctx, JSValueConst this_val,
     }
 
     ps = worker->send_pipe;
-    pthread_mutex_lock(&ps->mutex);
+    js_mutex_lock(&ps->mutex);
     /* indicate that data is present */
     if (list_empty(&ps->msg_queue)) {
         uint8_t ch = '\0';
@@ -3806,7 +3796,7 @@ static JSValue js_worker_postMessage(JSContext *ctx, JSValueConst this_val,
         }
     }
     list_add_tail(&msg->link, &ps->msg_queue);
-    pthread_mutex_unlock(&ps->mutex);
+    js_mutex_unlock(&ps->mutex);
     return JS_UNDEFINED;
  fail:
     if (msg) {
diff --git a/run-test262.c b/run-test262.c
@@ -35,12 +35,9 @@
 #ifdef _WIN32
 #include <windows.h>
 #include <process.h>
-typedef HANDLE js_thread_t;
 #else
 #include <dirent.h>
-#include <pthread.h>
 #include <unistd.h>
-typedef pthread_t js_thread_t;
 #endif
 
 #include "cutils.h"
@@ -530,49 +527,6 @@ static JSValue js_evalScript_262(JSContext *ctx, JSValueConst this_val,
     return ret;
 }
 
-static void start_thread(js_thread_t *thrd, void *(*start)(void *), void *arg)
-{
-    // musl libc gives threads 80 kb stacks, much smaller than
-    // JS_DEFAULT_STACK_SIZE (1 MB)
-    static const unsigned stacksize = 2 << 20; // 2 MB, glibc default
-#ifdef _WIN32
-    HANDLE h, cp;
-
-    cp = GetCurrentProcess();
-    h = (HANDLE)_beginthread((void (*)(void *))(void *)start, stacksize, arg);
-    if (!h)
-        fatal(1, "_beginthread error");
-    // _endthread() automatically closes the handle but we want to wait on
-    // it so make a copy. Race-y for very short-lived threads. Can be solved
-    // by switching to _beginthreadex(CREATE_SUSPENDED) but means changing
-    // |start| from __cdecl to __stdcall.
-    if (!DuplicateHandle(cp, h, cp, thrd, 0, false, DUPLICATE_SAME_ACCESS))
-        fatal(1, "DuplicateHandle error");
-#else
-    pthread_attr_t attr;
-
-    if (pthread_attr_init(&attr))
-        fatal(1, "pthread_attr_init");
-    if (pthread_attr_setstacksize(&attr, stacksize))
-        fatal(1, "pthread_attr_setstacksize");
-    if (pthread_create(thrd, &attr, start, arg))
-        fatal(1, "pthread_create error");
-    pthread_attr_destroy(&attr);
-#endif
-}
-
-static void join_thread(js_thread_t thrd)
-{
-#ifdef _WIN32
-    if (WaitForSingleObject(thrd, INFINITE))
-        fatal(1, "WaitForSingleObject error");
-    CloseHandle(thrd);
-#else
-    if (pthread_join(thrd, NULL))
-        fatal(1, "pthread_join error");
-#endif
-}
-
 static long cpu_count(void)
 {
 #ifdef _WIN32
@@ -621,7 +575,7 @@ typedef struct {
 static JSValue add_helpers1(JSContext *ctx);
 static void add_helpers(JSContext *ctx);
 
-static void *agent_start(void *arg)
+static void agent_start(void *arg)
 {
     ThreadLocalStorage *tls;
     Test262Agent *agent;
@@ -697,7 +651,6 @@ static void *agent_start(void *arg)
 
     JS_FreeContext(ctx);
     JS_FreeRuntime(rt);
-    return NULL;
 }
 
 static JSValue js_agent_start(JSContext *ctx, JSValueConst this_val,
@@ -721,7 +674,7 @@ static JSValue js_agent_start(JSContext *ctx, JSValueConst this_val,
     agent->tls = tls;
     JS_FreeCString(ctx, script);
     list_add_tail(&agent->link, &tls->agent_list);
-    start_thread(&agent->tid, agent_start, agent);
+    js_thread_create(&agent->tid, agent_start, agent, /*flags*/0);
     return JS_UNDEFINED;
 }
 
@@ -733,7 +686,7 @@ static void js_agent_free(JSContext *ctx)
 
     list_for_each_safe(el, el1, &tls->agent_list) {
         agent = list_entry(el, Test262Agent, link);
-        join_thread(agent->tid);
+        js_thread_join(agent->tid);
         JS_FreeValue(ctx, agent->broadcast_sab);
         list_del(&agent->link);
         free(agent);
@@ -2083,7 +2036,7 @@ int run_test262_harness_test(ThreadLocalStorage *tls, const char *filename,
 
 clock_t last_clock;
 
-void *show_progress(void *unused) {
+void show_progress(void *unused) {
     int interval = 1000*1000*1000 / 4; // 250 ms
 
     js_mutex_lock(&progress_mutex);
@@ -2096,7 +2049,6 @@ void *show_progress(void *unused) {
         fflush(stderr);
     }
     js_mutex_unlock(&progress_mutex);
-    return NULL;
 }
 
 enum { INCLUDE, EXCLUDE, SKIP };
@@ -2112,7 +2064,7 @@ int include_exclude_or_skip(int i) // naming is hard...
     return INCLUDE;
 }
 
-void *run_test_dir_list(void *arg)
+void run_test_dir_list(void *arg)
 {
     ThreadLocalStorage tls_s, *tls = &tls_s;
     const char *p;
@@ -2129,7 +2081,6 @@ void *run_test_dir_list(void *arg)
         if (verbose > 1 || (slow_test_threshold && msec >= slow_test_threshold))
             fprintf(stderr, "%s (%d ms)\n", p, msec);
     }
-    return NULL;
 }
 
 void help(void)
@@ -2304,15 +2255,17 @@ int main(int argc, char **argv)
         }
         js_cond_init(&progress_cond);
         js_mutex_init(&progress_mutex);
-        start_thread(&progress_thread, show_progress, NULL);
-        for (i = 0; i < nthreads; i++)
-            start_thread(&threads[i], run_test_dir_list, (void *)(uintptr_t)i);
+        js_thread_create(&progress_thread, show_progress, NULL, /*flags*/0);
+        for (i = 0; i < nthreads; i++) {
+            js_thread_create(&threads[i], run_test_dir_list,
+                             (void *)(uintptr_t)i, /*flags*/0);
+        }
         for (i = 0; i < nthreads; i++)
-            join_thread(threads[i]);
+            js_thread_join(threads[i]);
         js_mutex_lock(&progress_mutex);
         js_cond_signal(&progress_cond);
         js_mutex_unlock(&progress_mutex);
-        join_thread(progress_thread);
+        js_thread_join(progress_thread);
         js_mutex_destroy(&progress_mutex);
         js_cond_destroy(&progress_cond);
     } else {
diff --git a/tests/test_worker.js b/tests/test_worker.js
