Http:pool with concurrency limit?

[sfglobaldata]

Hey all,

Is there a way of allowing for a concurrency limit when using Http:pool ?

My issue is that I have several hundred requests to make, but I want to tread softly on the API, so I want to only have a certain number of requests in flight at one time.

I know that I can use Guzzle directly to do this using GuzzleHttp\Pool (as per https://docs.guzzlephp.org/en/stable/quickstart.html#concurrent-requests) but I would prefer to stick with the Http::pool method.

My current best solution is just to load up the pool with requests up to my concurrency limit, then wait() for them to finish like this:

Http::pool(function (Pool $pool) {
    $currentRequests = [];
    foreach ($this->requestsUrls as $requestUrl) {
        $currentRequests[] = $pool
            ->get(requestUrl)
            ->then(fn ($response) => $this->processResponse($response, $requestUrl));

        while (count($currentRequests) >= $this->concurrencyLimit) {
            $n = array_key_first($currentRequests);
            $currentRequests[$n]->wait();
            unset($currentRequests[$n]);
        }
    }
});

In the while loop I am just arbitrarily waiting on the first request. It does not seem to matter which one I wait for as any call to wait() will actually resolve all pending promises in the pool, so the very first wait() will wait for all requests to complete, not just the first call.

This has the effect of spooling up the max number of concurrent requests, then waiting on them all to complete and repeating. The solution works at keeping me below my concurrency limit, but it not very efficient.

Ideally I want to launch a new request when one request completes, therefore keeping at the concurrency limit until all calls are complete.

Does anyone have any ideas or pointers on how I might make this work?

Cheers,
Sab,

[joaoneres]

Hey! Did you manage to find anything related? I need to solve the same problem.

[henzeb]

I have a similar solution, although mine is a while loop at the outside, creating new pools as we go. Sadly Laravel's Http doesn't support concurrency limits as well as generator functions....

[henzeb]

I've solved it. code below has the same outcome as without Each::ofLimit, but when the concurrency of 10 is set to 1, it will run one by one. When set to 5 it will automatically add a new request whenever a request is fullfilled.

use Http;
use GuzzleHttp\Promise\Each;
use Illuminate\Http\Client\Pool;

Http::pool(function (Pool $pool) {
    return [
        Each::ofLimit(
            (function () use ($pool) {
                for ($i = 1; $i <= 10; $i++) {
                    yield $pool->async()
                        ->get('https://httpbin.org/delay/' . $i)
                        ->then(fn() => dump($i . ' done'));
                }
            })(),
            10 // this is the concurrency
        )
    ];
});

For anyone interested, it all boils down to the following Macroable mixin:

use Closure;
use GuzzleHttp\Promise\Each;
use Illuminate\Http\Client\Pool;
use Illuminate\Http\Client\Factory;

class HttpMixin
{
    public function concurrent(): Closure
    {
        return function (
            int $concurrency,
            callable $requests,
            callable $onFulfilled = null,
            callable $onRejected = null
        ): void {
            /**
             * @var $this Factory
             */
            $requests = $requests(...)(new Pool($this));

            Each::ofLimit(
                $requests,
                $concurrency,
                $onFulfilled,
                $onRejected
            )->wait();
        };
    }
}

usage:

use Http;
use Generator;
use Illuminate\Http\Client\Pool;

Http::mixin(new HttpMixin());

Http::concurrent(
    10,
    function (Pool $pool): Generator {
        for ($i = 1; $i <= 10; $i++) {
            yield $pool->async()
                ->get('https://httpbin.org/delay/' . $i)
                ->then(fn() => dump($i . ' done'));
        }
    }
);

No need for Http::pool anymore, as the only thing we need is the Pool instance and we can construct that ourselves. Under water the methods used on Pool are creating a PendingRequest. Each PendingRequest created by the Pool instance is receiving the same MultiCurlHandler instance as the rest. this finally returns a promise, which we put in an EachPromise using Each::forLimit which in his turn handles the concurrency.

Note: I used ->async() as I wanted to access the promise and needed the correct typehinting. It is not really needed if you do not want to access the promise from there.

[henzeb]

yes. Pool is just Pool you get when you would use Http::pool. what you can do inside Http::Pool, can be done here as well.

[ghost]

@henzeb well,. in my case when use a post request and add a parameter to the route and add an array of data, nobody appear on the request in the controller, when I use get, every work fine,

[henzeb]

I am not sure what you mean, but the macro I wrote is not meant to show output. It's more for crawler purposes. If you have 100s of requests, you will get the result of just the last 10 anyway. But if you really need output and throttle the concurrency, you can use the pool with the single promise instead.

[jonnywilliamson]

@henzeb Just wanted to drop you a note to say your code was such a help! Thank you.

For whatever reason, using the Guzzle Http pool from their documentation for using concurrent connection only gave me some marginal performance benefits...requests were being sent out concurrently but the entire process still took far too long...less than sequential request but still very slow.

As soon as I switched to your Each::ofLimit method it was like night and day, massive massive speed increase.

I know Taylor and @nunomaduro like to sometimes look into Laravel "papercuts" and things they can improve in the library. This certainly seems like a really good candidate for one of those. Yes the current HTTP client can cope with sending a pool of requests, but that doesn't scale at all, there's just no concurrency.

If someone has hundreds of requests to send out, and wants to maintain a steady amount of concurrent connections to the server but not bombard it with ALL of them at once, it's very ugly using the HTTP client at the moment. I hope the code above is a great stepping stone into having something 1st party in the framework sometime in the future.

[meduzen]

Ended up doing a slightly more simple and slightly less performant with chunks:

<?php

$ids->chunk(4) // concurrency is 4
    ->each(function ($chunk) {
        $responses = Http::pool(fn (Pool $pool) => $chunk
            ->map(fn ($id) => $pool
                ->as($id)
                ->get("https://domain.tld.com/endpoint/{$id}")
            ));

        collect($responses)->each(function ($res, $id) {
            if (! $res->ok()) {
                Log::notice('Ping to endpoint failed', [
                    'id' => $id,
                    'failing_url' => $res->effectiveUri(),
                    'status' => $res->status(),
                ]);
            } else {
                // success, do something
            }
        });
    });

For the record, someone opened and closed a PR 2 years ago.

I couldn't find any issue for a feature request.