async --> sync --> async: why is tokio freezing up?

[ibeckermayer]

I have a non-standard use case where I've managed to get tokio to break, however I've been unable to understand why. At a high level I'm writing an asynchronous program, which calls into a mostly-synchronous library, but that mostly-synchronous library needs to use async for a small subsection.

async    -->   sync --> async
|_program_|____library____|

Below is a full toy example that illustrates what I'm running into:

use std::net::ToSocketAddrs;
use std::time::Duration;
use tokio::net::TcpStream as TokioTcpStream;
use tokio::sync::oneshot;
use tokio::time::{sleep, timeout};

// main is analogous to the program I'm writing.
fn main() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let (tx, rx) = oneshot::channel();
    rt.spawn(async {
        let server_socket_addr = "google.com:80".to_socket_addrs().unwrap().next().unwrap();
        TokioTcpStream::connect(&server_socket_addr).await.unwrap();

        run_execute_future();

        tx.send(()).unwrap();
    });
    rx.blocking_recv().unwrap();
}

// run_execute_future is analogous to the synchronous function I'm calling from the library.
fn run_execute_future() {
    async fn sleep_for_a_bit() {
        sleep(Duration::from_millis(100)).await;
        println!("slept for a bit");
    }

    execute_future(sleep_for_a_bit());
}

// execute_future is exactly what the library uses to call the asynchronous function within run_execute_future.
fn execute_future<Fut>(fut: Fut) -> Fut::Output
where
    Fut: std::future::IntoFuture + Send,
    Fut::Output: Send,
{
    use std::thread;
    use tokio::runtime::{Builder, Handle};

    match Handle::try_current() {
        Ok(handle) => {
            // Tokio runtime already exists, cannot block again on the same thread.
            // Spawn a new thread to run the blocking code.
            thread::scope(|s| {
                s.spawn(move || handle.block_on(fut.into_future()))
                    .join()
                    .unwrap()
            })
        }
        Err(err) => {
            if err.is_missing_context() {
                // No existing tokio runtime context, block on a new one.
                let rt = Builder::new_current_thread().enable_all().build().unwrap();
                return rt.block_on(fut.into_future());
            }

            // ThreadLocalDestroyed error should never happen.
            panic!(
                "Unexpected error when trying to get current runtime: {}",
                err
            );
        }
    }
}

When the program is run as given, it hangs indefinitely, and from some print debugging I can see that it hangs indefinitely on the first call to await (in sleep_for_a_bit in the example). I can fix the program (get it to print "slept for a bit") by

1. Changing rt.spawn to rt.block_on.
2. Commenting out TokioTcpStream::connect(&server_socket_addr).await.unwrap();
3. Rewriting execute_future (omitted, because it's not important to my question).

This behavior baffles me. I understand that by blocking in execute_future via the call to thread::scope I'm doing something inadvisable (blocking an async program), but I don't understand why it's breaking the program entirely. Shouldn't I still be able to await a future? And why does block_on fix that?

Even more baffling is that commenting out TokioTcpStream::connect(&server_socket_addr).await.unwrap(); fixes the program. What does a random tcp connection have to do with tokio's sleep function (it's not just sleep, apparently any await call will do the trick).It seems as if there must be something deep within tokio's internals that I'm bumping into, and I'd like to understand more if anybody has the knowledge to help.

[Darksonn]

It's because all IO and timers depend on the runtime's IO driver. Since the IO driver is operated on Tokio's threads, it is one of the things you are blocking when you block the thread.

You can try using block_in_place instead of the thread::scope call, but it doesn't work in all cases, for example, it requires the multi-threaded runtime.

[ibeckermayer]

You can try using block_in_place instead of the thread::scope call, but it doesn't work in all cases, for example, it requires the multi-threaded runtime.

@Darksonn thanks for that suggestion, that's what we eventually decided to go with a la:

Code

fn execute_future<Fut>(fut: Fut) -> Fut::Output
where
    Fut: std::future::IntoFuture + Send,
    Fut::Output: Send,
{
    use std::thread;
    use tokio::runtime::{Builder, Handle, Runtime, RuntimeFlavor::MultiThread};
    use tokio::task;

    fn new_runtime() -> Runtime {
        Builder::new_current_thread().enable_all().build().unwrap()
    }

    match Handle::try_current() {
        Ok(handle) => {
            if handle.runtime_flavor() == MultiThread {
                task::block_in_place(move || handle.block_on(fut.into_future()))
            } else {
                thread::scope(|s| {
                    s.spawn(move || new_runtime().block_on(fut.into_future()))
                        .join()
                        .unwrap()
                })
            }
        }
        Err(_) => new_runtime().block_on(fut.into_future()),
    }
}

Regarding the original code, I'm still curious if there's some operating principle of Tokio that I'm missing or misunderstanding that explains the observed behavior. At a high level your explanation makes sense -- Tokio io/timers depend on the IO driver, the IO driver is operated Tokio's threads which we block, ergo the whole program locks up.

However some of the details still aren't quite clear for me: To narrow it down, when my main looks like this, the program fails.

fn main() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let (tx, rx) = oneshot::channel();
    rt.spawn(async {
        let server_socket_addr = "google.com:80".to_socket_addrs().unwrap().next().unwrap();
        TokioTcpStream::connect(&server_socket_addr).await.unwrap();

        run_execute_future();

        tx.send(()).unwrap();
    });
    rx.blocking_recv().unwrap();
}

however when I change rt.spawn to rt.block_on, the program runs to completion. In either case, I'm using the original execute_future function which blocks the thread

Code

// execute_future is exactly what the library uses to call the asynchronous function within run_execute_future.
fn execute_future<Fut>(fut: Fut) -> Fut::Output
where
    Fut: std::future::IntoFuture + Send,
    Fut::Output: Send,
{
    use std::thread;
    use tokio::runtime::{Builder, Handle};

    match Handle::try_current() {
        Ok(handle) => {
            // Tokio runtime already exists, cannot block again on the same thread.
            // Spawn a new thread to run the blocking code.
            thread::scope(|s| {
                s.spawn(move || handle.block_on(fut.into_future()))
                    .join()
                    .unwrap()
            })
        }
        Err(err) => {
            if err.is_missing_context() {
                // No existing tokio runtime context, block on a new one.
                let rt = Builder::new_current_thread().enable_all().build().unwrap();
                return rt.block_on(fut.into_future());
            }

            // ThreadLocalDestroyed error should never happen.
            panic!(
                "Unexpected error when trying to get current runtime: {}",
                err
            );
        }
    }
}

Ergo I suppose I can infer that Tokio's I/O driver runs on "worker" threads, hence when I block on the "worker" thread spawned by rt.spawn, the program locks up. On the other hand, rt.block_on creates the future on a separate thread pool, so blocking that thread doesn't prevent the I/O driver from advancing. Is that interpretation correct?

[hawkw]

The current_thread runtime does not execute spawned futures unless one or more threads are calling block_on. In your first example, the task you spawn using Runtime::spawn never actually executes, since nothing is calling Runtime::block_on or Handle::block_on. The first block_on call occurs inside of your execute_future function, but in your example code, that is never actually executed, since nothing actually calls block_on.

[ibeckermayer]

Makes sense, although in my first example I'm using the default runtime let rt = tokio::runtime::Runtime::new().unwrap(); which gives me the multi threaded runtime.