From 8c8c1107051d56326e00c1740e55d2a6d482294c Mon Sep 17 00:00:00 2001 From: Yevhen Sidelnyk Date: Sat, 27 Jun 2026 11:28:57 +0300 Subject: [PATCH] Add `concurrent()` and `settle()` functions --- README.md | 67 ++++++++++++++++---- src/Future/functions.php | 29 +++++++++ src/functions.php | 17 +++++ test/ConcurrentTest.php | 50 +++++++++++++++ test/Future/SettleTest.php | 126 +++++++++++++++++++++++++++++++++++++ 5 files changed, 276 insertions(+), 13 deletions(-) create mode 100644 test/ConcurrentTest.php create mode 100644 test/Future/SettleTest.php diff --git a/README.md b/README.md index 66f1b24c..3c53487f 100644 --- a/README.md +++ b/README.md @@ -219,7 +219,20 @@ throws an exception. #### Combinators -In concurrent applications, there will be multiple futures, where you might want to await them all or just the first one. +In concurrent applications, there will be multiple futures, where you might want to await them all or just the first +one. + +You can create a bunch of futures by applying `Amp\concurrent()` to an array of closures: +it returns a `Future` for each, preserving the keys. + +```php +$firstReachedApi = [ + fn () => $httpClient->request(new Request('https://a.api.com', 'HEAD')), + fn () => $httpClient->request(new Request('https://b.api.com', 'HEAD')), +] |> \Amp\concurrent(...) |> \Amp\Future\awaitAny(...); +``` + +The combinators below await such futures in different ways. ##### await @@ -241,17 +254,16 @@ use Amp\Http\Client\Request; require __DIR__ . '/vendor/autoload.php'; $httpClient = HttpClientBuilder::buildDefault(); -$uris = [ - "google" => "https://www.google.com", - "news" => "https://news.google.com", - "bing" => "https://www.bing.com", - "yahoo" => "https://www.yahoo.com", -]; + +$futures = Amp\concurrent([ + "google" => fn () => $httpClient->request(new Request("https://www.google.com", 'HEAD')), + "news" => fn () => $httpClient->request(new Request("https://news.google.com", 'HEAD')), + "bing" => fn () => $httpClient->request(new Request("https://www.bing.com", 'HEAD')), + "yahoo" => fn () => $httpClient->request(new Request("https://www.yahoo.com", 'HEAD')), +]); try { - $responses = Future\await(array_map(function ($uri) use ($httpClient) { - return Amp\async(fn () => $httpClient->request(new Request($uri, 'HEAD'))); - }, $uris)); + $responses = Future\await($futures); foreach ($responses as $key => $response) { printf( @@ -270,9 +282,38 @@ try { ##### awaitAnyN -`Amp\Future\awaitAnyN($count, $iterable, $cancellation)` is the same as `await()` except that it tolerates individual errors. A result is returned once -exactly `$count` instances in the `iterable` complete successfully. The return value is an array of values. The -individual keys in the component array are preserved from the `iterable` passed to the function for evaluation. +`Amp\Future\awaitAnyN($count, $iterable, $cancellation)` is the same as `await()` except that it tolerates individual errors. +A result is returned once exactly `$count` instances in the `iterable` complete successfully, or `CompositeException` is thrown otherwise. +The return value is an array of values with the individual keys preserved from the `iterable` passed to the function for evaluation. + +##### settle + +`Amp\Future\settle($iterable, $cancellation)` is the same as `await()` except that it waits for all the futures to finish (either successfully complete or error), +not failing right off with the first error. + +A result is returned only if all the futures complete successfully. +If at least one of them errors, a `CompositeException` comprised of all the errors is thrown. + +```php +use Amp\CompositeException; +use function Amp\concurrent; +use function Amp\Future\settle; + +try { + $services = [ + 'db' => fn () => $pool->connect(), + 'redis' => fn () => $cache->connect(), + 'broker' => fn () => $amqp->connect(), + 'secrets' => fn () => $vault->connect(), + ] |> concurrent(...) |> settle(...); +} catch (CompositeException $e) { + $failedServices = array_keys($e->getReasons()); + $message = sprintf('Services connection failed: %s.', implode(', ', $failedServices)); + + // Services connection failed: redis, broker. + throw new ServicesConnectionFailedException($message, $e->getReasons()); +} +``` ##### awaitAll diff --git a/src/Future/functions.php b/src/Future/functions.php index b383dd63..5f55154f 100644 --- a/src/Future/functions.php +++ b/src/Future/functions.php @@ -155,3 +155,32 @@ function await(iterable $futures, ?Cancellation $cancellation = null): array /** @var array */ return $values; } + +/** + * Waits for all futures to finish (either complete or error), unlike {@see await()}, which aborts on the first error. + * + * If any errored, a {@see CompositeException} holding all errors (keyed as given) is thrown. + * Otherwise, the unwrapped values are returned. + * + * The returned values are ordered by completion, not by the input iterable, just as with {@see await()}. + * + * @template Tk of array-key + * @template Tv + * + * @param iterable> $futures + * @param Cancellation|null $cancellation Optional cancellation. + * + * @return array Unwrapped values, keyed as given. + * + * @throws CompositeException If one or more futures errored. + */ +function settle(iterable $futures, ?Cancellation $cancellation = null): array +{ + [$errors, $values] = awaitAll($futures, $cancellation); + + if ($errors) { + throw new CompositeException($errors); + } + + return $values; +} diff --git a/src/functions.php b/src/functions.php index eed32ddb..ceccebc7 100644 --- a/src/functions.php +++ b/src/functions.php @@ -43,6 +43,23 @@ function async(\Closure $closure, mixed ...$args): Future return new Future($state); } +/** + * Concurrently evaluates the given closures, returning a {@see Future} for each. + * + * Pass or pipe the result into a combinator such as {@see Future\await()} or {@see Future\settle()} to await the values. + * + * @template Tk of array-key + * @template Tv + * + * @param array $closures + * + * @return array> A Future for each closure, with keys preserved. + */ +function concurrent(array $closures): array +{ + return \array_map(async(...), $closures); +} + /** * Returns the current time relative to an arbitrary point in time. * diff --git a/test/ConcurrentTest.php b/test/ConcurrentTest.php new file mode 100644 index 00000000..597154e0 --- /dev/null +++ b/test/ConcurrentTest.php @@ -0,0 +1,50 @@ + fn () => 1, 'two' => fn () => 2]; + + $futures = concurrent($closures); + + self::assertContainsOnlyInstancesOf(Future::class, $futures); + self::assertSame(['one', 'two'], \array_keys($futures)); + } + + public function testClosuresAreEvaluated(): void + { + self::assertSame( + ['one' => 1, 'two' => 2], + Future\await(concurrent(['one' => fn () => 1, 'two' => fn () => 2])) + ); + } + + public function testClosuresAreEvaluatedConcurrently(): void + { + $order = []; + + $futures = concurrent([ + static function () use (&$order): void { + delay(0.02); + $order[] = 'slow'; + }, + static function () use (&$order): void { + delay(0.01); + $order[] = 'fast'; + }, + ]); + + Future\await($futures); + + // "fast" is declared second but completes first, proving the closures run concurrently. + self::assertSame(['fast', 'slow'], $order); + } +} diff --git a/test/Future/SettleTest.php b/test/Future/SettleTest.php new file mode 100644 index 00000000..26dd72a2 --- /dev/null +++ b/test/Future/SettleTest.php @@ -0,0 +1,126 @@ + Future::complete(1), + 'two' => Future::complete(2), + ]; + + self::assertSame( + ['one' => 1, 'two' => 2], + settle($futures), + ); + } + + public function testSingleError(): void + { + $exception = new \Exception('foo'); + $futures = [ + 'one' => Future::error($exception), + 'two' => Future::complete(2), + ]; + + try { + settle($futures); + + self::fail('Expected ' . CompositeException::class . ' to be thrown'); + } catch (CompositeException $composite) { + self::assertSame( + ['one' => $exception], + $composite->getReasons(), + ); + } + } + + public function testMultipleErrors(): void + { + $first = new \Exception('foo'); + $second = new \RuntimeException('bar'); + $futures = [ + Future::error($first), + Future::error($second), + ]; + + try { + settle($futures); + + self::fail('Expected ' . CompositeException::class . ' to be thrown'); + } catch (CompositeException $composite) { + self::assertSame( + [$first, $second], + $composite->getReasons(), + ); + } + } + + public function testWaitsForAllBeforeThrowing(): void + { + $immediate = new \Exception('foo'); + $delayed = new \RuntimeException('bar'); + + $deferred = new DeferredFuture; + EventLoop::delay(0.01, fn () => $deferred->error($delayed)); + + $futures = [ + 'delayed' => $deferred->getFuture(), + 'immediate' => Future::error($immediate), + ]; + + try { + settle($futures); + + self::fail('Expected ' . CompositeException::class . ' to be thrown'); + } catch (CompositeException $composite) { + $reasons = $composite->getReasons(); + + self::assertSame([ + 'immediate' => $immediate, + 'delayed' => $delayed, + ], $reasons); + } + } + + public function testCancellation(): void + { + $this->expectException(CancelledException::class); + + $deferreds = \array_map(function (int $value) { + $deferred = new DeferredFuture; + EventLoop::delay($value / 10, fn () => $deferred->complete($value)); + return $deferred; + }, \range(1, 3)); + + settle(\array_map( + fn (DeferredFuture $deferred) => $deferred->getFuture(), + $deferreds + ), new TimeoutCancellation(0.05)); + } +}