Multiple function calls asynchronously in PHP

Question

I'm using Symfony 3 and ReactPHP library for control all my functionality and I need to execute multiple calls to same function (subFunction() in code) asynchronously. I have 2 projects (project1 and project2):

Project 1 code:

/**
* Loop an array of urls and call sub function.
**/
public function startFunction() {
  $finalResponse = [];
  $urls = ['www.google.es', 'www.github.com', 'www.bitbucket.org'];

  foreach ($urls as $url) {
    $res = $this->subFunction($url);    // subfunction call ( **IT MAY TAKE A LONG TIME !!** )
    $finalResponse[] = $res;
 }

 return $finalResponse;
}

/**
* Uses Factory loop to get the Promise returned by finalRequest function.
**/
private function subFunction($url) {
  $loop = \React\EventLoop\Factory::create();
  $classA = new Project2\ClassA();
  $finalResponse = null;

  // project 2 function call
  $classA->finalRequest($url)->then(function($response) use(   
     &$finalResponse
  ) {
     $finalResponse = $response;
  })

  return $finalResponse;
}

Project 2 code:

classA {

/**
* Makes an React\HttpClient request (GET) to sent url and return his value inside a Promise.
**/
public function finalRequest($url) {
   $generalDeferred = new Deferred();
   $generalPromise = $generalDeferred->promise();

   // make React\HttpClient request
   $request = $client->request('GET', $url);
   $request->on('response', function ($response) use($generalDeferred) {
   $response->on('data', function ($response) {
        $generalDeferred->resolve($response);
      });
   });
  $request->end();

  return $generalPromise;
}

}

Problem is that on every subFunction($url) call, the program stops until the sub Function gets the response, but I need to do this asynchronously because this subFunction could take many seconds. So I would like to launch all subFunction($url) calls at the same time, and get all responses asynchronously.

It's possible solve this problem? Thanks.

Answer 1

There are multiple options if your subFunction is a blocking function call.

Rewrite subFunction to use non-blocking I/O

This might work or not work depending on what the function is doing. If the function is mainly waiting for I/O such as waiting for HTTP responses, then this will probably be the easier way, but it entirely depends on the complexity of the code that needs to be rewritten.

ReactPHP provides an HTTP client that uses non-blocking I/O to do HTTP requests and receive responses that you're already using.

Use multi-processing

If your subFunction is CPU bound or there doesn't exist a non-blocking driver for the protocol you're trying to use, you can use multi-processing to execute the function in another process without blocking the main event loop.

There's a child process library for ReactPHP, but I'm currently not aware of a library for ReactPHP that makes parallel processing as easy as with amphp/parallel. While that library is based on Amp, there exists a compatibility adapter to use any ReactPHP library on top of Amp, so you can continue to use your current libraries.

With amphp/parallel-functions, which is build on top of amphp/parallel, you can simply execute your function in another process with a few lines of code. The only limitation is that your passed data (arguments and return values) must be serializable.

<?php

use Amp\Promise;
use function Amp\ParallelFunctions\parallel;

$callable = parallel('subFunction');

$promise = $callable($arg1, $arg2, $arg3);

// either yield $promise in a coroutine,
// or adapt it to a ReactPHP promise,
// or use Promise\wait for a blocking wait.

var_dump(Promise\wait($promise));

Answer 2

First of all you can only have one loop running in an app. Second you have to make the loop run: https://reactphp.org/event-loop/

You should create the app and then register all the services and events, start the loop and leave it running as a server.

$loop = React\EventLoop\Factory::create();

$server = stream_socket_server('tcp://127.0.0.1:8080');
stream_set_blocking($server, 0);

$loop->addReadStream($server, function ($server) use ($loop) {
[...]
});

$loop->addPeriodicTimer(5, function () {
[...]
});

$loop->run(); <---- you will not execute anything behind this point.

Why? https://github.com/reactphp/event-loop/blob/master/src/ExtLibeventLoop.php#L196

public function run()
{
    $this->running = true;
    while ($this->running) { <------------------------------
        $this->futureTickQueue->tick();
        $flags = EVLOOP_ONCE;
        if (!$this->running || !$this->futureTickQueue->isEmpty()) {
            $flags |= EVLOOP_NONBLOCK;
        } elseif (!$this->streamEvents && !$this->timerEvents->count()) {
            break;
        }
        event_base_loop($this->eventBase, $flags);
    }
}

For the use you do of the loop I would recommend to use Guzzle Async: http://docs.guzzlephp.org/en/stable/faq.html

$finalResponse = [];
$promises = [];
$urls = ['www.google.es', 'www.github.com', 'www.bitbucket.org'];
foreach ($urls as $index => $url) {
    $promise = $client->requestAsync('GET', $url);
    $promise->then(function ($response) use ($index, &$finalResponse) {
        $finalResponse[$index]=$response;
    });
    $promises[$index]=$promise;
}
foreach ($promises as $promise) {
    $promise->wait();
}
return $finalResponse;