Rust:複数スレッドで発生する非同期処理をあとでまとめてawaitしたい

Question

rustの非同期処理についてです。javascriptでいうPromise.allをrustでもやろうと思い、以下のコードを書きました。

#[tokio::main]
async fn main() {
    let mut process_list = vec![];

    let process_a = tokio::spawn(async {
        //処理
    });
    process_list.push(process_a);

    //色々処理

    let process_b = tokio::spawn(async {
        //処理
    });
    process_list.push(process_b);

    //色々処理

    for process in process_list {
        process.await.unwrap();
    }
}

このコードは上手くいってるのですが、process_listを複数スレッドで共有しようと思い、以下のようにprocess_listをArc<Mutex<Vec<tokio::task::JoinHandle<()>>>>としてみたのですがうまくいきません。

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

#[tokio::main]
async fn main() {
    let process_list = Arc::new(Mutex::new(vec![]));
    let process_list_cloned = process_list.clone();

    
    let process_a = tokio::spawn(async {
        //処理
    });
    let mut process_list_locked = process_list_cloned.lock().unwrap();
    process_list_locked.push(process_a);

    //色々処理

    let process_b = tokio::spawn(async {
        //処理
    });
    let mut process_list_locked = process_list_cloned.lock().unwrap();
    process_list_locked.push(process_b);

    //色々処理

    let process_list_locked = process_list_cloned.lock().unwrap();
    for process in process_list_locked.into_iter() {
        process.await.unwrap();
    }
}

error[E0507]: cannot move out of dereference of `std::sync::MutexGuard<'_, Vec<tokio::task::JoinHandle<()>>>`
  --> src/main.rs:69:20
   |
