Let's write in, we don't have much time.
My name is Paul Mazurel.
I go by the name of Filmi Coton on Mastonon on Twitter.
I've been a rest developer since 2016.
I spent most of my career as a search developer, so it was only natural for me as my first
pet project to develop a library to build search engines.
So if you're familiar with Lucine, that's kind of like Lucine, but for the rest of the world.
That was my first pet project.
It grew.
And two years ago, I co-founded a startup called Quit Quit, that is about building a search engine that is specialized for logs.
Just a word about Quit Quit.
I'm not here to do an advertisement.
It's not a commercial, but it's related to the tool, so I need to explain to you what's the problem, which is the problem that we are trying to solve.
Quit Quit is a Rust project.
It's under the AGPL license.
It's open source.
You can find the source code on GitHub.
And so we specialize on log search.
So what's specific about log search compared to, let's say, an e-commerce search engine is that the data that we get is more or less immutable.
So we assume that people will want to ingest documents into our search engine and won't modify it too much.
So after ingestion, the document stays there until it goes out of its retention period, at which point we will delete it.
Or maybe you might want to delete it if you have a request to comply to GDPR.
We handle that kind of stuff, but you cannot modify it like you would do for any e-commerce website.
And so one of the big differences in terms of efficiency is that when you deal with log search, the volume that you have to deal with has no limits.
The largest amount that we've seen so far is people indexing 100 terabytes a day.
So that's the volume of data.
Imagine if it was actually generated not by machine but by humans.
You would have to have a lot of people typing grief as to deal with that kind of volume.
So that's something that you will only get if you're doing log search.
And compared to any e-commerce website, most of the CPU is actually spent indexing and not searching because you have comparatively way less search and way more documents.
So indexing is actually crucial to our problems, and that's very different from usual search engines.
Indexing, what does it look like?
That's the problem that we are trying to solve.
Super oversimplified.
We get, as an input, a stream of documents.
It's interesting to have another idea for one pipeline of indexing.
We have to deal with around 40 megabytes per second.
And as an output, every 30 seconds, we write a file.
We put it somewhere, and usually we register it on some metadata backends.
And at this point, the file is searchable.
And the rules of the game here is we want to have the highest possible throughput.
And we want to keep what we call time to search as low as possible.
So time to search is at the moment when JSON file is entering the system, we start the clock,
and we measure how long it takes for it to go out of the system in the form of one of those files,
at which point it is searchable. We need that to be as low as possible,
and we need, it's very important to keep it very stable.
We don't want to have, like, a period of time where it goes through the roof.
So that's the whole game.
And in that black box, we do a lot of stuff.
I was voluntarily very simple.
I won't go through all of the stuff that we do,
but the important part here is every single of those steps is using different resources.
The time is spent on different stuff.
So, for instance, when we index things, when we build or in memory index,
we are spending CPU.
When we are writing the file, we use IO.
When we upload with network, and sometimes we are waiting for something that is outside of the system.
We spend no resources at all except we wait.
So you might think that the implementation is obvious.
We have one function for all of those steps, and we call them sequentially.
But if you do that, you are wasting the resources of our system, of course.
For instance, when you are uploading, you are spending your network resource,
but your CPU is not doing anything, so you're wasting money.
So the solution to this problem is relatively simple,
but it's not that simple to implement.
You want to streamline your pipeline.
What I mean by streamlining in a very concrete way is you might have two steps,
like indexing, spending CPU, upload, spending network.
They go sequentially, but what you want is you want indexing to work on building the first file,
and when it has finished, you start uploading, of course.
But as you upload, you want to start working on the second file,
so that you are spending CPU while you are doing your network stuff.
That's what the kind of behavior that we want.
And of course, this example is a little bit too simple
because the second step here is shorter than the first one.
It's a nice case, but if it was the other way around,
we would have to have some kind of mechanism to deal with, to have back pressure.
And in my experience, a lot of very good engineers are not familiar with the concept of back pressure,
so let me explain what it is about.
If you already know what it is, bear with me and enjoy the fine artworks that we have here.
So the idea of back pressure is, imagine you are cleaning dishes with a friend.
One of you is cleaning the plates, and the other one is wiping them dry.
And the person wiping the dishes dry is a little bit too slow compared to the person cleaning the dishes.
What's going to happen is that your plates will accumulate like forever.
And in the computer system, it's a very common problem,
and that's how you get out of memory errors.
So the solution is rather simple.
You need back pressure, which means you need some way to signal the person who is cleaning too fast
that they should slow down.
And the simplest mechanism to do this is you need to have some kind of limit on your stack of plates
or your work queue or whatever you are using in your system.
And then they stop once they reach the limit.
So it's the simplest way you could have back pressure.
So with all this said, the game here is how would you implement that?
What would be your go-to implementation if you had to have such a system in one hour?
And being Rust developers, I think that most of the people in this room will come with the following solution.
So the upload part, it's not CPUV, it's just dealing with networks.
So it's very natural to think, OK, I'm going to do that in a Tokyo task.
And back pressure, that's easy.
I already know what I'm going to use.
I feel that the good solution is I'm going to have a channel with some capacity.
And once it reaches capacity, of course, people sending work to my task, they will have to wait.
It's going to be very nice and natural.
And then on the indexing part, we will have the same mechanism.
And only right now, we will have to actually do a lot of CPU heavy work.
So maybe we won't run that in a Tokyo task and we will spawn our own thread
or maybe we will use a thread pool to do that work.
It would be better, right?
And we use the same mechanism.
We will have a channel to receive the work.
The capacity here is much larger because the type of stuff that we put in the channel is very different.
For the uploader, we are getting files, possibly they can be large, like 100 megabytes.
So it makes sense to have it as small as possible.
Here, it's documents.
So it's for many documents, you will emit one file.
You won't probably have capacity that this is larger than three.
And yeah, everything is fine and handy.
It's quite natural.
So we just reinvented Actors.
That's basically what Actors are.
So Actors is a programming paradigm that has been invented in the 70s by a researcher called Karlewit.
It has been popularized more recently with Erlong.
And the actual formal definition is here.
It's from Karlewit himself.
And I'm going to read it even if it's a little bit weird to read slides out loud.
This one is important.
So an actor is a computational entity that, in response to a message it receives, can concurrently.
A, send a finite number of messages to other Actors.
We've done that.
Two, create finite numbers of new Actors.
We haven't been spawning any Actors in our example yet,
but we do that in quick reads as something that we do especially for supervision or spawning a pipeline or stuff like that.
So we do that.
C, designates behavior to be used for the next message it receives.
That one is a little bit fuzzy.
Do we do this?
No, we definitely don't.
But if you water it down and you squint a little,
the fact that the Actors have actually a state,
and the whole point of having this Actors running on a specific thread is that
it will be possible to handle a message and mutate our state.
And mutating our state is a bit like designating the behavior that it will be able to use for the next message.
So we ended up building our own Actors framework.
So to be honest, I'm not trying to advertise for this framework.
It's under the AGPL license, so you can use it.
You're free to use it.
If you want to take over it for kids and make it better,
I'd be happy for us to use it.
If you want to use it as is and you would like to have it as a MIT license,
I'm perfectly happy to put it under MIT license as well, actually.
But right now, it's not redesigned to be reused by other people.
But I think I might be able to tell you about our journey
and maybe it could inspire other people to write their own framework.
There are actually a lot of Actors frameworks in Rust.
This one is Actix, there are many others.
So under our Actors framework, what does an Actor look like
compared to our original snippets?
Yeah, it looks like this.
So I implemented the uploader there.
So you have to implement first a trait called Actor
where you will define a bunch of small properties about your Actor,
especially the capacity that we have seen before.
So it will be like the capacity of the channel that we described before.
And then you will have to implement a handler trait
for each type of message that you deal with.
So, contrary to our example, now we can deal with several types of messages.
Same Actor can receive different kind of requests, if you will.
Another difference is most of the time you want to communicate with
an Actor in an asynchronous way.
That's how the Actor pattern is working usually.
But sometimes it's handy to actually have some kind of reply
when you do a request.
It's a bummer because if you really want a reply,
that means that you will be waiting for the Actor to process
the entire queue and execute your command and then return the result.
But you don't have to use it.
Most of our messages don't use it, but when we need it, it's handy.
And our indexer, it's about the same.
The one thing that I want to point out is this thing on the Actor trait
where we specialized what should be returned on the method RuntimeHandle.
So, you remember that in our example, we said that an indexer
spends a lot of CPU.
We wanted to run it on a dedicated thread.
We don't do that here.
What we do is we target a specific Tokyo runtime, which is weird.
So, that's an implementation shortcut that we used
instead of running stuff on a thread pool.
What we do is that we have several Tokyo runtime
and we have a Tokyo runtime that is dedicated to act as a thread pool.
The benefit of this is this implementation shortcut
gives us the possibility to write exactly the same code
for an Async Actor or Async Actor, which is neat.
And by the way, we're not the only one to use that trick.
InflexDB actually wrote a blog post about this a little bit of time ago.
Now that we have a framework, you might have noticed
that the code was not even shorter.
We have seen a couple of features that we got from the framework,
but what can we get like more?
Within QuickWit, we have 25 different actors.
What's cool when you have a framework is that you code stuff once
and the benefit is multiplied by 25.
What could be the benefit?
So, hopefully, we get better structure and code
that is a little bit more readable.
People open up files and they know what to expect.
I want to know what is the capacity associated to the cure of this actor.
Where should I look?
That's something that you get from having a framework.
So, we will talk a little bit about that,
but we get supervision from the framework.
We get a neat solution to deal with time,
which is probably the main reason why we don't use Actix today.
And then we have a bunch of recipes to write unique tests,
which is very important.
We also have some stuff to be able to see what our actors are doing
and we have a solution for discoverability also,
but we won't talk about this in this talk.
So, supervision.
Of course, I would like to tell you our code is perfect
and perfect code is, especially for telling the rest, it never fails.
And in a sense, we don't experience panics or stuff like that,
but we have to run our code, third-party code,
user-defined code, like user, they can write a VRI script
to transform their documents in the pipeline
and that's running on our indexing pipeline.
We have to do IO.
We have to interact with different systems.
For instance, we get our documents from a source.
We're running the pipeline. We send them to a storage.
We have different storage that are implemented.
That's a lot of components and any one of them can fail
and we want a very large amount of time.
So, one solution to this, it's not discoverability,
it's not like it works all of the time,
but just try to turn it off and on again.
It feels a little bit stupid,
but you just restart everything from a blank state
or blank slate and sometimes things work fine that way.
So, the supervision works as follows.
We have an actor that is in charge of supervising
all of the actors that are in our pipeline.
What he's doing is that it's pulling actors
and when it detects that they failed,
it will kill everyone and restart everyone.
The definition of failure is a little bit sophisticated
in our case, so it could be an actor that has returned
an error from a handler or it panicked
or we have some system to detect if an actor
has not been progressing for three seconds.
So, we have an option of progression in our framework.
That's an original way to do stuff.
And we use one for all supervision,
which means that if one actor failed,
we restart everyone.
Okay, that was for supervision.
Now, about handling time,
which is probably the most interesting part of our framework.
So, we need to be able to deal with the idea
that, for instance, in our indexer,
we want to emit a file after 30 seconds has passed.
So, we have a condition like this.
30 seconds after the first document was added in that batch.
And we cannot do time.sleep in the handler
because it will block the entire processing of documents.
So, the solution for this is rather simple.
We have a method so that actors can ask the framework
to send back a message to themselves
after a given amount of time, 30 seconds here.
And it seems like a very simple solution,
but it has a problem.
So, here I showed how it worked.
The actor is sending a message to the scheduler.
This is what is happening under the hood.
It's sending a message to actually an actor
that is run by the framework called the scheduler actor.
And 30 seconds later, the scheduler will stack a message
into the queue of the actor.
The trouble there is, imagine that you already had a lot
of messages in the queue of the actor.
Then, where are your 30 seconds?
Maybe we have one minute worth of messages in that queue.
And then, our entire contract of,
I want to emit a file every 30 seconds, it's broken.
We cannot do that.
So, the solution we went for is actually mailbox
are a tiny bit more complicated.
They have two queues.
One is a low priority queue, the usual one.
And we have a high priority queue.
And the scheduler will put that in the high priority queue.
So, as soon as the actor has finished dealing with the message
that it was processing at the time,
it will jump on this scheduled message.
And we will get our nice 30 seconds call back.
Testability, let me check the time to know.
If I'm good or not, 20 minutes, perfect.
Testability, we have a bunch of solutions to write tests.
Let's go through code to see, like,
actual RISD code to see how we can implement
complicated real-life stuff and unit test it.
So, the code that we will look at is a batch builder
that is mimicking pretty much what we do in indexing.
So, we have two possible conditions.
We emit a file either because we have enough data
and it's enough to cut a file.
So, let's say if you have 100 messages,
it's not 100 in relative,
but if you have 100 messages, then you emit a file.
Or, if 30 seconds has elapsed in the reception
of the first document of the batch.
So, let's start slow and easy.
So, we have our actor here.
So, this is the state of the actor,
but this is the actor as well.
So, something that is obvious is it will have to have some
mailbox to push the speed that it produced to.
It will have a document batch which will be a vex of string.
So, document will be just string.
It will append document to that.
When it's big enough, it's now flash it
and send it to the mailbox of the consumer.
And one thing that is new here is we added some counters
and we will be able in the unit test to do some assert
on this internal state.
I didn't talk about it,
but the actor trait actually has an associated type
which is called observable state.
Of course, the whole idea of actors is to encapsulate your
state into your thread or your token task
and you're not supposed to be able to mutate it
or even read it from the external world.
But we have some thing that makes it possible to ask
from outside the actor what is your observable state
and it will send a message to the actor
and the actor will send back the result
of the observable state method here
which is nifty for observability and for unit test.
And then there is our handler.
So, we will have two messages.
One message will be receiving a document here.
It was just a string as I told you.
I wanted to keep stuff simple.
And we will do several things.
The first thing that we do is if this was the first document
in the batch, we need to register our callback message
using the schedule self message.
We will append this document to our batch
and then we check for the conditions.
We have enough documents in the batch
to actually emit a batch using our second batch emission
condition.
And in that case, we call emit batch.
I didn't put the code of emit batch because it was too easy.
Not very interesting.
And then I didn't put the handler
of the time out message, but you can guess it basically
it's emitting the batch.
And then when we want to unit that stuff,
we write things like this.
So, we have a universe in our unit test.
It's a very important thing.
We want to isolate our unit test one from each other.
And the universe is in charge of this.
So, all of the actors of your program
have to belong to the same universe.
Otherwise, they're not supposed to communicate together.
And we will see that this isolation will make it possible
to do something really cool in the next slide.
And so, this universe makes it possible to make a fake mailbox
that we create like the consumer side of things.
We can create our batch builder and it's alone
and send message to it.
That's what we do there.
So, yeah, I usually like to jump and point at the screen,
but I've been told that I cannot cross the wait line.
So, yeah, and then what we do when we want to create an assert
is that we call this function called process pending and observe,
which just means that we wait for all of the messages
that are currently in the queue of the actor.
We have all of those process and then we call observe
and we get a snapshot of what is the observable state of the actor.
And here, the observable state was a counter,
so we check that it's equal to the number of documents that we wanted.
And then we check that the consumer mailbox does contain
our two batches, because 250 is 100 and 102 batches.
And we also want to be able to check the timeout,
because the timeout counter is working well.
So, here, what is interesting is we created our universe,
but with the method with accelerated time,
and we would be marking time at this point.
So, you won't have to wait 30 seconds to have your unit test
to run for 30 seconds.
It will do magic and the result will be exactly the same
as if you were not accelerating time,
but it will just be faster.
And I will explain a little bit how it works.
And so, for the unit test, obviously,
we have to call some way to wait,
and we call universe.sleep to do that.
And it's important to use the universe.sleep and time.sleep
because we are marking stuff, obviously.
We cannot use the marking facilities that we have in Tokyo
because we use several runtimes.
And also, what we do is similar in semantics
as pose and auto-advance if you're familiar with it,
except we never freeze time.
We keep time flowing,
but what we do is tiny bit different.
So, you can imagine that if you were not accelerating time,
your actor execution would look like this.
So, actor are processing stuff,
and sometimes you don't have any message in any queue,
or actor are either,
and the only thing that will resume the processing
is some time out to happen and the scheduler saying,
okay, I have a message for you,
you asked for a self-scheduled message.
It's happening now.
So, our framework detects that we are in a phase
where no one is working and waiting for the scheduler,
and in that case, and only in that case, we accelerate time.
And that's why we get a result that is exactly the same
as if we didn't accelerate time, but just faster.
So, we compress our execution before, after.
That's how it works.
I wanted to show you the actual indexing pipeline
in its full complexity.
I said 25 actors, it's not 25 actors here,
but we have other actors in other parts of the code
because the pattern got quite popular.
It's quite complex, but it makes me extremely good.
It makes me feel good to be able to show that
when we have to explain a new developer
what the indexing is doing.
We can point at things.
Every single one of these box is doing one very simple thing.
It has its own file, it has its own unit test.
It makes me happy to have this very simple figure
that we can discuss around.
One thing that I need to talk to you about is one problem
with Actors is if you have cycles in your Actors network,
you might experience deadlock.
And it's a pretty terrible thing that kind of deadlock
because it can happen at any time in production,
like it can work for one week
and then you experience a deadlock and it's a scary thing.
So there's a sufficient condition to have deadlocks.
If you don't have any cycles, right?
And usually that's the case when you are writing a pipeline.
In the graph before, there was actually a cycle.
We will have a look at it in a second.
There is another, there is a nicer condition to have deadlocks.
It's if the graph of your Actors where you removed all of the queue
where you had an infinite capacity,
if that one is a DAG, then you won't have deadlocks.
And that's what we are doing here.
So the loop, the cycle that we had was due to the fact
that we have like an auxiliary pipeline
that is merging the file together
and there is an arrow over there.
Sorry, I'm going to cross the line.
I did it, I apologize.
If you remove this arrow, then it's a DAG
and that's sufficient condition to never experience deadlock.
It helps me sleep at night.
And yeah, we have a bunch of other features.
The most important one I think I want to tell you about
is that we measure back pressure.
So the framework is automatically measuring back pressure
and expose it as a promoter counter.
That's really neat.
Very useful for us.
And let's use the rest of the time for questions.
So...
APPLAUSE
So are there any questions in the room?
Yes.
The last slide of the previous slide was
you didn't need parallel actors.
What did you need like Fanny and Fanna out
for having any parallel work?
Oh, yes.
So there is something that I didn't read
but Sylvain was very fast
and noticed that we don't have anything
to be able to have several actor work on the same queue
or work concurrently to process stuff faster.
So yeah, strongly, we haven't needed that
strongly enough to actually implement it.
I wrote an implementation and never managed it
because we didn't really need it.
So indexing, we just spawn several pipelines
on the same machine, not too much.
So that part, it's unparalleled.
But yeah, we just never need...
We haven't needed it yet.
I can't really tell where.
Yes, exactly.
So the parallel behavior.
Sorry, you want me to repeat.
So Sylvain was saying,
we use more than one core
just because within the pipeline
we do the streamlining thing.
So we may have different actors
that are consuming the CPU at the same time
but we don't have one actor going,
oh, there's actually five instance of the actor
and we are doing the work five times faster.
So we didn't need that.
Any more questions?
Yes?
Do you have the fairness system
so that one actor doesn't keep
on the processing restriction
other orders in the process?
So no, we don't have that.
So one thing that we have,
actually we don't want...
We have the opposite problem.
We don't want fairness.
So if you look at our pipeline,
the stuff that is taking a lot of CPU
would be the indexer.
Sylvain would take a lot of IO
and we want to give it priority
because it's the thing that we want it to use
as much as CPU as it can.
So we want to give it one core
and we want it to use as much IO as it needs.
And we would like to give it priority.
So the way we do that is that we run it
on a specific runtime
and over there it has its dedicated core.
For IO, the framework is actually not really helping.
So what we do is that we have some IO throttling
that makes it so that the other actors
are not able to actually more write on disk faster
and you can configure that
and there's some corner of the table computation
to compute what you should do.
But yeah, other actors will not be able to write on disk
faster than, let's say, 80 megabytes per second.
And the merge that you have below,
it's okay if it lags a little bit.
That's the part that we want to be low priority
and the part on the top we want to be high priority.
So we don't have any fairness
and we don't want any fairness.
Yes?
So I guess you're supervising that
because otherwise the timeouts may also be delayed.
So the supervisor is running on,
it's very fast, it doesn't do much.
So it's running on a Tokyo runtime
that has a dedicated core
and runs one bazillion actors,
but they're all very light.
So it doesn't matter at all.
Okay, yeah.
Yeah.
Because, I mean, if your actors are very heavy now,
of course, the supervisor will catch you.
Yeah.
At some point, because otherwise your timeouts
will still be delayed.
Yeah, absolutely.
You're absolutely right,
but the thing is it's running on its specific runtime
and it's not CPU-HV,
so there's plenty of core to work with.
Yes?
When you accelerate time in the testing universe,
do you have to specify the steps in time you take?
You mean the number of times?
I assume that when you accelerate time,
basically when nothing is happening,
you take a time step and then see if something would have happened
at that time point.
Does that mean that you have to specify,
we take steps of 100 milliseconds and then test every time
if something would be happening now?
No, it's not.
We don't need to say how many steps we take.
We don't need to say what is the resolution of the steps
that we take.
So the only thing that we do is that the scheduler,
when it detects, it needs to accelerate time.
It has some kind of heap that says,
OK, the next event is actually in 70 milliseconds.
So let's jump 70 milliseconds in the future
and it triggers that event.
And then the execution of the actor
that was supposed to receive this message will go
and if no actor is working anymore,
then we re-accidate them again.
So it's no steps, no resolution or nothing.
Yes?
How about reliability if you want to be sure
that the bot line will make it to the index,
you count them or how do you know they made it through?
So, yeah, it should be the subject of another talk.
Because now that's a super interesting question.
So the pipeline, you want to know,
to have an idea of what kind of semantics,
delivery semantics that you want to have.
We actually offer exactly one semantics
and the way we deal with that is,
so we didn't talk about that,
we have the documents that are coming from a source actor
and the source actors, when we spawn it,
we tell it, OK, you need to stream messages
from these specific checkpoints.
And when we publish stuff like downstream,
we publish stuff by running a transaction
on our metadata store backend
and that transaction updates the checkpoints
of the stuff that we have published
and it publishes the speed as well.
So that when we restart everything,
we can check in the metadata store
what is the last checkpoint and it starts from there.
And if there is an error,
the metadata store will just yell at us and return an error.
It will say, OK, no, something weird has been happening,
maybe we all had two pipelines working at the same time
and your checkpoints are overlapping and you have a problem.
That's the way we work with this problem.
Yes?
At the universe, is that a special crate
or is it in the standard category?
No, the universe thing is something within our framework
and that's what we use to be able to isolate
typically different programs or different unit tests
or different systems.
Yeah, it's within our active framework.
It is.
Do we still have time?
I think we have one more question or one more minute.
I think there was a...
Yes.
I understand on this graph,
the numbers on the rows indicate the capacity of the queue, right?
The capacity of the channel between the actors, right?
The numbers, yes.
Yes.
So we have a lot of tuning points in this system, right?
Yes.
With relation to the back pressure.
So the question is, from your experience,
how sensitive the performance of the system
is to the tuning of back pressure on the channels?
And maybe you have some kind of advice
or a rule of thumb on what to choose
for the best performance.
Yes.
So the question was, on this slide,
all of the little numbers that we have on the arrow
is the capacity of the different queues between actors
that's a lot of parameters to tune.
They probably have an impact on performance.
Is there a cool recipe to...
So the first question was,
how much do they impact performance?
And the second one is,
do we have a nice recipe to be able to tune them maybe automatically?
I'll go first because there is no more time.
So they don't impact performance all that much
as long as you got them a little bit correct.
So you usually need to identify the stuff
that should be at one,
and then you put it at one
and where you want a little bit of capacity.
It should be quite obvious if you know your system.
And I'm sure that there is a nice recipe
to auto-detect that.
I haven't found it.
So if you have ideas, I'd love to...
Usually that kind of question is someone
who is thinking about something.
So please come to me after the talk.
And I'd love to hear your thoughts.
Thank you, everyone. Time is up.
Thank you very much.