Fiberize error handler (#10)

Author: trowski

src/EventLoop.php

@@ -391,10 +391,11 @@ public static function run(): void
             self::$fiber = self::createFiber();
         }
 
-        if (self::$fiber->isStarted()) {
-            self::$fiber->resume();
-        } else {
-            self::$fiber->start();
+        $lambda = self::$fiber->isStarted() ? self::$fiber->resume() : self::$fiber->start();
+
+        if ($lambda) {
+            $lambda();
+            throw new \Error('Interrupt from event loop must throw an exception');
         }
     }
 

src/EventLoop/Internal/AbstractDriver.php

@@ -25,6 +25,7 @@ abstract class AbstractDriver implements Driver
 
     private \Fiber $callbackFiber;
     private \Fiber $queueFiber;
+    private \Closure $errorCallback;
 
     /** @var Callback[] */
     private array $callbacks = [];
@@ -51,15 +52,18 @@ abstract class AbstractDriver implements Driver
 
     private bool $running = false;
 
+    private bool $inFiber = false;
+
     private \stdClass $internalSuspensionMarker;
 
     public function __construct()
     {
         $this->internalSuspensionMarker = new \stdClass();
         $this->createCallbackFiber();
         $this->createQueueFiber();
+        $this->createErrorCallback();
         /** @psalm-suppress InvalidArgument */
-        $this->interruptCallback = \Closure::fromCallable([$this, 'interrupt']);
+        $this->interruptCallback = \Closure::fromCallable([$this, 'setInterrupt']);
     }
 
     /**
@@ -84,9 +88,14 @@ public function run(): void
         }
 
         $this->running = true;
+        $this->inFiber = \Fiber::getCurrent() !== null;
 
         try {
             while ($this->running) {
+                if ($this->interrupt) {
+                    $this->invokeInterrupt();
+                }
+
                 $this->invokeMicrotasks();
 
                 if ($this->isEmpty()) {
@@ -97,6 +106,7 @@ public function run(): void
             }
         } finally {
             $this->running = false;
+            $this->inFiber = false;
         }
     }
 
@@ -594,10 +604,7 @@ protected function invokeCallback(Callback $callback): void
         }
 
         if ($this->interrupt) {
-            $interrupt = $this->interrupt;
-            $this->interrupt = null;
-
-            \Fiber::suspend($interrupt);
+            $this->invokeInterrupt();
         }
 
         if ($this->microQueue) {
@@ -615,10 +622,12 @@ protected function invokeCallback(Callback $callback): void
     protected function error(\Throwable $exception): void
     {
         if ($this->errorHandler === null) {
-            throw $exception;
+            $this->setInterrupt(static fn () => throw $exception);
+            return;
         }
 
-        ($this->errorHandler)($exception);
+        $fiber = new \Fiber($this->errorCallback);
+        $fiber->start($this->errorHandler, $exception);
     }
 
     /**
@@ -698,18 +707,31 @@ private function invokeMicrotasks(): void
                 }
 
                 if ($this->interrupt) {
-                    $interrupt = $this->interrupt;
-                    $this->interrupt = null;
-
-                    \Fiber::suspend($interrupt);
+                    $this->invokeInterrupt();
                 }
             }
         }
     }
 
-    private function interrupt(callable $callback): void
+    private function setInterrupt(callable $interrupt): void
     {
-        $this->interrupt = $callback;
+        \assert($this->interrupt === null);
+        $this->interrupt = $interrupt;
+    }
+
+    private function invokeInterrupt(): void
+    {
+        \assert($this->interrupt !== null);
+
+        $interrupt = $this->interrupt;
+        $this->interrupt = null;
+
+        if (!$this->inFiber) {
+            $interrupt();
+            throw new \Error('Interrupt must throw if not executing in a fiber');
+        }
+
+        \Fiber::suspend($interrupt);
     }
 
     private function createCallbackFiber(): void
@@ -749,4 +771,15 @@ private function createQueueFiber(): void
 
         $this->queueFiber->start();
     }
+
+    private function createErrorCallback(): void
+    {
+        $this->errorCallback = function (callable $errorHandler, \Throwable $exception): void {
+            try {
+                $errorHandler($exception);
+            } catch (\Throwable $exception) {
+                $this->interrupt = static fn () => throw $exception;
+            }
+        };
+    }
 }

src/EventLoop/Suspension.php

@@ -21,6 +21,7 @@ final class Suspension
     private \Fiber $scheduler;
     private Driver $driver;
     private bool $pending = false;
+    private ?\FiberError $error = null;
     /** @var callable */
     private $interrupt;
 
@@ -45,7 +46,7 @@ public function __construct(Driver $driver, \Fiber $scheduler, callable $interru
     public function throw(\Throwable $throwable): void
     {
         if (!$this->pending) {
-            throw new \Error('Must call throw() before calling resume()');
+            throw $this->error ?? new \Error('Must call suspend() before calling throw()');
         }
 
         $this->pending = false;
@@ -61,7 +62,7 @@ public function throw(\Throwable $throwable): void
     public function resume(mixed $value): void
     {
         if (!$this->pending) {
-            throw new \Error('Must call suspend() before calling resume()');
+            throw $this->error ?? new \Error('Must call suspend() before calling resume()');
         }
 
         $this->pending = false;
@@ -88,7 +89,13 @@ public function suspend(): mixed
 
         // Awaiting from within a fiber.
         if ($this->fiber) {
-            return \Fiber::suspend();
+            try {
+                return \Fiber::suspend();
+            } catch (\FiberError $exception) {
+                $this->pending = false;
+                $this->error = $exception;
+                throw $exception;
+            }
         }
 
         // Awaiting from {main}.