69 |     for process in process_list_locked.into_iter() {
   |                    ^^^^^^^^^^^^^^^^^^^ move occurs because value has type `Vec<tokio::task::JoinHandle<()>>`, which does not implement the `Copy` trait

ここで質問なのですが、
1.そもそもやり方は間違えていないか(この方法が根本的に間違っていないか)
2.あっているのであれば、どのようにしたら動くか
3.自分がここで詰まらないために勉強しておくべきだったもの

上手く質問できていないかもしれませんがすみません。

Answer 1

こんにちは。

まずマルチスレッド版のエラーについてです。これはロックの中身である Vec をムーブ（元々あった Vec を無効化）しようとしています。これだと共有している他のスレッドからもデータが触れなくなってしまうのでエラーになってしまっています。 Vec はそのままに、 Vec の中身だけ取り出すような書き方をすればコンパイルが通ります。

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

#[tokio::main]
async fn main() {
    let process_list = Arc::new(Mutex::new(vec![]));
    let process_list_cloned = process_list.clone();

    // ロックはスコープの末尾まで有効なのでデッドロックを防ぐためにスコープを小さくする
    {
        let process_a = tokio::spawn(async {
            //処理
        });
        let mut process_list_locked = process_list.lock().unwrap();
        process_list_locked.push(process_a);
        //色々処理
    }
    {
        let process_b = tokio::spawn(async {
            //処理
        });
        let mut process_list_locked = process_list.lock().unwrap();
        process_list_locked.push(process_b);
        //色々処理
    }
    let mut process_list_locked = process_list_cloned.lock().unwrap();
    // drainでVectの中身のデータのみ取り出す。 `..` はFull Rangeのリテラル
    for process in process_list_locked.drain(..) {
        process.await.unwrap();
    }
}

この上で1, 2, 3に答えます。

1 やり方は間違えていないか

やりたいこと次第ですが、JavaScriptでいうPromise.allをやりたいなら間違えている可能性が高いです。 RustでPromise.allに対応するものはfuturesクレート のjoin_allです。for 式内で await をするとfutureを1つ1つ順番に待ちますが、 join_all を使うと同時に待ちます。

ちょっとコードを書いて実験してみましょう。タスクのあとに sleep をつけて tokio::spawn(async {}).then(sleep) とします。それを for 式で待つのと join_all で待つのを比べてみます。

use futures::prelude::*;
use std::time::{Duration, Instant};
use tokio::task::JoinError;

async fn sleep(_: Result<(), JoinError>) {
    tokio::time::sleep(Duration::from_secs(1)).await
}

#[tokio::main]
async fn main() {
    println!("waiting with `for`");
    let start = Instant::now();
    {
        let mut process_list = Vec::new();

        let process_a = tokio::spawn(async {}).then(sleep);
        process_list.push(process_a);
        //色々処理
        let process_b = tokio::spawn(async {}).then(sleep);
        process_list.push(process_b);
        //色々処理
        for process in process_list {
            process.await;
        }
    }
    let end = Instant::now();
    println!("time elapsed: {}ms", (end - start).as_millis());

    println!("waiting with `join_all`");
    let start = Instant::now();
    {
        let mut process_list = Vec::new();

        let process_a = tokio::spawn(async {}).then(sleep);
        process_list.push(process_a);
        //色々処理
        let process_b = tokio::spawn(async {}).then(sleep);
        process_list.push(process_b);
        //色々処理
        futures::future::join_all(process_list).await;
    }
    let end = Instant::now();
    println!("time elapsed: {}ms", (end - start).as_millis());
}

これを走らせると、 for 式で待つ方は1つ目の sleep が終わったあとに次の sleep が走ってそれを待つので2秒かかります。一方 join_all の方は2つの sleep が同時に走って同時に待つので1秒で終わります。

$ cargo run
    Finished dev [unoptimized + debuginfo] target(s) in 0.01s
     Running `target/debug/teratail-tokio`
waiting with `for`
time elapsed: 2002ms
waiting with `join_all`
time elapsed: 1001ms

なので十中八九 for 式で待つよりも join_all で待つ方がやりたいことでしょう。ただし、今回のコードに限っていえば tokio::spawn でタスクを裏で実行するようにしているので for 式で await してもほぼ同じことになります。もしそれを分かって書いているのであれば意図通りですが、恐らくは意図せずたまたま動いた類いのコードじゃないかなと思います。

2 あっているのであれば、どのようにしたら動くか

上記のとおりPromise.allのようなことをしたいのであればお題の並行版のコードは以下のように書くのが正しいです。

    // 略

    // drainでVectの中身のデータのみ取り出す。 `..` はFull Rangeのリテラル
    futures::future::join_all(process_list_locked.drain(..)).await;

もし、futureを1つ1つ await したいというのであればロックを使わずにチャネルを使うこともできます。Vec のように要素を入れ替えたり削除したりは難しいですが今回のように単純に push して1つ1つ取り出す目的であれば Arc<Mutex<Vec<T>>> よりも適しています。

use std::sync::mpsc::channel;

#[tokio::main]
async fn main() {
    let (sender, receiver) = channel();

    let process_a = tokio::spawn(async {
        //処理
    });
    sender.send(process_a).unwrap();
    //色々処理

    let process_b = tokio::spawn(async {
        //処理
    });
    sender.send(process_b).unwrap();
    //色々処理

    // senderをdropすることでchannelを終わらせる
    // そうしないと `for` 式が終わらない
    drop(sender);

    for process in receiver {
        process.await.unwrap();
    }
}

以上、join_all を使う方法と channel を使う方法を紹介しました。 channel に join_all を使うこともできます。

3 自分がここで詰まらないために勉強しておくべきだったもの

最初の所有権の問題に関しては正直なところこれを勉強していれば防げたというものは私は思い付きません。「データを保持するデータ型はデータ型そのもののムーブと中身のムーブの区別をつける」という一般論を覚えてあとは慣れるくらいでしょうか。

for 式と join_all の違いについては非同期の一般論とRust特有の挙動が絡みます。async book である程度カバーされているのでそちらを一読するとよいかもしれません。

チャネルについては公式ドキュメントで解説されています：メッセージ受け渡しを使ってスレッド間でデータを転送する - The Rust Programming Language 日本語版。

Rustはドキュメントの数も量も多くて全て目を通してから書き始めるのは難しいかもしれませんが困ったときは目次だけでも見返してみるといいかもしれません。困ることがあってもこういう質問サイトで質問できるので行き詰まったときは気負わずいつでも質問して下さい。

Comments

[bitt]

回答ありがとうございます！
forでawaitやってるのは代用だったのですが、drainで中身を取り出しつつ、join_allでやることでやりたいことが実現できるんですね、ありがとうございます。
またこの目的ならchannelの方がきれいに書けるんですね、こちらで書き換えようと思います。

Answer 2

1.そもそもやり方は間違えていないか(この方法が根本的に間違っていないか)

スレッド関連のコードが例示されていないので判断が難しいのですが、根本的に間違っていることはなさそうです。

2.あっているのであれば、どのようにしたら動くか

ご質問のコードには2点問題があります。まずエラーになっているところから解決しましょう。

error[E0507]: cannot move out of dereference of `std::sync::MutexGuard<'_, Vec<tokio::task::JoinHandle<()>>>`
  --> src/main.rs:69:20
   |
69 |     for process in process_list_locked.into_iter() {
   |                    ^^^^^^^^^^^^^^^^^^^ move occurs because value has type `Vec<tokio::task::JoinHandle<()>>`, which does not implement the `Copy` trait

Vecのinto_iter()でVec<_>の各要素の所有権を取ろうとしています。しかし、Mutexに入れたものは普通にアクセスしたのでは所有権が取れません。そのためにエラーになっています。

一方、Mutexに入れたものについて&や&mutなら簡単に得られます。.awaitは&mutを要求するので、それを渡してあげましょう。Vecのiter_mut()を使います。

    // &mut Vec<_>を得るためにはまずMutexGuardにmutが必要
    let mut process_list_locked = process_list_cloned.lock().unwrap();
    // Iteratorのiter_mut()で、&mut Vec<_>から、その要素に対する&mutを得る
    for process in process_list_locked.iter_mut() {
        process.await.unwrap();
    }

これでエラーが解消します。

なお、futuresクレート に join_allという関数 があって普段は便利なのですが今回のコードでは使えません。なぜならjoin_allは所有権を取るタイプのイテレーターが作れる型（IntoIteratorトレイトを実装した型）を引数に要求するからです。

もう一つの問題は実行時のデッドロックです。ご質問のコードはMutexのロックを計3回取得することになっていますが、前のロックが解除される前に次のロックを取得しようとしています。その場合にどういう動作をするのかはMutexの設計しだいなのですが、標準ライブラリーのMutexはデッドロックする仕様となっています。

Mutexのロックはそれに束縛されている変数がスコープを外れたときに解除されます。ですから以下のように{ }を使って変数のスコープを狭めてあげれば、適切なタイミングでロックが解除されるようになり、デッドロックが起こらなくなります。

#[tokio::main]
async fn main() {
    let process_list = Arc::new(Mutex::new(vec![]));
    let process_list_cloned = process_list.clone();

    {      
        let process_a = tokio::spawn(async {
            //処理
        });
        let mut process_list_locked = process_list_cloned.lock().unwrap();
        process_list_locked.push(process_a);
    } // ここでロックが解除される

    //色々処理

    {
        let process_b = tokio::spawn(async {
            //処理
        });
        let mut process_list_locked = process_list_cloned.lock().unwrap();
        process_list_locked.push(process_b);
    } // ここでロックが解除される

    //色々処理

    let mut process_list_locked = process_list_cloned.lock().unwrap();
    for process in process_list_locked.iter_mut() {
        process.await.unwrap();
    }

3.自分がここで詰まらないために勉強しておくべきだったもの

答えるのが難しい質問ですね。具体的にどういう理由で詰まったのかによって違う答えになりそうです。

例えば以下のような感じでしょうか。

エラーメッセージの意味がわからない

• → コンパイルエラーのドキュメントを読んでみる。Rust Compiler Error Index E0507

move outなど所有権まわりがわからない

• → 公式ドキュメント（The Rust Programming Language）の所有権の章（和訳）あたりか、Rustの入門書の該当の章をじっくり読んでみる

Vecにiter_mutがあることを知らなかった

• → Vecの リファレンス に何か使えそうなメソッドがないか眺める
• → iterのモジュールレベルのドキュメントを読んでみる。（ 参照をイテレートする方法 なども書かれており、そこでiter_mutが出てくる）

Comments

[bitt]

回答ありがとうございます！
なるほど、参照でも書き換え可能な&mutであればそのままawaitできるんですね、勉強になります。
mutexとかよく知らずに使っていたのですが、ロック状態は他が.lock()を呼び出すと自動で外れるのかと思ってたのですがそうではないのですね、勘違いしてました。
思ったより原因が多かったです…ドキュメントとか調べて読むのはよくても大量に読むのは苦手な人間なので、書いてみて止まったら調べるのを繰り返してるのですが、これからはちゃんと読むようにします。