不修改队列模块，用纯锁机制解决线程安全循环队列竞态问题

问题背景 #

我给enqueueTS()和dequeueTS()分别配置了独立互斥锁，原本想让写线程和读线程能并行执行、无需互相等待，但这两个函数都会修改原队列struct queue_t中的numOfElements成员，最终引发了竞态条件。

原队列模块（queue.c） #

struct queue_t {
  unsigned head;
  unsigned tail;
  unsigned size;
  unsigned numOfElements;
  unsigned elementSize;
  char data[];
};

int enqueue(queue_t *me, const void *item) {
  if (!me || !item) {
    queue_error_callback("%s: invalid input\n", __func__);
    return -1;
  }

  if (me->numOfElements == me->size) {
    queue_error_callback("%s: queue is full\n", __func__);
    return -1;
  }

  memcpy(&me->data[me->tail * me->elementSize], item, me->elementSize);

  if (++me->tail == me->size) {
    me->tail = 0;
  }

  me->numOfElements++;

  return 0;
}

int dequeue(queue_t *me, void *item) {
  if (!me || !item) {
    queue_error_callback("%s: invalid input\n", __func__);
    return -1;
  }

  if (me->numOfElements == 0) {
    queue_error_callback("%s: queue is empty\n", __func__);
    return -1;
  }

  memcpy(item, &me->data[me->head * me->elementSize], me->elementSize);

  if (++me->head == me->size) {
    me->head = 0;
  }

  me->numOfElements--;

  return 0;
}

当前线程安全封装（thread-safety queue.c） #

struct thread_safe_queue {
  queue_t *queue;
  pthread_mutex_t enqueue_mutex;
  pthread_mutex_t dequeue_mutex;
  pthread_cond_t queue_not_full;
  pthread_cond_t queue_not_empty;
};

int enqueueTS(ts_queue_t *tsq, const void *item) {
  if (!tsq || !item) {
    queue_error_callback("%s: invalid input\n", __func__);
    return -1;
  }

  pthread_mutex_lock(&tsq->enqueue_mutex);
  while (numOfEmptySlots(tsq->queue) == 0) {
    pthread_cond_wait(&tsq->queue_not_full, &tsq->enqueue_mutex);
  }
  int ret = enqueue(tsq->queue, item);
  pthread_cond_signal(&tsq->queue_not_empty);
  pthread_mutex_unlock(&tsq->enqueue_mutex);
  return ret;
}

int dequeueTS(ts_queue_t *tsq, void *item) {
  if (!tsq || !item) {
    queue_error_callback("%s: invalid input\n", __func__);
    return -1;
  }

  pthread_mutex_lock(&tsq->dequeue_mutex);
  while (numOfElements(tsq->queue) == 0) {
    pthread_cond_wait(&tsq->queue_not_empty, &tsq->dequeue_mutex);
  }
  int ret = dequeue(tsq->queue, item);
  pthread_cond_signal(&tsq->queue_not_full);
  pthread_mutex_unlock(&tsq->dequeue_mutex);
  return ret;
}

我不想把numOfElements改成原子类型，原因是：

• 希望只用纯锁机制实现；
• 希望将纯队列模块和线程安全功能完全分离，遵循开闭原则。

我的问题：有没有其他方法能让enqueueTS()和dequeueTS()并行执行且无竞态条件，同时完全不修改queue.c的内容？

解决方案：新增计数锁分离保护逻辑 #

可以通过在线程安全封装层新增一个计数互斥锁，单独保护对numOfElements的访问和条件变量的等待/通知，同时保留原有的enqueue/dequeue独立锁来保护队列指针（head/tail）和数据拷贝操作。这样读写线程可以并行执行核心的队列操作，同时避免计数的竞态问题，且完全不修改原队列模块。

修改后的线程安全队列结构体 #

struct thread_safe_queue {
  queue_t *queue;
  pthread_mutex_t enqueue_mutex;  // 保护tail修改、数据入队拷贝
  pthread_mutex_t dequeue_mutex;  // 保护head修改、数据出队拷贝
  pthread_mutex_t count_mutex;    // 保护numOfElements访问、条件变量同步
  pthread_cond_t queue_not_full;
  pthread_cond_t queue_not_empty;
};

修正后的enqueueTS()实现 #

int enqueueTS(ts_queue_t *tsq, const void *item) {
  if (!tsq || !item) {
    queue_error_callback("%s: invalid input\n", __func__);
    return -1;
  }

  // 第一步：用计数锁检查队列是否已满，等待可用空间
  pthread_mutex_lock(&tsq->count_mutex);
  while (tsq->queue->numOfElements == tsq->queue->size) {
    pthread_cond_wait(&tsq->queue_not_full, &tsq->count_mutex);
  }
  pthread_mutex_unlock(&tsq->count_mutex);

  // 第二步：用enqueue锁执行入队操作（此时已确保有空间）
  pthread_mutex_lock(&tsq->enqueue_mutex);
  int ret = enqueue(tsq->queue, item);
  pthread_mutex_unlock(&tsq->enqueue_mutex);

  // 第三步：用计数锁更新状态并通知读线程
  pthread_mutex_lock(&tsq->count_mutex);
  pthread_cond_signal(&tsq->queue_not_empty);
  pthread_mutex_unlock(&tsq->count_mutex);

  return ret;
}

修正后的dequeueTS()实现 #

int dequeueTS(ts_queue_t *tsq, void *item) {
  if (!tsq || !item) {
    queue_error_callback("%s: invalid input\n", __func__);
    return -1;
  }

  // 第一步：用计数锁检查队列是否为空，等待可用元素
  pthread_mutex_lock(&tsq->count_mutex);
  while (tsq->queue->numOfElements == 0) {
    pthread_cond_wait(&tsq->queue_not_empty, &tsq->count_mutex);
  }
  pthread_mutex_unlock(&tsq->count_mutex);

  // 第二步：用dequeue锁执行出队操作（此时已确保有元素）
  pthread_mutex_lock(&tsq->dequeue_mutex);
  int ret = dequeue(tsq->queue, item);
  pthread_mutex_unlock(&tsq->dequeue_mutex);

  // 第三步：用计数锁更新状态并通知写线程
  pthread_mutex_lock(&tsq->count_mutex);
  pthread_cond_signal(&tsq->queue_not_full);
  pthread_mutex_unlock(&tsq->count_mutex);

  return ret;
}

方案优势 #

1. 并发度保留：入队和出队的核心操作（数据拷贝、指针移动）由独立锁保护，读写线程可并行执行，不会互相阻塞；
2. 完全遵循开闭原则：原队列模块（queue.c）无任何修改，线程安全逻辑完全封装在独立层；
3. 纯锁实现：未使用原子操作，完全依赖互斥锁和条件变量保证线程安全。

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