Forem: Bruno Clemente

Stream: a library to lazily process data with PHP.

Bruno Clemente — Wed, 02 Feb 2022 12:28:56 +0000

First of all, when we are talking about streams inside the PHP universe, the thing that will probably go to your head is the streamable resources of PHP. This is not what I will talk about today. Stream is a PHP library that helps us to define a pipeline of data transformations ergonomically and lazily. It is named "stream" because it was heavily inspired by the Elixir Stream.

So, let's take a look at it. First, add it to your project using composer composer require ebanx/stream. Then, we can create a very simple example to illustrate how it works.

use EBANX\Stream\Stream;

$result = Stream::rangeInt(0, 5)
    ->map(function (int $value): int {
        return $value + 3;
    })
    ->filter(function (int $value): bool {
        return $value % 2 === 0;
    })
    ->collect();
print_r($result);

The output of this example should be:

Array 
( 
   [0] => 4
   [1] => 6
   [2] => 8
)

Ok, now we need to understand what is happening here. The first thing that we do is to create a Stream. There are 4 methods that allow us to create a stream: Stream::of, Stream::ofKeyValueMap, Stream::rangeInt, Stream::rangeFloat. The last two allow us to create a stream by providing a numeric range. Note that this is lazy, which means that each element of the range will be only created when you consume it. We can create the following Stream and it would not use a crazy amount of memory from your computer: Stream::rangeInt(PHP_INT_MIN, PHP_INT_MAX).

The other two stream creators allow us to create a stream providing an iterable to it. The Stream::of will ignore any key that each element might have. Stream::ofKeyValueMap will transform each element into an array tuple where the first element is the key and the second one is the element value.

Now, in the example, following the stream creation there are two transformation methods. The first one is the map. In the map, we pass to it a callable. This callable will be called for each element that is being processed and pass on the value that the callable returns to the stream chain. So, in our example, we add 3 to each element of the stream.

The next transformation method is the filter. Here we also pass a callable that will receive as a parameter each element of the stream. But this callable needs to return a boolean. If it returns true, the element will follow to the next step of the stream. If it returns false, the element will not continue and it will be filtered out. So in this case, we are only keeping the even numbers.

The last method in the chain is the collect. This is the only eager method that we saw this far. This method will consume the entire stream by collecting each element into an array. In the end, it returns this array. After calling this method, the original stream goes into an invalid state and you cannot work with it anymore. All methods that consume the stream have commentaries that point that out.

A little more focus on the Lazy part.

The Stream tries to be as lazy as it can. Because of that, the following code will not output anything.

use EBANX\Stream\Stream;

$result = Stream::of([1, 2])
    ->map(function (int $value): int {
        echo "Original Element $value\n";
        return $value * 5;
    })
    ->map(function (int $value): int {
        echo "After first map $value\n";
        return $value / 2;
    })

This happens because map is lazy. If no one tries to use/consume the stream, we will never apply the map. So, let's try again by adding a 'sum' to the end.

use EBANX\Stream\Stream;

$result = Stream::of([1, 2])
    ->map(function (int $value): int {
        echo "Original Element $value\n";
        return $value * 5;
    })
    ->map(function (int $value): int {
        echo "After first map $value\n";
        return $value / 2;
    })
    ->sum();
echo "The sum is $sum";

This should output the following:

Original Element 1
After first map 5
Original Element 2
After first map 10
The sum is 7

Note the order. We don't apply the first map to all elements and then apply the second one. We apply all transformations to each element as soon as they are consumed. This allows us to execute fewer transformations in case we are using something like take, where we can take only N elements of the stream.

The reduce.

There are a lot of different transformations and ways to consume your streams. But I want to focus a little bit on the reduce. The reduce is an eager method so it consumes the stream. It receives an accumulator initial value and a callable. Then, for each element of the stream the callable will receive the current value of the accumulator and the current element. The return of the callable will be the new value of the accumulator. After the last iteration, it returns whatever is in the accumulator.

The powerful thing about the reduce is that all other methods that consume the stream could be written using the reduce. The only reason that we added the other methods is for easy-of-use. For example, here we are implementing the sum using the reduce.

