use std::fmt::Write as _;
use crate::intrinsics::p3::waitable::WaitableIntrinsic;
use crate::intrinsics::{Intrinsic, RenderIntrinsicsArgs};
use crate::source::Source;
use crate::uwriteln;
#[derive(Debug, Copy, Clone, Ord, PartialOrd, Eq, PartialEq)]
pub enum ComponentIntrinsic {
GlobalAsyncStateMap,
GetOrCreateAsyncState,
BackpressureInc,
BackpressureDec,
ComponentAsyncStateClass,
ComponentStateSetAllError,
}
impl ComponentIntrinsic {
pub fn get_global_names() -> impl IntoIterator<Item = &'static str> {
[]
}
pub fn name(&self) -> &'static str {
match self {
Self::GlobalAsyncStateMap => "ASYNC_STATE",
Self::GetOrCreateAsyncState => "getOrCreateAsyncState",
Self::BackpressureInc => "backpressureInc",
Self::BackpressureDec => "backpressureDec",
Self::ComponentAsyncStateClass => "ComponentAsyncState",
Self::ComponentStateSetAllError => "_ComponentStateSetAllError",
}
}
pub fn render(&self, output: &mut Source, render_args: &RenderIntrinsicsArgs<'_>) {
match self {
Self::GlobalAsyncStateMap => {
let var_name = render_args.require_intrinsic(Self::GlobalAsyncStateMap);
uwriteln!(output, r#"const {var_name} = new Map();"#);
}
Self::BackpressureInc => {
let debug_log_fn = render_args.require_intrinsic(Intrinsic::DebugLog);
let backpressure_inc_fn = render_args.require_intrinsic(Self::BackpressureInc);
let get_or_create_async_state_fn =
render_args.require_intrinsic(Self::GetOrCreateAsyncState);
output.push_str(&format!(
r#"
function {backpressure_inc_fn}(componentIdx) {{
{debug_log_fn}('[{backpressure_inc_fn}()] args', {{ componentIdx }});
const state = {get_or_create_async_state_fn}(componentIdx);
if (!state) {{ throw new Error(`missing component state for component [${{componentIdx}}]`); }}
const newValue = state.incrementBackpressure();
{debug_log_fn}('[{backpressure_inc_fn}()] incremented', {{ componentIdx, newValue }});
}}
"#,
));
}
Self::BackpressureDec => {
let debug_log_fn = render_args.require_intrinsic(Intrinsic::DebugLog);
let backpressure_dec_fn = render_args.require_intrinsic(Self::BackpressureDec);
let get_or_create_async_state_fn =
render_args.require_intrinsic(Self::GetOrCreateAsyncState);
output.push_str(&format!(
"
function {backpressure_dec_fn}(componentIdx) {{
{debug_log_fn}('[{backpressure_dec_fn}()] args', {{ componentIdx }});
const state = {get_or_create_async_state_fn}(componentIdx);
const newValue = state.decrementBackpressure();
{debug_log_fn}('[{backpressure_dec_fn}()] decremented', {{ componentIdx, newValue }});
}}
"
));
}
Self::ComponentAsyncStateClass => {
let component_async_state_class = self.name();
let debug_log_fn = render_args.require_intrinsic(Intrinsic::DebugLog);
let rep_table_class = render_args.require_intrinsic(Intrinsic::RepTableClass);
let waitable_class =
render_args.require_intrinsic(WaitableIntrinsic::WaitableClass);
let promise_with_resolvers_fn =
render_args.require_intrinsic(Intrinsic::PromiseWithResolversPonyfill);
let runtime_error_class =
render_args.require_intrinsic(Intrinsic::WebAssemblyRuntimeError);
output.push_str(&format!(
r#"
class {component_async_state_class} {{
static EVENT_HANDLER_EVENTS = [ 'backpressure-change' ];
static TickResult = {{
// no suspended tasks remain
DONE: 'done',
// a suspended task was resumed (more may be ready)
RESUMED: 'resumed',
// suspended tasks remain but none were ready
IDLE: 'idle',
}};
#componentIdx;
#callingAsyncImport = false;
#syncImportWait = {promise_with_resolvers_fn}();
#lockHolderTaskID = null;
#lockWaiters = [];
#lockHandoffScheduled = false;
#parkedTasks = new Map();
#suspendedTasksByTaskID = new Map();
#suspendedTaskIDs = [];
#errored = null;
#trapped = null;
#backpressure = 0;
#backpressureWaiters = 0n;
#handlerMap = new Map();
#nextHandlerID = 0n;
#tickLoop = null;
#tickLoopInterval = null;
#onExclusiveReleaseHandlers = [];
mayLeave = true;
handles;
subtasks;
constructor(args) {{
this.#componentIdx = args.componentIdx;
this.handles = new {rep_table_class}({{ target: `component [${{this.#componentIdx}}] handles (waitable objects)` }});
this.subtasks = new {rep_table_class}({{ target: `component [${{this.#componentIdx}}] subtasks` }});
}};
componentIdx() {{ return this.#componentIdx; }}
errored() {{ return this.#errored !== null; }}
setErrored(err) {{
{debug_log_fn}('[{component_async_state_class}#setErrored()] component errored', {{ err, componentIdx: this.#componentIdx }});
if (this.#errored) {{ return; }}
if (!err) {{
err = new Error('error elswehere (see other component instance error)')
err.componentIdx = this.#componentIdx;
}}
this.#errored = err;
}}
markTrapped(err) {{
if (!(err instanceof {runtime_error_class})) {{
return false;
}}
{debug_log_fn}('[{component_async_state_class}#markTrapped()] component trapped', {{ err, componentIdx: this.#componentIdx }});
if (this.#trapped === null) {{ this.#trapped = err; }}
return true;
}}
throwIfTrapped() {{
if (this.#trapped !== null) {{ throw this.#trapped; }}
}}
callingSyncImport(val) {{
if (val === undefined) {{ return this.#callingAsyncImport; }}
if (typeof val !== 'boolean') {{ throw new TypeError('invalid setting for async import'); }}
const prev = this.#callingAsyncImport;
this.#callingAsyncImport = val;
if (prev === true && this.#callingAsyncImport === false) {{
this.#notifySyncImportEnd();
}}
}}
#notifySyncImportEnd() {{
const existing = this.#syncImportWait;
this.#syncImportWait = {promise_with_resolvers_fn}();
existing.resolve();
}}
async waitForSyncImportCallEnd() {{
await this.#syncImportWait.promise;
}}
setBackpressure(v) {{
this.#backpressure = v;
return this.#backpressure
}}
getBackpressure() {{ return this.#backpressure; }}
incrementBackpressure() {{
const current = this.#backpressure;
if (current < 0 || current > 2**16) {{
throw new Error(`invalid current backpressure value [${{current}}]`);
}}
const newValue = this.getBackpressure() + 1;
if (newValue >= 2**16) {{
throw new Error(`invalid new backpressure value [${{newValue}}], overflow`);
}}
return this.setBackpressure(newValue);
}}
decrementBackpressure() {{
const current = this.#backpressure;
if (current < 0 || current > 2**16) {{
throw new Error(`invalid current backpressure value [${{current}}]`);
}}
const newValue = Math.max(0, current - 1);
if (newValue < 0) {{
throw new Error(`invalid new backpressure value [${{newValue}}], underflow`);
}}
return this.setBackpressure(newValue);
}}
hasBackpressure() {{ return this.#backpressure > 0; }}
waitForBackpressure() {{
let backpressureCleared = false;
const cstate = this;
cstate.addBackpressureWaiter();
const handlerID = this.registerHandler({{
event: 'backpressure-change',
fn: (bp) => {{
if (bp === 0) {{
cstate.removeHandler(handlerID);
backpressureCleared = true;
}}
}}
}});
return new Promise((resolve) => {{
const interval = setInterval(() => {{
if (backpressureCleared) {{ return; }}
clearInterval(interval);
cstate.removeBackpressureWaiter();
resolve(null);
}}, 0);
}});
}}
registerHandler(args) {{
const {{ event, fn }} = args;
if (!event) {{ throw new Error("missing handler event"); }}
if (!fn) {{ throw new Error("missing handler fn"); }}
if (!{component_async_state_class}.EVENT_HANDLER_EVENTS.includes(event)) {{
throw new Error(`unrecognized event handler [${{event}}]`);
}}
const handlerID = this.#nextHandlerID++;
let handlers = this.#handlerMap.get(event);
if (!handlers) {{
handlers = [];
this.#handlerMap.set(event, handlers)
}}
handlers.push({{ id: handlerID, fn, event }});
return handlerID;
}}
removeHandler(args) {{
const {{ event, handlerID }} = args;
const registeredHandlers = this.#handlerMap.get(event);
if (!registeredHandlers) {{ return; }}
const found = registeredHandlers.find(h => h.id === handlerID);
if (!found) {{ return; }}
this.#handlerMap.set(event, this.#handlerMap.get(event).filter(h => h.id !== handlerID));
}}
getBackpressureWaiters() {{ return this.#backpressureWaiters; }}
addBackpressureWaiter() {{ this.#backpressureWaiters++; }}
removeBackpressureWaiter() {{
this.#backpressureWaiters--;
if (this.#backpressureWaiters < 0) {{
throw new Error("unexepctedly negative number of backpressure waiters");
}}
}}
// The per-slice mutual-exclusion lock for guest execution in this
// component instance. Guest slices (callback invocations and
// sync-lifted bodies) must be atomic per component even across the
// JSPI suspensions jco introduces for host imports: wit-bindgen's
// executors publish per-task state in single linear-memory cells
// (the wasip3-task pointer, context-local storage discipline) that
// an interleaved slice of the same component corrupts
//
// The lock is *owned*: acquisition records the holder task and
// release is a no-op for anyone else, so a task exiting can no
// longer drop a hold it does not own (blind acquire/release-any
// was the previous discipline). Contended acquisition queues
// FIFO; release hands the lock to the next waiter directly.
isExclusivelyLocked() {{ return this.#lockHolderTaskID !== null; }}
exclusivelyLockedBy(taskID) {{ return this.#lockHolderTaskID === taskID; }}
exclusiveLock(taskID) {{
{debug_log_fn}('[{component_async_state_class}#exclusiveLock()]', {{
holder: this.#lockHolderTaskID,
requester: taskID,
componentIdx: this.#componentIdx,
}});
if (taskID === undefined || taskID === null) {{
throw new Error('exclusive lock requires the acquiring task id');
}}
if (this.#lockHolderTaskID !== null) {{
throw new Error(`component [${{this.#componentIdx}}] exclusive lock held by task [${{this.#lockHolderTaskID}}], requested by [${{taskID}}]`);
}}
this.#lockHolderTaskID = taskID;
}}
// Awaitable acquisition: takes the lock immediately when free,
// otherwise queues FIFO behind the current holder and earlier
// waiters. The resolved promise implies ownership.
async acquireExclusiveLock(taskID) {{
if (taskID === undefined || taskID === null) {{
throw new Error('exclusive lock requires the acquiring task id');
}}
if (this.#lockHolderTaskID === null) {{
this.#lockHolderTaskID = taskID;
{debug_log_fn}('[{component_async_state_class}#acquireExclusiveLock()] acquired', {{
holder: taskID,
componentIdx: this.#componentIdx,
}});
return;
}}
if (this.#lockHolderTaskID === taskID) {{
throw new Error(`task [${{taskID}}] already holds the lock for component [${{this.#componentIdx}}]`);
}}
{debug_log_fn}('[{component_async_state_class}#acquireExclusiveLock()] waiting', {{
holder: this.#lockHolderTaskID,
requester: taskID,
componentIdx: this.#componentIdx,
queued: this.#lockWaiters.length,
}});
await new Promise((resolve) => {{
this.#lockWaiters.push({{ taskID, resolve }});
}});
}}
exclusiveRelease(taskID) {{
{debug_log_fn}('[{component_async_state_class}#exclusiveRelease()] args', {{
holder: this.#lockHolderTaskID,
releaser: taskID,
componentIdx: this.#componentIdx,
}});
if (this.#lockHolderTaskID !== taskID) {{
// Ownerless releases were the historical behavior; a foreign
// release now leaves the hold intact
{debug_log_fn}('[{component_async_state_class}#exclusiveRelease()] ignoring foreign release', {{
holder: this.#lockHolderTaskID,
releaser: taskID,
componentIdx: this.#componentIdx,
}});
return false;
}}
// Make the release observable before handing the lock to the next
// asynchronous guest slice.
//
// Release handlers may expose a lifted value whose consumer immediately
// performs a synchronous call on the same component; that call must run
// while the instance is genuinely unlocked, not via enterSync's
// lock-free fallback code.
this.#lockHolderTaskID = null;
this.#onExclusiveReleaseHandlers = this.#onExclusiveReleaseHandlers.filter(v => !!v);
for (const [idx, f] of this.#onExclusiveReleaseHandlers.entries()) {{
try {{
this.#onExclusiveReleaseHandlers[idx] = null;
f();
}} catch (err) {{
{debug_log_fn}("error while executing handler for next exclusive release", err);
throw err;
}}
}}
this.#scheduleLockHandoff();
return true;
}}
#scheduleLockHandoff() {{
if (this.#lockHandoffScheduled || this.#lockWaiters.length === 0) {{ return; }}
this.#lockHandoffScheduled = true;
queueMicrotask(() => {{
this.#lockHandoffScheduled = false;
// A synchronous call triggered by a release handler gets the
// first opportunity to use the unlocked component.
//
// Its release will leave this queued handoff in place.
if (this.#lockHolderTaskID !== null) {{
this.#scheduleLockHandoff();
return;
}}
const next = this.#lockWaiters.shift();
if (!next) {{ return; }}
this.#lockHolderTaskID = next.taskID;
next.resolve();
}});
}}
onNextExclusiveRelease(fn) {{
{debug_log_fn}('[{component_async_state_class}#()onNextExclusiveRelease] registering');
this.#onExclusiveReleaseHandlers.push(fn);
}}
async waitForExclusiveRelease() {{
while (this.isExclusivelyLocked()) {{
await new Promise(resolve => this.onNextExclusiveRelease(resolve));
}}
}}
#getSuspendedTaskMeta(taskID) {{
return this.#suspendedTasksByTaskID.get(taskID);
}}
#removeSuspendedTaskMeta(taskID) {{
{debug_log_fn}('[{component_async_state_class}#removeSuspendedTaskMeta()] removing suspended task', {{
taskID,
componentIdx: this.#componentIdx,
}});
const idx = this.#suspendedTaskIDs.findIndex(t => t === taskID);
const meta = this.#suspendedTasksByTaskID.get(taskID);
this.#suspendedTaskIDs[idx] = null;
this.#suspendedTasksByTaskID.delete(taskID);
return meta;
}}
#addSuspendedTaskMeta(meta) {{
if (!meta) {{ throw new Error('missing task meta'); }}
const taskID = meta.taskID;
this.#suspendedTasksByTaskID.set(taskID, meta);
this.#suspendedTaskIDs.push(taskID);
if (this.#suspendedTasksByTaskID.size < this.#suspendedTaskIDs.length - 10) {{
this.#suspendedTaskIDs = this.#suspendedTaskIDs.filter(t => t !== null);
}}
}}
// TODO(threads): readyFn is normally on the thread
suspendTask(args) {{
const {{ task, readyFn }} = args;
const taskID = task.id();
const componentIdx = task.componentIdx();
{debug_log_fn}('[{component_async_state_class}#suspendTask()]', {{
taskID,
componentIdx: this.#componentIdx,
taskEntryFnName: task.entryFnName(),
subtask: task.getParentSubtask(),
}});
if (componentIdx !== this.#componentIdx) {{
throw new Error('assert: task component idx should match async state');
}}
if (this.#getSuspendedTaskMeta(taskID)) {{
throw new Error(`task [${{taskID}}] already suspended`);
}}
const {{ promise, resolve, reject }} = {promise_with_resolvers_fn}();
this.#addSuspendedTaskMeta({{
task,
taskID,
readyFn,
resume: () => {{
{debug_log_fn}('[{component_async_state_class}] resuming suspended task', {{
taskID,
componentIdx: this.#componentIdx,
}});
// TODO(threads): it's thread cancellation we should be checking for below, not task
resolve(!task.isCancelled());
}},
}});
this.runTickLoop();
return promise;
}}
resumeTaskByID(taskID) {{
const meta = this.#removeSuspendedTaskMeta(taskID);
if (!meta) {{ return; }}
if (meta.taskID !== taskID) {{ throw new Error('task ID does not match'); }}
meta.resume();
}}
async runTickLoop() {{
if (this.#tickLoop !== null) {{ return; }}
this.#tickLoop = 1;
setTimeout(async () => {{
let result = this.tick();
while (result !== {component_async_state_class}.TickResult.DONE) {{
// After resuming a task, re-tick as soon as the resumed
// slice's microtask continuations have drained (timeout 0)
// so queued sibling resumptions aren't charged the idle
// polling interval; otherwise poll at the idle cadence.
const delay = result === {component_async_state_class}.TickResult.RESUMED ? 0 : 10;
await new Promise((resolve) => setTimeout(resolve, delay));
result = this.tick();
}}
this.#tickLoop = null;
}}, 10);
}}
tick() {{
// {debug_log_fn}('[{component_async_state_class}#tick()]', {{ suspendedTaskIDs: this.#suspendedTaskIDs }});
const resumableTasks = this.#suspendedTaskIDs.filter(t => t !== null);
for (const taskID of resumableTasks) {{
const meta = this.#suspendedTasksByTaskID.get(taskID);
if (!meta || !meta.readyFn) {{
throw new Error(`missing/invalid task despite ID [${{taskID}}] being present`);
}}
// If the task failed via any means, allow the task to resume because
// it's been cancelled -- the callback should immediately exit as well
if (meta.task.isRejected()) {{
{debug_log_fn}('[{component_async_state_class}#tick()] detected task rejection, leaving early', {{ meta }});
this.resumeTaskByID(taskID);
return {component_async_state_class}.TickResult.RESUMED;
}}
const isReady = meta.readyFn();
if (!isReady) {{ continue; }}
{debug_log_fn}('[{component_async_state_class}#tick()] resuming task via tick', {{
taskID,
componentIdx: this.#componentIdx,
}});
this.resumeTaskByID(taskID);
// NOTE: during single-flight resumption, we should resume at most one task per
// tick so that the resumed slice (a microtask continuation)
// runs -- and its current-task register window opens and
// closes -- before any sibling task of this component is
// resumed.
//
// Resuming multiple suspended tasks in one synchronous
// cascade interleaves their register save/restore windows
// ([restoreA, restoreB, resumeA, resumeB]), re-entering wasm
// with the register naming the wrong task, and the
// 'known residual' of the JSPI current-task register
// fix); with concurrent task lifetimes per component this
// corrupts guest context-local storage.
return {component_async_state_class}.TickResult.RESUMED;
}}
const idle = this.#suspendedTaskIDs.filter(t => t !== null).length > 0;
return idle
? {component_async_state_class}.TickResult.IDLE
: {component_async_state_class}.TickResult.DONE;
}}
createWaitable(args) {{
return new {waitable_class}({{ target: args?.target, }});
}}
}}
"#,
));
}
Self::GetOrCreateAsyncState => {
let get_state_fn = render_args.require_intrinsic(Self::GetOrCreateAsyncState);
let async_state_map = render_args.require_intrinsic(Self::GlobalAsyncStateMap);
let component_async_state_class =
render_args.require_intrinsic(Self::ComponentAsyncStateClass);
output.push_str(&format!(
r#"
function {get_state_fn}(componentIdx, init) {{
if (!{async_state_map}.has(componentIdx)) {{
const newState = new {component_async_state_class}({{ componentIdx }});
{async_state_map}.set(componentIdx, newState);
}}
return {async_state_map}.get(componentIdx);
}}
"#
));
}
Self::ComponentStateSetAllError => {
let debug_log_fn = render_args.require_intrinsic(Intrinsic::DebugLog);
let async_state_map = render_args.require_intrinsic(Self::GlobalAsyncStateMap);
let component_state_set_all_error_fn =
render_args.require_intrinsic(Self::ComponentStateSetAllError);
output.push_str(&format!(
r#"
function {component_state_set_all_error_fn}() {{
{debug_log_fn}('[{component_state_set_all_error_fn}()]');
for (const state of {async_state_map}.values()) {{
state.setErrored();
}}
}}
"#
));
}
}
}
}