async: Async\Coroutine::current()/resume() — engine tracks current from intercept binding, deferred wake

Author: EdmondDantes

Zend/tests/async/coroutine_current_resume.phpt

@@ -0,0 +1,57 @@
+--TEST--
+Async\Coroutine: current() tracks the running coroutine; resume() defers a wake
+--FILE--
+<?php
+Async\SchedulerHook::register('test', new class extends Async\AbstractScheduler {
+    public SplQueue $ready;
+    public function __construct() { $this->ready = new SplQueue(); }
+    public function interceptFiber(Fiber $fiber): ?object {
+        return new class($fiber) {
+            public function __construct(public readonly Fiber $fiber) {}
+        };
+    }
+    public function enqueue(object $coroutine): bool {
+        $this->ready->enqueue($coroutine);
+        return true;
+    }
+    public function suspend(bool $fromMain, bool $isBailout): bool {
+        if (!$fromMain) {
+            return true;
+        }
+        while (!$this->ready->isEmpty()) {
+            $fiber = $this->ready->dequeue()->fiber;
+            $fiber->isStarted() ? $fiber->resume() : $fiber->start();
+        }
+        return true;
+    }
+});
+
+// No coroutine is current in the main flow.
+var_dump(Async\Coroutine::current());
+
+$waiter = null;
+
+$w = new Fiber(function () use (&$waiter): void {
+    $waiter = Async\Coroutine::current();     // the running coroutine object
+    echo "waiter: is-coroutine=", var_export($waiter !== null, true), "\n";
+    Fiber::suspend();                          // await
+    echo "waiter: resumed\n";
+});
+
+$r = new Fiber(function () use (&$waiter): void {
+    echo "resolver: resume waiter\n";
+    Async\Coroutine::resume($waiter);          // deferred wake (no immediate switch)
+    echo "resolver: done\n";
+});
+
+$w->start();
+$r->start();
+echo "main: end\n";
+?>
+--EXPECT--
+NULL
+main: end
+waiter: is-coroutine=true
+resolver: resume waiter
+resolver: done
+waiter: resumed

Zend/zend_fibers.c

@@ -975,9 +975,16 @@ static void zend_fiber_scheduler_switch(zend_fiber *fiber, zval *return_value)
 	const bool saved_context = ZEND_ASYNC_IN_SCHEDULER_CONTEXT;
 	ZEND_ASYNC_IN_SCHEDULER_CONTEXT = false;
 
+	/* The coroutine bound to this fiber (via the scheduler's intercept-fiber
+	 * return) is the current one for the duration of its body. Restored when it
+	 * suspends and hands control back here. */
+	zend_coroutine_t *saved_coroutine = ZEND_ASYNC_CURRENT_COROUTINE;
+	ZEND_ASYNC_CURRENT_COROUTINE = coroutine;
+
 	zend_fiber_transfer transfer = zend_fiber_resume_internal(
 			fiber, Z_ISUNDEF(value) ? NULL : &value, is_error);
 
+	ZEND_ASYNC_CURRENT_COROUTINE = saved_coroutine;
 	ZEND_ASYNC_IN_SCHEDULER_CONTEXT = saved_context;
 
 	zval_ptr_dtor(&value);

Zend/zend_scheduler_hook.c

@@ -579,11 +579,87 @@ ZEND_METHOD(Async_AbstractScheduler, contextUnset)
 	RETURN_FALSE;
 }
 
+/////////////////////////////////////////////////////////////////////
+/// Async\Coroutine
+/////////////////////////////////////////////////////////////////////
+
+ZEND_METHOD(Async_Coroutine, current)
+{
+	ZEND_PARSE_PARAMETERS_NONE();
+
+	zend_coroutine_t *coro = ZEND_ASYNC_CURRENT_COROUTINE;
+
+	if (coro == NULL) {
+		RETURN_NULL();
+	}
+
+	zend_object *object = ZEND_COROUTINE_OBJECT(coro);
+
+	if (object == NULL) {
+		RETURN_NULL();
+	}
+
+	GC_ADDREF(object);
+	RETURN_OBJ(object);
+}
+
+ZEND_METHOD(Async_Coroutine, resume)
+{
+	zval *coroutine;
+	zend_object *error = NULL;
+
+	ZEND_PARSE_PARAMETERS_START(1, 2)
+		Z_PARAM_OBJECT(coroutine)
+		Z_PARAM_OPTIONAL
+		Z_PARAM_OBJ_OF_CLASS_OR_NULL(error, zend_ce_throwable)
+	ZEND_PARSE_PARAMETERS_END();
+
+	/* A deferred wake: route the coroutine back to the scheduler's resume()
+	 * hook to be re-queued (never an immediate fiber switch). Fall back to
+	 * enqueue() when the scheduler does not override resume(). */
+	php_async_hook_id id;
+
+	if (PHP_ASYNC_HOOK(PHP_ASYNC_HOOK_RESUME)->set) {
+		id = PHP_ASYNC_HOOK_RESUME;
+	} else if (PHP_ASYNC_HOOK(PHP_ASYNC_HOOK_ENQUEUE)->set) {
+		id = PHP_ASYNC_HOOK_ENQUEUE;
+	} else {
+		zend_throw_error(NULL, "The registered scheduler cannot resume coroutines");
+		RETURN_THROWS();
+	}
+
+	zval args[2];
+	ZVAL_COPY(&args[0], coroutine);
+	uint32_t argc = 1;
+
+	if (id == PHP_ASYNC_HOOK_RESUME) {
+		if (error != NULL) {
+			ZVAL_OBJ(&args[1], error);
+			GC_ADDREF(error);
+		} else {
+			ZVAL_NULL(&args[1]);
+		}
+		argc = 2;
+	}
+
+	php_async_hook_call_bool(id, argc, args);
+
+	zval_ptr_dtor(&args[0]);
+	if (argc == 2) {
+		zval_ptr_dtor(&args[1]);
+	}
+
+	if (UNEXPECTED(EG(exception))) {
+		RETURN_THROWS();
+	}
+}
+
 void zend_register_scheduler_hook(void)
 {
 	async_ce_Scheduler = register_class_Async_Scheduler();
 	async_ce_AbstractScheduler = register_class_Async_AbstractScheduler(async_ce_Scheduler);
 	register_class_Async_SchedulerHook();
+	register_class_Async_Coroutine();
 }
 
 void zend_scheduler_hook_request_shutdown(void)

Zend/zend_scheduler_hook.stub.php

@@ -104,3 +104,20 @@ public static function getModule(): ?string {}
      */
     public static function defer(callable $task): void {}
 }
+
+/**
+ * The coroutine currently on the stack, and the way to wake a suspended one.
+ *
+ * The engine tracks the current coroutine from the object the scheduler
+ * returned from Async\Scheduler::interceptFiber() — there is no setter.
+ * resume() is a *deferred* wake: it routes the coroutine back to the
+ * scheduler's resume() hook to be re-queued, never an immediate fiber switch.
+ */
+final class Coroutine
+{
+    /** The coroutine object running on the current stack, or null in the main flow. */
+    public static function current(): ?object {}
+
+    /** Wake a suspended coroutine: hand it to the scheduler to be run later. */
+    public static function resume(object $coroutine, ?\Throwable $error = null): void {}
+}