Matrix

Matrix is a PHP library that brings event-driven, asynchronous programming to PHP, inspired by JavaScript's async/await syntax. Built on top of ReactPHP's event loop, Matrix makes it easier to write non-blocking I/O operations and manage concurrency with a simple, intuitive API.

Understanding Async in PHP

Important: PHP runs in a single-threaded environment. Matrix doesn't create true parallelism but enables event-driven, non-blocking I/O operations through ReactPHP's event loop. This means:

• ✅ Non-blocking I/O: Network requests, file operations, and timers don't block execution
• ✅ Concurrent operations: Multiple I/O operations can run simultaneously
• ❌ CPU-bound tasks: Heavy computations will still block the event loop
• ❌ True parallelism: No multiple threads or processes

Matrix shines when dealing with I/O-heavy applications like API clients, web scrapers, or microservices.

Why Choose Matrix?

Matrix simplifies ReactPHP development by providing a familiar async/await syntax while maintaining full compatibility with ReactPHP's ecosystem. It handles the complexity of promise management and event loop integration behind a clean, intuitive API.

Key Features

• JavaScript-like API: Use async() and await() for straightforward asynchronous programming
• Powered by ReactPHP: Built on ReactPHP's battle-tested event loop for true non-blocking I/O
• Robust Error Handling: Catch and handle exceptions with .catch() or try-catch
• Automatic Loop Management: The event loop runs automatically to handle promise resolution
• Concurrent Operations: Run multiple I/O operations simultaneously
• Rate Limiting: Control the frequency of asynchronous operations
• Promise Cancellation: Cancel pending operations when they're no longer needed
• Retry Mechanism: Automatically retry failed operations with configurable backoff strategies
• Batch Processing: Process items in batches for improved performance
• Enhanced Error Handling: Add context to errors for better debugging

Installation

Install Matrix via Composer:

composer require jerome/matrix

Requirements

• PHP 8.0 or higher
• sockets extension enabled

ReactPHP promises and the event loop will be installed automatically via Composer.

API Overview

Core Functions

async(callable $callable): PromiseInterface

Wraps a callable into an asynchronous function that returns a promise.

$func = async(fn () => 'Success');

$func->then(fn ($value) => echo $value) // Outputs: Success
    ->catch(fn ($e) => echo 'Error: ' . $e->getMessage());

await(PromiseInterface $promise, ?float $timeout = null): mixed

Awaits the resolution of a promise and returns its value. Optionally accepts a timeout in seconds.

try {
    $result = await(async(fn () => 'Success'));
    echo $result; // Outputs: Success

    // With timeout
    $result = await(async(fn () => sleep(2) && 'Delayed Success'), 3.0);
    echo $result; // Outputs: Delayed Success (or throws TimeoutException if it takes too long)
} catch (\Throwable $e) {
    echo 'Error: ' . $e->getMessage();
}

Promise Combination

all(array $promises): PromiseInterface

Runs multiple promises concurrently and returns a promise that resolves with an array of all results.

$promises = [
    async(fn () => 'Result 1'),
    async(fn () => 'Result 2'),
    async(fn () => 'Result 3'),
];

$results = await(all($promises));
// $results = ['Result 1', 'Result 2', 'Result 3']

race(array $promises): PromiseInterface

Returns a promise that resolves with the value of the first resolved promise in the array.

$promises = [
    async(function () { sleep(2); return 'Slow'; }),
    async(function () { sleep(1); return 'Medium'; }),
    async(function () { return 'Fast'; }),
];

$result = await(race($promises));
// $result = 'Fast'

any(array $promises): PromiseInterface

Returns a promise that resolves when any promise resolves, or rejects when all promises reject.

$promises = [
    async(function () { throw new \Exception('Error 1'); }),
    async(function () { return 'Success'; }),
    async(function () { throw new \Exception('Error 2'); }),
];

$result = await(any($promises));
// $result = 'Success'