$result = Stream::rangeInt(0, 5)
    ->reduce(0, function(int $acc, int $value): int {
        return $acc + $value;
    })

Arrays in the parameters.

Let's try to implement the collect method using the reduce:

$result = Stream::rangeInt(0, 5)
    ->reduce([], function(array $acc, int $value): array {
        $acc[] = $value;
        return $acc;
    })

If you execute it, you will see that it will work as expected. But what about performance? Let's create a range of 100000 entities and check the performance using reduce and using collect. These are the results on my machine:

Using collect:

real    0m0.128s
user    0m0.076s
sys 0m0.032s

Using reduce:

real    1m29.543s
user    0m46.033s
sys 0m43.108s

Wow, that is a BIG difference. Why does it take so much more to execute it using reduce?? Well, that is because of PHP and arrays.

PHP will pass an array by value in functions parameters. What this means is that every time that we try to change and array that we received as a parameter, PHP will clone the entire array to a new memory location. In our example, we are doing this 100k times and every time the array is a little bit longer.

A way around this issue is to receive the array as a reference. Here is the same example using references and the amount of time that it took to process the same amount of entities.

use EBANX\Stream\Stream;

$result = Stream::rangeInt(0, 100000)
    ->reduce([], function(array &$acc, int $value): array {
        $acc[] = $value;
        return $acc;
    });

real    0m0.136s
user    0m0.085s
sys 0m0.033s

To finish it all up

Stream is a very useful and addictive library. You can get more details of all methods on it by taking a look at our unit tests.

It is under a MIT license so feel free to go wild with it. Cheers.

Flaws of PHP Iterators

Bruno Clemente — Thu, 23 Jul 2020 01:05:12 +0000

First of all, I need to point out that as it normally is when we talk about programming and design, this is my personal opinion. But, I will try to explain my position the best that I can.

So, what is an Iterator? As the wiki page says, an iterator is an object that allows us to traverse a container. Another definition that I found on the internet is that an iterator is an abstraction of a pointer to an element of a sequence.

Imagine that you have a container with some elements. Maybe a PHP array. An iterator will point to an element of that array and also provide ways so you can get a pointer to the next element of that array.

It worth mentioning that iterators can also be used in containers that not really exist in memory. For example, you could create an iterator that each element is the next possible odd number. This iterator pointer to an imaginary container that holds an infinite amount of all possible odd numbers. Iterators are also used for lazy loading where, for example, you iterate over data that is on a server but you only load it at each iteration.

Iterators appear in many languages like JAVA, C++, JavaScript, and Rust. But the journey for iterators in PHP started in PHP 4. That is when the foreach loop was added and you could use it with arrays with this syntax:

$array = [
    'first' => 1,
    'second'=> 2,
]

foreach ($array as $key => $value) {
  ...
}

In the next major version, iterators were added using two interfaces: Iterator and IteratorAggregate. I will not focus on the IteratorAggregate right now because it does not represent an iterator in itself but actually something that we can get an iterator from.

The main focus here will be the Iterator interface. So, let's take a look into its methods:

interface Iterator extends Traversable {
    /* Methods */
    public current ( void ) : mixed
    public key ( void ) : scalar
    public next ( void ) : void
    public rewind ( void ) : void
    public valid ( void ) : bool
}

If you implement this interface, your object can be iterated both by manually calling the next, current, valid methods or by using the same foreach loop that you used for arrays. Inside a foreach, the following methods will be called:

/** First Iteration */
rewind();
valid();
current();
key();

/** Remaining Iterations until valid() returns false */
next();
valid();
current();
key();

And, to demonstrate, here is an implementation of an iterator that read a file line by line:

class FileIterator implements \Iterator {

    private $resource;
    private $read_lines;
    private $current_line;

    public function __construct(string $filename) {
        $this->resource = fopen($filename, 'r');
        $this->read_lines = 0;
        $this->current_line = "";
    }

    public function rewind(): void {
        rewind($this->resource);
        $this->read_lines = 0;
        $this->fetchLine();
    }

    public function next() {
        $this->fetchLine();
        $this->read_lines++;
    }

    public function current(): string {
        return $this->current_line;
    }

    public function key(): int {
        return $this->read_lines;
    }

