c - 为什么有些线程没有收到 pthread_cond_broadcast？

Question

我有一个工作线程池。每个 worker 执行这个例程:

void* worker(void* args){
  ...
  pthread_mutex_lock(&mtx);

  while (queue == NULL && stop == 0){
    pthread_cond_wait(&cond, &mtx);
  }

  el = pop(queue);
  pthread_mutex_unlock(&mtx);

  ...
}

主线程:

int main(){

   ...
   while (stop == 0){
     ...
     pthread_mutex_lock(&mtx);  
     insert(queue, el);
     pthread_cond_signal(&cond);
     pthread_mutex_unlock(&mtx);
     ...
   }
...
}

然后我有一个信号处理程序，它在收到信号时执行此代码:

void exit_handler(){
    stop = 1;   
    pthread_mutex_lock(&mtx);
    pthread_cond_broadcast(&cond);
    pthread_mutex_unlock(&mtx); 
}

我省略了声明和初始化，但原始代码有它们。

大部分时间收到信号后一切正常，但有时似乎某些工作线程停留在等待循环中，因为它们没有看到变量 stop 发生变化和/或它们没有被唤醒广播。

所以线程永远不会结束。我缺少什么？

编辑:stop=1 移入了 exit_handler 的关键部分。问题依然存在。

EDIT2:我正在使用 Ubuntu 的虚拟机上执行该程序。由于代码看起来完全正确，我尝试更改 VM 和操作系统 (XUbuntu)，现在它似乎可以正常工作。仍然不知道为什么，有人有想法吗？

Answer 1

这里有些猜测，但是评论太长了，所以如果有错，我会删除。我认为您可能对 pthread_cond_broadcast 的工作原理有误解(至少我过去一直被它困扰)。来自man page :

The pthread_cond_broadcast() function shall unblock all threads currently blocked on the specified condition variable cond.

好的，这是有道理的，_broadcast 唤醒当前阻塞在 cond 上的所有线程。 但是，只有一个被唤醒的线程在全部被唤醒后能够锁定互斥体。同样来自手册页:

The thread(s) that are unblocked shall contend for the mutex according to the scheduling policy (if applicable), and as if each had called pthread_mutex_lock().

所以这意味着如果 3 个线程在 cond 上被阻塞并且 _broadcast 被调用，所有 3 个线程都会被唤醒，但只有 1 个线程可以获取互斥量。另外 2 个仍将停留在 pthread_cond_wait 中，等待信号。因此，他们看不到 stop 设置为 1，并且 exit_handler(我假设是 Ctrl+c 软件信号？)完成信号，所以剩下的输掉 _broadcast 竞争的线程陷入困境，等待永远不会到来的信号，并且无法读取已设置的 stop 标志。

我认为有 2 个选项可以解决/解决这个问题:

1. 使用pthread_cond_timedwait .即使没有收到信号，这也会在指定的时间间隔从等待返回，看到 stop == 1，然后退出。
2. 在 worker 函数的末尾添加 pthread_cond_signal 或 pthread_cond_broadcast。这样，就在线程退出之前，它将向 cond 变量发出信号，允许任何其他等待的线程获取互斥锁并完成处理。如果没有线程在等待条件变量，则向它发送信号没有任何坏处，因此即使对于最后一个线程也应该没问题。

---

编辑:这是一个 MCVE 证明(据我所知)我上面的回答是错误的，呵呵。只要我按下 Ctrl+c，程序就会“立即”退出，这对我来说意味着所有线程在广播后迅速获取互斥锁，看到 stop 为假，然后退出。然后 main 加入线程并结束进程。

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <stdbool.h>
#include <signal.h>
#include <unistd.h>


#define NUM_THREADS 3
#define STACK_SIZE 10

pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t c = PTHREAD_COND_INITIALIZER;
volatile bool stop = false;
int stack[STACK_SIZE] = { 0 };
int sp = 0; // stack pointer,, also doubles as the current stack size

void SigHandler(int sig)
{
  if (sig == SIGINT)
  {
    stop = true;
  }
  else
  {
    printf("Received unexcepted signal %d\n", sig);
  }
}

void* worker(void* param)
{
  long tid = (long)(param);
  while (stop == false)
  {
    // acquire the lock
    pthread_mutex_lock(&m);
    while (sp <= 0)  // sp should never be < 0
    {
      // there is no data in the stack to consume, wait to get signaled
      // this unlocks the mutex when it is called, and locks the
      // mutex before it returns
      pthread_cond_wait(&c, &m);
    }

    // when we get here we should be guaranteed sp >= 1
    printf("thread %ld consuming stack[%d] = %d\n", tid, sp-1, stack[sp-1]);
    sp--;

    pthread_mutex_unlock(&m);

    int sleepVal = rand() % 10;
    printf("thread %ld sleeping for %d seconds...\n", tid, sleepVal);
    sleep(sleepVal);
  }
  pthread_exit(NULL);
}

int main(void)
{
  pthread_t threads[NUM_THREADS];
  pthread_attr_t attr;

  pthread_attr_init(&attr);
  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

  srand(time(NULL));

  for (long i=0; i<NUM_THREADS; i++)
  {
    int rc = pthread_create(&threads[i], &attr, worker, (void*)i);
    if (rc != 0)
    {
      fprintf(stderr, "Failed to create thread %ld\n", i);
    }
  }

  while (stop == false)
  {
    // produce data in bursts
    int numValsToInsert = rand() % (STACK_SIZE - sp);
    printf("main producing %d values\n", numValsToInsert);
    // acquire the lock
    pthread_mutex_lock(&m);

    for (int i=0; i<numValsToInsert; i++)
    {
      // produce values for the stack
      int val = rand() % 10000;
      // I think this should already be guaranteed..?
      if (sp+1 < STACK_SIZE)
      {
        printf("main pushing stack[%d] = %d\n", sp, val);
        stack[sp++] = val;
        // signal the workers that data is ready
        //printf("main signaling threads...\n");
        //pthread_cond_signal(&c);
      }
      else
      {
        printf("stack full!\n");
      }
    }

    pthread_mutex_unlock(&m);

    // signal the workers that data is ready
    printf("main signaling threads...\n");
    pthread_cond_broadcast(&c);  

    int sleepVal = 1;//rand() % 5;
    printf("main sleeping for %d seconds...\n", sleepVal);
    sleep(sleepVal);    
  }


  for (long i=0; i<NUM_THREADS; i++)
  {
    pthread_join(threads[i], NULL);
  }

  return 0;
}