int next_timer_id; /* for setTimeout() */
/* not used in the main thread */
JSWorkerMessagePipe *recv_pipe, *send_pipe;
+ /* used by Atomics.notify to wake up other JSRuntimes for
+ Atomics.waitAsync() */
+ JSWaker atomics_async_waker;
#if !defined(_WIN32)
struct pollfd *poll_fds;
int poll_fds_size;
/* XXX: handle signals if useful */
+ delay = JS_AtomicsPollWaiters(ctx);
+
if (list_empty(&ts->os_rw_handlers) && list_empty(&ts->os_timers) &&
- list_empty(&ts->port_list)) {
+ list_empty(&ts->port_list) && delay < 0) {
return -1; /* no more events */
}
+ min_delay = 10000;
+ if (delay >= 0 && delay < min_delay)
+ min_delay = delay;
+
if (!list_empty(&ts->os_timers)) {
cur_time = get_time_ms();
- min_delay = 10000;
list_for_each(el, &ts->os_timers) {
JSOSTimer *th = list_entry(el, JSOSTimer, link);
delay = th->timeout - cur_time;
min_delay = delay;
}
}
- } else {
- min_delay = -1;
}
count = 0;
list_for_each(el, &ts->os_rw_handlers) {
rh = list_entry(el, JSOSRWHandler, link);
if (rh->fd == 0 && !JS_IsNull(rh->rw_func[0])) {
- handles[count++] = (HANDLE)_get_osfhandle(rh->fd); // stdin
if (count == (int)countof(handles))
break;
+ handles[count++] = (HANDLE)_get_osfhandle(rh->fd); // stdin
}
}
JSWorkerMessageHandler *port = list_entry(el, JSWorkerMessageHandler, link);
if (JS_IsNull(port->on_message_func))
continue;
- handles[count++] = port->recv_pipe->waker.handle;
if (count == (int)countof(handles))
break;
+ handles[count++] = port->recv_pipe->waker.handle;
}
+ if (count < (int)countof(handles)) {
+ handles[count++] = ts->atomics_async_waker.handle;
+ }
+
if (count > 0) {
DWORD ret, timeout = INFINITE;
if (min_delay != -1)
}
}
}
+
+ if (ts->atomics_async_waker.handle == handles[ret]) {
+ js_waker_clear(&ts->atomics_async_waker);
+ }
}
} else {
Sleep(min_delay);
}
}
+ delay = JS_AtomicsPollWaiters(ctx);
+
if (list_empty(&ts->os_rw_handlers) && list_empty(&ts->os_timers) &&
- list_empty(&ts->port_list))
+ list_empty(&ts->port_list) && delay < 0)
return -1; /* no more events */
+ min_delay = 10000;
+ if (delay >= 0 && delay < min_delay)
+ min_delay = delay;
+
if (!list_empty(&ts->os_timers)) {
cur_time = get_time_ms();
- min_delay = 10000;
list_for_each(el, &ts->os_timers) {
JSOSTimer *th = list_entry(el, JSOSTimer, link);
delay = th->timeout - cur_time;
min_delay = delay;
}
}
- } else {
- min_delay = -1; /* infinite */
}
-
+
nfds = 0;
list_for_each(el, &ts->os_rw_handlers) {
int events;
}
}
+ if (js_poll_add_poll_fd(ts, &nfds, ts->atomics_async_waker.read_fd, POLLIN))
+ return -1;
+
nfds = poll(ts->poll_fds, nfds, min_delay);
if (nfds > 0) {
list_for_each(el, &ts->os_rw_handlers) {
}
}
}
+ if (ts->poll_fds[ts->atomics_async_waker.read_fd].revents != 0) {
+ js_waker_clear(&ts->atomics_async_waker);
+ }
}
done:
return 0;
JS_FreeValue(ctx, global_obj);
}
+static void js_std_atomics_waker(JSRuntime *rt, void *opaque)
+{
+ JSThreadState *ts = JS_GetRuntimeOpaque(rt);
+ if (ts) {
+ js_waker_signal(&ts->atomics_async_waker);
+ }
+}
+
void js_std_init_handlers(JSRuntime *rt)
{
JSThreadState *ts;
init_list_head(&ts->rejected_promise_list);
ts->next_timer_id = 1;
+ if (js_waker_init(&ts->atomics_async_waker)) {
+ fprintf(stderr, "Could not create atomics_async_waker\n");
+ exit(1);
+ }
+ /* Note: could be done once */
+ JS_AtomicsSetAsyncWaker(js_std_atomics_waker, NULL);
+
JS_SetRuntimeOpaque(rt, ts);
#ifdef USE_WORKER
}
#endif
+ js_waker_close(&ts->atomics_async_waker);
+
#if !defined(_WIN32)
free(ts->poll_fds);
#endif
static void js_iterator_zip_finalizer(JSRuntime *rt, JSValue val);
static void js_iterator_zip_mark(JSRuntime *rt, JSValueConst val,
JS_MarkFunc *mark_func);
+static void js_atomics_free_waiters(JSRuntime *rt);
static void js_array_from_async_finalizer(JSRuntime *rt, JSValue val);
static void js_array_from_async_mark(JSRuntime *rt, JSValueConst val,
JS_MarkFunc *mark_func);
js_free_rt(rt, e);
}
init_list_head(&rt->job_list);
+ js_atomics_free_waiters(rt);
/* don't remove the weak objects to avoid create new jobs with
FinalizationRegistry */
BOOL linked;
pthread_cond_t cond;
int32_t *ptr;
+ /* asynchronous waiters (Atomics.waitAsync) */
+ BOOL is_async;
+ JSRuntime *owner_rt;
+ JSValue resolving_funcs[2];
+ int64_t timeout_time; /* ms since the epoch, INT64_MAX if none */
+ int async_state; /* 0: waiting, 1: notified, 2: timed out */
} JSAtomicsWaiter;
static pthread_mutex_t js_atomics_mutex = PTHREAD_MUTEX_INITIALIZER;
static struct list_head js_atomics_waiter_list =
LIST_HEAD_INIT(js_atomics_waiter_list);
+static JS_AtomicsAsyncWaker *js_atomics_async_waker_cb;
+static void *js_atomics_async_waker_opaque;
#if defined(__aarch64__)
static inline void cpu_pause(void)
return JS_UNDEFINED;
}
+static int js_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex,
+ int64_t timeout_ms)
+{
+#if defined(__APPLE__)
+ struct timespec ts;
+ ts.tv_sec = timeout_ms / 1000;
+ ts.tv_nsec = (timeout_ms % 1000) * 1000000;
+ return pthread_cond_timedwait_relative_np(cond, mutex, &ts);
+#else
+ struct timespec ts;
+ clock_gettime(CLOCK_MONOTONIC, &ts);
+ ts.tv_sec += timeout_ms / 1000;
+ ts.tv_nsec += (timeout_ms % 1000) * 1000000;
+ if (ts.tv_nsec >= 1000000000) {
+ ts.tv_nsec -= 1000000000;
+ ts.tv_sec++;
+ }
+ return pthread_cond_timedwait(cond, mutex, &ts);
+#endif
+}
+
static JSValue js_atomics_wait(JSContext *ctx,
JSValueConst this_obj,
int argc, JSValueConst *argv)
uint64_t idx;
void *ptr;
int64_t timeout;
- struct timespec ts;
JSAtomicsWaiter waiter_s, *waiter;
int ret, size_log2, res;
double d;
waiter = &waiter_s;
waiter->ptr = ptr;
+#if defined(__APPLE__)
pthread_cond_init(&waiter->cond, NULL);
+#else
+ {
+ pthread_condattr_t attr;
+ pthread_condattr_init(&attr);
+ pthread_condattr_setclock(&attr, CLOCK_MONOTONIC);
+ pthread_cond_init(&waiter->cond, &attr);
+ pthread_condattr_destroy(&attr);
+ }
+#endif
waiter->linked = TRUE;
+ waiter->is_async = FALSE;
+ waiter->async_state = 0;
list_add_tail(&waiter->link, &js_atomics_waiter_list);
if (timeout == INT64_MAX) {
pthread_cond_wait(&waiter->cond, &js_atomics_mutex);
ret = 0;
} else {
- /* XXX: use clock monotonic */
- clock_gettime(CLOCK_REALTIME, &ts);
- ts.tv_sec += timeout / 1000;
- ts.tv_nsec += (timeout % 1000) * 1000000;
- if (ts.tv_nsec >= 1000000000) {
- ts.tv_nsec -= 1000000000;
- ts.tv_sec++;
- }
- ret = pthread_cond_timedwait(&waiter->cond, &js_atomics_mutex,
- &ts);
+ ret = js_cond_timedwait(&waiter->cond, &js_atomics_mutex, timeout);
}
if (waiter->linked)
list_del(&waiter->link);
}
}
+static int64_t js_atomics_get_time_ms(void)
+{
+ struct timespec ts;
+ clock_gettime(CLOCK_MONOTONIC, &ts);
+ return (int64_t)ts.tv_sec * 1000 + ts.tv_nsec / 1000000;
+}
+
+/* create the { async, value } result object of Atomics.waitAsync() */
+static JSValue js_atomics_wait_result(JSContext *ctx,
+ BOOL async, JSValue value)
+{
+ JSValue result_obj;
+ result_obj = JS_NewObject(ctx);
+ if (JS_DefinePropertyValue(ctx, result_obj, JS_ATOM_async,
+ JS_NewBool(ctx, async),
+ JS_PROP_C_W_E) < 0 ||
+ JS_DefinePropertyValue(ctx, result_obj, JS_ATOM_value,
+ value, JS_PROP_C_W_E) < 0) {
+ JS_FreeValue(ctx, result_obj);
+ JS_FreeValue(ctx, value);
+ return JS_EXCEPTION;
+ }
+ return result_obj;
+}
+
+/* Atomics.waitAsync(): DoWait(async, ...) */
+static JSValue js_atomics_waitAsync(JSContext *ctx,
+ JSValueConst this_obj,
+ int argc, JSValueConst *argv)
+{
+ JSObject *p;
+ JSValueConst ta, idx_val, expected_val, timeout_val;
+ JSValue result_obj, promise, resolving_funcs[2];
+ JSAtomicsWaiter *waiter;
+ uint64_t idx;
+ void *ptr;
+ int size_log2, res;
+ int32_t v32;
+ int64_t v, timeout, now;
+ double d;
+
+ ta = argc > 0 ? argv[0] : JS_UNDEFINED;
+ idx_val = argc > 1 ? argv[1] : JS_UNDEFINED;
+ expected_val = argc > 2 ? argv[2] : JS_UNDEFINED;
+ timeout_val = argc > 3 ? argv[3] : JS_UNDEFINED;
+
+ p = js_atomics_get_buf(ctx, ta, idx_val, &idx, 2);
+ if (!p)
+ return JS_EXCEPTION;
+ size_log2 = typed_array_size_log2(p->class_id);
+ ptr = p->u.array.u.uint8_ptr + ((uintptr_t)idx << size_log2);
+
+ if (size_log2 == 3) {
+ if (JS_ToBigInt64(ctx, &v, expected_val))
+ return JS_EXCEPTION;
+ } else {
+ if (JS_ToInt32(ctx, &v32, expected_val))
+ return JS_EXCEPTION;
+ v = v32;
+ }
+ if (JS_ToFloat64(ctx, &d, timeout_val))
+ return JS_EXCEPTION;
+ if (isnan(d) || d >= 0x1p63)
+ timeout = INT64_MAX;
+ else if (d < 0)
+ timeout = 0;
+ else
+ timeout = (int64_t)d;
+
+
+ waiter = js_mallocz(ctx, sizeof(*waiter));
+ if (!waiter)
+ return JS_EXCEPTION;
+
+ promise = JS_NewPromiseCapability(ctx, resolving_funcs);
+ if (JS_IsException(promise)) {
+ js_free(ctx, waiter);
+ return JS_EXCEPTION;
+ }
+
+ result_obj = js_atomics_wait_result(ctx, TRUE, promise);
+ if (JS_IsException(result_obj)) {
+ JS_FreeValue(ctx, resolving_funcs[0]);
+ JS_FreeValue(ctx, resolving_funcs[1]);
+ js_free(ctx, waiter);
+ return JS_EXCEPTION;
+ }
+
+ pthread_mutex_lock(&js_atomics_mutex);
+ if (size_log2 == 3) {
+ res = *(int64_t *)ptr != v;
+ } else {
+ res = *(int32_t *)ptr != v;
+ }
+ if (res || timeout == 0) {
+ pthread_mutex_unlock(&js_atomics_mutex);
+ JS_FreeValue(ctx, result_obj);
+ JS_FreeValue(ctx, resolving_funcs[0]);
+ JS_FreeValue(ctx, resolving_funcs[1]);
+ js_free(ctx, waiter);
+ return js_atomics_wait_result(ctx, FALSE,
+ JS_AtomToString(ctx, res ?
+ JS_ATOM_not_equal :
+ JS_ATOM_timed_out));
+ }
+ now = js_atomics_get_time_ms();
+ waiter->ptr = ptr;
+ waiter->linked = TRUE;
+ waiter->is_async = TRUE;
+ waiter->owner_rt = ctx->rt;
+ waiter->resolving_funcs[0] = resolving_funcs[0];
+ waiter->resolving_funcs[1] = resolving_funcs[1];
+ waiter->async_state = 0;
+ if (timeout >= INT64_MAX - now)
+ waiter->timeout_time = INT64_MAX;
+ else
+ waiter->timeout_time = now + timeout;
+ list_add_tail(&waiter->link, &js_atomics_waiter_list);
+ pthread_mutex_unlock(&js_atomics_mutex);
+
+ return result_obj;
+}
+
+static void notify_async_waiters(JSContext *ctx, struct list_head *ready_list)
+{
+ struct list_head *el, *el1;
+ JSAtomicsWaiter *waiter;
+ JSValue value, ret;
+
+ list_for_each_safe(el, el1, ready_list) {
+ waiter = list_entry(el, JSAtomicsWaiter, link);
+ list_del(&waiter->link);
+ value = JS_AtomToString(ctx, waiter->async_state == 1 ?
+ JS_ATOM_ok : JS_ATOM_timed_out);
+ if (!JS_IsException(value)) {
+ ret = JS_Call(ctx, waiter->resolving_funcs[0], JS_UNDEFINED, 1,
+ vc(&value));
+ JS_FreeValue(ctx, ret);
+ }
+ JS_FreeValue(ctx, value);
+ JS_FreeValue(ctx, waiter->resolving_funcs[0]);
+ JS_FreeValue(ctx, waiter->resolving_funcs[1]);
+ js_free(ctx, waiter);
+ }
+}
+
+/* Resolve the asynchronous waiters of this runtime which were
+ notified or whose timeout expired. Return the delay in milliseconds
+ before the next call must be done, 0 if promise callbacks were
+ called or -1 if there is no asynchronous waiter left for this
+ runtime. */
+int64_t JS_AtomicsPollWaiters(JSContext *ctx)
+{
+ struct list_head *el, *el1, ready_list;
+ JSAtomicsWaiter *waiter;
+ JSRuntime *rt = ctx->rt;
+ int64_t now, delay = -1, d;
+
+ now = js_atomics_get_time_ms();
+ init_list_head(&ready_list);
+ pthread_mutex_lock(&js_atomics_mutex);
+ list_for_each_safe(el, el1, &js_atomics_waiter_list) {
+ waiter = list_entry(el, JSAtomicsWaiter, link);
+ if (!waiter->is_async || waiter->owner_rt != rt)
+ continue;
+ if (waiter->async_state == 0) {
+ if (waiter->timeout_time != INT64_MAX) {
+ d = waiter->timeout_time - now;
+ } else {
+ d = INT64_MAX;
+ }
+ if (d <= 0) {
+ waiter->async_state = 2;
+ goto make_ready;
+ } else {
+ if (delay < 0 || d < delay)
+ delay = d;
+ }
+ } else {
+ make_ready:
+ list_del(&waiter->link);
+ waiter->linked = FALSE;
+ list_add_tail(&waiter->link, &ready_list);
+ delay = 0;
+ }
+ }
+ pthread_mutex_unlock(&js_atomics_mutex);
+
+ if (!list_empty(&ready_list))
+ notify_async_waiters(ctx, &ready_list);
+
+ return delay;
+}
+
+/* Set the waker function which is used by Atomics.notify() to wake
+ another runtime in the Atomics.waitAsync() case. */
+void JS_AtomicsSetAsyncWaker(JS_AtomicsAsyncWaker *cb, void *opaque)
+{
+ pthread_mutex_lock(&js_atomics_mutex);
+ js_atomics_async_waker_cb = cb;
+ js_atomics_async_waker_opaque = opaque;
+ pthread_mutex_unlock(&js_atomics_mutex);
+}
+
+/* free the asynchronous waiters of a runtime (JS_FreeRuntime) */
+static void js_atomics_free_waiters(JSRuntime *rt)
+{
+ struct list_head *el, *el1;
+ JSAtomicsWaiter *waiter;
+
+ pthread_mutex_lock(&js_atomics_mutex);
+ list_for_each_safe(el, el1, &js_atomics_waiter_list) {
+ waiter = list_entry(el, JSAtomicsWaiter, link);
+ if (!waiter->is_async || waiter->owner_rt != rt)
+ continue;
+ list_del(&waiter->link);
+ waiter->linked = FALSE;
+ JS_FreeValueRT(rt, waiter->resolving_funcs[0]);
+ JS_FreeValueRT(rt, waiter->resolving_funcs[1]);
+ js_free_rt(rt, waiter);
+ }
+ pthread_mutex_unlock(&js_atomics_mutex);
+}
+
static JSValue js_atomics_notify(JSContext *ctx,
JSValueConst this_obj,
int argc, JSValueConst *argv)
{
- struct list_head *el, *el1, waiter_list;
+ struct list_head *el, *el1, waiter_list, async_waiter_list;
int32_t count, n;
uint64_t idx;
int size_log2;
if (abuf->shared && count > 0) {
/* 'argv[0]' is a SharedArrayBuffer so it cannot be detached nor reduced */
ptr = p->u.array.u.uint8_ptr + ((uintptr_t)idx << size_log2);
- pthread_mutex_lock(&js_atomics_mutex);
init_list_head(&waiter_list);
+ init_list_head(&async_waiter_list);
+ pthread_mutex_lock(&js_atomics_mutex);
list_for_each_safe(el, el1, &js_atomics_waiter_list) {
waiter = list_entry(el, JSAtomicsWaiter, link);
if (waiter->ptr == ptr) {
- list_del(&waiter->link);
- waiter->linked = FALSE;
- list_add_tail(&waiter->link, &waiter_list);
- n++;
+ if (waiter->is_async) {
+ /* the waiter stays in the list until its owning
+ thread resolves the promise */
+ if (waiter->async_state != 0)
+ continue;
+ waiter->async_state = 1;
+ if (waiter->owner_rt != ctx->rt) {
+ /* wake the runtime if necessary */
+ if (js_atomics_async_waker_cb) {
+ js_atomics_async_waker_cb(waiter->owner_rt,
+ js_atomics_async_waker_opaque);
+ }
+ } else {
+ list_del(&waiter->link);
+ waiter->linked = FALSE;
+ list_add_tail(&waiter->link, &async_waiter_list);
+ }
+ n++;
+ } else {
+ list_del(&waiter->link);
+ waiter->linked = FALSE;
+ list_add_tail(&waiter->link, &waiter_list);
+ n++;
+ }
if (n >= count)
break;
}
pthread_cond_signal(&waiter->cond);
}
pthread_mutex_unlock(&js_atomics_mutex);
+
+ /* better to resolve the promise outside the mutex */
+ if (!list_empty(&async_waiter_list))
+ notify_async_waiters(ctx, &async_waiter_list);
}
return JS_NewInt32(ctx, n);
}
JS_CFUNC_DEF("isLockFree", 1, js_atomics_isLockFree ),
JS_CFUNC_DEF("pause", 0, js_atomics_pause ),
JS_CFUNC_DEF("wait", 4, js_atomics_wait ),
+ JS_CFUNC_DEF("waitAsync", 4, js_atomics_waitAsync ),
JS_CFUNC_DEF("notify", 3, js_atomics_notify ),
JS_PROP_STRING_DEF("[Symbol.toStringTag]", "Atomics", JS_PROP_CONFIGURABLE ),
};
projects / quickjs.git / commitdiff