Multi-threaded downloading files with Guzzle HTTP client: EachPromises vs Pool objects

Question

for testing purposes, I have an array of 2000 image URIs (strings) with I download asynchronously with this functions. After some googling & testing & trying I've come up with 2 functions that both of them work (well to be honest downloadFilesAsync2 throws a InvalidArgumentException at the last line).

The function downloadFilesAsync2 is based on the class GuzzleHttp\Promise\EachPromise and downloadFilesAsync1 is based on the GuzzleHttp\Pool class.

Both functions download pretty well the 2000 files asynchronously, with the limit of 10 threads at the same time.

I know that they work, but nothing else. I wonder if someone could explain both aproaches, if one is better than the other, implications, etc.

// for the purpose of this question i've reduced the array to 5 files!
$uris = array /
  "https://cdn.enchufix.com/media/catalog/product/u/n/unix-48120.jpg",
  "https://cdn.enchufix.com/media/catalog/product/u/n/unix-48120-01.jpg",
  "https://cdn.enchufix.com/media/catalog/product/u/n/unix-48120-02.jpg",
  "https://cdn.enchufix.com/media/catalog/product/u/n/unix-48120-03.jpg",
  "https://cdn.enchufix.com/media/catalog/product/u/n/unix-48120-04.jpg",
);

function downloadFilesAsync2(array $uris, string $dir, $overwrite=true) {
    $client   = new \GuzzleHttp\Client();
    $requests = array();
    foreach ($uris as $i => $uri) {
        $loc = $dir . DIRECTORY_SEPARATOR . basename($uri);
        if ($overwrite && file_exists($loc)) unlink($loc);
        $requests[] = new GuzzleHttp\Psr7\Request('GET', $uri, ['sink' => $loc]);
        echo "Downloading $uri to $loc" . PHP_EOL;
    }
    $pool = new \GuzzleHttp\Pool($client, $requests, [
        'concurrency' => 10,
        'fulfilled' => function (\Psr\Http\Message\ResponseInterface $response, $index) {
            // this is delivered each successful response
            echo 'success: '.$response->getStatusCode().PHP_EOL;
        },
        'rejected' => function ($reason, $index) {
            // this is delivered each failed request
            echo 'failed: '.$reason.PHP_EOL;
        },
    ]);
    $promise = $pool->promise();  // Start transfers and create a promise
    $promise->wait();   // Force the pool of requests to complete.
}

function downloadFilesAsync1(array $uris, string $dir, $overwrite=true) {
    $client = new \GuzzleHttp\Client();
    $promises = (function () use ($client, $uris, $dir, $overwrite) {
        foreach ($uris as $uri) {
            $loc = $dir . DIRECTORY_SEPARATOR . basename($uri);
            if ($overwrite && file_exists($loc)) unlink($loc);
            yield $client->requestAsync('GET', $uri, ['sink' => $loc]);
            echo "Downloading $uri to $loc" . PHP_EOL;
        }
    })();
    (new \GuzzleHttp\Promise\EachPromise(
        $promises, [
        'concurrency' => 10,
        'fulfilled'   => function (\Psr\Http\Message\ResponseInterface $response) {
            //            echo "\t=>\tDONE! status:" . $response->getStatusCode() . PHP_EOL;
        },
        'rejected'    => function ($reason, $index) {
            echo 'ERROR => ' . strtok($reason->getMessage(), "\n") . PHP_EOL;
        },
    ])
    )->promise()->wait();
}

Answer 1

First, I will address the InvalidArgumentException within the downloadFilesAsync2 method. There are actually a pair of issues with this method. Both relate to this:

$requests[] = $client->request('GET', $uri, ['sink' => $loc]);

The first issue is the fact that Client::request() is a synchronous utility method which wraps $client->requestAsync()->wait(). $client->request() will return an instance of Psr\Http\Message\ResponseInterface, as a result $requests[] will actually be populated with ResponseInterface implementations. This is what, ultimately causes the InvalidArgumentException as the $requests does not contain any Psr\Http\Message\RequestInterface's, and the exception is thrown from within Pool::__construct().

A corrected version of this method should contain code which looks more like:

$requests = [
    new Request('GET', 'www.google.com', [], null, 1.1),
    new Request('GET', 'www.ebay.com', [], null, 1.1),
    new Request('GET', 'www.cnn.com', [], null, 1.1),
    new Request('GET', 'www.red.com', [], null, 1.1),
];

$pool = new Pool($client, $requests, [
    'concurrency' => 10,
    'fulfilled' => function(ResponseInterface $response) {
        // do something
    },
    'rejected' => function($reason, $index) {
        // do something error handling
    },
    'options' => ['sink' => $some_location,],
]);

$promise = $pool->promise();
$promise->wait();

To answer your second question, "What is the difference between these two methods", the answer is simply, there is none. To explain this, let me copy and paste Pool::__construct():

/**
 * @param ClientInterface $client   Client used to send the requests.
 * @param array|\Iterator $requests Requests or functions that return
 *                                  requests to send concurrently.
 * @param array           $config   Associative array of options
 *     - concurrency: (int) Maximum number of requests to send concurrently
 *     - options: Array of request options to apply to each request.
 *     - fulfilled: (callable) Function to invoke when a request completes.
 *     - rejected: (callable) Function to invoke when a request is rejected.
 */
public function __construct(
    ClientInterface $client,
    $requests,
    array $config = []
) {
    // Backwards compatibility.
    if (isset($config['pool_size'])) {
        $config['concurrency'] = $config['pool_size'];
    } elseif (!isset($config['concurrency'])) {
        $config['concurrency'] = 25;
    }

    if (isset($config['options'])) {
        $opts = $config['options'];
        unset($config['options']);
    } else {
        $opts = [];
    }

    $iterable = \GuzzleHttp\Promise\iter_for($requests);
    $requests = function () use ($iterable, $client, $opts) {
        foreach ($iterable as $key => $rfn) {
            if ($rfn instanceof RequestInterface) {
                yield $key => $client->sendAsync($rfn, $opts);
            } elseif (is_callable($rfn)) {
                yield $key => $rfn($opts);
            } else {
                throw new \InvalidArgumentException('Each value yielded by '
                    . 'the iterator must be a Psr7\Http\Message\RequestInterface '
                    . 'or a callable that returns a promise that fulfills '
                    . 'with a Psr7\Message\Http\ResponseInterface object.');
            }
        }
    };

    $this->each = new EachPromise($requests(), $config);
}

now if we compare that to an a simplified version of the code within the downloadFilesAsync1 method:

$promises = (function () use ($client, $uris) {
    foreach ($uris as $uri) {
        yield $client->requestAsync('GET', $uri, ['sink' => $some_location]);
    }
})();
(new \GuzzleHttp\Promise\EachPromise(
    $promises, [
    'concurrency' => 10,
    'fulfilled'   => function (\Psr\Http\Message\ResponseInterface $response) {
        // do something
    },
    'rejected'    => function ($reason, $index) {
        // do something
    },
])
)->promise()->wait();

In both examples, there is a generator which yields promises that resolve to instances of ResponseInterface and that generator along with the configuration array (fulfilled callable, rejected callable, concurrency) is also fed into a new instance of EachPromise.

In summary:

1. downloadFilesAsync1 is functionally the same thing as using Pool only without the error checking that has been built into Pool::__construct().

2. There are a few errors within downloadFilesAsync2 which will cause the files to be downloaded in a synchronous fashion prior to receiving an InvalidArgumentException when the Pool is instantiated.

My only recommendation is: use whichever feels more intuitive for you to use.