I have no idea to which category this post belongs to – to C++ or to
KDE development – because it is about one component in Plasma Blade. But
the post is mostly about the approach to writing a distributed system I
went for while implementing it, and a C++ framework that will come out
of it.
Ranges
One of the most powerful parts of the C++ standard library is the
<algorithm> header. It provides quite a few useful
algorithms for processing collections. The main problem is that
algorithms take iterator pairs to denote the start and end of a
collection which should be processed instead of taking the collection
directly.
While this has some useful applications, most of the time it just
makes algorithms more difficult to use and difficult to compose.
There were multiple attempts at fixing this by introducing an
abstraction over sequence collections called ranges. There are
several 3rd party libraries that implement ranges, and one of them (Eric
Niebler’s range-v3 library) is on its way to become a part of the C++
standard library.
There are numerous articles written about ranges available online
(and I even dedicated a whole chapter to them in my book), so I’m not
going to cover them in more detail here. I’m just going to say that
ranges allow us to easily create sequences of transformations that
should be applied to a collection.
Imagine the following scenario – we have the output of the
ping command, and we want to convert the whole output to
uppercase, then extract the number of miliseconds each response took,
and then filter out all the responses which took longer than some
predefined time limit.
64 bytes from localhost (::1): icmp_seq=1 ttl=64 time=0.015 ms
64 bytes from localhost (::1): icmp_seq=2 ttl=64 time=0.041 ms
64 bytes from localhost (::1): icmp_seq=3 ttl=64 time=0.041 ms
In order to demonstrate the range transformation chaining, we are
going to make this a bit more complicated than it needs to be:
- we will convert each line to uppercase;
- find the last
= character in each line;
- chop off everything before the
= sign (including the
sign);
- keep only results that are less than
0.045.
Written in the pipe notation supported by the range-v3
library, it would look something like this:
auto results =
ping_output
| transform([] (std::string&& value) {
std::transform(value.begin(), value.end(), value.begin(), toupper);
return value;
})
| transform([] (std::string&& value) {
const auto pos = value.find_last_of('=');
return std::make_pair(std::move(value), pos);
})
| transform([] (std::pair<std::string, size_t>&& pair) {
auto [ value, pos ] = pair;
return pos == std::string::npos
? std::move(value)
: std::string(value.cbegin() + pos + 1, value.cend());
})
| filter([] (const std::string& value) {
return value < "0.045"s;
});
Abstractions
In one of my previous
posts, I wrote that using the for_each algorithm
instead of the range-based for loop provides us with an abstraction high
enough to be used for processing asynchronous data streams instead of
being able to work only with regular data collections.
Some people argued that for_each was not meant to be a
customization point in C++, which might be true, but it works very well
as one. Now, it suffers from the same problems as other STL algorithms,
and if we want to reach new heights of abstraction, it truly is not the
best customization point out there. So, we might find something else to
customize instead.
If you look at the previous code snippet, you’ll see several
transformations performed on ping_output. But it does not
say what ping_output is. It can be a vector of strings, it
can be an input stream tokenized on newlines, it can be a
QFuture, etc. It can be anything that
transform and filter can exist for.
Reactive streams
We can create an abstraction similar to ranges, but instead of trying
to create an abstraction over collections, we will create abstractions
over series of events.
Imagine if ping_output was a range that reads the data
from an input stream like std::cin. Whenever we request a
result from the results range, it would block the execution
of our whole program until a whole line is read from the input
stream.
This is a huge problem. The ping command will send our
program one line each second which means that our program will be
suspended for most of its lifetime waiting for those seconds to
pass,
Instead, if would be better if it could continue working on other
tasks until the ping command sends it the new data to
process.
This is where event processing comes into play. Instead of requesting
a value from the results, and then blocking the execution
until that value appears, we want to react to new values
(events) that are received from the ping command, and
process them when they arrive – without ever blocking our program.
This is what reactive streams are meant to model – an asynchronous
stream of values (events). If we continue with the ping
example, the transformations it defines should also be able to work on
reactive streams. When a new value arrives, it goes through the first
transformation which converts it to uppercase, and sends it to the
second transformation. The second transformation processes the value and
sends it to the third. And so on.
The code would look like this:
auto pipeline =
system_cmd("ping"s, "localhost"s)
| transform([] (std::string&& value) {
std::transform(value.begin(), value.end(), value.begin(), toupper);
return value;
})
| transform([] (std::string&& value) {
const auto pos = value.find_last_of('=');
return std::make_pair(std::move(value), pos);
})
| transform([] (std::pair&& pair) {
auto [ value, pos ] = pair;
return pos == std::string::npos
? std::move(value)
: std::string(value.cbegin() + pos + 1, value.cend());
})
| filter([] (const std::string& value) {
return value < "0.045"s;
});
The only thing that changed is the source of the values. Instead of
using a range (a vector, or another sequence collection), this uses a
reactive stream which emits a line every time the ping
command outputs it.
This is the power of abstraction – using the code we have written for
one thing, for something completely different. In this case, using the
code that was written to process a collection synchronously, to process
asynchronous event streams.
Distributed stream
processing
While it is nice to be able to write asynchronous software systems in
the same way we write synchronous systems, that is not enough for this
post.
One of the great things about defining programs as series of pure
transformations to be performed on the data is that those
transformations are independent from one another – they can be isolated
from each other and even moved into different processes (even processes
on separate computers in a network).
For this case, I’ve implemented in the Voy library a special type of
transformation called a bridge which is used to transparently
send the data from one process to the other.
Let’s split the data processing in ping example into
three parts (similar to what the Plasma Blade will need) – to execute
the first and the last part in the main program, and to execute the
middle part in the backend. It will look like this (added the
voy namespace for completeness):
auto pipeline =
voy::system_cmd("ping"s, "localhost"s)
| voy::transform([] (std::string&& value) {
std::transform(value.begin(), value.end(),
value.begin(), toupper);
return value;
})
| voy_bridge(to_backend)
| voy::transform([] (std::string&& value) {
const auto pos = value.find_last_of('=');
return std::make_pair(std::move(value), pos);
})
| voy::transform([] (std::pair&& pair) {
auto [ value, pos ] = pair;
return pos == std::string::npos
? std::move(value)
: std::string(value.cbegin() + pos + 1,
value.cend());
})
| voy_bridge(from_backend)
| voy::filter([] (const std::string& value) {
return value < "0.045"s;
});
This data pipeline is defined once for both the main program and the
backend. For the main program, the middle part of the pipeline will be
disabled (no code generated for it), while for the backend only the
middle part will be compiled.
A note on the implementation
At the last Akademy, I talked to Tomaz Canabrava about making KDE
software more appealing to students to join our development efforts.
Now, this project is probably going to be overly complex for an
average student to join in, but I’m trying to make it as readable and as
clean as possible for more adventurous students. It uses all the new and
cool features of C++17, along with void_t and the detection
idiom to simulate concepts, etc.
If you desire to have a chance to work on a real-world C++17 project,
just send me an e-mail, and I’ll try to get you up to speed. The only
pre-requirement is for you to do some investigation and find out which
repository the code resides in. ;)
