linux - pthread_cond_broadcast 问题

Question

在 linux 2.6.30 中使用 pthreads 我试图发送一个信号，这将导致多个线程开始执行。广播似乎只被一个线程接收到。我已经尝试过 pthread_cond_signal 和 pthread cond_broadcast 两者似乎具有相同的行为。对于pthread_cond_wait中的mutex，我尝试了common mutex和separate (local) mutex，没有明显区别。

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()。可能会发生虚假唤醒，您必须能够处理它们。

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

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

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

如果当最后一个工作线程在那个条件下发出信号时调度线程还没有休眠，它会发现计数已经匹配所需的计数并且不会费心等待，而是立即广播共享条件以唤醒工作人员，他们现在保证所有人都在 sleep 。

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

#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;
}