rust - 如何通过异步任务获取 tokio_postgres::Client 互斥体的所有权？

Question

我想创建一个返回 tokio_postgres 客户端的函数。但是，我找不到在异步任务(连接到数据库)中获取变量所有权(来自库 tokio_postgres 的数据库连接)的解决方案。

这是我的代码(Playground):

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

它表示 future 无法在线程之间安全地发送。可以实现我想要的吗？

使用的 crate :

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

Answer 1

标准库Mutex旨在在单个线程的上下文中持有锁。由于任务可以由不同的线程拾取，因此跨 await 点的值必须是 Send。此外，即使它确实有效，在异步程序中使用阻塞互斥体也是一个坏主意，因为获取互斥体可能需要任意长的时间，在此期间其他任务无法在同一执行器线程上运行。

您可以通过切换到 tokio's mutex 来解决这两个问题。其守卫是 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)

Playground

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