    public function valid(): bool {
        return !feof($this->resource);
    }

    private function fetchLine(): void {
        $this->current_line = rtrim(fgets($this->resource), PHP_EOL);
    }
}

Now that we know what are iterators, where do we use them and how do they work in PHP, I want to point out 3 design flaws of PHP iterators. I will enumerate them from what I think is the least serious, to the biggest flaw.

The key() method

The first design flaw is the key() method. My problem with it is that by being on the interface, everyone that wants to create an iterator needs to implement some type of key for each iteration.

There are some times where this is useful. For example, in an iterator over a database table, we can use the table primary key as the key.

But imagine another iterator that iterates over a list of prime numbers. What would we use as a key? Look at the FileIterator example. There, we are just incrementing a local property as key and this is a very common approach to this problem. Although, we may never actually need to use it.

I imagine that the reason for the key(), is because of the foreach. As I said earlier, the foreach came as a syntax sugar to iterate over an array. Since every element of an array has an index, it makes sense to also add to it a way to easily access the index of each element in the iteration. When the iterator interface was added, the PHP core team decided to also use the foreach with the iterator, and to avoid changing the current syntax, they added a key to the iterator.

The current() and valid() method

The problem with current() and valid() methods is the horrible time coupling. We always need to call valid() to check if the iterator is in a valid state before calling current(). Calling current() without checking if the iterator is in a valid state is undefined behavior.

A slightly better approach would be to get rid of the valid() method and add to the current() method a way to inform that the iteration is over. An even better approach would be to also remove the current() method and change the next() method to return the value. As is done in other languages like JavaScript, Rust, and Python.

Another thing that is kind of annoying about having a valid() and current() method is that is quite common to have to save extra data inside the iterator. We can see this in the FileIterator where we save the data of each line read when next() is called to be returned in the current() method.

The rewind() method

Now, is time for what I think is the worst design flaw, the rewind() method. The rewind resets the iterator so we can iterate over it again. The problem is that not all iterable things can or should be rewindable. Probably the correct approach would have been to have another interface, something like \RewindableIterator, that extends the iterator interface with the rewind method.

You might ask, which iterator is not rewindable? Well, a good example can be found in the \Generator class. As is stated in the PHP manual, generators are forward-only iterators. This means that we cannot rewind it after we start to iterate over it. And what happens if we try to do so? We receive a horrible exception resulting in a very nasty infringement of the Liskov Substitution Principle.

Also, another thing that really grinds my gears is that we always need to call rewind before we start to iterate over an iterator. Calling any method of an iterator without calling rewind first is also undefined behavior. Now, imagine that we have a method that receives an iterator. The problem is if we check if the iterator is valid and it is not, we don't know if it is invalid because the iteration is over or if it is invalid because no one called rewind on it. This gets worse since we can't call a rewind if it is an already started generator as doing so would throw an exception.

Better approaches

Looking at the approach of other languages I personally like two other approaches.

The first one comes from Rust. There, the iterator has only one method: next(). The next method returns an option that can be either Some(Value) or None. When None is returned it means that iteration is over. I played a little bit with this approach in this project, where I basically rewrite the rust iterator using PHP. I also added some adapters to better integrate it with the current PHP approach.

Another solution is the python way. The python iterator also has only the next() method. What differs from Rust, is that an exception is thrown when we call a next in a completed iterator. I personally have my problems with this because we are using an exception to control the flow of the program. But in a controlled way, it might be a good approach.

Summing up

As stated above, I think that the PHP iterator has some major design flaws that make working with it quite annoying. I tried to explain why I think this in this article. Also, I have shown other approaches. Nevertheless, I have to say thanks to the PHP core team because overall, I think that the PHP language as a whole has evolved a lot during the last years.

Playing with Rust to solve data-intensive problem

Bruno Clemente — Tue, 31 Mar 2020 12:33:11 +0000

As programmers, we need to be able to choose the correct tools to solve the problems that we face every day. One of those tools is which programming language should we use. Sometimes a scripting language like Python, PHP or JS will be best. At others, we will need the robustness of a compiled language like Java, C++ or Rust. In this post, I will talk about my adventure in changing the stack to help me solve a problem.

