在某些情况下，满足这两个条件很容易，而在其他情况下则更困难。在复杂情况下，一个不错的方法是将 ConcurrentOrderedBag 与一个 ConcurrentIter 配对，以极大地减轻复杂性和安全风险，请参见下面的并行映射示例。

示例

手动示例

在下面的示例中，我们将计算分配到两个线程：第一个线程处理索引为偶数的输入，第二个线程处理索引为奇数的输入。这满足了上面提到的安全要求。

use orx_concurrent_ordered_bag::*;

let n = 1024;

let evens_odds = ConcurrentOrderedBag::new();

// just take a reference and share among threads
let bag = &evens_odds;

std::thread::scope(|s| {
    s.spawn(move || {
        for i in (0..n).filter(|x| x % 2 == 0) {
            unsafe { bag.set_value(i, i as i32) };
        }
    });

    s.spawn(move || {
        for i in (0..n).filter(|x| x % 2 == 1) {
            unsafe { bag.set_value(i, -(i as i32)) };
        }
    });
});

let vec = unsafe { evens_odds.into_inner().unwrap_only_if_counts_match() };
assert_eq!(vec.len(), n);
for i in 0..n {
    if i % 2 == 0 {
        assert_eq!(vec[i], i as i32);
    } else {
        assert_eq!(vec[i], -(i as i32));
    }
}

请注意，只要满足无间隙和一次写入保证，ConcurrentOrderedBag 在写入顺序方面非常灵活。考虑以下示例。我们仅启动一个线程来两次写入集合的末尾。然后我们启动许多其他线程来填充集合的开头。这可以正常工作，无需任何锁或等待。

use orx_concurrent_ordered_bag::*;

let n = 1024;
let num_additional_threads = 4;

let bag = ConcurrentOrderedBag::new();
let con_bag = &bag;

std::thread::scope(|s| {
    s.spawn(move || {
        // start writing to the end
        unsafe { con_bag.set_value(n - 1, 42) };
    });

    for thread in 0..num_additional_threads {
        s.spawn(move || {
            // then fill the rest concurrently from the beginning
            for i in (0..(n - 1)).filter(|i| i % num_additional_threads == thread) {
                unsafe { con_bag.set_value(i, i as i32) };
            }
        });
    }
});

let vec = unsafe { bag.into_inner().unwrap_only_if_counts_match() };
assert_eq!(vec.len(), n);
for i in 0..(n - 1) {
    assert_eq!(vec[i], i as i32);
}
assert_eq!(vec[n - 1], 42);

这些示例代表了一些可以在线程之间简单分割工作并满足安全要求的情况。在一般情况下，需要特别注意才能满足安全要求。通过在输入端将 ConcurrentOrderedBag 与 ConcurrentIter 配对，可以显著避免这种复杂性和安全风险。

使用 ConcurrentIter 的并行映射

并行映射操作是我们关心收集元素顺序的情况之一，因此，ConcurrentBag 就不再适用了。另一方面，使用 ConcurrentOrderedBag 和 ConcurrentIter 可以实现一个非常简单但高效的实现。

use orx_concurrent_ordered_bag::*;
use orx_concurrent_iter::*;

fn parallel_map<In, Out, Map, Inputs>(
    num_threads: usize,
    inputs: Inputs,
    map: &Map,
) -> ConcurrentOrderedBag<Out>
where
    Inputs: ConcurrentIter<Item = In>,
    Map: Fn(In) -> Out + Send + Sync,
    Out: Send + Sync,
{
    let outputs = ConcurrentOrderedBag::new();
    let inputs = &inputs;
    let out = &outputs;
    std::thread::scope(|s| {
        for _ in 0..num_threads {
            s.spawn(|| {
                while let Some(next) = inputs.next_id_and_value() {
                    unsafe { out.set_value(next.idx, map(next.value)) };
                }
            });
        }
    });
    outputs
}

let len = 2465;
let input: Vec<_> = (0..len).map(|x| x.to_string()).collect();

let bag = parallel_map(4, input.into_con_iter(), &|x| x.to_string().len());

let output = unsafe { bag.into_inner().unwrap_only_if_counts_match() };
assert_eq!(output.len(), len);
for (i, value) in output.iter().enumerate() {
    assert_eq!(value, &i.to_string().len());
}

如您所见，无需手动工作或关注即可满足安全要求。迭代器中的每个元素都按顺序处理和写入，就像在顺序实现中一样。

使用 ConcurrentIter 的并行映射

对上述并行映射实现进行进一步性能改进的方法是将任务在多个线程中分块分配。这种方法的目标是避免伪共享，您可以在 此处 了解更多详情。这可以通过将 set_values 方法与 ConcurrentOrderedBag 配对，而不是与 ConcurrentIter 配对来实现。

use orx_concurrent_ordered_bag::*;
use orx_concurrent_iter::*;

fn parallel_map<In, Out, Map, Inputs>(
    num_threads: usize,
    inputs: Inputs,
    map: &Map,
    chunk_size: usize,
) -> ConcurrentOrderedBag<Out>
where
    Inputs: ConcurrentIter<Item = In>,
    Map: Fn(In) -> Out + Send + Sync,
    Out: Send + Sync,
{
    let outputs = ConcurrentOrderedBag::new();
    let inputs = &inputs;
    let out = &outputs;
    std::thread::scope(|s| {
        for _ in 0..num_threads {
            s.spawn(|| {
                while let Some(next) = inputs.next_chunk(chunk_size) {
                    unsafe { out.set_values(next.begin_idx, next.values.map(map)) };
                }
            });
        }
    });
    outputs
}

let len = 2465;
let input: Vec<_> = (0..len).map(|x| x.to_string()).collect();
let bag = parallel_map(4, input.into_con_iter(), &|x| x.to_string().len(), 64);

let output = unsafe { bag.into_inner().unwrap_only_if_counts_match() };
for (i, value) in output.iter().enumerate() {
    assert_eq!(value, &i.to_string().len());
}

并发状态和属性

并发状态简单地通过原子容量进行建模。此状态与 PinnedConcurrentCol 的组合导致以下属性

• 写入集合的位置不会阻止其他写入，可以同时发生多个写入。
• 调用者需要保证每个位置只写入一次。
• ⇒ 调用者负责避免写 & 写冲突。
• 一次只能发生一个增长。
• 读取只能在将袋子转换为底层的 PinnedVec 之后进行。
• ⇒ 不存在读 & 写冲突。

并发友元集合

许可

此库根据 MIT 许可证授权。有关详细信息，请参阅 LICENSE。