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