diff --git a/src/EventLoop/Internal/AbstractDriver.php b/src/EventLoop/Internal/AbstractDriver.php index ef7513b..e11c22b 100644 --- a/src/EventLoop/Internal/AbstractDriver.php +++ b/src/EventLoop/Internal/AbstractDriver.php @@ -439,7 +439,7 @@ final protected function error(\Closure $closure, \Throwable $exception): void */ abstract protected function now(): float; - private function invokeMicrotasks(): void + private function invokeMicrotasks(\Fiber $fiber): void { while (!$this->microtaskQueue->isEmpty()) { [$callback, $args] = $this->microtaskQueue->dequeue(); @@ -455,6 +455,10 @@ private function invokeMicrotasks(): void unset($callback, $args); + if ($fiber !== $this->callbackFiber) { + return; + } + if ($this->interrupt) { /** @noinspection PhpUnhandledExceptionInspection */ \Fiber::suspend($this->internalSuspensionMarker); @@ -587,10 +591,13 @@ private function createLoopFiber(): void private function createCallbackFiber(): void { $this->callbackFiber = new \Fiber(function (): void { + /** @var \Fiber $fiber */ + $fiber = \Fiber::getCurrent(); + do { - $this->invokeMicrotasks(); + $this->invokeMicrotasks($fiber); - while (!$this->callbackQueue->isEmpty()) { + while ($fiber === $this->callbackFiber && !$this->callbackQueue->isEmpty()) { $callback = $this->callbackQueue->dequeue(); if (!isset($this->callbacks[$callback->id]) || !$callback->invokable) { @@ -636,12 +643,20 @@ private function createCallbackFiber(): void unset($callback); + if ($fiber !== $this->callbackFiber) { + return; + } + if ($this->interrupt) { /** @noinspection PhpUnhandledExceptionInspection */ \Fiber::suspend($this->internalSuspensionMarker); } - $this->invokeMicrotasks(); + $this->invokeMicrotasks($fiber); + } + + if ($fiber !== $this->callbackFiber) { + return; } /** @noinspection PhpUnhandledExceptionInspection */ diff --git a/test/EventLoopTest.php b/test/EventLoopTest.php index 9aacf59..b2427cb 100644 --- a/test/EventLoopTest.php +++ b/test/EventLoopTest.php @@ -411,6 +411,314 @@ public function testSuspensionWithinQueue(): void self::assertSame($send, $received); } + public function testResumedQueueFibersDoNotNestInEachOther(): void + { + $fibers = 25; + $suspensions = []; + $depths = []; + + for ($i = 0; $i < $fibers; $i++) { + EventLoop::queue(static function () use (&$suspensions, &$depths): void { + $suspension = EventLoop::getSuspension(); + $suspensions[] = $suspension; + $suspension->suspend(); + + $exception = new \Exception(); + $depths[] = \count($exception->getTrace()); + }); + } + + EventLoop::defer(static function () use (&$suspensions): void { + foreach ($suspensions as $suspension) { + $suspension->resume(); + } + }); + + EventLoop::run(); + + self::assertCount($fibers, $depths); + self::assertCount(1, \array_unique($depths)); + } + + public function testResumedCallbackFibersDoNotNestInEachOther(): void + { + $fibers = 25; + $suspensions = []; + $depths = []; + + for ($i = 0; $i < $fibers; $i++) { + EventLoop::defer(static function () use (&$suspensions, &$depths): void { + $suspension = EventLoop::getSuspension(); + $suspensions[] = $suspension; + $suspension->suspend(); + + $exception = new \Exception(); + $depths[] = \count($exception->getTrace()); + }); + } + + EventLoop::delay(0, static function () use (&$suspensions): void { + foreach ($suspensions as $suspension) { + $suspension->resume(); + } + }); + + EventLoop::run(); + + self::assertCount($fibers, $depths); + self::assertCount(1, \array_unique($depths)); + } + + public function testDeferredCallbackDoesNotRunInsideResumedFiber(): void + { + EventLoop::queue(static function () use (&$suspension): void { + $suspension = EventLoop::getSuspension(); + $suspension->suspend(); + }); + + EventLoop::defer(static function () use (&$suspension, &$resumingDepth): void { + $exception = new \Exception(); + $resumingDepth = \count($exception->getTrace()); + + $suspension->resume(); + }); + + EventLoop::defer(static function () use (&$nextDepth): void { + $exception = new \Exception(); + $nextDepth = \count($exception->getTrace()); + }); + + EventLoop::run(); + + self::assertSame($resumingDepth, $nextDepth); + } + + public function testResumedFibersRunInOrderWithQueuedMicrotasks(): void + { + $order = []; + + EventLoop::queue(static function () use (&$first, &$order): void { + $first = EventLoop::getSuspension(); + $first->suspend(); + $order[] = 'first'; + }); + + EventLoop::queue(static function () use (&$second, &$order): void { + $second = EventLoop::getSuspension(); + $second->suspend(); + $order[] = 'second'; + }); + + EventLoop::defer(static function () use (&$first, &$second, &$order): void { + $first->resume(); + EventLoop::queue(static function () use (&$order): void { + $order[] = 'microtask'; + }); + $second->resume(); + }); + + EventLoop::run(); + + self::assertSame(['first', 'microtask', 'second'], $order); + } + + public function testExceptionFromResumedFiberLeavesNextMicrotaskToNextRun(): void + { + $exception = new \Exception(); + $invoked = false; + + EventLoop::queue(static function () use (&$suspension, $exception): void { + $suspension = EventLoop::getSuspension(); + $suspension->suspend(); + + throw $exception; + }); + + EventLoop::defer(static function () use (&$suspension, &$invoked): void { + $suspension->resume(); + EventLoop::queue(static function () use (&$invoked): void { + $invoked = true; + }); + }); + + try { + EventLoop::run(); + self::fail('The exception must leave the event loop'); + } catch (UncaughtThrowable $throwable) { + self::assertSame($exception, $throwable->getPrevious()); + } + + self::assertFalse($invoked); + + EventLoop::run(); + + self::assertTrue($invoked); + } + + public function testFiberResumedDirectlyFromMainLeavesMicrotasksToNextRun(): void + { + $log = []; + + EventLoop::defer(static function () use (&$fiber, &$log): void { + $fiber = \Fiber::getCurrent(); + \Fiber::suspend(); + + EventLoop::queue(static function () use (&$log): void { + $log[] = 'microtask'; + }); + + $log[] = 'callback'; + }); + + EventLoop::run(); + + $fiber->resume(); + $log[] = 'resumed'; + + EventLoop::run(); + + self::assertSame(['callback', 'resumed', 'microtask'], $log); + self::assertTrue($fiber->isTerminated()); + } + + public function testFiberResumedDirectlyFromCallbackLeavesMicrotasksUntilCallbackReturns(): void + { + $log = []; + + EventLoop::defer(static function () use (&$fiber, &$log): void { + $fiber = \Fiber::getCurrent(); + \Fiber::suspend(); + + EventLoop::queue(static function () use (&$log): void { + $log[] = 'microtask'; + }); + + $log[] = 'callback'; + }); + + EventLoop::delay(0, static function () use (&$fiber, &$log): void { + $fiber->resume(); + $log[] = 'resumed'; + }); + + EventLoop::run(); + + self::assertSame(['callback', 'resumed', 'microtask'], $log); + self::assertTrue($fiber->isTerminated()); + } + + public function testTimerCallbackDoesNotRunInsideDirectlyResumedFiber(): void + { + $log = []; + + EventLoop::defer(static function () use (&$fiber, &$log): void { + $fiber = \Fiber::getCurrent(); + \Fiber::suspend(); + + $log[] = 'callback'; + }); + + EventLoop::delay(0, static function () use (&$fiber, &$log): void { + $fiber->resume(); + $log[] = 'resumed'; + }); + + EventLoop::delay(0, static function () use (&$log): void { + $log[] = 'other callback'; + }); + + EventLoop::run(); + + self::assertSame(['callback', 'resumed', 'other callback'], $log); + self::assertTrue($fiber->isTerminated()); + } + + /** + * Switching fibers in destructors is only supported since PHP 8.4. + * + * @requires PHP >= 8.4 + */ + public function testDeferredCallbackDoesNotRunInsideFiberSuspendedByDestructor(): void + { + // Suspending in a destructor replaces the callback fiber the destructor runs in. + $suspendOnDestruct = new class (static function () use (&$replacedFiber, &$suspension): void { + $replacedFiber = \Fiber::getCurrent(); + $suspension = EventLoop::getSuspension(); + $suspension->suspend(); + }) { + public function __construct(private readonly \Closure $onDestruct) + { + } + + public function __destruct() + { + ($this->onDestruct)(); + } + }; + + // Runs first and cancels the next callback while it is already queued for this tick. + EventLoop::defer(static function () use (&$callbackId): void { + EventLoop::cancel($callbackId); + }); + + $callbackId = EventLoop::defer(static function () use ($suspendOnDestruct): void { + self::fail('Callback must not be invoked after being cancelled'); + }); + + // The callback now holds the only reference, so the destructor runs when the loop discards it after the cancel. + unset($suspendOnDestruct); + + EventLoop::defer(static function () use (&$suspension): void { + $suspension->resume(); + }); + + EventLoop::defer(static function () use (&$lastCallbackFiber): void { + $lastCallbackFiber = \Fiber::getCurrent(); + }); + + EventLoop::run(); + + self::assertNotSame($replacedFiber, $lastCallbackFiber); + self::assertTrue($replacedFiber->isTerminated()); + } + + public function testExceptionFromDirectlyResumedFiberIsThrownAfterResumingCallbackReturns(): void + { + $exception = new \Exception(); + $log = []; + + EventLoop::defer(static function () use (&$fiber, $exception): void { + $fiber = \Fiber::getCurrent(); + \Fiber::suspend(); + + throw $exception; + }); + + EventLoop::delay(0, static function () use (&$fiber, &$log): void { + $fiber->resume(); + $log[] = 'resumed'; + + EventLoop::queue(static function () use (&$log): void { + $log[] = 'microtask'; + }); + }); + + try { + EventLoop::run(); + self::fail('The exception must leave the event loop'); + } catch (UncaughtThrowable $throwable) { + self::assertSame($exception, $throwable->getPrevious()); + } + + // A raw Fiber::resume() cannot be stopped, so the resuming callback finishes before the interrupt is thrown. + self::assertTrue($fiber->isTerminated()); + self::assertSame(['resumed'], $log); + + EventLoop::run(); + + self::assertSame(['resumed', 'microtask'], $log); + } + public function testSuspensionThrowingErrorViaInterrupt(): void { $suspension = EventLoop::getSuspension();