广播不适用于障碍

Question

我正在尝试使用 pthread 实现一个基本的工作池。 场景是我想要固定数量的工人，他们在我的项目期间一直存在。

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

在主程序继续之前，我需要等待所有线程完成，所以我决定在每个工作线程中使用 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 被锁定，它将保持锁定状态，直到线程函数到达 _unlock(); _等待（）;一对。只有在那之后，下一个单线程才能运行并到达它的屏障。清洗、冲洗、重复……

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 会更安全。

Answer 2

从问题中不清楚输入数据是什么，以及它们与线程和结果的关系。我无法从发布的代码中看出，因为我看不到实际工作应该在哪里完成。

假设您有 N（子）任务、N 个线程并希望主线程等待 N 个结果：您不需要真的需要一个屏障，你可以这样做：

• 主线程
  1. 在输入队列中推送 N 个任务
  2. 等到您在输出队列中收到 N 个结果
• 工作线程
  1. 从输入队列中弹出一个任务
  2. 计算结果
  3. 将结果推送到输出队列

最简单的同步队列只会一次推送/弹出一个项目（如果队列为空则推送信号，如果队列为空则弹出等待，仅此而已）。

作为优化，您可以轻松添加类似push_n(vector<task> const &input) 广播的pop_n(int count, vector<result> &output) 等待所有count 结果的内容，但基本模式是相同的。

Answer 3

你让自己变得比你必须做的更难。摆脱障碍。

如果您想等到所有工作都完成后，只需计算还有多少工作要做。用互斥锁保护它。使用条件变量触发它。然后只需使用pthread_cond_wait 等待它达到零。 （您可以使用已用于处理作业队列的相同逻辑。）

或者，将线程编码为在它们没有更多工作要做时终止。然后等到所有工作线程都以pthread_join 终止。