单消费者MPSC队列中已出队但std::counting_semaphore::try_acquire()失败排查

问题描述 #

实现了单消费者多生产者无锁队列（MPSCQueue），搭配std::counting_semaphore在元素入队时通知消费者：生产者完成入队后调用sema.release()，消费者先尝试从队列取元素，若成功则调用sema.try_acquire()消耗信号量。但极少情况下触发断言失败：队列返回有效元素，try_acquire()却返回false。

已知条件：

• 仅单个消费者线程调用dequeue()
• 队列所有原子操作使用std::memory_order_seq_cst
• 信号量使用标准release()/acquire()接口

核心成因 #

问题根源是队列元素可见性与信号量计数可见性未绑定同步：

1. 生产者线程中，队列入队的最终原子操作（prev_head->next.store(seq_cst)）确实先于sema.release()执行，但两者仅依赖线程内的happens-before关系，未对消费者线程形成强制跨线程同步约束。
2. 消费者线程中，读取队列元素（tail->next.load(seq_cst)）和读取信号量计数（try_acquire()）是独立操作。即使消费者通过队列的原子操作同步看到了新元素，信号量计数的更新可能因CPU缓存未同步，仍停留在旧值，导致try_acquire()误判失败。

简言之：队列的原子操作触发了缓存同步，让消费者看到新元素，但信号量计数的更新尚未同步到消费者CPU缓存，就执行了try_acquire()。

解决方案 #

方案1：调整消费者逻辑（推荐，最安全） #

修改消费者dequeue()逻辑，先获取信号量，再从队列取元素。该逻辑完全符合信号量同步语义，从根源上避免可见性不一致问题：

T dequeue() {
  sema.acquire(); // 先等待信号量，确保队列必有元素
  auto re = queue.dequeue();
  assert(re.has_value()); // 此时队列一定存在元素，断言不会触发
  return re.value();
}

此方案逻辑简单，无内存序隐患。即使队列已有多个元素，acquire()会直接返回（因信号量计数大于0），性能损失可忽略。

方案2：添加内存屏障（适配原有逻辑） #

若需保留“先尝试取元素，再消耗信号量”的逻辑，可在消费者读取到队列元素后添加全局内存屏障，强制同步所有缓存状态，确保信号量计数更新对消费者可见：

T dequeue() {
  auto re = queue.dequeue();
  if (re.has_value()) {
    std::atomic_thread_fence(std::memory_order_seq_cst); // 强制缓存同步
    if (!sema.try_acquire()) {
      std::cerr << __FILE__ << ":" << __FUNCTION__
                << " sema.try_acquire() should succeed, please check\n";
      assert(false);
    }
    return re.value();
  }
  sema.acquire();
  return queue.dequeue().value();
}

std::memory_order_seq_cst栅栏会强制消费者CPU刷新所有缓存，确保能看到生产者所有已完成的操作（包括sema.release()），从而保证try_acquire()结果与队列状态一致。

额外注意点 #

• 原MPSCQueue的size()实现存在线程安全问题：enqueue_count是原子变量，但dequeue_count是普通变量，消费者修改dequeue_count时，生产者读取enqueue_count可能导致大小计算不准确。若需要正确的队列大小统计，应将dequeue_count改为std::atomic_size_t。
• 无锁队列的节点内存管理逻辑正确，析构函数能避免内存泄漏或野指针问题。

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