Add spawn_pinned to tokio-util

[AzureMarker]

Motivation

Sometimes when running a task the program has to handle !Sync data. For example, this could be a database connection (Send + !Sync) which will be passed to other async functions by reference when processing a web server request. Because the connection is !Sync and it is captured as an argument to an async function, the resulting future is !Send. Currently the only way tokio can execute !Send futures is by starting a LocalSet and using spawn_local. This does not work very well when the multi-threaded runtime is used. More context is in #2545.

Solution

This PR introduces tokio_util::task::LocalPoolHandle based on the discussion in #2545 and this PR:

#[derive(Clone)]
pub struct LocalPoolHandle { /* ... */ }

impl LocalPoolHandle {
    pub fn new(pool_size: usize) -> LocalPoolHandle { /* ... */ }

    pub fn spawn_pinned<F, Fut>(&self, create_task: F) -> JoinHandle<Fut::Output>
    where
        F: FnOnce() -> Fut,
        F: Send + 'static,
        Fut: Future + 'static,
        Fut::Output: Send + 'static,
    { /* ... */ }
}

The idea for spawn_pinned is to take in an FnOnce which will create the !Send future. The closure will be sent to a thread which runs a LocalSet on the single-threaded executor. The thread is chosen from a small group of worker threads based on load.

[sfackler]

It seems like this is a lot of complexity that could be avoided if your database client was Sync. That might be a more reasonable route?

[AzureMarker]

I would like to avoid that. The specific DB in my use case is SQLite, but there are other Send + !Sync DB connections (see the connection types used by diesel for example). tokio-postgres gets around this by moving the connection to its own future and passing around handlers to it in the form of Client, but that is not a pattern available everywhere. There are also other situations besides DB connections which require !Send futures.

This is a hard to work around limitation of the multithreaded executor, so I think it's important to provide a general solution instead of only fixing my specific case.

[MikailBag]

Global pool approach is something new for Tokio and Tokio-util.

I'd consider instead:

fn new_local_pool(/*some config, e.g. thread count*/) -> Handle;

struct Handle: Clone;

impl Handle {
    fn spawn(&self, fut) -> JoinHandle;
}

This way user can configure local pool (and use multiple pools if they wish).
Additionally you now have simple shutdown logic: when all Handles are dropped, all worker threads stop.

[AzureMarker]

Thanks for the insight and suggestion @MikailBag! I'll rework the code to implement that.

[AzureMarker]

@MikailBag Is there anything blocking review of this PR?

[MikailBag]

IMHO your PR looks good.

Please note that I'm not Tokio maintainer though.

[Darksonn]

I haven't had time to look closely at this PR yet, sorry.

[Noah-Kennedy]

I'll try and make time to go through this later.

[AzureMarker]

@Darksonn I patched up the issues related to task cancellation and task count decrementing. I think it's much more robust now.

[AzureMarker]

@Darksonn I think this PR is pretty close to being merged. Could you take another look?

[Darksonn]

Thanks, it seems good.

[Darksonn]

Sorry about the many delays in reviewing this.

[AzureMarker]

Thanks for reviewing!