基于C++原子变量的自旋锁队列数据丢失问题求助

你当前的SingleQueue实现存在核心逻辑错误，导致无法正确读取队列元素，和消费者线程速度无关，具体问题及修复方案如下：

1. 核心错误点 #

Pop函数逻辑完全错误 #

• Push操作时，你将元素存入_q[_count]后执行_count++，因此有效元素的索引范围是0 ~ _count-1
• 但Pop操作时，你却读取_q[_count]——这是未被赋值的内存位置，完全不是队首元素
• 同时Pop后既未减少_count，也未正确移动队首指针，队列元素计数完全混乱

队列状态判断缺失 #

• 消费者线程的while(true)会无限制调用Pop，即使队列是空的，会导致读取无效数据
• Push时未检查队列是否已满，可能触发数组越界

修复后的代码实现 #

修正后的自旋锁（保留原逻辑，补充标准内存序） #

#pragma once
#include<atomic>

class spinlock_mutex {
  std::atomic_flag flag;

public:
  spinlock_mutex() : flag(ATOMIC_FLAG_INIT) {}
  void lock() {
    // 添加memory_order_acquire，符合C++原子操作内存模型规范
    while (flag.test_and_set(std::memory_order_acquire))
      ;
  }
  void unlock() {
    // 添加memory_order_release，确保临界区内存操作可见性
    flag.clear(std::memory_order_release);
  }
};

修正后的SingleQueue队列 #

#include <stdexcept>

template <class T, uint32_t MAX_SIZE = UCHAR_MAX>
class SingleQueue {
protected:
  T _q[MAX_SIZE]{};
  uint16_t _front = 0; // 队首元素索引
  uint16_t _back = 0;  // 队尾下一个可用位置索引
  uint16_t _count = 0; // 当前元素总数
  spinlock_mutex mtx{};

public:
  SingleQueue() = default;

  // 禁止拷贝与移动，避免锁和数组的错误复制
  SingleQueue(const SingleQueue&) = delete;
  SingleQueue& operator=(const SingleQueue&) = delete;

  // 检查队列是否为空
  bool is_empty() const {
    mtx.lock();
    bool empty = (_count == 0);
    mtx.unlock();
    return empty;
  }

  // 检查队列是否已满
  bool is_full() const {
    mtx.lock();
    bool full = (_count == MAX_SIZE);
    mtx.unlock();
    return full;
  }

  void push(T &&val) {
    mtx.lock();
    if (_count >= MAX_SIZE) {
      mtx.unlock();
      throw std::runtime_error("Queue is full");
    }
    _q[_back] = std::move(val);
    _back = (_back + 1) % MAX_SIZE; // 循环队列处理，复用数组空间
    _count++;
    mtx.unlock();
  }

  void push(const T &val) {
    mtx.lock();
    if (_count >= MAX_SIZE) {
      mtx.unlock();
      throw std::runtime_error("Queue is full");
    }
    _q[_back] = val;
    _back = (_back + 1) % MAX_SIZE;
    _count++;
    mtx.unlock();
  }

  // 返回值表示是否读取成功（队列为空时返回false）
  bool pop(T &val) {
    mtx.lock();
    if (_count == 0) {
      mtx.unlock();
      return false;
    }
    val = _q[_front];
    _front = (_front + 1) % MAX_SIZE;
    _count--;
    mtx.unlock();
    return true;
  }
};

修正后的测试代码 #

#include <thread>
#include <vector>
#include <iostream>
#include <gtest/gtest.h>

class DummyComm {
public:
  int l_val1{};
  int l_val2{};
};

SingleQueue<DummyComm,200000> q{};

void producer_m() {
  for (int32_t idx = 0; idx < 100000; idx++) {
    // 频繁cout会严重拖慢线程速度，测试时可注释
    // std::cout << "Tx: " << idx << "\n";
    q.push(DummyComm{idx, idx});
  }
}

void consumer_m() {
  std::vector<DummyComm> vec{};
  DummyComm res{};
  int received_count = 0;

  // 通过计数判断是否完成，避免依赖特定元素值的偶然情况
  while (received_count < 100000) {
    if (q.pop(res)) {
      // std::cout << "Rx: " << res.l_val1 << "\n";
      vec.push_back(res);
      received_count++;
    } else {
      // 队列为空时主动出让CPU，减少空自旋的资源消耗
      std::this_thread::yield();
    }
  }

  // 验证最后一个元素正确性
  ASSERT_EQ(vec.back().l_val1, 99999);
  std::cout << "Received all " << vec.size() << " elements, last value: " << vec.back().l_val1 << std::endl;
}

TEST(testOne, basic) {
  std::thread prod(producer_m);
  std::thread cons(consumer_m);
  prod.join();
  cons.join();
}

关键修复说明 #

1. 修正Pop逻辑：改为读取队首_front位置的元素，更新队首指针并减少_count，确保每次出队都是最早入队的元素
2. 添加状态检查：Push时判断队列是否满，Pop时判断队列是否空，避免越界和无效读取
3. 循环队列优化：用取模运算处理队首/队尾指针，充分利用数组空间
4. 测试逻辑优化：通过计数判断任务完成，添加yield()减少空自旋的CPU消耗
5. 规范内存序：给原子操作添加标准内存序，符合C++内存模型规范

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