C++原子操作实现并发素数筛提升core_count后小素数丢失问题

你观察到的素数丢失和compare_exchange的原子性无关，是代码逻辑和并发规则使用错误导致的，核心问题有3个：

1. CAS操作预期值未重置（最核心漏洞）
   C++标准明确规定：compare_exchange_weak/strong操作失败时，会将第一个入参的预期值自动更新为原子变量的当前实际值。你现有代码中写入output的逻辑是：

int zero = 0;
while (!output.compare_exchange_weak(zero, claimed, order))
  ;

当线程A成功将output从0改为素数P1后，线程B进入这段逻辑：第一次CAS用0比对失败，zero被自动更新为P1，下一次循环会直接尝试将当前值为P1的output替换为自己的素数，直接覆盖了前一个线程的写入结果，主线程根本没有机会读取P1，直接导致素数丢失。核心数越高，并发写入冲突概率越大，丢失的素数越多，完全匹配你观察到的现象。
2. 收尾阶段数据漏读
主线程的收集循环条件是while (finished_worker_count < core_count)，所有worker执行完成后循环会立刻退出，此时output中可能还残留最后一个写入的素数没有被读取，也会导致数据丢失。
3. 内存序选择的潜在问题
全场景使用memory_order_relaxed仅保证操作本身的原子性，不保证多线程间的操作可见性和执行顺序，可能出现合数标记对其他线程不可见、素数claim操作和output写入顺序被重排的问题，进一步放大数据异常概率。

可直接运行的修复代码 #

#include <algorithm>
#include <atomic>
#include <future>
#include <iostream>
#include <iterator>
#include <thread>
#include <vector>

int main() {
  using namespace std;
  // 调整为acq_rel保证操作顺序和可见性，不需要最强的seq_cst也能保证正确性
  constexpr memory_order order = memory_order_acq_rel;
  atomic<int> output{0};
  vector<atomic_bool> sieve(10000);
  for (auto& each : sieve) atomic_init(&each, false);
  atomic<unsigned> finished_worker_count{0};

  auto const worker = [&output, &sieve, &finished_worker_count]() {
    for (auto current = next(sieve.begin(), 2); current != sieve.end();) {
      current = find_if(current, sieve.end(), [](atomic_bool& value) {
        bool untrue = false;
        return value.compare_exchange_strong(untrue, true, order);
      });
      if (current == sieve.end()) break;
      int const claimed = static_cast<int>(distance(sieve.begin(), current));
      // 修复：每次CAS都用新的0作为预期值，不会用被修改过的旧值覆盖已有结果
      while (true) {
        int zero = 0;
        if (output.compare_exchange_weak(zero, claimed, order)) {
          break;
        }
        this_thread::yield(); // 减少空转CPU占用
      }
      for (auto product = 2 * claimed; product < static_cast<int>(sieve.size());
           product += claimed)
        sieve[product].store(true, order);
    }
    finished_worker_count.fetch_add(1, order);
  };

  const auto core_count = thread::hardware_concurrency();
  vector<future<void>> futures;
  futures.reserve(core_count);
  generate_n(back_inserter(futures), core_count,
             [&worker]() { return async(worker); });
  vector<int> result;
  while (finished_worker_count < core_count) {
    auto current = output.exchange(0, order);
    if (current > 0) result.push_back(current);
  }
  // 修复：收尾读取最后一个残留的素数
  auto last = output.exchange(0, order);
  if (last > 0) result.push_back(last);

  sort(result.begin(), result.end());
  for (auto each : result) cout << each << " ";
  cout << '\n';
  return 0;
}

并发编程最佳实践建议 #

• 使用compare_exchange系列函数时，必须明确预期值会被失败操作修改的特性，不要复用被修改过的预期值，除非你的逻辑确实需要基于当前值做更新。
• 多线程间传递数据不要用单个原子变量作为通道，本质是长度为1的队列，冲突概率极高、性能损耗大，优先采用无锁队列/分段锁的实现。
• 不要随意使用memory_order_relaxed，除非你完全确定不需要任何顺序保证，优先用memory_order_acq_rel或者默认的memory_order_seq_cst，后者性能略低但逻辑正确性更容易保证，排查问题成本更低。
• 多线程任务收尾阶段，必须检查公共数据通道中是否还有未处理的残留数据，避免最后一批数据丢失。
• 并发素数筛的更合理实现是给每个线程分配独立的筛法区间，减少CAS操作的冲突次数，当前实现每个素数都要抢一次公共写入通道，核心数超过4之后性能提升会非常有限。

内容的提问来源于stack exchange，提问作者Krokeledocus