Async core

Zend/tests/async/continuation_current_main.phpt

@@ -0,0 +1,52 @@
+--TEST--
+Async\Continuation: currentContinuation captures main, a continuation switches back into it
+--FILE--
+<?php
+final class CaptureScheduler implements Async\Scheduler {
+    public function __construct(
+        public readonly Closure $create,   // createContinuation(callable $entry): Continuation
+        public readonly Closure $capture,  // currentContinuation(): Continuation
+    ) {}
+
+    public function onEnqueue(object $coroutine, ?Throwable $error = null): bool { return true; }
+    public function onSuspend(bool $fromMain, bool $isBailout): ?object { return null; }
+    public function onShutdown(): void {}
+    public function onWaitInfo(object $coroutine, string $info): void {}
+    public function onFiber(Fiber $fiber): ?object { return null; }
+    public function onDefer(callable $task): void {}
+    public function getContext(object $coroutine): object { return new stdClass(); }
+    public function getInternalContext(object $coroutine): object { return new stdClass(); }
+    public function contextFind(object $context, mixed $key): mixed { return null; }
+    public function contextSet(object $context, mixed $key, mixed $value): void {}
+    public function contextUnset(object $context, mixed $key): bool { return true; }
+}
+
+$sched = null;
+Async\SchedulerHook::register('test',
+    function (callable $create, callable $capture) use (&$sched): CaptureScheduler {
+        return $sched = new CaptureScheduler($create, $capture);
+    });
+
+// The main flow becomes addressable like any other coroutine: this is the
+// capture that normalisation ("the main flow is a coroutine too") builds on.
+$main = ($sched->capture)();
+var_dump($main instanceof Async\Continuation);
+
+$c = ($sched->create)(function () use ($main): void {
+    echo "  in continuation\n";
+    $back = $main->switchTo('back to main');   // symmetric jump into captured main
+    echo "  finishing ($back)\n";
+});
+
+var_dump($c->switchTo());
+echo "main resumed\n";
+$c->switchTo('bye');    // let the body run to completion
+echo "done\n";
+?>
+--EXPECT--
+bool(true)
+  in continuation
+string(12) "back to main"
+main resumed
+  finishing (bye)
+done

Zend/tests/async/continuation_switch.phpt

@@ -0,0 +1,26 @@
+--TEST--
+Async\Continuation: switchTo runs the body on its own stack and returns to the switcher
+--FILE--
+<?php
+$a = Async\Continuation::create(function (): void {
+    echo "  A\n";
+});
+
+$b = Async\Continuation::create(function (): void {
+    $sum = 0;
+    for ($i = 0; $i < 1000; $i++) {
+        $sum += $i;   // real VM work on the continuation's own stack
+    }
+    echo "  B sum=$sum\n";
+});
+
+echo "before\n";
+$a->switchTo();
+$b->switchTo();
+echo "after\n";
+?>
+--EXPECT--
+before
+  A
+  B sum=499500
+after

Zend/tests/async/continuation_switchto_error.phpt

@@ -0,0 +1,52 @@
+--TEST--
+Async\Continuation: switchTo() delivers $error at the suspension point; boundary cases
+--FILE--
+<?php
+// 1. An error is thrown from the switchTo() the target is suspended in.
+$main = Async\Continuation::current();
+
+$worker = Async\Continuation::create(function () use ($main): void {
+    echo "  worker started\n";
+    try {
+        $main->switchTo("parked");
+    } catch (\RuntimeException $e) {
+        echo "  caught at suspension point: {$e->getMessage()}\n";
+    }
+});
+
+var_dump($worker->switchTo());
+$worker->switchTo(null, new \RuntimeException("cancelled"));
+
+// 2. Passing both a non-null value and an error is a ValueError.
+$other = Async\Continuation::create(function (): void {});
+try {
+    $other->switchTo("value", new \RuntimeException("boom"));
+} catch (\ValueError $e) {
+    echo "ValueError\n";
+}
+
+// 3. A first entry with an error finishes the continuation without starting
+// the body; the error surfaces at the switch site.
+$never = Async\Continuation::create(function (): void {
+    echo "  never printed\n";
+});
+try {
+    $never->switchTo(null, new \LogicException("cancel before start"));
+} catch (\LogicException $e) {
+    echo "first entry: {$e->getMessage()}\n";
+}
+
+// 4. Switching into a finished continuation throws an Error.
+try {
+    $never->switchTo();
+} catch (\Error $e) {
+    echo $e->getMessage(), "\n";
+}
+?>
+--EXPECT--
+  worker started
+string(6) "parked"
+  caught at suspension point: cancelled
+ValueError
+first entry: cancel before start
+Cannot switch into a finished continuation

Zend/tests/async/continuation_symmetric_nested.phpt

