Weblog | Quickwit, Tantivy, Rust, and extra.

At Quickwit, we’re constructing probably the most cost-efficient search engine for logs and traces. Such an engine usually ingests large quantities of knowledge whereas serving a relatively low variety of search queries. Underneath this workload, most of your CPU is spent on indexing, making our indexing pipeline an important part.

Like most search engines like google like Elasticsearch or Solr, Quickwit builds its inverted index in small batches.
The indexing pipeline’s function is to obtain a steady stream of paperwork, lower it into batches of usually 30s, and carry out a sequence of operations to provide an new piece of the index we name break up.

This part is stuffed with fascinating engineering challenges:

• It ought to be environment friendly. In observe, this implies the entire pipeline ought to be fastidiously streamlined to verify we don’t waste CPU whereas ready for one more useful resource.
• The pipeline ought to be strong and have a transparent and easy method to deal with errors.
• The code must be simple to check, navigate, and preserve.

On this article, we delve into how we ended up adopting an actor mannequin base resolution for our indexing pipeline and talk about among the distinctive options of our implementation. Paul Masurel one in all our co-founders additionally gave a chat on this matter at FOSDEM23.

First, let’s crack open the indexing pipeline field and briefly describe what it does. The next is a abstract of the steps concerned within the indexing pipeline.
Every steps is annotated with the useful resource it consumes.

