Multi Threading #

<?php
declare(strict_types=1);

// Make sure the pcntl extension is available
if (!function_exists('pcntl_fork')) {
    die("The pcntl extension is not available\n");
}

$urls = [
    'https://api.example.com/data/1',
    'https://api.example.com/data/2',
    'https://api.example.com/data/3',
    'https://api.example.com/data/4',
];

$pids = []; // store the PIDs of all child processes

foreach ($urls as $i => $url) {
    $pid = pcntl_fork();

    if ($pid === -1) {
        // Fork failed
        throw new \RuntimeException("Failed to create a process for URL $i");

    } elseif ($pid === 0) {
        // This is the CHILD process
        // Each child handles one URL
        $data = file_get_contents($url);
        file_put_contents("/tmp/result_$i.json", $data);
        echo "Child process $i finished: " . strlen($data) . " bytes\n";
        exit(0); // IMPORTANT: the child must exit, not continue to the next iteration!

    } else {
        // This is the PARENT process
        $pids[$pid] = $url;
        echo "Spawned child process PID $pid for $url\n";
    }
}

// The parent waits for all children to finish
foreach ($pids as $pid => $url) {
    $status = 0;
    pcntl_waitpid($pid, $status);

    if (pcntl_wifexited($status)) {
        $exitCode = pcntl_wexitstatus($status);
        echo "PID $pid finished with exit code $exitCode\n";
    }
}

echo "All child processes finished\n";

<?php
declare(strict_types=1);

function processInParallel(array $items, callable $handler, int $maxWorkers = 4): void
{
    $queue = $items;
    $active = []; // PIDs of currently running processes

    while (!empty($queue) || !empty($active)) {
        // Spawn new workers if there are slots and items
        while (count($active) < $maxWorkers && !empty($queue)) {
            $item = array_shift($queue);
            $pid  = pcntl_fork();

            if ($pid === 0) {
                // Child process
                try {
                    $handler($item);
                } catch (\Throwable $e) {
                    error_log("Worker error: " . $e->getMessage());
                    exit(1);
                }
                exit(0);
            } elseif ($pid > 0) {
                $active[$pid] = $item;
            }
        }

        // Wait for one child to finish before spawning the next
        if (!empty($active)) {
            $status = 0;
            $pid    = pcntl_wait($status); // wait for any child
            if ($pid > 0) {
                unset($active[$pid]);
            }
        }
    }
}

// Usage
$fileList = glob('/data/csv/*.csv');

processInParallel($fileList, function(string $file) {
    $rowCount = 0;
    $handle = fopen($file, 'r');
    while (($row = fgetcsv($handle)) !== false) {
        // process each row
        $rowCount++;
    }
    fclose($handle);
    echo basename($file) . ": $rowCount rows processed\n";
}, maxWorkers: 8);

<?php
// 1. Shared Memory — fastest
$shmKey  = ftok(__FILE__, 'a');
$shmId   = shmop_open($shmKey, 'c', 0644, 1024);

// Write to shared memory
shmop_write($shmId, json_encode(['status' => 'ok', 'count' => 42]), 0);

// Read from shared memory (possible from a different process)
$data = json_decode(shmop_read($shmId, 0, 1024), true);
shmop_close($shmId);

// 2. File — simplest, use locks
function safeWrite(string $path, string $data): void
{
    $fp = fopen($path, 'a');
    flock($fp, LOCK_EX);  // exclusive lock
    fwrite($fp, $data . "\n");
    flock($fp, LOCK_UN);  // release the lock
    fclose($fp);
}

// 3. Database or Redis — most flexible for production

<?php
use parallel\{Runtime, Future, Channel};

// Runtime is a thread worker
$runtime1 = new Runtime();
$runtime2 = new Runtime();

// Future represents a value that will arrive (async result)
$future1 = $runtime1->run(function(): int {
    // This code runs in a separate thread
    $total = 0;
    for ($i = 1; $i <= 1_000_000; $i++) {
        $total += $i;
    }
    return $total;
});

