主页 / coding / 问题 / 78479584
Accepted
rbriggs
Asked: 2024-05-15 00:52:09 +0800 CST

Rust Rayon ThreadPool：“不能借用可变的，因为它是 Fn 闭包中捕获的变量”

  • 772

我正在尝试通过制作一个简单的向量加法函数来学习 Rust 的 Rayon 库。我当前的代码是这样的，假设a、b和c是初始化为相同长度的向量，c是可变的并且num_threads是一个usize变量：

let pool = ThreadPoolBuilder::new().num_threads(num_threads).build()
           .expect("Could not create thread pool");
pool.install(|| {
    (0..c.len()).into_par_iter().for_each(|x| {
        c[x] = a[x] + b[x];
    });
});

但我得到了错误

error[E0596]: cannot borrow c as mutable, as it is a captured variable in a Fn closure
    c[x] = a[x].wrap_add(b[x]);
    ^ cannot borrow as mutable

我还想指出，最终我的目标是制作基准软件，这就是为什么我使用线程池来指定线程数，但我认为额外的闭包pool.install()使用是问题根源的一部分。修改全局线程池也不是一个选项，因为这只能完成一次，而且我想使用不同的线程数重新运行基准测试。我还想避免使用范围 - 如果这是唯一的解决方案，那就这样吧 - 因为这会增加性能损失。

我从根本上理解 Rayon 在这里不喜欢什么：Rust Book 第 4.2 章说你不能对一个变量有多个可变引用，这本质上是每个线程都会得到的。然而，除了如何让这段代码工作的特定问题之外，这还引发了一些其他问题。

我似乎甚至无法将引用移至c线程池闭包中。为什么这是一个限制？当然，多线程的强大之处在于让多个线程同时对相关数据进行一些工作，那么为什么我不能以这种方式将数据传递给线程呢？

假设我可以获得对线程池的引用c，是否有某种方法可以实现，例如，线程 0 gets &mut c[0]、线程 1 gets&mut c[1]等等？如果 Rayon 的目的是抽象 Rust 基本多线程库的一些样板，那么 Rayon 不应该尝试让这变得更简单吗？

我看到的其他一些答案暗示迭代向量本身的内容会有所帮助，但由于我需要所有三个向量，所以我需要使用izip。这样做（替换(0..c.len().into_par_iter()...为izip!(&a.mat, &b.mat, &mut c.mat).into_par_iter()...）给了我错误

error[E0599]: the method into_par_iter exists for struct Map<Zip<Zip<Iter<'_, T>, Iter<'_, T>>, IterMut<'_, T>>, {[email protected]:303:9}>, but its trait bounds were not satisfied

izip!(&a.mat, &b.mat, &mut c.mat).into_par_iter().for_each(|x| {
   |                                                   ^^^^^^^^^^^^^
   |
   = note: the following trait bounds were not satisfied:
           `std::iter::Map<std::iter::Zip<std::iter::Zip<std::slice::Iter<'_, T>, std::slice::Iter<'_, T>>, std::slice::IterMut<'_, T>>, {closure@/home/richard/.cargo/registry/src/index.crates.io-6f17d22bba15001f/itertools-0.12.1/src/lib.rs:303:9: 303:10}>: rayon::iter::ParallelIterator`
           which is required by `std::iter::Map<std::iter::Zip<std::iter::Zip<std::slice::Iter<'_, T>, std::slice::Iter<'_, T>>, std::slice::IterMut<'_, T>>, {closure@/home/richard/.cargo/registry/src/index.crates.io-6f17d22bba15001f/itertools-0.12.1/src/lib.rs:303:9: 303:10}>: rayon::iter::IntoParallelIterator`
           `&std::iter::Map<std::iter::Zip<std::iter::Zip<std::slice::Iter<'_, T>, std::slice::Iter<'_, T>>, std::slice::IterMut<'_, T>>, {closure@/home/richard/.cargo/registry/src/index.crates.io-6f17d22bba15001f/itertools-0.12.1/src/lib.rs:303:9: 303:10}>: rayon::iter::ParallelIterator`
           which is required by `&std::iter::Map<std::iter::Zip<std::iter::Zip<std::slice::Iter<'_, T>, std::slice::Iter<'_, T>>, std::slice::IterMut<'_, T>>, {closure@/home/richard/.cargo/registry/src/index.crates.io-6f17d22bba15001f/itertools-0.12.1/src/lib.rs:303:9: 303:10}>: rayon::iter::IntoParallelIterator`
           `&mut std::iter::Map<std::iter::Zip<std::iter::Zip<std::slice::Iter<'_, T>, std::slice::Iter<'_, T>>, std::slice::IterMut<'_, T>>, {closure@/home/richard/.cargo/registry/src/index.crates.io-6f17d22bba15001f/itertools-0.12.1/src/lib.rs:303:9: 303:10}>: rayon::iter::ParallelIterator`
           which is required by `&mut std::iter::Map<std::iter::Zip<std::iter::Zip<std::slice::Iter<'_, T>, std::slice::Iter<'_, T>>, std::slice::IterMut<'_, T>>, {closure@/home/richard/.cargo/registry/src/index.crates.io-6f17d22bba15001f/itertools-0.12.1/src/lib.rs:303:9: 303:10}>: rayon::iter::IntoParallelIterator`

这似乎意味着引用现在正在被移至关闭中。是什么赋予了？之前为什么不搬走呢？

multithreading

2 个回答

  1. Best Answer

    您应该做的第一件事是将代码编写为普通迭代器，严格使用迭代器方法链。

    c.iter_mut()
    .zip(&a)
    .zip(&b)
    .for_each(|((c, &a), &b)| *c = a + b);

    这可以通过添加 直接转换为多线程版本par_，无需其他更改。这是 Rayon 的理想场景。

    pool.install(|| {
    c.par_iter_mut()
        .zip(&a)
        .zip(&b)
        .for_each(|((c, &a), &b)| *c = a + b);
});

    Rayon 提供了for tuples的实现IntoParallelIterator，它允许您制作更简单的等效版本。

    pool.install(|| {
    (&mut c, &a, &b)
        .into_par_iter()
        .for_each(|(c, &a, &b)| *c = a + b);
});

    请注意，并非所有情况都会如此简单。特别是，如果每个线程只需要其中之一，请查看_initor_with方法。需要累积为单个值的迭代器通常需要使用foldand reduce。请务必通读ParallelIterator并IndexedParallelIterator找到最适合您情况的方法。

    • 2

  2. 我已经想出了一个解决办法。我需要c通过在线程池的闭包中使用迭代器来更明确地进行借用 - 这似乎也解释了我的问题的“假设我可以......”部分 - 以及 @user2407038 的并行 zip 建议：

    let pool = ThreadPoolBuilder::new().num_threads(num_threads).build()
           .expect("Could not create thread pool");
pool.install(|| {
    (&mut c).into_par_iter().zip(0..c.len()).for_each(|x| {
        *x.0 = a[x.1] + b[x.1];
    });
});
    • 1

相关问题