diff --git a/apps/nest-node-probe/correctness/set-immediate.ts b/apps/nest-node-probe/correctness/set-immediate.ts new file mode 100644 index 0000000..7b02a79 --- /dev/null +++ b/apps/nest-node-probe/correctness/set-immediate.ts @@ -0,0 +1,33 @@ +import { setImmediate as setImmediateFromTimers } from 'node:timers' + +const log = (line: string): void => { + console.log(line) +} + +log('start') +const a = setImmediate(() => { + log('immediate a') + setImmediate(() => log('immediate queued by a (next turn)')) + process.nextTick(() => log('tick from a')) + Promise.resolve().then(() => log('microtask from a')) +}) +setImmediate(() => log('immediate b')) +const c = setImmediate(() => log('immediate c (cleared)')) +clearImmediate(c) +log(`hasRef ${a.hasRef()} ${a.unref().hasRef()} ${a.ref().hasRef()}`) +setTimeout(() => { + log('timer 10ms') + setImmediate(() => log('immediate from timer')) + setTimeout(() => log('timer 0 from timer'), 0) +}, 10) +setImmediate(() => { + const d = setImmediate(() => log('immediate d (cleared from immediate b2)')) + clearImmediate(d) + log('immediate b2') +}) +setImmediateFromTimers(() => log('immediate from node:timers')) +setTimeout(() => { + setImmediate(() => log('unref immediate ran')).unref() + log('timer 20ms, last referenced work') +}, 20) +log('end') diff --git a/plugin/intrinsics.mjs b/plugin/intrinsics.mjs index a12ac5e..c9efca1 100644 --- a/plugin/intrinsics.mjs +++ b/plugin/intrinsics.mjs @@ -482,6 +482,26 @@ export const intrinsics = { returnType: 'void', decl: 'void __gea_node_clear_timer(double id);' }, + __gea_node_immediate_start: { + emit: '__gea_node_immediate_start', + returnType: 'double', + decl: 'double __gea_node_immediate_start(std::function callback);' + }, + __gea_node_immediate_clear: { + emit: '__gea_node_immediate_clear', + returnType: 'void', + decl: 'void __gea_node_immediate_clear(double id);' + }, + __gea_node_immediate_ref: { + emit: '__gea_node_immediate_ref', + returnType: 'void', + decl: 'void __gea_node_immediate_ref(double id);' + }, + __gea_node_immediate_unref: { + emit: '__gea_node_immediate_unref', + returnType: 'void', + decl: 'void __gea_node_immediate_unref(double id);' + }, __gea_node_process_env: { emit: 'gea_cpp_process_env', returnType: 'gea_cpp_value' diff --git a/runtime/gea_node.cpp b/runtime/gea_node.cpp index 3968c6a..3ca7987 100644 --- a/runtime/gea_node.cpp +++ b/runtime/gea_node.cpp @@ -408,6 +408,77 @@ inline void fireDue() { } // namespace gea::node::timers +namespace gea::node::immediates { + +struct Immediate { + double id; + bool referenced; + std::function callback; +}; + +inline std::vector &queue() { + static std::vector immediates; + return immediates; +} + +inline double nextImmediateId() { + static double id = 0; + return ++id; +} + +inline bool hasPending() { return !queue().empty(); } +inline bool hasReferenced() { + for (const Immediate &immediate : queue()) + if (immediate.referenced) return true; + return false; +} + +inline double add(std::function callback) { + const double id = nextImmediateId(); + queue().push_back(Immediate{id, true, std::move(callback)}); + return id; +} + +inline std::unordered_set &running() { + static std::unordered_set ids; + return ids; +} + +inline void remove(double id) { + running().erase(id); + auto &immediates = queue(); + immediates.erase( + std::remove_if(immediates.begin(), immediates.end(), [&](const Immediate &immediate) { return immediate.id == id; }), + immediates.end()); +} + +inline void setReferenced(double id, bool referenced) { + for (Immediate &immediate : queue()) { + if (immediate.id == id) { + immediate.referenced = referenced; + return; + } + } +} + +inline void runAll(bool loopAlive) { + if (queue().empty()) return; + if (!loopAlive && !hasReferenced()) return; + std::vector batch; + batch.swap(queue()); + gea::detail::HostCallbackScope scope; + running().clear(); + for (const Immediate &immediate : batch) running().insert(immediate.id); + for (Immediate &immediate : batch) { + if (running().erase(immediate.id) == 0) continue; + if (immediate.callback) immediate.callback(); + gea::node::drain_microtasks(); + } + running().clear(); +} + +} + // --------------------------------------------------------------------------- // Size-class free-list allocator (global operator new/delete override). // @@ -1070,7 +1141,7 @@ class EventLoop { }); const bool timersPending = gea::node::timers::hasPending(); if ((pfds.empty() || (done == nullptr && !haveReferencedWatcher)) && - !gea::node::timers::hasReferenced()) { + !gea::node::timers::hasReferenced() && !gea::node::immediates::hasReferenced()) { purgeClosedIfSafe(); return stopRequested(done); // only unref work remains } @@ -1081,7 +1152,7 @@ class EventLoop { // times per request, re-tracing the request's live objects each time. gea::collectCyclesAtQuiescence(); if (!gea::node::microtasks().empty() || !gea::node::next_ticks().empty()) continue; - const int timeout = timersPending ? gea::node::timers::nextDelayMs() : -1; + const int timeout = gea::node::immediates::hasReferenced() ? 0 : timersPending ? gea::node::timers::nextDelayMs() : -1; const int events = platform::poll(pfds.data(), pfds.size(), timeout); if (events < 0) { if (errno == EINTR) continue; @@ -1116,6 +1187,7 @@ class EventLoop { // Handler promises progress here: continuations queued during dispatch // (async handlers, body-delivery microtasks) run before the next poll. gea::node::drain_microtasks(); + gea::node::immediates::runAll(haveReferencedWatcher || gea::node::timers::hasReferenced()); const bool stopped = stopRequested(done); --dispatchDepth_; purgeClosedIfSafe(); @@ -1155,14 +1227,14 @@ class EventLoop { }); const bool timersPending = gea::node::timers::hasPending(); if ((!haveWatchers || (done == nullptr && !haveReferencedWatcher)) && - !gea::node::timers::hasReferenced()) { + !gea::node::timers::hasReferenced() && !gea::node::immediates::hasReferenced()) { purgeClosedIfSafe(); return stopRequested(done); } gea::collectCyclesAtQuiescence(); // see `runPoll` if (!gea::node::microtasks().empty() || !gea::node::next_ticks().empty()) continue; - const int timeout = timersPending ? gea::node::timers::nextDelayMs() : -1; + const int timeout = gea::node::immediates::hasReferenced() ? 0 : timersPending ? gea::node::timers::nextDelayMs() : -1; const int count = ::epoll_wait(epollFd_, events, 128, timeout); if (count < 0) { if (errno == EINTR) continue; @@ -1192,6 +1264,7 @@ class EventLoop { if (stopRequested(done)) break; } gea::node::drain_microtasks(); + gea::node::immediates::runAll(haveReferencedWatcher || gea::node::timers::hasReferenced()); const bool stopped = stopRequested(done); --dispatchDepth_; purgeClosedIfSafe(); @@ -1233,7 +1306,7 @@ class EventLoop { }); const bool timersPending = gea::node::timers::hasPending(); if ((!haveWatchers || (done == nullptr && !haveReferencedWatcher)) && - !gea::node::timers::hasReferenced()) { + !gea::node::timers::hasReferenced() && !gea::node::immediates::hasReferenced()) { applyKqueueChanges(changes); purgeClosedIfSafe(); return stopRequested(done); @@ -1245,7 +1318,7 @@ class EventLoop { continue; } struct timespec wait{}; - const int delay = timersPending ? gea::node::timers::nextDelayMs() : -1; + const int delay = gea::node::immediates::hasReferenced() ? 0 : timersPending ? gea::node::timers::nextDelayMs() : -1; if (delay >= 0) { wait.tv_sec = delay / 1000; wait.tv_nsec = static_cast(delay % 1000) * 1000000L; @@ -1300,6 +1373,7 @@ class EventLoop { if (stopRequested(done)) break; } gea::node::drain_microtasks(); + gea::node::immediates::runAll(haveReferencedWatcher || gea::node::timers::hasReferenced()); const bool stopped = stopRequested(done); --dispatchDepth_; purgeClosedIfSafe(); @@ -4055,6 +4129,17 @@ inline void __gea_node_clear_timer(double id) { gea::node::timers::remove(id); } inline void __gea_node_timer_unref(double id) { gea::node::timers::unref(id); } +inline double __gea_node_immediate_start(std::function callback) { + if (!callback) return 0; + return gea::node::immediates::add(std::move(callback)); +} + +inline void __gea_node_immediate_clear(double id) { gea::node::immediates::remove(id); } + +inline void __gea_node_immediate_ref(double id) { gea::node::immediates::setReferenced(id, true); } + +inline void __gea_node_immediate_unref(double id) { gea::node::immediates::setReferenced(id, false); } + inline void __gea_node_run_pending() { gea::node::clientEventLoop().run(); } inline bool __gea_node_pump_until(const std::function &done) { diff --git a/runtime/node/generated/facades/timers.ts b/runtime/node/generated/facades/timers.ts index f9507ee..c57b752 100644 --- a/runtime/node/generated/facades/timers.ts +++ b/runtime/node/generated/facades/timers.ts @@ -1,16 +1,5 @@ // Generated from the pinned Node 24 declaration inventory. Do not edit. // Canonical builtin: node:timers -import { nodeNotImplemented } from "../../not-implemented" -export { clearInterval, clearTimeout, setInterval, setTimeout } from "../../timers" - -export function clearImmediate(...args: unknown[]): never { - void args - return nodeNotImplemented("node:timers", "clearImmediate") -} +export { clearImmediate, clearInterval, clearTimeout, setImmediate, setInterval, setTimeout } from "../../timers" export const promises: unknown = undefined - -export function setImmediate(...args: unknown[]): never { - void args - return nodeNotImplemented("node:timers", "setImmediate") -} diff --git a/runtime/node/generated/node24-compatibility.json b/runtime/node/generated/node24-compatibility.json index 22cc2ae..0cf05ca 100644 --- a/runtime/node/generated/node24-compatibility.json +++ b/runtime/node/generated/node24-compatibility.json @@ -18096,7 +18096,11 @@ { "name": "clearImmediate", "kind": "function", - "status": "stubbed" + "status": "partial", + "overlay": "timers.ts", + "supportedCallSignatures": [ + "(immediate: NodeJS.Immediate | undefined): void" + ] }, { "name": "clearInterval", @@ -18126,7 +18130,12 @@ { "name": "setImmediate", "kind": "function", - "status": "stubbed" + "status": "partial", + "overlay": "timers.ts", + "supportedCallSignatures": [ + "(callback: (...args: TArgs) => void, ...args: TArgs): NodeJS.Immediate", + "(callback: (_: void) => void): NodeJS.Immediate" + ] }, { "name": "setInterval", diff --git a/runtime/node/global-timers.ts b/runtime/node/global-timers.ts index c4dbd5a..3b6b449 100644 --- a/runtime/node/global-timers.ts +++ b/runtime/node/global-timers.ts @@ -44,6 +44,14 @@ declare function __gea_node_timer_start_interval(callback: () => void, delay: nu declare function __gea_node_timer_clear(id: number): void /** @gea-host-inert */ declare function __gea_node_timer_unref(id: number): void +/** @gea-host-no-property-writes */ +declare function __gea_node_immediate_start(callback: () => void): number +/** @gea-host-inert */ +declare function __gea_node_immediate_clear(id: number): void +/** @gea-host-inert */ +declare function __gea_node_immediate_ref(id: number): void +/** @gea-host-inert */ +declare function __gea_node_immediate_unref(id: number): void class Timeout { private id_: number @@ -112,3 +120,50 @@ function clearTimeout(handle: Timeout | number | null | undefined): void { function clearInterval(handle: Timeout | number | null | undefined): void { clearTimeout(handle) } + +class Immediate { + private id_: number + private referenced_: boolean + + constructor(id: number) { + this.id_ = id + this.referenced_ = true + } + + id(): number { + return this.id_ + } + + ref(): Immediate { + if (!this.referenced_) { + this.referenced_ = true + __gea_node_immediate_ref(this.id_) + } + return this + } + + unref(): Immediate { + if (this.referenced_) { + this.referenced_ = false + __gea_node_immediate_unref(this.id_) + } + return this + } + + hasRef(): boolean { + return this.referenced_ + } +} + +function setImmediate(callback: (...args: any[]) => void): Immediate { + return new Immediate( + __gea_node_immediate_start((): void => { + callback() + }) + ) +} + +function clearImmediate(handle: Immediate | null | undefined): void { + if (handle === null || handle === undefined) return + __gea_node_immediate_clear(handle.id()) +} diff --git a/runtime/node/node-globals.ts b/runtime/node/node-globals.ts index 7fb11f5..b89a57d 100644 --- a/runtime/node/node-globals.ts +++ b/runtime/node/node-globals.ts @@ -180,6 +180,7 @@ declare global { // package's unrelated object graph. type BufferConstructor = globalThis.BufferConstructor type Timeout = import('./timers.js').Timeout + type Immediate = import('./timers.js').Immediate type WriteStream = import('./process.js').WriteStream type Platform = typeof import('./process.js').platform diff --git a/runtime/node/timers.ts b/runtime/node/timers.ts index ed432c0..82eef39 100644 --- a/runtime/node/timers.ts +++ b/runtime/node/timers.ts @@ -18,15 +18,23 @@ const TimeoutAlias = Timeout type TimeoutAlias = Timeout +const ImmediateAlias = Immediate +type ImmediateAlias = Immediate + const setTimeoutAlias = setTimeout const setIntervalAlias = setInterval const clearTimeoutAlias = clearTimeout const clearIntervalAlias = clearInterval +const setImmediateAlias = setImmediate +const clearImmediateAlias = clearImmediate export { TimeoutAlias as Timeout, + ImmediateAlias as Immediate, setTimeoutAlias as setTimeout, setIntervalAlias as setInterval, clearTimeoutAlias as clearTimeout, - clearIntervalAlias as clearInterval + clearIntervalAlias as clearInterval, + setImmediateAlias as setImmediate, + clearImmediateAlias as clearImmediate } diff --git a/scripts/test-set-immediate.mjs b/scripts/test-set-immediate.mjs new file mode 100644 index 0000000..47a556d --- /dev/null +++ b/scripts/test-set-immediate.mjs @@ -0,0 +1,26 @@ +import assert from 'node:assert/strict' +import { execFileSync, spawnSync } from 'node:child_process' +import { mkdirSync } from 'node:fs' +import { resolve } from 'node:path' + +const root = resolve(import.meta.dirname, '..') +const output = resolve(root, 'apps/nest-node-probe/dist') +const source = resolve(root, 'apps/nest-node-probe/correctness/set-immediate.ts') +const executable = resolve(output, 'set-immediate') +const options = { cwd: root, encoding: 'utf8', env: { ...process.env, TMPDIR: output } } +mkdirSync(output, { recursive: true }) + +execFileSync(process.execPath, [resolve(root, 'scripts/build.mjs'), source, '--out', output, '--exe', executable], { + ...options, + stdio: 'inherit' +}) +const expected = spawnSync(process.execPath, [source], { ...options, timeout: 15000 }) +const actual = spawnSync(executable, [], { ...options, timeout: 15000 }) +assert.equal(expected.status, 0, `node status\n${expected.stderr}`) +assert.equal(actual.status, 0, `compiled status\n${actual.stderr}`) +const expectedLines = expected.stdout.trim().split('\n') +const actualLines = actual.stdout.trim().split('\n') +assert.equal(actualLines.length, expectedLines.length, `line count\n--- node\n${expected.stdout}\n--- compiled\n${actual.stdout}`) +for (let i = 0; i < expectedLines.length; i++) assert.equal(actualLines[i], expectedLines[i], `line ${i + 1}`) +assert.ok(expectedLines.includes('immediate queued by a (next turn)') && !expectedLines.includes('unref immediate ran'), 'the oracle covered the phases') +console.log(`SET_IMMEDIATE_PARITY_OK:${expectedLines.length}`) diff --git a/tools/node-api/generate-surface.mjs b/tools/node-api/generate-surface.mjs index e516570..de8154e 100644 --- a/tools/node-api/generate-surface.mjs +++ b/tools/node-api/generate-surface.mjs @@ -47,7 +47,7 @@ const internalOverlayExports = new Map( 'node:process': ['WriteStream'], 'node:stream': ['duplexToWeb', 'finishedPromise', 'pipelinePromise'], 'node:stream/promises': ['nodeStreamPromiseFinished', 'nodeStreamPromisePipeline'], - 'node:timers': ['Timeout'] + 'node:timers': ['Timeout', 'Immediate'] }).map(([moduleName, names]) => [moduleName, new Set(names)]) ) const overlayPublicExportTargets = new Map([