Odd Send + async error

1

I'm getting an odd error I can't explain. Rust seems to have a problem when the following code contains the variable completely_unrelated, but not when the variable is gone, despite the variable not being used. I believe the problem is that Workbook is not Send, but I don't understand why this is a problem or how to fix it. Any help is appreciated, thank you.

I've reduced this down to the most minimal example I can.

Exact error:

future cannot be sent between threads safely
the trait `std::marker::Send` is not implemented for `dyn for<'a> Fn(&'a Workbook) -> serde_json::Value`, which is required by `{async block@src/bin/rocket_server.rs:122:43: 131:6}: std::marker::Send`
required for the cast from `Pin<Box<{async block@src/bin/rocket_server.rs:122:43: 131:6}>>` to `Pin<Box<dyn std::future::Future<Output = Result<(), rocket_ws::result::Error>> + std::marker::Send>>`rustcClick for full compiler diagnostic
rocket_server.rs(114, 42): future is not `Send` as this value is used across an await

Code

pub async fn test<F, Fut>(send_message: F) -> Result<(), String>
where
F: Fn((bool, HashMap<u64, serde_json::Value>)) -> Fut,
Fut: std::future::Future<Output = Result<(), String>>
{
    let completely_unrelated: HashMap<usize, Box<dyn Fn(&Workbook) -> serde_json::Value>> = HashMap::new();
    // Suppose you insert and do some computation on completely_unrelated, then drop it
    send_message((true, HashMap::new())).await.map_err(|e| e.to_string())?;
    drop(completely_unrelated);
    Ok(())
}

// <channel>&
#[get("/workbook/<owner>/<id>?<username>&<auth>")]
async fn workbook_ws<'a>(ws: ws::WebSocket, owner: &'a str, id: &'a str, /* channel: &'a str, */ username: &'a str, auth: &'a str, state: &'a State<Arc<WsState>>) ->  ws::Channel<'a> {
    ws.channel(move |mut stream| Box::pin(async move {                
        let send_response_to_users = |(should_save_workbook, conn_id_to_output): (bool, HashMap<u64, serde_json::Value>)| {
            async move {
                Ok(())
            }
        };

        let handle = test(send_response_to_users).await;
        Ok(())
    }))
}
2

What's the definition of types used? I guess I can minimize this further, but not sure if my guesses are correct, and the current latest version of ws::WebSocket, for example, doesn't seem to have the channel method.

3

The only type not specified is &Workbook, and all that matters about it is that it is not Send. You can use any non-Send type.

4

As I said, I can't find what ws::WebSocket is being used here, so can't check workbook_ws, and function test in isolation compiles.

5

The "ws" is "use rocket_ws as ws;"

6

Aha, so it's this channel. Stubbing it (and everything else unrelated), we get this:

use futures::future::BoxFuture;
use std::collections::HashMap;

pub async fn test()
{
    let completely_unrelated: HashMap<usize, Box<dyn Fn()>> = HashMap::new();
    // Suppose you insert and do some computation on completely_unrelated, then drop it
    tokio::task::yield_now().await;
    drop(completely_unrelated);
}

fn ws_channel<F>(_: F) where F: FnOnce() -> BoxFuture<'static, ()> + Send {}

fn workbook_ws() {
    ws_channel(|| Box::pin(async move { test().await }))
}

...which fails to compile with essentially the same error:

error: future cannot be sent between threads safely
  --> src/lib.rs:15:19
   |
15 |     ws_channel(|| Box::pin(async move { test().await }))
   |                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ future created by async block is not `Send`
   |
   = help: the trait `std::marker::Send` is not implemented for `dyn Fn()`, which is required by `{async block@src/lib.rs:15:28: 15:55}: std::marker::Send`
note: future is not `Send` as this value is used across an await
  --> src/lib.rs:8:30
   |
6  |     let completely_unrelated: HashMap<usize, Box<dyn Fn()>> = HashMap::new();
   |         -------------------- has type `HashMap<usize, Box<dyn Fn()>>` which is not `Send`
7  |     // Suppose you insert and do some computation on completely_unrelated, then drop it
8  |     tokio::task::yield_now().await;
   |                              ^^^^^ await occurs here, with `completely_unrelated` maybe used later
   = note: required for the cast from `Pin<Box<{async block@src/lib.rs:15:28: 15:55}>>` to `Pin<Box<dyn futures::Future<Output = ()> + std::marker::Send>>`

That's a lot to digest, to be honest, but the gist is simple:

  • you hold completely_unrelated over the await point;
  • completely_unrelated is expected to hold Box<dyn Fn()>s, which are not Send, therefore it isn't Send itself;
  • therefore, the future returned by test cannot be Send too (since it must store completely_unrelated during await);
  • but ws_channel requires its callback to return something Send, hence the error.

Note that this error is not connected to Workbook in any way - Send closures can very well handle non-Send data, and vice versa.

The simplest way to go is to add the relevant bound:

-let completely_unrelated: HashMap<usize, Box<dyn Fn()>> = HashMap::new();
+let completely_unrelated: HashMap<usize, Box<dyn Fn() + Send>> = HashMap::new();

...but of course, this might become problematic if some of these functions are actually non-Send themselves.

4 Likes
7

This is exactly it. But—how can this work when Workbook is not Send?

8

Why wouldn't it? Workbook is not being sent anywhere, as far as I can see. The closure is called at the same thread where Workbook is placed and doesn't send it anywhere. Yes, the closure itself might have been moved between threads, but it doesn't matter, as long as it ends wherever it needs to be.

1 Like
9

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.