$future2 = $runtime2->run(function(): int {
    $total = 0;
    for ($i = 1_000_001; $i <= 2_000_000; $i++) {
        $total += $i;
    }
    return $total;
});

// Wait for both threads to finish and get their results
$total = $future1->value() + $future2->value();
echo "Total: $total\n"; // 2000001000000

// Channel — communication between threads
$channel = new Channel();

$producer = new Runtime();
$producer->run(function(Channel $ch): void {
    for ($i = 0; $i < 10; $i++) {
        $ch->send("message-$i");
        usleep(10000); // 10ms
    }
    $ch->close();
}, [$channel]);

// Consume messages from the channel
while (true) {
    try {
        $message = $channel->recv();
        echo "Received: $message\n";
    } catch (\parallel\Channel\Error\Closed $e) {
        break; // channel closed, done
    }
}

<?php
use parallel\{Runtime, Future};

function parallelMap(array $items, \Closure $fn, int $workers = 4): array
{
    $runtimes = [];
    $futures  = [];

    // Create a pool of runtimes
    for ($i = 0; $i < $workers; $i++) {
        $runtimes[] = new Runtime();
    }

    // Distribute work to workers round-robin
    foreach ($items as $i => $item) {
        $runtime      = $runtimes[$i % $workers];
        $futures[$i] = $runtime->run($fn, [$item]);
    }

    // Collect all results
    $results = [];
    foreach ($futures as $i => $future) {
        $results[$i] = $future->value();
    }

    return $results;
}

$urls = ['url1', 'url2', 'url3', 'url4', 'url5', 'url6', 'url7', 'url8'];

$downloadResults = parallelMap($urls, function(string $url): string {
    // Simulated download — runs in parallel across several threads
    sleep(1);
    return "Content from $url";
}, workers: 4);

// 8 URLs each taking 1 second, with 4 workers: finishes in ~2 seconds
// (compared to 8 seconds if sequential)

<?php
// Basic Fiber
$fiber = new Fiber(function(): string {
    echo "Fiber started\n";

    // suspend() — pause the fiber, return a value to the caller
    $valueFromResume = Fiber::suspend('first suspend result');
    echo "Resumed with: $valueFromResume\n";

    Fiber::suspend('second suspend result');
    echo "Fiber finished\n";

    return 'final return value';
});

// start() — begin the fiber, runs until the first suspend()
$value1 = $fiber->start();
echo "First suspend: $value1\n"; // "first suspend result"

// resume() — continue the fiber, runs until the next suspend()
$value2 = $fiber->resume('data from main');
echo "Second suspend: $value2\n";   // "second suspend result"

// Final resume() — the fiber runs until completion
$fiber->resume();
echo "Return: " . $fiber->getReturn() . "\n"; // "final return value"

<?php
class SimpleEventLoop
{
    private array $fibers = [];
    private array $queue  = [];

    public function add(Fiber $fiber): void
    {
        $this->fibers[] = $fiber;
    }

    public function schedule(callable $fn): void
    {
        $this->add(new Fiber($fn));
    }

    public function run(): void
    {
        // Start all fibers
        foreach ($this->fibers as $fiber) {
            if (!$fiber->isStarted()) {
                $fiber->start();
            }
        }

        // Keep running until all are done
        while (true) {
            $stillActive = false;
            foreach ($this->fibers as $fiber) {
                if ($fiber->isSuspended()) {
                    $fiber->resume();
                    $stillActive = true;
                } elseif (!$fiber->isTerminated()) {
                    $stillActive = true;
                }
            }
            if (!$stillActive) break;
        }
    }
}

$loop = new SimpleEventLoop();

$loop->schedule(function(): void {
    echo "Task 1: start\n";
    Fiber::suspend();
    echo "Task 1: continue\n";
    Fiber::suspend();
    echo "Task 1: done\n";
});

$loop->schedule(function(): void {
    echo "Task 2: start\n";
    Fiber::suspend();
    echo "Task 2: done\n";
});

$loop->run();
// Task 1: start
// Task 2: start
// Task 1: continue
// Task 2: done
// Task 1: done

