c - 为 pthread_cond_broadcast 调用同步线程

Question

我有一个带有“管理器”线程的简单应用程序，它产生十个简单的“工作器”线程。我希望所有“工作”线程都阻塞在同一个条件变量(即:condvar)上，并且我希望通过 pthread_cond_broadcast 调用手动通知所有十个线程同时唤醒。

在我的应用程序的情况下，线程可能会遇到错误条件并提前终止，因此可能并非所有 10 个线程都到达同步点。

一种简单的机制是创建一个 pthread_barrier_t 并让所有 10 个线程调用 pthread_barrier_wait，当所有 10 个线程完成此调用时，它们都可以自由地继续执行。但是，这将要求线程能够修改屏障解除阻塞所需的线程数。我不知道这是否可以安全地修改。

此外，我想保证所有仍在工作的线程不会像使用屏障那样自动启动，我想用 pthread_cond_broadcast 调用手动启动它们。在我进行广播调用之前，我如何保证所有仍然存活的线程(最好是 10 个)在 condvar 上被阻塞？

谢谢!

Answer 1

下面显示了一种方法，使用条件变量和其他一些变量；虽然可能有更好的方法。评论应该显示它是如何工作的。当然，您必须修改内容以适合您的实际情况；例如可能涉及循环等。

int activeThreads = 0;   /* number of threads currently going */
int waitingThreads = 0;  /* number of threads waiting on the condition */
int readyFlag = 0;       /* flag to tell the threads to proceed when signaled */
pthread_cond_t  cond;    /* condition to wait on / signal */
pthread_mutex_t mtx;     /* mutex for the above */

pthread_cond_t  condWaiting; /* EDIT: additional condition variable to signal 
                              * when each thread starts waiting */

void *threadFunc(void *arg)
{
  /* Edit: Rather than incrementing 'activeThreads' here, it should be done
   * in the main thread when each thread is created (to avoid a race) */

  /* ...do stuff... */

  /* When the threads should wait, do this (they wait for 'readyFlag' to be 
   * set, but also adjust the waiting thread count so the main thread can
   * determine whether to broadcast) */
  pthread_mutex_lock(&mtx);
    if (readyFlag == 0) {
      waitingThreads++;
      do {
        pthread_cond_signal(&condWaiting); /* EDIT: signal the main thread when
                                            * a thread begins waiting */
        pthread_cond_wait(&cond,&mtx);
      } while (readyFlag == 0);
      waitingThreads--;
    }
  pthread_mutex_unlock(&mtx);

  /* ...more stuff... */

  /* When threads terminate, they decrement the active thread count */
  pthread_mutex_lock(&mtx);
    activeThreads--;
    pthread_cond_signal(&condWaiting); /* EDIT: also signal the main thread
                                        * when a thread exits to make it 
                                        * recheck the waiting thread count if
                                        * waiting for all threads to wait */
  pthread_mutex_unlock(&mtx);

  return NULL;
}

int main(int argc, char *argv[])
{
  /* Edit: Showing some code to initialize the mutex, condition variable(s),
   * etc., and create some threads -- modify as needed */
  pthread_mutex_init(&mtx,NULL);
  pthread_cond_init(&cond,NULL);
  pthread_cond_init(&condWaiting,NULL); /* EDIT: if main thread should block
                                         * until all threads are waiting */
  activeThreads = waitingThreads = readyFlag = 0;

  /* Edit: Increment 'activeThreads' here rather than in the thread function,
   * to avoid a race (if the main thread started waiting for the others
   * when not all had incremented the count yet, the main thread might end
   * up waiting for fewer threads to be ready -- though it's unlikely */
  #define NUM_THREADS 10
  pthread_t workers[NUM_THREADS];
  for (int i = 0; i < NUM_THREADS; i++) {
    /* should use appropriate thread attributes */
    if (pthread_create(&workers[i],NULL,threadFunc,NULL) == 0)
      activeThreads++;
  }

  /* ...do stuff... */

  /* Set 'readyFlag' and do condition broadcast IF all threads are waiting,
   * or just carry on if they aren't */
  pthread_mutex_lock(&mtx);
    if ((activeThreads != 0) && (activeThreads == waitingThreads)) {
      readyFlag = 1;
      pthread_cond_broadcast(&cond);
    }
  pthread_mutex_unlock(&mtx);

  /* EDIT: OR.. to wait until all threads are waiting and then broadcast, do 
   * this instead: */
  pthread_mutex_lock(&mtx);
    while (waitingThreads < activeThreads) { /* wait on 'condWaiting' until all
                                              * active threads are waiting */
      pthread_cond_wait(&condWaiting,&mtx);
    }
    if (waitingThreads != 0) {
      readyFlag = 1;
      pthread_cond_broadcast(&cond);
    }
  pthread_mutex_unlock(&mtx);

  /* ...more stuff... */

  /* If needed, you can clear the flag when NO threads are waiting.. */
  pthread_mutex_lock(&mtx);
    if (waitingThreads == 0)
      readyFlag = 0;
  pthread_mutex_unlock(&mtx);

  /* ...even more stuff... */

  return 0;
}

不过，我想补充一点，我不知道何时有充分的理由这样做，而不仅仅是以更直接的方式保护资源。

编辑:在代码中添加了一些内容，显示了第二个条件变量，用于让主线程等待所有工作线程准备就绪。更改的部分在注释中标有“编辑:”，如果不需要，可以省略。我还通过移动 activeThreads 的增量更正了竞争条件。在线程函数之外，并显示了互斥锁等的初始化(没有错误处理)。