Forem: Gabriel Olivério

Streams in PHP: What you really need to know

Gabriel Olivério — Sun, 25 Aug 2024 18:28:10 +0000

Whether you've ever had to deal with local files, HTTP requests or compressed files, you dealt with streams but... did you really get to know them?

I think that this is one of the most misunderstood concepts in PHP, and, as a consequence I've seen quite a few bugs being introduced because the lack of some fundamental knowledge.

In this article I'll try to explain what streams really are and how to work with them. We will see many functions used to work with streams as well as lots of examples, but it is not my intention to "redocument" all of them in any way.

Before learning what streams are, we first need to approach resources.

Resources

Resources are simply references or pointers to external resources, such as a file, a database, network or SSH connection, for example.

There are several types of resources, such as curl - created by curl_init(), process, created by proc_open and stream, created by functions like fopen(), opendir among others.

Streams

Streams are the way that PHP generalises types of resources that have common behaviour, that is, resources can be read from and written to linearly, like a cassette tape (damn, I am getting old). Some examples of streams are file resources, HTTP response bodies and compressed files, just to name a few.

Streams are incredibly useful as they enable us to work with resources that range from a few bytes to several GBs in size and, an attempt of read them entirely, for example, would exhaust our available memory.

Creating a stream with `fopen`

 fopen(
    string $filename,
    string $mode,
    bool $use_include_path = false,
    ?resource $context = null
): resource|false

fopen opens a file or network resource[1], depending on the path provided to its first parameter. As said before, this resource is of type stream:

$fileStream = fopen('/tmp/test', 'w');
echo get_resource_type($fileStream); // 'stream'

If $filename is provided in the form scheme://, it is assumed to be a URL and PHP will try to find a supported protocol handlers/wrappers that matches the path, such as file:// - to handle local files, http:// - to work on remote HTTP/S resources, ssh2:// - to handle SSH connections or php:// - that allows us access PHP's own input and output streams, such as php://stdin, php://stdout and php://stderr.

$mode defines the type of access you require to the stream, that is, whether you need only read access, only write, read and write, read/write from the beginning of the stream or end, and so on.

The mode also depends on the type of resource you are working on. For example:

$fileStream = fopen('/tmp/test', 'w');
$networkStream = fopen('https://google.com', 'r');

Opening a writable stream using the wrapper https://, for example, does not work:

fopen('https://google.com', 'w'); // Failed to open stream: HTTP wrapper does not support writeable connections

[1] Using fopen with network or remote resources only works when allow_url_fopen is enabled on php.ini. For more information, check the documentation.

So, now we've got a stream resource, what can we do with them?

Writing into a file stream with `fwrite`

fwrite(resource $stream, string $data, ?int $length = null): int|false

fwrite enables us to write the contents provided to $data into a stream. If $length is supplied, it writes only the given supplied number of bytes. Let's see an example:

$fileStream = fopen('/tmp/test', 'w');

fwrite($fileStream,  "The quick brown fox jumps over the lazy dog", 10);

In this example, as we provided $length = 10, so only part of the content was written - "The quick " - ignoring the rest.

Notice that we opened the file stream with $mode = 'w', which enabled us to write content into the file. If, instead, we had opened the file with $mode = 'r', we would get a message such as fwrite(): Write of 8192 bytes failed with errno=9 Bad file descriptor.

Let's see another example, now writing the whole content into the file stream:

$fileStream = fopen('/tmp/test', 'w');

fwrite($fileStream,  "The quick brown fox jumps over the lazy dog");

Now, since we haven't provided $length, the whole content was written into the file.

Writing into a stream moves the position of the read/write pointer to end end of the sequence. In this case, the string written into the stream has 44 characters, therefore, the pointer's position now should be 43.

Besides writing into a file, fwrite can write in other types of streams, such as sockets. Example extracted from the docs:

$sock = fsockopen("ssl://secure.example.com", 443, $errno, $errstr, 30);
if (!$sock) die("$errstr ($errno)\n");

$data = "foo=" . urlencode("Value for Foo") . "&bar=" . urlencode("Value for Bar");

fwrite($sock, "POST /form_action.php HTTP/1.0\r\n");
fwrite($sock, "Host: secure.example.com\r\n");
fwrite($sock, "Content-type: application/x-www-form-urlencoded\r\n");
fwrite($sock, "Content-length: " . strlen($data) . "\r\n");
fwrite($sock, "Accept: */*\r\n");
fwrite($sock, "\r\n");
fwrite($sock, $data);

$headers = "";
while ($str = trim(fgets($sock, 4096)))
$headers .= "$str\n";

echo "\n";

$body = "";
while (!feof($sock))
$body .= fgets($sock, 4096);

fclose($sock);

