orx-concurrent-iter

线程安全、易用且轻量级的并发迭代器特性和高效实现。

• 易用：实现 ConcurrentIter 的迭代器可以在线程之间安全地共享。它可以被多个线程并发迭代，使用 for 或 while let。它还提供了更高层次的方法，如 for_each 和 fold，这些方法允许安全、简单和高效地并行处理。
• 高效 和 轻量级：此 crate 中提供的所有并发迭代器实现都扩展了原子迭代器，这些迭代器是无锁的，只依赖于原子原语。

示例

基本用法

ConcurrentIter 可以在线程之间安全地共享，并且可以并发迭代。如预期的那样，它将按顺序仅生成每个元素一次。生成的元素将在基于先到先得的方式下在线程之间共享。换句话说，线程将并发地从迭代器中拉取剩余的元素。

use orx_concurrent_iter::*;
use std::fmt::Debug;

fn fake_work<T: Debug>(_x: T) {
    std::thread::sleep(std::time::Duration::from_nanos(10));
}

/// `process` elements of `iter` concurrently using `num_threads` threads
fn process_concurrently<T, I, F>(process: &F, num_threads: usize, iter: I)
where
    T: Send + Sync,
    F: Fn(T) + Send + Sync,
    I: ConcurrentIter<Item = T>,
{
    std::thread::scope(|s| {
        for _ in 0..num_threads {
            s.spawn(|| {
                // concurrently iterate over values in a `for` loop
                for value in iter.values() {
                    process(value);
                }
            });
        }
    });
}

/// executes `fake_work` concurrently on all elements of the `concurrent_iter`
fn run<T: Send + Sync + Debug>(concurrent_iter: impl ConcurrentIter<Item = T>) {
    process_concurrently(&fake_work, 8, concurrent_iter)
}

// non-consuming iteration over references
let names: [String; 3] = [
    String::from("foo"),
    String::from("bar"),
    String::from("baz"),
];
run::<&String>(names.con_iter());

let values: Vec<i32> = (0..8).map(|x| 3 * x + 1).collect();
run::<&i32>(values.con_iter());

let slice: &[i32] = values.as_slice();
run::<&i32>(slice.con_iter());
run::<i32>(slice.con_iter().cloned());

// consuming iteration over values
run::<String>(names.into_con_iter());
run::<i32>(values.into_con_iter());

// any Iterator into ConcurrentIter
let values: Vec<i32> = (0..1024).collect();

let evens = values.iter().filter(|x| *x % 2 == 0);
run::<&i32>(evens.into_con_iter());

let evens = values.iter().filter(|x| *x % 2 == 0);
run::<i32>(evens.into_con_iter().cloned());

let iter_val = values
    .iter()
    .filter(|x| *x % 2 == 0)
    .map(|x| (7 * x + 3) as usize)
    .skip(2)
    .take(5);
run::<usize>(iter_val.into_con_iter());

循环方式

ConcurrentIter 实现了 next 方法，这是 Iterator::next 的并发对应物。因此，迭代器可以在多个线程中几乎像常规 Iterator 一样安全地使用。以下示例演示并解释了迭代 ConcurrentIter 的不同方式。

use orx_concurrent_iter::*;
use std::fmt::Debug;

fn process_one_by_one<T, I, F>(process: &F, num_threads: usize, iter: &I)
where
    T: Send + Sync + Debug,
    F: Fn(T) + Send + Sync,
    I: ConcurrentIter<Item = T>,
{
    std::thread::scope(|s| {
        for _ in 0..num_threads {
            s.spawn(|| {
                // pull values 1 by 1
                for value in iter.values() {
                    process(value);
                }

                while let Some(value) = iter.next() {
                    process(value);
                }

                // pull values and corresponding index 1 by 1
                for (idx, value) in iter.ids_and_values() {
                    dbg!(idx);
                    process(value);
                }

                while let Some(x) = iter.next_id_and_value() {
                    dbg!(x.idx);
                    process(x.value);
                }
            });
        }
    });
}

fn process_in_chunks<T, I, F>(
    process: &F,
    num_threads: usize,
    iter: &I,
    chunk_size: usize,
) where
    T: Send + Sync + Debug,
    F: Fn(T) + Send + Sync,
    I: ConcurrentIter<Item = T>,
{
    std::thread::scope(|s| {
        for _ in 0..num_threads {
            s.spawn(|| {
                // pull values in chunks of `chunk_size`
                let mut chunk_iter = iter.buffered_iter(16);
                while let Some(chunk) = chunk_iter.next() {
                    assert!(chunk.values.len() <= chunk_size);

                    for (i, value) in chunk.values.enumerate() {
                        let idx = chunk.begin_idx + i;

                        dbg!(idx);
                        process(value);
                    }
                }
            });
        }
    });
}

let process = |x| {
    dbg!(x);
};

process_one_by_one(&process, 8, &(0..1024).con_iter());
process_in_chunks(&process, 8, &(0..1024).con_iter(), 64);

