[Prototype] Node 24: run await continuations from fibers in hidden fibers (port of #6)

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;

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');
 	}

src/fibers.cc

@@ -7,6 +7,44 @@
 #include <vector>
 #include <iostream>
 
+// 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(<v8-microtask-dispatch.h>)
+#include <v8-microtask-dispatch.h>
+#include <dlfcn.h>
+#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<v8::Value> GetCped(v8::Isolate* isolate, v8::Local<v8::Context> context) {
+#if V8_AT_LEAST(13, 0)
+	return isolate->GetContinuationPreservedEmbedderData();
+#else
+	return context->GetContinuationPreservedEmbedderData();
+#endif
+}
+
+static void SetCped(v8::Isolate* isolate, v8::Local<v8::Context> context, v8::Local<v8::Value> 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<Fiber*> orphaned_fibers;
 		static Persistent<Value> fatal_stack;
+#ifdef FIBERS_AWAIT_DISPATCH
+		static Persistent<Function> microtask_dispatcher;
+		static Persistent<Function> run_dispatched;
+		static Persistent<ObjectTemplate> token_template;
+		static double microtasks_dispatched;
+#endif
 
 		Isolate* isolate;
 		Persistent<Object> 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<Context> cped_context;
+			Local<Value> previous_cped;
+			if (tag_reactions) {
+				cped_context = Local<Context>::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> context = isolate->GetCurrentContext();
+			Local<Object> token;
+			if (!Local<ObjectTemplate>::New(isolate, token_template)->NewInstance(context).ToLocal(&token)) {
+				return false;
+			}
+			token->SetAlignedPointerInInternalField(0, task);
+			Local<Value> argv[2] = { Local<Function>::New(isolate, run_dispatched), token };
+			{
+				uni::TryCatch try_catch(isolate);
+				Local<Value> ignored;
+				if (!Local<Function>::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 : "<unknown>") << "\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<Object>()->InternalFieldCount() != 1) {
+				THROW(Exception::TypeError, "runDispatched expects a dispatch token");
+			}
+			Local<Object> token = args[0].As<Object>();
+			v8::DispatchedMicrotask* task = static_cast<v8::DispatchedMicrotask*>(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<Value> 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<Function>());
+			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<String> 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<ObjectTemplate> 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*> Fiber::orphaned_fibers;
 Persistent<Value> Fiber::fatal_stack;
+#ifdef FIBERS_AWAIT_DISPATCH
+Persistent<Function> Fiber::microtask_dispatcher;
+Persistent<Function> Fiber::run_dispatched;
+Persistent<ObjectTemplate> Fiber::token_template;
+double Fiber::microtasks_dispatched = 0;
+#endif
 bool did_init = false;
 
 #if !NODE_VERSION_AT_LEAST(0,10,0)