orx-concurrent-bag

一个高效、便捷且轻量级的只增并发数据结构，允许高性能并发收集。

• 便捷: ConcurrentBag 可以简单地作为一个共享引用在多个线程之间安全地共享。它是一个具有特殊并发状态实现的 PinnedConcurrentCol。底层的 PinnedVec 和并发袋可以相互转换。
• 高效: ConcurrentBag 是一个无锁结构，适合并发、无复制和高性能增长。您可以查看 基准测试 和进一步的 性能说明 以获取详细信息。

示例

底层的 PinnedVec 保证使得使用共享引用安全增长变得简单，从而提供了如以下所示的便捷 API。

use orx_concurrent_bag::*;

let bag = ConcurrentBag::new();

let (num_threads, num_items_per_thread) = (4, 1_024);

let bag_ref = &bag;
std::thread::scope(|s| {
   for i in 0..num_threads {
       s.spawn(move || {
           for j in 0..num_items_per_thread {
               // concurrently collect results simply by calling `push`
               bag_ref.push(i * 1000 + j);
           }
       });
   }
});

let mut vec_from_bag = bag.into_inner().to_vec();
vec_from_bag.sort();
let mut expected: Vec<_> = (0..num_threads).flat_map(|i| (0..num_items_per_thread).map(move |j| i * 1000 + j)).collect();
expected.sort();
assert_eq!(vec_from_bag, expected);

并发状态和属性

并发状态通过一个原子长度简单地建模。这个状态与 PinnedConcurrentCol 的组合导致以下属性

• 向集合中的位置写入不会阻塞其他写入，可以并发发生多个写入。
• 每个位置恰好写入一次。
• ⟹ 不存在写和写的竞争条件。
• 一次只能发生一个增长。增长是无复制的，且不改变已推入元素的内存位置。
• 底层固定向量始终有效，可以通过 into_inner(self) 在任何时间取出。
• 只有在将袋转换为底层 PinnedVec 之后才能进行读取。
• ⟹ 不存在读和写的竞争条件。

基准测试

使用 push 的性能

您可以在 benches/collect_with_push.rs 中找到基准测试的详细信息。

在实验中，使用了 rayon 的并行迭代器、AppendOnlyVec 和 ConcurrentBag 的 push 方法来从多个线程收集结果。此外，还评估了 ConcurrentBag 的不同底层固定向量。

// reserve and push one position at a time
for j in 0..num_items_per_thread {
    bag.push(i * 1000 + j);
}

我们观察到 ConcurrentBag 允许高效地进行元素的并发收集

• 并发包的默认增长策略 Doubling，由于其无需任何先验知识而非常灵活，似乎已经超越了其他选择。因此，在大多数情况下都可以使用。
• Linear 增长策略需要一个参数来确定底层 SplitVec 的均匀片段容量。当我们希望更保守地进行分配时，可能会选择这种策略。记住，与标准 Vec 类似，Doubling 的容量呈指数增长；而正如其名称所暗示的，Linear 则线性增长。
• 最后，Fixed 增长策略最不灵活，需要完美了解硬约束的容量（如果超过则会引发恐慌）。由于它并不优于 Doubling 或 Linear，因此除了在极少数情况下我们想要预先分配已知所需的确切内存之外，我们并不一定需要使用 Fixed。

通过使用 extend 方法而不是 push 方法，可以进一步提高性能。您可以在下一小节中看到结果，并在性能注释中找到详细信息。

使用 extend 的性能

您可以在benches/collect_with_extend.rs中找到基准测试的详细信息。

在这个后续实验中，唯一的区别是我们使用 extend 而不是 push 与 ConcurrentBag 一起。期望这种方法将解决在 小数据 & 少工作量 情况下由于伪共享导致的性能下降。

在完全均匀的场景中，我们可以如下平均地将工作分配给线程。

// reserve num_items_per_thread positions at a time
// and then push as the iterator yields
let iter = (0..num_items_per_thread).map(|j| i * 100000 + j);
bag.extend(iter);

然而，我们不需要拥有完美的均匀性或对每个线程要推送的项目数量的完美信息，就可以获得 extend 的好处。我们可以简单地按 step_by 分步进行，并以 batch_size 元素为扩展单位。足够大的 batch_size，使得批处理大小元素超过缓存行，就可以防止因伪共享而导致的性能急剧下降。

// reserve batch_size positions at each iteration
// and then push as the iterator yields
for j in (0..num_items_per_thread).step_by(batch_size) {
    let iter = (j..(j + batch_size)).map(|j| i * 100000 + j);
    bag.extend(iter);
}

虽然通过 ConcurrentBag::push 进行并发收集非常高效，但只要可能，就应该考虑使用 ConcurrentBag::extend 进行收集，因为它会改变规模。如图所示，仅在收集数千个元素时以每批64个元素的批次收集，就能提供数量级的改进。

并发友元集合

性能注意事项

我们旋转多少次和多久？

push 和 extend 方法的唯一等待或旋转条件：当底层的 PinnedVec 需要增长时。请注意，固定向量的增长是无复制的。因此，当它旋转时，它等待的只是分配。由于增长次数是确定的，所以旋转次数也是确定的。

例如，假设我们将总共 15_000 个元素并发地推送到一个空的袋子中。

• 进一步假设我们使用默认的 SplitVec<_, Doubling> 作为底层固定向量。在整个执行过程中，我们将分配容量为 [4, 8, 16, ..., 4096, 8192] 的片段，总容量将达到 16_380。换句话说，在整个执行过程中我们将恰好分配 12 次。
• 如果我们使用具有恒定片段长度 1_024 的 SplitVec<_, Linear>，我们将分配 15 个等容量的片段。
• 如果我们使用严格的 FixedVec<_>，我们必须预分配一个安全数量，并且永远不会超过这个数字。因此，将永远不会有任何旋转。

use orx_concurrent_bag::*;

let (num_threads, num_items_per_thread) = (4, 1_024);
let bag = ConcurrentBag::new();
let batch_size = 64;

let bag_ref = &bag;
std::thread::scope(|s| {
    for i in 0..num_threads {
        s.spawn(move || {
            for j in (0..num_items_per_thread).step_by(batch_size) {
                let iter = (j..(j + batch_size)).map(|j| i * 1000 + j);
                bag_ref.extend(iter);
            }
        });
    }
});