@@ -0,0 +1,25 @@
+--TEST--
+Async\Continuation: a coroutine switches into another mid-run; the inner returns to the outer (symmetric)
+--FILE--
+<?php
+$b = Async\Continuation::create(function (): void {
+    echo "  B: runs to completion\n";
+    // returns -> control goes back to B's switcher (A)
+});
+
+$a = Async\Continuation::create(function () use ($b): void {
+    echo "  A: 1\n";
+    $b->switchTo();            // A -> B mid-run; B completes -> back to A
+    echo "  A: 2 (after B)\n";
+});
+
+echo "main -> A\n";
+$a->switchTo();               // main -> A -> B(done) -> A(done) -> main
+echo "main: done\n";
+?>
+--EXPECT--
+main -> A
+  A: 1
+  B: runs to completion
+  A: 2 (after B)
+main: done

Zend/tests/async/continuation_value_exception.phpt

@@ -0,0 +1,22 @@
+--TEST--
+Async\Continuation: switchTo() returns the body's value and re-raises its exception
+--FILE--
+<?php
+var_dump(Async\Continuation::create(fn () => 42)->switchTo());
+var_dump(Async\Continuation::create(fn () => "hi")->switchTo());
+
+try {
+    Async\Continuation::create(function () {
+        throw new RuntimeException("boom");
+    })->switchTo();
+} catch (RuntimeException $e) {
+    echo "caught: ", $e->getMessage(), "\n";
+}
+
+echo "survived\n";
+?>
+--EXPECT--
+int(42)
+string(2) "hi"
+caught: boom
+survived

Zend/tests/async/fiber_lowlevel_null.phpt

@@ -0,0 +1,43 @@
+--TEST--
+interceptFiber returning null keeps the fiber on the low-level path
+--FILE--
+<?php
+Async\SchedulerHook::register('test', fn () => new class implements \Async\Scheduler {
+    public function onShutdown(): void {}
+    public function onWaitInfo(object $coroutine, string $info): void {}
+    public function onDefer(callable $task): void {}
+    public function getContext(object $coroutine): object { return new \stdClass(); }
+    public function getInternalContext(object $coroutine): object { return new \stdClass(); }
+    public function contextFind(object $context, mixed $key): mixed { return null; }
+    public function contextSet(object $context, mixed $key, mixed $value): void {}
+    public function contextUnset(object $context, mixed $key): bool { return true; }
+    public function onFiber(Fiber $fiber): ?object {
+        echo "intercept: null\n";
+        return null;
+    }
+    public function onEnqueue(object $coroutine, ?Throwable $error = null): bool { return true; }
+    public function onSuspend(bool $fromMain, bool $isBailout): ?object {
+        // Never called for the low-level fiber itself; the engine invokes it
+        // for the after-main handover (script end, then after destructors).
+        echo "suspend(fromMain: ", var_export($fromMain, true), ")\n";
+        return null;
+    }
+});
+
+// A low-level fiber behaves exactly as classic Fiber even with a scheduler.
+$fiber = new Fiber(function (int $x): int {
+    $y = Fiber::suspend($x + 1);
+    return $y * 10;
+});
+
+var_dump($fiber->start(5));
+var_dump($fiber->resume(4));
+var_dump($fiber->getReturn());
+?>
+--EXPECT--
+intercept: null
+int(6)
+NULL
+int(40)
+suspend(fromMain: true)
+suspend(fromMain: true)

Zend/tests/async/fiber_managed_after_main.phpt

@@ -0,0 +1,54 @@
+--TEST--
+After-main handover runs deferred managed coroutines; abandoned ones die cleanly
+--FILE--
+<?php
+Async\SchedulerHook::register('test', fn () => new class implements \Async\Scheduler {
+    public function onShutdown(): void {}
+    public function onWaitInfo(object $coroutine, string $info): void {}
+    public function onDefer(callable $task): void {}
+    public function getContext(object $coroutine): object { return new \stdClass(); }
+    public function getInternalContext(object $coroutine): object { return new \stdClass(); }
+    public function contextFind(object $context, mixed $key): mixed { return null; }
+    public function contextSet(object $context, mixed $key, mixed $value): void {}
+    public function contextUnset(object $context, mixed $key): bool { return true; }
+    public SplQueue $queue;
+    public function __construct() { $this->queue = new SplQueue(); }
+    public function onFiber(Fiber $fiber): ?object {
+        return new class($fiber) {
+            public function __construct(public readonly Fiber $fiber) {}
+        };
+    }
+    public function onEnqueue(object $coroutine, ?Throwable $error = null): bool {
+        $this->queue->enqueue($coroutine);
+        return true;
+    }
+    public function onSuspend(bool $fromMain, bool $isBailout): ?object {
+        // This scheduler defers everything: nothing runs until after main.
+        if (!$fromMain) {
+            return null;
+        }
+        while (!$this->queue->isEmpty()) {
+            $fiber = $this->queue->dequeue()->fiber;
+            $fiber->isStarted() ? $fiber->resume() : $fiber->start();
+        }
+        return null;
+    }
+});
+
+$fiber = new Fiber(function (): void {
+    echo "ran after main\n";
+    Fiber::suspend();
+    echo "never printed: nobody resumes\n";
+});
+
+// The deferring scheduler does not run the fiber here.
+var_dump($fiber->start());
+var_dump($fiber->isStarted());
+
+echo "end of script\n";
+?>
+--EXPECT--
+NULL
+bool(false)
+end of script
+ran after main

