/** Random Delay in microseonds */ std::random_device rd; thread_local std::mt19937 gen(rd()); thread_local std::uniform_int_distribution<> dis(0, 4095);
voidunpredictableDelay(int extra = 0){ if (dis(gen) == 0) { this_thread::sleep_for(chrono::nanoseconds(2000 + extra)); } }
/** thread function for counting */ voidworker(int id){ for (int i = 0; i < increments_per_thread; ++i) { uint64_t old = counter.load(memory_order_relaxed); // 如果 load(relaxed) 不能看到当前最新值 // 那么 CAS 就会加多次，最终结果会大于 max_count while (old < max_count && !counter.compare_exchange_weak(old, old + 1, memory_order_relaxed)) { // old is updated with the current value of counter unpredictableDelay(dis(gen)); } } // cout << "Worker " << id << " done." << endl; }
/** main function */ intmain(){ for (int run = 0; run < run_times; ++run) { counter.store(0, memory_order_relaxed); cout << "Run " << run << ": "; thread threads[num_threads];
for (int i = 0; i < num_threads; ++i) { threads[i] = thread(worker, i); }
for (int i = 0; i < num_threads; ++i) { threads[i].join(); }
voidtry_advance_epoch(){ int current = global_epoch.load(std::memory_order_relaxed); for (int i = 0; i < MAX_THREADS; ++i) { if (active[i].load(std::memory_order_relaxed) && local_epoch[i].load(std::memory_order_relaxed) != current) { return; // 有线程还在旧 epoch，不能推进 } }
int next_epoch = (current + 1) % 3; global_epoch.store(next_epoch, std::memory_order_relaxed);
像 Michael-Scott 队列这种链表结构，节点被 pop 出队后地址可能被重用。如果没有安全的内存回收，另一个线
程可能 CAS 成功指向了一个“已经被释放并重用的地址”，这就是 ABA 的根源。所以 hazard pointer 或 epoch
GC 是在链表队列里用来避免这种 悬空引用 + ABA 的。
# Fails if not enough memory to enqueue try_enqueue(item) : bool try_enqueue(prod_token, item) : bool try_enqueue_bulk(item_first, count) : bool try_enqueue_bulk(prod_token, item_first, count) : bool
# Attempts to dequeue from the queue (never allocates) try_dequeue(item&) : bool try_dequeue(cons_token, item&) : bool try_dequeue_bulk(item_first, max) : size_t try_dequeue_bulk(cons_token, item_first, max) : size_t
# If you happen to know which producer you want to dequeue from try_dequeue_from_producer(prod_token, item&) : bool try_dequeue_bulk_from_producer(prod_token, item_first, max) : size_t
# A not-necessarily-accurate count of the total number of elements size_approx() : size_t
auto start = std::chrono::high_resolution_clock::now();
// 启动生产者线程 std::vector<std::thread> producers; for (int p = 0; p < NUM_PRODUCERS; ++p) { producers.emplace_back([&queue, p]() { for (int i = 0; i < ITEMS_PER_PRODUCER; ++i) { queue.enqueue(i + p * ITEMS_PER_PRODUCER); } }); }
// 启动消费者线程 std::vector<std::thread> consumers; for (int c = 0; c < NUM_CONSUMERS; ++c) { consumers.emplace_back([&queue, &totalConsumed, c, &threadWaitTimes]() { int item; auto localStart = std::chrono::high_resolution_clock::now(); while (true) { auto t0 = std::chrono::high_resolution_clock::now(); while (!queue.try_dequeue(item)) { // busy wait } auto t1 = std::chrono::high_resolution_clock::now(); threadWaitTimes[c] += std::chrono::duration<double>(t1 - t0).count();
if (++totalConsumed >= N) break; } auto localEnd = std::chrono::high_resolution_clock::now(); double threadTime = std::chrono::duration<double>(localEnd - localStart).count(); std::cout << "Consumer " << c << " finished in " << threadTime << "s, wait time: " << threadWaitTimes[c] << "s\n"; }); }
for (auto& t : producers) t.join(); for (auto& t : consumers) t.join();
auto end = std::chrono::high_resolution_clock::now(); std::chrono::duration<double> elapsed = end - start; opsPerSec = N / elapsed.count();