mirror of
https://github.com/vercel-labs/scriptc.git
synced 2026-10-29 07:04:09 +08:00
Make short async calls and awaits cheap
This commit is contained in:
@@ -95,6 +95,7 @@ import { RUNTIME_ABI_MARKER } from "../runtime-abi.js";
|
||||
import { computeMayThrow } from "../may-throw.js";
|
||||
import {
|
||||
mangleArgPack,
|
||||
mangleAsyncAwaitCall,
|
||||
mangleAsyncSpawn,
|
||||
mangleBorrowedFunction,
|
||||
mangleClassObj,
|
||||
@@ -659,6 +660,9 @@ export class LlEmitter {
|
||||
/** Method names with at least one may-throw implementation — the
|
||||
* virtualCall pending check's key. */
|
||||
readonly mayThrowMethods = new Set<string>();
|
||||
/** Async functions awaited directly at some `await f(...)` site; each
|
||||
* gets an await-call entry (emitAsyncAwaitCall). */
|
||||
readonly awaitCallees = new Set<string>();
|
||||
/** setTimeout and friends appeared somewhere: main must run the event
|
||||
* loop even in programs with no async functions. */
|
||||
usesTimers = false;
|
||||
@@ -2996,8 +3000,14 @@ export class LlEmitter {
|
||||
const definitions = [`%${pack} = type { ${fieldTys.join(", ") || "i8"} } ; ${fn.name} args`];
|
||||
const sizeOf = `ptrtoint (ptr getelementptr (%${pack}, ptr null, i32 1) to ${this.sizeType})`;
|
||||
|
||||
this.declare(`declare void @free(ptr)`);
|
||||
this.declare(`declare ptr @malloc(${this.sizeType})`);
|
||||
// Async argument packs live from the spawn to the trampoline's first
|
||||
// instructions, so they take the runtime's small-block allocator.
|
||||
// Generator packs keep malloc: the runtime frees never-started ones.
|
||||
const smallPack = fn.generator === undefined;
|
||||
const packAlloc = smallPack ? "scr_rt_calloc" : "malloc";
|
||||
const packFree = smallPack ? "scr_rt_free" : "free";
|
||||
this.declare(`declare void @${packFree}(ptr)`);
|
||||
this.declare(`declare ptr @${packAlloc}(${this.sizeType})`);
|
||||
|
||||
const tr: string[] = [
|
||||
`define internal void @${mangleTrampoline(fn.name)}(ptr %self, ptr %ap) ${FN_ATTRS} {`,
|
||||
@@ -3011,7 +3021,7 @@ export class LlEmitter {
|
||||
);
|
||||
loads.push(`${ty} %a${i}`);
|
||||
});
|
||||
tr.push(` call void @free(ptr %ap)`);
|
||||
tr.push(` call void @${packFree}(ptr %ap)`);
|
||||
const ret = fn.returnType;
|
||||
const retTy = this.llType(ret);
|
||||
const bodyCall = `call ${retTy} @${mangleFunction(fn.name)}(${loads.join(", ")})`;
|
||||
@@ -3022,7 +3032,7 @@ export class LlEmitter {
|
||||
|
||||
const spawnParams = fieldTys.map((ty, i) => `${ty} %a${i}`);
|
||||
const argPackLines = [
|
||||
` %ap = call ptr @malloc(${this.sizeType} ${sizeOf})`,
|
||||
` %ap = call ptr @${packAlloc}(${this.sizeType} ${sizeOf})`,
|
||||
` %isnull = icmp eq ptr %ap, null`,
|
||||
` br i1 %isnull, label %oom, label %ok`,
|
||||
`oom:`,
|
||||
@@ -3094,7 +3104,7 @@ export class LlEmitter {
|
||||
const eligible = new Set<string>();
|
||||
if (this.wasi || this.mod.workers === true) return eligible;
|
||||
const suspends =
|
||||
/@scr_(?:await_|module_await\b|promise_await_settled\b|test_sub\b|gen_yield_|async_gen_|fiber_)/;
|
||||
/@(?:scr_(?:await_|module_await\b|promise_await_settled\b|test_sub\b|gen_yield_|async_gen_|fiber_)|sc_aw_)/;
|
||||
for (const decl of this.decls) if (/@scr_test_sub\b/.test(decl)) return eligible;
|
||||
for (const text of this.fnDefText.values()) if (/@scr_test_sub\b/.test(text)) return eligible;
|
||||
const byMangled = new Map<string, string>();
|
||||
@@ -3180,6 +3190,118 @@ export class LlEmitter {
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Whether `await f(...)` may enter f through its await-call entry. The
|
||||
* entry's result is the awaited value of f's own body type; direct calls
|
||||
* never carry a lifted environment. */
|
||||
awaitCallEligible(fnName: string): boolean {
|
||||
if (this.wasi || this.mod.workers === true) return false;
|
||||
const fn = this.fnByName.get(fnName);
|
||||
if (fn === undefined || fn.async !== true || fn.generator !== undefined) return false;
|
||||
if (fn.captures !== undefined || fn.returnType.kind === "dyn") return false;
|
||||
return fn.asyncCacheGlobal === undefined && fn.asyncCycleCacheGlobal === undefined;
|
||||
}
|
||||
|
||||
/** The await-call entry of an async function (`sc_aw_*`): one direct
|
||||
* `await f(...)` as a call returning the awaited value, so the call site
|
||||
* needs no promise of its own. A fiberless body runs on the caller's
|
||||
* stack as in its spawn wrapper; a clean result then takes the await's
|
||||
* one microtask turn (scr_await_hop) and is returned without ever
|
||||
* materializing the promise, which nothing else could observe. A throw
|
||||
* builds the rejected promise exactly as the spawn wrapper would and
|
||||
* awaits it, so the rejection reaches the awaiter on the same turn. Other
|
||||
* bodies pack their arguments as the spawn wrapper does and enter
|
||||
* scr_async_spawn_await_*, which parks the caller on the promise as soon
|
||||
* as the body first suspends. */
|
||||
private emitAsyncAwaitCall(fn: IrFunction, argPackLines: string[] | null): string[] {
|
||||
const ret = fn.returnType;
|
||||
const retTy = this.llType(ret);
|
||||
const params = fn.params.map((p, i) => `${this.llType(p.type)} %a${i}`);
|
||||
const awaitLine = (dest: string, promise: string): string => {
|
||||
switch (ret.kind) {
|
||||
case "void":
|
||||
this.declare(`declare void @scr_await_void(ptr)`);
|
||||
return ` call void @scr_await_void(ptr ${promise})`;
|
||||
case "f64":
|
||||
case "date":
|
||||
this.declare(`declare double @scr_await_f64(ptr)`);
|
||||
return ` ${dest} = call double @scr_await_f64(ptr ${promise})`;
|
||||
case "bool":
|
||||
this.declare(`declare zeroext i1 @scr_await_bool(ptr)`);
|
||||
return ` ${dest} = call zeroext i1 @scr_await_bool(ptr ${promise})`;
|
||||
case "string":
|
||||
this.declare(`declare ptr @scr_await_str(ptr)`);
|
||||
return ` ${dest} = call ptr @scr_await_str(ptr ${promise})`;
|
||||
default:
|
||||
this.declare(`declare ptr @scr_await_ref(ptr)`);
|
||||
return ` ${dest} = call ptr @scr_await_ref(ptr ${promise})`;
|
||||
}
|
||||
};
|
||||
const retLine = (v: string): string =>
|
||||
retTy === "void" ? ` ret void` : ` ret ${retTy} ${v}`;
|
||||
const head = `define internal ${retTy} @${mangleAsyncAwaitCall(fn.name)}(${params.join(", ")}) ${FN_ATTRS} { ; await ${fn.name}`;
|
||||
const args = params.join(", ");
|
||||
if (argPackLines !== null) {
|
||||
const kind =
|
||||
ret.kind === "void"
|
||||
? "void"
|
||||
: ret.kind === "f64" || ret.kind === "date"
|
||||
? "f64"
|
||||
: ret.kind === "bool"
|
||||
? "bool"
|
||||
: ret.kind === "string"
|
||||
? "str"
|
||||
: "ref";
|
||||
const cret = kind === "bool" ? "zeroext i1" : retTy;
|
||||
this.declare(`declare ${cret} @scr_async_spawn_await_${kind}(ptr, ptr)`);
|
||||
this.needOom();
|
||||
const call = `call ${cret} @scr_async_spawn_await_${kind}(ptr @${mangleTrampoline(fn.name)}, ptr %ap)`;
|
||||
return [
|
||||
head,
|
||||
`entry:`,
|
||||
...argPackLines,
|
||||
kind === "void" ? ` ${call}` : ` %v = ${call}`,
|
||||
retLine("%v"),
|
||||
`}`,
|
||||
``,
|
||||
];
|
||||
}
|
||||
this.declare(`declare void @scr_async_inline_enter(ptr)`);
|
||||
this.declare(`declare void @scr_async_inline_exit(ptr)`);
|
||||
this.declare(`declare void @scr_await_hop()`);
|
||||
this.declare(`declare ptr @scr_promise_new()`);
|
||||
this.declare(`declare void @scr_promise_reject_pending(ptr)`);
|
||||
this.declare(`declare void @scr_promise_release(ptr)`);
|
||||
const bodyCall = `call ${retTy} @${mangleFunction(fn.name)}(${args})`;
|
||||
const out = [
|
||||
head,
|
||||
`entry:`,
|
||||
` %frame = alloca { ptr, ptr }`,
|
||||
` call void @scr_async_inline_enter(ptr %frame)`,
|
||||
retTy === "void" ? ` ${bodyCall}` : ` %r = ${bodyCall}`,
|
||||
...this.pendingTestLines("%pend"),
|
||||
` call void @scr_async_inline_exit(ptr %frame)`,
|
||||
` br i1 %pend, label %thrown, label %clean`,
|
||||
`clean:`,
|
||||
` call void @scr_await_hop()`,
|
||||
retLine("%r"),
|
||||
`thrown:`,
|
||||
];
|
||||
if (ret.kind !== "void" && isRefCounted(ret)) {
|
||||
// An escaping throw means %r is the never-read dummy (NULL).
|
||||
out.push(` call void ${releaseSym(this.shapeHost, ret)}(ptr %r)`);
|
||||
}
|
||||
out.push(
|
||||
` %p = call ptr @scr_promise_new()`,
|
||||
` call void @scr_promise_reject_pending(ptr %p)`,
|
||||
awaitLine("%v", "%p"),
|
||||
` call void @scr_promise_release(ptr %p)`,
|
||||
retLine("%v"),
|
||||
`}`,
|
||||
``,
|
||||
);
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Per-async-function machinery — async.ts's scaffolding, .ll
|
||||
* flavored: an argument-pack struct type, a fiber trampoline (unpacks,
|
||||
* frees the pack, runs the ordinary compiled body, settles the
|
||||
@@ -3209,6 +3331,9 @@ export class LlEmitter {
|
||||
const fiberless = this.fiberlessAsyncFunctions();
|
||||
for (const fn of this.mod.functions) {
|
||||
if (fn.async !== true || fn.generator !== undefined) continue;
|
||||
if (fiberless.has(fn.name) && this.awaitCallees.has(fn.name)) {
|
||||
for (const line of this.emitAsyncAwaitCall(fn, null)) out.push(line);
|
||||
}
|
||||
if (fiberless.has(fn.name)) {
|
||||
const fieldTys = [
|
||||
...(fn.captures !== undefined ? ["ptr"] : []),
|
||||
@@ -3221,6 +3346,9 @@ export class LlEmitter {
|
||||
const { definitions, ret, tr, spawnParams, argPackLines } =
|
||||
this.emitArgPackAndTrampolinePrologue(fn);
|
||||
for (const line of definitions) out.push(line);
|
||||
if (this.awaitCallees.has(fn.name)) {
|
||||
for (const line of this.emitAsyncAwaitCall(fn, argPackLines)) out.push(line);
|
||||
}
|
||||
this.declare(`declare ptr @scr_fiber_promise(ptr)`);
|
||||
this.declare(`declare ptr @scr_async_spawn(ptr, ptr)`);
|
||||
this.needOom();
|
||||
@@ -7443,6 +7571,12 @@ export class LlEmitter {
|
||||
return emitCallExpr(this, e);
|
||||
}
|
||||
|
||||
/** A direct async call an enclosing `await` consumes whole: emitted
|
||||
* through the callee's await-call entry, yielding the awaited value. */
|
||||
emitAwaitedCall(e: ExprOf<"call">): LlValue {
|
||||
return emitCallExpr(this, e, true);
|
||||
}
|
||||
|
||||
emitRecordExpr(
|
||||
e: ExprOf<
|
||||
| "fieldGet"
|
||||
|
||||
@@ -626,6 +626,18 @@ export function emitAsyncExpr(
|
||||
// re-throw here (hence the pending check). Promise temp borrowed;
|
||||
// refcounted results arrive +1 and join the frame pre-check so an
|
||||
// unwind releases the dummy (NULL) harmlessly.
|
||||
// `await f(...)` of a direct async call: the call and the await go
|
||||
// through f's await-call entry, which returns the awaited value
|
||||
// (owned, with the pending check done by the call emission).
|
||||
if (e.value.kind === "call" && host.awaitCallEligible(e.value.callee)) {
|
||||
const callee = host.fnByName.get(e.value.callee)!;
|
||||
if (
|
||||
typeEquals(callee.returnType, e.type) &&
|
||||
!host.callLifetimes.borrowed.has(e.value.callee)
|
||||
) {
|
||||
return host.emitAwaitedCall(e.value);
|
||||
}
|
||||
}
|
||||
const pr = host.emitExpr(e.value);
|
||||
if (host.wasi) host.emitWasiSuspend(pr.name);
|
||||
if (e.type.kind === "void") {
|
||||
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
} from "../../ir/ir.js";
|
||||
import { collectFfiRetainedOps, parseFfiCallbackKey } from "../ffi-callbacks.js";
|
||||
import {
|
||||
mangleAsyncAwaitCall,
|
||||
mangleBorrowedFunction,
|
||||
mangleClassNew,
|
||||
mangleClassRetain,
|
||||
@@ -51,6 +52,7 @@ export function emitCallExpr(
|
||||
| "instanceOf"
|
||||
| "virtualCall"
|
||||
>,
|
||||
awaitEntry = false,
|
||||
): LlValue {
|
||||
const B = host.B;
|
||||
switch (e.kind) {
|
||||
@@ -120,6 +122,24 @@ export function emitCallExpr(
|
||||
const argList = args
|
||||
.map((a, i) => `${host.llType(callee.params[i]!.type)} ${a.name}`)
|
||||
.join(", ");
|
||||
if (awaitEntry) {
|
||||
// The enclosing await routes this call through the callee's
|
||||
// await-call entry: the result is the awaited value, and the
|
||||
// await's rejection re-throw is the pending check.
|
||||
host.awaitCallees.add(e.callee);
|
||||
const awaited = callee.returnType;
|
||||
const entry = `@${mangleAsyncAwaitCall(e.callee)}`;
|
||||
if (awaited.kind === "void") {
|
||||
B.line(`call void ${entry}(${argList})`);
|
||||
host.emitPendingCheck();
|
||||
return { name: "", type: awaited };
|
||||
}
|
||||
const t = B.tmp();
|
||||
B.line(`${t} = call ${host.llType(awaited)} ${entry}(${argList})`);
|
||||
const out = host.own({ name: t, type: awaited });
|
||||
host.emitPendingCheck();
|
||||
return out;
|
||||
}
|
||||
// Async callee: the spawn wrapper runs the body eagerly to its
|
||||
// first suspension and returns the promise (+1). The call itself
|
||||
// never unwinds — rejections surface at await (computeMayThrow's
|
||||
|
||||
@@ -166,6 +166,11 @@ export function mangleGlobal(globalId: string): string {
|
||||
export function mangleAsyncSpawn(fnName: string): string {
|
||||
return `sc_as_${sanitize(fnName)}`;
|
||||
}
|
||||
/** `await f(...)` of a direct async call: the call and its await as one
|
||||
* entry that returns the awaited value. */
|
||||
export function mangleAsyncAwaitCall(fnName: string): string {
|
||||
return `sc_aw_${sanitize(fnName)}`;
|
||||
}
|
||||
export function mangleTrampoline(fnName: string): string {
|
||||
return `sc_tr_${sanitize(fnName)}`;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { afterEach, expect, test } from "vitest";
|
||||
import { compile } from "../src/index.js";
|
||||
|
||||
const directories: string[] = [];
|
||||
afterEach(async () => {
|
||||
await Promise.all(directories.splice(0).map((dir) => rm(dir, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
async function emit(source: string): Promise<string> {
|
||||
const outDir = await mkdtemp(join(tmpdir(), "scriptc-await-call-"));
|
||||
directories.push(outDir);
|
||||
const entry = join(outDir, "main.ts");
|
||||
await writeFile(entry, source);
|
||||
const outPath = join(outDir, "program.ll");
|
||||
const result = await compile(entry, { entry, outDir, outPath, outputKind: "llvm" });
|
||||
expect(result.ok, JSON.stringify(result)).toBe(true);
|
||||
return readFile(outPath, "utf8");
|
||||
}
|
||||
|
||||
function definition(llvm: string, symbol: string): string {
|
||||
const match = new RegExp(`^define internal [^\\n]*@${symbol}\\([^]*?^}`, "m").exec(llvm);
|
||||
expect(match, symbol).not.toBeNull();
|
||||
return match![0];
|
||||
}
|
||||
|
||||
const source = `
|
||||
const cache = new Map<number, number>();
|
||||
async function leaf(x: number): Promise<number> {
|
||||
const hit = cache.get(x);
|
||||
return hit ?? x + 1;
|
||||
}
|
||||
async function middle(x: number): Promise<number> {
|
||||
const y = await leaf(x);
|
||||
return y * 2;
|
||||
}
|
||||
async function main(): Promise<void> {
|
||||
const pending = leaf(1);
|
||||
let sum = await middle(2);
|
||||
sum += await pending;
|
||||
console.log(sum);
|
||||
}
|
||||
void main();
|
||||
`;
|
||||
|
||||
test("awaiting a direct call to a body that never suspends skips the promise", async () => {
|
||||
const llvm = await emit(source);
|
||||
expect(definition(llvm, "sc_f_middle")).toContain("call double @sc_aw_leaf(double");
|
||||
const entry = definition(llvm, "sc_aw_leaf");
|
||||
expect(entry).toContain("call double @sc_f_leaf(double %a0)");
|
||||
expect(entry).toContain("call void @scr_async_inline_exit(ptr %frame)");
|
||||
// The clean result takes the await's microtask turn and returns directly;
|
||||
// only the throw path builds the rejected promise.
|
||||
const clean = entry.slice(entry.indexOf("clean:"), entry.indexOf("thrown:"));
|
||||
expect(clean).toContain("call void @scr_await_hop()");
|
||||
expect(clean).not.toContain("@scr_promise_new");
|
||||
expect(entry.slice(entry.indexOf("thrown:"))).toContain("call void @scr_promise_reject_pending");
|
||||
});
|
||||
|
||||
test("awaiting a direct call to a suspending body spawns and awaits in one call", async () => {
|
||||
const llvm = await emit(source);
|
||||
const main = definition(llvm, "sc_f_main");
|
||||
expect(main).toContain("call double @sc_aw_middle(double");
|
||||
// A call whose promise is kept keeps the promise-returning entry.
|
||||
expect(main).toContain("call ptr @sc_as_leaf(double");
|
||||
expect(definition(llvm, "sc_aw_middle")).toContain(
|
||||
"call double @scr_async_spawn_await_f64(ptr @sc_tr_middle, ptr %ap)",
|
||||
);
|
||||
// middle awaits, so it keeps its fiber.
|
||||
expect(definition(llvm, "sc_as_middle")).toContain("@scr_async_spawn(");
|
||||
});
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"order": [
|
||||
"<repo>/tests/corpus/4880-await-call-turn-order.ts"
|
||||
],
|
||||
"diags": []
|
||||
}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"order": [
|
||||
"<repo>/tests/corpus/4881-await-call-exceptions.ts"
|
||||
],
|
||||
"diags": []
|
||||
}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"order": [
|
||||
"<repo>/tests/corpus/4882-await-call-request-chain.ts"
|
||||
],
|
||||
"diags": []
|
||||
}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"order": [
|
||||
"<repo>/tests/corpus/4883-await-values-and-hops.ts"
|
||||
],
|
||||
"diags": []
|
||||
}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"order": [
|
||||
"<repo>/tests/corpus/4884-await-call-unhandled-sibling.ts"
|
||||
],
|
||||
"diags": []
|
||||
}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"order": [
|
||||
"<repo>/tests/corpus/4885-await-call-checkpoints.ts"
|
||||
],
|
||||
"diags": []
|
||||
}
|
||||
@@ -163,6 +163,8 @@ typedef struct { size_t rc; } ScrPromiseIdentity;
|
||||
|
||||
struct ScrPromise {
|
||||
size_t rc; /* The cycle collector reads the reference count at offset zero. */
|
||||
/* Shared identity cell for promises adopted across inner types (NULL
|
||||
* until scr_promise_share_identity first links this promise). */
|
||||
ScrPromiseIdentity *identity;
|
||||
int state;
|
||||
bool top_level_handled; /* uncaughtException handled entry evaluation */
|
||||
@@ -178,8 +180,9 @@ struct ScrPromise {
|
||||
void (*release_fn)(void *);
|
||||
ScrTraceFn trace_fn;
|
||||
/* Fibers parked on this promise. */
|
||||
ScrFiber **waiters;
|
||||
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
|
||||
@@ -235,6 +238,37 @@ static void scr_oom(void) {
|
||||
abort();
|
||||
}
|
||||
|
||||
static void scr_promise_free_side(ScrPromise *p) {
|
||||
if (p->identity && --p->identity->rc == 0) free(p->identity);
|
||||
if (p->waiters != &p->waiter_inline) free(p->waiters);
|
||||
}
|
||||
|
||||
static void scr_promise_add_waiter(ScrPromise *p, ScrFiber *f) {
|
||||
if (p->nwaiters == p->waiters_cap) {
|
||||
if (p->waiters_cap == 0) {
|
||||
p->waiters = &p->waiter_inline;
|
||||
p->waiters_cap = 1;
|
||||
} else {
|
||||
size_t cap = p->waiters_cap < 4 ? 4 : p->waiters_cap * 2;
|
||||
ScrFiber **grown = p->waiters == &p->waiter_inline
|
||||
? malloc(cap * sizeof *grown)
|
||||
: realloc(p->waiters, cap * sizeof *grown);
|
||||
if (!grown) scr_oom();
|
||||
if (p->waiters == &p->waiter_inline) grown[0] = p->waiter_inline;
|
||||
p->waiters = grown;
|
||||
p->waiters_cap = cap;
|
||||
}
|
||||
}
|
||||
p->waiters[p->nwaiters++] = f;
|
||||
}
|
||||
|
||||
static bool scr_promise_has_traced_children(const ScrPromise *p) {
|
||||
return p->ncbs != 0 ||
|
||||
((p->payload_kind == SCR_EXC_REF || p->payload_kind == SCR_EXC_OBJ ||
|
||||
p->payload_kind == SCR_EXC_PRIMITIVE_REF) &&
|
||||
p->trace_fn && p->payload);
|
||||
}
|
||||
|
||||
static void scr_promise_trace(void *o, ScrTraceVisit visit, void *ctx) {
|
||||
ScrPromise *p = (ScrPromise *)o;
|
||||
/* Waiters are fibers (not refcounted objects) — the settled payload and
|
||||
@@ -266,8 +300,7 @@ static void scr_promise_gcfree(void *o) {
|
||||
for (size_t i = 0; i < p->ncbs; i++) {
|
||||
if (p->cbs[i].all) scr_promise_all_state_release(p->cbs[i].all);
|
||||
}
|
||||
if (--p->identity->rc == 0) free(p->identity);
|
||||
free(p->waiters);
|
||||
scr_promise_free_side(p);
|
||||
free(p->cbs);
|
||||
#ifdef SCR_RC_AUDIT
|
||||
scr_live_promises--;
|
||||
@@ -279,9 +312,6 @@ ScrPromise *scr_promise_new(void) {
|
||||
static const ScrCycType cyc_type = { &scr_promise_trace, &scr_promise_gcfree, SCR_CYC_KIND_PROMISE };
|
||||
ScrPromise *p = scr_cyc_alloc(sizeof *p, &cyc_type);
|
||||
p->rc = 1;
|
||||
p->identity = malloc(sizeof *p->identity);
|
||||
if (!p->identity) scr_oom();
|
||||
p->identity->rc = 1;
|
||||
#ifdef SCR_RC_AUDIT
|
||||
scr_live_promises++;
|
||||
#endif
|
||||
@@ -289,14 +319,20 @@ ScrPromise *scr_promise_new(void) {
|
||||
}
|
||||
|
||||
void scr_promise_share_identity(ScrPromise *dst, ScrPromise *src) {
|
||||
if (dst == src) return;
|
||||
if (!src->identity) {
|
||||
src->identity = malloc(sizeof *src->identity);
|
||||
if (!src->identity) scr_oom();
|
||||
src->identity->rc = 1;
|
||||
}
|
||||
if (dst->identity == src->identity) return;
|
||||
if (--dst->identity->rc == 0) free(dst->identity);
|
||||
if (dst->identity && --dst->identity->rc == 0) free(dst->identity);
|
||||
dst->identity = src->identity;
|
||||
dst->identity->rc++;
|
||||
}
|
||||
|
||||
bool scr_promise_identity_equal(ScrPromise *a, ScrPromise *b) {
|
||||
return a->identity == b->identity;
|
||||
return a == b || (a->identity != NULL && a->identity == b->identity);
|
||||
}
|
||||
|
||||
/* The 'rejectionHandled' hook (scr_async_dyn.c installs it at listener
|
||||
@@ -352,8 +388,7 @@ void scr_promise_release(ScrPromise *p) {
|
||||
if (--p->rc == 0) {
|
||||
scr_cyc_dead(p);
|
||||
scr_promise_release_payload(p);
|
||||
if (--p->identity->rc == 0) free(p->identity);
|
||||
free(p->waiters);
|
||||
scr_promise_free_side(p);
|
||||
for (size_t i = 0; i < p->ncbs; i++) {
|
||||
scr_promise_release(p->cbs[i].dst);
|
||||
if (p->cbs[i].all) scr_promise_all_state_release(p->cbs[i].all);
|
||||
@@ -363,9 +398,14 @@ void scr_promise_release(ScrPromise *p) {
|
||||
scr_live_promises--;
|
||||
#endif
|
||||
scr_cyc_free(p);
|
||||
} else {
|
||||
} else if (scr_promise_has_traced_children(p)) {
|
||||
scr_cyc_released(p); /* possible cycle root; may collect — p is done */
|
||||
}
|
||||
/* A promise the trace walks nothing from cannot be on a cycle at this
|
||||
* decrement, so it is not a candidate root. If it later gains traced
|
||||
* children (a cycle-capable payload or a combinator destination) while
|
||||
* still referenced, the decrement that could strand a cycle through it
|
||||
* sees them and buffers it then. */
|
||||
}
|
||||
|
||||
void *scr_promise_retain_v(void *p) { return scr_promise_retain((ScrPromise *)p); }
|
||||
@@ -557,6 +597,14 @@ struct ScrFiber {
|
||||
/* Async-generator settlement jobs: stackless READY-queue envelopes.
|
||||
* The retained generator is pumped on the main stack. */
|
||||
ScrGen *gen_job;
|
||||
/* 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
|
||||
* promise right away (scr_fiber_park_awaiters) instead of switching back
|
||||
* just so it can park itself. */
|
||||
struct ScrFiber *awaiter;
|
||||
/* This fiber was parked by such a child; set before the switch away. */
|
||||
bool await_parked;
|
||||
#ifdef SCR_ASAN_FIBERS
|
||||
void *fake_stack;
|
||||
#endif
|
||||
@@ -599,6 +647,8 @@ void scr_als_ctx_release(ScrAlsCtx *c) {
|
||||
|
||||
static SCR_TL ScrCtx scr_loop_ctx; /* the main stack (scheduler home) */
|
||||
static SCR_TL ScrFiber *scr_current = NULL;
|
||||
/* A finished fiber whose last switch went to the loop (scr_resume_fiber). */
|
||||
static SCR_TL ScrFiber *scr_loop_reap = NULL;
|
||||
static SCR_TL long scr_fibers_live = 0;
|
||||
static SCR_TL long scr_fibers_abandoned = 0;
|
||||
|
||||
@@ -1548,6 +1598,10 @@ static void scr_trampoline(void) {
|
||||
ScrFiber *self = scr_current;
|
||||
self->entry(self, self->argpack);
|
||||
scr_fiber_finish(self);
|
||||
/* The loop reaps the fiber that comes back to it finished, which need
|
||||
* not be the one it resumed: a fiber that continued in place for the
|
||||
* awaiters it parked (scr_await_yield) returns to the loop too. */
|
||||
if (self->gen == NULL && self->return_to == &scr_loop_ctx) scr_loop_reap = self;
|
||||
/* Dead fiber: hop back to whoever resumed us. The loop frees us. */
|
||||
scr_switch(&self->ctx, self->return_to, NULL);
|
||||
/* unreachable */
|
||||
@@ -1626,13 +1680,24 @@ static void scr_fiber_destroy(ScrFiber *f) {
|
||||
#else
|
||||
free(f->stack);
|
||||
#endif
|
||||
free(f);
|
||||
scr_mem_free(f);
|
||||
}
|
||||
|
||||
#if !defined(_WIN32) && !defined(__wasi__) && !defined(SCR_WORKERS)
|
||||
#define SCR_SPAWN_AWAIT 1
|
||||
#endif
|
||||
|
||||
/* Spawns and EAGERLY runs an async body until its first suspension (JS's
|
||||
* synchronous-prefix rule). Returns the fiber's promise, +1. */
|
||||
ScrPromise *scr_async_spawn(void (*entry)(ScrFiber *, void *), void *argpack) {
|
||||
ScrFiber *f = calloc(1, sizeof *f);
|
||||
* synchronous-prefix rule). Returns the fiber's promise, +1. A non-NULL
|
||||
* `awaiter` (the current fiber, SCR_SPAWN_AWAIT only) awaits that promise
|
||||
* immediately: the child saves into and returns to the awaiter's own
|
||||
* context, so a child suspension can park the awaiter and switch past
|
||||
* it (scr_fiber_park_awaiters). The promise is retained before the child
|
||||
* runs, as the child may then finish and be reaped by the loop before the
|
||||
* awaiter resumes. */
|
||||
static ScrPromise *scr_async_spawn_impl(void (*entry)(ScrFiber *, void *), void *argpack,
|
||||
ScrFiber *awaiter) {
|
||||
ScrFiber *f = scr_mem_calloc(sizeof *f);
|
||||
if (!f) scr_oom();
|
||||
scr_fiber_track(f);
|
||||
f->promise = scr_promise_new();
|
||||
@@ -1651,6 +1716,7 @@ ScrPromise *scr_async_spawn(void (*entry)(ScrFiber *, void *), void *argpack) {
|
||||
f->ctx = CreateFiberEx(4096, 0, 0, scr_trampoline, NULL);
|
||||
if (f->ctx == NULL) scr_oom();
|
||||
#elif defined(__wasi__)
|
||||
(void)awaiter;
|
||||
ScrFiber *spawner = scr_current;
|
||||
scr_current = f;
|
||||
scr_exc_swap_cell(&f->exc);
|
||||
@@ -1672,9 +1738,21 @@ ScrPromise *scr_async_spawn(void (*entry)(ScrFiber *, void *), void *argpack) {
|
||||
ScrCtx here;
|
||||
#endif
|
||||
#ifndef __wasi__
|
||||
f->return_to = &here;
|
||||
ScrCtx *home = &here;
|
||||
ScrPromise *awaited = NULL;
|
||||
#ifdef SCR_SPAWN_AWAIT
|
||||
if (awaiter != NULL) {
|
||||
f->awaiter = awaiter;
|
||||
awaiter->await_parked = false;
|
||||
home = &awaiter->ctx;
|
||||
awaited = scr_promise_retain(f->promise);
|
||||
}
|
||||
#else
|
||||
(void)awaiter;
|
||||
#endif
|
||||
f->return_to = home;
|
||||
ScrFiber *spawner = scr_current;
|
||||
scr_switch(&here, &f->ctx, f);
|
||||
scr_switch(home, &f->ctx, f);
|
||||
/* back: the fiber suspended or finished. The switch back targeted NULL
|
||||
* (the child doesn't know who spawned it), which pointed both the current
|
||||
* fiber AND the exception machinery at main — restore the spawner's cell
|
||||
@@ -1684,7 +1762,12 @@ ScrPromise *scr_async_spawn(void (*entry)(ScrFiber *, void *), void *argpack) {
|
||||
scr_current = spawner;
|
||||
scr_exc_swap_cell(spawner ? &spawner->exc : NULL);
|
||||
scr_als_active = spawner ? &spawner->als : &scr_als_main_slot;
|
||||
ScrPromise *result = scr_promise_retain(f->promise);
|
||||
if (awaited != NULL) {
|
||||
/* Parked by the child: the loop owns (and may have reaped) it. */
|
||||
if (awaiter->await_parked) return awaited;
|
||||
f->awaiter = NULL;
|
||||
}
|
||||
ScrPromise *result = awaited != NULL ? awaited : scr_promise_retain(f->promise);
|
||||
if (f->done) {
|
||||
scr_promise_release(f->promise);
|
||||
scr_fiber_destroy(f);
|
||||
@@ -1694,6 +1777,81 @@ ScrPromise *scr_async_spawn(void (*entry)(ScrFiber *, void *), void *argpack) {
|
||||
#endif
|
||||
}
|
||||
|
||||
ScrPromise *scr_async_spawn(void (*entry)(ScrFiber *, void *), void *argpack) {
|
||||
return scr_async_spawn_impl(entry, argpack, NULL);
|
||||
}
|
||||
|
||||
static void scr_await_park(ScrPromise *p);
|
||||
static void scr_promise_rethrow(ScrPromise *p);
|
||||
|
||||
/* `await f(...)` of a fiber-backed async f in one call (the compiler's
|
||||
* await-call entry): spawn, then await the promise, with the same turns
|
||||
* as the two steps. When the child suspends in its eager prefix, the
|
||||
* awaiter is already parked on the pending promise, so the settle wake
|
||||
* resumes it with the result and no further hop; a child that finished
|
||||
* eagerly leaves a settled promise, which the ordinary await hops on.
|
||||
* Returns the promise +1; `ok` false means the rejection was re-thrown. */
|
||||
|
||||
static ScrPromise *scr_async_spawn_await(void (*entry)(ScrFiber *, void *), void *argpack,
|
||||
bool *ok) {
|
||||
#ifdef SCR_SPAWN_AWAIT
|
||||
ScrFiber *self = scr_current;
|
||||
if (self != NULL && self->gen == NULL && self->inline_depth == 0) {
|
||||
ScrPromise *p = scr_async_spawn_impl(entry, argpack, self);
|
||||
if (self->await_parked) {
|
||||
self->await_parked = false;
|
||||
while (p->state == SCR_PROM_PENDING) scr_await_park(p);
|
||||
scr_prom_observe(p);
|
||||
if (p->state == SCR_PROM_REJECTED) {
|
||||
scr_promise_rethrow(p);
|
||||
*ok = false;
|
||||
} else {
|
||||
*ok = true;
|
||||
}
|
||||
return p;
|
||||
}
|
||||
*ok = scr_await_settled(p);
|
||||
return p;
|
||||
}
|
||||
#endif
|
||||
ScrPromise *p = scr_async_spawn(entry, argpack);
|
||||
*ok = scr_await_settled(p);
|
||||
return p;
|
||||
}
|
||||
|
||||
void scr_async_spawn_await_void(void (*entry)(ScrFiber *, void *), void *argpack) {
|
||||
bool ok;
|
||||
scr_promise_release(scr_async_spawn_await(entry, argpack, &ok));
|
||||
}
|
||||
double scr_async_spawn_await_f64(void (*entry)(ScrFiber *, void *), void *argpack) {
|
||||
bool ok;
|
||||
ScrPromise *p = scr_async_spawn_await(entry, argpack, &ok);
|
||||
double v = ok ? p->f64 : 0;
|
||||
scr_promise_release(p);
|
||||
return v;
|
||||
}
|
||||
bool scr_async_spawn_await_bool(void (*entry)(ScrFiber *, void *), void *argpack) {
|
||||
bool ok;
|
||||
ScrPromise *p = scr_async_spawn_await(entry, argpack, &ok);
|
||||
bool v = ok ? p->b : false;
|
||||
scr_promise_release(p);
|
||||
return v;
|
||||
}
|
||||
ScrStr *scr_async_spawn_await_str(void (*entry)(ScrFiber *, void *), void *argpack) {
|
||||
bool ok;
|
||||
ScrPromise *p = scr_async_spawn_await(entry, argpack, &ok);
|
||||
ScrStr *v = ok && p->payload ? scr_str_retain((ScrStr *)p->payload) : NULL;
|
||||
scr_promise_release(p);
|
||||
return v;
|
||||
}
|
||||
void *scr_async_spawn_await_ref(void (*entry)(ScrFiber *, void *), void *argpack) {
|
||||
bool ok;
|
||||
ScrPromise *p = scr_async_spawn_await(entry, argpack, &ok);
|
||||
void *v = ok && p->payload ? p->retain_fn(p->payload) : NULL;
|
||||
scr_promise_release(p);
|
||||
return v;
|
||||
}
|
||||
|
||||
/* ── fiberless async calls ─────────────────────────────────────────────
|
||||
* An async function whose body the compiler proved can never park its
|
||||
* execution context (no await, no other suspending runtime call, directly
|
||||
@@ -1721,7 +1879,7 @@ void scr_async_inline_enter(ScrAsyncInline *frame) {
|
||||
if (scr_current != NULL) scr_current->inline_depth++;
|
||||
}
|
||||
|
||||
ScrPromise *scr_async_inline_leave(ScrAsyncInline *frame) {
|
||||
void scr_async_inline_exit(ScrAsyncInline *frame) {
|
||||
if (scr_current != NULL) scr_current->inline_depth--;
|
||||
scr_exc_current_cell()->stack = (ScrStackFrame *)frame->stack;
|
||||
ScrAlsCtx *saved = (ScrAlsCtx *)frame->als;
|
||||
@@ -1732,6 +1890,10 @@ ScrPromise *scr_async_inline_leave(ScrAsyncInline *frame) {
|
||||
} else {
|
||||
scr_als_ctx_release(saved);
|
||||
}
|
||||
}
|
||||
|
||||
ScrPromise *scr_async_inline_leave(ScrAsyncInline *frame) {
|
||||
scr_async_inline_exit(frame);
|
||||
return scr_promise_new();
|
||||
}
|
||||
|
||||
@@ -1747,7 +1909,7 @@ ScrPromise *scr_async_spawn_after(ScrPromise *dependency,
|
||||
(void)dependency;
|
||||
return scr_async_spawn(entry, argpack);
|
||||
#else
|
||||
ScrFiber *f = calloc(1, sizeof *f);
|
||||
ScrFiber *f = scr_mem_calloc(sizeof *f);
|
||||
if (!f) scr_oom();
|
||||
f->promise = scr_promise_new();
|
||||
f->entry = entry;
|
||||
@@ -1756,13 +1918,7 @@ ScrPromise *scr_async_spawn_after(ScrPromise *dependency,
|
||||
scr_fibers_live++;
|
||||
|
||||
if (dependency->state == SCR_PROM_PENDING) {
|
||||
if (dependency->nwaiters == dependency->waiters_cap) {
|
||||
dependency->waiters_cap = dependency->waiters_cap ? dependency->waiters_cap * 2 : 4;
|
||||
dependency->waiters = realloc(
|
||||
dependency->waiters, dependency->waiters_cap * sizeof *dependency->waiters);
|
||||
if (!dependency->waiters) scr_oom();
|
||||
}
|
||||
dependency->waiters[dependency->nwaiters++] = f;
|
||||
scr_promise_add_waiter(dependency, f);
|
||||
} else {
|
||||
scr_ready_push(f);
|
||||
}
|
||||
@@ -1770,6 +1926,29 @@ ScrPromise *scr_async_spawn_after(ScrPromise *dependency,
|
||||
#endif
|
||||
}
|
||||
|
||||
/* Where a suspending async fiber switches to. During the eager prefix
|
||||
* under an await-call entry (awaiter set), control would return to the
|
||||
* awaiter only for it to park on this fiber's still-pending promise and
|
||||
* switch to its own resumer: park it here, for each such awaiter up the
|
||||
* chain, and switch there directly. Nothing runs in between either way. */
|
||||
static ScrCtx *scr_fiber_park_awaiters(ScrFiber *self) {
|
||||
ScrCtx *to = self->return_to;
|
||||
#ifdef SCR_SPAWN_AWAIT
|
||||
for (ScrFiber *cur = self; cur->awaiter != NULL;) {
|
||||
ScrFiber *aw = cur->awaiter;
|
||||
cur->awaiter = NULL;
|
||||
scr_promise_add_waiter(cur->promise, aw);
|
||||
aw->await_parked = true;
|
||||
to = aw->return_to;
|
||||
cur = aw;
|
||||
}
|
||||
/* Whoever resumes this fiber next (the loop) becomes its return target;
|
||||
* also right when the caller continues in place without switching. */
|
||||
self->return_to = to;
|
||||
#endif
|
||||
return to;
|
||||
}
|
||||
|
||||
/* Parks the current fiber on `p` until it settles. Only fibers await. */
|
||||
static void scr_await_park(ScrPromise *p) {
|
||||
ScrFiber *self = scr_current;
|
||||
@@ -1778,16 +1957,11 @@ static void scr_await_park(ScrPromise *p) {
|
||||
abort();
|
||||
}
|
||||
if (self->inline_depth != 0) scr_inline_park_trap();
|
||||
if (p->nwaiters == p->waiters_cap) {
|
||||
p->waiters_cap = p->waiters_cap ? p->waiters_cap * 2 : 4;
|
||||
p->waiters = realloc(p->waiters, p->waiters_cap * sizeof *p->waiters);
|
||||
if (!p->waiters) scr_oom();
|
||||
}
|
||||
p->waiters[p->nwaiters++] = self;
|
||||
scr_promise_add_waiter(p, self);
|
||||
#ifdef SCR_WORKERS
|
||||
self->waiting = p;
|
||||
#endif
|
||||
scr_switch(&self->ctx, self->return_to, NULL);
|
||||
scr_switch(&self->ctx, scr_fiber_park_awaiters(self), NULL);
|
||||
#ifdef SCR_WORKERS
|
||||
/* Shutdown resumes suspended stacks without settling their dependencies.
|
||||
* Remove the borrowed waiter before the stack releases its promise. */
|
||||
@@ -1809,12 +1983,7 @@ static void scr_await_park(ScrPromise *p) {
|
||||
* the continuation through their waiter list. */
|
||||
void scr_wasi_await_prepare(ScrFiber *self, ScrPromise *p) {
|
||||
if (p->state == SCR_PROM_PENDING) {
|
||||
if (p->nwaiters == p->waiters_cap) {
|
||||
p->waiters_cap = p->waiters_cap ? p->waiters_cap * 2 : 4;
|
||||
p->waiters = realloc(p->waiters, p->waiters_cap * sizeof *p->waiters);
|
||||
if (!p->waiters) scr_oom();
|
||||
}
|
||||
p->waiters[p->nwaiters++] = self;
|
||||
scr_promise_add_waiter(p, self);
|
||||
} else {
|
||||
scr_ready_push(self);
|
||||
}
|
||||
@@ -1851,8 +2020,22 @@ static void scr_await_yield(void) {
|
||||
abort();
|
||||
}
|
||||
if (self->inline_depth != 0) scr_inline_park_trap();
|
||||
if (self->gen != NULL) {
|
||||
scr_ready_push(self);
|
||||
scr_switch(&self->ctx, self->return_to, NULL);
|
||||
return;
|
||||
}
|
||||
/* The hop would return control to the loop's microtask drain with
|
||||
* nothing else queued (this async fiber, or the awaiters parked for it,
|
||||
* were resumed by the drain), so this fiber would be the very next job:
|
||||
* the drain runs to exhaustion with no checkpoint in between. Continuing
|
||||
* in place is the same turn order without the two switches. A fiber
|
||||
* resumed by its spawner (the eager prefix) must still hop: the spawner
|
||||
* has code to run first. */
|
||||
ScrCtx *to = scr_fiber_park_awaiters(self);
|
||||
if (to == &scr_loop_ctx && scr_ready_len == 0) return;
|
||||
scr_ready_push(self);
|
||||
scr_switch(&self->ctx, self->return_to, NULL);
|
||||
scr_switch(&self->ctx, to, NULL);
|
||||
}
|
||||
|
||||
/* The emitted promise-or-absent await's unit arm (`await u` where u holds
|
||||
@@ -2782,7 +2965,7 @@ ScrDyn *scr_set_immediate_dyn_value(void) {
|
||||
* main stack; a throw is an uncaught exception (Node's queueMicrotask),
|
||||
* never a rejection. */
|
||||
void scr_queue_microtask(ScrClosure *cb) {
|
||||
ScrFiber *f = calloc(1, sizeof *f);
|
||||
ScrFiber *f = scr_mem_calloc(sizeof *f);
|
||||
if (!f) scr_oom();
|
||||
f->micro_cb = cb; /* ownership moves in */
|
||||
scr_ready_push(f);
|
||||
@@ -2840,7 +3023,7 @@ void scr_loop_set_events(bool (*pending)(void), bool (*watching)(void),
|
||||
static void scr_resume_fiber(ScrFiber *f) {
|
||||
if (f->gen_job != NULL) {
|
||||
ScrGen *g = f->gen_job;
|
||||
free(f);
|
||||
scr_mem_free(f);
|
||||
scr_async_gen_job(g);
|
||||
scr_gen_release(g);
|
||||
return;
|
||||
@@ -2851,7 +3034,7 @@ static void scr_resume_fiber(ScrFiber *f) {
|
||||
* semantics (uncaughtException, never an unhandled rejection). */
|
||||
if (f->micro_cb != NULL) {
|
||||
ScrClosure *cb = f->micro_cb;
|
||||
free(f);
|
||||
scr_mem_free(f);
|
||||
((void (*)(ScrClosure *))cb->fn)(cb);
|
||||
scr_closure_release(cb);
|
||||
return;
|
||||
@@ -2884,6 +3067,9 @@ static void scr_resume_fiber(ScrFiber *f) {
|
||||
scr_als_active = &scr_als_main_slot;
|
||||
#else
|
||||
scr_switch(&scr_loop_ctx, &f->ctx, f);
|
||||
f = scr_loop_reap;
|
||||
if (f == NULL) return;
|
||||
scr_loop_reap = NULL;
|
||||
#endif
|
||||
if (f->done) {
|
||||
scr_promise_release(f->promise);
|
||||
@@ -3868,7 +4054,7 @@ static ScrGen *scr_gen_new_common(void (*entry)(ScrFiber *, void *), void *argpa
|
||||
g->settle_async = settle_async;
|
||||
scr_obj_alloc_note();
|
||||
|
||||
ScrFiber *f = calloc(1, sizeof *f);
|
||||
ScrFiber *f = scr_mem_calloc(sizeof *f);
|
||||
if (!f) scr_oom();
|
||||
f->gen = g;
|
||||
scr_fiber_track(f);
|
||||
@@ -4273,7 +4459,7 @@ static void scr_async_gen_start_queued(ScrGen *g);
|
||||
|
||||
static void scr_async_gen_schedule_settle(ScrGen *g) {
|
||||
if (g->settle_queued) return;
|
||||
ScrFiber *job = calloc(1, sizeof *job);
|
||||
ScrFiber *job = scr_mem_calloc(sizeof *job);
|
||||
if (!job) scr_oom();
|
||||
job->gen_job = scr_gen_retain(g);
|
||||
g->settle_queued = true;
|
||||
@@ -4485,12 +4671,12 @@ static void scr_context_ready_discard(void) {
|
||||
scr_ready_len--;
|
||||
if (fiber->micro_cb) {
|
||||
scr_closure_release(fiber->micro_cb);
|
||||
free(fiber);
|
||||
scr_mem_free(fiber);
|
||||
} else if (fiber->gen_job) {
|
||||
ScrGen *generator = fiber->gen_job;
|
||||
generator->settle_queued = false;
|
||||
scr_gen_release(generator);
|
||||
free(fiber);
|
||||
scr_mem_free(fiber);
|
||||
}
|
||||
/* Real fibers belong to the context registry, including queued ones. */
|
||||
}
|
||||
|
||||
@@ -5351,6 +5351,13 @@ void scr_promise_release_v(void *p);
|
||||
typedef struct ScrFiber ScrFiber;
|
||||
/* Spawn + run eagerly to the first suspension; returns the promise, +1. */
|
||||
ScrPromise *scr_async_spawn(void (*entry)(ScrFiber *, void *), void *argpack);
|
||||
/* `await f(...)` of a fiber-backed async f: spawn and await in one call,
|
||||
* returning the awaited value (a rejection re-throws into the caller). */
|
||||
void scr_async_spawn_await_void(void (*entry)(ScrFiber *, void *), void *argpack);
|
||||
double scr_async_spawn_await_f64(void (*entry)(ScrFiber *, void *), void *argpack);
|
||||
bool scr_async_spawn_await_bool(void (*entry)(ScrFiber *, void *), void *argpack);
|
||||
ScrStr *scr_async_spawn_await_str(void (*entry)(ScrFiber *, void *), void *argpack);
|
||||
void *scr_async_spawn_await_ref(void (*entry)(ScrFiber *, void *), void *argpack);
|
||||
/* Runtime-authored C waiters cannot themselves become LLVM switched
|
||||
* coroutines on wasm32. Spawn after `dependency` settles so their first
|
||||
* (and only) await is extraction from an already-settled promise, while
|
||||
@@ -5370,6 +5377,10 @@ typedef struct ScrAsyncInline {
|
||||
} ScrAsyncInline;
|
||||
void scr_async_inline_enter(ScrAsyncInline *frame);
|
||||
ScrPromise *scr_async_inline_leave(ScrAsyncInline *frame);
|
||||
/* Leaves the frame without creating the call's promise: `await f()` of a
|
||||
* fiberless f consumes the result directly (the emitted await-call entry
|
||||
* still takes the await's microtask turn). */
|
||||
void scr_async_inline_exit(ScrAsyncInline *frame);
|
||||
|
||||
/* wasm32-wasi lowers async functions and generators through LLVM switched
|
||||
* coroutines instead of the native stack-switching implementations. The
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
// Directly awaited async calls (bodies that never suspend, bodies that do,
|
||||
// and chains of them) keep Node's microtask order against then-callbacks,
|
||||
// queueMicrotask, and settled and pending promises.
|
||||
const log: string[] = [];
|
||||
const cache = new Map<number, number>();
|
||||
|
||||
async function hit(key: number): Promise<number> {
|
||||
const v = cache.get(key);
|
||||
if (v !== undefined) return v;
|
||||
cache.set(key, key * 10);
|
||||
return key * 10;
|
||||
}
|
||||
async function flag(n: number): Promise<boolean> {
|
||||
return n % 2 === 0;
|
||||
}
|
||||
async function label(n: number): Promise<string> {
|
||||
return "L" + n;
|
||||
}
|
||||
async function nothing(n: number): Promise<void> {
|
||||
log.push("nothing " + n);
|
||||
}
|
||||
async function suspends(n: number): Promise<number> {
|
||||
log.push("suspends enter " + n);
|
||||
const a = await hit(n);
|
||||
log.push("suspends mid " + n);
|
||||
const b = await hit(n + 1);
|
||||
return a + b;
|
||||
}
|
||||
async function chain(n: number): Promise<number> {
|
||||
log.push("chain enter " + n);
|
||||
const v = await suspends(n);
|
||||
log.push("chain got " + n + " " + v);
|
||||
return v + 1;
|
||||
}
|
||||
async function deep(n: number): Promise<number> {
|
||||
return (await chain(n)) * 2;
|
||||
}
|
||||
|
||||
async function worker(name: string, n: number): Promise<number> {
|
||||
log.push(name + " start");
|
||||
const a = await hit(n);
|
||||
log.push(name + " a=" + a);
|
||||
const f = await flag(n);
|
||||
log.push(name + " flag=" + f);
|
||||
const s = await label(n);
|
||||
log.push(name + " label=" + s);
|
||||
await nothing(n);
|
||||
log.push(name + " after nothing");
|
||||
const d = await deep(n);
|
||||
log.push(name + " deep=" + d);
|
||||
return a + d;
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
const settled = Promise.resolve(7);
|
||||
const pending = new Promise<number>((resolve) => {
|
||||
setTimeout(() => {
|
||||
log.push("timer resolves");
|
||||
resolve(9);
|
||||
}, 0);
|
||||
});
|
||||
settled.then((v) => log.push("then settled " + v));
|
||||
queueMicrotask(() => log.push("microtask 1"));
|
||||
const w1 = worker("w1", 1);
|
||||
const w2 = worker("w2", 2);
|
||||
queueMicrotask(() => log.push("microtask 2"));
|
||||
Promise.resolve()
|
||||
.then(() => log.push("then A1"))
|
||||
.then(() => log.push("then A2"))
|
||||
.then(() => log.push("then A3"))
|
||||
.then(() => log.push("then A4"));
|
||||
log.push("main sync end");
|
||||
const x = await settled;
|
||||
log.push("main settled " + x);
|
||||
const t = await (async (): Promise<number> => await hit(5))();
|
||||
log.push("main arrow " + t);
|
||||
const both: number[] = await Promise.all([w1, w2]);
|
||||
log.push("workers " + both[0] + " " + both[1]);
|
||||
const p = await pending;
|
||||
log.push("main pending " + p);
|
||||
const r = await deep(3);
|
||||
log.push("main deep " + r);
|
||||
}
|
||||
|
||||
main().then(() => {
|
||||
for (const line of log) console.log(line);
|
||||
});
|
||||
console.log("top-level sync end");
|
||||
@@ -0,0 +1,82 @@
|
||||
// Exceptions thrown by directly awaited async calls, before and after the
|
||||
// callee's first await, reach the awaiter's catch on Node's turn, and
|
||||
// finally blocks and rethrows keep their order against other microtasks.
|
||||
const log: string[] = [];
|
||||
|
||||
async function failsNow(n: number): Promise<number> {
|
||||
if (n > 0) throw new Error("now " + n);
|
||||
return n;
|
||||
}
|
||||
async function failsString(n: number): Promise<string> {
|
||||
if (n > 0) throw "plain " + n;
|
||||
return "ok";
|
||||
}
|
||||
async function failsLater(n: number): Promise<number> {
|
||||
log.push("later enter " + n);
|
||||
await failsNow(0);
|
||||
log.push("later resumed " + n);
|
||||
throw new Error("later " + n);
|
||||
}
|
||||
async function failsVoid(): Promise<void> {
|
||||
throw new RangeError("void failure");
|
||||
}
|
||||
async function rethrows(n: number): Promise<number> {
|
||||
try {
|
||||
return await failsLater(n);
|
||||
} catch (e) {
|
||||
log.push("rethrows caught " + (e as Error).message);
|
||||
throw new Error("wrapped " + (e as Error).message);
|
||||
} finally {
|
||||
log.push("rethrows finally " + n);
|
||||
}
|
||||
}
|
||||
|
||||
async function probe(name: string): Promise<void> {
|
||||
try {
|
||||
await failsNow(1);
|
||||
} catch (e) {
|
||||
log.push(name + " caught " + (e as Error).message);
|
||||
}
|
||||
try {
|
||||
await failsString(2);
|
||||
} catch (e) {
|
||||
log.push(name + " caught " + String(e));
|
||||
}
|
||||
try {
|
||||
await failsVoid();
|
||||
} catch (e) {
|
||||
log.push(name + " caught " + (e as Error).name + " " + (e as Error).message);
|
||||
}
|
||||
try {
|
||||
await rethrows(3);
|
||||
} catch (e) {
|
||||
log.push(name + " caught " + (e as Error).message);
|
||||
}
|
||||
const ok = await failsNow(0);
|
||||
log.push(name + " ok " + ok);
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
queueMicrotask(() => log.push("microtask before"));
|
||||
const a = probe("a");
|
||||
const b = probe("b");
|
||||
Promise.resolve()
|
||||
.then(() => log.push("tick 1"))
|
||||
.then(() => log.push("tick 2"))
|
||||
.then(() => log.push("tick 3"));
|
||||
await a;
|
||||
await b;
|
||||
let caught = "";
|
||||
try {
|
||||
await (async (): Promise<number> => {
|
||||
throw new Error("inline arrow");
|
||||
})();
|
||||
} catch (e) {
|
||||
caught = (e as Error).message;
|
||||
}
|
||||
log.push("arrow " + caught);
|
||||
}
|
||||
|
||||
main().then(() => {
|
||||
for (const line of log) console.log(line);
|
||||
});
|
||||
@@ -0,0 +1,53 @@
|
||||
// A request handler chain of small async helpers that usually complete
|
||||
// without suspending (cache hits), awaited directly, with concurrent
|
||||
// requests interleaving on the microtask queue.
|
||||
interface Session {
|
||||
user: string;
|
||||
role: string;
|
||||
}
|
||||
const sessions = new Map<string, Session>();
|
||||
const quotas = new Map<string, number>();
|
||||
const order: string[] = [];
|
||||
|
||||
async function lookupSession(token: string): Promise<Session | undefined> {
|
||||
return sessions.get(token);
|
||||
}
|
||||
async function loadSession(token: string): Promise<Session> {
|
||||
const hit = await lookupSession(token);
|
||||
if (hit !== undefined) return hit;
|
||||
await new Promise<void>((resolve) => setTimeout(resolve, 0));
|
||||
const fresh = { user: "u-" + token, role: token.length % 2 === 0 ? "admin" : "user" };
|
||||
sessions.set(token, fresh);
|
||||
return fresh;
|
||||
}
|
||||
async function canWrite(session: Session): Promise<boolean> {
|
||||
return session.role === "admin";
|
||||
}
|
||||
async function quota(user: string): Promise<number> {
|
||||
const q = quotas.get(user) ?? 3;
|
||||
quotas.set(user, q - 1);
|
||||
return q;
|
||||
}
|
||||
async function render(status: number, body: string): Promise<string> {
|
||||
return status + " " + body;
|
||||
}
|
||||
async function handle(id: number, token: string): Promise<string> {
|
||||
order.push("start " + id);
|
||||
const session = await loadSession(token);
|
||||
if (!(await canWrite(session))) return await render(403, session.user);
|
||||
const left = await quota(session.user);
|
||||
if (left <= 0) return await render(429, session.user);
|
||||
order.push("write " + id);
|
||||
return await render(200, session.user + " left " + left);
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
const tokens = ["ab", "abc", "ab", "abcd", "ab", "abc", "abcd", "ab", "ab", "ab"];
|
||||
const first = await Promise.all(tokens.map((t, i) => handle(i, t)));
|
||||
for (const r of first) console.log(r);
|
||||
const second: string[] = [];
|
||||
for (let i = 0; i < tokens.length; i++) second.push(await handle(100 + i, tokens[i]!));
|
||||
for (const r of second) console.log(r);
|
||||
console.log(order.join(","));
|
||||
}
|
||||
void main();
|
||||
@@ -0,0 +1,32 @@
|
||||
// await of non-promise values and of directly called async functions each
|
||||
// take exactly one microtask turn, interleaved with two counters.
|
||||
const log: string[] = [];
|
||||
async function one(): Promise<number> {
|
||||
return 1;
|
||||
}
|
||||
async function ticker(name: string, n: number): Promise<void> {
|
||||
for (let i = 0; i < n; i++) {
|
||||
log.push(name + i);
|
||||
await null;
|
||||
}
|
||||
}
|
||||
async function mixed(): Promise<void> {
|
||||
log.push("m0");
|
||||
const a = await 5;
|
||||
log.push("m1 " + a);
|
||||
const b = await one();
|
||||
log.push("m2 " + b);
|
||||
const c = await "text";
|
||||
log.push("m3 " + c);
|
||||
const d = await undefined;
|
||||
log.push("m4 " + d);
|
||||
const e = await one();
|
||||
log.push("m5 " + e);
|
||||
const f = await Promise.resolve(await one());
|
||||
log.push("m6 " + f);
|
||||
}
|
||||
async function main(): Promise<void> {
|
||||
await Promise.all([ticker("t", 10), mixed(), ticker("s", 4)]);
|
||||
console.log(log.join(" "));
|
||||
}
|
||||
void main();
|
||||
@@ -0,0 +1,22 @@
|
||||
// @exit: 1
|
||||
// A rejection from a directly awaited async call is handled by that await,
|
||||
// while the same function's rejected promise left unawaited still reports
|
||||
// as an unhandled rejection after the microtask queue drains.
|
||||
async function fails(n: number): Promise<number> {
|
||||
if (n > 0) throw new Error("fail " + n);
|
||||
return n;
|
||||
}
|
||||
async function main(): Promise<void> {
|
||||
try {
|
||||
await fails(1);
|
||||
} catch (e) {
|
||||
console.log("handled", (e as Error).message);
|
||||
}
|
||||
process.on("unhandledRejection", (reason) => {
|
||||
console.log("unhandled", (reason as Error).message);
|
||||
process.exitCode = 1;
|
||||
});
|
||||
void fails(2);
|
||||
console.log("after", await fails(0));
|
||||
}
|
||||
void main();
|
||||
@@ -0,0 +1,40 @@
|
||||
// Directly awaited async calls against process.nextTick, setImmediate,
|
||||
// a timer and queueMicrotask: ticks run before promise jobs at each
|
||||
// checkpoint, and every await still waits for the jobs queued before it.
|
||||
const log: string[] = [];
|
||||
async function quick(n: number): Promise<number> {
|
||||
return n * 3;
|
||||
}
|
||||
async function once(n: number): Promise<number> {
|
||||
const v = await quick(n);
|
||||
process.nextTick(() => log.push("tick from once " + n));
|
||||
return v + 1;
|
||||
}
|
||||
async function runner(name: string): Promise<void> {
|
||||
log.push(name + " begin");
|
||||
for (let i = 0; i < 3; i++) {
|
||||
const v = await once(i);
|
||||
log.push(name + " step " + i + " = " + v);
|
||||
if (i === 1) {
|
||||
queueMicrotask(() => log.push(name + " microtask"));
|
||||
process.nextTick(() => log.push(name + " tick"));
|
||||
}
|
||||
}
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
log.push(name + " after immediate " + (await quick(7)));
|
||||
await new Promise<void>((resolve) => setTimeout(resolve, 1));
|
||||
log.push(name + " after timer " + (await once(9)));
|
||||
}
|
||||
async function main(): Promise<void> {
|
||||
process.nextTick(() => log.push("first tick"));
|
||||
setImmediate(() => log.push("immediate"));
|
||||
const a = runner("a");
|
||||
const b = runner("b");
|
||||
log.push("main sync");
|
||||
await Promise.all([a, b]);
|
||||
const p = quick(1);
|
||||
const q = Promise.resolve(p);
|
||||
log.push("same promise " + (p === q));
|
||||
log.push("value " + (await q));
|
||||
}
|
||||
main().then(() => console.log(log.join("\n")));
|
||||
Reference in New Issue
Block a user