Non-static, Non-Send Tokio Spawn Blocking with &self and Scoped Threads

help
1

I have a function like this:

pub async fn work_async(&self) -> Result<(), Error> {

}

I also have a synchronous implementation of this that uses std::thread::scope to spawn threads in a way where I don't have to deal with 'static lifetimes or Send:

pub fn work_threaded(&self) -> Result<(), Error> {
    std::thread::scope(|s| {
        let a = s.spawn(|| self.work_task_a());
        let b = s.spawn(|| self.work_task_b());

        let _a = a.join().unwrap();
        let _b = b.join().unwrap();

        Ok(())
   }).unwrap()
}

My constraints here are:

  1. I cannot make self Send or 'static.
  2. My work tasks are black boxes, they cannot be made into async functions.
  3. My work tasks absolutely block the calling thread
  4. The work done by both tasks should proceed concurrently and not serially, as each of these operations take around 10 seconds to complete and I can't wait 20 seconds.
  5. I want to not block in my async implementation so that other work on the executors can continue.
  6. My work tasks absolutely need &self, but I could make them &mut self if need be.

I'm trying to look into tokio-scoped but I'm still having trouble with lifetimes.

So I essentially have a future which must live as long as &self but which must be able to spawn (scoped probably) threads.

Is there a way for me to use Pin to somehow get around the lifetime issues? I essentially need to go from a future to tokio::task::spawn_blocking into std::thread::scope and express that this future has the same lifetime as &self so as to ensure everything lives long enough without going to 'static or having to make things Send.

Is this at all possible?

2

You could use block_in_place in tokio::task - Rust

3

What you’ve hit here is the scoped task trilemma:

Any sound API can only provide at most two of the following three desirable properties:

  1. Concurrency: Child tasks proceed concurrently with the parent.
  2. Parallelizability: Child tasks can be made to proceed in parallel with the parent.
  3. Borrowing: Child tasks can borrow data from the parent without synchronization.

You must give up on having one of these properties. Presuming that the most important things are async API and the speedup gained by executing your blocking code in parallel, you must give up on borrowing: this means that you must arrange so that the code managing the tasks (work_threaded() or its caller) must own all of the data it needs to use, but the tasks can still borrow from that manager. Once you have arranged this, you can use tokio’s spawn_blocking() to run work_threaded().

(block_in_place() solves this problem by giving up on concurrency; block_in_place() will cause any code that uses join() or select!() on your future that uses block_in_place() to stop running its other branches until the blocking ends. All it achieves is preventing other Tokio tasks from being blocked by your blocking.)

4

the scoped thread api removes the 'static limitation, but you always need Send for multi-threaded work. you maybe thinking about something else.

the problem with async scoped task spawing is what's called leakpocalypse: you cannot rely on destructors being run for safety invariants.

so what you end is either:

  • the api is safe but the thread must be blocked, like tokio_scoped::scoped(), or:

  • you don't block the caller but the api is unsound, or at least very unsafe, like many other attempts for the problem.

if your tasks are io bound, you may use the "join" combinator to achieve some concurrency (but cpu work is interleved/multiplexed, NOT in parallel).

5

From the docs:

Runs the provided blocking function on the current thread without blocking the executor.
...
Calling this function informs the executor that the currently executing task is about to block the thread, so the executor is able to hand off any other tasks it has to a new worker thread before that happens.

And from your answer:

So in my case I ultimately decided that block_in_place was sufficient for my needs, as the synchronous threaded code that it calls is essentially a flush or shutdown operation that mainly needs to be called at the very end of a program's lifecycle.

I have no futures whatsoever in my call to block_in_place, and it seems okay to me that due to the purpose of these APIs that the likelihood of select! or join! happening on the same worker is low and it is acceptable that these can be starved during the flush/shutdown until that call returns.

(by the way @kpreid thanks for all that you do in the Rust community! you've helped me personally countless times and I'm sure the same is true for many others. thank you!)

6

I think you misunderstand what the constraint for correct use of block_in_place() is. It's not about what you do in block_in_place(); it's about what the caller of your async function does. If the caller uses select!() or join() on your function and another function, they will be blocked.

That is, if the control flow is

tokio:spawn(async {
    join(
        your_function_that_calls_block_in_place(),
        other_function(),
    ).await
})

then other_function()s execution will be blocked. This is why block_in_place() is hazardous to use: it doesn't compose like a Future should, so callers of it have to be restricted to avoid stalls/deadlocks.