amphp/amp

AMPHP is a collection of event-driven libraries for PHP designed with fibers and concurrency in mind. amphp/amp specifically provides futures and cancellations as fundamental primitives for asynchronous programming. We're now using Revolt instead of shipping an event loop implementation with amphp/amp.

Amp makes heavy use of fibers shipped with PHP 8.1 to write asynchronous code just like synchronous, blocking code. In contrast to earlier versions, there's no need for generator based coroutines or callbacks. Similar to threads, each fiber has its own call stack, but fibers are scheduled cooperatively by the event loop. Use Amp\async() to run things concurrently.

Motivation

Traditionally, PHP follows a sequential execution model. The PHP engine executes one line after the other in sequential order. Often, however, programs consist of multiple independent sub-programs which can be executed concurrently.

If you query a database, you send the query and wait for the response from the database server in a blocking manner. Once you have the response, you can start doing the next thing. Instead of sitting there and doing nothing while waiting, we could already send the next database query, or do an HTTP call to an API. Let's make use of the time we usually spend on waiting for I/O!

Revolt allows such concurrent I/O operations. We keep the cognitive load low by avoiding callbacks. Our APIs can be used like any other library, except that things also work concurrently, because we use non-blocking I/O under the hood. Run things concurrently using Amp\async() and await the result using Future::await() where and when you need it!

There have been various techniques for implementing concurrency in PHP over the years, e.g. callbacks and generators shipped in PHP 5. These approaches suffered from the "What color is your function" problem, which we solved by shipping Fibers with PHP 8.1. They allow for concurrency with multiple independent call stacks.

Fibers are cooperatively scheduled by the event-loop, which is why they're also called coroutines. It's important to understand that only one coroutine is running at any given time, all other coroutines are suspended in the meantime.

You can compare coroutines to a computer running multiple programs using a single CPU core. Each program gets a timeslot to execute. Coroutines, however, are not preemptive. They don't get their fixed timeslot. They have to voluntarily give up control to the event loop.

Any blocking I/O function blocks the entire process while waiting for I/O. You'll want to avoid them. If you haven't read the installation guide, have a look at the Hello World example that demonstrates the effect of blocking functions. The libraries provided by AMPHP avoid blocking for I/O.

Installation

This package can be installed as a Composer dependency.

composer require amphp/amp

If you use this library, it's very likely you want to schedule events using Revolt, which you should require separately, even if it's automatically installed as a dependency.

composer require revolt/event-loop

These packages provide the basic building blocks for asynchronous / concurrent applications in PHP. We offer a lot of packages building on top of these, e.g.

• amphp/byte-stream providing a stream abstraction
• amphp/socket providing a socket layer for UDP and TCP including TLS
• amphp/parallel providing parallel processing to utilize multiple CPU cores and offload blocking operations
• amphp/http-client providing an HTTP/1.1 and HTTP/2 client
• amphp/http-server providing an HTTP/1.1 and HTTP/2 application server
• amphp/mysql and amphp/postgres for non-blocking database access
• and many more packages

Requirements

This package requires PHP 8.1 or later. No extensions required!

Extensions are only needed if your app necessitates a high numbers of concurrent socket connections, usually this limit is configured up to 1024 file descriptors.

Usage

Coroutines

Coroutines are interruptible functions. In PHP, they can be implemented using fibers.

Note Previous versions of Amp used generators for a similar purpose, but fibers can be interrupted anywhere in the call stack making previous boilerplate like Amp\call() unnecessary.

At any given time, only one fiber is running. When a coroutine suspends, execution of the coroutine is temporarily interrupted, allowing other tasks to be run. Execution is resumed once a timer expires, stream operations are possible, or any awaited Future completes.

Low-level suspension and resumption of coroutines is handled by Revolt's Suspension API.

<?php

use Revolt\EventLoop;

require __DIR__ . '/vendor/autoload.php';

$suspension = EventLoop::getSuspension();

EventLoop::delay(5, function () use ($suspension): void {
    print '++ Executing callback created by EventLoop::delay()' . PHP_EOL;

    $suspension->resume(null);
});

print '++ Suspending to event loop...' . PHP_EOL;

$suspension->suspend();

print '++ Script end' . PHP_EOL;

Callbacks registered on the Revolt event-loop are automatically run as coroutines. It is safe to suspend within those callbacks. Apart from the event-loop API, Amp\async() can be used to start a coroutine (that is, a new fiber or an independent call stack).

<?php

require __DIR__ . '/vendor/autoload.php';

Amp\async(function () {
    print '++ Executing callback passed to async()' . PHP_EOL;

    Amp\delay(3);

    print '++ Finished callback passed to async()' . PHP_EOL;
});

print '++ Suspending to event loop...' . PHP_EOL;
Amp\delay(5);

print '++ Script end' . PHP_EOL;

Future

A Future is an object representing the eventual result of an asynchronous operation. Such placeholders are also called a "promise" in other frameworks or languages such as JavaScript. We chose to not use the "promise" name since a Future does not have a then method, which is typical of most promise implementations. Futures are primarily designed to be awaited in coroutines, though Future also has methods which act upon the result, returning another future.

A future may be in one of three states:

• Completed: The future has been completed successfully.
• Errored: The future failed with an exception.
• Pending: The future is still pending.

A successfully completed future is analog to a return value, while an errored future is analog to throwing an exception.

One way to approach asynchronous APIs is using callbacks that are passed when the operation is started and called once it completes:

doSomething(function ($error, $value) {
    if ($error) {
        /* ... */
    } else {
        /* ... */
    }
});

The callback approach has several drawbacks.

• Passing callbacks and doing further actions in them that depend on the result of the first action gets messy really quickly.
• An explicit callback is required as input parameter to the function, and the return value is simply unused. There's no way to use this API without involving a callback.

That's where futures come into play. They're placeholders for the result that are returned like any other return value. The caller has the choice of awaiting the result using Future::await() or registering one or several callbacks.

try {
    $value = doSomething()->await();
} catch (...) {
    /* ... */
}

public function await(): mixed

Suspends the current coroutine until the future is completed or errors. The future result is returned or an exception thrown if the future errored.

/** @param Closure(mixed $value): mixed $map */
public function map(Closure $map): Future

Attaches a callback which is invoked if the future completes successfully, passing the future result as an argument. Another future is returned, which either completes with the return value of the callback, or errors if the callback throws an exception.

/** @param Closure(Throwable $exception): mixed $catch */
public function catch(Closure $catch): Future

Attaches a callback which is invoked if the future errors, passing the exception as the callback parameter. Another future is returned, which either completes with the return value of the callback, or errors if the callback throws an exception.

/** @param Closure(): void $finally */
public function finally(Closure $finally): Future

Attaches a callback which is always invoked, whether the future completes or errors. Another future is returned, which either completes with same value as the future, or errors if the callback throws an exception.

Combinators

In concurrent applications, there will be multiple futures, where you might want to await them all or just the first one.

await

Amp\Future\await($iterable, $cancellation) awaits all Future objects of an iterable. If one of the Future instances errors, the operation will be aborted with that exception. Otherwise, the result is an array matching keys from the input iterable to their completion values.

The await() combinator is extremely powerful because it allows you to concurrently execute many asynchronous operations at the same time. Let's look at an example using amphp/http-client to retrieve multiple HTTP res