Cancel fiber

Author: WyriHaximus

composer.json

@@ -31,7 +31,8 @@
         "react/promise": "^2.8 || ^1.2.1"
     },
     "require-dev": {
-        "phpunit/phpunit": "^9.3"
+        "phpunit/phpunit": "^9.3",
+        "react/promise-timer": "^1.8"
     },
     "autoload": {
         "psr-4": {

src/FiberMap.php

@@ -0,0 +1,39 @@
+<?php
+
+namespace React\Async;
+
+use Fiber;
+use React\Promise\PromiseInterface;
+
+/**
+ * @internal
+ */
+final class FiberMap
+{
+    private array $map = [];
+
+    public function register(Fiber $fiber): void
+    {
+        $this->map[spl_object_hash($fiber)] = [];
+    }
+
+    public function attachPromise(Fiber $fiber, PromiseInterface $promise): void
+    {
+        $this->map[spl_object_hash($fiber)][] = $promise;
+    }
+
+    public function has(Fiber $fiber): bool
+    {
+        return array_key_exists(spl_object_hash($fiber), $this->map);
+    }
+
+    public function getPromises(Fiber $fiber): array
+    {
+        return $this->map[spl_object_hash($fiber)];
+    }
+
+    public function unregister(Fiber $fiber): void
+    {
+        unset($this->map[spl_object_hash($fiber)]);
+    }
+}

src/functions.php

@@ -2,6 +2,7 @@
 
 namespace React\Async;
 
+use Fiber;
 use React\EventLoop\Loop;
 use React\Promise\CancellablePromiseInterface;
 use React\Promise\Deferred;
@@ -20,16 +21,36 @@
  */
 function async(callable $function): callable
 {
-    return static fn (mixed ...$args): PromiseInterface => new Promise(function (callable $resolve, callable $reject) use ($function, $args): void {
-        $fiber = new \Fiber(function () use ($resolve, $reject, $function, $args): void {
+    $fiber = null;
+    return static fn (mixed ...$args): PromiseInterface => new Promise(function (callable $resolve, callable $reject) use ($function, $args, &$fiber): void {
+        $fiber = new \Fiber(function () use ($resolve, $reject, $function, $args, &$fiber): void {
+            if (!fiberMap()->has($fiber)) {
+                $reject(new \Exception('Fiber has been cancelled before execution started'));
+                return;
+            }
+
             try {
                 $resolve($function(...$args));
             } catch (\Throwable $exception) {
                 $reject($exception);
+            } finally {
+                fiberMap()->unregister($fiber);
             }
         });
 
+        fiberMap()->register($fiber);
+
         Loop::futureTick(static fn() => $fiber->start());
+    }, function () use (&$fiber): void {
+        if ($fiber instanceof Fiber) {
+            $promises = fiberMap()->getPromises($fiber);
+            fiberMap()->unregister($fiber);
+            foreach ($promises as $promise) {
+                if (method_exists($promise, 'cancel')) {
+                    $promise->cancel();
+                }
+            }
+        }
     });
 }
 
@@ -82,6 +103,10 @@ function await(PromiseInterface $promise): mixed
     $rejected = false;
     $resolvedValue = null;
     $rejectedThrowable = null;
+    $lowLevelFiber = \Fiber::getCurrent();
+    if ($lowLevelFiber !== null) {
+        fiberMap()->attachPromise($lowLevelFiber, $promise);
+    }
 
     $promise->then(
         function (mixed $value) use (&$resolved, &$resolvedValue, &$fiber): void {
@@ -433,3 +458,17 @@ function waterfall(array $tasks): PromiseInterface
 
     return $deferred->promise();
 }
+
+/**
+ * @internal
+ */
+function fiberMap(): FiberMap
+{
+    static $wm = null;
+
+    if ($wm === null) {
+        $wm = new FiberMap();
+    }
+
+    return $wm;
+}

tests/AsyncTest.php

@@ -8,6 +8,7 @@
 use function React\Async\async;
 use function React\Async\await;
 use function React\Promise\all;
+use function React\Promise\Timer\sleep;
 
 class AsyncTest extends TestCase
 {
@@ -84,4 +85,39 @@ public function testAsyncReturnsPromiseThatFulfillsWithValueWhenCallbackReturnsA
         $this->assertGreaterThan(0.1, $time);
         $this->assertLessThan(0.12, $time);
     }
+
+    public function testCancel()
+    {
+        self::expectOutputString('a');
+
+        $promise = async(function (): int {
+            echo 'a';
+            await(sleep(2));
+            echo 'b';
+
+            return time();
+        })();
+        Loop::addTimer(0.5, function () use ($promise) {
+            $promise->cancel();
+        });
+
+        Loop::run();
+    }
+
+    public function testCancelNeverStartedFiber()
+    {
+        self::expectOutputString('');
+        $this->expectException(\Exception::class);
+        $this->expectExceptionMessage('Fiber has been cancelled before execution started');
+
+        $promise = async(function (): int {
+            echo 'a';
+            await(sleep(2));
+            echo 'b';
+
+            return time();
+        })();
+        $promise->cancel();
+        await($promise);
+    }
 }