Add support for customized future executor

[gaoqiangz]

Support to customize the runtime associated with spawn to solve the problem of across runtimes.
Scenario:
The connection connected by the runtime A, use by the runtime B, destroy A before B while connection are using will cause IO driver has terminated.

[djc]

What runtime are you using? Why do you want to use bb8 instead of another connection pool that already supports your runtime?

[gaoqiangz]

Please see this example:

• Cargo.toml

bb8 = "0.8.0"
bb8-tiberius = { version = "0.13.0", features = ["sql-browser"] }

• main.rs

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let host = ".\\JQZ"; //local instance `JQZ`
    let conn_str = format!(
        "Server={host};Database=master;TrustServerCertificate=true;IntegratedSecurity=true;"
    );

    let mgr = bb8_tiberius::ConnectionManager::build(conn_str.as_str())?.using_named_connection();
    let pool = bb8::Pool::builder().max_size(1).build_unchecked(mgr);

    let (tx_connected, rx_connected) = tokio::sync::oneshot::channel();
    let thread_1 = std::thread::spawn({
        let pool = pool.clone();
        move || {
            let rt = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .unwrap();
            rt.block_on(async {
                //connect in this runtime
                println!("thread_1 connecting");
                let conn = pool.get().await.unwrap();
                println!("thread_1 connected");
                tx_connected.send(());
                //make `thread_2` block at the `get` call
                tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
                //release `thread_2`
                println!("thread_1 release");
                drop(conn);
                tokio::time::sleep(tokio::time::Duration::from_secs(0)).await;
            });
            println!("thread_1 runtime dropping");
            //cause `thread_2` error
            drop(rt);
            println!("thread_1 runtime droped");
        }
    });

    let thread_2 = std::thread::spawn({
        let pool = pool.clone();
        move || {
            let rt = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .unwrap();
            rt.block_on(async {
                rx_connected.await;
                //use here
                println!("thread_2 getting connection");
                let mut conn = pool.get().await.unwrap();
                println!("thread_2 querying");
                //sleep 10 seconds, should have a error `IO driver has terminated`
                let e = conn.simple_query("WAITFOR DELAY '00:00:10'").await.err();
                println!("thread_2 queried: {e:?}");
            });
        }
    });

    thread_1.join().unwrap();
    println!("thread_1 exited");
    thread_2.join().unwrap();
    println!("thread_2 exited");

    Ok(())
}

• log

thread_1 connecting
thread_1 connected
thread_2 getting connection
thread_1 release
thread_2 querying
thread_1 runtime dropping
thread_2 queried: Some(Io { kind: Other, message: "IO driver has terminated" })
thread_1 runtime droped
thread_1 exited
thread_2 exited

[djc]

Okay, so you want to use a per-thread single-threaded Tokio runtime. Could have just said so...

[gaoqiangz]

Yes, there are multiple runtimes in our program,such as independent timer threads.

[djc]

I won't be able to merge this PR with this stuff in it, please revert these changes.

[djc]

Would prefer to call this spawn().

[djc]

Why does this need a trait object? It would be nice if we didn't have to Box every future just to support this fairly niche use case.