Sled: future cannot be sent between threads safely, probably due Send not being implemented in iter methods

[nbari]

First of all, thanks for all your time making and sharing this library, I started to use it and so far have been very happy with it.

I am learning rust and don't know if this is indeed a BUG bug after asking (https://stackoverflow.com/q/63529755/1135424) and researching seems to be an issue in the implementation of the methods from iter (https://docs.rs/sled/0.34.2/sled/struct.Iter.html) missing Send.

For example, I can reproduce this when trying to use the output of db.iter().values() the coded compiles but if running clippy returns:

future cannot be sent between threads safely

snipped of code:

```rust
let mut tasks = FuturesUnordered::new();

let bin_parts = db_parts.iter().values();
for bin_part in bin_parts {
    if let Ok(p) = bin_part {
        let part: Part = from_reader(&p[..])?;
        tasks.push(async { upload_part(s3, key, file, &upload_id, sdb, part).await });

        // limit to N threads (4 for example)
        if tasks.len() == threads {
            while let Some(r) = tasks.next().await {
                if r.is_ok() {
                    pb.inc(1)
                }
            }
        }
    }
}

As a workaround, I am iterating and saving the output into a vector and then using the vector to push the tasks to `FuturesUnordered`

 ```rust
    let mut tasks = FuturesUnordered::new();

    let mut xx: Vec<Part> = Vec::new();
    {
        let bin_parts = db_parts.iter().values();

        for bin_part in bin_parts {
            if let Ok(p) = bin_part {
                let part: Part = from_reader(&p[..])?;
                xx.push(part);
            }
        }
    }

    for x in xx {
        tasks.push(async { upload_part(s3, key, file, &upload_id, sdb, x).await });

        // limit to N threads
        if tasks.len() == threads {
            while let Some(r) = tasks.next().await {
                if r.is_ok() {
                    pb.inc(1)
                }
            }
        }
    }

By doing this I don't get anymore the future cannot be sent between threads safely warning but I need to allocate all the output of the tree into a Vector and iterate over the vector and this is something I would like to prevent.

Any better workaround?

Thanks in advance.

[spacejam]

Hey @nbari :) The issue of Send not being implemented can be kind of tricky to understand. Here's how I understand this error:

1. some async executors will poll a Future on thread T1, and at some point that Future::poll may return Poll::Pending, giving control of the current thread back to the executor before the Future has completed.
2. the async executor (smol, tokio, etc...) may move the Future to another thread, T2 for the next call to Future::poll, as a means of increasing parallelism of the overall system.
3. if the Future was created by the compiler from an async block, the Future may jump from one thread to another any time there is a .await because that's where the sub-Future is polled, and that may return Pending, which could cause the executor to move the Future from T1 to T2 during execution.
4. if the Future jumps from T1 to T2, any items that are in-scope at that point of the async code block need to be capable of being sent from T1 to T2, and implement Send.

So, the issue is that there is some object in-scope during a call to .await which does not implement Send, while using the Future with an async executor that requires the Future implements Send. We can fix this error by changing any of these facts:

1. use an executor that does not require Send, such as smol::Task::local which will cause a task to be executed without a work-stealing executor that will pull it to another thread. This actually increases performance in many cases due to improving cache locality. Always measure realistic workloads on realistic hardware before deciding whether this is a good idea for your workload or not. It's surprising to me how often it's much better than using multiple threads.
2. drop the objects that were in-scope at the point of calling .await. For example:
   let rng = rand::thread_rng(); // uses thread-local storage, NOT Send let gen = rng.gen::<u64>(); my_future.await; // compile error: Future does not implement Send as rng is in-scope
   to
   let gen: u64 = { rand::thread_rng().gen() }; // rng is dropped at the end of the sub-scope above my_future.await; // compile success, nothing !Send in-scope at point of yield
3. Use versions of the !Send object that actually do implement Send. A common source of !Send is having an unlocked Mutex in scope, and the MutexGuard is not Send. However, parking_lot::MutexGuard is Send, and can be used to avoid some issues.

This said, sled only has a single async block (fn) currently, and it is Send: Tree::flush_async. So, I think this is a class of errors that sled has simply avoided by not using async blocks, and the only thing that implements Future is also Send: Subscriber.

I suspect that the Send issue may be due to some object that is in-scope during your call to .await. Note that this could be present in an async block several levels above the specific .await call. If a !Send Future .await's a Send Future, the overall Future is still !Send. It only takes a single !Send child Future to make the overall Future !Send, so it's viral.

Feel free to send me a dm on discord if you'd like me to take a closer look at your code where this issue is happening! I may also be wrong, due to my understanding being incomplete, but this is my current expectation.

[nbari]

Hi @spacejam thanks for the reply, I will try to debug deeper also by removing other objects and let you know.

[nbari]

I tried to test removing all possible objects indeed only sending an usize:

 let mut tasks = FuturesUnordered::new();

 let bin_parts = db_parts.iter().values();
 for bin_part in bin_parts {
     if let Ok(p) = bin_part {
         let _part: Part = from_reader(&p[..])?;
         tasks.push(async { upload_part(1).await });
     }
 }

or also with move and and inner scope:

let tasks = FuturesUnordered::new();

 let bin_parts = db_parts.iter().values();
 for bin_part in bin_parts {
     {
         if let Ok(p) = bin_part {
             let _part: Part = from_reader(&p[..])?;
            let uid = upload_id.clone();
            tasks.push(async move { upload_part(s3, key, file, uid, sdb, part).await });
         }
     }
 }

I am testing with this code

[spacejam]

the fix for this is released in 0.34.4