Concurrency Control

map(array $items, callable $callback, int $concurrency = 0, ?callable $onProgress = null): PromiseInterface

Maps an array of items through an async function with optional concurrency control and progress tracking.

use React\Http\Browser;

$urls = ['https://example.com', 'https://example.org', 'https://example.net'];
$browser = new Browser();

$results = await(map(
    $urls,
    function ($url) use ($browser) {
        // Non-blocking HTTP request
        return $browser->get($url)->then(function ($response) use ($url) {
            return [
                'url' => $url,
                'status' => $response->getStatusCode(),
                'size' => strlen($response->getBody())
            ];
        });
    },
    2, // Process 2 URLs at a time
    function ($done, $total) {
        echo "Processed $done of $total URLs\n";
    }
));

print_r($results); // Array of response data

batch(array $items, callable $batchCallback, int $batchSize = 10, int $concurrency = 1): PromiseInterface

Processes items in batches rather than one at a time for improved performance.

$items = range(1, 100); // 100 items to process

$results = await(batch(
    $items,
    function ($batch) {
        return async(function () use ($batch) {
            // Process the entire batch at once
            return array_map(fn ($item) => $item * 2, $batch);
        });
    },
    25, // 25 items per batch
    2   // Process 2 batches concurrently
));

print_r($results); // Array of processed items

pool(array $callables, int $concurrency = 5, ?callable $onProgress = null): PromiseInterface

Executes an array of callables with limited concurrency.

$tasks = [
    fn () => async(fn () => performTask(1)),
    fn () => async(fn () => performTask(2)),
    fn () => async(fn () => performTask(3)),
    // ...more tasks
];

$results = await(pool(
    $tasks,
    3, // Run 3 tasks concurrently
    function ($done, $total) {
        echo "Completed $done of $total tasks\n";
    }
));

print_r($results); // Array of task results

Error Handling and Control

timeout(PromiseInterface $promise, float $seconds, string $message = 'Operation timed out'): PromiseInterface

Creates a promise that times out after a specified period.

try {
    $result = await(timeout(
        async(function () {
            sleep(5); // Long operation
            return 'Done';
        }),
        2.0, // 2 second timeout
        'The operation took too long'
    ));
} catch (\Matrix\Exceptions\TimeoutException $e) {
    echo $e->getMessage(); // "The operation took too long"
    echo "Duration: " . $e->getDuration() . " seconds"; // "Duration: 2 seconds"
}

retry(callable $factory, int $maxAttempts = 3, ?callable $backoffStrategy = null): PromiseInterface

Retries a promise-returning function multiple times until success or max attempts reached.

use React\Http\Browser;

$browser = new Browser();

try {
    $result = await(retry(
        function () use ($browser) {
            // Non-blocking HTTP request with potential for failure
            return $browser->get('https://unreliable-api.com/data')
                ->then(function ($response) {
                    if ($response->getStatusCode() !== 200) {
                        throw new \RuntimeException('API returned ' . $response->getStatusCode());
                    }
                    return $response->getBody()->getContents();
                });
        },
        5, // Try up to 5 times
        function ($attempt, $error) {
            // Exponential backoff with jitter
            $delay = min(pow(2, $attempt - 1) * 0.1, 5.0) * (0.8 + 0.4 * mt_rand() / mt_getrandmax());
            echo "Attempt $attempt failed: {$error->getMessage()}, retrying in {$delay}s...\n";
            return $delay; // Return null to stop retrying
        }
    ));

    echo "Finally succeeded: $result\n";
} catch (\Matrix\Exceptions\RetryException $e) {
    echo "All {$e->getAttempts()} attempts failed\n";

    foreach ($e->getFailures() as $index => $failure) {
        echo "Failure " . ($index + 1) . ": " . $failure->getMessage() . "\n";
    }
}

cancellable(PromiseInterface $promise, callable $onCancel): CancellablePromise

Creates a c