From 1f48cc0c716b889c512d0a5af967a30f693e1b26 Mon Sep 17 00:00:00 2001 From: lav45 Date: Tue, 3 Nov 2026 02:34:52 +0300 Subject: [PATCH 1/7] Stop replaced callback fibers from running the event loop queues A fiber suspended by a callback is replaced as the callback fiber, but it stayed in the callback fiber loop once resumed. After finishing its callback it ran the next microtask itself, so fibers resumed in the same tick nested inside each other. A trace in the last one grew by about 4 frames per fiber, and exceptions held across a suspension cost O(N^2) memory. A replaced fiber now terminates after its callback and returns to the fiber that resumed it, which carries on with the queues. Microtask and callback order is unchanged. Fixes #139 --- src/EventLoop/Internal/AbstractDriver.php | 19 ++- test/EventLoopTest.php | 144 ++++++++++++++++++++++ 2 files changed, 162 insertions(+), 1 deletion(-) diff --git a/src/EventLoop/Internal/AbstractDriver.php b/src/EventLoop/Internal/AbstractDriver.php index ef7513b..734d7f9 100644 --- a/src/EventLoop/Internal/AbstractDriver.php +++ b/src/EventLoop/Internal/AbstractDriver.php @@ -455,6 +455,10 @@ private function invokeMicrotasks(): void unset($callback, $args); + if ($this->isCallbackFiberReplaced()) { + return; + } + if ($this->interrupt) { /** @noinspection PhpUnhandledExceptionInspection */ \Fiber::suspend($this->internalSuspensionMarker); @@ -590,7 +594,7 @@ private function createCallbackFiber(): void do { $this->invokeMicrotasks(); - while (!$this->callbackQueue->isEmpty()) { + while (!$this->isCallbackFiberReplaced() && !$this->callbackQueue->isEmpty()) { $callback = $this->callbackQueue->dequeue(); if (!isset($this->callbacks[$callback->id]) || !$callback->invokable) { @@ -636,6 +640,10 @@ private function createCallbackFiber(): void unset($callback); + if ($this->isCallbackFiberReplaced()) { + return; + } + if ($this->interrupt) { /** @noinspection PhpUnhandledExceptionInspection */ \Fiber::suspend($this->internalSuspensionMarker); @@ -644,12 +652,21 @@ private function createCallbackFiber(): void $this->invokeMicrotasks(); } + if ($this->isCallbackFiberReplaced()) { + return; + } + /** @noinspection PhpUnhandledExceptionInspection */ \Fiber::suspend($this->internalSuspensionMarker); } while (true); }); } + private function isCallbackFiberReplaced(): bool + { + return \Fiber::getCurrent() !== $this->callbackFiber; + } + private function createErrorCallback(): void { $this->errorCallback = function (\Closure $errorHandler, \Throwable $exception): void { diff --git a/test/EventLoopTest.php b/test/EventLoopTest.php index 9aacf59..3483487 100644 --- a/test/EventLoopTest.php +++ b/test/EventLoopTest.php @@ -411,6 +411,150 @@ 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::assertSame($depths[0], \max($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::assertSame($depths[0], \max($depths)); + } + + public function testResumedQueueFiberLeavesCallbacksToCurrentCallbackFiber(): 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 testResumedFibersKeepMicrotaskOrder(): 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 testExceptionFromResumedFiberStopsNextMicrotask(): 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 testSuspensionThrowingErrorViaInterrupt(): void { $suspension = EventLoop::getSuspension(); From 16bb768e1b255bd595d6a3572c0b3ffa298c5b54 Mon Sep 17 00:00:00 2001 From: lav45 Date: Tue, 3 Nov 2026 11:06:19 +0300 Subject: [PATCH 2/7] Rename isCallbackFiberReplaced() to isCurrentCallbackFiber() --- src/EventLoop/Internal/AbstractDriver.php | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/EventLoop/Internal/AbstractDriver.php b/src/EventLoop/Internal/AbstractDriver.php index 734d7f9..0d24494 100644 --- a/src/EventLoop/Internal/AbstractDriver.php +++ b/src/EventLoop/Internal/AbstractDriver.php @@ -455,7 +455,7 @@ private function invokeMicrotasks(): void unset($callback, $args); - if ($this->isCallbackFiberReplaced()) { + if (!$this->isCurrentCallbackFiber()) { return; } @@ -594,7 +594,7 @@ private function createCallbackFiber(): void do { $this->invokeMicrotasks(); - while (!$this->isCallbackFiberReplaced() && !$this->callbackQueue->isEmpty()) { + while ($this->isCurrentCallbackFiber() && !$this->callbackQueue->isEmpty()) { $callback = $this->callbackQueue->dequeue(); if (!isset($this->callbacks[$callback->id]) || !$callback->invokable) { @@ -640,7 +640,7 @@ private function createCallbackFiber(): void unset($callback); - if ($this->isCallbackFiberReplaced()) { + if (!$this->isCurrentCallbackFiber()) { return; } @@ -652,7 +652,7 @@ private function createCallbackFiber(): void $this->invokeMicrotasks(); } - if ($this->isCallbackFiberReplaced()) { + if (!$this->isCurrentCallbackFiber()) { return; } @@ -662,9 +662,9 @@ private function createCallbackFiber(): void }); } - private function isCallbackFiberReplaced(): bool + private function isCurrentCallbackFiber(): bool { - return \Fiber::getCurrent() !== $this->callbackFiber; + return \Fiber::getCurrent() === $this->callbackFiber; } private function createErrorCallback(): void From 4e9acb686795a10e478962d578cb56a07f1cbafe Mon Sep 17 00:00:00 2001 From: lav45 Date: Tue, 3 Nov 2026 11:24:25 +0300 Subject: [PATCH 3/7] Compare the callback fiber captured at start instead of calling Fiber::getCurrent() The identity check runs after every microtask and callback. As a method calling Fiber::getCurrent() it cost 9-12% on callbacks that don't suspend. The callback fiber now keeps itself in a local variable and compares it with the callback fiber property, which brings the cost down to 2-3%. --- src/EventLoop/Internal/AbstractDriver.php | 22 ++++++++++------------ 1 file changed, 10 insertions(+), 12 deletions(-) diff --git a/src/EventLoop/Internal/AbstractDriver.php b/src/EventLoop/Internal/AbstractDriver.php index 0d24494..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,7 +455,7 @@ private function invokeMicrotasks(): void unset($callback, $args); - if (!$this->isCurrentCallbackFiber()) { + if ($fiber !== $this->callbackFiber) { return; } @@ -591,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->isCurrentCallbackFiber() && !$this->callbackQueue->isEmpty()) { + while ($fiber === $this->callbackFiber && !$this->callbackQueue->isEmpty()) { $callback = $this->callbackQueue->dequeue(); if (!isset($this->callbacks[$callback->id]) || !$callback->invokable) { @@ -640,7 +643,7 @@ private function createCallbackFiber(): void unset($callback); - if (!$this->isCurrentCallbackFiber()) { + if ($fiber !== $this->callbackFiber) { return; } @@ -649,10 +652,10 @@ private function createCallbackFiber(): void \Fiber::suspend($this->internalSuspensionMarker); } - $this->invokeMicrotasks(); + $this->invokeMicrotasks($fiber); } - if (!$this->isCurrentCallbackFiber()) { + if ($fiber !== $this->callbackFiber) { return; } @@ -662,11 +665,6 @@ private function createCallbackFiber(): void }); } - private function isCurrentCallbackFiber(): bool - { - return \Fiber::getCurrent() === $this->callbackFiber; - } - private function createErrorCallback(): void { $this->errorCallback = function (\Closure $errorHandler, \Throwable $exception): void { From 5b1446d1bdcd91bfe2040f62081d2e81f66ece59 Mon Sep 17 00:00:00 2001 From: lav45 Date: Tue, 3 Nov 2026 22:57:32 +0300 Subject: [PATCH 4/7] Name the test after what it checks: the next deferred callback doesn't run inside the resumed fiber --- test/EventLoopTest.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/EventLoopTest.php b/test/EventLoopTest.php index 3483487..de241b4 100644 --- a/test/EventLoopTest.php +++ b/test/EventLoopTest.php @@ -469,7 +469,7 @@ public function testResumedCallbackFibersDoNotNestInEachOther(): void self::assertSame($depths[0], \max($depths)); } - public function testResumedQueueFiberLeavesCallbacksToCurrentCallbackFiber(): void + public function testDeferredCallbackDoesNotRunInsideResumedFiber(): void { EventLoop::queue(static function () use (&$suspension): void { $suspension = EventLoop::getSuspension(); From 7e039cb5262983088aa7c4a9ce5633e11594f6c8 Mon Sep 17 00:00:00 2001 From: lav45 Date: Tue, 3 Nov 2026 23:05:51 +0300 Subject: [PATCH 5/7] Name the order and exception tests after what they check --- test/EventLoopTest.php | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/EventLoopTest.php b/test/EventLoopTest.php index de241b4..9eabde9 100644 --- a/test/EventLoopTest.php +++ b/test/EventLoopTest.php @@ -493,7 +493,7 @@ public function testDeferredCallbackDoesNotRunInsideResumedFiber(): void self::assertSame($resumingDepth, $nextDepth); } - public function testResumedFibersKeepMicrotaskOrder(): void + public function testResumedFibersRunInOrderWithQueuedMicrotasks(): void { $order = []; @@ -522,7 +522,7 @@ public function testResumedFibersKeepMicrotaskOrder(): void self::assertSame(['first', 'microtask', 'second'], $order); } - public function testExceptionFromResumedFiberStopsNextMicrotask(): void + public function testExceptionFromResumedFiberLeavesNextMicrotaskToNextRun(): void { $exception = new \Exception(); $invoked = false; From 1261d90615faebdd94cf5eb28238ae2d07db6501 Mon Sep 17 00:00:00 2001 From: Niklas Keller Date: Tue, 3 Nov 2026 21:08:10 +0100 Subject: [PATCH 6/7] Add tests for resuming replaced callback fibers Cover replaced fibers resumed directly via Fiber::resume() from {main} and from another callback, an exception thrown after such a resume, and a fiber replaced by a suspending destructor of a cancelled callback. Also assert that all resumed fiber trace depths are equal. Co-Authored-By: Claude Opus 5.5 --- test/EventLoopTest.php | 168 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 166 insertions(+), 2 deletions(-) diff --git a/test/EventLoopTest.php b/test/EventLoopTest.php index 9eabde9..fa0d397 100644 --- a/test/EventLoopTest.php +++ b/test/EventLoopTest.php @@ -437,7 +437,7 @@ public function testResumedQueueFibersDoNotNestInEachOther(): void EventLoop::run(); self::assertCount($fibers, $depths); - self::assertSame($depths[0], \max($depths)); + self::assertCount(1, \array_unique($depths)); } public function testResumedCallbackFibersDoNotNestInEachOther(): void @@ -466,7 +466,7 @@ public function testResumedCallbackFibersDoNotNestInEachOther(): void EventLoop::run(); self::assertCount($fibers, $depths); - self::assertSame($depths[0], \max($depths)); + self::assertCount(1, \array_unique($depths)); } public function testDeferredCallbackDoesNotRunInsideResumedFiber(): void @@ -555,6 +555,170 @@ public function testExceptionFromResumedFiberLeavesNextMicrotaskToNextRun(): voi self::assertTrue($invoked); } + public function testReplacedFiberResumedFromMainLeavesMicrotasksToNextRun(): 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 testReplacedFiberResumedFromCallbackLeavesMicrotasksToCurrentCallbackFiber(): 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 testReplacedFiberResumedFromCallbackDoesNotRunOtherCallbacks(): 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 testFiberReplacedByDestructorOfCancelledCallbackLeavesCallbacksToCurrentCallbackFiber(): 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 testExceptionFromReplacedFiberResumedFromCallbackInterruptsAfterResumingCallback(): 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(); From 657e11ce35e56592d50493ffc23e257e17c9576d Mon Sep 17 00:00:00 2001 From: Niklas Keller Date: Tue, 3 Nov 2026 21:15:58 +0100 Subject: [PATCH 7/7] Name the direct resume tests after what they check Co-Authored-By: Claude Opus 5.5 --- test/EventLoopTest.php | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/test/EventLoopTest.php b/test/EventLoopTest.php index fa0d397..b2427cb 100644 --- a/test/EventLoopTest.php +++ b/test/EventLoopTest.php @@ -555,7 +555,7 @@ public function testExceptionFromResumedFiberLeavesNextMicrotaskToNextRun(): voi self::assertTrue($invoked); } - public function testReplacedFiberResumedFromMainLeavesMicrotasksToNextRun(): void + public function testFiberResumedDirectlyFromMainLeavesMicrotasksToNextRun(): void { $log = []; @@ -581,7 +581,7 @@ public function testReplacedFiberResumedFromMainLeavesMicrotasksToNextRun(): voi self::assertTrue($fiber->isTerminated()); } - public function testReplacedFiberResumedFromCallbackLeavesMicrotasksToCurrentCallbackFiber(): void + public function testFiberResumedDirectlyFromCallbackLeavesMicrotasksUntilCallbackReturns(): void { $log = []; @@ -607,7 +607,7 @@ public function testReplacedFiberResumedFromCallbackLeavesMicrotasksToCurrentCal self::assertTrue($fiber->isTerminated()); } - public function testReplacedFiberResumedFromCallbackDoesNotRunOtherCallbacks(): void + public function testTimerCallbackDoesNotRunInsideDirectlyResumedFiber(): void { $log = []; @@ -638,7 +638,7 @@ public function testReplacedFiberResumedFromCallbackDoesNotRunOtherCallbacks(): * * @requires PHP >= 8.4 */ - public function testFiberReplacedByDestructorOfCancelledCallbackLeavesCallbacksToCurrentCallbackFiber(): void + 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 { @@ -682,7 +682,7 @@ public function __destruct() self::assertTrue($replacedFiber->isTerminated()); } - public function testExceptionFromReplacedFiberResumedFromCallbackInterruptsAfterResumingCallback(): void + public function testExceptionFromDirectlyResumedFiberIsThrownAfterResumingCallbackReturns(): void { $exception = new \Exception(); $log = [];