Using futures::executor::block_on inside Tokio

[fdubois1]

Version
└── tokio v0.2.21
└── tokio-macros v0.2.5

Platform
Linux COMPUTERNAME 5.3.0-62-generic #56~18.04.1-Ubuntu SMP Wed Jun 24 16:17:03 UTC 2020 x86_64 x86_64 x86_64 GNU/Linux

Description
In my application, I had all tokio threads blocked so my application was not responding.

[code sample that causes the bug]

use std::time::Duration;

async fn update() {
    tokio::time::delay_for(Duration::from_millis(10)).await;
}

#[tokio::main]
async fn main() {
    futures::executor::block_on(async {
        for i in 0..100 {
            dbg!(i);
            update().await;
        }
    });
}

I expected to see this happen: [explanation]
Normally, I should see a print for 0 to 99 and the application should stop right away. But it prints only i = 0 to i = 64 and after, the application is locked. Nothing happen.

If I remove the futures::executor::block_on() block and I just keep the loop, it works well. But as soon as I add a block_on block, if lot of operation occur, it blocks somewhere.

[davidbarsky]

I see the issue. You shouldn’t be using futures::executor alongside with Tokio as neither executor (Tokio or futures) are aware of one another, which explains the blocking behavior you’re seeing.

I’ll also note that #[tokio::main] expands into tokio::Runtime::block_on, which means this code has two nested block_ons.

[fdubois1]

Thank you for your quick reply. I understand your point. But what I can do to block on a async function call in a such case ? Of course, in the example, it doesn't make sense to add a block_on() block there. But in my real case, I have a callback defined like this :
Fn(u32, &[u8]) -> (Vec, Option)

And into the callback, I want to call an async function. The callback is called from a thread running within a runtime. Even if I use tokio::runtime::block_on, it doesn't work because you can't start a runtime from within a runtime. Do I have any option there ?

Of course, I could change my callback to look like Fn(u32, &[u8]) -> BoxFuture<(Vec, Option)> and remove the block_on() block, but it is not as simple as that since the callback is from a future 1 that is launch on the tokio runtime with a call to compat(). So I though that I could block on the async call as a quick fix. Any idea how I could block in a such case ?

Thanks

[davidbarsky]

I'm not entirely sure what you mean by "block on an async call", as I can see two possible interpretations for that. However, I think you're looking for the second usecase:

• You want to top-level (e.g., fn main()-level) block_on, which runs the Tokio runtime until the application is signaled to shut down. This is what #[tokio::main] does.
• You want to make a blocking call within an asynchronous context on something that is CPU-bound. In that case, I'd suggest using Tokio's task::spawn_blocking, which moves the blocking call into a dedicated threadpool. Your would therefore look like this:

use std::time::Duration;
use tokio::{task, time};

fn callback(i: u32, byte: &[u8]) -> (Vec<i32>, Option) {
   // skipping details...
}

#[tokio::main]
async fn main() {
    task::spawn_blocking(|| {
        for i in 0..100 {
            let bytes = vec![];
            callback(i, bytes)
        }
    });
}

Did I get that right?

[fdubois1]

Not exactly. I want to call an async function in my callback. And that function is not async so I can't do a simple .await on my call.

Here is a simple example of what I have. I spawn a future version 1 on the tokio runtime with the compat() call. And into that future, in the poll function, I call my callback that is not async. But in my callback, I want to call an async function. Is it possible ?

use futures::compat::Future01CompatExt;
use std::time::Duration;
use tokio::runtime::Handle;

#[tokio::main]
async fn main() {

    let my_future_01 = MyFuture{};
    tokio::spawn(my_future_01.compat());
    println!("End of program");
}

struct MyFuture{}

impl futures_01::future::Future for MyFuture {
    type Item = ();
    type Error = ();

    fn poll(&mut self) -> std::result::Result<futures_01::Async<<Self as futures_01::Future>::Item>, <Self as futures_01::Future>::Error> {
        callback();
        Ok(futures_01::Async::Ready(()))
    }
}

fn callback() {
    // I want to call some async function here. Since my function is not async, I want to block on those call.
    // Calling handle.block_on doesn't work of course. But how can I do something similar ?
    // I was using futures::executor::block_on, but when it is called often, doesn't work.
    let handle = Handle::current();
    handle.block_on( async {
        // I want to call some async function here
        tokio::time::delay_for(Duration::from_millis(10)).await
    });
}

And even if would change my callback function to be async, let say :

async fn callback() {...}

I can't call .await in my poll() function.

I hope it is clear. Of course, I would like to remove all those future 1, but I have to understand how to do it and learn more how to migrate from 1 to 3.

Thanks

[Darksonn]

In this specific case, you can do this:

use futures::compat::{Future01CompatExt, Compat};
use futures::future::BoxFuture;
use std::time::Duration;

#[tokio::main]
async fn main() {

    let my_future_01 = MyFuture {
        callback_future: None,
    };
    let join = tokio::spawn(my_future_01.compat());
    println!("End of program");

    join.await.expect("Panic in spawned task.").unwrap();
}

struct MyFuture {
    callback_future: Option<Compat<BoxFuture<'static, Result<(), ()>>>>,
}

impl futures_01::future::Future for MyFuture {
    type Item = ();
    type Error = ();

    fn poll(&mut self) -> Result<futures_01::Async<<Self as futures_01::Future>::Item>, <Self as futures_01::Future>::Error> {
        match &mut self.callback_future {
            Some(future) => future.poll(),
            None => {
                self.callback_future = Some(Compat::new(Box::pin(callback())));
                self.poll()
            }
        }
    }
}

async fn callback() -> Result<(), ()> {
    tokio::time::delay_for(Duration::from_millis(10)).await;
    println!("In callback");
    Ok(())
}

[fdubois1]

Thank you very much. I will try to integrate that in my code. It is not as simple as the example, but I understand how it is possible. But soon or later, I would like to get rid of future version 1 and avoid that kind of issue. Thanks a lot for your time.

[Darksonn]

Apparently I cannot mark the comment above as answer, so I'll mark this one instead.