pthread_cond_broadcast 问题

Question

在 linux 2.6.30 中使用 pthreads 我试图发送一个信号，这将导致多个线程开始执行。广播似乎只被一个线程接收。我已经尝试过 pthread_cond_signal 和 pthread cond_broadcast ，它们似乎都有相同的行为。对于 pthread_cond_wait 中的互斥体，我尝试了普通互斥体和单独（本地）互斥体，没有明显区别。

worker_thread(void *p)
{
   // setup stuff here

   printf("Thread %d ready for action \n", p->thread_no);
   pthread_cond_wait(p->cond_var, p->mutex);
   printf("Thread %d off to work \n", p->thread_no);

   // work stuff
}

dispatch_thread(void *p)
{
   // setup stuff

   printf("Wakeup, everyone ");
   pthread_cond_broadcast(p->cond_var);
   printf("everyone should be working \n");

   // more stuff
}

main()
{
   pthread_cond_init(cond_var);

   for (i=0; i!=num_cores; i++) {
      pthread_create(worker_thread...);
   }

   pthread_create(dispatch_thread...);
}

输出：

  Thread 0 ready for action
  Thread 1 ready for action
  Thread 2 ready for action
  Thread 3 ready for action
  Wakeup, everyone
  everyone should be working
  Thread 0 off to work

向所有线程发送信号的好方法是什么？

Answer 1

首先，您应该将互斥锁锁定在您调用pthread_cond_wait() 的位置。当您调用 pthread_cond_broadcast() 时，通常最好保留互斥锁。

其次，您应该在等待条件为真时循环调用pthread_cond_wait()。可能会发生虚假唤醒，您必须能够处理它们。

最后，您的实际问题是：您正在向所有线程发出信号，但在发送信号时，其中一些线程还没有等待。您的主线程和调度线程正在与您的工作线程竞争：如果主线程可以启动调度线程，并且调度线程可以在工作线程之前获取互斥锁并在其上广播，那么这些工作线程将永远不会唤醒。

在发出信号之前，您需要一个同步点，直到所有线程都在等待信号为止。那，或者您可以一直发出信号，直到您知道所有线程都已被唤醒。

在这种情况下，您可以使用互斥锁来保护休眠线程的数量。每个线程获取互斥量并增加计数。如果计数与工作线程的计数匹配，那么它是增加计数的最后一个线程，因此在另一个共享相同互斥锁的条件变量上向睡眠调度线程发出信号，表明所有线程都已准备好。然后线程等待原始条件，这导致它释放互斥锁。

如果在最后一个工作线程发出信号时调度线程还没有休眠，它会发现计数已经与期望的计数匹配，无需等待，而是立即在共享条件上广播以唤醒工作人员，工作人员现在保证所有人都在睡觉。

无论如何，这里有一些工作源代码充实了您的示例代码并包含了我的解决方案：

#include <stdio.h>
#include <pthread.h>
#include <err.h>

static const int num_cores = 8;

struct sync {
  pthread_mutex_t *mutex;
  pthread_cond_t *cond_var;
  int thread_no;
};

static int sleeping_count = 0;
static pthread_cond_t all_sleeping_cond = PTHREAD_COND_INITIALIZER;

void *
worker_thread(void *p_)
{
  struct sync *p = p_;
   // setup stuff here

   pthread_mutex_lock(p->mutex);
   printf("Thread %d ready for action \n", p->thread_no);

   sleeping_count += 1;
   if (sleeping_count >= num_cores) {
       /* Last worker to go to sleep. */
       pthread_cond_signal(&all_sleeping_cond);
   }

   int err = pthread_cond_wait(p->cond_var, p->mutex);
   if (err) warnc(err, "pthread_cond_wait");
   printf("Thread %d off to work \n", p->thread_no);
   pthread_mutex_unlock(p->mutex);

   // work stuff
   return NULL;
}

void *
dispatch_thread(void *p_)
{
  struct sync *p = p_;
   // setup stuff

   pthread_mutex_lock(p->mutex);
   while (sleeping_count < num_cores) {
       pthread_cond_wait(&all_sleeping_cond, p->mutex);
   }
   printf("Wakeup, everyone ");
   int err = pthread_cond_broadcast(p->cond_var);
   if (err) warnc(err, "pthread_cond_broadcast");
   printf("everyone should be working \n");
   pthread_mutex_unlock(p->mutex);

   // more stuff
   return NULL;
}

int
main(void)
{
   pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
   pthread_cond_t cond_var = PTHREAD_COND_INITIALIZER;

   pthread_t worker[num_cores];
   struct sync info[num_cores];
   for (int i = 0; i < num_cores; i++) {
      struct sync *p = &info[i];
      p->mutex = &mutex;
      p->cond_var = &cond_var;
      p->thread_no = i;
      pthread_create(&worker[i], NULL, worker_thread, p);
   }

   pthread_t dispatcher;
   struct sync p = {&mutex, &cond_var, num_cores};
   pthread_create(&dispatcher, NULL, dispatch_thread, &p);

   pthread_exit(NULL);
   /* not reached */
   return 0;
}