5

以下の内容をお聞きしたいです。

背景

マルチスレッドについて少し触ったので配列を分割して和を、各スレッドで取得できるようにしようと考えました。

そこで以下のコードを作成しました。

use rand::prelude::*;
use std::thread;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

fn main() {
    const N: usize = 10;
    const SIZE: usize = 10000000;
    let mut tasks = vec![];

    println!("[make array]");
    let arr = create_rand_array(SIZE);
    let result = Arc::new(AtomicUsize::new(0));
    for i in 0..N - 1 {
        let picked_arr = pick_array(i * N, SIZE / N, &arr);
        let result = Arc::clone(&result);
        let handle = thread::spawn(move || {
            result.fetch_add(sum(picked_arr), Ordering::SeqCst);
        });
        tasks.push(handle);
    }

    for task in tasks {
        task.join().unwrap();
    }

    println!("sum: {:?}", result);
}

fn sum(arr: &[usize]) -> usize {
    let mut result: usize = 0;
    arr.iter().for_each(|num| {
        result += num
    });
    result
}

fn create_rand_array(num: usize) -> Vec<usize> {
    let mut arr: Vec<usize> = Vec::with_capacity(num);
    let mut rng = rand::thread_rng();
    (1..num).for_each(|_| {
        arr.push(rng.gen::<usize>() % 1000);
    });
    arr
}

fn pick_array(start: usize, num: usize, arr: &[usize]) -> &[usize] {
    if start >= 0 && start + num < arr.len() {
        &arr[start..start + num - 1]
    } else {
        panic!("Invalid Args: start -> {}, num -> {}, arr.len() -> {}", start, num, arr.len());
    }
}

このコードを実行したところ

   Compiling playground v0.0.1 (/playground)
error[E0597]: `arr` does not live long enough
  --> src/main.rs:15:54
   |
15 |         let picked_arr = pick_array(i * N, SIZE / N, &arr);
   |                          ----------------------------^^^^-
   |                          |                           |
   |                          |                           borrowed value does not live long enough
   |                          argument requires that `arr` is borrowed for `'static`
...
28 | }
   | - `arr` dropped here while still borrowed

となりました。

自分としては

for task in tasks {
    task.join().unwrap();
}

があるからarrの参照であるpicked_arrがdropされるまで(thread::spawnのクロージャを抜けるまで)、arrは存在できるのではないかと考えています。

お聞きしたいこと

  • このコンパイルエラーを解消する方法をお聞きしたいです

1 件の回答 1

6

thread::spawnで生成したスレッドに変数をムーブ(所有権を移動)するには、その変数が 'static ライフタイムを持っている必要があります。生成されたスレッドの処理は、おおもとのプロセスが終了する(メインスレッドが終了する)まではどこまでも続く可能性があるからです。

pick_array関数のライフタイムを明示的に書くと次のようになり、質問にある通りpicked_arr&arr のライフタイムは等しくなります。

fn pick_array<'a>(start: usize, num: usize, arr: &'a [usize]) -> 
 &'a [usize] {

ここではpicked_arrがスレッドにムーブされているので 'static が必要になり、pick_arrayの定義によりライフタイムが伝搬されたため &arr にも 'static が要求されている、ということになります。

とりあえずコンパイルが通るようにするには、pick_arrayからVec<usize>を返すようにする方法があります(Playground)。効率の良い方法ではありませんが…

fn pick_array(start: usize, num: usize, arr: &[usize]) -> Vec<usize> {
    if start >= 0 && start + num < arr.len() {
        arr[start..start + num - 1].into()

また、pick_arrayのようにスライスの要素をまとめて取ってくるには、chunksが便利です。他にもfoldmapを使って書き換えると次のようになります。こちらではBox::leakを使って 'static ライフタイムを持つ参照を生成しています。

use rand::prelude::*;
use std::thread::{self, JoinHandle};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

fn main() {
    const N: usize = 10;
    const SIZE: usize = 1000;

    println!("[make array]");
    let result = Arc::new(AtomicUsize::new(0));
    let arr = Box::new(create_rand_array(SIZE));
    let arr = Box::leak(arr);
    arr.chunks(N).map(|chunk| {
        let result = Arc::clone(&result);
        thread::spawn(move || {
            let sum = chunk.iter().fold(0, |sum, num| sum + *num);
            result.fetch_add(sum, Ordering::SeqCst);
        })
    })
   .collect::<Vec<JoinHandle<()>>>().into_iter()
    .for_each(|handle| handle.join().unwrap());

    println!("sum: {:?}", result);
}

fn create_rand_array(num: usize) -> Vec<usize> {
    let mut rng = rand::thread_rng();
    (1..num).map(|_| rng.gen::<usize>() % 1000).collect()
}

コメントの指摘にある通り、collectを呼ばないとイテレータが遅延評価されるため、それぞれのスレッドがspawnされた後にすぐjoinされてしまいます(比較: Playground)。

ちなみに、以前は thread::scopedという「スコープ付きスレッド」のようなものがあり、'staticではない参照を使えたのですが、いろいろとややこしい問題があり廃止されました(下記リンクを参照)。現時点ではcrossbeam, thread-scopedクレートから同等の機能を使えます。2022/8/31更新:Rust 1.63.0でスコープ付きスレッド std::thread::scope が復活しました。

参考リンク

7
  • ご回答ありがとうございます。Box::leakchunkを知らなかったので、勉強になりました。1点お聞きしたいのですが、提示いただいたプログラムだとforeachでjoinしているから、mapでspawn→join→mapでspawn→join...という結局シリアルな処理になってしまうのかと思ったのですが、どうでしょうか?実行時間を測ったところ、分割しないでfoldした方が100倍程度早くなっていました。
    – tbt
    Commented 2021年12月15日 14:05
  • 上記ですがchunk_sizeをSIZE/Nにして、foreachではtasksにpushしその後join()を呼び出すとやったら、早くなりました。お騒がせしましたmm
    – tbt
    Commented 2021年12月15日 14:24
  • 連続ですみません。SIZE=100000000, N=10までは上記の対処した方が早かったのですが、SIZE=1000000000, N=10になったらarr.chunk.for_each内でjoinした方が早くなりました。arr.chunk.for_each内でjoinした場合(Playground)と、arr.chunk.for_each内ではtasksにpushして後からtasksのfor文内でjoinした場合(Playground)の挙動の違いを教えていただけたらと思います。
    – tbt
    Commented 2021年12月15日 14:46
  • そうですね。上記の状態では遅延評価のおかげで並行処理ができていませんね。回答を修正します。SIZE/N を使うとプログラムの意味が変わってしまうので、それで速くなるのは関係ない気がします。
    – sei0o
    Commented 2021年12月17日 1:19
  • 1
    tasksというあらかじめ生成された Vecpush() する場合には、tasks 用のメモリ領域が必要です。一方、collect() を使って一旦 Vec にまとめる場合は内部で FromIterator::from_iter() が呼び出されます(イテレータからVecにする関数)。Vecfrom_iter() の実装はイテレータが持っていた領域をなるべく使い回すようになっていて、メモリを節約できます。Vecの内部実装には詳しくないので解釈には自信がありませんが…
    – sei0o
    Commented 2021年12月17日 1:41

この質問に回答するには、ログインする必要があります。

求めていた回答ではありませんか? のタグが付いた他の質問を参照する。