无法从存储在向量中的 JoinHandles 加入线程 - Rust答案

【问题标题】：Unable to join threads from JoinHandles stored in a Vector - Rust无法从存储在向量中的 JoinHandles 加入线程 - Rust
【发布时间】：2021-08-28 18:00:23
【问题描述】：

我正在编写一个程序，它从网站列表中抓取数据并将其存储到一个名为 Listing 的结构中，然后将其收集到一个名为 Listings 的最终结构中。

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

fn main() {
    // ... some declarations
    let sites_count = site_list.len(); // site_list is a vector containing the list of websites

    // The variable to be updated by the thread instances ( `Listing` is a struct holding the information ) 
    let listings: Arc<Mutex<Vec<Vec<types::Listing<String>>>>> = Arc::new(Mutex::new(Vec::new()));

    // A vector containing all the JoinHandles for the spawned threads
    let mut fetch_handle: Vec<thread::JoinHandle<()>> = Vec::new();

    // Spawn a thread for each concurrent website
    for i in 0..sites_count { 
        let slist = Arc::clone(&site_list);
        let listng = Arc::clone(&listings);
        fetch_handle.push(
            thread::spawn(move || {
                println!("⌛ Spawned Thread: {}",i);
                let site_profile = read_profile(&slist[i]);
                let results = function1(function(2)) // A long list of functions from a submodule that make the http request and parse the data into `Listing`
                listng.lock().unwrap().push(results);
            }));
    }
    
    for thread in fetch_handle.iter_mut() { 
        thread.join().unwrap();
    }

    // This is the one line version of the above for loop - yields the same error.
    // fetch_handle.iter().map(|thread| thread.join().unwrap()); 

    // The final println to just test feed the target struct `Listings` with the values
    println!("{}",types::Listings{ date_time: format!("{}", chrono::offset::Local::now()),
                                   category: category.to_string(),
                                   query: (&search_query).to_string(),
                                   listings: listings.lock().unwrap() // It prevents me from owning this variable
                                 }.to_json());
}

我偶然发现了这个错误

error[E0507]: cannot move out of `*thread` which is behind a mutable reference
   --> src/main.rs:112:9
    |
112 |         thread.join().unwrap();
    |         ^^^^^^ move occurs because `*thread` has type `JoinHandle<()>`, which does not implement the `Copy` trait

它阻止我在 thread.join() for 循环之后拥有该变量。

当我尝试分配检查输出类型时

let all_listings = listings.lock().unwrap()

all_listings 报告了一种 MutexGuard（在线程 for 循环中也是如此，但它允许我在其上调用向量方法）并且不允许我拥有数据。我更改了Listings 结构中的数据类型以保存引用而不是拥有它。但似乎我对.to_json() 中的结构执行的操作要求我拥有它的价值。 listings 在 Listings 结构中的类型声明是 Vec<Vec<Listing<T>>。

但是，当我将 .join().unwrap() 移动到 thread::spawn() 块的末尾或将其应用于 for 循环内的句柄（同时禁用外部 .join() ）时，此代码可以正常工作。但这使得所有线程都在一条链中执行，这是不可取的，因为使用线程的主要目的是同时执行具有不同数据值的相同功能。

总的来说，我对 Rust 还是很陌生（我使用它已经 3 周了），这是我第一次实现多线程。在此之前，我只用 java 和 python 编写过单线程程序，所以如果可能的话，对菜鸟友好一点。然而，任何帮助表示赞赏:)。

【问题讨论】：

