Escaping from sync to async world with tokio, and vice-versa

, , , ,

Motivation

Writing rust is a great pleasure when the code is either purely synchronous or asynchronous. However, the awkward thing kicks in if you try to jump between the two worlds. Although there is a standalone document page explaining how to bridge the code, we could always find people having a hard time doing it everywhere, including tokio issues, rust user forum and stackoverflow. This blog post tries to serve as a cheatsheet to glue the code between the two worlds.

Async -> Sync

Usually, this direction is not an issue.

Find tokio wrappers

Except for runtime implementation, tokio also has a large number of wrappers of common sync operations like reading a file or network socket. If there is one, it is usually the best choice.

Use tokio::task::spawn_blocking

If the blocking operation doesn’t have a wrapper, we can simply do it with tokio::task::spawn_blocking as suggested from the official docs.

Use tokio::task::block_in_place

spawn_blocking sometimes is not ideal because it requires the future to be 'static, i.e. you can’t have any borrows. An alternative would be tokio::task::block_in_place, which internally spawns another thread to replace the current thread to avoid blocking the whole runtime.

Caveat: Don’t “Just Block”

In any case, blocking should be avoided in worker threads. In other words, long blocking without yielding to the underlying runtime by any await can lead to various issues, like deadlocks and missing signals, etc.

Sync -> Async

This direction is much more common (imagine you want to call async functions in a trait) but easier to mess up as well. Using the code below as an example, we illustrate different solutions.

1
2
3
4
5
6
7
8
9
10
async fn to_bridge() {
    for _ in 0..3 {
        tokio::time::sleep(Duration::from_secs(1)).await;
        println!("slept");
    }
}

fn block() {
    // How to call to_bridge here?
}

If the calling thread is managed by tokio

If the main entry is annotated by tokio::main or current thread is managed by tokio, a perfectly safe approach will be wrapping block in a spawn_blocking. You can also try this in the playground.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
async fn to_bridge() {
    for _ in 0..3 {
        tokio::time::sleep(Duration::from_secs(1)).await;
        println!("slept");
    }
}

fn block() {
    Handle::current().block_on(to_bridge());
}

async fn run() {
    let hdl = tokio::task::spawn_blocking(|| {
        block()
    });

    hdl.await.unwrap();
}

#[tokio::main]
async fn main() {
    run().await
}

But this approach is not perfect, sometimes things could be complex, for example, the call stack is deep and you can’t ensure the block function is indeed wrapped with spawn_blocking. In this case, the program above without spawn_blocking will panic with (see playground):

Cannot start a runtime from within a runtime. This happens because a function (like block_on) attempted to block the current thread while the thread is being used to drive asynchronous tasks.

Thankfully, block_in_place again comes to the rescue. We can do (see playground):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
async fn to_bridge() {
    for _ in 0..3 {
        tokio::time::sleep(Duration::from_secs(1)).await;
        println!("slept");
    }
}

fn block() {
    let handle = Handle::current();
    tokio::task::block_in_place(move || {
        handle.block_on(to_bridge());
    });
}

async fn run() {
    block()
}

#[tokio::main]
async fn main() {
    run().await
}

Note block_in_place can also block if the block thread reaches the limit.

This trick is also widely used to call async functions when in Drop.

If the calling thread is not managed by tokio

Sometimes, the calling code is the pure synchronous code and you even don’t find tokio in their dependency. In this case, the Handle::current() in previous approach will panic because there is no runtime. In this case, we can spin up a current_thread runtime. See it in playground.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
use std::time::Duration;
use tokio::runtime::Builder;

async fn to_bridge() {
    for _ in 0..3 {
        tokio::time::sleep(Duration::from_secs(1)).await;
        println!("slept");
    }
}

fn block() {
    Builder::new_current_thread().enable_all().build().unwrap().block_on(to_bridge())
}

fn main() {
    block();
}

If necessary, the current_thread runtime can be created once and saved for later use, though creating one is actually very lightweight.

All-in-one takeaway?

Sure, here is the takeaway:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
fn escape_to_async<F, O>(fut: F) -> O
where
    F: Future<Output = O>
{
    match Handle::try_current() {
        Ok(handle) => {
            tokio::task::block_in_place(move || {
                handle.block_on(fut)
            })
        },
        Err(_) => {
            Builder::new_current_thread().enable_all().build().unwrap().block_on(fut)
        }
    }
}

You can even play it with:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
async fn jump_sync(level: usize) {
    println!("in async...");

    jump_async(level + 1);
} 

fn jump_async(level: usize) {
    println!("in block...");
    escape_to_async(async move {
        if level < 256 {
            jump_sync(level + 1).await
        }
    })
}

Note if you would like escape_to_async to be compatible with current_thread runtime, O and F must be Send because tokio currently doesn’t allow current_thread to block_in_place. That is:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
fn escape_to_async<F, O>(fut: F) -> O
where
    F: Future<Output = O> + Send,
    O: Send
{
    match Handle::try_current() {
        Ok(handle) => {
            match handle.runtime_flavor() {
                RuntimeFlavor::CurrentThread => {
                    std::thread::scope(move |t| {
                        t.spawn(move || {
                            Builder::new_current_thread().enable_all().build().unwrap().block_on(fut)
                        }).join().unwrap()
                    })
                },
                _ => {
                    tokio::task::block_in_place(move || {
                        handle.block_on(fut)
                    })
                }
            }
            
        },
        Err(_) => {
            Builder::new_current_thread().enable_all().build().unwrap().block_on(fut)
        }
    }
}

It somehow seems not perfect but I failed to figure it out in the scope of tokio. I’m happy to hear about a better solution!

Caveat: Don’t mix with other executors

It can be tempered to use other executors like futures::executor::block_on to escape from sync to async world. However, it is generally a bad idea, which is known to cause deadlocks.

Conclusion

Generally speaking, because async inexplicitly requires an executor (or runtime) while sync code doesn’t, users should be very cautious when transiting from sync to async, while the other direction is much more straightforward.

Reference