并行递归修复

Question

对 Rust 很陌生，并试图解决玩具问题。尝试编写仅使用 Rayon 的目录遍历。

struct Node {
    path: PathBuf,
    files: Vec<PathBuf>,
    hashes: Vec<String>,
    folders: Vec<Box<Node>>,
}

impl Node {
    pub fn new(path: PathBuf) -> Self {
        Node {
            path: path,
            files: Vec::new(),
            hashes: Vec::new(),
            folders: Vec::new(),
        }
    }
    
    pub fn burrow(&mut self) {
        let mut contents: Vec<PathBuf> = ls_dir(&self.path);

        contents.par_iter().for_each(|item| 
                                if item.is_file() {
                                    self.files.push(*item);
                                } else if item.is_dir() {
                                    let mut new_folder = Node::new(*item);
                                    new_folder.burrow();
                                    self.folders.push(Box::new(new_folder));
                                });
        
    }
}

我收到的错误是

error[E0596]: cannot borrow `*self.files` as mutable, as it is a captured variable in a `Fn` closure
  --> src/main.rs:40:37
   |
40 | ...                   self.files.push(*item);
   |                       ^^^^^^^^^^^^^^^^^^^^^^ cannot borrow as mutable

error[E0507]: cannot move out of `*item` which is behind a shared reference
  --> src/main.rs:40:53
   |
40 | ...                   self.files.push(*item);
   |                                       ^^^^^ move occurs because `*item` has type `PathBuf`, which does not implement the `Copy` trait

error[E0507]: cannot move out of `*item` which is behind a shared reference
  --> src/main.rs:42:68
   |
42 | ...                   let mut new_folder = Node::new(*item);
   |                                                      ^^^^^ move occurs because `*item` has type `PathBuf`, which does not implement the `Copy` trait

error[E0596]: cannot borrow `*self.folders` as mutable, as it is a captured variable in a `Fn` closure
  --> src/main.rs:44:37
   |
44 | ...                   self.folders.push(Box::new(new_folder));
   |                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ cannot borrow as mutable

这些错误很明显，因为它们阻止了不同的线程访问可变内存，但我只是不确定如何开始解决这些错误。

以下是burrow 的原始（非并行）版本

pub fn burrow(&mut self) {
    let mut contents: Vec<PathBuf> = ls_dir(&self.path);

    for item in contents {
        if item.is_file() {
            self.files.push(item);
        } else if item.is_dir() {
            let mut new_folder = Node::new(item);
            new_folder.burrow();
            self.folders.push(Box::new(new_folder));
        }
    }
}

Answer 1

在这种情况下，最好的选择是使用ParallelIterator::partition_map()，它允许您将并行迭代器转换为两个不同的根据某些条件收集，这正是您需要做的。

示例程序：

use rayon::iter::{Either, IntoParallelIterator, ParallelIterator};

fn main() {
    let input = vec!["a", "bb", "c", "dd"];

    let (chars, strings): (Vec<char>, Vec<&str>) =
        input.into_par_iter().partition_map(|s| {
            if s.len() == 1 {
                Either::Left(s.chars().next().unwrap())
            } else {
                Either::Right(s)
            }
        });

    dbg!(chars, strings);
}

如果你有三个不同的输出，很遗憾 Rayon 不支持。我还没有研究是否可以使用 Rayon 的特征进行构建，但我建议作为更通用（尽管效率不高）的解决方案是使用渠道.像std::sync::mpsc 这样的通道允许任意数量的线程插入项目，而另一个线程删除它们——在你的情况下，将它们移动到集合中。这不会像并行收集那样有效，但是在像您这样的以 IO 为主的问题中，它并不重要。

Answer 2

我将跳过文件和文件夹的分离，忽略结构，并演示一种简单的递归方法，递归获取目录中的所有文件：

fn burrow(dir: &Path) -> Vec<PathBuf> {
    let mut contents = vec![];

    for entry in std::fs::read_dir(dir).unwrap() {
        let entry = entry.unwrap().path();
        if entry.is_dir() {
            contents.extend(burrow(&entry));
        } else {
            contents.push(entry);
        }
    }

    contents
}

如果要使用来自 rayon 的并行迭代器，第一步是将此循环转换为非并行迭代器链。最好的方法是使用.flat_map()展平产生多个元素的结果：

fn burrow(dir: &Path) -> Vec<PathBuf> {
    std::fs::read_dir(dir)
        .unwrap()
        .flat_map(|entry| {
            let entry = entry.unwrap().path();
            if entry.is_dir() {
                burrow(&entry)
            } else {
                vec![entry] // use a single-element Vec if not a directory
            }
        })
        .collect()
}

那么使用 rayon 来并行处理这个迭代就是使用.par_bridge() 将一个迭代器转换为一个并行迭代器。实际上就是这样：

use rayon::iter::{ParallelBridge, ParallelIterator};

fn burrow(dir: &Path) -> Vec<PathBuf> {
    std::fs::read_dir(dir)
        .unwrap()
        .par_bridge()
        .flat_map(|entry| {
            let entry = entry.unwrap().path();
            if entry.is_dir() {
                burrow(&entry)
            } else {
                vec![entry]
            }
        })
        .collect()
}

在playground 上查看它。您可以对此进行扩展以收集更复杂的结果（如文件夹和哈希等）。