Rebuilding ReactPHP from scratch
Table of Contents
In this article we’ll write a relatively simple, and certainly naive, event loop in pure PHP. While I hope to never encounter this exact solution in production, I find “rediscovering” existing solutions helps to get a better understanding of the underlying fundamentals. The article focuses on the concepts rather than clean code. Things like proper error handling are often ignored to make the concepts and code samples themselves easier to understand. You have been warned!
This post is heavily inspired by David Beazley’s legendary talk “Python Concurrency from the Ground Up” at PyCon 2015, but has been adapted to use PHP concepts. I have presented this concept previously at my company as an internal talk (a “spreekbeurt”, as we call it), which you can find on my Github profile.
Something to do #
In this example we will build the next billion idea, FaaS; fibonacci as a service. The core of which will be this function.
function fibonacci($n) {
if ($n <= 2) {
return 1;
}
return fibonacci($n - 1) + fibonacci($n - 2);
}
This fibonacci implementation intentionally takes exponential time. This allows us to simulate work on the CPU, people tend to have an intuition of “something” happening.
TCP server #
To illustrate the concepts, we will be building a simple TCP server using raw sockets. Don’t worry if you haven’t worked with sockets before. This example is just a request / response cycle, but without the overhead of a transport layer such as HTTP. Anyone familiar with web development in PHP should be able to follow along.
I encourage everyone to make a simple program using raw sockets in your language of choice. If you have never worked with sockets before you’ll see sockets have a tendency to challenge your assumptions and force you to think of failure states. Doing this is bound to spark a newfound appreciation of cURL.
function fib_server() {
$sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
socket_bind($sock, '127.0.0.1', 12021);
socket_listen($sock, 5);
try {
while (true) {
$client = socket_accept($sock);
fib_handler($client);
}
} finally {
socket_close($sock);
}
}
A couple things to note in this example, we open a socket server on port 12021. The integer 5 in socket_listen($sock, 5) tells PHP to use a backlog of 5 connections. Any attempted connection after the backlog is full will still stall or fail depending on the OS. Lastly, we wrap the whole server in a try / finally block.
As PHP developers we don’t usually consider resource cleanup as the resources are cleaned up automatically after the process is done. In practice this means the resources are freed up after the request is done and the response has been sent. However, when you start to develop long running processes you’ll quickly realize that the automatic cleanup never actually gets the chance to run. Which means our service will consume “unlimited” RAM, until the OS has had enough and sends the OOM killer to stop the process.
TCP handler #
Our fibonacci service does one thing and one thing only, this means we don’t have to implement any kind of routing. We define a single fib_handler function, which takes an accepted connection. The handler then reads a maximum of 1024 bytes from the socket to be considered our input, which is naively cast to an integer. We calculate the fibonacci number and write it back to the client. To stay with HTTP framework terms, $data is our request, and the call to socket_write is our response.
Error handling is left as an exercise for the reader to keep the relevant logic clear. What happens when the client sends more than 1024 bytes? What happens when the client sends a character that cannot be converted to an integer? What happens when the client sends nothing at all?
function fib_handler($client)
{
while (true) {
$data = socket_read($client, 1024);
if ($data === '') {
break;
}
$n = (int)($data);
$result = fibonacci($n);
socket_write($client, $result . "\n");
}
}
Benchmarking #
Every good piece of server needs a benchmark, we must be blazingly fast™ after all. The benchmarking script is as simple and naive as the server, again to keep the focus on the relevant bits. In this case we use a small snippet to connect to our fibonacci server, continuously write the same integer and calculate the time it took to get a response.
$sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
socket_connect($sock, '127.0.0.1', 12021);
while (true) {
$start = microtime(true);
socket_write($sock, "30");
$result = socket_read($sock, 1024);
$end = microtime(true);
echo ($end - $start) . "\n";
}
Initial results #
The result of this benchmark will vary depending on the machine the server is running and to some extent the machine running the benchmark. The important thing to note here is the time it takes the server to process each request. There are some small fluctuations, but the time per request is roughly constant.
0.036442995071411
0.036579847335815
0.03670597076416
...
0.036795854568481
0.036497831344604
0.036513090133667
Waiting for the network #
If we were to run a second benchmark while the first script is still running, we’d see that the second benchmark hangs until the first process is stopped. The server in its current state can only handle a single connection at a time, regardless of what it’s actually doing. In practice this means the server is spending a lot of time waiting for the network. If you take a look back at the fib_handler you’ll quickly see two crucial points where we are waiting for the network, notably socket_accept and socket_read. We call this blocking. We block the execution of the program until we have what we need to continue. You might assume socket_write is not blocking, in practice this is not always true and depends on buffers and availability of the underlying network.
In this post we’re using a form of cooperative multitasking. A common alternative is to use threads which will give each thread “roughly” the same amount of time to do what it needs to do. While every thread gets the same amount of time, the scheduler has no idea what each thread is actually doing, meaning it could be waiting for the network while another thread actually could have used the time to do something productive.
Ideally, we would put the handler in an idle state while it’s waiting for the network and tell the server to move on to more pressing matters such as accepting new clients. This concept is called cooperative multitasking. It just so happens PHP has a way for code to communicate with its caller since PHP 5.5 in the form of generators.
Note: while generators were introduced in PHP 5.5, later in this post we’ll use features from PHP 7 and PHP 8.1
A brief generator refresh #
Generators are a useful tool to have available. In their simplest form they are an iterator with a construct phase that can be looped over, almost as if it’s an array or a Collection. This allows the programmer to do some setup before the loop actually starts as well as cleanup after the loop is done. The yield statement is used to suspend the loop to send a value back to the caller, in this case the variable $n.
function countdown($n)
{
while ($n > 0) {
yield $n;
$n--;
}
}
$generator = countdown(3);
foreach ($generator as $value) {
echo $value . "\n";
}
Output:
3
2
1
Manual control #
A lesser known feature of generators is that they also have a “manual control” option. Much like driving a manual car, this means the operator is responsible for advancing the generator when applicable. In the example below we call current and next 3 times, with the same output as before.
$generator = countdown(3);
echo $generator->current() . "\n";
$generator->next();
echo $generator->current() . "\n";
$generator->next();
echo $generator->current() . "\n";
$generator->next();
Output:
3
2
1
In this example we have no way of knowing when the generator has been exhausted, when you use a foreach loop, this is handled for you by PHP itself. Luckily, a generator object has a valid method, which returns true when the generator still has a new value to yield and false when the generator has been exhausted.
While you might have intuitively guessed such a manual option exists, people often overlook the option of sending data and even exceptions into the generator from the caller. These three methods allow the caller full control over the flow of data and crucially when the data is allowed to flow.
function receiver()
{
while (true) {
$value = yield;
echo "Received: $value\n";
}
}
$recv = receiver();
$recv->send('Hello');
$recv->send('Hello');
function receiver()
{
while (true) {
try {
$value = yield;
echo "Received: $value\n";
} catch (Throwable $e) {
echo "Caught: " . $e->getMessage() . "\n";
}
}
}
$recv = receiver();
$recv->send('Hello');
$recv->throw(new Exception('¯\_(ツ)_/¯'));
Scheduling generators #
We can rephrase the yield to mean something like “hold up, I need to wait for a bit”. This is effectively what a Task in other languages represents, the yield then becomes a different way of writing await. We can build upon this concept by writing a scheduler that takes an array of “tasks”; our generator objects and advances the generator one by one until there’s none left. A scheduler in this context is just a piece of code that decides which paused generator should run next.
Here we use a tasks array as a way of keeping track of incomplete work, also known as valid generators. During each iteration of the scheduler, we pull the next task from the array and advance it. If the task is still valid at the end of the iteration, we push it back onto the tasks array. We call this a round-robin scheduler.
This type of scheduling is easy to understand and explain making it a prime candidate for this post. The simplicity comes at the cost of naivety, the scheduler has no concept of prioritization and is susceptible to spending more time context switching than doing actual work. A round-robin scheduler also relies on the tasks themselves being “fair”. If the task never gives control back to the scheduler, no other tasks will ever get a chance to run.
$tasks = [
countdown(10),
countdown(5),
countdown(20),
];
while ($tasks) {
$task = array_shift($tasks);
$result = $task->current();
echo $result, PHP_EOL;
$task->next();
if ($task->valid()) {
$tasks[] = $task;
} else {
echo "Task complete", PHP_EOL;
}
}
IO multiplexing #
Recall the main fibonacci service and specifically the blocking operations. If we could somehow indicate what we need to wait for and crucially who is waiting, we could update the scheduler to check if that condition has been met and skip the task.
As you might have expected by now, generators allow us to do just that. By yielding an array with the action and the socket we have both nuggets of information to implement this. By placing the yield right before the blocking call we signal to the scheduler “don’t resume this task until the socket is ready for the requested action”.
function fib_handler($client)
{
while (true) {
yield ['read', $client];
// blocking
$data = socket_read($client, 1024);
// Connection has been broken, either gracefully or due to an error
if ($data === false || $data === '') {
break;
}
$n = (int)($data);
$result = fibonacci($n);
yield ['write', $client];
// potentially blocking
socket_write($client, $result . "\n");
}
}
We do the same in the fibonacci server itself. Though it now also needs the ability to kick off new tasks. Since we practice the single responsibility principle, the server does not need to know how to run tasks. It just needs to know how to schedule new tasks. In this case the scheduling “API” is an array of tasks. The server can schedule a new task by pushing a generator object onto the array.
function fib_server(&$tasks) {
$sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
socket_bind($sock, '127.0.0.1', 12021);
socket_listen($sock, 5);
try {
while (true) {
yield ['read', $sock];
// blocking
$client = socket_accept($sock);
$tasks[] = fib_handler($client);
}
} finally {
socket_close($sock);
}
}
If you recall generators also have
sendandthrowmethods which allow you to send data into the generator. If we were to build on this concept further, we could use these methods to hand control over the sockets completely to the scheduler.
We now know what tasks we need to execute, what and who they are waiting for. As it turns out, this is everything we need to update the scheduler to accept multiple clients at the same time. To do this we split up the scheduler into two distinct sections; the “waiting for IO” section and the task running section. The latter we have seen before and is responsible for keeping track of tasks, advancing and rescheduling generators.
To know when the sockets are ready for the task to resume, we use the PHP function socket_select which takes arrays of sockets and returns new arrays of sockets that are ready. This might be all of the sockets, none of them or a subset.
socket_selectalso takes an “exceptional” array to indicate sockets that have some sort of problem and a timeout in seconds and micro seconds. By passingnullas the timeout, we are telling the OS to block until there’s activity on the sockets. In “real” async frameworkssocket_selectis only used when better alternatives such asepollorkqueuearen’t available.socket_selectremains available as a portable alternative.socket_selectreports readiness, meaning we do not have to callsocket_set_nonblockon our sockets, though in a real program you would do so defensively.
Putting everything together #
Here we create a run function to act as our scheduler. If you take a look at async frameworks, you’ll often find a very similar function to kick off async functionality. The run function contains 2 different loops, a loop to run the tasks as well as a loop to wait for IO. Note how we only wait for IO if we do not have any tasks that need to do work.
Within the IO loop, we group the sockets by operation and hand them over to socket_select. Once we have arrays of readable and writable sockets, we determine which task was waiting for which socket and advance the relevant task(s). Looking at the signature of socket_select we see the function has int|false as its return type. It either returns the amount of socket resources that have had something of interest happen to them or false when some kind of error occurred. In the real world sockets are unreliable; the client might not even be connected anymore by the time we’re handling their socket.
Note the loop conditions in the example below. The scheduler only exits when there are no runnable tasks left and nothing is waiting on IO. In this demo server this never happens, the server task keeps the listening socket in the readWait array forever. A scheduler can’t tell difference between “everyone is waiting” and “everyone is done”. Either scenario will use all the CPU available or hang on the last
socket_select.
function run(&$tasks)
{
$readWait = [];
$writeWait = [];
// Run while we have _anything_ to do
while ($tasks || $readWait || $writeWait) {
// No active tasks, wait for I/O — but only if something is actually waiting
while (!$tasks && ($readWait || $writeWait)) {
$readable = array_map(static fn($task) => $task[0], $readWait);
$writable = array_map(static fn($task) => $task[0], $writeWait);
$except = [];
// Poll the OS for socket activity
$result = socket_select($readable, $writable, $except, null);
if ($result === false) {
continue;
}
// Pull the tasks from the waiting area and add them back to the main queue
// Note, we must do this for both read and write
foreach ($readable as $socket) {
$id = spl_object_id($socket);
[, $task] = $readWait[$id];
unset($readWait[$id]);
$task->next();
if ($task->valid()) {
$tasks[] = $task;
}
}
foreach ($writable as $socket) {
$id = spl_object_id($socket);
[, $task] = $writeWait[$id];
unset($writeWait[$id]);
$task->next();
if ($task->valid()) {
$tasks[] = $task;
}
}
}
// Nothing runnable and nothing waiting: we're done
if (!$tasks) {
continue;
}
/** @var Generator $task */
$task = array_shift($tasks);
/**
* @var $why string
* @var $who Socket
*/
[$why, $who] = $task->current();
if ($who === null) {
continue;
}
$id = spl_object_id($who);
if ($why === 'read') {
$readWait[$id] = [$who, $task];
} elseif ($why === 'write') {
$writeWait[$id] = [$who, $task];
} else {
throw new \Exception("¯\_(ツ)_/¯");
}
}
}
$tasks = [];
// Must start the server as a task to be able to pause and resume it
$tasks[] = fib_server($tasks);
run($tasks);
Results #
Now we can start the benchmark as many times as we want and keep getting results in each terminal. Note however that for each additional benchmark process the time per calculation increases. This is an important caveat of the asynchronous programming, the socket operations are no longer blocking, but calculating fibonacci still takes CPU time which will still block.
0.03693413734436
0.037462949752808
0.03684401512146
...
0.073081016540527
0.072849988937378
0.072762012481689
Cleaning up #
Remembering to yield a magic array each time your task is going to do a blocking operation is not very ergonomic nor very DRY. Let’s clean that up a little and make our async code a little more recognizable for people unfamiliar with this kind of generator hacking.
In this snippet we define an AsyncSocket which wraps a native socket and has methods for each of the whats the scheduler supports.
class AsyncSocket
{
public function __construct(
private Socket $socket
)
{
}
public function read(int $maxsize = 1024)
{
yield ['read', $this->socket];
return socket_read($this->socket, $maxsize);
}
public function write(string $data)
{
yield ['write', $this->socket];
return socket_write($this->socket, $data);
}
public function accept()
{
yield ['read', $this->socket];
return new AsyncSocket(socket_accept($this->socket));
}
}
Using this, we can clean up the fib_server and introduce the final concept; yield from. This allows us to create a generator from another generator.
function fib_handler(AsyncSocket $client)
{
while (true) {
$data = yield from $client->read(1024);
if ($data === '' || $data === false) {
break;
}
$n = (int)($data);
$result = fibonacci($n);
yield from $client->write($result . "\n");
}
echo "Connection closed\n";
}
function fib_server(&$tasks)
{
$sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
socket_bind($sock, '127.0.0.1', 12021);
socket_listen($sock, 5);
try {
$asock = new AsyncSocket($sock);
while (true) {
$client = yield from $asock->accept();
$tasks[] = fib_handler($client);
}
} finally {
socket_close($sock);
}
}
Tying up loose threads with fibers #
If you take a look at older async code, you’ll see a lot of yield from calls. Even in other languages, such as Python, using yield and yield from used to be the way to implement asynchronous calls. While this works, it does feel like a hack. Nor is it immediately obvious what is happening, especially to the uninitiated.
Luckily fibers were introduced in PHP 8.1. These give us a native way to represent tasks and allow us to suspend the program, removing the need for ugly yield from calls. As it turns out our little fibonacci service can be easily adapted to use fibers. Here we replace yield from statements with calls to Fiber::suspend with the same array of data.
function fib_handler($client)
{
while (true) {
Fiber::suspend(['read', $client]);
$data = socket_read($client, 1024);
if ($data === '' || $data === false) {
break;
}
$n = (int)($data);
$result = fibonacci($n);
Fiber::suspend(['write', $client]);
socket_write($client, $result . "\n");
}
echo "Connection closed\n";
}
function fib_server(SplQueue $fibers)
{
$sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
socket_bind($sock, '127.0.0.1', 12021);
socket_listen($sock, 5);
try {
while (true) {
Fiber::suspend(['read', $sock]);
$client = socket_accept($sock);
// Start a new fiber (task) to handle the client
$fiber = new Fiber(fn () => fib_handler($client));
$fibers->enqueue($fiber);
}
} finally {
socket_close($sock);
}
}
function run(SplQueue $fibers)
{
$readWait = [];
$writeWait = [];
while ($fibers->count() || $readWait || $writeWait) {
// No active tasks, wait for I/O — but only if something is actually waiting
while (!$fibers->count() && ($readWait || $writeWait)) {
$readable = array_map(static fn($fiber) => $fiber[0], $readWait);
$writable = array_map(static fn($fiber) => $fiber[0], $writeWait);
$except = [];
// Poll the OS for socket activity
$result = socket_select($readable, $writable, $except, null);
if ($result === false) {
continue;
}
// Pull the tasks from the waiting area and add them back to the main queue
// Note, we must do this for both read and write
foreach ($readable as $socket) {
$id = spl_object_id($socket);
$fibers->enqueue($readWait[$id][1]);
unset($readWait[$id]);
}
foreach ($writable as $socket) {
$id = spl_object_id($socket);
$fibers->enqueue($writeWait[$id][1]);
unset($writeWait[$id]);
}
}
// Nothing runnable and nothing waiting: we're done
if (!$fibers->count()) {
continue;
}
/** @var Fiber $fiber */
$fiber = $fibers->dequeue();
if ($fiber->isTerminated()) {
continue;
}
$payload = $fiber->isStarted() ? $fiber->resume() : $fiber->start();
if ($payload === null) {
// Fiber ran to completion
continue;
}
[$why, $who] = $payload;
if (null === $why) {
continue;
}
$id = spl_object_id($who);
if ($why === 'read') {
$readWait[$id] = [$who, $fiber];
} elseif ($why === 'write') {
$writeWait[$id] = [$who, $fiber];
} else {
throw new \Exception("¯\_(ツ)_/¯");
}
}
}
To start the fiber based service, we have to instantiate a new Fiber object and pass in a callable. In our case we wrap the fib_server in a closure, which enables us to pass the SplQueue containing fibers as an argument to the server.
Async enabled programs are structured like a tree and need a starting task, I like to think of this as the entrypoint of the service. In practice you often have a synchronous bootstrapping phase, before starting the async mode.
$fibers = new SplQueue();
// Must start the server as a task to be able to pause and resume it
$fiber = new Fiber(fn () => fib_server($fibers));
$fibers->enqueue($fiber);
run($fibers);
Making the code read like sync code #
Using fibers to suspend the call stack is neat, but what if we rewrite our AsyncSocket from before to use fibers instead? We would still need to announce that we’re going to do IO, but crucially the developer doesn’t need to know that. In that case we could make the code read like sync code, while behaving like async code. Giving us the best of both worlds.
In this case we could write async aware building blocks that wrap a Socket. Each of these building blocks is responsible for deciding how to interact with the outside world, but crucially the developer doesn’t need to be aware that they are async. They just call a method on an object.
class AsyncSocket
{
public function __construct(private Socket $socket) {}
public function read(int $maxsize = 1024)
{
Fiber::suspend(['read', $this->socket]);
return socket_read($this->socket, $maxsize);
}
public function write(string $data)
{
Fiber::suspend(['write', $this->socket]);
return socket_write($this->socket, $data);
}
}
For the sake of brevity updating the server and scheduler to use the updated
AsyncSocketclass is left as an exercise for the reader.
function fib_handler(AsyncSocket $client)
{
while (true) {
$data = $client->read(1024);
if ($data === '' || $data === false) {
break;
}
$n = (int)($data);
$result = fibonacci($n);
$client->write($result . "\n");
}
}
So, should you build your own event loop? #
No. But you should know why the things you use look and work the way they do.
Everything we have built in this post already exists in a production ready form. ReactPHP and AMPHP follow this exact model; a scheduler, a waiting area and an interface with the OS to determine which sockets are ready. These frameworks also come with building blocks allowing you as the developer to focus on adding value to your program. Traditional async code built with coroutines “colors” the codebase, async functions may only ever be called from other async functions. Fibers solve this issue by allowing the call stack to be suspended from anywhere. Mature fiber based frameworks, such as AMPHP, hide this from the user offering a collection of packages (like our AsyncSocket) to make your async life less painful. Ranging from a DNS client to database drivers for MySQL and Postgres.
The more important lesson is the one hiding in plain sight in the benchmark. While the sockets stopped blocking, the fibonacci function never did. With 2 clients connected, both got slower. Adding more clients would make this slower still. An event loop doesn’t buy you CPU, it buys you the time you were wasting on the network. One slow request in an async application and the entire system grinds to a halt. This is why async frameworks are obsessive about moving CPU work off the loop as soon as possible and they often include tools to move work to a separate process pool. This of course comes with its own pitfalls; serialization and startup costs to name a few.
For the average PHP application where a request comes in, some database queries happen and a response goes out, you do not need any of this. You would be much better off with a way to offload work to a separate worker process via a broker such as Redis or RabbitMQ. Every framework has standard options for this. FPM’s share nothing model is still great and prevents a horde of problems you never even thought of.
Where async PHP earns its complexity is when you’re holding on to many, mostly idle, connections. Think websockets, long polling or proxying to slow upstreams. If that is not your problem, the boring solution is still the right one.