How to use amphp/parallel library for non-blocking process

Question

I want to use the amphp/parallel library for non-blocking process. I have a simple download file function which does a curl hit to the remote image file and save it to the local. I'm hitting this method through a REST API. Basically I want a process where aysnc download should be done on backend and it could be said as, REST API hit the function and function says "Hey, OK I'm downloading in background you can proceed ahead". Means non-blocking and API gets response as ok , not to wait. Meanwhile, if there is some network failure onto download, worker can restart the process in some time. How do I start?

I have tried the following code, but did not work.

require_once "vendor/autoload.php";
use Amp\Loop;
use Amp\Parallel\Worker\CallableTask;
use Amp\Parallel\Worker\DefaultWorkerFactory;

\Amp\Loop::run(function () {
  $remote_file_url = "some remote image url"; //http://example.com/some.png
  $file_save_path = "save path for file"; //var/www/html/some.png
  $factory = new DefaultWorkerFactory();
  $worker = $factory->create();
  $result = yield $worker->enqueue(new CallableTask('downloadFile', [$remote_file_url, $file_save_path]));

  $code = yield $worker->shutdown();

  });

//downloadFile is a simple download function

function downloadFile($remoteFile, $localFile) {
    if (!$remoteFile || !$localFile) {
        return;
    }
    set_time_limit(0);
    $fp = fopen($localFile, 'w+');
    $ch = curl_init();
    curl_setopt($ch, CURLOPT_URL, $remoteFile);
    curl_setopt($ch, CURLOPT_TIMEOUT, 50);
    curl_setopt($ch, CURLOPT_FILE, $fp);
    curl_setopt($ch, CURLOPT_FOLLOWLOCATION, true);
    $result = curl_exec($ch);
    curl_close($ch);
    fclose($fp);

    return $result ? true : false;
}

I'm getting this error:

PHP Fatal error:  Uncaught Amp\\Parallel\\Worker\\TaskError: Uncaught Error in worker with message "Call to undefined function downloadFile()" and code "0" in /var/www/html/test/vendor/amphp/parallel/lib/Worker/Internal/TaskFailure.php:45\nStack trace:\n#0 /var/www/html/test/vendor/amphp/parallel/lib/Worker/TaskWorker.php(126): Amp\\Parallel\\Worker\\Internal\\TaskFailure->promise()\n#1 [internal function]: Amp\\Parallel\\Worker\\TaskWorker->Amp\\Parallel\\Worker\\{closure}()\n#2 /var/www/html/test/vendor/amphp/amp/lib/Coroutine.php(76): Generator->send(Object(Amp\\Parallel\\Worker\\Internal\\TaskFailure))\n#3 /var/www/html/test/vendor/amphp/amp/lib/Internal/Placeholder.php(130): Amp\\Coroutine->Amp\\{closure}(NULL, Object(Amp\\Parallel\\Worker\\Internal\\TaskFailure))\n#4 /var/www/html/test/vendor/amphp/amp/lib/Coroutine.php(81): Amp\\Coroutine->resolve(Object(Amp\\Parallel\\Worker\\Internal\\TaskFailure))\n#5 /var/www/html/test/vendor/amphp/amp/lib/Internal/Placeholder.php(130): Amp\\Coroutine->Amp\\{closure}(NULL, Object(Amp\\Parallel\\Worker\\Internal\\TaskFailur in /var/www/html/test/vendor/amphp/parallel/lib/Worker/Internal/TaskFailure.php on line 45

I have similar requirement as asked in How does amphp work regarding the background running process.

Please add more detail about what you have tried an why it did not work and what you were expecting, please see https://stackoverflow.com/help/mcve — CoderJoe, May 14 '19 at 13:45

score 3 · Answer 1 · answered Jun 14 '19 at 06:12

Generally, Amp doesn't work magically in the background. If you use PHP via PHP-FPM or alike Amp will be shut down once the response is done, just like anything else.

If you want to move work from these requests into background processes, you need some kind of queue (e.g. beanstalkd) and a (permanent) worker to process these queued jobs. You can write such a daemonized worker with Amp, but it will have to be started out-of-band.

That said, if you just want concurrent downloads amphp/artax is better suited than using amphp/parallel, as it has a way lower overhead compared to a separate PHP process per HTTP request.

I understand what you are saying. Can you please guide me like how can I create a Worker with amphp/parallel and simultaneously it can be enqueued with [beanstalkd](https://github.com/amphp/beanstalk)? How to create a queue with it? — Neeraj Krishna Maurya, Oct 17 '19 at 11:42

score 0 · Answer 2 · answered Sep 09 '22 at 05:08

The question doesn't clarify where the downloadFile() function has been defined. As per amphp/parallel documentation, the callback must be auto-loaded so that Amphp can find it when the task is executed.

Here's a suggestion:

Put the downloadFile() function in a separate file, say functions.inc.
In your composer.json, under autoload/files, add an entry for functions.inc.
```
{
  "autoload": {
    "files": ["functions.inc"]
  }
}
```
Run composer install so that the autoload.php is regenerated to reflect the above change.
Try executing the file containing your first code snippet containing Loop::run() etc.

I think this should do the trick. Apart from this, please refer to kelunik's comment which contains valuable information.

How to use amphp/parallel library for non-blocking process

2 Answers2