PHP Asynchronous programming Spawning non-blocking processes with proc_open()
Example
PHP has no support for running code concurrently unless you install extensions such as pthread. This can be sometimes bypassed by using proc_open() and stream_set_blocking() and reading their output asynchronously.
If we split code into smaller chunks we can run it as multiple suprocesses. Then using stream_set_blocking() function we can make each subprocess also non-blocking. This means we can spawn multiple subprocesses and then check for their output in a loop (similarly to an even loop) and wait until all of them finish.
As an example we can have a small subprocess that just runs a loop and in each iteration sleeps randomly for 100 - 1000ms (note, the delay is always the same for one subprocess).
<?php
// subprocess.php
$name = $argv[1];
$delay = rand(1, 10) * 100;
printf("$name delay: ${delay}ms\n");
for ($i = 0; $i < 5; $i++) {
usleep($delay * 1000);
printf("$name: $i\n");
}
Then the main process will spawn subprocesses and read their output. We can split it into smaller blocks:
- Spawn subprocesses with proc_open() .
- Make each subprocess non-blocking with
stream_set_blocking(). - Run a loop until all subprocesses finish using
proc_get_status(). - Properly close file handles with the output pipe for each subprocess using
fclose()and close process handles withproc_close().
<?php
// non-blocking-proc_open.php
// File descriptors for each subprocess.
$descriptors = [
0 => ['pipe', 'r'], // stdin
1 => ['pipe', 'w'], // stdout
];
$pipes = [];
$processes = [];
foreach (range(1, 3) as $i) {
// Spawn a subprocess.
$proc = proc_open('php subprocess.php proc' . $i, $descriptors, $procPipes);
$processes[$i] = $proc;
// Make the subprocess non-blocking (only output pipe).
stream_set_blocking($procPipes[1], 0);
$pipes[$i] = $procPipes;
}
// Run in a loop until all subprocesses finish.
while (array_filter($processes, function($proc) { return proc_get_status($proc)['running']; })) {
foreach (range(1, 3) as $i) {
usleep(10 * 1000); // 100ms
// Read all available output (unread output is buffered).
$str = fread($pipes[$i][1], 1024);
if ($str) {
printf($str);
}
}
}
// Close all pipes and processes.
foreach (range(1, 3) as $i) {
fclose($pipes[$i][1]);
proc_close($processes[$i]);
}
The output then contains mixture from all three subprocesses as they we're read by fread() (note, that in this case proc1 ended much earlier than the other two):
$ php non-blocking-proc_open.php
proc1 delay: 200ms
proc2 delay: 1000ms
proc3 delay: 800ms
proc1: 0
proc1: 1
proc1: 2
proc1: 3
proc3: 0
proc1: 4
proc2: 0
proc3: 1
proc2: 1
proc3: 2
proc2: 2
proc3: 3
proc2: 3
proc3: 4
proc2: 4
