Download the PHP package shawty81/multiprocessor without Composer

MultiProcessor

Process a large dataset in parallel by handing chunks of it to forked child processes.

The parent keeps a pool of children busy, hands each one a chunk of your data, and reaps them as they finish. Because every chunk is processed in a process of its own, a fatal error, a segfault or an out of memory kill in one chunk cannot take the run down: the parent notices the child died, retries the chunk, and carries on with the rest.

This is a command line tool. It forks with pcntl_fork() and signals with posix_kill(), so it needs ext-pcntl and ext-posix and must not be run under a web server, PHP-FPM or any other environment where the process is not yours to fork.

Requirements

• PHP 8.4 or newer
• ext-pcntl and ext-posix
• A CLI SAPI

Installation

Quick start

Two interfaces make up everything you write: an iterator that hands out chunks of work, and a child processor that does the work for one chunk. ArrayIterator ships with the library, so this runs as it stands.

Run it and you get four children resizing five images each, so the twenty quarter-second jobs take about a second and a quarter instead of five seconds:

The usleep() is there because forking is only worth it when a record costs real time. Forking is not free puts a number on that.

ArrayIterator accepts any array. A plain list, an associative array, rows keyed by their database id, the output of array_filter(), an array with gaps left by unset(), one holding null values: all of them hand out every record they contain.

What it does not carry through is the keys. A chunk holds values, in the array's own order, so rows keyed by their database id arrive without those ids. When a key means something, put it inside the value as well and the child processor will have it.

The two interfaces you implement

IteratorInterface

Lives entirely in the parent, except for dropConnections(). It is the single source of work: the parent asks it for a chunk, forks a child for it, and asks for the next one.

dropConnections() exists because a fork copies the process, file descriptors and all. A database handle that was open when the fork happened is now referred to by two processes that both think they own it, and they will interleave on the same socket: results arrive at the wrong process, and the connection is torn down for both when either one closes it. The fix is for the child to throw the inherited handle away and, if it needs one, connect again for itself. That is what this method is the place for.

ChildProcessorInterface

process() should return when it is done and throw when it cannot finish. Do not call exit() yourself: the library gives the child its exit status, and it uses that status to tell the parent whether the chunk succeeded.

Chunk

A final readonly value object carrying one unit of work.

Settings

Settings is a final readonly value object. Build it with named arguments; only the first two have no default.

A value outside those bounds throws InvalidSettingsException from the constructor, so a misconfigured run fails before anything is forked.

Forking is not free

Every chunk costs one pcntl_fork() and one reap, and that cost is roughly fixed however much or little the chunk goes on to do. On this machine it is about 390 microseconds per chunk, measured by running 20000 records of trivial work (one multiplication each) through the library at maxChildren: 8 and varying only chunkSize:

The same 20000 records in a plain foreach take 0.0001 s in total, about 0.007 us each.

Measured on 16 cores under the php:8.5-cli image, so treat the numbers as indicative of the shape rather than as a guarantee: the constant differs per machine, kernel and container runtime, but it is a constant either way.

What it implies:

• A chunk has to do substantially more than half a millisecond of work before forking it pays for itself. Work that is faster than that runs quicker in a plain foreach, and this library is pure overhead.
• The point of a chunk is to amortise that fixed cost over many records, so the default chunkSize of 10 is low for most real workloads. Raise it until a chunk takes long enough that the fork disappears into it, while staying short enough that a chunk lost to a failure is not much work to redo.

What happens when a child fails

A child fails in one of two ways, and the parent treats them the same.

• It threw. The exception never reaches the parent, because it happened in another process. The child catches it, logs its class, message and stack trace, and exits with status 255. The parent logs Child (pid: ...) exited with an error.
• It was killed. A segfault, an out of memory kill or a kill -9 leaves the child no chance to report anything. The parent sees the signal in the wait status and logs Child (pid: ...) was killed by signal 9.

Either way the chunk that child was holding is then handled like this:

1. If exitOnFatal is true, the run stops immediately: every remaining child is killed with SIGKILL and the parent exits. finish() is not called and no summary is logged.
2. Otherwise, if retryOnFatal is false, the chunk is dropped and the run continues.
3. Otherwise, if the chunk has already been retried maxRetries times, the parent logs Chunk from Child (pid: ...) failed N times, giving up on it., drops it, and continues.
4. Otherwise the chunk goes back on the queue with its retries count raised by one, and is handed to the next free child.

Retried chunks are taken from the queue before new ones are read from the iterator, so a failure is retried promptly rather than at the end of the run.

Note the order: exitOnFatal is checked first, so setting it to true disables retrying whatever retryOnFatal says.

A run that a child cannot be forked for at all is not recoverable, and the parent throws ForkFailedException.

Signals

The parent installs handlers for SIGINT, SIGTERM and SIGCHLD.

SIGINT (Ctrl-C) or SIGTERM on the parent starts an orderly shutdown: it logs what it is doing, sends SIGKILL to every child still in the pool, and exits. It does not wait for children to finish their chunk, and it does not call finish() or log a summary. The handlers are installed so that they interrupt the blocking wait the parent spends nearly all of its time in, so the shutdown happens at once rather than when a child next exits.

Children inherit those handlers across the fork but have no business shutting the run down on the parent's behalf. So the first thing a child does with an inherited signal is put the default handler back and raise the signal on itself again, which leaves it dying exactly as it would have without the fork. A child of your own that you fork inside process() is unaffected by any of this.

Logging

Anything implementing PSR-3's LoggerInterface will do. Pass it as the logger setting, or set it afterwards, since MultiProcessor implements LoggerAwareInterface:

That logger is the MultiProcessor's own. Your iterator and child processor log through whatever you give them; the quick start passes the same instance to both.

CommandLineLogger is bundled for when you just want to see the run on your terminal. It writes one line per message, prefixed with the time and the first letter of the level, and fills in the PSR-3 {placeholders} from the context:

Exceptions

Every exception this library throws implements MultiProcessor\Exception\ExceptionInterface, so one catch covers all of it:

Exceptions your own code throws are not caught by the library in the parent. In a child they are caught, logged and turned into a failed chunk, as described above.

License

MIT. See LICENSE.