Reading streams with `fread`

fread(resource $stream, int $length): string|false

With fread you can read up to $length bytes from a stream, starting from the current read pointer. It is binary-safe and it works with local and network resources, as we will see in the examples.

Calling fread consecutively will read a chunk and then move the read pointer to the end of this chunk. Example, considering the file written in the previous example:

# Content: "The quick brown fox jumps over the lazy dog"
$fileStream = fopen('/tmp/test', 'r');

echo fread($fileStream, 10) . PHP_EOL;      // 'The quick '
echo ftell($fileStream); // 10
echo fread($fileStream, 10) . PHP_EOL;      // 'brown fox '
echo ftell($fileStream); // 20

We will come back to ftell soon, but what it does is simply return the current position of the read pointer.

The reading stops (returning false), as soon as one of the following occurs (copied from the docs, you'll understand later):

length bytes have been read

EOF (end of file) is reached

a packet becomes available or the socket timeout occurs (for network streams)

if the stream is read buffered and it does not represent a plain file, at most one read of up to a number of bytes equal to the chunk size (usually 8192) is made; depending on the previously buffered data, the size of the returned data may be larger than the chunk size.

I don't know if you had the same felling, but this last part is pretty cryptic, so let's break it down.

"if the stream is read buffered"

Stream reads and writes can be buffered, that is, the content may be stored internally. It is possible to disable/enable the buffering, as well as set their sizes using stream_set_read_buffer and stream-set-write-buffer, but according to this comment on the PHP doc's Github, the description of these functions can be misleading.

This is where things get interesting, as this part of the documentation is really obscure. As per the comment, setting stream_set_read_buffer($stream, 0) would disable the read buffering, whereas stream_set_read_buffer($stream, 1) or stream_set_read_buffer($stream, 42) would simply enable it, ignoring its size (depending on the stream wrapper, which can override this default behaviour).

"... at most one read of up to a number of bytes equal to the chunk size (usually 8192) is made"

The chunk size is usually 8192 bytes or 8 KiB, as we will confirm in a bit. We can change this value using stream_set_chunk_size. Let's see it in action:

$f = fopen('https://dl-cdn.alpinelinux.org/alpine/v3.20/releases/x86_64/alpine-standard-3.20.2-x86_64.iso', 'rb');

$previousPos = 0;
$chunkSize = 1024;
$i = 1;

while ($chunk = fread($f, $chunkSize)) {
    $bytesRead = (ftell($f) - $previousPos);
    $previousPos = ftell($f);

    echo "Iteration: {$i}. Bytes read: {$bytesRead}" . PHP_EOL;

    $i++;
}

Output:

Iteration: 1. Bytes read: 1024
Iteration: 2. Bytes read: 1024
Iteration: 3. Bytes read: 1024
...
Iteration: 214016. Bytes read: 1024
Iteration: 214017. Bytes read: 169

What happened in this case was clear:

We wanted up to 1024 bytes in each fread call and that's what we got
In the last call there were only 169 bytes remainder, which were returned
When there was nothing else to return, that is, EOF was reached fread returned false and the loop finished.

Now let's increase considerably the length provided to fread to 1 MiB and see what happens:

$f = fopen('https://dl-cdn.alpinelinux.org/alpine/v3.20/releases/x86_64/alpine-standard-3.20.2-x86_64.iso', 'rb');

$previousPos = 0;
$chunkSize = 1048576; // 1 MiB
$i = 1;

while ($chunk = fread($f, $chunkSize)) {
    $bytesRead = (ftell($f) - $previousPos);
    $previousPos = ftell($f);

    echo "Iteration: {$i}. Bytes read: {$bytesRead}" . PHP_EOL;

    $i++;
}

Output:

Iteration: 1. Bytes read: 1378
Iteration: 2. Bytes read: 1378
Iteration: 3. Bytes read: 1378
...
Iteration: 24. Bytes read: 1074
Iteration: 25. Bytes read: 8192
Iteration: 26. Bytes read: 8192
...
Iteration: 26777. Bytes read: 8192
Iteration: 26778. Bytes read: 8192
Iteration: 26779. Bytes read: 293

So, even though we tried to read 1 MiB using fread, it read up to 8192 bytes - same value that the docs said it would. Interesting. Let's see another experiment:

$f = fopen('https://dl-cdn.alpinelinux.org/alpine/v3.20/releases/x86_64/alpine-standard-3.20.2-x86_64.iso', 'rb');

$previousPos = 0;
$chunkSize = 1048576; // 1 MiB
$i = 1;

stream_set_chunk_size($f, $chunkSize); // Just added this line

while ($chunk = fread($f, $chunkSize)) {
    $bytesRead = (ftell($f) - $previousPos);
    $previousPos = ftell($f);

    echo "Iteration: {$i}. Bytes read: {$bytesRead}" . PHP_EOL;

    $i++;
}

And the output:

Iteration: 1. Bytes read: 1378
Iteration: 2. Bytes read: 1378
Iteration: 3. Bytes read: 1378
...
Iteration: 12. Bytes read: 533
Iteration: 13. Bytes read: 16384
Iteration: 14. Bytes read: 16384
...
Iteration: 13386. Bytes read: 16384
Iteration: 13387. Bytes read: 16384
Iteration: 13388. Bytes read: 13626

Notice that now fread read up to 16 KiB - not even close to what we wanted, but we've seen that stream_set_chunk_size did work, but there are some hard limits, that I suppose that depends also on the wrapper. Let's put that in practice with another experiment, using a local file this time:

$f = fopen('alpine-standard-3.20.2-x86_64.iso', 'rb');

$previousPos = 0;
$chunkSize = 1048576; // 1 MiB
$i = 1;

while ($chunk = fread($f, $chunkSize)) {
    $bytesRead = (ftell($f) - $previousPos);
    $previousPos = ftell($f);

    echo "Iteration: {$i}. Bytes read: {$bytesRead}" . PHP_EOL;

    $i++;
}

Output:

Iteration: 1. Bytes read: 1048576
Iteration: 2. Bytes read: 1048576
...
Iteration: 208. Bytes read: 1048576
Iteration: 209. Bytes read: 1048576

Aha! So using the local file handler we were able to fread 1 MiB as we wanted, and we did not even need to increase the buffer/chunk size with stream_set_chunk_size.

Wrapping up

I think that now the description is less cryptic, at least. Let's read it again (with some interventions):

if the stream is read buffered ...

and it does not represent a plain file (that is, local, not a network resource), ...

at most one read of up to a number of bytes equal to the chunk size (usually 8192) is made (and in our experiments we could confirm that this is true, at least one read of the chunk size was made); ...

depending on the previously buffered data, the size of the returned data may be larger than the chunk size (we did not experience that, but I assume it may happen depending on the wrapper).

There is definitely some room to play here, but I will challenge you. What would happen if you disable the buffers while reading a file? And a network resource? What if you write into a file?

`ftell`

ftell(resource $stream): int|false

ftell returns the position of the read/write pointer (or null when the resource is not valid).

# Content: "The quick brown fox jumps over the lazy dog"
$fileStream = fopen('/tmp/test', 'r');

fread($fileStream, 10); # "The quick "
echo ftell($fileStream); 10

`stream_get_meta_data`

stream_get_meta_data(resource $stream): array

stream_get_meta_data returns information about the stream in form of an array. Let's see an example:

# Content: "The quick brown fox jumps over the lazy dog"
$fileStream = fopen('/tmp/test', 'r');
var_dump(stream_get_meta_data($fileStream)):

The previous example would return in something like this:

array(9) {
  ["timed_out"]=>
  bool(false)
  ["blocked"]=>
  bool(true)
  ["eof"]=>
  bool(false)
  ["wrapper_type"]=>
  string(9) "plainfile"
  ["stream_type"]=>
  string(5) "STDIO"
  ["mode"]=>
  string(1) "r"
  ["unread_bytes"]=>
  int(0)
  ["seekable"]=>
  bool(true)
  ["uri"]=>
  string(16) "file:///tmp/test"
}

This function's documentation is pretty honest describing each value ;)

`fseek`

fseek(resource $stream, int $offset, int $whence = SEEK_SET): int

fseek sets the read/write pointer on the opened stream to the value provided to $offset.
The position will be updated based on $whence:

SEEK_SET: Position is set to $offset, that is, if you call
SEEK_CUR: Position is set based on the current one, that is, current + $offset
SEEK_END: Position is set to End Of File + $offset.

Using SEEK_END we can provide a negative value to $offset and go backwards from EOF. Its return value can be used to assess if the position has been set successfully (0) or has failed (-1).

Let's see some examples:

# Content: "The quick brown fox jumps over the lazy dog\n"
$fileStream = fopen('/tmp/test', 'r+');

fseek($fileStream, 4, SEEK_SET);
echo fread($fileStream, 5);         // 'quick'
echo ftell($fileStream);            // 9

fseek($fileStream, 7, SEEK_CUR);
echo ftell($fileStream);            // 16, that is, 9 + 7
echo fread($fileStream, 3);         // 'fox'  

fseek($fileStream, 5, SEEK_END);    // Sets the position past the End Of File
echo ftell($fileStream);            // 49, that is, EOF (at 44th position) + 5
echo fread($fileStream, 3);         // ''  
echo ftell($fileStream);            // 49, nothing to read, so read/write pointer hasn't changed
fwrite($fileStream, 'foo');
ftell($fileStream);                 // 52, that is, previous position + 3
fseek($fileStream, -3, SEEK_END);
ftell($fileStream);                 // 49, that is, 52 - 3
echo fread($fileStream, 3);         // 'foo'

Some important considerations

As we've seen in this example, it is possible we seek past the End Of File and even read in an unwritten area (which returns 0 bytes), but some types of streams do not support it.
An important consideration is that not all streams can be seeked, for instance, you cannot fseek a remote resource:

$f = fopen('https://dl-cdn.alpinelinux.org/alpine/v3.20/releases/x86_64/alpine-standard-3.20.2-x86_64.iso', 'rb');
fseek($f, 10); WARNING  fseek(): Stream does not support seeking

This obviously makes total sense, as we cannot "fast-forward" and set a position on a remote resource. The stream in this case is only read sequentially, like a cassette tape.
We can determine if the stream is seekable or not via the seekable value returned by stream_get_meta_data that we've seen before.

We can fseek a resource opened in append mode (a or a+), but the data will always be appended.

`rewind`

rewind(resource $stream): bool

This is a pure analogy of rewinding a videotape before returning it to video store. As expected, rewind sets the position of the read/write pointer to 0, which is basically the same as calling fseek with $offset 0.

The same considerations we've seen for fseek applies for rewind, that is:

You cannot rewind an unseekable stream
rewind on a resource opened in append mode will still write from the current position - the write pointer is not updated.

How about `file_get_contents`?

So far we've been working directly with resources. file_get_contents is a bit different, as it accepts the file path and returns the whole file content as a string, that is, it implicitly opens the resource.

file_get_contents(
   string $filename,
   bool $use_include_path = false,
   ?resource $context = null,
   int $offset = 0,
   ?int $length = null
): string|false

Similar to fread, file_get_contents can work on local and remote resources, depending on the $filename we provide:

# Content: "The quick brown fox jumps over the lazy dog"

echo file_get_contents('/tmp/test'); // "The quick brown fox jumps over the lazy dog\n"

echo file_get_contents('https://www.php.net/images/logos/php-logo.svg'); // "<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 -1 100 50">\n ..."

With $offset we can set the starting point to read the content, whereas with length we can get a given amount of bytes.

# Content: "The quick brown fox jumps over the lazy dog"
echo file_get_contents('/tmp/test', offset: 16, size: 3); // 'fox'

offset also accepts negative values, which counts from the end of the stream.

# Content: "The quick brown fox jumps over the lazy dog"
echo file_get_contents('/tmp/test', offset: -4, size: 3); // 'dog'

Notice that the same rules that govern fseek are also applied for $offset, that is - you cannot set an $offset while reading remote files, as the function would be basically fseek the stream, and we've seen that it does not work well.

The parameter context makes file_get_contents really flexible, enabling us set, for example:

Set a different HTTP method, such as POST instead of the default GET
Provide headers and content to a POST or PUT request
Disable SSL verification and allow self-signed certificates

We create a context using stream_context_create, example:

$context = stream_context_create(['http' => ['method' => "POST"]]);
file_get_contents('https://a-valid-resource.xyz', context: $context);

You can find the list of options you can provide to stream_context_create in this page.

$networkResource = fopen('https://releases.ubuntu.com/24.04/ubuntu-24.04-desktop-amd64.iso', 'r');

while ($chunk = fread($networkResource, 1024)) {
   doSomething($chunk);
}

Which one to use? `fread`, `file_get_contents`, `fgets`, another one?

The list of functions that we can use to read local or remote contents is lengthy, and each function can be seen as a tool in your tool belt, suitable for a specific purpose.

According to the docs, file_get_contents is the preferred way of reading contents of a file into a string, but, is it appropriate for all purposes?

What if you know that the content is large? Will it fit into memory?
Do you need it entirely in memory or can you work in chunks?
Are you going to work on local or remote files?

Ask yourself these (and other questions), make some performance benchmark tests and select the function that suits your needs the most.

PSR-7's `StreamInterface`

PSR defines the StreamInterface, which libraries such as Guzzle use to represent request and response bodies. When you send a request, the body is an instance of StreamInterface. Let's see an example, extracted from the Guzzle docs:

$client = new \GuzzleHttp\Client();
$response = $client->request('GET', 'http://httpbin.org/get');

$body = $response->getBody();
$body->seek(0);
$body->read(1024);

I suppose that the methods available on $body look familiar for you now :D

StreamInterface implements methods that resemble a lot the functions we've just seen, such as:

seek()
tell()
eof()
read
write
isSeekable
isReadable()
isWritable
and so on.

Last but not least, we can use GuzzleHttp\Psr7\Utils::streamFor to create streams from strings, resources opened with fopen and instances of StreamInterface:

use GuzzleHttp\Psr7;

$stream = Psr7\Utils::streamFor('string data');
echo $stream;                   // string data
echo $stream->read(3);          // str
echo $stream->getContents();    // ing data
var_export($stream->eof());     // true
var_export($stream->tell());    // 11

Summary

In this article we've seen what streams really are, learned how to create them, read from them, write to them, manipulate their pointers as well as clarified some obscured parts regarding read a write buffers.

If I did a good job, some of the doubts you might have had regarding streams are now a little bit clearer and, from now on, you'll write code more confidently, as you know what you are doing.

Should you noticed any errors, inaccuracies or there is any topic that is still unclear, let me know in the comments and I'd be glad to try to help.

Concurrency and parallelism in PHP (part 1)

Gabriel Olivério — Mon, 29 Jul 2024 18:30:57 +0000

You may be "Yeah, big deal, another article about Fibers", but NO! I mean, yes, you are partly right, I am also going to approach Fibers, but also other features you may not be aware of.

Before we dive into how to achieve concurrency and parallelism with PHP, we need to talk about what these two things are, and no, they are not exactly the same thing. If you are a computer science sorcerer that knows everything about this topic, you can skip this first part, you'll be forgiven 😛

Concurrency and parallelism are terms that are sometimes used interchangeably, however, they're conceptually different. Let's go to the kitchen and use an analogy to understand these terms.

First, a bit of theory

Suppose we are functional adults and we're going to prepare hamburgers for dinner, but you are like me, who can only do one thing well at a time. Our simplified "main process" would be something like this:

Get the minced meat
Get the buns
Shape the hamburgers
Cut the buns
Separate the cheese
Fry the hamburgers
Add the cheese to melt
Assemble our hamburgers

Following this sequential flow and having a single worker, namely "you", each task can only start once the previous one has finished.

This is basically how our PHP scripts work when they do not use parallelism and concurrency features - you can only iterate over entries you obtained from the database after the script has connected to the database, executed the query, waited for the results and closed the connection. You are blocked or, I/O blocked, as some literature puts it.

Back to the kitchen, now suppose you are not like me 🙂

You shape the hamburgers
Get the buns
Separate the cheese
Do until all the hamburgers and are assembled
1. Fry a hamburger
2. While one is frying, you can cut one or more buns
3. Assemble the hamburgers as they are fried and the buns are cut
4. Repeat until all hamburgers are done

Now we are capable of dealing with multiple things at once, checking visually if parts of our process have completed so we can start the other and, more importantly, we are not blocked while the hamburgers are fried!

This workflow demonstrates exactly what concurrency is. In concurrency we are not necessarily doing multiple things at once, indeed, multiple things are happening at the same time, but YOU are doing only one thing at a time, either cutting buns or taking care of the hamburgers while they fry.

Using concurrency features in your PHP script you would be, for example, capable of executing multiple queries into a database at once, do other processing and, when these queries return, you can then process them and keep your usual flow, for example. We are not executing each and every query sequentially and waiting for each one return results to be able to execute the next one. We are not that blocked anymore!

Now back once again to the kitchen, and I promise this will be the last time. This time you are not cooking alone anymore, you have a kitchen partner. I'll be generic and call you and your partner as worker 1 and worker 2, and both are working at the same time.

Worker 1

Get the minced meat
Shape one hamburger
Start frying. Once one hamburger is in our frying pan:
We can shape other hamburgers
Take care of our hamburgers being fried
Add the cheese that hopefully our worker 2 has separate

Worker 2

Get the buns
Cut the buns
Separate the cheese
Assemble the hamburgers as they are fried

Now we are effectively doing multiple things at once, thanks to the help of our second worker!

Notice now there is some coordinating work to do, such as checking if the other worker has finished one step before proceeding with your task. In a similar fashion, our PHP script could have two workers, for example, one performing one task while the other performs another. They can (actually they need to) communicate with one another and synchronise their tasks.

Parallelism and concurrency actually have two distinct objectives.
Concurrency aims at dealing with multiple things at once, not getting blocked whenever we need to wait on external outcomes. Parallelism, on the other hand, focuses on doing multiple things at once, potentially maximising the performance and use of our resources without (hopefully) compromising the end result.

Now going back to PHP

In this article I am only going to cover concurrency, parallelism will be explored in the second part. First, we will see what building blocks PHP itself provides to us and afterwards we will check out some libraries that can make our lives much easier.

Generators (and coroutines)

According to the docs, Generators "provide an easy way to implement iterators without the overhead of creating a class that implements the Iterator interface" (and the overall complexity of managing its state). Generators are not something necessarily new, it was added back in 2013, on version 5.5.

Conceptually, a Generator function is any function that contains a yield statement. The yield statement is similar to the return, except that instead of stopping the execution of the function and returning a result to the caller, it provides a value and pauses its execution.

Since Generator objects implement the Iterator interface, we can do things like this, for example (extracted from the Generators's RFC):

As getLinesFromFile has a yield statement, it is therefore a Generator function. $lines, on the other hand, is a Generator object and, since the Generator class implements the Iterator interface, it can be iterated like we did on line 14.

Generators work passing control back and forth between the generator function and the caller, in our example, the foreach. It works this way:

When the Generator is created on line 13, it is not executed - it is only executed when we call the method next, something that our loop does implicitly. The execution happens as usual until it reaches the yield statement, on line 7. At this point, it pauses the execution and its current value is handed over to the caller. In the next foreach iteration, the method next is called again, resuming the generator at the point it had been paused.

At this point you may be thinking: So, what does generators have to do with concurrency? Turns out that, even though the articles cover more this particular aspect of Generators, they can do much more! “Hidden” in the RFC, there is the following paragraph:

Generators can also be used the other way around, i.e. instead of producing values they can also consume them. When used in this way they are often referred to as enhanced generators, reverse generators or coroutines.

That's right. Since generator functions are interruptible, we can use them as part of cooperative multitasking, also known as non-preemptive multitasking. In this approach, each executing unit (the generator or coroutine, in our case) voluntarily yields control periodically when blocked or idle. It is called cooperative because every generator must cooperate in order for the whole thing to work. With this approach, we need something that executes our generators and lets them return control back to it voluntarily.

Enough of chit-chat, let's check some code.

You can also check out this and all the next examples in this repository.

First let’s clone the repo and run a container with PHP and Apache:



docker run -p 9000:80 -v "$(pwd)"/src:/var/www/html -d php:8.2-apache

The class Task is merely a wrapper for our tasks, it accepts an ID and a generator. The ID will be set by the Scheduler, as follows.

The class Scheduler is responsible for executing the tasks. We use its newTask method to create the task, which stores it in a queue and in map, to control which tasks to run.

In each iteration, the method run obtains one task from the queue, executes it and, if the task is terminated - that is, if the generator is not valid anymore - the task is removed from the queue.

The function fetchResource, uses curl_multi* functions to fetch multiple resources at the same time. The function curl_multi_exec does not block our execution while it waits for its response; if instead the function curl_exec had been used, this code simply would not have worked. This is a very important point, which we will come back to shortly.

We are using bare curl functions here on purpose, in order not to hide anything using libraries. As we will see later in this article, libraries such as Guzzle allow sending multiple concurrent requests, returning Promises similar to those we find in other languages.

We execute our do/while loop while our request is still running, but we interrupt the function's execution and return the control back to our scheduler once the execution reaches the yield statement. This interruption makes that the next loop iteration only happens after we give another task a chance to execute. The next time that the first task has been given a chance to process, it will merely check if the resource has returned, if so, it will terminate, otherwise, it will pause itself once again and return the control back to the scheduler.

Executing this code you'd see something like this (shortened for brevity):

It is beyond the scope of this article to cover all the use cases of Generators, but at this point you may have noticed that they can do much more than allow you to iterate on stuff, right?

If you are interested in going deep in this subject, I strongly recommend this Nikita Popov's article. He is simply the Generators RFC's author and proponent of many other very cool features we all use every day. Part of the code we’ve just seen was borrowed from the aforementioned article.

Even though this Generator's side is quite obscure for most PHP developers, it has been extensively explored by well-known libraries, such as ReactPHP and AMPHP, which we will see later on.

A word about blocking and non-blocking functions

Unfortunately, most PHP functions are blocking, with some rare exceptions. Besides curl_multi* functions that we’ve just seen, we can set streams as non-blocking, with the function stream_set_blocking, and then use stream_select to wait for them to change status for some time.

We can also work with sockets in a similar way, setting the socket as non-blocking with socket_set_nonblock, and then using socket_select to wait for the sockets to change the status.

Fibers

As we’ve just seen, it is already possible to achieve concurrency in our applications via Generators, so… what’s the real utility of Fibers, and what’s the difference between Fibers and Generators?

The Fibers RFC was proposed in order to address one very common problem that other languages also suffer from, the problem known as the “What colour is your function”. Summarising, this problem is characterised by the distinction we've got between synchronous and asynchronous functions:

Asynchronous functions need to be called in a different way (such as with the await keyword, in JavaScript)
Synchronous functions cannot call asynchronous ones (even though the other way around is possible)

If you have ever worked with Promises/async/await in JavaScript, you surely noticed this effect. Whenever we use await in a function, this function also needs to be asynchronous and, therefore, needs to be marked as async.

Fibers represent full-stack, interruptible functions, which can be suspended from anywhere in the call-stack and have its execution paused from within the fiber until it is finally resumed at a later time. In a sense, they are very similar to Generators, but there are a few important distinctions:

Fibers pause the entire execution stack, so the direct caller of the function does not need to change how it invokes the function
Fibers have their own call-stack, whereas Generators do not. This allows them to be paused within deeply nested function calls

Important to say that, in the same way as Generators, we can use Fibers to achieve concurrency, but not parallelism, that is, even though you may have multiple fibers, only one is actively being executed. Fibers are also known as green-threads or coroutines in other languages, but don't be misled by its name - fibers exist within a single process thread, and that's the reason that only one Fiber is executed at a time.

Example 1 - our hello world.

Exactly, this is our hello world using Fibers. I admit, nothing fancy, and definitely no concurrency, but we are close, I promise. Here's what's happening:

Fibers are created passing a callable to it - it is not executed though, only created. Inside the fiber, during its execution, we suspend it with a value - the string "Hello". On line 4 we print the value we have passed as an argument to the resume method, which we use to resume the fiber’s execution on line 11.

On line 7 the fiber is actually executed, until we suspend it on line 2 with "Hello". The instruction on line 9 prints “Value from fiber suspending: Hello”. On line 11 we resume the fiber execution, which will execute instruction present on line 4, completing our Hello World.

The whole output of our example is the following:



Value from fiber suspending: Hello
Value used to resume fiber: World

Fibers can be started, suspended from anywhere in the call stack, throw an exception/error and terminate. We can also check for their current states, using the isSuspended(), isStarted(), isRunning() and isTerminated(), and that is basically its API.

Example 2 - sending concurrent requests

Let’s modify a bit our classes Task and Scheduler to use Fibers instead of generators.

The task now accepts a callable (not a Generator object anymore) and, internally, it creates a Fiber out of the defined callable. In the run method, we first check if the fiber has been already started and, if not, we start it, otherwise, the fiber is just resumed.

Our simple scheduler now also accepts a callable rather than a Generator function as a parameter to the newTask method. Other than that, it remains unchanged.

The function fetchResource is also basically the same as previously presented, but instead of yielding the control to the caller, we suspend itself, on line 17.

As a result, we should see the same result as we’ve seen with the Generators example:

A final word on Fibers

You might have noticed that working with Fibers and Generators can add a good level of complexity to our code base if we use them directly.

Even though they are the building blocks that PHP provides to us, if you notice in the Fibers’ RFC, the author was quite explicit about its purpose and audience:

Fibers are an advanced feature that most users will not use directly. This feature is primarily targeted at library and framework authors to provide an event loop and an asynchronous programming API.

The Fiber API is not expected to be used directly in application-level code. Fibers provide a basic, low-level flow-control API to create higher-level abstractions that are then used in application code.

Having said that, let’s have a look at some libraries to make our lives a bit easier.

Guzzle

As mentioned before, Guzzle provides a very convenient way to send concurrent requests, which under the hood, as this article is being written, is powered by Generators. If this is your only need for concurrency, go for it. Let's see an example:

The method getAsync, as well as their similar postAsync, putAsync, deleteAsync and so on, return a Promise, which implements the Promises/A+ spec. You can chain their calls or wait for all of them concurrently, as we've done. Executing this script would result in something like:

AMPHP

AMPHP is a collection of libraries to deal with concurrency and, as of version 3.0, is powered by Fibers.

One of the nicest things about this project is that it also provides non-blocking alternatives for many PHP functions, such as for handling files, executing MySQL queries or communicating with Redis, without requiring the installation of extensions, like Swoole does. Even though this approach may have some performance limitations, it can fit well in many scenarios.

Let's have a look in an example that uses the non-blocking MySQL connector:

We start defining the connection pool and set up our example database. Both tables, tmp1 and tmp2 are created synchronously.

The foreach defined in lines 17 to 20 we execute our INSERT statements using the async method. This method returns a Future object, which represents the eventual result of an asynchronous operation. A Future can be in pending, completed successfully or errored states. This is basically the same concept we find in other languages, such as JavaScript.

In line 22 we use the await combinator to wait for all Futures to complete, then we do the same process to query the results from line 24 to 27. In line 27 we can see that the async function returns an array whose keys match the ones from the iterable provided to it and their completion values.

Later on we use the function async and the fetchRow to get the obtained results.

The example is quite simple and it probably would not make sense to execute them asynchronously when we have 10 insertions to each table, but can be useful when dealing with multiple databases, for example.

Similar to the await combinator, we've got:

awaitAll, which awaits all the Future objects and returns their results as [$errors, $values]
awaitFirst, that returns the first completed Future object, whether successfully completed or errored
awaitAny, which returns the first successfully completed Future
awaitAnyN, which is similar to the await, but tolerates individual failures. It accepts the $count parameter, that we can set the number of Future objects that must be returned once successfully completed.

The next example will make use of the HttpClient library, which is similar to Guzzle in a sense that allows us to make asynchronous requests, but with some special powers. We will try to obtain an HTTP resource and, if it is too slow for our standards, we fallback on a secondary resource:

As we can see in line 12, we are providing a TimeoutCancellation to the function await. This causes a CancelledException once we reach 2 seconds of execution, making us to fall back on the secondary resource, which could be read a possibly-outdated read-only database replica, for example. Executing this script would print the "I am waaay faster", echoed by the script fast.php, instead of the 10 seconds that would take if we had insisted on the main resource - slow.php.

There are other Cancellation exceptions we can use, such as the SignalCancellation, which cancels the Future after receiving a specified signal and DeferredCancellation, which allows manual cancellation when calling the DeferredCancellation::cancel().

It is clear that it is way easier to work with Fibers consuming AMPHP than manually, right?

ReactPHP

ReactPHP is a low-level library for event-driven programming in PHP. It provides an event loop and utilities to work with streams, DNS resolver, network client/server, HTTP client/server and interaction with processes.

As AMPHP, it is highly modular, so we can install only the packages we need. Let's check our first example, similar to what we've done with AMPHP and Guzzle, consuming multiple resources simultaneously:

This example uses the HTTP package, which provides the ability to consume and serve HTTP resources. The method get returns a Promise, an implementation of CommonJS Promises/A for PHP, which is provided by one of the ReactPHP packages - Promise.

Executing this file in our terminal, we can see this output:

That is, even though we have sent the request to the slow resource first, it was the promise associated with the second request that was resolved first.

We can use the combinator all, similarly to what we've done with AMPHP:

Or use the race function, and guess who's the winner 😛:

Besides enabling us to send concurrent requests, ReactPHP's HTTP package enables us to write HTTP and HTTPS servers that can handle multiple concurrent requests without blocking. Let's see an example:

In this example we create an HttpServer object, which responds immediately to GET requests and, for any other HTTP methods, we defer the response for 2 whole seconds using the await function, provided by the package Async, along with the function sleep, provided by the package PromiseTimer. This way we simulate an asynchronous operation, such as obtaining external HTTP resources or executing database queries.

This function sleep, unlike the one provided natively by PHP, is non-blocking. Executing this PHP script will launch an HTTP server that listens to the port 9090. With the server running, we can send multiple GET and POST requests and check that, even though there is a single process running, our script is capable of dealing with multiple requests at the same time.

ReactPHP is a great library and it can do much more things that I could cover with this introductory article, nor was this my intention. Should you need to process streams of any type, launch HTTP or Socket, consume HTTP resources, interact with cache, resolve DNS or any other asynchronous operation, you’ll be in good hands with ReactPHP.

Conclusion

As we’ve seen, it may not be as easy as with JavaScript or Python to achieve concurrency, but it is possible, even though it requires some external packages and additional effort. This means you don’t need to rewrite your entire application if you need it to deal with concurrency, isn’t that great?

There is still a lot of ground to cover, as it is not possible to explore all use cases and libraries available out in the wild in this introductory article. Only the aforementioned OpenSwoole deserves its own article!

In the next part of this article we will explore parallelism with PHP. Are you excited? Cause I am.

Forem: Gabriel Olivério

Streams in PHP: What you really need to know

Resources

Streams

Creating a stream with fopen

Writing into a file stream with fwrite

Reading streams with fread

"if the stream is read buffered"

"... at most one read of up to a number of bytes equal to the chunk size (usually 8192) is made"

Wrapping up

ftell

stream_get_meta_data

fseek

Some important considerations

rewind

How about file_get_contents?

Which one to use? fread, file_get_contents, fgets, another one?

PSR-7's StreamInterface

Summary

Concurrency and parallelism in PHP (part 1)

First, a bit of theory

Now going back to PHP

Generators (and coroutines)

A word about blocking and non-blocking functions

Fibers

Example 1 - our hello world.

Example 2 - sending concurrent requests

A final word on Fibers

Guzzle

AMPHP

ReactPHP

Conclusion

Creating a stream with `fopen`

Writing into a file stream with `fwrite`

Reading streams with `fread`

`ftell`

`stream_get_meta_data`

`fseek`

`rewind`

How about `file_get_contents`?

Which one to use? `fread`, `file_get_contents`, `fgets`, another one?

PSR-7's `StreamInterface`