Match Node's turn order for Promise.all, Promise.race and async generator returns (#942)

This commit is contained in:
Chris Tate
2026-10-13 16:55:00 -05:00
committed by GitHub
17 changed files with 909 additions and 110 deletions
@@ -859,6 +859,8 @@ export class LlEmitter {
* generator bodies): yieldExpr emission reads them, and emitTryCatch's
* catch prologue emits the GENRET sentinel re-unwind exactly here. */
currentGenerator: { yieldT: IrType; nextT: IrType } | null = null;
/** The function being emitted is an async generator body. */
private currentAsyncGenerator = false;
/** Active optional-chain bind slots, by chain id (chainRecv reads). */
readonly chainSlots = new Map<string, LlValue>();
logArgSlots = 0;
@@ -5464,6 +5466,7 @@ export class LlEmitter {
const int32Abi = this.int32Abis.get(fn.name);
this.currentInt32Result = int32Abi?.result === true;
this.currentGenerator = fn.generator ?? null;
this.currentAsyncGenerator = fn.async === true && fn.generator !== undefined;
this.currentBorrowedReturn = this.borrowedReturns.has(fn.name);
this.inConstructor = this.constructorFunctions.has(fn.name);
this.currentCoro = null;
@@ -5654,6 +5657,8 @@ export class LlEmitter {
// whose unwind released everything down to depth 0).
if (fn.returnType.kind === "void" && !B.isTerminated()) {
this.releaseScope(this.scopes[0]!);
// Completing by falling off the end has nothing to await.
if (this.currentAsyncGenerator) this.emitAsyncGenSettled();
if (this.currentCoro !== null) {
this.emitWasiFulfill(null);
B.terminate(`br label %${this.currentCoro.finalLabel}`);
@@ -6048,6 +6053,13 @@ export class LlEmitter {
this.tryStack = savedTry;
}
/** An async generator completing without a further turn: its request
* settles as the body returns (scr_async_gen_hop_done). */
private emitAsyncGenSettled(): void {
this.declare(`declare void @scr_async_gen_hop_done()`);
this.B.line(`call void @scr_async_gen_hop_done()`);
}
/** `return x` of a local whose slot is an owned scope entry of the same
* type: the return can take over the local's reference. */
private movableReturn(
@@ -7332,6 +7344,19 @@ export class LlEmitter {
v = this.emitExpr(s.value);
this.moveTemp(v);
}
// An async generator's `return v` awaits v before any crossed
// finally runs. Past finally regions that turn is taken here and
// the request settles at completion; otherwise the completion's
// own turn is the await (nothing runs in between).
const genSettles =
this.currentAsyncGenerator && (s.awaited === true || this.finallyStack.length > 0);
if (this.currentAsyncGenerator && s.awaited !== true && this.finallyStack.length > 0) {
if (this.wasi) this.emitWasiSuspend(null);
else {
this.declare(`declare void @scr_await_hop()`);
B.line(`call void @scr_await_hop()`);
}
}
if (this.finallyStack.length > 0) {
let pretSlot: string | null = null;
if (v !== null && isRefCounted(v.type)) {
@@ -7363,6 +7388,7 @@ export class LlEmitter {
this.releaseForJump(0, 0);
}
if (!B.isTerminated()) {
if (genSettles) this.emitAsyncGenSettled();
if (this.currentCoro !== null) {
this.emitWasiFulfill(v);
B.terminate(`br label %${this.currentCoro.finalLabel}`);
@@ -159,14 +159,14 @@ export function emitIntrinsicExpr(host: LlvmEmitterContext, e: ExprOf<"intrinsic
return out;
}
if (e.name === "promise.race") {
// A fresh result promise + one race_add per entry: settled
// entries settle it immediately (first add wins), pending ones
// park a callback waiter. Entry temps stay frame-owned
// (race_add retains what it keeps).
// A fresh result promise + one race_join per entry: each entry
// settles it from a reaction job (first job wins), queued when the
// entry settles or at the join for a settled entry. Entry temps
// stay frame-owned (race_join retains what it keeps).
if (e.type.kind !== "promise")
throw new InternalCompilerError("llvm emitter bug: promise.race type");
host.declare(`declare ptr @scr_promise_new()`);
host.declare(`declare void @scr_promise_race_add(ptr, ptr, ptr)`);
host.declare(`declare void @scr_promise_race_join(ptr, ptr, ptr)`);
const result = B.tmp();
B.line(`${result} = call ptr @scr_promise_new()`);
const out = host.own({ name: result, type: e.type });
@@ -175,7 +175,7 @@ export function emitIntrinsicExpr(host: LlvmEmitterContext, e: ExprOf<"intrinsic
throw new InternalCompilerError("llvm emitter bug: promise.race entry");
const p = host.emitExpr(entry);
const adapter = host.raceAdapterFor(entry.type.inner, e.type.inner);
B.line(`call void @scr_promise_race_add(ptr ${result}, ptr ${p.name}, ptr @${adapter})`);
B.line(`call void @scr_promise_race_join(ptr ${result}, ptr ${p.name}, ptr @${adapter})`);
}
return out;
}
@@ -1312,6 +1312,9 @@ export function lowerStmt(lowerer: Lowerer, stmt: ts.Statement): IrStmt | IrStmt
`bare 'return' where the declared return type is '${lowerer.fmt(lowerer.ctx.returnType)}' (JS hands the caller undefined, which this representation cannot hold — return a value, or widen the declared return type with undefined)`,
);
}
// An async generator's bare return has no operand to await.
if (lowerer.ctx.isAsync === true && lowerer.ctx.generator)
return { kind: "return", value, awaited: true, loc: locOf(stmt) };
return { kind: "return", value, loc: locOf(stmt) };
}
if (ts.isBreakStatement(stmt) || ts.isContinueStatement(stmt)) {
@@ -2473,6 +2473,9 @@ export class Lowerer {
/** Set while a subclass constructor's super() call notes its base
* constructor edge: that edge does not instantiate the base exactly. */
private superConstructorEdge = false;
/** Set when the async generator `return` being lowered awaited its
* promise operand (returnedPromiseAwait). */
private returnAwaitedPromise = false;
private readonly virtualEdges = new Map<ClassInfo, Set<string>>();
classCanBeConstructed(info: ClassInfo): boolean {
@@ -8602,8 +8605,10 @@ export class Lowerer {
returnedPromiseAwait(e: IrExpr): IrExpr {
if (e.type.kind !== "promise")
throw new InternalCompilerError("returned-promise await of a non-promise");
if (!this.adoptsReturnedPromise())
if (!this.adoptsReturnedPromise()) {
this.returnAwaitedPromise = true;
return { kind: "awaitExpr", value: e, type: e.type.inner, loc: e.loc };
}
return { kind: "awaitExpr", value: e, type: e.type.inner, adopt: true, loc: e.loc };
}
@@ -8614,6 +8619,24 @@ export class Lowerer {
* caller never sees a value, so the return goes out bare. Async
* void-inner returns still resolve a returned promise first. */
lowerReturnStmt(node: ts.Expression, loc: SrcLoc): IrStmt {
if (!(this.ctx.isAsync === true && this.ctx.generator)) return this.lowerReturnOf(node, loc);
// An async generator's `return x` awaits x; for a promise that await
// is the returned-promise await, and the return needs no other turn.
const saved = this.returnAwaitedPromise;
this.returnAwaitedPromise = false;
const s = this.lowerReturnOf(node, loc);
const awaited = this.returnAwaitedPromise;
this.returnAwaitedPromise = saved;
if (!awaited) return s;
if (s.kind === "return") return { ...s, awaited: true };
if (s.kind === "block" && s.body.at(-1)?.kind === "return") {
const last = s.body.at(-1) as IrStmt & { kind: "return" };
return { ...s, body: [...s.body.slice(0, -1), { ...last, awaited: true }] };
}
return s;
}
private lowerReturnOf(node: ts.Expression, loc: SrcLoc): IrStmt {
const expected = this.ctx.returnType;
if (expected.kind === "void") {
let e = this.lowerExpr(node);
+6 -2
View File
@@ -1692,7 +1692,11 @@ export type IrStmt =
labels?: string[];
loc: SrcLoc;
}
| { kind: "return"; value: IrExpr | null; loc: SrcLoc }
/** `awaited`: a return of an async generator that needs no Await of its
* own — its operand's await performed it (`return p` of a promise), or
* it has no operand (`return;`). The request settles as the body
* completes instead of one microtask turn later. */
| { kind: "return"; value: IrExpr | null; awaited?: true; loc: SrcLoc }
/** Field write `obj.f = v` — statement-only, like `assign`/`arraySet`.
* Evaluation order: obj, then value. The old value is released; ownership
* of a refcounted new value MOVES into the object. */
@@ -3310,7 +3314,7 @@ export type IrExpr =
* promise.race: every arg is a PROMISE (the array literal's
* entries, lowered individually — the array never materializes), the
* type is the checker's combined result promise; the backend emits a
* fresh promise plus one scr_promise_race_add per entry with an
* fresh promise plus one scr_promise_race_join per entry with an
* interned per-(entry-inner → result-inner) adapter (raceAdapterFor) —
* same-type entries share the runtime's copy adapter, arm entries wrap
* into the result union, sub-union entries re-tag arm-wise. First
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5040-promise-all-race-turns.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5041-promise-all-turns.js"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5042-async-generator-return-promise.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5043-async-generator-return-forms.ts"
],
"diags": []
}
+199 -85
View File
@@ -157,6 +157,9 @@ static void scr_fiber_stack_free(void *stack) {
enum { SCR_PROM_PENDING = 0, SCR_PROM_FULFILLED = 1, SCR_PROM_REJECTED = 2 };
#define SCR_CB_SYNC SIZE_MAX
#define SCR_CB_BATCH (SIZE_MAX - 1)
typedef struct ScrFiber ScrFiber;
typedef struct { size_t rc; } ScrPromiseIdentity;
@@ -183,18 +186,22 @@ struct ScrPromise {
ScrFiber **waiters; /* waiter_inline until a second fiber parks */
size_t nwaiters, waiters_cap;
ScrFiber *waiter_inline;
/* Combinator callbacks (Promise.race / Promise.all): at settle,
* fulfilled promises run adapt(dst, this) — an emitted adapter
* converting the payload into the destination's inner type — and
* rejections copy raw (reasons are dynamically tagged). Each entry owns
* a reference on its dst. Promise.all entries carry the shared
* countdown state (`all` non-NULL, `all_idx` the entry's INPUT index);
* race entries leave both zeroed and keep the adapt dispatch. */
/* Callback waiters: fulfilled promises run adapt(dst, this) — an
* emitted adapter converting the payload into the destination's inner
* type — and rejections copy raw (reasons are dynamically tagged). Each
* entry owns a reference on its dst. Promise.all entries carry the
* shared countdown state (`all` non-NULL, `all_idx` the entry's INPUT
* index); the others leave both zeroed and keep the adapt dispatch.
* Combinator entries (Promise.all / Promise.race) are reaction jobs:
* `job_at` is the number of waiters parked before the entry subscribed,
* so settlement queues the job in subscription order among the waking
* fibers. Conversions (SCR_CB_SYNC) run inside the settle itself. */
struct ScrPromiseCbWaiter {
void (*adapt)(ScrPromise *dst, ScrPromise *src);
ScrPromise *dst;
struct ScrAllState *all;
size_t all_idx;
size_t job_at;
} *cbs;
size_t ncbs, cbs_cap;
/* Unhandled-rejection tracking: set when rejected, cleared on await. */
@@ -229,7 +236,7 @@ typedef struct ScrAllState {
static void scr_promise_all_state_release(ScrAllState *st) {
if (--st->rc == 0) {
if (st->values) scr_arr_release(st->values);
free(st);
scr_mem_free(st);
}
}
@@ -601,6 +608,14 @@ struct ScrFiber {
/* Async-generator settlement jobs: stackless READY-queue envelopes.
* The retained generator is pumped on the main stack. */
ScrGen *gen_job;
/* Promise.all / Promise.race reaction jobs (cb_job set): stackless
* READY-queue envelopes carrying the settled entry (owned) and its
* callback waiter (which owns its destination and countdown state). A
* Promise.all batch job (cb.job_at == SCR_CB_BATCH) carries the first
* rejected entry, or none. */
bool cb_job;
ScrPromise *cb_src;
struct ScrPromiseCbWaiter cb;
/* Set during the eager prefix of a fiber spawned by an await-call entry
* (scr_async_spawn_await_*): the spawner that awaits this fiber's promise
* as soon as the spawn returns. The first suspension parks it on the
@@ -1371,68 +1386,156 @@ void scr_promise_adapt_copy(ScrPromise *dst, ScrPromise *src) {
scr_promise_settle_from(dst, src);
}
/* One Promise.all entry settling: a fulfillment stores its payload into
/* One Promise.all entry's reaction job: a fulfillment stores its payload into
* the values array at the entry's INPUT index (input order regardless of
* settlement order) and the LAST missing fulfillment fulfills the result
* with the array (void-element all fulfills void); a rejection copies raw
* into the result — the first one in SETTLEMENT order wins through the
* destination's own pending check, and later ones still count as HANDLED
* on their entries (settle_from marks them observed), exactly Node's
* into the result — the first job to run wins through the destination's
* own pending check, and later ones still count as HANDLED on their
* entries (marked at the subscription), exactly Node's
* subscribe-to-everything behavior. */
static void scr_promise_all_fulfill(ScrAllState *st, ScrPromise *result) {
if (st->values) {
scr_promise_fulfill_ref(result, scr_arr_retain(st->values), scr_arr_retain_v,
scr_arr_release_v, st->values->cyc ? scr_arr_trace_v : NULL);
} else {
scr_promise_fulfill_void(result);
}
}
static void scr_promise_all_count(ScrAllState *st, ScrPromise *result, size_t fulfilled) {
st->remaining -= fulfilled;
if (st->remaining == 0 && result->state == SCR_PROM_PENDING) scr_promise_all_fulfill(st, result);
}
static void scr_promise_all_settle(ScrAllState *st, ScrPromise *result, size_t idx,
ScrPromise *src) {
if (src->state == SCR_PROM_FULFILLED) {
if (st->store) st->store(st->values, (double)idx, src);
if (--st->remaining == 0 && result->state == SCR_PROM_PENDING) {
if (st->values) {
scr_promise_fulfill_ref(result, scr_arr_retain(st->values), scr_arr_retain_v,
scr_arr_release_v,
st->values->cyc ? scr_arr_trace_v : NULL);
} else {
scr_promise_fulfill_void(result);
}
}
scr_promise_all_count(st, result, 1);
} else {
scr_promise_settle_from(result, src);
}
}
/* Settle helpers wake waiters into the microtask queue and run the
* combinator callbacks: adapters for fulfillments (they convert the
* payload to the destination's inner type), the raw copy for rejections
* (reasons are dynamically tagged). first-settle-wins rides the
* destination's own pending check. Promise.all entries dispatch through
* their shared countdown state instead of the adapt pair. */
static void scr_promise_settle_wake(ScrPromise *p) {
for (size_t i = 0; i < p->nwaiters; i++) {
static inline void scr_promise_wake_one(ScrFiber *f) {
#ifdef SCR_WORKERS
p->waiters[i]->waiting = NULL;
f->waiting = NULL;
#endif
scr_ready_push(p->waiters[i]);
scr_ready_push(f);
}
/* Runs one callback waiter against its settled source and drops the
* waiter's references (its destination and countdown state). */
static void scr_promise_cb_run(struct ScrPromiseCbWaiter *cb, ScrPromise *src) {
if (cb->all) {
scr_promise_all_settle(cb->all, cb->dst, cb->all_idx, src);
scr_promise_all_state_release(cb->all);
} else if (src->state == SCR_PROM_FULFILLED) {
cb->adapt(cb->dst, src);
} else {
scr_promise_settle_from(cb->dst, src);
}
scr_promise_release(cb->dst);
}
/* Queues a combinator's reaction job for a settled entry. The waiter's
* references move into the envelope, which also retains the entry. */
static void scr_promise_cb_queue(const struct ScrPromiseCbWaiter *cb, ScrPromise *src) {
ScrFiber *job = scr_mem_calloc(sizeof *job);
if (!job) scr_oom();
job->cb_job = true;
job->cb_src = scr_promise_retain(src);
job->cb = *cb;
scr_ready_push(job);
}
/* Runs and frees a reaction job envelope. */
static void scr_promise_job_run(ScrFiber *f) {
ScrPromise *src = f->cb_src;
struct ScrPromiseCbWaiter cb = f->cb;
scr_mem_free(f);
if (cb.job_at != SCR_CB_BATCH) {
scr_promise_cb_run(&cb, src);
} else {
if (src != NULL) scr_promise_settle_from(cb.dst, src);
else scr_promise_all_count(cb.all, cb.dst, cb.all_idx);
scr_promise_all_state_release(cb.all);
scr_promise_release(cb.dst);
}
scr_promise_release(src);
}
static struct ScrPromiseCbWaiter *scr_promise_cb_push(ScrPromise *in) {
if (in->ncbs == in->cbs_cap) {
in->cbs_cap = in->cbs_cap ? in->cbs_cap * 2 : 4;
in->cbs = realloc(in->cbs, in->cbs_cap * sizeof *in->cbs);
if (!in->cbs) scr_oom();
}
return &in->cbs[in->ncbs++];
}
/* A combinator subscribing to an entry: Node's `then` on the entry. The
* entry counts as handled from here on; a settled entry queues its
* reaction job now, a pending one when it settles. */
static void scr_promise_cb_subscribe(ScrPromise *in, struct ScrPromiseCbWaiter cb) {
scr_prom_observe(in);
if (in->state != SCR_PROM_PENDING) {
scr_promise_cb_queue(&cb, in);
} else {
cb.job_at = in->nwaiters;
*scr_promise_cb_push(in) = cb;
}
}
/* Settlement with callback waiters: combinator jobs enter the microtask
* queue interleaved with the waking fibers in subscription order, then
* the conversions run in place (they settle their destinations now, so
* those destinations' waiters queue behind this promise's). */
static void scr_promise_settle_wake_cbs(ScrPromise *p) {
size_t w = 0;
for (size_t i = 0; i < p->ncbs; i++) {
const struct ScrPromiseCbWaiter *cb = &p->cbs[i];
if (cb->job_at == SCR_CB_SYNC) continue;
for (; w < cb->job_at && w < p->nwaiters; w++) scr_promise_wake_one(p->waiters[w]);
scr_promise_cb_queue(cb, p);
}
for (; w < p->nwaiters; w++) scr_promise_wake_one(p->waiters[w]);
p->nwaiters = 0;
for (size_t i = 0; i < p->ncbs; i++) {
if (p->cbs[i].all) {
scr_promise_all_settle(p->cbs[i].all, p->cbs[i].dst, p->cbs[i].all_idx, p);
scr_promise_all_state_release(p->cbs[i].all);
} else if (p->state == SCR_PROM_FULFILLED) {
p->cbs[i].adapt(p->cbs[i].dst, p);
} else {
scr_promise_settle_from(p->cbs[i].dst, p);
}
scr_promise_release(p->cbs[i].dst);
if (p->cbs[i].job_at == SCR_CB_SYNC) scr_promise_cb_run(&p->cbs[i], p);
}
p->ncbs = 0;
}
/* Settle helpers wake waiters into the microtask queue and dispatch the
* callback waiters. */
static void scr_promise_settle_wake(ScrPromise *p) {
if (p->ncbs != 0) {
scr_promise_settle_wake_cbs(p);
} else {
for (size_t i = 0; i < p->nwaiters; i++) scr_promise_wake_one(p->waiters[i]);
p->nwaiters = 0;
}
if (p->state == SCR_PROM_REJECTED) scr_track_rejection(p);
}
/* ── Promise.race ─────────────────────────────────────────────────────
* The compiler emits: a fresh result promise, then one race_add per
* entry. A settled entry settles the result immediately (first add
* wins); pending entries park a callback waiter that fires inside the
* entry's settle. Fibers awaiting the result still wake through the
* microtask queue. Losing entries keep their own settlements (payloads
* are retained, not moved). */
* The compiler emits: a fresh result promise, then one race_join per
* entry. Each entry settles the result from its reaction job (first job
* wins), one microtask turn after the entry settles or, for an entry
* already settled, after the join. Losing entries keep their own
* settlements (payloads are retained, not moved).
*
* race_add is the same callback without the job: the destination is a
* view of the entry at another inner type (dyn boxes, typed reads of
* dyn promises), so it settles inside the entry's own settlement. */
void scr_promise_race_join(ScrPromise *race, ScrPromise *in,
void (*adapt)(ScrPromise *dst, ScrPromise *src)) {
scr_promise_cb_subscribe(
in, (struct ScrPromiseCbWaiter){ adapt, scr_promise_retain(race), NULL, 0, 0 });
}
void scr_promise_race_add(ScrPromise *race, ScrPromise *in,
void (*adapt)(ScrPromise *dst, ScrPromise *src)) {
@@ -1441,16 +1544,8 @@ void scr_promise_race_add(ScrPromise *race, ScrPromise *in,
else scr_promise_settle_from(race, in);
return;
}
if (in->ncbs == in->cbs_cap) {
in->cbs_cap = in->cbs_cap ? in->cbs_cap * 2 : 4;
in->cbs = realloc(in->cbs, in->cbs_cap * sizeof *in->cbs);
if (!in->cbs) scr_oom();
}
in->cbs[in->ncbs].adapt = adapt;
in->cbs[in->ncbs].dst = scr_promise_retain(race);
in->cbs[in->ncbs].all = NULL;
in->cbs[in->ncbs].all_idx = 0;
in->ncbs++;
*scr_promise_cb_push(in) =
(struct ScrPromiseCbWaiter){ adapt, scr_promise_retain(race), NULL, 0, SCR_CB_SYNC };
}
/* ── Promise.all ──────────────────────────────────────────────────────
@@ -1459,12 +1554,14 @@ void scr_promise_race_add(ScrPromise *race, ScrPromise *in,
* the result promise +1. `values` arrives EMPTY with cap >= ps->len (the
* emitted construction passes the length) and is pre-sized here: every
* slot's zero bits are the per-kind placeholder (0.0 / false / NULL),
* legal to overwrite through the store helpers and to release. Entries
* already settled settle inline (a rejection settles the result NOW —
* first rejection in settlement order — and later rejected entries are
* still marked handled); pending entries park a countdown waiter. The
* empty array fulfills immediately (awaiters take the settled-await
* microtask hop, like Node's one-tick resolve). */
* legal to overwrite through the store helpers and to release. Every
* entry subscribes a countdown reaction job (Node's `then` on each entry):
* it runs one microtask turn after the entry settles, or after this call
* for an entry already settled, so the last fulfillment's job fulfills
* the result and the first rejection's job rejects it. Every entry is
* marked handled at the subscription. The empty array fulfills
* immediately (awaiters take the settled-await microtask hop, like Node's
* one-tick resolve). */
ScrPromise *scr_promise_all(ScrArr *ps, ScrArr *values,
void (*store)(ScrArr *a, double i, ScrPromise *src)) {
size_t n = ps->len;
@@ -1475,41 +1572,48 @@ ScrPromise *scr_promise_all(ScrArr *ps, ScrArr *values,
* the array intentionally has no dense allocation. */
scr_arr_set_len(values, (double)n);
}
ScrAllState *st = malloc(sizeof *st);
ScrAllState *st = scr_mem_alloc(sizeof *st);
if (!st) scr_oom();
st->rc = 1; /* the builder's reference, dropped at the end */
st->remaining = n;
st->values = values ? scr_arr_retain(values) : NULL;
st->store = store;
/* The jobs of entries already settled would run back to back with
* nothing in between, so they run as one batch job: fulfillments store
* their values now (the array is unobservable until the result
* fulfills) and count down in the job; the first rejected entry in input
* order rejects the result there. */
size_t fulfilled = 0;
ScrPromise *rejected = NULL;
for (size_t i = 0; i < n; i++) {
ScrPromise *in = (ScrPromise *)scr_arr_get_ref(ps, (double)i); /* +1 */
if (in->state != SCR_PROM_PENDING) {
scr_promise_all_settle(st, result, i, in);
} else {
if (in->ncbs == in->cbs_cap) {
in->cbs_cap = in->cbs_cap ? in->cbs_cap * 2 : 4;
in->cbs = realloc(in->cbs, in->cbs_cap * sizeof *in->cbs);
if (!in->cbs) scr_oom();
}
in->cbs[in->ncbs].adapt = NULL;
in->cbs[in->ncbs].dst = scr_promise_retain(result);
in->cbs[in->ncbs].all = st;
in->cbs[in->ncbs].all_idx = i;
in->ncbs++;
if (in->state == SCR_PROM_PENDING) {
st->rc++;
scr_promise_cb_subscribe(
in, (struct ScrPromiseCbWaiter){ NULL, scr_promise_retain(result), st, i, 0 });
} else {
scr_prom_observe(in);
if (in->state == SCR_PROM_FULFILLED) {
if (st->store) st->store(st->values, (double)i, in);
fulfilled++;
} else if (rejected == NULL) {
rejected = scr_promise_retain(in);
}
}
scr_promise_release(in);
}
/* n == 0, or every entry was already fulfilled inline. */
if (st->remaining == 0 && result->state == SCR_PROM_PENDING) {
if (st->values) {
scr_promise_fulfill_ref(result, scr_arr_retain(st->values), scr_arr_retain_v,
scr_arr_release_v,
st->values->cyc ? scr_arr_trace_v : NULL);
} else {
scr_promise_fulfill_void(result);
}
if (fulfilled > 0 || rejected != NULL) {
ScrFiber *job = scr_mem_calloc(sizeof *job);
if (!job) scr_oom();
st->rc++;
job->cb_job = true;
job->cb_src = rejected; /* the reference moves in */
job->cb = (struct ScrPromiseCbWaiter){ NULL, scr_promise_retain(result), st, fulfilled,
SCR_CB_BATCH };
scr_ready_push(job);
}
/* The empty array. */
if (n == 0) scr_promise_all_fulfill(st, result);
scr_promise_all_state_release(st);
return result;
}
@@ -3073,6 +3177,11 @@ static void scr_resume_fiber(ScrFiber *f) {
scr_closure_release(cb);
return;
}
/* A Promise.all / Promise.race reaction job, also on the main stack. */
if (f->cb_job) {
scr_promise_job_run(f);
return;
}
if (f->gen != NULL) {
scr_async_gen_resume_ready(f->gen);
return;
@@ -4779,6 +4888,11 @@ static void scr_context_ready_discard(void) {
generator->settle_queued = false;
scr_gen_release(generator);
scr_mem_free(fiber);
} else if (fiber->cb_job) {
scr_promise_release(fiber->cb.dst);
if (fiber->cb.all) scr_promise_all_state_release(fiber->cb.all);
scr_promise_release(fiber->cb_src);
scr_mem_free(fiber);
}
/* Real fibers belong to the context registry, including queued ones. */
}
+4 -7
View File
@@ -914,13 +914,10 @@ ScrPromise *scr_dyn_promise_all(const ScrDyn *entries) {
if (scr_exc_pending()) scr_promise_reject_pending(source);
else scr_promise_fulfill_ref(source, scr_dyn_retain(value), scr_dyn_retain_v, scr_dyn_release_v, scr_dyn_trace_v);
}
/* Each element contributes a reaction job, even an already fulfilled
* plain value. The aggregate preserves input order and Node's hops. */
scr_promise_mark_handled(source);
ScrDyn *reaction = scr_dyn_promise_then(source, NULL, NULL, NULL);
scr_arr_push_ref(promises, scr_promise_retain(reaction->v.promise));
scr_dyn_release(reaction);
scr_promise_release(source);
/* scr_promise_all subscribes a reaction job to each element, even an
* already fulfilled plain value, preserving input order and Node's
* hops. */
scr_arr_push_ref(promises, source);
}
ScrPromise *all = scr_promise_all(promises, values, scr_promise_all_store_ref);
ScrPromise *result = scr_promise_new();
+15 -9
View File
@@ -5656,13 +5656,18 @@ ScrPromise *scr_promise_settled_void(void);
ScrPromise *scr_promise_settled_ref(void *v, void *(*retain)(void *), void (*release)(void *), ScrTraceFn trace);
/* ── Promise.race (compiler-emitted combinator plumbing) ──────────────
* race_add: settle the result from an already-settled entry, or park a
* callback waiter that fires inside the entry's settle — fulfillments run
* `adapt` (an emitted per-type adapter building the result's inner type
* from the entry's payload; scr_promise_adapt_copy is the same-type
* one), rejections copy raw and count as HANDLED on the entry. The
* race_join: subscribe the result to an entry with a reaction job that
* runs one microtask turn after the entry settles (or after the join, for
* an entry already settled) — fulfillments run `adapt` (an emitted
* per-type adapter building the result's inner type from the entry's
* payload; scr_promise_adapt_copy is the same-type one), rejections copy
* raw; the entry counts as HANDLED from the join. race_add is the same
* callback run inside the entry's own settlement (or now, if settled),
* for destinations that view the entry at another inner type. The
* payload accessors return retained/by-value fulfillment payloads for
* the emitted adapters (losing entries keep their settlements). */
void scr_promise_race_join(ScrPromise *race, ScrPromise *in,
void (*adapt)(ScrPromise *dst, ScrPromise *src));
void scr_promise_race_add(ScrPromise *race, ScrPromise *in,
void (*adapt)(ScrPromise *dst, ScrPromise *src));
@@ -5670,10 +5675,11 @@ void scr_promise_race_add(ScrPromise *race, ScrPromise *in,
* BORROWS the Promise<T>[] entries array and the pre-capacity (empty,
* cap >= ps->len) values array — NULL values/store for Promise<void>
* entries, whose result fulfills void — retains what it keeps, and
* returns the result promise +1. Node-exact: values land at their INPUT
* index regardless of settlement order, the first rejection in
* SETTLEMENT order wins (later ones count as handled on their entries),
* and the empty array fulfills immediately. The store helpers are the
* returns the result promise +1. Node-exact: each entry settles the
* result through a reaction job one microtask turn after it settles,
* values land at their INPUT index regardless of settlement order, the
* first rejection's job wins (every entry counts as handled), and the
* empty array fulfills immediately. The store helpers are the
* per-element-kind payload writers the compiler picks from. */
ScrPromise *scr_promise_all(ScrArr *ps, ScrArr *values,
void (*store)(ScrArr *a, double i, ScrPromise *src));
+182
View File
@@ -0,0 +1,182 @@
// Promise.all and Promise.race settle their result from a reaction job on
// each entry, one microtask turn after the entry settles (or after the
// call, for an entry already settled). Settled and pending entries,
// fulfillments and rejections interleave with then-callbacks and
// queueMicrotask in Node's order.
const log: string[] = [];
// A then-chain that logs one step per microtask turn.
function ticker(label: string, steps: number): void {
let i = 0;
const step = (): void => {
log.push(label + i);
i++;
if (i < steps) Promise.resolve().then(step);
};
Promise.resolve().then(step);
}
function later(value: number): Promise<number> {
return new Promise<number>((resolve) => queueMicrotask(() => resolve(value)));
}
function laterReject(message: string): Promise<number> {
return new Promise<number>((_, reject) => queueMicrotask(() => reject(new Error(message))));
}
async function allSettledEntries(): Promise<void> {
ticker("a", 5);
const ps: Promise<number>[] = [Promise.resolve(1), Promise.resolve(2)];
const values = await Promise.all(ps);
log.push("all settled " + values.join(","));
}
async function allPendingEntries(): Promise<void> {
ticker("b", 6);
const ps: Promise<number>[] = [later(1), Promise.resolve(2), later(3)];
const values = await Promise.all(ps);
log.push("all pending " + values.join(","));
}
async function allRejected(): Promise<void> {
ticker("c", 6);
const ps: Promise<number>[] = [later(1), laterReject("no"), Promise.resolve(3)];
try {
await Promise.all(ps);
} catch (e) {
log.push("all rejected " + (e as Error).message);
}
}
async function allSettledRejection(): Promise<void> {
ticker("l", 5);
const first: Promise<number> = Promise.reject(new Error("first"));
const second: Promise<number> = Promise.reject(new Error("second"));
const ps: Promise<number>[] = [Promise.resolve(1), first, later(2), second];
const all = Promise.all(ps);
all.then(
() => log.push("not reached"),
(e) => log.push("then rejected " + (e as Error).message),
);
try {
await all;
} catch (e) {
log.push("all settled rejection " + (e as Error).message);
}
}
async function allVoid(): Promise<void> {
ticker("d", 5);
const ps: Promise<void>[] = [Promise.resolve(), new Promise<void>((r) => queueMicrotask(r))];
await Promise.all(ps);
log.push("all void");
}
async function allEmpty(): Promise<void> {
ticker("e", 4);
const ps: Promise<number>[] = [];
const values = await Promise.all(ps);
log.push("all empty " + values.length);
}
async function raceSettled(): Promise<void> {
ticker("f", 5);
const winner = await Promise.race([Promise.resolve(1), Promise.resolve(2)]);
log.push("race settled " + winner);
}
async function racePending(): Promise<void> {
ticker("g", 6);
const winner = await Promise.race([later(1), later(2)]);
log.push("race pending " + winner);
}
async function raceRejected(): Promise<void> {
ticker("h", 6);
try {
await Promise.race([laterReject("lost"), later(2)]);
} catch (e) {
log.push("race rejected " + (e as Error).message);
}
}
async function raceSettledRejection(): Promise<void> {
ticker("i", 5);
const rejected: Promise<number> = Promise.reject(new Error("first"));
try {
await Promise.race([rejected, Promise.resolve(2)]);
} catch (e) {
log.push("race settled rejection " + (e as Error).message);
}
}
// Reactions on one entry run in subscription order, whether they come
// from then, await, or the combinator.
async function sharedEntry(): Promise<void> {
ticker("j", 6);
const shared = later(7);
shared.then((v) => {
log.push("then before " + v);
queueMicrotask(() => log.push("micro before"));
});
const ps: Promise<number>[] = [shared];
const all = Promise.all(ps);
const race = Promise.race([shared]);
shared.then((v) => {
log.push("then after " + v);
queueMicrotask(() => log.push("micro after"));
});
all.then((vs) => log.push("all then " + vs.join(",")));
race.then((v) => log.push("race then " + v));
const [first] = await all;
log.push("shared all " + first + " race " + (await race));
}
async function nested(): Promise<void> {
ticker("k", 8);
const inner: Promise<number[]>[] = [
Promise.all([Promise.resolve(1), later(2)] as Promise<number>[]),
Promise.all([later(3)] as Promise<number>[]),
];
const outer = await Promise.all(inner);
log.push("nested " + outer.map((vs) => vs.join("+")).join(","));
const winner = await Promise.race([Promise.race([later(4)]), later(5)]);
log.push("nested race " + winner);
}
async function main(): Promise<void> {
const cases: (() => Promise<void>)[] = [
allSettledEntries,
allPendingEntries,
allRejected,
allSettledRejection,
allVoid,
allEmpty,
raceSettled,
racePending,
raceRejected,
raceSettledRejection,
sharedEntry,
nested,
];
for (const run of cases) {
log.length = 0;
await run();
// Let the tickers finish before printing the case.
await new Promise<void>((resolve) => setTimeout(resolve, 0));
console.log(log.join(" "));
}
// Concurrent: two combinators racing each other's turns.
log.length = 0;
const left = Promise.all([later(1), Promise.resolve(2)] as Promise<number>[]).then((vs) =>
log.push("left " + vs.join(",")),
);
const right = Promise.race([later(3), Promise.resolve(4)]).then((v) => log.push("right " + v));
queueMicrotask(() => log.push("micro 1"));
Promise.resolve().then(() => queueMicrotask(() => log.push("micro 2")));
await left;
await right;
console.log(log.join(" "));
}
main();
+54
View File
@@ -0,0 +1,54 @@
// Promise.all over checked dynamic values settles one microtask turn
// after its last entry's reaction job, for promises, plain values, and
// rejections alike, interleaved with then-callbacks and queueMicrotask.
const log = [];
function ticker(label, steps) {
let i = 0;
const step = () => {
log.push(label + i);
i++;
if (i < steps) Promise.resolve().then(step);
};
Promise.resolve().then(step);
}
function later(value) {
return new Promise((resolve) => queueMicrotask(() => resolve(value)));
}
async function settle() {
await new Promise((resolve) => setTimeout(resolve, 0));
console.log(log.join(" "));
log.length = 0;
}
ticker("a", 5);
const plain = await Promise.all([1, "two", Promise.resolve(3)]);
log.push("plain " + plain.join(","));
await settle();
ticker("b", 6);
const pending = await Promise.all([later(1), 2, later(3)]);
log.push("pending " + pending.join(","));
await settle();
ticker("c", 6);
try {
await Promise.all([later(1), Promise.reject(new Error("no")), 3]);
} catch (e) {
log.push("rejected " + e.message);
}
await settle();
const shared = later("s");
shared.then((v) => log.push("then before " + v));
const all = Promise.all([shared, "x"]);
shared.then((v) => {
log.push("then after " + v);
queueMicrotask(() => log.push("micro after"));
});
all.then((vs) => log.push("all " + vs.join(",")));
queueMicrotask(() => log.push("micro"));
await all;
await settle();
@@ -0,0 +1,212 @@
// An async generator's `return x` awaits x where the return statement runs,
// before any enclosing finally blocks, and the generator's request settles
// as the body completes. A returned promise therefore takes the same turns
// as a returned value (one turn once it settles), and finally blocks see
// the turn of that await, as in Node.
const log: string[] = [];
function ticker(label: string, steps: number): void {
let i = 0;
const step = (): void => {
log.push(label + i);
i++;
if (i < steps) Promise.resolve().then(step);
};
Promise.resolve().then(step);
}
function later(value: number): Promise<number> {
return new Promise<number>((resolve) => queueMicrotask(() => resolve(value)));
}
function laterReject(message: string): Promise<number> {
return new Promise<number>((_, reject) => queueMicrotask(() => reject(new Error(message))));
}
async function* returnValue(): AsyncGenerator<number, number> {
yield 1;
return 2;
}
async function* returnSettled(): AsyncGenerator<number, number> {
yield 1;
return Promise.resolve(2);
}
async function* returnPending(): AsyncGenerator<number, number> {
yield 1;
return later(2);
}
async function* returnFirst(): AsyncGenerator<number, number> {
return Promise.resolve(5);
}
async function* returnAfterAwait(): AsyncGenerator<number, number> {
yield 1;
const base = await later(10);
return later(base + 1);
}
async function* returnValueFinally(): AsyncGenerator<number, number> {
yield 1;
try {
return 2;
} finally {
log.push("finally");
}
}
async function* returnPromiseFinally(): AsyncGenerator<number, number> {
yield 1;
try {
return Promise.resolve(2);
} finally {
log.push("finally");
}
}
async function* returnPromiseNestedFinally(): AsyncGenerator<number, number> {
yield 1;
try {
try {
return later(2);
} finally {
log.push("inner finally");
}
} finally {
log.push("outer finally");
}
}
async function* returnPromiseFinallyAwait(): AsyncGenerator<number, number> {
yield 1;
try {
return later(2);
} finally {
log.push("finally");
await null;
log.push("finally after await");
}
}
async function* returnPromiseFinallyYield(): AsyncGenerator<number, number> {
yield 1;
try {
return Promise.resolve(2);
} finally {
log.push("finally");
yield 9;
log.push("after finally yield");
}
}
async function* returnPromiseFinallyOverride(): AsyncGenerator<number, number> {
yield 1;
try {
return later(2);
} finally {
log.push("finally");
return later(3);
}
}
async function* returnRejectedCaught(): AsyncGenerator<number, number> {
yield 1;
try {
return laterReject("caught");
} catch (e) {
log.push("catch " + (e as Error).message);
return Promise.resolve(3);
}
}
async function* returnRejected(): AsyncGenerator<number, number> {
yield 1;
try {
return laterReject("escaped");
} finally {
log.push("finally");
}
}
async function* returnVoidPromise(): AsyncGenerator<number, void> {
yield 1;
return Promise.resolve();
}
async function* fallOff(): AsyncGenerator<number, void> {
yield 1;
log.push("end");
}
async function settle(name: string): Promise<void> {
await new Promise<void>((resolve) => setTimeout(resolve, 0));
console.log(name + ": " + log.join(" "));
log.length = 0;
}
async function drive(name: string, it: AsyncGenerator<number, number>): Promise<void> {
ticker("t", 8);
try {
for (let k = 0; k < 4; k++) {
const r = await it.next();
log.push("next " + r.value + " " + r.done);
if (r.done) break;
}
} catch (e) {
log.push("rejected " + (e as Error).message);
}
await settle(name);
}
async function driveVoid(name: string, it: AsyncGenerator<number, void>): Promise<void> {
ticker("t", 6);
for (let k = 0; k < 4; k++) {
const r = await it.next();
log.push("next " + r.value + " " + r.done);
if (r.done) break;
}
await settle(name);
}
async function main(): Promise<void> {
await drive("value", returnValue());
await drive("settled", returnSettled());
await drive("pending", returnPending());
await drive("first", returnFirst());
await drive("after await", returnAfterAwait());
await drive("value finally", returnValueFinally());
await drive("promise finally", returnPromiseFinally());
await drive("nested finally", returnPromiseNestedFinally());
await drive("finally await", returnPromiseFinallyAwait());
await drive("finally yield", returnPromiseFinallyYield());
await drive("finally override", returnPromiseFinallyOverride());
await drive("rejected caught", returnRejectedCaught());
await drive("rejected", returnRejected());
await driveVoid("void promise", returnVoidPromise());
await driveVoid("fall off", fallOff());
// Requests queued before the body runs settle in order, with the
// returned promise's turn between them.
const it = returnSettled();
ticker("q", 6);
const a = it.next();
const b = it.next();
const c = it.next();
a.then((r) => log.push("a " + r.value + " " + r.done));
b.then((r) => log.push("b " + r.value + " " + r.done));
c.then((r) => log.push("c " + r.done));
await c;
await settle("queued");
// return(v) from outside resumes the generator's finally.
const outside = returnPromiseFinally();
await outside.next();
ticker("r", 6);
const done = await outside.return(7);
log.push("return " + done.value + " " + done.done);
await settle("return()");
}
main();
@@ -0,0 +1,151 @@
// The other ways an async generator completes, against then-callbacks:
// a bare `return;` and falling off the end settle as the body completes,
// `return undefined` awaits its operand, and methods, generators driving
// other generators, and for-await loops see the same turns.
const log: string[] = [];
function ticker(label: string, steps: number): void {
let i = 0;
const step = (): void => {
log.push(label + i);
i++;
if (i < steps) Promise.resolve().then(step);
};
Promise.resolve().then(step);
}
function later(value: number): Promise<number> {
return new Promise<number>((resolve) => queueMicrotask(() => resolve(value)));
}
async function settle(name: string): Promise<void> {
await new Promise<void>((resolve) => setTimeout(resolve, 0));
console.log(name + ": " + log.join(" "));
log.length = 0;
}
async function* bareReturn(stop: boolean): AsyncGenerator<number, void> {
yield 1;
if (stop) return;
log.push("not stopped");
}
async function* bareReturnFinally(): AsyncGenerator<number, void> {
yield 1;
try {
return;
} finally {
log.push("finally");
}
}
async function* returnUndefined(): AsyncGenerator<number, void> {
yield 1;
return undefined;
}
async function* returnUndefinedFinally(): AsyncGenerator<number, void> {
yield 1;
try {
return undefined;
} finally {
log.push("finally");
}
}
async function* returnCaughtValue(): AsyncGenerator<number, number> {
yield 1;
try {
return 2;
} catch {
return 0;
}
}
async function* inner(): AsyncGenerator<number, number> {
yield 1;
return later(20);
}
async function* outer(): AsyncGenerator<number, number> {
const it = inner();
let r = await it.next();
while (r.done !== true) {
yield r.value;
r = await it.next();
}
log.push("inner returned " + r.value);
return Promise.resolve(r.value + 1);
}
class Source {
readonly base: number;
constructor(base: number) {
this.base = base;
}
async *values(): AsyncGenerator<number, number> {
yield this.base;
try {
return later(this.base + 1);
} finally {
log.push("method finally");
}
}
}
async function* counted(n: number): AsyncGenerator<number, number> {
for (let i = 0; i < n; i++) yield i;
return Promise.resolve(n);
}
async function drive(name: string, it: AsyncGenerator<number, number>): Promise<void> {
ticker("t", 8);
for (let k = 0; k < 4; k++) {
const r = await it.next();
log.push("next " + r.value + " " + r.done);
if (r.done) break;
}
await settle(name);
}
async function driveVoid(name: string, it: AsyncGenerator<number, void>): Promise<void> {
ticker("t", 6);
for (let k = 0; k < 4; k++) {
const r = await it.next();
log.push("next " + r.value + " " + r.done);
if (r.done) break;
}
await settle(name);
}
async function main(): Promise<void> {
await driveVoid("bare return", bareReturn(true));
await driveVoid("no return", bareReturn(false));
await driveVoid("bare return finally", bareReturnFinally());
await driveVoid("return undefined", returnUndefined());
await driveVoid("return undefined finally", returnUndefinedFinally());
await drive("return caught value", returnCaughtValue());
await drive("driving", outer());
await drive("method", new Source(4).values());
ticker("t", 10);
let sum = 0;
for await (const v of counted(3)) {
log.push("got " + v);
sum += v;
}
log.push("for await " + sum);
await settle("for await");
ticker("t", 8);
for await (const v of new Source(7).values()) {
log.push("got " + v);
break;
}
log.push("after break");
await settle("for await break");
}
main();
+3
View File
@@ -201,6 +201,9 @@ describe.skipIf(!zigOnPath())("wasm32-wasi differential", () => {
"2015-generators-yieldstar.ts",
"2016-generators-rc-stress.ts",
"2017-generators-async.ts",
"5040-promise-all-race-turns.ts",
"5042-async-generator-return-promise.ts",
"5043-async-generator-return-forms.ts",
"2018-generators-uncaught.ts",
"2019-generators-loops.ts",
"2093-timers-promises.ts",