Rustの "future cannot be sent between threads safely" エラーの原因

Question

Rustの "future cannot be sent between threads safely" エラーについて

RustとtokioでMutexを利用した非同期コードを書いていたところ，以下のようなエラーが出てしまいました．
Mutexがdropされていないことによるエラーだと考え明示的にdropしたのですがやはり同じエラーが出てしまいます．
このエラーの出る理由と，このエラーを回避する方法を教えていただけると嬉しいです．
初めての質問ですので至らぬ点があるかとは思いますがよろしくお願いいたします．

発生している問題・エラーメッセージ

future cannot be sent between threads safely

future created by async block is not `Send`

help: within `impl std::future::Future`, the trait `std::marker::Send` is not implemented for `std::sync::MutexGuard<'_, i32>`rustc
main.rs(1, 1): required by a bound in this
main.rs(92, 22): future created by async block is not `Send`
main.rs(102, 13): has type `std::sync::MutexGuard<'_, i32>` which is not `Send`
main.rs(105, 19): await occurs here, with `mut a` maybe used later
main.rs(112, 5): `mut a` is later dropped here
spawn.rs(127, 21): required by this bound in `tokio::spawn`

エラーの再現コード

use std::sync::{Arc, Mutex};

#[tokio::test]
async fn async_test() {
    let handle = tokio::spawn(async move {
        let a = Arc::new(Mutex::new(0));
        let ret = funcA(&a).await;
        println!("ret: {}, a: {}", ret, a.lock().unwrap());
    });

    handle.await;
}

async fn funcA(this: &Arc<Mutex<i32>>) -> i32 {
    let mut a = this.lock().unwrap();
    *a += 1;

    let ret = funcB(this).await;

    println!("funcA: {}",a);

    drop(a);

    ret
}

pub async fn funcB(this: &Arc<Mutex<i32>>) -> i32 {
    let mut a = this.lock().unwrap();
    *a += 2;

    println!("funcB: {}",a);
        
    drop(a);

    100
}

補足情報

rustc: version 1.54.0
tokio: version 1.11.0

Answer 1

Tokioではマルチスレッドの非同期ランタイムを使用することを前提にAPIが設計されています。そのようなランタイムで非同期タスクを実行すると、.awaitするたびに、そのタスクを実行するexecutorスレッド（OSが提供するスレッド）が切り替わる可能性があります。つまりタスクがあるスレッドから、別のスレッドに送信されることがあります。そのためtokio::spawnは非同期タスクが、Sendトレイトを実装していることを要求します。

stdのMutexをlock()すると、Mutexガードと呼ばれるオブジェクトが返されます。このオブジェクトはスレッド間で安全に送信できないためSendを実装していません。問題となっているコードのfuncAでは、変数aが束縛されたMutexガードが、.awaitを超えて生存しています。このように書くとfuncAが返すFutureもSendでなくなります。一方でtokio::spawnはfuncAが返すFutureに対してSendを要求するためにエラーになります。

async fn funcA(this: &Arc<Mutex<i32>>) -> i32 {
    // Mutexガードを取得し、aを束縛
    let mut a = this.lock().unwrap();
    *a += 1;

    // マルチスレッドの非同期ランタイムでは、ここでこのタスクを実行する
    // スレッドが切り替わる可能性がある
    let ret = funcB(this).await;

    // Mutexガードを、.awaitを通り越して使用している
    println!("funcA: {}",a);

Mutexガードを.awaitを超えて使いたいときにはstdのMutexではなく、非同期に対応したMutexを使用します。そういうMutexは、Tokioやfuturesクレートに用意されています。

TokioのMutexを使う場合は以下のようにします。

Cargo.toml

[dependencies]
# Mutexを使うためにsyncフィーチャーを追加する
tokio = { version = "1.11.0", features = ["rt-multi-thread", "macros", "sync"] }

// stdのMutexを、tokioのMutexに変更する
use std::sync::Arc;
use tokio::sync::Mutex;

#[tokio::test]
async fn async_test() {
    let handle = tokio::spawn(async move {
        let a = Arc::new(Mutex::new(0));
        let ret = funcA(&a).await;
        // lock().unwrap()としていたところをlock().awaitに変更する
        println!("ret: {}, a: {}", ret, a.lock().await);
    });

    handle.await;
}

async fn funcA(this: &Arc<Mutex<i32>>) -> i32 {
    let mut a = this.lock().await;
    *a += 1;

    let ret = funcB(this).await;

    println!("funcA: {}", a);

    ret
}

pub async fn funcB(this: &Arc<Mutex<i32>>) -> i32 {
    let mut a = this.lock().await;
    *a += 2;

    println!("funcB: {}", a);

    100
}

これでコンパイルエラーは解消しますが、実行するとデッドロックが起こり、実行が進まなくなってしまいます。理由はfuncAの冒頭でロックを取得したままfuncBをコールし、funcBの中でまたロックを取得しようとしているからです。

これを解決するためには、funcBをコールする前にロックを解除します。

async fn funcA(this: &Arc<Mutex<i32>>) -> i32 {
    let mut a = this.lock().await;
    *a += 1;
    // デッドロックを避けるためにロックを解除する
    drop(a)

    // funcBをコールする
    let ret = funcB(this).await;

    // 再度ロックを取得する
    let a = this.lock().await;
    println!("funcA: {}", a);

    ret
}

これで完成ですが、結果的にMutexガードが.awaitを超えなくなったので、stdのMutexに戻しても問題なくコンパイルできるはずです。上のようにdrop(a)を使って書くと、最初の変数aのスコープが残っているので引き続きコンパイルエラーになるのですが、以下のようにブロックを作って変数aのスコープを制限すると、コンパイルエラーが解消します。

use std::sync::{Arc, Mutex};
// use tokio::sync::Mutex;

async fn funcA(this: &Arc<Mutex<i32>>) -> i32 {
    // スコープを制限して、Mutexガードが.awaitを超えないようにすれば
    // stdのMutexでも問題なくコンパイルできる
    {
        let mut a = this.lock().unwrap();
        *a += 1;
    }

    let ret = funcB(this).await;

    let a = this.lock().unwrap();
    println!("funcA: {}", a);

    ret
}

今回のケースではstdのMutexでも動きましたが、async fnやasyncブロック内では常に非同期に対応したMutexを使うほうがいいです。なぜなら、非同期に対応していないMutexはロックを取得できなくて待たされる状況になったときに、非同期タスクが切り替わらず、非同期ランタイムのexecutorスレッドが待機状態になってしまうからです。

非同期に対応しているMutexなら、ロック待ちになると非同期タスクが切り替わります。これにより非同期ランタイムのexecutorスレッドを無駄に待機状態にすることなく、効率的に使用できます。

Comments

[退会済みユーザー]

tokioにもMutexがあったのですね．
おかげさまで解決しました．
また，非同期処理についてもう少し勉強してみようと思います．
ありがとうございました！