ForkJoin

一个基于工作窃取的 Fork-Join 并行库。

受博客文章 Rust 中的数据并行 启发，作为硕士学位论文的一部分实现。仓库托管在 github.com/faern/forkjoin

库文档托管在 此处

这个库是为了满足三种类型算法的需求而开发的，这些算法非常适合 Fork-Join 并行。

Reduce 风格

Reduce 风格是算法接收一个参数，递归地从该参数计算一个值，并返回一个答案。这种风格的一个例子是递归查找第 n 个斐波那契数和树结构的求和。这种风格的特点是算法不需要修改其参数，并且结果仅在所有子任务完全计算后才能获得。

在 reduce 风格算法中，每个子任务的返回值传递给一个特殊的 join 函数，该函数在所有子任务完成后执行。如果算法具有 ReduceStyle::Arg，则可以直接从任务向该 join 函数发送额外的参数。这在本例中可以体现出来。

Reduce 风格示例 (ReduceStyle::NoArg)

use forkjoin::{FJData,TaskResult,ForkPool,AlgoStyle,ReduceStyle,Algorithm};

fn fib_30_with_4_threads() {
    let forkpool = ForkPool::with_threads(4);
    let fibpool = forkpool.init_algorithm(Algorithm {
        fun: fib_task,
        style: AlgoStyle::Reduce(ReduceStyle::NoArg(fib_join)),
    });

    let job = fibpool.schedule(30);
    let result: usize = job.recv().unwrap();
    assert_eq!(1346269, result);
}

fn fib_task(n: usize, _: FJData) -> TaskResult<usize, usize> {
    if n < 2 {
        TaskResult::Done(1)
    } else {
        TaskResult::Fork(vec![n-1,n-2], None)
    }
}

fn fib_join(values: &[usize]) -> usize {
    values.iter().fold(0, |acc, &v| acc + v)
}

Reduce 风格示例 (ReduceStyle::Arg)

use forkjoin::{FJData,TaskResult,ForkPool,AlgoStyle,ReduceStyle,Algorithm};

struct Tree {
    value: usize,
    children: Vec<Tree>,
}

fn sum_tree(t: &Tree) -> usize {
    let forkpool = ForkPool::new();
    let sumpool = forkpool.init_algorithm(Algorithm {
        fun: sum_tree_task,
        style: AlgoStyle::Reduce(ReduceStyle::Arg(sum_tree_join)),
    });
    let job = sumpool.schedule(t);
    job.recv().unwrap()
}

fn sum_tree_task(t: &Tree, fj: FJData) -> TaskResult<&Tree, usize> {
    if t.children.is_empty() {
        TaskResult::Done(t.value)
    } else if fj.depth > fj.workers { // Bad example of serial threshold
        TaskResult::Done(sum_tree_seq(t))
    } else {
        let mut fork_args: Vec<&Tree> = vec![];
        for c in t.children.iter() {
            fork_args.push(c);
        }
        TaskResult::Fork(fork_args, Some(t.value)) // Pass current nodes value to join
    }
}

fn sum_tree_seq(t: &Tree) -> usize {
    t.value + t.children.iter().fold(0, |acc, t2| acc + sum_tree_seq(t2))
}

fn sum_tree_join(value: &usize, values: &[usize]) -> usize {
    *value + values.iter().fold(0, |acc, &v| acc + v)
}

Search 风格

Search 风格会持续返回结果，有时可以无参数开始，或从某种初始状态开始。算法在执行过程中会生成一个或多个输出值，可能在中间任何地方终止。例如，n皇后和数独求解器等算法具有这种风格。Search 风格的特点是它们可以生成多个结果，并且可以在树中的所有任务计算完成之前终止。

Search 风格示例

use forkjoin::{FJData,ForkPool,TaskResult,AlgoStyle,Algorithm};

type Queen = usize;
type Board = Vec<Queen>;
type Solutions = Vec<Board>;

fn search_nqueens() {
    let n: usize = 8;
    let empty = vec![];

    let forkpool = ForkPool::with_threads(4);
    let queenpool = forkpool.init_algorithm(Algorithm {
        fun: nqueens_task,
        style: AlgoStyle::Search,
    });

    let job = queenpool.schedule((empty, n));

    let mut solutions: Vec<Board> = vec![];
    loop {
        match job.recv() {
            Err(..) => break, // Job has completed
            Ok(board) => solutions.push(board),
        };
    }
    let num_solutions = solutions.len();
    println!("Found {} solutions to nqueens({}x{})", num_solutions, n, n);
}

fn nqueens_task((q, n): (Board, usize), fj: FJData) -> TaskResult<(Board,usize), Board> {
    if q.len() == n {
        TaskResult::Done(q)
    } else {
        let mut fork_args: Vec<(Board, usize)> = vec![];
        for i in 0..n {
            let mut q2 = q.clone();
            q2.push(i);

            if ok(&q2[..]) {
                fork_args.push((q2, n));
            }
        }
        TaskResult::Fork(fork_args, None)
    }
}

fn ok(q: &[usize]) -> bool {
    for (x1, &y1) in q.iter().enumerate() {
        for (x2, &y2) in q.iter().enumerate() {
            if x2 > x1 {
                let xd = x2-x1;
                if y1 == y2 || y1 == y2 + xd || (y2 >= xd && y1 == y2 - xd) {
                    return false;
                }
            }
        }
    }
    true
}

原地修改风格

注意：这种风格在当前库版本中工作，但它需要非常丑陋的不安全代码！

原地修改风格接收一个可变参数，递归地修改这个值，结果是参数本身。排序算法，如排序其输入数组的算法，就是这种风格的例子。这种风格的特点是它们修改其输入参数而不是生成任何输出。

当可以很好地实现时，将会提供这种风格的示例。

任务

可以执行并可以选择分叉或返回值的微小单元是 TaskFun。TaskFun永远不能阻塞，因为那样会阻塞其正在执行的内核线程。相反，它应该决定是否已经完成计算或需要分叉。这个决定通过返回值来体现，向用户表明TaskFun需要在任何事件发生之前返回。

TaskFun返回一个 TaskResult。如果它已完成计算，则返回 TaskResult::Done(value)。如果需要分叉，则返回 TaskResult::Fork(args)。

待办事项

• 在不提供join函数的情况下使修改风格算法正常工作
• 实现一个排序算法。快速排序？
• 在NoArg分叉时不需要返回None
• 使能够在同一个池中使用具有不同Arg和Ret的算法。
• 使ForkJoin在稳定Rust中工作。
• 在搜索风格的channel周围删除互斥锁。