• for 和 while let 循环的 process_one_by_one 示例展示了最基本的使用，其中线程将在完成前一个元素的处理后，从迭代器中拉取下一个元素。
• 请注意，每个线程将根据它们完成循环内任务的速度，在迭代器的不同位置拉取不同的元素。因此，线程内部的对 enumerate 的调用，或者通过特定线程计数拉取的元素，**不**提供原始数据源中元素的索引。《ConcurrentIter》还通过 ids_and_values 或 next_id_and_value 方法提供了原始索引。
• 当循环内部要执行的工作很小（例如上面示例中的dbg调用）时，逐个取元素可能不是最优的。在这种情况下，更好的想法是批量取元素。在process_in_chunks中，我们创建了一个缓冲区块迭代器，在每次next调用时取出chunk_size（或更少，如果剩余不足）个**连续**的元素。请注意，由chunk_iter.next()返回的chunk是一个具有已知len的ExactSizeIterator。
• 在批量迭代时，我们仍然可以访问元素的原始idx。chunk.begin_idx代表返回的chunk.values迭代器的第一个元素的原始索引。请注意，chunk.values始终不为空；即，始终至少有一个元素，否则next返回None。此外，块迭代器包含连续的元素。因此，我们可以通过将chunk.begin_idx与当前chunk的局部索引结合来获取所有元素的原始索引，该索引是通过chunk.values.enumerate获得的；即，let idx = chunk.begin_idx + i。

并行折叠

将迭代的元素视为过程的输入，ConcurrentIter方便地允许将任务分配给多个线程。下面是一个使用并发迭代器的并行折叠实现示例。

use orx_concurrent_iter::*;

fn compute(input: u64) -> u64 {
    input * 2
}

fn fold(aggregated: u64, value: u64) -> u64 {
    aggregated + value
}

fn par_fold(num_threads: usize, inputs: impl ConcurrentIter<Item = u64>) -> u64 {
    std::thread::scope(|s| {
        (0..num_threads)
            .map(|_| s.spawn(|| inputs.values().map(compute).fold(0u64, fold)))
            .collect::<Vec<_>>()
            .into_iter()
            .map(|x| x.join().expect("-_-"))
            .fold(0u64, fold)
    })
}

// validate
for num_threads in [1, 2, 4, 8] {
    let values = (0..1024).map(|x| 2 * x);
    let par_result = par_fold(num_threads, values.into_con_iter());
    assert_eq!(par_result, 2 * 1023 * 1024);
}

关于实现的说明

• 并发迭代器允许简单的7行并行折叠实现。
• 并行折叠操作定义为两个折叠操作。
  • 第一个.map(_).fold(_)定义了由num_threads个线程执行的并行折叠操作。每个线程返回自己的聚合结果。
  • 第二个map(_).fold(_)定义了最终的对每个线程获得的num_threads结果执行的顺序折叠操作。

use orx_concurrent_iter::*;
use orx_concurrent_bag::*;

fn map(input: u64) -> String {
    input.to_string()
}

fn parallel_map(num_threads: usize, iter: impl ConcurrentIter<Item = u64>) -> SplitVec<String> {
    let outputs = ConcurrentBag::new();
    std::thread::scope(|s| {
        for _ in 0..num_threads {
            s.spawn(|| {
                for output in iter.values().map(map) {
                    outputs.push(output);
                }
            });
        }
    });
    outputs.into_inner()
}

// test
for num_threads in [1, 2, 4, 8] {
    let inputs = (0..1024).map(|x| 2 * x);
    let outputs = parallel_map(num_threads, inputs.into_con_iter());
    assert_eq!(1024, outputs.len());
}

use orx_concurrent_iter::*;

fn par_find<I, P>(iter: I, predicate: P, n_threads: usize) -> Option<(usize, I::Item)>
where
    I: ConcurrentIter,
    P: Fn(&I::Item) -> bool + Send + Sync,
{
    std::thread::scope(|s| {
        (0..n_threads)
            .map(|_| {
                s.spawn(|| {
                    for (i, x) in iter.ids_and_values() {
                        if predicate(&x) {
                            iter.skip_to_end();
                            return Some((i, x));
                        }
                    }
                    None
                })
            })
            .collect::<Vec<_>>()
            .into_iter()
            .flat_map(|x| x.join().expect("(-)"))
            .min_by_key(|x| x.0)
    })
}

let mut names: Vec<_> = (0..8785).map(|x| x.to_string()).collect();
names[42] = "foo".to_string();

let result = par_find(names.con_iter(), |x| x.starts_with('x'), 4);
assert_eq!(result, None);

let result = par_find(names.con_iter(), |x| x.starts_with('f'), 4);
assert_eq!(result, Some((42, &"foo".to_string())));

names[43] = "foo_second_match".to_string();
let result = par_find(names.con_iter(), |x| x.starts_with('f'), 4);
assert_eq!(result, Some((42, &"foo".to_string())));