worker_threads.c
8755416ca476master

worker_threads.c
View Options

			// stub: minimal node:worker_threads implementation
			// just enough for rolldown to run transforms

			#include <compat.h> // IWYU pragma: keep

			#include <uv.h>
			#include <stdio.h>
			#include <stdlib.h>
			#include <string.h>
			#include <signal.h>

			#ifdef _WIN32
			#include <io.h>
			#define WT_WRITE _write
			#define WT_STDOUT_FD 1
			extern char **_environ;
			#define environ _environ
			#else
			#include <unistd.h>
			#define WT_WRITE write
			#define WT_STDOUT_FD STDOUT_FILENO
			extern char **environ;
			#endif

			#include "ant.h"
			#include "internal.h"
			#include "runtime.h"
			#include "descriptors.h"
			#include "silver/engine.h"
			#include "modules/json.h"
			#include "modules/symbol.h"
			#include "modules/worker_threads.h"

			#define WT_ENV_MODE "ANT_WORKER_THREADS_MODE"
			#define WT_ENV_DATA_JSON "ANT_WORKER_DATA_JSON"
			#define WT_ENV_STORE_JSON "ANT_WORKER_ENV_DATA_JSON"
			#define WT_MSG_PREFIX "ANT_WT_MSG:"

			typedef struct ant_worker_thread {
			ant_t *js;
			uv_process_t process;
			uv_pipe_t stdout_pipe;
			bool spawned;
			bool exited;
			bool closing;
			bool closed;
			bool refed;
			int close_pending;
			int64_t exit_status;
			int term_signal;
			char *line_buf;
			size_t line_len;
			size_t line_cap;
			jshdl_t self_root;
			jshdl_t terminate_root;
			bool has_terminate_root;
			} ant_worker_thread_t;

			static inline bool wt_is_callable(jsval_t v) {
			uint8_t t = vtype(v);
			return t == T_FUNC \|\| t == T_CFUNC;
			}

			static bool wt_is_worker_mode(void) {
			const char *mode = getenv(WT_ENV_MODE);
			return mode && strcmp(mode, "1") == 0;
			}

			static jsval_t wt_get_or_create_env_store(ant_t *js) {
			jsval_t store = js_get_slot(js, js->global, SLOT_WT_ENV_STORE);
			if (!is_object_type(store)) {
			store = js_mkobj(js);
			js_set_slot(js, js->global, SLOT_WT_ENV_STORE, store);
			}
			return store;
			}

			static void wt_init_env_store(ant_t *js, bool is_worker) {
			jsval_t store = js_mkobj(js);
			if (is_worker) {
			const char *raw = getenv(WT_ENV_STORE_JSON);
			if (raw && raw[0]) {
			jsval_t input = js_mkstr(js, raw, strlen(raw));
			jsval_t parse_args[1] = {input};
			jsval_t parsed = js_json_parse(js, parse_args, 1);
			if (is_object_type(parsed)) store = parsed;
			}
			}
			js_set_slot(js, js->global, SLOT_WT_ENV_STORE, store);
			}

			static ant_worker_thread_t wt_get_worker(ant_t js, jsval_t this_obj) {
			if (!is_object_type(this_obj)) return NULL;
			jsval_t data = js_get_slot(js, this_obj, SLOT_DATA);
			if (vtype(data) != T_NUM) return NULL;
			return (ant_worker_thread_t *)(uintptr_t)js_getnum(data);
			}

			static bool wt_is_message_port(ant_t *js, jsval_t obj) {
			if (!is_object_type(obj)) return false;
			jsval_t tag = js_get_slot(js, obj, SLOT_WT_PORT_TAG);
			return js_truthy(js, tag);
			}

			static jsval_t wt_get_message_port_proto(ant_t *js) {
			jsval_t proto = js_get_slot(js, js->global, SLOT_WT_PORT_PROTO);
			if (!is_object_type(proto)) proto = js_mkobj(js);
			return proto;
			}

			static jsval_t wt_make_message_port(ant_t *js) {
			jsval_t port = js_mkobj(js);
			js_set_proto(js, port, wt_get_message_port_proto(js));
			js_set_slot(js, port, SLOT_WT_PORT_TAG, js_true);
			js_set_slot(js, port, SLOT_WT_PORT_QUEUE, js_mkarr(js));
			js_set_slot(js, port, SLOT_WT_PORT_HEAD, js_mknum(0));
			js_set_slot(js, port, SLOT_WT_PORT_PEER, js_mknull());
			js_set_slot(js, port, SLOT_WT_PORT_CLOSED, js_false);
			js_set_slot(js, port, SLOT_WT_PORT_STARTED, js_false);
			js_set_slot(js, port, SLOT_WT_PORT_ON_MESSAGE, js_mkundef());
			js_set_slot(js, port, SLOT_WT_PORT_ONCE_MESSAGE, js_mkundef());
			js_set(js, port, "onmessage", js_mkundef());
			return port;
			}

			static bool wt_port_is_closed(ant_t *js, jsval_t port) {
			return js_truthy(js, js_get_slot(js, port, SLOT_WT_PORT_CLOSED));
			}

			static void wt_port_set_closed(ant_t *js, jsval_t port, bool closed) {
			js_set_slot(js, port, SLOT_WT_PORT_CLOSED, js_bool(closed));
			}

			static bool wt_port_queue_dequeue(ant_t js, jsval_t port, jsval_t out) {
			jsval_t queue = js_get_slot(js, port, SLOT_WT_PORT_QUEUE);
			if (vtype(queue) != T_ARR) return false;

			jsval_t head_val = js_get_slot(js, port, SLOT_WT_PORT_HEAD);
			jsoff_t head = (vtype(head_val) == T_NUM) ? (jsoff_t)js_getnum(head_val) : 0;
			jsoff_t len = js_arr_len(js, queue);
			if (len <= 0 \|\| head >= len) {
			js_set_slot(js, port, SLOT_WT_PORT_QUEUE, js_mkarr(js));
			js_set_slot(js, port, SLOT_WT_PORT_HEAD, js_mknum(0));
			return false;
			}

			if (out) *out = js_arr_get(js, queue, head);
			jsoff_t next_head = head + 1;

			if (next_head >= len) {
			js_set_slot(js, port, SLOT_WT_PORT_QUEUE, js_mkarr(js));
			js_set_slot(js, port, SLOT_WT_PORT_HEAD, js_mknum(0));
			return true;
			}

			if (next_head > 32 && next_head * 2 >= len) {
			jsval_t compact = js_mkarr(js);
			for (jsoff_t i = next_head; i < len; i++) js_arr_push(js, compact, js_arr_get(js, queue, i));
			js_set_slot(js, port, SLOT_WT_PORT_QUEUE, compact);
			js_set_slot(js, port, SLOT_WT_PORT_HEAD, js_mknum(0));
			return true;
			}

			js_set_slot(js, port, SLOT_WT_PORT_HEAD, js_mknum((double)next_head));
			return true;
			}

			static void wt_port_queue_push(ant_t *js, jsval_t port, jsval_t value) {
			jsval_t queue = js_get_slot(js, port, SLOT_WT_PORT_QUEUE);
			if (vtype(queue) != T_ARR) {
			queue = js_mkarr(js);
			js_set_slot(js, port, SLOT_WT_PORT_QUEUE, queue);
			js_set_slot(js, port, SLOT_WT_PORT_HEAD, js_mknum(0));
			}
			js_arr_push(js, queue, value);
			}

			static void wt_port_call_listener(ant_t *js, jsval_t this_obj, jsval_t fn, jsval_t arg) {
			if (!wt_is_callable(fn)) return;
			jsval_t argv[1] = {arg};

			sv_vm_call(js->vm, js, fn, this_obj, argv, 1, NULL, false);
			}

			static bool wt_port_should_deliver(ant_t *js, jsval_t port) {
			if (wt_port_is_closed(js, port)) return false;
			bool started = js_truthy(js, js_get_slot(js, port, SLOT_WT_PORT_STARTED));
			jsval_t on_fn = js_get_slot(js, port, SLOT_WT_PORT_ON_MESSAGE);
			jsval_t once_fn = js_get_slot(js, port, SLOT_WT_PORT_ONCE_MESSAGE);
			jsval_t onmessage = js_get(js, port, "onmessage");

			bool has_event_listener = wt_is_callable(on_fn) \|\| wt_is_callable(once_fn);
			if (wt_is_callable(onmessage)) return true;
			return started && has_event_listener;
			}

			static void wt_port_drain(ant_t *js, jsval_t port) {
			if (!wt_is_message_port(js, port)) return;

			while (wt_port_should_deliver(js, port)) {
			jsval_t msg = js_mkundef();
			if (!wt_port_queue_dequeue(js, port, &msg)) break;

			jsval_t on_fn = js_get_slot(js, port, SLOT_WT_PORT_ON_MESSAGE);
			wt_port_call_listener(js, port, on_fn, msg);

			jsval_t once_fn = js_get_slot(js, port, SLOT_WT_PORT_ONCE_MESSAGE);
			if (wt_is_callable(once_fn)) {
			wt_port_call_listener(js, port, once_fn, msg);
			js_set_slot(js, port, SLOT_WT_PORT_ONCE_MESSAGE, js_mkundef());
			}

			jsval_t onmessage = js_get(js, port, "onmessage");
			if (wt_is_callable(onmessage)) {
			jsval_t event_obj = js_mkobj(js);
			js_set(js, event_obj, "data", msg);
			wt_port_call_listener(js, port, onmessage, event_obj);
			}

			if (wt_port_is_closed(js, port)) break;
			}
			}

			static jsval_t wt_make_resolved_promise(ant_t *js, jsval_t value) {
			jsval_t p = js_mkpromise(js);
			js_resolve_promise(js, p, value);
			return p;
			}

			static void wt_call_listener(ant_t *js, jsval_t this_obj, jsval_t fn, jsval_t arg) {
			if (!wt_is_callable(fn)) return;
			jsval_t argv[1] = {arg};

			sv_vm_call(js->vm, js, fn, this_obj, argv, 1, NULL, false);
			}

			static void wt_emit(ant_worker_thread_t wt, const char event, jsval_t arg) {
			if (!wt \|\| !wt->js) return;
			ant_t *js = wt->js;
			jsval_t this_obj = js_deref(js, wt->self_root);
			if (!is_object_type(this_obj)) return;

			internal_slot_t on_slot, once_slot;
			if (strcmp(event, "message") == 0) {
			on_slot = SLOT_WT_ON_MESSAGE;
			once_slot = SLOT_WT_ONCE_MESSAGE;
			} else if (strcmp(event, "exit") == 0) {
			on_slot = SLOT_WT_ON_EXIT;
			once_slot = SLOT_WT_ONCE_EXIT;
			} else return;

			jsval_t on_fn = js_get_slot(js, this_obj, on_slot);
			wt_call_listener(js, this_obj, on_fn, arg);

			jsval_t once_fn = js_get_slot(js, this_obj, once_slot);
			if (wt_is_callable(once_fn)) {
			wt_call_listener(js, this_obj, once_fn, arg);
			js_set_slot(js, this_obj, once_slot, js_mkundef());
			}
			}

			static void wt_free_env(char **env) {
			if (!env) return;
			for (char *p = env; p; p++) free(*p);
			free(env);
			}

			static char *wt_build_worker_env(const char worker_data_json, const char *env_store_json) {
			size_t count = 0;
			if (environ) {
			while (environ[count]) count++;
			}

			size_t extra = 2;
			if (worker_data_json) extra++;
			if (env_store_json) extra++;
			char env = (char )calloc(count + extra, sizeof(char *));
			if (!env) return NULL;

			size_t out = 0;
			for (size_t i = 0; i < count; i++) {
			env[out] = strdup(environ[i]);
			if (!env[out]) {
			wt_free_env(env);
			return NULL;
			}
			out++;
			}

			env[out++] = strdup(WT_ENV_MODE "=1");
			if (!env[out - 1]) {
			wt_free_env(env);
			return NULL;
			}

			if (worker_data_json) {
			size_t key_len = strlen(WT_ENV_DATA_JSON);
			size_t val_len = strlen(worker_data_json);
			char entry = (char )malloc(key_len + 1 + val_len + 1);
			if (!entry) {
			wt_free_env(env);
			return NULL;
			}
			memcpy(entry, WT_ENV_DATA_JSON, key_len);
			entry[key_len] = '=';
			memcpy(entry + key_len + 1, worker_data_json, val_len);
			entry[key_len + 1 + val_len] = '\0';
			env[out++] = entry;
			}

			if (env_store_json) {
			size_t key_len = strlen(WT_ENV_STORE_JSON);
			size_t val_len = strlen(env_store_json);
			char entry = (char )malloc(key_len + 1 + val_len + 1);
			if (!entry) {
			wt_free_env(env);
			return NULL;
			}
			memcpy(entry, WT_ENV_STORE_JSON, key_len);
			entry[key_len] = '=';
			memcpy(entry + key_len + 1, env_store_json, val_len);
			entry[key_len + 1 + val_len] = '\0';
			env[out++] = entry;
			}

			env[out] = NULL;
			return env;
			}

			static void wt_cleanup(ant_worker_thread_t *wt) {
			if (!wt) return;
			if (wt->js) {
			if (wt->self_root) js_unroot(wt->js, wt->self_root);
			if (wt->has_terminate_root) js_unroot(wt->js, wt->terminate_root);
			}
			free(wt->line_buf);
			free(wt);
			}

			static void wt_detach(ant_worker_thread_t *wt) {
			if (!wt) return;
			if (wt->js) {
			if (wt->self_root) js_unroot(wt->js, wt->self_root);
			if (wt->has_terminate_root) js_unroot(wt->js, wt->terminate_root);
			}
			wt->self_root = 0;
			wt->terminate_root = 0;
			wt->has_terminate_root = false;
			free(wt->line_buf);
			wt->line_buf = NULL;
			wt->line_len = 0;
			wt->line_cap = 0;
			wt->spawned = false;
			wt->refed = false;
			wt->closed = true;
			wt->js = NULL;
			}

			static void wt_on_handle_closed(uv_handle_t *h) {
			ant_worker_thread_t wt = (ant_worker_thread_t )h->data;
			if (!wt) return;
			if (wt->close_pending > 0) wt->close_pending--;
			if (wt->close_pending == 0) wt_detach(wt);
			}

			static void wt_finish_exit(ant_worker_thread_t *wt) {
			if (!wt) return;
			if (wt->closed \|\| wt->closing) return;
			wt->closing = true;
			wt->spawned = false;
			wt->refed = false;
			wt->close_pending = 0;

			if (!uv_is_closing((uv_handle_t *)&wt->stdout_pipe)) {
			wt->close_pending++;
			uv_close((uv_handle_t *)&wt->stdout_pipe, wt_on_handle_closed);
			}
			if (!uv_is_closing((uv_handle_t *)&wt->process)) {
			wt->close_pending++;
			uv_close((uv_handle_t *)&wt->process, wt_on_handle_closed);
			}

			if (wt->close_pending == 0) wt_detach(wt);
			}

			static void wt_on_process_exit(uv_process_t *proc, int64_t exit_status, int term_signal) {
			ant_worker_thread_t wt = (ant_worker_thread_t )proc->data;
			if (!wt \|\| !wt->js) return;

			wt->exited = true;
			wt->exit_status = exit_status;
			wt->term_signal = term_signal;

			if (wt->has_terminate_root) {
			jsval_t p = js_deref(wt->js, wt->terminate_root);
			js_resolve_promise(wt->js, p, js_mknum((double)exit_status));
			js_root_update(wt->js, wt->terminate_root, js_mkundef());
			wt->has_terminate_root = false;
			}

			wt_emit(wt, "exit", js_mknum((double)exit_status));
			wt_finish_exit(wt);
			}

			static void wt_alloc_cb(uv_handle_t handle, size_t suggested_size, uv_buf_t buf) {
			buf->base = (char *)malloc(suggested_size);
			buf->len = buf->base ? suggested_size : 0;
			}

			static void wt_emit_message_from_json(ant_worker_thread_t wt, const char json, size_t len) {
			if (!wt \|\| !wt->js) return;
			ant_t *js = wt->js;

			jsval_t s = js_mkstr(js, json, len);
			jsval_t parse_args[1] = {s};
			jsval_t msg = js_json_parse(js, parse_args, 1);
			if (is_err(msg)) msg = s;

			wt_emit(wt, "message", msg);
			}

			static void wt_process_lines(ant_worker_thread_t *wt) {
			if (!wt \|\| !wt->line_buf) return;

			for (;;) {
			char *nl = memchr(wt->line_buf, '\n', wt->line_len);
			if (!nl) break;

			size_t line_len = (size_t)(nl - wt->line_buf);
			if (line_len > 0 && wt->line_buf[line_len - 1] == '\r') line_len--;

			size_t prefix_len = strlen(WT_MSG_PREFIX);
			if (line_len >= prefix_len && memcmp(wt->line_buf, WT_MSG_PREFIX, prefix_len) == 0) {
			const char *payload = wt->line_buf + prefix_len;
			size_t payload_len = line_len - prefix_len;
			wt_emit_message_from_json(wt, payload, payload_len);
			}

			size_t consumed = (size_t)(nl - wt->line_buf) + 1;
			size_t remain = wt->line_len - consumed;
			memmove(wt->line_buf, wt->line_buf + consumed, remain);
			wt->line_len = remain;
			}
			}

			static bool wt_append_stdout(ant_worker_thread_t wt, const char data, size_t len) {
			if (!wt \|\| !data \|\| len == 0) return true;

			size_t needed = wt->line_len + len;
			if (needed + 1 > wt->line_cap) {
			size_t cap = wt->line_cap ? wt->line_cap : 1024;
			while (cap < needed + 1) cap *= 2;
			char next = (char )realloc(wt->line_buf, cap);
			if (!next) return false;
			wt->line_buf = next;
			wt->line_cap = cap;
			}

			memcpy(wt->line_buf + wt->line_len, data, len);
			wt->line_len += len;
			wt->line_buf[wt->line_len] = '\0';
			return true;
			}

			static void wt_read_cb(uv_stream_t stream, ssize_t nread, const uv_buf_t buf) {
			ant_worker_thread_t wt = (ant_worker_thread_t )stream->data;
			if (!wt) {
			free(buf->base);
			return;
			}

			if (nread > 0) {
			if (wt_append_stdout(wt, buf->base, (size_t)nread)) wt_process_lines(wt);
			} else if (nread < 0) {
			uv_read_stop(stream);
			}

			free(buf->base);
			}

			static char wt_path_from_specifier(ant_t js, jsval_t spec) {
			const char *raw = NULL;
			size_t len = 0;

			if (vtype(spec) == T_STR) {
			raw = js_getstr(js, spec, &len);
			} else if (is_object_type(spec)) {
			jsval_t pathname = js_get(js, spec, "pathname");
			if (vtype(pathname) == T_STR) raw = js_getstr(js, pathname, &len);
			if (!raw) {
			jsval_t href = js_get(js, spec, "href");
			if (vtype(href) == T_STR) raw = js_getstr(js, href, &len);
			}
			}

			if (!raw \|\| len == 0) return NULL;

			if (len >= 7 && strncmp(raw, "file://", 7) == 0) {
			const char *p = raw + 7;
			if (strncmp(p, "localhost/", 10) == 0) p += 9;
			if (*p == '\0') return NULL;
			return strndup(p, len - (size_t)(p - raw));
			}

			return strndup(raw, len);
			}

			static int wt_spawn_worker(
			ant_worker_thread_t *wt,
			const char *script_path,
			const char *worker_data_json,
			const char *env_store_json
			) {
			if (!wt \|\| !wt->js \|\| !script_path \|\| !rt \|\| !rt->argv \|\| rt->argc <= 0) return UV_EINVAL;

			uv_loop_t *loop = uv_default_loop();
			uv_pipe_init(loop, &wt->stdout_pipe, 0);

			uv_stdio_container_t stdio[3];
			stdio[0].flags = UV_IGNORE;
			stdio[1].flags = UV_CREATE_PIPE \| UV_WRITABLE_PIPE;
			stdio[1].data.stream = (uv_stream_t *)&wt->stdout_pipe;
			stdio[2].flags = UV_INHERIT_FD;
			stdio[2].data.fd = 2;

			char *argv0 = strdup(rt->argv[0]);
			char *argv1 = strdup(script_path);
			if (!argv0 \|\| !argv1) {
			free(argv0);
			free(argv1);
			uv_close((uv_handle_t *)&wt->stdout_pipe, NULL);
			return UV_ENOMEM;
			}

			char *args[3] = {argv0, argv1, NULL};
			char **env = wt_build_worker_env(worker_data_json, env_store_json);
			if (!env) {
			free(argv0);
			free(argv1);
			uv_close((uv_handle_t *)&wt->stdout_pipe, NULL);
			return UV_ENOMEM;
			}

			uv_process_options_t options;
			memset(&options, 0, sizeof(options));
			options.file = argv0;
			options.args = args;
			options.env = env;
			options.stdio_count = 3;
			options.stdio = stdio;
			options.exit_cb = wt_on_process_exit;

			wt->process.data = wt;
			wt->stdout_pipe.data = wt;

			int rc = uv_spawn(loop, &wt->process, &options);

			wt_free_env(env);
			free(argv0);
			free(argv1);

			if (rc != 0) {
			uv_close((uv_handle_t *)&wt->stdout_pipe, NULL);
			return rc;
			}

			wt->spawned = true;
			wt->refed = true;

			rc = uv_read_start((uv_stream_t *)&wt->stdout_pipe, wt_alloc_cb, wt_read_cb);
			if (rc != 0) {
			uv_process_kill(&wt->process, SIGTERM);
			}
			return rc;
			}

			static jsval_t worker_threads_worker_on(ant_t js, jsval_t args, int nargs) {
			if (nargs < 2 \|\| vtype(args[0]) != T_STR \|\| !wt_is_callable(args[1])) {
			return js_mkerr(js, "Worker.on(event, listener) requires (string, function)");
			}

			jsval_t this_obj = js_getthis(js);
			if (!is_object_type(this_obj)) return js_mkerr(js, "invalid Worker receiver");

			size_t len = 0;
			const char *event = js_getstr(js, args[0], &len);
			if (!event) return js_mkerr(js, "invalid event name");

			if (len == 7 && memcmp(event, "message", 7) == 0) {
			js_set_slot(js, this_obj, SLOT_WT_ON_MESSAGE, args[1]);
			} else if (len == 4 && memcmp(event, "exit", 4) == 0) {
			js_set_slot(js, this_obj, SLOT_WT_ON_EXIT, args[1]);
			}

			return this_obj;
			}

			static jsval_t worker_threads_worker_once(ant_t js, jsval_t args, int nargs) {
			if (nargs < 2 \|\| vtype(args[0]) != T_STR \|\| !wt_is_callable(args[1])) {
			return js_mkerr(js, "Worker.once(event, listener) requires (string, function)");
			}

			jsval_t this_obj = js_getthis(js);
			if (!is_object_type(this_obj)) return js_mkerr(js, "invalid Worker receiver");

			size_t len = 0;
			const char *event = js_getstr(js, args[0], &len);
			if (!event) return js_mkerr(js, "invalid event name");

			if (len == 7 && memcmp(event, "message", 7) == 0) {
			js_set_slot(js, this_obj, SLOT_WT_ONCE_MESSAGE, args[1]);
			} else if (len == 4 && memcmp(event, "exit", 4) == 0) {
			js_set_slot(js, this_obj, SLOT_WT_ONCE_EXIT, args[1]);
			}

			return this_obj;
			}

			static jsval_t worker_threads_worker_unref(ant_t js, jsval_t args, int nargs) {
			jsval_t this_obj = js_getthis(js);
			ant_worker_thread_t *wt = wt_get_worker(js, this_obj);
			if (!wt) return js_mkerr(js, "invalid Worker receiver");
			if (wt->spawned && wt->refed) {
			if (!uv_is_closing((uv_handle_t )&wt->process)) uv_unref((uv_handle_t )&wt->process);
			if (!uv_is_closing((uv_handle_t )&wt->stdout_pipe)) uv_unref((uv_handle_t )&wt->stdout_pipe);
			wt->refed = false;
			}
			return this_obj;
			}

			static jsval_t worker_threads_worker_ref(ant_t js, jsval_t args, int nargs) {
			jsval_t this_obj = js_getthis(js);
			ant_worker_thread_t *wt = wt_get_worker(js, this_obj);
			if (!wt) return js_mkerr(js, "invalid Worker receiver");
			if (wt->spawned && !wt->refed) {
			if (!uv_is_closing((uv_handle_t )&wt->process)) uv_ref((uv_handle_t )&wt->process);
			if (!uv_is_closing((uv_handle_t )&wt->stdout_pipe)) uv_ref((uv_handle_t )&wt->stdout_pipe);
			wt->refed = true;
			}
			return this_obj;
			}

			static jsval_t worker_threads_worker_terminate(ant_t js, jsval_t args, int nargs) {
			jsval_t this_obj = js_getthis(js);
			ant_worker_thread_t *wt = wt_get_worker(js, this_obj);
			if (!wt) return js_mkerr(js, "invalid Worker receiver");

			if (wt->exited) return wt_make_resolved_promise(js, js_mknum((double)wt->exit_status));

			if (!wt->has_terminate_root) {
			jsval_t p = js_mkpromise(js);
			wt->terminate_root = js_root(js, p);
			wt->has_terminate_root = true;
			}

			int rc = uv_process_kill(&wt->process, SIGTERM);
			if (rc != 0) {
			jsval_t p = js_deref(js, wt->terminate_root);
			js_reject_promise(js, p, js_mkerr(js, "terminate failed: %s", uv_strerror(rc)));
			js_root_update(js, wt->terminate_root, js_mkundef());
			wt->has_terminate_root = false;
			return p;
			}

			return js_deref(js, wt->terminate_root);
			}

			static jsval_t worker_threads_worker_post_message(ant_t js, jsval_t args, int nargs) {
			return js_mkerr(js, "Worker.postMessage is not implemented yet");
			}

			static jsval_t worker_threads_message_port_post_message(ant_t js, jsval_t args, int nargs) {
			jsval_t this_obj = js_getthis(js);
			if (!wt_is_message_port(js, this_obj)) return js_mkerr(js, "invalid MessagePort receiver");
			if (wt_port_is_closed(js, this_obj)) return js_mkundef();

			jsval_t peer = js_get_slot(js, this_obj, SLOT_WT_PORT_PEER);
			if (!wt_is_message_port(js, peer) \|\| wt_port_is_closed(js, peer)) return js_mkundef();

			jsval_t value = (nargs > 0) ? args[0] : js_mkundef();
			wt_port_queue_push(js, peer, value);
			wt_port_drain(js, peer);
			return js_mkundef();
			}

			static jsval_t worker_threads_message_port_on(ant_t js, jsval_t args, int nargs) {
			if (nargs < 2 \|\| vtype(args[0]) != T_STR \|\| !wt_is_callable(args[1])) {
			return js_mkerr(js, "MessagePort.on(event, listener) requires (string, function)");
			}
			jsval_t this_obj = js_getthis(js);
			if (!wt_is_message_port(js, this_obj)) return js_mkerr(js, "invalid MessagePort receiver");

			size_t len = 0;
			const char *event = js_getstr(js, args[0], &len);
			if (!event) return js_mkerr(js, "invalid event name");
			if (len == 7 && memcmp(event, "message", 7) == 0) {
			js_set_slot(js, this_obj, SLOT_WT_PORT_ON_MESSAGE, args[1]);
			js_set_slot(js, this_obj, SLOT_WT_PORT_STARTED, js_true);
			wt_port_drain(js, this_obj);
			}
			return this_obj;
			}

			static jsval_t worker_threads_message_port_once(ant_t js, jsval_t args, int nargs) {
			if (nargs < 2 \|\| vtype(args[0]) != T_STR \|\| !wt_is_callable(args[1])) {
			return js_mkerr(js, "MessagePort.once(event, listener) requires (string, function)");
			}
			jsval_t this_obj = js_getthis(js);
			if (!wt_is_message_port(js, this_obj)) return js_mkerr(js, "invalid MessagePort receiver");

			size_t len = 0;
			const char *event = js_getstr(js, args[0], &len);
			if (!event) return js_mkerr(js, "invalid event name");
			if (len == 7 && memcmp(event, "message", 7) == 0) {
			js_set_slot(js, this_obj, SLOT_WT_PORT_ONCE_MESSAGE, args[1]);
			js_set_slot(js, this_obj, SLOT_WT_PORT_STARTED, js_true);
			wt_port_drain(js, this_obj);
			}
			return this_obj;
			}

			static jsval_t worker_threads_message_port_start(ant_t js, jsval_t args, int nargs) {
			jsval_t this_obj = js_getthis(js);
			if (!wt_is_message_port(js, this_obj)) return js_mkerr(js, "invalid MessagePort receiver");
			js_set_slot(js, this_obj, SLOT_WT_PORT_STARTED, js_true);
			wt_port_drain(js, this_obj);
			return js_mkundef();
			}

			static jsval_t worker_threads_message_port_close(ant_t js, jsval_t args, int nargs) {
			jsval_t this_obj = js_getthis(js);
			if (!wt_is_message_port(js, this_obj)) return js_mkerr(js, "invalid MessagePort receiver");
			wt_port_set_closed(js, this_obj, true);

			jsval_t peer = js_get_slot(js, this_obj, SLOT_WT_PORT_PEER);
			js_set_slot(js, this_obj, SLOT_WT_PORT_PEER, js_mknull());
			if (wt_is_message_port(js, peer)) js_set_slot(js, peer, SLOT_WT_PORT_PEER, js_mknull());
			return js_mkundef();
			}

			static jsval_t worker_threads_message_port_ref(ant_t js, jsval_t args, int nargs) {
			return js_mkundef();
			}

			static jsval_t worker_threads_message_port_unref(ant_t js, jsval_t args, int nargs) {
			return js_mkundef();
			}

			static jsval_t worker_threads_message_port_ctor(ant_t js, jsval_t args, int nargs) {
			return js_mkerr(js, "MessagePort constructor is not public");
			}

			static jsval_t worker_threads_message_channel_ctor(ant_t js, jsval_t args, int nargs) {
			if (vtype(js->new_target) == T_UNDEF) {
			return js_mkerr(js, "MessageChannel constructor requires 'new'");
			}

			jsval_t this_obj = js_getthis(js);
			jsval_t port1 = wt_make_message_port(js);
			jsval_t port2 = wt_make_message_port(js);
			js_set_slot(js, port1, SLOT_WT_PORT_PEER, port2);
			js_set_slot(js, port2, SLOT_WT_PORT_PEER, port1);
			js_set(js, this_obj, "port1", port1);
			js_set(js, this_obj, "port2", port2);
			return this_obj;
			}

			static jsval_t worker_threads_worker_ctor(ant_t js, jsval_t args, int nargs) {
			if (vtype(js->new_target) == T_UNDEF) {
			return js_mkerr(js, "Worker constructor requires 'new'");
			}
			if (nargs < 1) return js_mkerr(js, "Worker() requires a filename or URL");

			char *script_path = wt_path_from_specifier(js, args[0]);
			if (!script_path) return js_mkerr(js, "Invalid Worker filename/URL");

			const char *worker_data_json = NULL;
			const char *env_store_json = NULL;
			char *worker_data_heap = NULL;
			char *env_store_heap = NULL;
			if (nargs >= 2 && is_object_type(args[1])) {
			jsval_t worker_data = js_get(js, args[1], "workerData");
			if (!is_undefined(worker_data)) {
			jsval_t stringify_args[1] = {worker_data};
			jsval_t json = js_json_stringify(js, stringify_args, 1);
			if (vtype(json) != T_STR) {
			free(script_path);
			return js_mkerr(js, "Worker options.workerData must be JSON-serializable");
			}
			size_t len = 0;
			const char *raw = js_getstr(js, json, &len);
			if (!raw) {
			free(script_path);
			return js_mkerr(js, "Failed to stringify workerData");
			}
			worker_data_heap = strndup(raw, len);
			if (!worker_data_heap) {
			free(script_path);
			return js_mkerr(js, "Out of memory");
			}
			worker_data_json = worker_data_heap;
			}
			}

			jsval_t env_store = wt_get_or_create_env_store(js);
			jsval_t env_stringify_args[1] = {env_store};
			jsval_t env_json = js_json_stringify(js, env_stringify_args, 1);
			if (vtype(env_json) != T_STR) {
			free(script_path);
			free(worker_data_heap);
			return js_mkerr(js, "setEnvironmentData values must be JSON-serializable");
			}
			size_t env_len = 0;
			const char *env_raw = js_getstr(js, env_json, &env_len);
			if (!env_raw) {
			free(script_path);
			free(worker_data_heap);
			return js_mkerr(js, "Failed to snapshot environment data");
			}
			env_store_heap = strndup(env_raw, env_len);
			if (!env_store_heap) {
			free(script_path);
			free(worker_data_heap);
			return js_mkerr(js, "Out of memory");
			}
			env_store_json = env_store_heap;

			jsval_t this_obj = js_getthis(js);
			ant_worker_thread_t wt = (ant_worker_thread_t )calloc(1, sizeof(*wt));
			if (!wt) {
			free(script_path);
			free(worker_data_heap);
			free(env_store_heap);
			return js_mkerr(js, "Out of memory");
			}

			wt->js = js;
			wt->self_root = js_root(js, this_obj);
			js_set_slot(js, this_obj, SLOT_DATA, ANT_PTR(wt));

			int rc = wt_spawn_worker(wt, script_path, worker_data_json, env_store_json);
			free(script_path);
			free(worker_data_heap);
			free(env_store_heap);

			if (rc != 0) {
			js_set_slot(js, this_obj, SLOT_DATA, js_mkundef());
			wt_cleanup(wt);
			return js_mkerr(js, "Failed to spawn Worker: %s", uv_strerror(rc));
			}

			return this_obj;
			}

			static jsval_t worker_threads_parent_post_message(ant_t js, jsval_t args, int nargs) {
			jsval_t value = (nargs > 0) ? args[0] : js_mkundef();
			jsval_t stringify_args[1] = {value};
			jsval_t json = js_json_stringify(js, stringify_args, 1);
			if (vtype(json) != T_STR) return js_mkerr(js, "parentPort.postMessage payload must be JSON-serializable");

			size_t json_len = 0;
			const char *json_str = js_getstr(js, json, &json_len);
			if (!json_str) return js_mkerr(js, "Failed to serialize message");

			size_t prefix_len = strlen(WT_MSG_PREFIX);
			size_t line_len = prefix_len + json_len + 1;
			char line = (char )malloc(line_len + 1);
			if (!line) return js_mkerr(js, "Out of memory");

			memcpy(line, WT_MSG_PREFIX, prefix_len);
			memcpy(line + prefix_len, json_str, json_len);
			line[prefix_len + json_len] = '\n';
			line[prefix_len + json_len + 1] = '\0';

			ssize_t wrote = WT_WRITE(WT_STDOUT_FD, line, (unsigned int)line_len);
			free(line);
			if (wrote < 0) return js_mkerr(js, "parentPort.postMessage failed");
			return js_mkundef();
			}

			static jsval_t worker_threads_parent_unref(ant_t js, jsval_t args, int nargs) {
			return js_mkundef();
			}

			static jsval_t worker_threads_mark_as_untransferable(ant_t js, jsval_t args, int nargs) {
			if (nargs < 1) return js_mkundef();
			return args[0];
			}

			static jsval_t worker_threads_receive_message_on_port(ant_t js, jsval_t args, int nargs) {
			if (nargs < 1 \|\| !wt_is_message_port(js, args[0])) {
			return js_mkerr(js, "receiveMessageOnPort(port) requires a MessagePort");
			}
			jsval_t msg = js_mkundef();
			if (!wt_port_queue_dequeue(js, args[0], &msg)) return js_mkundef();

			jsval_t out = js_mkobj(js);
			js_set(js, out, "message", msg);
			return out;
			}

			static jsval_t worker_threads_set_environment_data(ant_t js, jsval_t args, int nargs) {
			if (nargs < 2) return js_mkerr(js, "setEnvironmentData(key, value) requires 2 arguments");

			jsval_t key_stringify_args[1] = {args[0]};
			jsval_t key_json = js_json_stringify(js, key_stringify_args, 1);
			if (vtype(key_json) != T_STR) return js_mkerr(js, "setEnvironmentData key must be JSON-serializable");

			jsval_t value_stringify_args[1] = {args[1]};
			jsval_t value_json = js_json_stringify(js, value_stringify_args, 1);
			if (vtype(value_json) != T_STR) return js_mkerr(js, "setEnvironmentData value must be JSON-serializable");

			jsval_t parse_args[1] = {value_json};
			jsval_t cloned = js_json_parse(js, parse_args, 1);
			if (is_err(cloned)) return js_mkerr(js, "setEnvironmentData value must be JSON-serializable");

			size_t key_len = 0;
			const char *key_ptr = js_getstr(js, key_json, &key_len);
			if (!key_ptr) return js_mkerr(js, "Failed to serialize environment data key");

			char *key = strndup(key_ptr, key_len);
			if (!key) return js_mkerr(js, "Out of memory");
			js_set(js, wt_get_or_create_env_store(js), key, cloned);
			free(key);
			return js_mkundef();
			}

			static jsval_t worker_threads_get_environment_data(ant_t js, jsval_t args, int nargs) {
			if (nargs < 1) return js_mkundef();

			jsval_t key_stringify_args[1] = {args[0]};
			jsval_t key_json = js_json_stringify(js, key_stringify_args, 1);
			if (vtype(key_json) != T_STR) return js_mkundef();

			size_t key_len = 0;
			const char *key_ptr = js_getstr(js, key_json, &key_len);
			if (!key_ptr) return js_mkundef();

			char *key = strndup(key_ptr, key_len);
			if (!key) return js_mkerr(js, "Out of memory");

			jsval_t store = wt_get_or_create_env_store(js);
			jsoff_t off = lkp(js, store, key, key_len);
			if (off == 0) {
			free(key);
			return js_mkundef();
			}

			jsval_t value = js_get(js, store, key);
			free(key);
			return value;
			}

			static jsval_t worker_threads_move_message_port_to_context(ant_t js, jsval_t args, int nargs) {
			if (nargs < 1 \|\| !wt_is_message_port(js, args[0])) {
			return js_mkerr(js, "moveMessagePortToContext(port, context) requires a MessagePort");
			}
			return args[0];
			}

			jsval_t worker_threads_library(ant_t *js) {
			jsval_t lib = js_mkobj(js);
			bool is_worker = wt_is_worker_mode();

			wt_init_env_store(js, is_worker);

			js_set(js, lib, "isMainThread", js_bool(!is_worker));
			js_set(js, lib, "threadId", js_mknum((double)(is_worker ? rt->pid : 0)));
			js_set(js, lib, "SHARE_ENV", js_mksym(js, "SHARE_ENV"));

			jsval_t message_port_ctor_obj = js_mkobj(js);
			jsval_t message_port_proto = js_mkobj(js);
			js_set(js, message_port_proto, "postMessage", js_mkfun(worker_threads_message_port_post_message));
			js_set(js, message_port_proto, "on", js_mkfun(worker_threads_message_port_on));
			js_set(js, message_port_proto, "once", js_mkfun(worker_threads_message_port_once));
			js_set(js, message_port_proto, "start", js_mkfun(worker_threads_message_port_start));
			js_set(js, message_port_proto, "close", js_mkfun(worker_threads_message_port_close));
			js_set(js, message_port_proto, "ref", js_mkfun(worker_threads_message_port_ref));
			js_set(js, message_port_proto, "unref", js_mkfun(worker_threads_message_port_unref));
			js_set_sym(js, message_port_proto, get_toStringTag_sym(), js_mkstr(js, "MessagePort", 11));

			js_set_slot(js, message_port_ctor_obj, SLOT_CFUNC, js_mkfun(worker_threads_message_port_ctor));
			js_mkprop_fast(js, message_port_ctor_obj, "prototype", 9, message_port_proto);
			js_mkprop_fast(js, message_port_ctor_obj, "name", 4, js_mkstr(js, "MessagePort", 11));
			js_set_descriptor(js, message_port_ctor_obj, "name", 4, 0);
			js_set(js, lib, "MessagePort", js_obj_to_func(message_port_ctor_obj));
			js_set_slot(js, js->global, SLOT_WT_PORT_PROTO, message_port_proto);

			jsval_t message_channel_ctor_obj = js_mkobj(js);
			jsval_t message_channel_proto = js_mkobj(js);
			js_set_sym(js, message_channel_proto, get_toStringTag_sym(), js_mkstr(js, "MessageChannel", 14));
			js_set_slot(js, message_channel_ctor_obj, SLOT_CFUNC, js_mkfun(worker_threads_message_channel_ctor));
			js_mkprop_fast(js, message_channel_ctor_obj, "prototype", 9, message_channel_proto);
			js_mkprop_fast(js, message_channel_ctor_obj, "name", 4, js_mkstr(js, "MessageChannel", 14));
			js_set_descriptor(js, message_channel_ctor_obj, "name", 4, 0);
			js_set(js, lib, "MessageChannel", js_obj_to_func(message_channel_ctor_obj));

			if (is_worker) {
			jsval_t parent_port = js_mkobj(js);
			js_set(js, parent_port, "postMessage", js_mkfun(worker_threads_parent_post_message));
			js_set(js, parent_port, "unref", js_mkfun(worker_threads_parent_unref));
			js_set(js, parent_port, "ref", js_mkfun(worker_threads_parent_unref));
			js_set(js, lib, "parentPort", parent_port);

			const char *worker_data_json = getenv(WT_ENV_DATA_JSON);
			if (worker_data_json && worker_data_json[0]) {
			jsval_t raw = js_mkstr(js, worker_data_json, strlen(worker_data_json));
			jsval_t parse_args[1] = {raw};
			jsval_t parsed = js_json_parse(js, parse_args, 1);
			js_set(js, lib, "workerData", is_err(parsed) ? js_mkundef() : parsed);
			} else js_set(js, lib, "workerData", js_mkundef());
			} else {
			js_set(js, lib, "parentPort", js_mknull());
			js_set(js, lib, "workerData", js_mkundef());
			}

			jsval_t worker_ctor_obj = js_mkobj(js);
			jsval_t worker_proto = js_mkobj(js);

			js_set(js, worker_proto, "on", js_mkfun(worker_threads_worker_on));
			js_set(js, worker_proto, "once", js_mkfun(worker_threads_worker_once));
			js_set(js, worker_proto, "terminate", js_mkfun(worker_threads_worker_terminate));
			js_set(js, worker_proto, "unref", js_mkfun(worker_threads_worker_unref));
			js_set(js, worker_proto, "ref", js_mkfun(worker_threads_worker_ref));
			js_set(js, worker_proto, "postMessage", js_mkfun(worker_threads_worker_post_message));
			js_set_sym(js, worker_proto, get_toStringTag_sym(), js_mkstr(js, "Worker", 6));

			js_set_slot(js, worker_ctor_obj, SLOT_CFUNC, js_mkfun(worker_threads_worker_ctor));
			js_mkprop_fast(js, worker_ctor_obj, "prototype", 9, worker_proto);
			js_mkprop_fast(js, worker_ctor_obj, "name", 4, js_mkstr(js, "Worker", 6));
			js_set_descriptor(js, worker_ctor_obj, "name", 4, 0);
			js_set(js, lib, "Worker", js_obj_to_func(worker_ctor_obj));

			js_set(js, lib, "markAsUntransferable", js_mkfun(worker_threads_mark_as_untransferable));
			js_set(js, lib, "receiveMessageOnPort", js_mkfun(worker_threads_receive_message_on_port));
			js_set(js, lib, "setEnvironmentData", js_mkfun(worker_threads_set_environment_data));
			js_set(js, lib, "getEnvironmentData", js_mkfun(worker_threads_get_environment_data));
			js_set(js, lib, "moveMessagePortToContext", js_mkfun(worker_threads_move_message_port_to_context));

			js_set_sym(js, lib, get_toStringTag_sym(), js_mkstr(js, "worker_threads", 14));
			return lib;
			}

Owner Packages

No Owners

ant/src/modules/worker_threads.c8755416ca476master

worker_threads.cView Options

Owner Packages

ant/src/modules/worker_threads.c
8755416ca476master

worker_threads.c
View Options