如何获得具有异步任务的tokio_postgres::Client的互斥对象的所有权？

Question

我想要创建一个返回tokio_postgres客户机的函数。但是，我无法在异步任务(连接到数据库)中获得变量(来自库tokio_postgres的数据库连接)的所有权。

下面是我的代码(游乐场)：

use std::sync::{Arc, Mutex};
use tokio_postgres::tls::NoTlsStream;
use tokio_postgres::{Client, Connection, Error, NoTls, Socket};

#[tokio::main] // By default, tokio_postgres uses the tokio crate as its runtime.
async fn main() -> Result<(), Error> {
    let pg_client = create_pg_client("postgres://postgres:root@localhost:5432");
    // Use pg_client for the program
    Ok(())
}

//
// Creates a pg client
//
pub async fn create_pg_client(
    config: &str,
) -> Result<Arc<Mutex<(Client, Connection<Socket, NoTlsStream>)>>, Error> {
    // Connect to the database.
    let mut connect_result = Arc::new(Mutex::new(tokio_postgres::connect(config, NoTls).await?));

    let connect_result_thread = connect_result.clone();
    // The connection object performs the actual communication with the database,
    // so spawn it off to run on its own.
    tokio::spawn(async move {
        let mut result = connect_result_thread.lock().unwrap();
        if let Err(e) = (&mut result.1).await {
            eprintln!("An error occured while trying to connect to the database");
        }
    });

    Ok(connect_result)
}

我的代码没有编译：

error: future cannot be sent between threads safely
   --> src\pg_client.rs:18:5
    |
18  |     tokio::spawn(async move {
    |     ^^^^^^^^^^^^ future created by async block is not `Send`
    | 
   ::: src\github.com-1ecc6299db9ec823\tokio-1.5.0\src\task\spawn.rs:129:21
    |
129 |         T: Future + Send + 'static,
    |                     ---- required by this bound in `tokio::spawn`
    |
    = help: within `impl Future`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, (Client, tokio_postgres::Connection<Socket, NoTlsStream>)>`
note: future is not `Send` as this value is used across an await
   --> src\pg_client.rs:20:25
    |
19  |         let mut result = connect_result_thread.lock().unwrap();
    |             ---------- has type `std::sync::MutexGuard<'_, (Client, tokio_postgres::Connection<Socket, NoTlsStream>)>` which is not `Send`
20  |         if let Err(e) = (*result).1.await {
    |                         ^^^^^^^^^^^^^^^^^ await occurs here, with `mut result` maybe used later
...
23  |     });
    |     - `mut result` is later dropped here

它说未来不能安全地在线程之间传送。有可能实现我想要的吗？

使用的箱箱：

tokio = { version = "1.5.0", features = ["full"]}
tokio-postgres = "0.7.2"

Answer 1

标准库Mutex被设计为在单个线程的上下文中保存锁。由于任务可以被不同的线程捕获，所以跨await点的值必须是Send。此外，即使它确实工作，在异步程序中使用阻塞互斥是个坏主意，因为获取互斥可能会花费任意长的时间，在此期间，其他任务不能在同一个执行器线程上运行。

您可以通过切换到东京互斥来解决这两个问题，后者的保护是Send，lock()方法是异步的。例如，它编译：

// Pick up an async aware Mutex
use tokio::sync::Mutex;

let connect_result = Arc::new(Mutex::new(tokio_postgres::connect(config, NoTls).await?));
let connect_result_thread = connect_result.clone();
tokio::spawn(async move {
    let mut result = connect_result_thread.lock().await;
    if let Err(e) = (&mut result.1).await {
        eprintln!("An error occured while trying to connect to the database: {}", e);
    }
});

Ok(connect_result)

游乐场

目前尚不清楚的是这一设计的最终目标。如果连接位于互斥体后面，则可以有效地序列化来自各种任务的对它的访问。这样的连接一开始就不应该在任务之间共享。