diff --git a/README.md b/README.md index 8a43856f..d84b944c 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,32 @@ Every package here is versioned and released alongside the core. Add what your a Thank you for considering contributing to the Doppar framework! The contribution guide can be found in the [Doppar documentation](https://doppar.com/versions/4.x/contributions). +## Concurrent tasks + +Independent non-blocking operations can run together in a named task group: + +```php +use Phaseolies\Support\Concurrency\Task; +use Phaseolies\Support\Concurrency\TaskGroup; + +$results = TaskGroup::run([ + 'profile' => function (Task $task) use ($client) { + return $task->await(function ($resolve, $reject) use ($client) { + $request = $client->fetch('/profile', $resolve, $reject); + + return fn () => $request->cancel(); + }); + }, + 'settings' => fn () => loadSettings(), +], timeout: 2.0); +``` + +The cleanup callback returned by `Task::await()` cancels the pending operation +when a sibling fails or the shared deadline expires. Task groups use the +configured React `LoopInterface`; when that loop is already running, call the +group from a Fiber so the parent Fiber can suspend without restarting or +stopping the loop. Blocking PDO, filesystem, and HTTP calls remain blocking. + ## Code of Conduct In order to ensure that the Doppar community is welcoming to all, please review and abide by the [Code of Conduct](https://doppar.com/versions/4.x/contributions#code-of-conduct). diff --git a/src/Phaseolies/Application.php b/src/Phaseolies/Application.php index f31b5034..7b1c2004 100644 --- a/src/Phaseolies/Application.php +++ b/src/Phaseolies/Application.php @@ -16,6 +16,9 @@ use Phaseolies\Config\Config; use Phaseolies\Support\DeferScope; use Phaseolies\Support\ExecutionContext; +use Phaseolies\Support\Concurrency\TaskGroup; +use React\EventLoop\Loop; +use React\EventLoop\LoopInterface; use Phaseolies\ApplicationBuilder; use Devium\Toml\Toml; use RuntimeException; @@ -821,6 +824,8 @@ protected function bindSingletonClasses(): void $this->singleton('request', fn() => Request::capture()); + $this->singleton(LoopInterface::class, fn() => Loop::get()); + $this->bindHttpGateway(); $this->singleton('route', Router::class); @@ -1547,6 +1552,8 @@ public function dispatch($request): DispatchResult $response = $this->handle($request); + TaskGroup::assertNothingPending($context->id()); + $response->prepare($request)->send(); $this->suspendDeferScope($deferScope); @@ -1554,6 +1561,8 @@ public function dispatch($request): DispatchResult return new DispatchResult($this, $request, $response, null, $deferScope, $context); } catch (HttpException $e) { + TaskGroup::cancelPending($e, $context->id()); + try { Response::dispatchHttpException($e); } catch (\Throwable $renderFailure) { @@ -1568,6 +1577,7 @@ public function dispatch($request): DispatchResult return new DispatchResult($this, $request, null, $e, $deferScope, $context); } catch (\Throwable $e) { + TaskGroup::cancelPending($e, $context->id()); $this->endDeferScope($deferScope, $e); $this->endContext($context, $e); diff --git a/src/Phaseolies/Support/Concurrency/Exceptions/DeadlineExceededException.php b/src/Phaseolies/Support/Concurrency/Exceptions/DeadlineExceededException.php new file mode 100644 index 00000000..ce5ee0e8 --- /dev/null +++ b/src/Phaseolies/Support/Concurrency/Exceptions/DeadlineExceededException.php @@ -0,0 +1,24 @@ +timeout; + } +} diff --git a/src/Phaseolies/Support/Concurrency/Exceptions/TaskCancelledException.php b/src/Phaseolies/Support/Concurrency/Exceptions/TaskCancelledException.php new file mode 100644 index 00000000..cd609b6e --- /dev/null +++ b/src/Phaseolies/Support/Concurrency/Exceptions/TaskCancelledException.php @@ -0,0 +1,17 @@ +getMessage()}", + 0, + $failure + ); + } + + /** + * Get the name of the failed task. + * + * @return string + */ + public function taskName(): string + { + return $this->taskName; + } +} diff --git a/src/Phaseolies/Support/Concurrency/Task.php b/src/Phaseolies/Support/Concurrency/Task.php new file mode 100644 index 00000000..c5a782d8 --- /dev/null +++ b/src/Phaseolies/Support/Concurrency/Task.php @@ -0,0 +1,323 @@ +name; + } + + /** + * Suspend the task until an asynchronous operation settles. + * + * The subscriber receives resolve and reject callbacks and may return a + * cleanup callback that is invoked if the task is cancelled while waiting. + * + * @param callable(callable(mixed): void, callable(\Throwable): void): mixed $subscriber + * @return mixed + */ + public function await(callable $subscriber): mixed + { + if ($this->cancellationRequested) { + throw $this->cancellation; + } + + if (\Fiber::getCurrent() !== $this->fiber || $this->state !== self::RUNNING) { + throw new \LogicException('Task::await() may only be called by its running task.'); + } + + $settled = false; + $value = null; + $error = null; + + $resolve = function (mixed $resolved = null) use (&$settled, &$value): void { + if ($settled || $this->terminal()) { + return; + } + + $settled = true; + $value = $resolved; + + if ($this->state === self::SUSPENDED) { + $this->group->resume($this, $resolved); + } + }; + + $reject = function (\Throwable $failure) use (&$settled, &$error): void { + if ($settled || $this->terminal()) { + return; + } + + $settled = true; + $error = $failure; + + if ($this->state === self::SUSPENDED) { + $this->group->reject($this, $failure); + } + }; + + $cleanup = $subscriber($resolve, $reject); + + if ($cleanup !== null && !is_callable($cleanup)) { + throw new \LogicException('A task await subscriber must return a callable or null.'); + } + + if ($settled) { + if ($error !== null) { + throw $error; + } + + return $value; + } + + $this->cancelSuspension = $cleanup; + $this->state = self::SUSPENDED; + + try { + return \Fiber::suspend(); + } finally { + $this->cancelSuspension = null; + + if (!$this->terminal()) { + $this->state = self::RUNNING; + } + } + } + + /** + * Start executing the task. + * + * @return void + */ + public function start(): void + { + if ($this->state !== self::PENDING) { + return; + } + + $this->fiber = new \Fiber(function (): mixed { + return ($this->callback)($this); + }); + + $this->state = self::RUNNING; + + try { + $this->fiber->start(); + $this->captureCompletion(); + } catch (\Throwable $failure) { + $this->fail($failure); + } + } + + /** + * Resume a suspended task with a resolved value. + * + * @param mixed $value + * @return void + */ + public function resumeWith(mixed $value): void + { + if ($this->state !== self::SUSPENDED || $this->fiber === null) { + return; + } + + $this->state = self::RUNNING; + + try { + $this->fiber->resume($value); + $this->captureCompletion(); + } catch (\Throwable $failure) { + $this->fail($failure); + } + } + + /** + * Resume a suspended task by throwing an asynchronous failure into it. + * + * @param \Throwable $failure + * @return void + */ + public function rejectWith(\Throwable $failure): void + { + if ($this->state !== self::SUSPENDED || $this->fiber === null) { + return; + } + + $this->state = self::RUNNING; + + try { + $this->fiber->throw($failure); + $this->captureCompletion(); + } catch (\Throwable $thrown) { + $this->fail($thrown); + } + } + + /** + * Cancel this task if it has not already settled. + * + * @param \Throwable|null $reason + * @return void + */ + public function cancel(?\Throwable $reason = null): void + { + if ($this->terminal()) { + return; + } + + $cancellation = new TaskCancelledException($this->name, $reason); + + $this->cancellationRequested = true; + $this->cancellation = $cancellation; + + if ($this->state === self::SUSPENDED && $this->fiber !== null) { + if ($this->cancelSuspension !== null) { + try { + ($this->cancelSuspension)(); + } catch (\Throwable) { + // Cancellation cleanup must not replace the group failure. + } + } + + $this->rejectWith($cancellation); + + return; + } + + $this->state = self::CANCELLED; + $this->failure = $cancellation; + $this->group->settled($this); + } + + /** + * Determine whether the task has reached a terminal state. + * + * @return bool + */ + public function terminal(): bool + { + return in_array($this->state, [self::SUCCEEDED, self::FAILED, self::CANCELLED], true); + } + + /** + * Get the successful task result. + * + * @return mixed + */ + public function result(): mixed + { + return $this->result; + } + + /** + * Get the task failure, if one exists. + * + * @return \Throwable|null + */ + public function failure(): ?\Throwable + { + return $this->failure; + } + + /** + * Store a Fiber return value when execution has completed. + * + * @return void + */ + private function captureCompletion(): void + { + if ($this->fiber === null || !$this->fiber->isTerminated()) { + return; + } + + if ($this->cancellationRequested) { + $this->state = self::CANCELLED; + $this->failure = $this->cancellation; + $this->group->settled($this); + + return; + } + + $this->result = $this->fiber->getReturn(); + $this->state = self::SUCCEEDED; + $this->group->settled($this); + } + + /** + * Mark the task as failed. + * + * @param \Throwable $failure + * @return void + */ + private function fail(\Throwable $failure): void + { + if ($failure instanceof TaskCancelledException) { + $this->state = self::CANCELLED; + } else { + $this->state = self::FAILED; + } + + $this->failure = $failure; + $this->group->settled($this); + } +} diff --git a/src/Phaseolies/Support/Concurrency/TaskGroup.php b/src/Phaseolies/Support/Concurrency/TaskGroup.php new file mode 100644 index 00000000..e26834cc --- /dev/null +++ b/src/Phaseolies/Support/Concurrency/TaskGroup.php @@ -0,0 +1,382 @@ +> + */ + private static array $active = []; + + /** + * Tasks keyed by their unique names. + * + * @var array + */ + private array $tasks = []; + + /** Whether this group has started. */ + private bool $started = false; + + /** Whether every task has settled. */ + private bool $finished = false; + + /** Primary failure that ended the group. */ + private ?\Throwable $failure = null; + + /** Name of the task that produced the primary failure. */ + private ?string $failedTask = null; + + /** Shared deadline timer for the group. */ + private ?TimerInterface $deadlineTimer = null; + + /** Event loop used to schedule this group. */ + private LoopInterface $loop; + + /** Fiber waiting while an already-running loop owns execution. */ + private ?\Fiber $waitingFiber = null; + + /** Execution-context identifier that owns this group. */ + private ?string $scopeId = null; + + /** Whether this group started and therefore owns the loop run. */ + private bool $drivesLoop = false; + + /** + * @param float|null $timeout Maximum runtime in seconds + * @param LoopInterface|null $loop + */ + public function __construct( + private readonly ?float $timeout = null, + ?LoopInterface $loop = null + ) { + if ($timeout !== null && $timeout < 0) { + throw new \InvalidArgumentException('A task group timeout cannot be negative.'); + } + + $this->loop = $loop ?? self::resolveLoop(); + } + + /** + * Run a named set of tasks and return their results. + * + * @param array $tasks + * @param float|null $timeout Maximum runtime in seconds + * @param LoopInterface|null $loop + * @return array + */ + public static function run( + array $tasks, + ?float $timeout = null, + ?LoopInterface $loop = null + ): array { + $group = new self($timeout, $loop); + + foreach ($tasks as $name => $task) { + $group->spawn((string) $name, $task); + } + + return $group->await(); + } + + /** + * Add a named task to the group before it starts. + * + * @param string $name + * @param callable $task + * @return $this + */ + public function spawn(string $name, callable $task): static + { + if ($this->started) { + throw new \LogicException('Tasks cannot be added after a task group has started.'); + } + + if (trim($name) === '') { + throw new \InvalidArgumentException('A task name cannot be empty.'); + } + + if (isset($this->tasks[$name])) { + throw new \InvalidArgumentException("Task [{$name}] is already registered."); + } + + $this->tasks[$name] = new Task($name, $task, $this); + + return $this; + } + + /** + * Start every task and wait until the group settles. + * + * @return array + * + * @throws TaskFailedException + * @throws DeadlineExceededException + */ + public function await(): array + { + if ($this->started) { + throw new \LogicException('A task group may only be awaited once.'); + } + + $this->started = true; + + if ($this->tasks === []) { + $this->finished = true; + + return []; + } + + $this->scopeId = self::currentScopeId(); + + if ((self::$active[$this->scopeId] ?? []) !== []) { + throw new \LogicException('Nested task groups are not supported.'); + } + + $id = spl_object_id($this); + self::$active[$this->scopeId][$id] = $this; + + try { + if ($this->timeout !== null) { + $this->deadlineTimer = $this->loop->addTimer($this->timeout, function (): void { + $this->deadlineExceeded(); + }); + } + + $this->loop->futureTick(function (): void { + foreach ($this->tasks as $task) { + $task->start(); + } + }); + + $currentFiber = \Fiber::getCurrent(); + + if ($currentFiber !== null) { + $this->waitingFiber = $currentFiber; + \Fiber::suspend(); + } else { + $this->drivesLoop = true; + $this->loop->run(); + } + + if (!$this->finished) { + $failure = new \RuntimeException( + 'The event loop stopped before every task in the group settled.' + ); + $this->cancel($failure); + $this->failure = $failure; + } + + if ($this->failure instanceof DeadlineExceededException) { + throw $this->failure; + } + + if ($this->failure !== null && $this->failedTask !== null) { + throw new TaskFailedException($this->failedTask, $this->failure); + } + + if ($this->failure !== null) { + throw $this->failure; + } + + $results = []; + + foreach ($this->tasks as $name => $task) { + $results[$name] = $task->result(); + } + + return $results; + } finally { + $this->cancelDeadlineTimer(); + unset(self::$active[$this->scopeId][$id]); + + if (self::$active[$this->scopeId] === []) { + unset(self::$active[$this->scopeId]); + } + } + } + + /** + * Schedule a suspended task to resume with a value. + * + * @param Task $task + * @param mixed $value + * @return void + * @internal + */ + public function resume(Task $task, mixed $value): void + { + $task->resumeWith($value); + } + + /** + * Schedule an asynchronous failure inside a suspended task. + * + * @param Task $task + * @param \Throwable $failure + * @return void + * @internal + */ + public function reject(Task $task, \Throwable $failure): void + { + $task->rejectWith($failure); + } + + /** + * Record a terminal task and settle or cancel the group as needed. + * + * @param Task $task + * @return void + * @internal + */ + public function settled(Task $task): void + { + $failure = $task->failure(); + + if ($failure !== null && !$failure instanceof TaskCancelledException && $this->failure === null) { + $this->failure = $failure; + $this->failedTask = $task->name(); + $this->cancel($failure, $task); + } + + foreach ($this->tasks as $candidate) { + if (!$candidate->terminal()) { + return; + } + } + + $this->finished = true; + $this->cancelDeadlineTimer(); + + if ($this->waitingFiber !== null) { + $waitingFiber = $this->waitingFiber; + $this->loop->futureTick(function () use ($waitingFiber): void { + if ($waitingFiber->isSuspended()) { + $waitingFiber->resume(); + } + }); + } elseif ($this->drivesLoop) { + $this->loop->stop(); + } + } + + /** + * Cancel every unsettled task in the group. + * + * @param \Throwable|null $reason + * @param Task|null $except + * @return void + */ + public function cancel(?\Throwable $reason = null, ?Task $except = null): void + { + if ($except === null && $this->failure === null) { + $this->failure = $reason ?? new \RuntimeException('The task group was cancelled.'); + } + + foreach ($this->tasks as $task) { + if ($task !== $except) { + $task->cancel($reason); + } + } + } + + /** + * Cancel every task group still active in the current process. + * + * @param \Throwable|null $reason + * @param string|null $scopeId + * @return void + */ + public static function cancelPending(?\Throwable $reason = null, ?string $scopeId = null): void + { + $scopeId ??= self::currentScopeId(); + + foreach (self::$active[$scopeId] ?? [] as $group) { + $group->cancel($reason); + } + } + + /** + * Ensure no child task group remains active before a request finishes. + * + * @param string|null $scopeId + * @return void + */ + public static function assertNothingPending(?string $scopeId = null): void + { + $scopeId ??= self::currentScopeId(); + + if ((self::$active[$scopeId] ?? []) !== []) { + throw new \LogicException('A request cannot finish while child tasks remain registered.'); + } + } + + /** + * Fail the group when its shared deadline expires. + * + * @return void + */ + private function deadlineExceeded(): void + { + if ($this->finished || $this->failure !== null) { + return; + } + + $this->failure = new DeadlineExceededException($this->timeout ?? 0.0); + $this->cancel($this->failure); + } + + /** + * Cancel the shared deadline timer once the group settles. + * + * @return void + */ + private function cancelDeadlineTimer(): void + { + if ($this->deadlineTimer === null) { + return; + } + + $this->loop->cancelTimer($this->deadlineTimer); + $this->deadlineTimer = null; + } + + /** + * Resolve the configured event loop, falling back to React's default loop. + * + * @return LoopInterface + */ + private static function resolveLoop(): LoopInterface + { + $container = Container::getInstance(); + + if ($container->scopeOf(LoopInterface::class) !== null) { + return $container->make(LoopInterface::class); + } + + return Loop::get(); + } + + /** + * Resolve the current execution-context identifier. + * + * @return string + */ + private static function currentScopeId(): string + { + $context = Container::getInstance()->executionContext(); + + return $context?->id() ?? 'standalone'; + } +} diff --git a/tests/Application/ApplicationTest.php b/tests/Application/ApplicationTest.php index 3384768e..3718fc0f 100644 --- a/tests/Application/ApplicationTest.php +++ b/tests/Application/ApplicationTest.php @@ -21,6 +21,11 @@ use PHPUnit\Framework\Attributes\AllowMockObjectsWithoutExpectations; use Phaseolies\Support\StringService; use Phaseolies\Support\View\Factory as ViewFactory; +use Phaseolies\Support\Concurrency\Task; +use Phaseolies\Support\Concurrency\TaskGroup; +use React\EventLoop\Loop; +use React\EventLoop\LoopInterface; +use React\EventLoop\StreamSelectLoop; if (!class_exists('App\Http\Gateway')) { class_alias(Gateway::class, 'App\Http\Gateway'); @@ -450,6 +455,73 @@ function ( $this->assertNull($captured[2]); } + public function testDispatchWaitsForTaskGroupsBeforeSendingTheResponse(): void + { + $request = new Request(); + $taskFinished = false; + + $response = $this->getMockBuilder(Response::class) + ->disableOriginalConstructor() + ->onlyMethods(['prepare', 'send']) + ->getMock(); + $response->method('prepare')->willReturnSelf(); + $response->expects($this->once()) + ->method('send') + ->willReturnCallback(function () use (&$taskFinished, $response) { + $this->assertTrue($taskFinished); + + return $response; + }); + + $router = $this->createMock(Router::class); + $router->method('resolve')->willReturnCallback( + function () use ($response, &$taskFinished) { + $requestContext = $this->app->executionContext(); + + TaskGroup::run([ + 'required-data' => function (Task $task) use (&$taskFinished, $requestContext) { + $this->assertSame($requestContext, $this->app->executionContext()); + + return $task->await(function ($resolve) use (&$taskFinished) { + $timer = Loop::addTimer(0.001, function () use ($resolve, &$taskFinished) { + $taskFinished = true; + $resolve(); + }); + + return fn () => Loop::cancelTimer($timer); + }); + }, + ]); + + return $response; + } + ); + $this->app->router = $router; + + $this->app->dispatch($request)->terminate(); + } + + public function testApplicationBindsTheConfiguredEventLoopContract(): void + { + $this->assertSame(Loop::get(), $this->app->make(LoopInterface::class)); + } + + public function testTaskGroupsResolveAnOverriddenEventLoopContract(): void + { + $loop = new StreamSelectLoop(); + $this->app->instance(LoopInterface::class, $loop); + + $results = TaskGroup::run([ + 'configured' => function (Task $task) use ($loop) { + return $task->await(function ($resolve) use ($loop) { + $loop->futureTick(fn () => $resolve('result')); + }); + }, + ]); + + $this->assertSame(['configured' => 'result'], $results); + } + public function testDispatchStillTerminatesWhenResultIsIgnored(): void { $request = new Request(); diff --git a/tests/Support/Concurrency/TaskGroupTest.php b/tests/Support/Concurrency/TaskGroupTest.php new file mode 100644 index 00000000..81ef5b6c --- /dev/null +++ b/tests/Support/Concurrency/TaskGroupTest.php @@ -0,0 +1,211 @@ + function (Task $task) use (&$order) { + $order[] = 'slow:start'; + + return $task->await(function ($resolve) { + Loop::futureTick(fn () => $resolve('slow result')); + }); + }, + 'fast' => function () use (&$order) { + $order[] = 'fast'; + + return 'fast result'; + }, + ]); + + $this->assertSame([ + 'slow' => 'slow result', + 'fast' => 'fast result', + ], $results); + $this->assertSame(['slow:start', 'fast'], $order); + } + + public function testNamedFailureCancelsSuspendedSiblings(): void + { + $cancelled = false; + $timer = null; + + try { + TaskGroup::run([ + 'waiting' => function (Task $task) use (&$cancelled, &$timer) { + try { + return $task->await(function ($resolve) use (&$timer) { + $timer = Loop::addTimer(10, fn () => $resolve('late')); + + return fn () => Loop::cancelTimer($timer); + }); + } catch (TaskCancelledException $cancellation) { + $cancelled = true; + + throw $cancellation; + } + }, + 'broken' => fn () => throw new \RuntimeException('boom'), + ]); + + $this->fail('The task group should have failed.'); + } catch (TaskFailedException $failure) { + $this->assertSame('broken', $failure->taskName()); + $this->assertSame('boom', $failure->getPrevious()?->getMessage()); + $this->assertTrue($cancelled); + } + } + + public function testCancellationCannotBeSwallowedByASibling(): void + { + $cleanupRan = false; + $continued = false; + + try { + TaskGroup::run([ + 'waiting' => function (Task $task) use (&$cleanupRan, &$continued) { + try { + $task->await(function ($resolve) use (&$cleanupRan) { + $timer = Loop::addTimer(10, fn () => $resolve('late')); + + return function () use ($timer, &$cleanupRan): void { + Loop::cancelTimer($timer); + $cleanupRan = true; + }; + }); + } catch (TaskCancelledException) { + $continued = true; + } + + return 'ignored'; + }, + 'broken' => fn () => throw new \RuntimeException('boom'), + ]); + + $this->fail('The task group should have failed.'); + } catch (TaskFailedException $failure) { + $this->assertSame('broken', $failure->taskName()); + $this->assertTrue($cleanupRan); + $this->assertTrue($continued); + } + } + + public function testSharedDeadlineCancelsTheGroup(): void + { + $this->expectException(DeadlineExceededException::class); + + TaskGroup::run([ + 'waiting' => function (Task $task) { + return $task->await(function ($resolve) { + $timer = Loop::addTimer(10, fn () => $resolve('late')); + + return fn () => Loop::cancelTimer($timer); + }); + }, + ], 0.0); + } + + public function testItRejectsDuplicateAndLateTasks(): void + { + $group = new TaskGroup(); + $group->spawn('one', fn () => 1); + + try { + $group->spawn('one', fn () => 2); + $this->fail('Duplicate task names should be rejected.'); + } catch (\InvalidArgumentException) { + $this->assertSame(['one' => 1], $group->await()); + } + + $this->expectException(\LogicException::class); + $group->spawn('late', fn () => 3); + } + + public function testItRejectsNestedTaskGroups(): void + { + try { + TaskGroup::run([ + 'outer' => fn () => TaskGroup::run([ + 'inner' => fn () => 'result', + ]), + ]); + + $this->fail('Nested task groups should be rejected.'); + } catch (TaskFailedException $failure) { + $this->assertSame('outer', $failure->taskName()); + $this->assertInstanceOf(\LogicException::class, $failure->getPrevious()); + } + } + + public function testItUsesAnInjectedEventLoop(): void + { + $loop = new StreamSelectLoop(); + + $results = TaskGroup::run([ + 'injected' => function (Task $task) use ($loop) { + return $task->await(function ($resolve) use ($loop) { + $loop->futureTick(fn () => $resolve('result')); + }); + }, + ], loop: $loop); + + $this->assertSame(['injected' => 'result'], $results); + } + + public function testItSuspendsInsideAnAlreadyRunningLoopWithoutStoppingIt(): void + { + $loop = new StreamSelectLoop(); + $results = null; + $loopContinued = false; + + $loop->futureTick(function () use ($loop, &$results): void { + $fiber = new \Fiber(function () use ($loop, &$results): void { + $results = TaskGroup::run([ + 'embedded' => function (Task $task) use ($loop) { + return $task->await(function ($resolve) use ($loop) { + $timer = $loop->addTimer(0.001, fn () => $resolve('result')); + + return fn () => $loop->cancelTimer($timer); + }); + }, + ], loop: $loop); + }); + + $fiber->start(); + }); + + $loop->futureTick(function (): void { + TaskGroup::cancelPending(new \RuntimeException('other request'), 'other-scope'); + TaskGroup::assertNothingPending('other-scope'); + }); + + $loop->addTimer(0.002, function () use (&$loopContinued): void { + $loopContinued = true; + }); + + $loop->run(); + + $this->assertSame(['embedded' => 'result'], $results); + $this->assertTrue($loopContinued); + } + + protected function tearDown(): void + { + TaskGroup::cancelPending(); + TaskGroup::assertNothingPending(); + } +}