把for thread in fetch_handle.iter_mut() { 改成for thread in fetch_handle.into_iter() {
我没有足够的经验来弄清楚你必须做什么，但@PiRocks 的建议可能会奏效，因为 JoinHandle<()> 然后完全移入当前迭代。您的方式可能不起作用，因为当您join 和unwrap（我不确定哪个部分更重要）时，向量中还剩下什么？这是我自己不确定的生命周期，但无论如何，这可能是错误的“来源”。 into_iter“破坏”原始向量，允许在加入后明确所有权。我不确定做你想做的事情的正确方法。
只要for thread in fetch_handle 应该可以工作，因为Vec 实现了IntoIterator。单行版只需要into_iter()。
另见How to take ownership of T from Arc<Mutex<T>>?你可能不想在最后使用.lock().unwrap()，因为lock借用了互斥锁内的项目而不是将其移出（就像into_inner一样） .
@PiRocks 是的，这似乎完全解决了问题。 @Kevin 好的。我还没有进入任何 into* 方法。我还不完全明白它的作用，但我会查一下。 @trentcl 我在单行版本中尝试了into_iter()，但它弹出警告Unused Map that must be used, iterators are lazy and do nothing until consumed。 @trentcl 好的。因此，仅在线程内使用.lock().unwrap() 并在不再需要互斥锁时使用try_unwrap() 和.into_inner() 组合，明白了。谢谢@PiRocks @Kevin 和 @trentcl ！

标签： multithreading rust

【解决方案1】：

我知道需要发生什么。首先，对于这种事情，我同意into_iter 做你想做的事，但它在 IMO 中掩盖了为什么。 为什么是当你借用它时，它不拥有该值，这对于JoinHandle<()> 结构上的join() 方法是必需的。你会注意到它的签名是self 而不是&mut self 或类似的东西。所以它需要 real 对象。

为此，您需要将对象从其内部的Vec<thread::JoinHandle<()>> 中取出。如前所述，into_iter 这样做是因为它“破坏”了现有的Vec 并接管了它，因此它完全拥有内容，并且迭代返回要连接的“实际”对象而没有副本。但是您也可以使用remove 一次拥有一个内容，如下所示：

while fetch_handle.len() > 0 {
    let cur_thread = fetch_handle.remove(0); // moves it into cur_thread
    cur_thread.join().unwrap();
}

这不是上面的 for 循环。 complete example in the playground is linked 如果你想试试的话。

我希望这更清楚地说明如何处理无法复制的东西，但方法需要完全拥有它们，以及将它们从集合中取出的问题。想象一下，如果您需要只结束其中一个线程，并且您知道要结束哪个线程，但又不想全部结束？ Vec<_>::remove 可以，但into_iter 不行。

感谢您提出一个让我思考的问题，并促使我自己去查找答案（并尝试一下）。我还在学习 Rust，所以这对我很有帮助。

编辑：

使用pop() 和while let 的另一种方法：

while let Some(cur_thread) = fetch_handle.pop() {
    cur_thread.join().unwrap();
}

这会从末端穿过它（pop 将其从末端拉出，而不是从前面拉出），但也不会通过将矢量内容从前面拉出来重新分配或移动矢量内容。

【讨论】：

这比我自己的回答更清楚。我会接受这个答案。
我个人会使用for cur_thread in fetch_handle.drain(..)。但通往罗马的路有很多。
@Arthur 如果使用.drain(..)，那么您也可以使用into_iter。如果您只参与其中的一部分，排水听起来不错，但如果您正在耗尽整个事情，我看不出这样做的好处。可能是因为在使用drain 之后它是一个可用的Vec，而into_iter 对原来的Vec 进行了改进，但总的来说这似乎是一个不确定的情况，尽管在某些特定情况下可能会出现。
@KevinAnderson 我认为它更明确。您可以更轻松地从名称中看出，无需接触文档，循环完成后元素应该消失。这意味着循环可以取得所有权，这意味着它可能确实取得了所有权。然后可以阅读文档，发现确实如此。

【解决方案2】：

好的，@PiRocks 指出的问题似乎在加入线程的 for 循环中。

 for thread in fetch_handle.iter_mut() {
        thread.join().unwrap();
    }

问题是iter_mut()。改用into_iter()

 for thread in fetch_handle.into_iter() {
        thread.join().unwrap();
    }

不会产生错误，并且程序会根据需要同时跨线程运行。

@Kevin Anderson 对此的解释是：

使用into_iter() 会导致JoinHandle<()> 进入for 循环。

同时查看文档(std::iter) 我发现 iter() 和 iter_mut() 遍历 self 的引用，而 into_iter() 遍历 self 直接（拥有它）。

所以iter_mut() 正在迭代&mut thread::JoinHandle<()> 而不是thread::JoinHandle<()>。

【讨论】：