diff --git a/fibers_async.js b/fibers_async.js index 966ea58..700d778 100644 --- a/fibers_async.js +++ b/fibers_async.js @@ -10,7 +10,8 @@ const asyncResourceWeakMap = new WeakMap(); function Fiber(fn, ...args) { const ar = new AsyncResource('Fiber'); const actualFn = (...args1) => ar.runInAsyncScope(() => { - Fiber.current._meteor_dynamics = undefined; + // _Fiber.current: Fiber.current hides the fibers that run code after an await. + _Fiber.current._meteor_dynamics = undefined; // return the fiber function's value so run() resolves to it when the fiber finishes (fibers README semantics) return fn(...args1); }); @@ -22,7 +23,20 @@ function Fiber(fn, ...args) { Fiber.__proto__ = _Fiber; Fiber.prototype = _Fiber.prototype; +// Fibers that run promise reactions (see below) are hidden: `Fiber.current` reports them as +// undefined, exactly what code after an `await` sees today, so `Fiber.current`-based mode switches +// ("in a fiber? return sync, else return a Promise") keep behaving as before. Blocking primitives +// (Future.wait, Promise.await) use `Fiber.currentIncludingHidden` so they can still yield. +const kHidden = Symbol.for('fibers.hiddenAwaitContinuation'); + Object.defineProperty(Fiber, 'current', { + get() { + const current = _Fiber.current; + return current && current[kHidden] ? undefined : current; + } +}) + +Object.defineProperty(Fiber, 'currentIncludingHidden', { get() { return _Fiber.current; } @@ -51,4 +65,97 @@ _Fiber[Symbol.hasInstance] = function(obj) { return obj instanceof Fiber || obj.run; }; +// Prototype (Qualia): with a patched node, V8 offers us every promise reaction (e.g. the code after +// an `await`) that was registered inside a fiber. Run it in a hidden fiber so it can still block. +// The job's own promise hooks restore its async context inside the fiber. +// FIBERS_AWAIT_DISPATCH=0 turns this off. +_Fiber.awaitDispatchEnabled = false; +if (typeof _Fiber.__setMicrotaskDispatcher === 'function' && process.env.FIBERS_AWAIT_DISPATCH !== '0' + && !asyncContextFrameInUse()) { + _Fiber.awaitDispatchEnabled = true; + const runNextDispatchable = process.env.FIBERS_AWAIT_REUSE !== '0' ? _Fiber.__runNextDispatchable : undefined; + _Fiber.__setMicrotaskDispatcher(typeof runNextDispatchable === 'function' + ? reusingDispatcher(runNextDispatchable) + : dispatchOnNewFiber); +} + +// Node 24's AsyncContextFrame (on by default) keeps AsyncLocalStorage state in the V8 slot we tag +// reactions with, so dispatch needs `--no-async-context-frame` (fibers needs it on node 24 anyway). +// Quiet unless FIBERS_AWAIT_DISPATCH=1 asks for dispatch: helper processes (build tools, child +// processes) load fibers without the flag. Fiber.awaitDispatchEnabled tells you which way it went. +function asyncContextFrameInUse() { + if (typeof _Fiber.__cpedInUse !== 'function') { + return false; + } + const { AsyncLocalStorage } = require('async_hooks'); + const probe = new AsyncLocalStorage(); + const inUse = probe.run(true, () => _Fiber.__cpedInUse()); + // Without AsyncContextFrame, an AsyncLocalStorage turns on promise hooks for the whole process + // (every promise pays for them); disabling the probe turns them off again if nothing else uses one. + probe.disable(); + if (inUse && process.env.FIBERS_AWAIT_DISPATCH === '1') { + console.warn('[fibers] not running await continuations on fibers: AsyncContextFrame is on (start node with --no-async-context-frame)'); + } + return inUse; +} + +function dispatchOnNewFiber(runDispatched, token) { + const fiber = Fiber(() => runDispatched(token)); + fiber[kHidden] = true; + fiber.run(); +} + +// Runs jobs on one long-lived hidden fiber instead of a new fiber per job. After each job it keeps +// running the dispatchable jobs queued right behind it, so a run of `await`s in fibered code costs +// one switch onto the fiber and one back, not two per `await`. A job that blocks keeps the fiber +// (it's parked inside that job), and the next dispatch starts a new one. FIBERS_AWAIT_REUSE=0 turns +// this off. +// Kept on the native module, which every copy of this file shares (an app can load more than one). +_Fiber.__microtasksBatched = _Fiber.__microtasksBatched || 0; +function reusingDispatcher(runNextDispatchable) { + const kIdle = {}; + let dispatchFiber = null; + let runDispatched; + + function dispatchLoop(token) { + const self = _Fiber.current; + for (;;) { + runDispatched(token); + // Only while this is still the dispatch fiber, i.e. we're nested inside the microtask loop. + // A fiber that was parked by a job and later resumed must not pull in unrelated jobs. + while (self === dispatchFiber && runNextDispatchable()) { + _Fiber.__microtasksBatched++; + } + if (self !== dispatchFiber) { + return; + } + token = _Fiber.yield(kIdle); + } + } + + return function dispatchMicrotask(run, token) { + runDispatched = run; + if (dispatchFiber && dispatchFiber === _Fiber.current) { + // A nested microtask drain on the dispatch fiber itself; we can't switch into a running fiber. + dispatchOnNewFiber(run, token); + return; + } + if (!dispatchFiber) { + dispatchFiber = Fiber(dispatchLoop); + dispatchFiber[kHidden] = true; + } + const fiber = dispatchFiber; + if (fiber.run(token) !== kIdle && dispatchFiber === fiber) { + // A job blocked on this fiber; it belongs to that job now. + dispatchFiber = null; + } + }; +} + +Object.defineProperty(Fiber, 'microtasksBatched', { + get() { + return _Fiber.__microtasksBatched || 0; + } +}) + module.exports = _Fiber.Fiber = Fiber; diff --git a/future.js b/future.js index 8d907f9..9dcd5d5 100644 --- a/future.js +++ b/future.js @@ -154,7 +154,8 @@ Future.wait = function wait(/* ... */) { } // Resumes current fiber - var fiber = Fiber.current; + // Includes the hidden fibers that run code after an `await` (see fibers_async.js). + var fiber = Fiber.currentIncludingHidden || Fiber.current; if (!fiber) { throw new Error('Can\'t wait without a fiber. Most likely you called `Promise.await()` after calling `await somePromiseAPI()`. Read https://engdocs.qualia.io/qualia/advanced-topics/fibers'); } diff --git a/src/fibers.cc b/src/fibers.cc index cf86a51..edb34af 100644 --- a/src/fibers.cc +++ b/src/fibers.cc @@ -7,6 +7,44 @@ #include #include +// Prototype (Qualia): patched node exposes v8::SetMicrotaskDispatchCallback so the code after an +// `await` (or a `.then` callback) registered inside a fiber also runs inside a fiber. +#if defined(__has_include) +#if __has_include() +#include +#include +#define FIBERS_AWAIT_DISPATCH 1 +#endif +#endif + +#ifdef FIBERS_AWAIT_DISPATCH +// Looked up at load time so this binary still works (with the feature off) on unpatched node. +typedef void (*SetMicrotaskDispatchCallbackFn)(v8::Isolate*, v8::MicrotaskDispatchCallback, void*); +typedef void (*RunDispatchedMicrotaskFn)(v8::Isolate*, v8::DispatchedMicrotask*); +typedef bool (*RunNextDispatchableMicrotaskFn)(v8::Isolate*); +static SetMicrotaskDispatchCallbackFn set_microtask_dispatch_callback = NULL; +static RunDispatchedMicrotaskFn run_dispatched_microtask = NULL; +static RunNextDispatchableMicrotaskFn run_next_dispatchable_microtask = NULL; + +// V8 keeps the continuation-preserved embedder data (CPED) slot per context up to V8 12, per +// isolate from V8 13 (node 24). Only node 18 and node 24 are tested. +static v8::Local GetCped(v8::Isolate* isolate, v8::Local context) { +#if V8_AT_LEAST(13, 0) + return isolate->GetContinuationPreservedEmbedderData(); +#else + return context->GetContinuationPreservedEmbedderData(); +#endif +} + +static void SetCped(v8::Isolate* isolate, v8::Local context, v8::Local value) { +#if V8_AT_LEAST(13, 0) + isolate->SetContinuationPreservedEmbedderData(value); +#else + context->SetContinuationPreservedEmbedderData(value); +#endif +} +#endif + #define THROW(x, m) return uni::Return(uni::ThrowException(Isolate::GetCurrent(), x(uni::NewLatin1String(Isolate::GetCurrent(), m))), args) using namespace std; @@ -433,6 +471,12 @@ class Fiber { static Fiber* current; static vector orphaned_fibers; static Persistent fatal_stack; +#ifdef FIBERS_AWAIT_DISPATCH + static Persistent microtask_dispatcher; + static Persistent run_dispatched; + static Persistent token_template; + static double microtasks_dispatched; +#endif Isolate* isolate; Persistent handle; @@ -721,6 +765,24 @@ class Fiber { Fiber* last_fiber = current; current = this; +#ifdef FIBERS_AWAIT_DISPATCH + // While this fiber runs, the context's continuation-preserved embedder data (CPED) is the + // marker. V8 copies CPED into every promise reaction registered meanwhile, and offers + // those reactions to DispatchMicrotask() instead of running them on whatever stack is + // draining the microtask queue. + bool tag_reactions = !microtask_dispatcher.IsEmpty(); + Local cped_context; + Local previous_cped; + if (tag_reactions) { + cped_context = Local::New(isolate, v8_context); + previous_cped = GetCped(isolate, cped_context); + // The marker is `true`: it lives in V8's read-only space and never moves. On V8 13 the + // slot is a GC root visited once per archived thread (one per fiber), and visiting a + // young object more than once in a scavenge corrupts the heap. + SetCped(isolate, cped_context, v8::True(isolate)); + } +#endif + // This will jump into either `RunFiber()` or `Yield()`, depending on if the fiber was // already running. { @@ -729,6 +791,12 @@ class Fiber { this_fiber->run(); } +#ifdef FIBERS_AWAIT_DISPATCH + if (tag_reactions) { + SetCped(isolate, cped_context, previous_cped); + } +#endif + // At this point the fiber either returned or called `yield()`. current = last_fiber; } @@ -891,6 +959,99 @@ class Fiber { return uni::Return(uni::NewNumber(Isolate::GetCurrent(), Coroutine::coroutines_created()), info); } +#ifdef FIBERS_AWAIT_DISPATCH + /** + * Called by V8 for a promise reaction (e.g. the code after an `await`) that was registered + * while a fiber was running. Hands the job to the JS dispatcher, which runs it in a fiber. + * Returns false (V8 runs the job inline) if the dispatcher didn't start it. + */ + static bool DispatchMicrotask(Isolate* isolate, v8::DispatchedMicrotask* task, void* data) { + if (microtask_dispatcher.IsEmpty()) { + return false; + } + HandleScope scope(isolate); + Local context = isolate->GetCurrentContext(); + Local token; + if (!Local::New(isolate, token_template)->NewInstance(context).ToLocal(&token)) { + return false; + } + token->SetAlignedPointerInInternalField(0, task); + Local argv[2] = { Local::New(isolate, run_dispatched), token }; + { + uni::TryCatch try_catch(isolate); + Local ignored; + if (!Local::New(isolate, microtask_dispatcher)->Call(context, Undefined(isolate), 2, argv).ToLocal(&ignored) && try_catch.HasCaught()) { + String::Utf8Value message(isolate, try_catch.Exception()); + cerr << "fibers: microtask dispatcher threw: " << (*message ? *message : "") << "\n"; + } + } + if (token->GetAlignedPointerFromInternalField(0) != NULL) { + // Never started: make the token inert and let V8 run the job inline. + token->SetAlignedPointerInInternalField(0, NULL); + return false; + } + ++microtasks_dispatched; + return true; + } + + /** + * runDispatched(token): runs the job behind `token` on the current stack. Only the first call + * does anything. + */ + static uni::FunctionType RunDispatched(const uni::Arguments& args) { + Isolate* isolate = args.GetIsolate(); + if (args.Length() != 1 || !args[0]->IsObject() || args[0].As()->InternalFieldCount() != 1) { + THROW(Exception::TypeError, "runDispatched expects a dispatch token"); + } + Local token = args[0].As(); + v8::DispatchedMicrotask* task = static_cast(token->GetAlignedPointerFromInternalField(0)); + if (task != NULL) { + token->SetAlignedPointerInInternalField(0, NULL); + run_dispatched_microtask(isolate, task); + } + return uni::Return(uni::Undefined(isolate), args); + } + + /** + * Fiber.__cpedInUse(): whether the CPED slot currently holds anything. Used to detect node 24's + * AsyncContextFrame, which keeps AsyncLocalStorage state in the same slot. + */ + static uni::FunctionType CpedInUse(const uni::Arguments& args) { + Isolate* isolate = args.GetIsolate(); + Local cped = GetCped(isolate, isolate->GetCurrentContext()); + return uni::Return(uni::NewBoolean(isolate, !cped->IsUndefined()), args); + } + + /** + * Fiber.__setMicrotaskDispatcher(fn): fn(runDispatched, token) must call + * runDispatched(token) synchronously, normally inside a new fiber. + */ + static uni::FunctionType SetMicrotaskDispatcher(const uni::Arguments& args) { + Isolate* isolate = args.GetIsolate(); + if (args.Length() != 1 || !args[0]->IsFunction()) { + THROW(Exception::TypeError, "__setMicrotaskDispatcher expects a function"); + } + microtask_dispatcher.Reset(isolate, args[0].As()); + set_microtask_dispatch_callback(isolate, DispatchMicrotask, NULL); + return uni::Return(uni::Undefined(isolate), args); + } + + /** + * Fiber.__runNextDispatchable(): if the next queued microtask is also one that would be + * dispatched, run it on the current stack and return true. Lets the dispatch fiber run a + * run of such jobs without switching stacks for each one. + */ + static uni::FunctionType RunNextDispatchable(const uni::Arguments& args) { + Isolate* isolate = args.GetIsolate(); + bool ran = run_next_dispatchable_microtask(isolate); + return uni::Return(uni::NewBoolean(isolate, ran), args); + } + + static uni::FunctionType GetMicrotasksDispatched(Local property, const uni::GetterCallbackInfo& info) { + return uni::Return(uni::NewNumber(Isolate::GetCurrent(), microtasks_dispatched), info); + } +#endif + public: /** * Initialize the Fiber library. @@ -941,6 +1102,26 @@ class Fiber { uni::SetAccessor(isolate, fn, uni::NewLatin1Symbol(isolate, "current"), GetCurrent); uni::SetAccessor(isolate, fn, uni::NewLatin1Symbol(isolate, "poolSize"), GetPoolSize, SetPoolSize); uni::SetAccessor(isolate, fn, uni::NewLatin1Symbol(isolate, "fibersCreated"), GetFibersCreated); +#ifdef FIBERS_AWAIT_DISPATCH + set_microtask_dispatch_callback = (SetMicrotaskDispatchCallbackFn)dlsym(RTLD_DEFAULT, "v8_qualia_SetMicrotaskDispatchCallback"); + run_dispatched_microtask = (RunDispatchedMicrotaskFn)dlsym(RTLD_DEFAULT, "v8_qualia_RunDispatchedMicrotask"); + if (set_microtask_dispatch_callback && run_dispatched_microtask) { + Local token_tmpl = ObjectTemplate::New(isolate); + token_tmpl->SetInternalFieldCount(1); + token_template.Reset(isolate, token_tmpl); + run_dispatched.Reset(isolate, uni::GetFunction(uni::NewFunctionTemplate(isolate, RunDispatched))); + fn->Set(context, uni::NewLatin1Symbol(isolate, "__setMicrotaskDispatcher"), + uni::GetFunction(uni::NewFunctionTemplate(isolate, SetMicrotaskDispatcher))).FromJust(); + fn->Set(context, uni::NewLatin1Symbol(isolate, "__cpedInUse"), + uni::GetFunction(uni::NewFunctionTemplate(isolate, CpedInUse))).FromJust(); + uni::SetAccessor(isolate, fn, uni::NewLatin1Symbol(isolate, "microtasksDispatched"), GetMicrotasksDispatched); + run_next_dispatchable_microtask = (RunNextDispatchableMicrotaskFn)dlsym(RTLD_DEFAULT, "v8_qualia_RunNextDispatchableMicrotask"); + if (run_next_dispatchable_microtask) { + fn->Set(context, uni::NewLatin1Symbol(isolate, "__runNextDispatchable"), + uni::GetFunction(uni::NewFunctionTemplate(isolate, RunNextDispatchable))).FromJust(); + } + } +#endif // Global Fiber target->Set(context, uni::NewLatin1Symbol(isolate, "Fiber"), fn).FromJust(); @@ -954,6 +1135,12 @@ Locker* Fiber::global_locker; Fiber* Fiber::current = NULL; vector Fiber::orphaned_fibers; Persistent Fiber::fatal_stack; +#ifdef FIBERS_AWAIT_DISPATCH +Persistent Fiber::microtask_dispatcher; +Persistent Fiber::run_dispatched; +Persistent Fiber::token_template; +double Fiber::microtasks_dispatched = 0; +#endif bool did_init = false; #if !NODE_VERSION_AT_LEAST(0,10,0)