Is it possible to launch this worker in a separate thread instead of blocking the main one?

help
1

I have a Rust app which is basically an axum web server with some routes.

But now I need to periodically check if in a database table there are some new rows.

If there are I need to do some HEAVY computations that can last even minutes (especially in an undersized docker container).

The code I'm using is the below and the output is:


The main thread should not be blocked and should print every second: Instant { t: 6479.5889821s }

The main thread should not be blocked and should print every second: Instant { t: 6480.5996495s }

The main thread should not be blocked and should print every second: Instant { t: 6481.6152853s }

Starting an heavy CPU computation...

The main thread should not be blocked and should print every second: Instant { t: 6502.5748215s }

The main thread should not be blocked and should print every second: Instant { t: 6503.5917731s }

The main thread should not be blocked and should print every second: Instant { t: 6504.5990575s }

As you can see the Starting an heavy CPU computation... blocks the main thread.

Is there a way to avoid this?

Should I use tokio::task::spawn_blocking() for each heavy job?

Can I start the entire "worker" in a separate thread? Because I have many different jobs.

I mean this code in main():


let worker = Queue::new();

tokio::spawn(async move { worker.run().await }); // Is there a way to launch this in a separate thread?

The code is here: Rust Playground

Or here: Rust Explorer

2

You could create a second runtime dedicated to the job queue I guess. Tokio doesn't allow you control over the threadpool of the multi threaded runtime such that you can dedicate a whole thread just to your job. Also, I wonder if the same footgun applies to for_each_concurrent as it does to FuturesUnordered.

1 Like
3

spawn_blocking is made exactly for this usecase. Another solution might be inserting yield_now().await spread across your CPU bound work, making sure to hit one of them at least once per second, though this is easier to mess up and end up with blocking code anyway.

1 Like
4

Yeah. This is my second question. I don't know how unfortunately. Can you point me in that direction?

5

Multiple strategies for doing that are discussed in the second half of this blog post.

It needs to be way more frequent than once per second.

1 Like
6

Thanks. I saw it. Except for the third solution "Spawn a dedicated thread" there is no example code.

Can I ask you for help?

7

Ok, let's say I have this code:

#[tokio::main]
async fn main() {
    let worker = Queue::new();

    std::thread::spawn(move || {
        // What should I call here?
        tokio::spawn(async move { worker.run().await });
    });

    loop {
        println!(
            "The main thread should not be blocked and should print every second: {:?}",
            Instant::now()
        );
        tokio::time::sleep(Duration::from_secs(1)).await;
    }
}

If I runs this I get this panic:

there is no reactor running, must be called from the context of a Tokio 1.x runtime

8

Spawning a dedicated thread can be done with thread::spawn. You can create a tokio runtime on the new thread with tokio::runtime::Builder::new_current_thread:

use std::thread;

use tokio::runtime;

thread::spawn(move || {
    let rt = runtime::Builder::new_current_thread().build().unwrap();

    rt.block_on(async move {
        // your job worker
    });
});
1 Like
9

Thanks @jofas. Why new_current_thread?

10

Because otherwise you'd spawn a new multi-threaded runtime with its own dedicated threadpool, which is probably not what you want.

1 Like
11

If I run your code @jofas I get this panic:

A Tokio 1.x context was found, but timers are disabled. Call enable_time on the runtime builder to enable timers.

12

Sorry, I didn't test my code. let rt = runtime::Builder::new_current_thread().enable_all().build().unwrap(); should hopefully work.

1 Like
13

Thanks. And what is the difference between block_on() you suggested and spawn()?

14

You can't spawn tasks onto a tokio runtime from somewhere else but other tasks or the future that block_on executes to completion:

pub fn block_on<F: Future>(&self, future: F) -> F::Output

Runs a future to completion on the Tokio runtime. This is the runtime’s entry point.

This runs the given future on the current thread, blocking until it is complete, and yielding its resolved result. Any tasks or timers which the future spawns internally will be executed on the runtime.

1 Like
15

This topic was automatically closed 90 days after the last reply. We invite you to open a new topic if you have further questions or comments.