c++ - 广播不适用于障碍

Question

我正在尝试使用 pthreads 实现一个基本的工作线程池。场景是我想要固定数量的 worker ，他们在我的计划期间一直生活。

我永远不需要向单个线程发送信号，而是一次向所有线程发送信号，这就是我想要进行单个广播的原因。

在主程序继续之前，我需要等待所有线程完成，因此我决定在每个工作线程中使用 barrier_wait。

问题是，如果我的线程调用 barrier_wait，广播将不起作用。

完整示例和可编译代码如下所示。这只是为了广播的单一触发，在我的完整版本中，我会循环播放类似的内容

while(conditionMet){
  1.prepare data
  2.signal threads using data
  3.post processing of thread results (because of barrier all threads finished)
  4.modify conditionMet if needed
}

谢谢

#include <pthread.h>
#include <stdio.h>
#include <unistd.h>
void checkResults(char *str,int i){
  fprintf(stdout,"%s:%d\n",str,i);
}
void checkResults(char *str,size_t n,int i){
  fprintf(stdout,"%s[%lu]:%d\n",str,n,i);
}

/* For safe condition variable usage, must use a boolean predicate and  */
/* a mutex with the condition.                                          */
int                 conditionMet = 0;
pthread_cond_t      cond  = PTHREAD_COND_INITIALIZER;
pthread_mutex_t     mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_barrier_t barr;

#define NTHREADS    3

void *threadfunc(void *parm)
{
  size_t i = (size_t) parm;
  int           rc;

  rc = pthread_mutex_lock(&mutex);
  checkResults("\tpthread_mutex_lock()",i, rc);

  while (0==conditionMet) {
    printf("\tThread blocked[%d]\n",(int)i);
    rc = pthread_cond_wait(&cond, &mutex);
    checkResults("\tpthread_cond_wait()",i, rc);
    checkResults("\tbefore barrier",i);
   rc = pthread_barrier_wait(&barr);//broadcast works if this is commented out
    if(rc)
      fprintf(stdout,"problems waiting for baarr\n");
    checkResults("\tafter  barrier",i);
  }

  rc = pthread_mutex_unlock(&mutex);
  checkResults("\tpthread_mutex_lock()",i, rc);
  return NULL;
}

int main(int argc, char **argv)
{
  int                   rc=0;
  int                   i;
  pthread_t             threadid[NTHREADS];

  if(pthread_barrier_init(&barr, NULL,NTHREADS))
    {
      printf("Could not create a barrier\n");
    }



  printf("Enter Testcase - %s\n", argv[0]);

  printf("Create %d threads\n", NTHREADS);
  for(i=0; i<NTHREADS; ++i) {
    rc = pthread_create(&threadid[i], NULL, threadfunc,(void *) i);
    if(rc)
      checkResults("pthread_create()", rc);
  }

  sleep(5);  /* Sleep isn't a very robust way to serialize threads */
  rc = pthread_mutex_lock(&mutex);
  checkResults("pthread_mutex_lock()", rc);

  /* The condition has occured. Set the flag and wake up any waiters */
  conditionMet = 1;
  printf("\nWake up all waiters...\n");
  rc = pthread_cond_broadcast(&cond);
  checkResults("pthread_cond_broadcast()", rc);

  rc = pthread_mutex_unlock(&mutex);
  checkResults("pthread_mutex_unlock()", rc);

  printf("Wait for threads and cleanup\n");
  for (i=0; i<NTHREADS; ++i) {
    rc = pthread_join(threadid[i], NULL);
    checkResults("pthread_join()", rc);
  }
  pthread_cond_destroy(&cond);
  pthread_mutex_destroy(&mutex);

  printf("Main completed\n");
  return 0;
}

Answer 1

线程函数会在收到信号后立即锁定mutex。因此，只有一个线程函数将在屏障上等待(mutex 仍处于锁定状态)并且永远不会满足屏障条件。

您应该重新设计应用程序的逻辑才能使用屏障。 mutex 必须在等待屏障之前立即解锁。此外，考虑到您的代码中使用了 pthread_cond_wait()，您的应用程序中只会有一个线程处于事件状态，这完全消除了多线程的需要。

编辑:

我想详细说明一下最后一句话。假设我们像这样修改线程函数:

while (0==conditionMet) {     
    printf("\tThread blocked[%d]\n",(int)i);     
    rc = pthread_cond_wait(&cond, &mutex);     
    checkResults("\tpthread_cond_wait()",i, rc);     
    checkResults("\tbefore barrier",i);

    pthread_mutex_unlock(&mutex); //added    

    rc = pthread_barrier_wait(&barr);//broadcast works if this is commented out     
    if(rc)
        fprintf(stdout,"problems waiting for baarr\n");     
    checkResults("\tafter  barrier",i);   
}

这样我们就可以消除死锁，当只有一个线程能够到达屏障时 mutex 被锁定。但是在给定的时刻仍然只有一个线程会在临界区运行:当它的 pthread_cond_wait() 返回时 mutex 被锁定并且它将保持锁定状态直到线程函数到达_开锁(); _等待();一对。只有在下一个线程能够运行并到达其屏障之后。清洗、漂洗、重复……

OP prolly 想要的是让线程函数同时运行(否则为什么会有人想要线程池？)。在这种情况下，函数可能如下所示:

void *threadfunc(void *parm)
{
/*...*/
struct ThreadRuntimeData {
} rtd;
while (0==conditionMet) {     
    printf("\tThread blocked[%d]\n",(int)i);     
    rc = pthread_cond_wait(&cond, &mutex);     
    checkResults("\tpthread_cond_wait()",i, rc);

    GetWorkData(&rtd); //Gets some data from critical section and places it in rtd
    pthread_mutex_unlock(&mutex);

    ProcessingOfData(&rtd); //here we do the thread's job 
    //without the modification of global data; this may take a while

    pthread_mutex_lock(&mutex);
    PublishProcessedData(&rtd); //Here we modify some global data 
    //with the results of thread's work. 
    //Other threads may do the same, so we had to enter critical section again
    pthread_mutex_unlock(&mutex);   
    checkResults("\tbefore barrier",i);
    rc = pthread_barrier_wait(&barr);//broadcast works if this is commented out     
    if(rc)
        fprintf(stdout,"problems waiting for baarr\n");     
    checkResults("\tafter  barrier",i);   
}
/*...*/
}

这当然只是一个草图。线程函数的优化设计取决于OP希望线程做什么。

作为旁注，用于检查 pthread_barrier_wait() 返回结果的代码必须考虑到 PTHREAD_BARRIER_SERIAL_THREAD 返回。此外，将 conditionMet 声明为 volatile 会更安全。