composer require react/event-loop react/http react/promise

<?php
require 'vendor/autoload.php';

use React\EventLoop\Loop;
use React\Http\Browser;
use React\Promise\Promise;

// ReactPHP's Browser is a non-blocking HTTP client
$browser = new Browser();

// Fetch several URLs in parallel (non-blocking)
$promises = [
    $browser->get('https://api.example.com/users'),
    $browser->get('https://api.example.com/products'),
    $browser->get('https://api.example.com/orders'),
];

// React\Promise\all() — wait for all promises to finish
\React\Promise\all($promises)->then(
    function(array $responses): void {
        foreach ($responses as $i => $response) {
            $data = json_decode((string)$response->getBody(), true);
            echo "Response $i: " . count($data) . " items\n";
        }
    },
    function(\Throwable $e): void {
        echo "Error: " . $e->getMessage() . "\n";
    }
);

// The event loop runs until all async operations finish
Loop::run();

<?php
require 'vendor/autoload.php';

use React\Http\HttpServer;
use React\Http\Message\Response;
use React\Socket\SocketServer;
use Psr\Http\Message\ServerRequestInterface;

$server = new HttpServer(function(ServerRequestInterface $request): Response {
    $path = $request->getUri()->getPath();

    return match(true) {
        $path === '/'         => Response::plaintext("Hello from ReactPHP!\n"),
        $path === '/api/ping' => Response::json(['status' => 'ok', 'timestamp' => time()]),
        default               => new Response(404, [], "Not found\n"),
    };
});

$socket = new SocketServer('0.0.0.0:8080');
$server->listen($socket);

echo "Server running at http://localhost:8080\n";

\React\EventLoop\Loop::run();

<?php
// A simple job — a class describing the work
class SendEmailJob
{
    public function __construct(
        public readonly string $to,
        public readonly string $subject,
        public readonly string $body,
    ) {}

    public function handle(): void
    {
        // Email sending logic
        $mailer = new SmtpMailer();
        $mailer->send($this->to, $this->subject, $this->body);
        echo "Email sent to {$this->to}\n";
    }
}

// A simple queue using Redis
class SimpleQueue
{
    private \Redis $redis;

    public function __construct()
    {
        $this->redis = new \Redis();
        $this->redis->connect('127.0.0.1', 6379);
    }

    public function dispatch(object $job): void
    {
        $payload = serialize($job);
        $this->redis->rpush('queue:default', $payload);
        echo "Job " . get_class($job) . " queued\n";
    }

    public function process(): void
    {
        echo "Worker starting...\n";

        while (true) {
            // BLPOP — blocking pop, wait until there's an item
            $result = $this->redis->blpop(['queue:default'], 5); // 5 second timeout

            if ($result === null) {
                continue; // timeout, try again
            }

            $payload = $result[1];
            $job     = unserialize($payload);

            try {
                $job->handle();
                echo "Job " . get_class($job) . " succeeded\n";
            } catch (\Throwable $e) {
                error_log("Job failed: " . $e->getMessage());
                // Could push to a 'failed' queue for later retry
                $this->redis->rpush('queue:failed', $payload);
            }
        }
    }
}

// In a web controller/handler — dispatch the job
$queue = new SimpleQueue();
$queue->dispatch(new SendEmailJob(
    to:      '[email protected]',
    subject: 'Welcome!',
    body:    'Thank you for registering.',
));
// The request finishes instantly — the email is sent by a worker, not this request

// In the worker (separate CLI, run as a daemon)
// php worker.php
$queue = new SimpleQueue();
$queue->process();

; /etc/supervisor/conf.d/queue-worker.conf
[program:php-queue-worker]
process_name=%(program_name)s_%(process_num)02d
command=php /var/www/app/worker.php
directory=/var/www/app
autostart=true
autorestart=true
numprocs=4          ; 4 workers running in parallel
redirect_stderr=true
stdout_logfile=/var/log/queue-worker.log

# Manage workers with supervisord
supervisorctl reread
supervisorctl update
supervisorctl start php-queue-worker:*
supervisorctl status