I work as a software engineer at EBANX. Here, most of our backend is written in PHP. Even though there is a lot of criticism about it out there, PHP is a great language that is scaling pretty well to attend our needs. But, the lack of good multithreading can be a negative point when you want to do some very data or CPU intensive tasks.

So, here at EBANX, we have a very critical operation that happens every day and it needs to be finished by noon. Since it is very important for our business, we are constantly trying to make it faster without sacrificing the business rules. One of the steps of this process was taking over 30 minutes, so I decided to play with it to see if I could get it to be faster.

Now, this step is very data-intensive. Imagine that we have a payment(transaction) with its unique identifier, a date, a customer identifier and an amount. We need to check that if we sum the amount from all payments with the same customer identifier in the same month (month/year) it does not surpass a specific threshold. To make matters worse, every day we may have more than a million transactions from a variety of customers and dates that need to be validated against the entire history of the company.

In the past, we tried to solve this by using our relational database. But as the company grew, it started to be too slow. So, we changed to a different approach. We started to write a file every day with all payments that entered during that day operation with a month-year identifier, the payment amount and a customer identifier. So, now we open all files from the company history, we create a compound key using the month-year and customer identifier. Then we add up all amounts from payments that match the keys from our current process and alert it goes over our limit. All of this is done using PHP.

So, my first decision was that I would not change this method. Even though we could think of ways of pre-caching some of those calculations, I wanted to keep the same approach in a way that I could simply swap the existent code with mine. Also, I decided to try writing it using Rust since it is a compiled language focused on performance and safety.

To have a baseline, I started by running the current code on my computer to see how much time it was taking on my machine. I ran it 5 times and it took on average 9.3 minutes. Now, It was way faster than in our production environment and I think it is due to my SSD being faster than whatever storage device is being used in production.

My next step was to rewrite all the current code from PHP to Rust. I wrote it as a standalone program that could be called by our current PHP code and just wait for the result reading the standard output.
I used some cool crates like flate2 that allows you to stream over a GZ compressed file. Also, I wrote integration tests using assert_cmd to call my program and assert that it outputs the expected payments.

Once it was finished, I did the same thing that I did with the original PHP code. I ran it 5 times and got the average execution time. And it was... 9.1 minutes. Only 2.1% faster. Quite a disappointment right? It makes sense since it does not perform a lot of heavy computations, the PHP interpreter was not being a big bottleneck. But I did not give up. My next step was to use some multithreading.

If you never used Rust, it tries to be a fast language like C or C++, but with a lot of guarantees that help you as a programmer to not mess up. It does not make multithreading easy, but it makes it safer.

I was using a HashMap to accumulate my values where the key was a tuple with the first element being the month-year of the payment and the second element the customer identifier. For each key, I was accumulating the total amount. To do the multithreading I could go in two ways, the first would be to create a separate HashMap to each thread and then unify them at the end. This would avoid concurrency problems and having one thread have to wait for another one to stop modifying the data. Another approach would be to use the same hashmap across all threads and handle the clashes. I decided to go with the second option using a DashMap. A Dashmap is a Rust crate that offers a concurrent hashmap implementation that tries to minimize the number of locks. It has some pretty good benchmarks as you can see in the following image taken from the official crate documentation. One factor that made me prefer this approach is that in my use case it should not happen very often that two threads are processing payments from the same customer opened in the same month so clashes should be rare.

Another thing that I needed to think about is since the company has grown a lot, the newer files are way bigger than the old ones. So, a thread might get all small files so it ends fast while another gets stuck with all the big ones and take a lot of time. To try to minimize this, I shuffle all files before dividing them into the threads. I can do this safely since the order of the files does not affect the result.

After it was done it was the time to test it. Going from 2 threads up to 8(The number of virtual cores on my machine), I ran it 5 times and took the average. Bellow is the result of all tests. Now, the average execution time for 8 threads was 2.6 minutes. 72% faster. Yay, real improvement :). I did not test it yet on one of our production machines to see how it will perform there.

My next step will be to rewrite everything to use async I/O. See if this will make it even faster or not. After it, I will try changing the approach and do some caching to try getting it from minutes to seconds.