PHP Multiprocessing Library

This library provides a robust abstraction layer for handling multiprocessing in PHP. It simplifies the creation of independent processes, child process management (supervision), worker pools for parallel task execution, and inter-process communication (IPC) using queues.

📋 Requirements and PHP Extensions

To use this library, your PHP environment must have the following extensions installed and enabled:

• pcntl (Process Control) - For fork creation and signal management.
• posix (POSIX Functions) - For process ID management and POSIX signals.
• sysvmsg (System V Message Queues) - For IPC message queues.

🚀 Key Features

1. Process

The Process class is the foundation for executing code in isolated processes. It handles the complexity of pcntl_fork, signal management, and capturing results and exceptions.

Basic Example:

use MultiprocessingProcess;

// Creates a process that executes an anonymous function with arguments
$process = new Process('MyProcess', function ($name) {
    sleep(2); // Simulate heavy work
    return "Hello, $name! Process finished.";
}, 'World');

// Starts the process (fork)
$process->start();

echo "Process started with PID: " . $process->getPid() . "\n";

// Waits for the process to finish and gets the result
// The result() method blocks execution until the child process returns or throws an exception
try {
    $result = $process->result();
    echo $result; // Output: Hello, World! Process finished.
} catch (\Throwable $e) {
    echo "Error in process: " . $e->getMessage();
}

2. MultiprocessPool

The MultiprocessPool allows parallel task processing by automatically distributing them among a pre-allocated set of workers. It is ideal for batch processing or heavy tasks that should not block the main flow.

Worker Pool Example:

use MultiprocessingPool\MultiprocessPool;

// Creates a pool with 4 workers
$pool = new MultiprocessPool(4);
$pool->start();

echo "Submitting tasks to the pool...\n";

// Submits tasks for execution
$futures = [];
for ($i = 1; $i <= 5; $i++) {
    // submit() returns a Future object immediately
    $futures[] = $pool->submit(function ($num) {
        $wait = rand(1, 3);
        sleep($wait);
        return "Task $num took $wait seconds";
    }, $i);
}

// Processes results as they become ready (asynchronously)
foreach ($pool->asCompleted() as $id => $future) {
    echo "Completed: " . $future->getResult() . "\n";
}

// Shuts down the pool and cleans up resources
$pool->shutdown();

3. ParentProcess (Supervisor)

The ParentProcess class acts as a supervisor managing multiple child processes. It monitors registered processes and restarts them if they terminate, keeping services always active.

Supervisor Example:

First, define a worker class extending Process:

use MultiprocessingProcess;

class MyDaemonWorker extends Process
{
    protected function run()
    {
        // Worker logic
        // ... perform tasks ...
        sleep(5);
        return "Cycle finished";
    }
}

Configure and start the supervisor:

use MultiprocessingParentProcess;
use MultiprocessingPidfile;

$pidfile = new Pidfile('/tmp/supervisor.pid');
$supervisor = new ParentProcess('SupervisorApp', $pidfile);

// Registers child processes
$supervisor->registerChild(MyDaemonWorker::class, []);

// Starts the supervisor
$supervisor->start();

// The main script can wait for the supervisor (which runs in background/fork)
$supervisor->join();

4. Queue (IPC)

The Queue class facilitates the use of System V message queues to safely exchange complex data between processes.

Usage Example:

use MultiprocessingQueue\Queue;

$queueId = 12345;
$queue = new Queue($queueId);

// Publishes data to the queue (can be any serializable data)
$queue->publish(['job' => 'send_email', 'email' => 'test@example.com']);

// In another process...
$message = $queue->retrieve();
if ($message) {
    // Process the message
    print_r($message);
}

// Close the queue when no longer needed
$queue->close();