• Parse and remodel paperwork. (CPU)
• Index JSON paperwork in an in-memory index. (CPU)
• Set off the tip of the break up in line with commit_timeout_secs or split_num_docs_target. This step is what breaks the stream into batches. (CPU, Disk IO)
• Compress and serializes the in-memory break up to the disk. (CPU, Disk IO)
• Stage the break up. (We have to file the break up earlier than beginning the add to have the ability to clear up on failures. (Look ahead to Metastore)
• Add the break up to the goal storage (Native FS, Object storage). (Disk IO or Community)
• Publish the break up to make it out there for search queries. (Look ahead to Metastore)

As you may see, a doc must undergo a bunch of steps earlier than it may be searchable. A naive implementation would run all of those steps sequentially.

let active_split = ingest_enougth_docs_to_make_split();


let serialized_split = save_split(active_split, local_working_directory);


let split_metadata = serialized_split.metadata();
stage_split(&split_metadata, metastore);


upload_split(serialized_split, storage);


publish_split(split_metadata.split_id, metastore);

Nonetheless, we noticed that these steps are working on completely different assets: CPU, disk, community, and typically we’re merely ready for one more service’s response. Merely operating these steps one after the opposite is a waste of assets. As an illustration, our CPU can be idle whereas we’re writing our break up to disk. We have to run these steps concurrently to be able to get probably the most out of the out there computing assets. We will obtain this by dedicating staff to every step and streamlining our pipeline.

Now, if a employee can’t sustain with the incoming duties, it’ll accumulate them in its work queue, and ultimately crash the system. This requires some backpressure mechanism the place we set a most variety of gadgets that may be held in a employee’s backlog. When that most is reached, all staff making an attempt to ship a job to it should wait, thus slowing down and permitting the lagging employee to catch up.

In Rust, bounded mpsc (multiple-producer-single-consumer) channels are a pure resolution for this activity queue. Here’s what such a employee may appear like:

use tokio::sync::mpsc::*;

struct Break up {  }

fn start_worker() -> io::Outcome<Sender<Break up>> {
    
    let (tx, mut rx) = channel::<Break up>(3);
    tokio::activity::spawn(async transfer { 
        for break up in rx.recv().await {
            process_upload(break up).await?;
        }
        io::Outcome::Okay(())
    });
    Okay(tx)
}

It seems this employee setup is what’s conceptually known as an actor.

As framed by Carl Hewitt in 1973, an actor is a computational entity that, in response to a message it receives, can concurrently:

• ship a finite variety of messages to different actors;
• create a finite variety of new actors
• designate the conduct for use for the subsequent message it receives.

Earlier than leaping into the specificity of our actor framework, We should suggest the weblog publish from Alice Ryhl titled Actors with Tokio which reveals the way to use Actors in Tokio rust, as a programming sample and with none framework.

Implementing an actor-based resolution requires an excessive amount of wiring to handle the actors themselves. We don’t wish to combine this with the applying code. A number of of the actor’s administration chores that must be taken care of are:

• Monitor or supervise actors and restart them once they fail.
• Monitor actors’ jobs and schedule retries if essential.
• Handle message scheduling together with time and precedence elements.
• Present testing and observability utilities (ex. mocking time, efficiency metrics).

Actor frameworks might be useful to summary away these elements.

We ended up implementing our personal framework known as quickwit-actors.
However why implement our personal framework when there are a plethora of actor frameworks within the Rust ecosystem (Actix, meio, riker and so on …)? We discovered that the majority out there choices besides Actix weren’t actively maintained.

Our predominant difficulty with Actix was the dearth of message precedence. Our indexer, for example, is answerable for chopping batches each 30 seconds or so. We wish to have the ability to have a message processed as quickly as attainable after the 30 seconds have elapsed. In Actix and in quickwit-actors, that is completed by asking the framework to ship a message to our actor after 30 seconds has elapsed. In Actix, that message is enqueued, and can solely be processed as soon as the entire pending messages have been processed. In Quickwit, this might imply 10 seconds later for this particular actor. As an answer, in our actor framework, every actor has a high-priority and a low-priority queue. Scheduled messages are delivered to the high-priority queue.

Implementing our personal framework additionally made it attainable to bake in some treasured options. Corresponding to measuring and reporting backpressure as metrics, and routinely detecting actors which are blocked on some operation.

From a Rust developer perspective, you outline your actor kind and implement the Actor trait, which solely requires one methodology to be applied. Subsequent, you implement the Handler trait which is generic over the Message kind it ought to deal with;

#[derive(Debug)]
struct HealthCheckMessage;

#[derive(Debug, Default)]
struct HealthActor(usize);

#[async_trait]
impl Actor for HealthActor {
    kind ObservableState = usize;

    fn observable_state(&self) -> Self::ObservableState {
            self.0
    }

    async fn initialize(&mut self, ctx: &ActorContext<Self>) -> Outcome<(),ActorExitStatus> {
        ctx.send_self_message(HealthCheckMessage).await?;
        Okay(())
    }
}

impl Handler<HealthCheckMessage> for HealthActor {
    kind Reply = ();

    async fn deal with(&mut self, health_check_msg: HealthCheckMessage, ctx: &ActorContext<Self>) -> ActorResult<()> {
        
        check_health_status().await?;

        
        self.0 += 1;
        
        ctx.schedule_self_msg(Length::from_secs(2), HealthCheckMessage).await;
        Okay(())
    }
}

#[tokio::main]
async fn predominant() {
    
    let universe = Universe::new();

        
    let health_checker = HealthActor::default();
    let (_health_checker_mailbox, health_checker_handle) = universe
        .spawn_actor(health_checker)
        .spawn();

    health_checker_handle.be a part of().await;
}

The above is an actor that checks the well being standing of an imaginary service each two seconds and maintains a state that counts the variety of checks it has carried out since its inception. This actor sends the HealthCheckMessage to itself.

• First within the initialize methodology of the Actor trait: the strategy that’s initially run when an actor begins.
• Then within the deal with methodology of the Handler trait, however this time with a two seconds delay. This creates an infinite loop of HealthCheckMessage dealing with.

In predominant, we create a Universe for our actors to dwell in. When an actor is spawned, we get again two issues:

• A sort-safe Mailbox that can be utilized to ship messages to the corresponding actor.
• A deal with that can be utilized to observe in addition to wait, stop, kill, pause, and resume the actor.

All Quickwit actors are async by default. To take care of the CPU-intensive actors, we merely schedule them on a special runtime. This trick can also be utilized by InfluxDB as uncovered on this blog post. Usually, RuntimeType::Blocking.get_runtime_handle() is used to request an actor to be scheduled as synchronous actor.

As a result of our actors might be operating in numerous runtimes, we needed to introduce the idea of Universe to isolate teams of actors inside the identical course of. As an illustration, every unit check usually begins by instantiating its personal universe. A universe holds:

• An ActorRegistry that retains observe of actors instantiated on this universe.
• A universe’s KillSwitch to close down the entire actors on this universe.
• A SchedulerClient to deal with and mock time.

One other function of the universe that’s useful when testing is the strategy Universe::with_accelerated_time(). It supplies a universe that accelerates time every time no actor has any message to course of.

Concretely, right here is how we use the actor framework inside the Quickwit indexer service: when JSON paperwork arrive in Quickwit by way of the supply actor, they’re parsed by the doc-processor and buffered in an in-memory break up (indexer actor). When the commit_timeout_secs has expired or the split_num_docs_target is reached, a brand new lively break up is created to maintain receiving ingested paperwork whereas the earlier break up is finalized within the subsequent actors earlier than being marked as searchable by the writer. You may as well distinguish the indexing pipeline (purple group) for indexing paperwork from the merge pipeline (inexperienced group) for merging splits that aren’t matured but.

quickwit-actors has confirmed to be a precious asset for Quickwit. Although it lacks many options and documentation, it supplies us the pliability to mould it into what we’d like.

We lately launched the framework below the MIT Licensed. We do not actually count on it to grow to be a preferred crate, however we hope the MIT License will make it simple for folks to experiment with it and scrap no matter items of code they see match.