一个消费者和两个生产者中的任何一个答案

【问题标题】：One consumer and any of two producers一个消费者和两个生产者中的任何一个
【发布时间】：2015-05-28 04:49:46
【问题描述】：

我在编写这个 C 代码时遇到了麻烦，因为我需要有两个生产者有自己的缓冲区。然后一个消费者从任一缓冲区消费一项，并且仅在两个缓冲区都为空时才等待。如果任一生产者缓冲区为空，我知道如何编写消费者是否等待的代码，

sem_wait(Xfull);
item1 = BufferX[outX];
sem_signal(Xempty);

sem_wait(Yfull);
item2 = BufferB[outY];
sem_signal(BufferBEmpty);

但我坚持这一点。

【问题讨论】：

你写的行是不是用同样的方法和顺序？

标签： c multithreading semaphore producer-consumer

【解决方案1】：

您可以有一个信号量来计算两个队列中未使用项目的总数。生产者发布到它，消费者等待它。一旦消费者成功等待信号量，它就会使用互斥锁检查每个队列以找到要消费的项目。

sem_t ProducedItems;  // counts total # of unconsumed items in Q's A and B
sem_t Q_A;            // mutex on Q A
sem_t Q_B;            // mutex on Q B

void *consumer(void *arg)
{
  int bias = 0;  // NOTE: could use to not always favor reading from Q A over Q B

  while (1)
  {
    // wait for an item to be produced
    sem_wait(&ProducedItems);

    // find an item to consume
    do 
    {
      sem_wait(&Q_A);
      {
        if (Q A not empty) 
        {
          remove an item from Q A;
          bias = 1;
        }
      }
      sem_post(&Q_A);

      if (got item from A)
        break;

      sem_wait(&Q_B);
      {
        remove an item from Q B;
        bias = 0;
      }
      sem_post(&Q_B);
    } 
    while (0);

    // consume item
  }

  return NULL;
}

如果消费者还需要向生产者发回信号以确保他们的 Q 不会变得太大，那么您还可以为每个生产者设置信号量，指示他们的 Q 中还剩下多少空间。生产者会等待它的信号量，而消费者会在他们从各自的 Q 中删除项目时向其发布。

如果您想将其推广到 N 个生产 Q，那么您可能希望保留一个受互斥体保护的位掩码，以指示哪些 Q 上有未消费的项目。

一旦消费者获得 ProducedItems 信号量，它就会锁定掩码以查看哪些 Q 上有未消费的项目，并对它们进行循环处理以确定下一次应该从哪个 Q 读取。然后它会从那个 Q 中锁定 + 抓取一个项目，如果 Q 变为空，则更新位掩码，释放互斥体并最终使用该项目。

生产者会将项目放入他们的 Q 中，如果它是空的，那么他们还将锁定全局位掩码并将其 Q 位设置在那里，释放互斥体，然后发布 ProducedItems 信号量。

或者不是一个位掩码，你可以有一个 Q 大小的数组，尽管这需要生产者在每次生产时锁定互斥锁，等等。

【讨论】：

【解决方案2】：

一种选择是在两个信号量中实现忙等待：

int sem;
do {
    if (sem_trywait(sem1) == 0) {
        sem = 1;
        break;
    }
    if (sem_trywait(sem2) == 0) {
        sem = 2;
        break;
    }

    usleep(1);
} while (1);

if (sem == 1) {
    ...
}

我在this 答案中找到的另一个选项是查看sun 对WaitForMultipleObjects api 的实现，我引用：

基本思想是将条件变量列表与句柄（受互斥体保护），并发出所有条件的信号每当句柄发出信号时变量。每次调用模拟 WaitForMultipleObjects，创建一个新的条件变量并添加到您感兴趣的所有句柄的列表中。在 WaitForMultipleObjects 仿真，你阻塞条件变量，并在您醒来时检查您的每个把手。

论文可以在here找到。

【讨论】：

【解决方案3】：

条件变量适用于这种情况。

如果我正确理解您的问题。你有一个消费者消费两个生产者。因此，您可以

pthread_cond_t *cond;
pthread_mutex_t *cond_mutex;
sem_t *x, *y;

在消费者中：

while(exit_cond) {
    pthread_cond_wait(cond, cond_mutex);
    if(sem_trywait(x) == 0)
        consumeX(); // then post x
    if(sem_trywait(y) == 0)
        consumeY(); // then post y
}

在两个生产者中：

while(exit_cond) {
    produce(); //  hold and release semaphore
    pthread_cond_signal(cond);
}

通过这种方式，所有消费者和生产者都可以尽可能并行工作。

【讨论】：