C中线程同步和条件变量的问题

Question

我有三个线程，一个是主线程，另外两个是工作线程。当有工作要做时，第一个线程会唤醒两个线程之一。每个线程在被唤醒时都会执行一些计算，并且在执行此操作时，如果它发现有更多工作要做，则可以唤醒其他工作线程或简单地决定自行完成该工作（例如，通过将工作添加到本地队列）。 当工作线程有工作要做时，主线程必须等待工作完成。我用条件变量实现了如下（这里报告的代码隐藏了很多细节，如果有什么难以理解的地方请询问）：

MAIN THREAD (pseudocode):

//this function can be called from the main several time. It blocks the main thread till the work is done.
void new_work(){

//signaling to worker threads if work is available

    //Now, the threads have been awakened, it's time to sleep till they have finished.
    pthread_mutex_lock(&main_lock);
    while (work > 0)    //work is a shared atomic integer, incremented each time there's work to do and decremented when finished executing some work unit
       pthread_cond_wait(&main_cond);
    pthread_mutex_unlock(&main_lock);

}

WORKER THREADS:

while (1){

   pthread_mutex_lock(&main_lock);
    if (work == 0)
       pthread_cond_signal(&main_cond);
    pthread_mutex_unlock(&main_lock);  

    //code to let the worker thread wait again -- PROBLEM!

   while (I have work to do, in my queue){
       do_work()
   }

}

这是问题：当一个工作线程唤醒主线程 我不确定工作线程是否调用 wait 将其自身置于等待新工作的状态。即使我使用另一个条件变量实现此等待，主线程也可能处于唤醒状态，执行一些工作，直到达到必须唤醒尚未调用等待的线程的程度...这可以导致不好的结果。我尝试了多种方法来解决这个问题，但我找不到解决方案，也许有一个明显的方法可以解决它，但我错过了。

您能提供解决此类问题的方案吗？我使用的是 C 语言，并且可以使用您认为适合的任何同步机制，例如 pthreads 或 posix 信号量。

谢谢

Answer 1

处理此问题的常用方法是使用单个工作队列并防止其溢出和下溢。像这样的东西（我省略了“pthread_”前缀）：

mutex queue_mutex;
cond_t queue_not_full, queue_not_empty;

void enqueue_work(Work w) {
    mutex_lock(&queue_mutex);
    while (queue_full())
        cond_wait(&queue_not_full, &queue_mutex);
    add_work_to_queue(w);
    cond_signal(&queue_not_empty);
    mutex_unlock(&queue_mutex);
}

Work dequeue_work() {
    mutex_lock(&queue_mutex);
    while (queue_empty())
        cond_wait(&queue_not_empty, &queue_mutex);
    Work w = remove_work_from_queue();
    cond_signal(&queue_not_full);
    mutex_unlock(&queue_mutex);
}

注意这些函数之间的对称性：enqueue <->出队，清空<->满，not_empty <->不满。

这为任意数量的产生工作的线程和任意数量的消耗工作的线程提供了一个线程安全的有界大小队列。 （实际上，这是使用条件变量的典型示例。）如果您的解决方案看起来与此不完全一样，那么它可能应该非常接近......

Answer 2

如果您希望主线程将工作分配给其他两个线程，然后等到两个线程都完成其工作后再继续，您也许可以通过屏障来完成此任务。

屏障是一种同步构造，您可以使用它使线程在代码中的某个点等待，直到一定数量的线程准备好继续前进。本质上，您初始化一个 pthread 屏障，表示 x 个线程必须等待它，然后才允许任何线程继续。当每个线程完成其工作并准备继续时，它将在屏障上等待，一旦x个线程到达屏障，它们都被允许继续。

在您的情况下，您可能可以执行以下操作：

pthread_barrier_t barrier;
pthread_barrier_init(&barrier, 3);

master()
{
  while (work_to_do) {
    put_work_on_worker_queues();
    pthread_barrier_wait(&barrier);
  }
}

worker()
{
  while(1) {
    while (work_on_my_queue()) {
      do_work();
    }
    pthread_barrier_wait(&barrier);
  }
}

这应该使您的主线程完成工作，然后等待两个工作线程完成它们所分配的工作（如果有），然后再继续。

Answer 3

你能有一个由主线程管理的“新作业”队列吗？主线程可以一次向每个工作线程分发 1 个作业。主线程还将侦听工作人员完成的作业。如果工作线程发现需要执行的新作业，只需将其添加到“新作业”队列中，主线程将分发它。

伪代码：

JobQueue NewJobs;
Job JobForWorker[NUM_WORKERS];

workerthread()
{
  while(wait for new job)
  {
    do job (this may include adding new jobs to NewJobs queue)
    signal job complete to main thread
  }
}

main thread()
{
  while(whatever)
  {
    wait for job completion on any worker thread
    now a worker thread is free put a new job on it
  }
}

Answer 4

我相信您所遇到的是生产者-消费者问题的变体。您正在做的是编写计数信号量的临时实现（用于提供的不仅仅是互斥）。

如果我正确地阅读了您的问题，那么您要做的就是让工作线程阻塞，直到有可用的工作单元，然后在工作单元可用后执行该工作单元。您的问题是有太多可用工作并且主线程尝试解锁已经在工作的工作人员的情况。我将按如下方式构建您的代码。

sem_t main_sem;
sem_init(&main_sem, 0, 0);

void new_work() {
    sem_post(&main_sem);
    pthread_cond_wait(&main_cond);
}

void do_work() {
    while (1) {
        sem_wait(&main_sem);
        // do stuff
        // do more stuff
        pthread_cond_signal(&main_sem);
    }
}

现在，如果工作线程生成更多工作，那么它们可以简单地将 sem_post 发送到信号量，并简单地推迟 pthread_cond_signal 直到所有工作完成。

但请注意，如果您实际上需要主线程在工作线程工作时始终阻塞，那么当您只需调用执行该工作的函数时，将工作推送到另一个线程是没有用的。