Zend/tests/async/fiber_managed_basic.phpt

@@ -0,0 +1,61 @@
+--TEST--
+A fiber adopted by the scheduler runs through the coroutine path
+--FILE--
+<?php
+$scheduler = new class implements \Async\Scheduler {
+    public function onShutdown(): void {}
+    public function onWaitInfo(object $coroutine, string $info): void {}
+    public function onDefer(callable $task): void {}
+    public function getContext(object $coroutine): object { return new \stdClass(); }
+    public function getInternalContext(object $coroutine): object { return new \stdClass(); }
+    public function contextFind(object $context, mixed $key): mixed { return null; }
+    public function contextSet(object $context, mixed $key, mixed $value): void {}
+    public function contextUnset(object $context, mixed $key): bool { return true; }
+    public SplQueue $queue;
+    public array $log = [];
+    public function __construct() { $this->queue = new SplQueue(); }
+    public function onFiber(Fiber $fiber): ?object {
+        $this->log[] = 'intercept';
+        return new class($fiber) {
+            public function __construct(public readonly Fiber $fiber) {}
+        };
+    }
+    public function onEnqueue(object $coroutine, ?Throwable $error = null): bool {
+        $this->log[] = 'enqueue';
+        $this->queue->enqueue($coroutine);
+        return true;
+    }
+    public function onSuspend(bool $fromMain, bool $isBailout): ?object {
+        $this->log[] = 'suspend';
+        while (!$this->queue->isEmpty()) {
+            $fiber = $this->queue->dequeue()->fiber;
+            $fiber->isStarted() ? $fiber->resume() : $fiber->start();
+            if (!$fromMain) {
+                return null;
+            }
+        }
+        return null;
+    }
+};
+
+Async\SchedulerHook::register('test', fn () => $scheduler);
+
+$fiber = new Fiber(function (int $x): int {
+    $y = Fiber::suspend($x + 1);
+    return $y * 10;
+});
+
+var_dump($fiber->start(5));
+var_dump($fiber->isSuspended());
+var_dump($fiber->resume(4));
+var_dump($fiber->isTerminated());
+var_dump($fiber->getReturn());
+echo implode(',', $scheduler->log), "\n";
+?>
+--EXPECT--
+int(6)
+bool(true)
+NULL
+bool(true)
+int(40)
+intercept,enqueue,suspend,enqueue,suspend

Zend/tests/async/fiber_managed_throw.phpt

@@ -0,0 +1,56 @@
+--TEST--
+Fiber::throw() on a managed fiber delivers the exception at the suspension point
+--FILE--
+<?php
+Async\SchedulerHook::register('test', fn () => new class implements \Async\Scheduler {
+    public function onShutdown(): void {}
+    public function onWaitInfo(object $coroutine, string $info): void {}
+    public function onDefer(callable $task): void {}
+    public function getContext(object $coroutine): object { return new \stdClass(); }
+    public function getInternalContext(object $coroutine): object { return new \stdClass(); }
+    public function contextFind(object $context, mixed $key): mixed { return null; }
+    public function contextSet(object $context, mixed $key, mixed $value): void {}
+    public function contextUnset(object $context, mixed $key): bool { return true; }
+    public SplQueue $queue;
+    public function __construct() { $this->queue = new SplQueue(); }
+    public function onFiber(Fiber $fiber): ?object {
+        return new class($fiber) {
+            public function __construct(public readonly Fiber $fiber) {}
+        };
+    }
+    public function onEnqueue(object $coroutine, ?Throwable $error = null): bool {
+        if ($error !== null) {
+            echo "enqueue hook error: ", get_class($error), "\n";
+        }
+        $this->queue->enqueue($coroutine);
+        return true;
+    }
+    public function onSuspend(bool $fromMain, bool $isBailout): ?object {
+        while (!$this->queue->isEmpty()) {
+            $fiber = $this->queue->dequeue()->fiber;
+            $fiber->isStarted() ? $fiber->resume() : $fiber->start();
+            if (!$fromMain) {
+                return null;
+            }
+        }
+        return null;
+    }
+});
+
+$fiber = new Fiber(function (): string {
+    try {
+        Fiber::suspend('waiting');
+    } catch (RuntimeException $e) {
+        return 'caught: ' . $e->getMessage();
+    }
+    return 'not reached';
+});
+
+var_dump($fiber->start());
+$fiber->throw(new RuntimeException('boom'));
+var_dump($fiber->getReturn());
+?>
+--EXPECT--
+string(7) "waiting"
+enqueue hook error: RuntimeException
+string(12) "caught: boom"