pthread 的生产者-消费者问题

Question

我试图使用 pthread 和信号量解决生产者-消费者问题，但看起来生产者线程没有生产，消费者线程也没有消费。看起来正在创建线程：

  /* Do actual work from this point forward */
  /* Create the producer threads */
  for(c1=1; c1<=argarray[1]; c1++)
  {
    pthread_create(&tid, &attr, producer, NULL);
    printf("Creating producer #%d\n", c1);    
  }

  /* Create the consumer threads */
  for(c1=1; c1<=argarray[2]; c1++)
  {
    pthread_create(&tid, &attr, consumer, NULL);
    printf("Creating consumer #%d\n", c1);    
  }

因为“Creating Producer #x”和“Creating Consumer #x”被打印到屏幕上。然而，它不会从线程本身内部打印：

if(insert_item(item))
{
  fprintf(stderr, "Producer error.");
}
else
{
  printf("Producer produced %d\n", item);
}

对于消费者线程来说也是如此。完整代码：

#include "buffer.h"
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <semaphore.h>

/* Create Buffer */
buffer_item buffer[BUFFER_SIZE];

/* Semaphore and Mutex lock */
sem_t cEmpty;
sem_t cFull;
pthread_mutex_t mutex;

/* Threads */
pthread_t tid; /* Thread ID */
pthread_attr_t attr; /* Thread attributes */

void *producer(void *param);
void *consumer(void *param);
void init();

/* Progress Counter */
int cg;

main(int argc, char *argv[])
{
  /* Variables */
  int argarray[3], c1;

  /* Argument counter checks */
  if(argc != 4)
  {
    fprintf(stderr, "usage: main [sleep time] [# of producer threads] [# of consumer threads]\n");
    return -1;
  }

  /* Get args from command line and change them into integers */
  argarray[0] = atoi(argv[1]); /* How long to sleep before ending */
  argarray[1] = atoi(argv[2]); /* Producer threads */
  argarray[2] = atoi(argv[3]); /* Consumer threads */

  /* Error check */
  if(argarray[1]<1)
  {
    fprintf(stderr, "argument 2 must be > 0\n");
    return -1;
  }
  if(argarray[2]<1)
  {
    fprintf(stderr, "argument 3 must be > 0\n");
    return -1;
  }    

  init();

  /* Do actual work from this point forward */
  /* Create the producer threads */
  for(c1=1; c1<=argarray[1]; c1++)
  {
    pthread_create(&tid, &attr, producer, NULL);
    printf("Creating producer #%d\n", c1);    
  }

  /* Create the consumer threads */
  for(c1=1; c1<=argarray[2]; c1++)
  {
    pthread_create(&tid, &attr, consumer, NULL);
    printf("Creating consumer #%d\n", c1);    
  }

  /* Ending it */
  sleep(argarray[0]);

  printf("Production complete.\n");
  exit(0);
}

void init()
{
  int c2;

  pthread_mutex_init(&mutex, NULL); /* Initialize mutex lock */
  pthread_attr_init(&attr); /* Initialize pthread attributes to default */
  sem_init(&cFull, 0, 0); /* Initialize full semaphore */
  sem_init(&cEmpty, 0, BUFFER_SIZE); /* Initialize empty semaphore */
  cg = 0; /* Initialize global counter */ 
  for(c2=0;c2<BUFFER_SIZE;c2++) /* Initialize buffer */
  {
    buffer[c2] = 0;
  }
}

void *producer(void *param)
{
  /* Variables */
  buffer_item item;

  while(1)
  { 
    sleep(rand());      
    item = (rand()); /* Generates random item */ 

    sem_wait(&cEmpty); /* Lock empty semaphore if not zero */
    pthread_mutex_lock(&mutex);

    if(insert_item(item))
    {
      fprintf(stderr, "Producer error."); 
    }
    else
    {
      printf("Producer produced %d\n", item); 
    }

    pthread_mutex_unlock(&mutex);
    sem_post(&cFull); /* Increment semaphore for # of full */
  }
}

void *consumer(void *param)
{
  buffer_item item;

  while(1)
  {
    sleep(rand());
    sem_wait(&cFull); /* Lock empty semaphore if not zero */
    pthread_mutex_lock(&mutex);
    if(remove_item(&item))
    {
      fprintf(stderr, "Consumer error."); 
    }
    else
    {
      printf("Consumer consumed %d\n", item);
    }

    pthread_mutex_unlock(&mutex);
    sem_post(&cEmpty); /* Increments semaphore for # of empty */
  }
}

int insert_item(buffer_item item)
{
  if(cg < BUFFER_SIZE) /* Buffer has space */
  {
    buffer[cg] = item;
    cg++;
    return 0;
  }
  else /* Buffer full */
  {
    return -1;
  }
}

int remove_item(buffer_item *item)
{
  if(cg > 0) /* Buffer has something in it */
  {
    *item = buffer[(cg-1)];
    cg--;
    return 0;
  }
  else /* Buffer empty */
  {
    return -1;
  }
}

终端输出：

user@isanacom:~/Desktop/PCthreads$ ./main 10 10 10
Creating producer #1
Creating producer #2
Creating producer #3
Creating producer #4
Creating producer #5
Creating producer #6
Creating producer #7
Creating producer #8
Creating producer #9
Creating producer #10
Creating consumer #1
Creating consumer #2
Creating consumer #3
Creating consumer #4
Creating consumer #5
Creating consumer #6
Creating consumer #7
Creating consumer #8
Creating consumer #9
Creating consumer #10
Production complete.

作为多线程的初学者，我确信它可能是我忽略的简单内容，并且我很感谢您的帮助。

Answer 1

消费者和生产者都会执行 sleep(rand()) ，它将休眠 0 到 MAX_INT 之间的随机秒数，在您给出的示例中，主线程将在 10 秒后终止。如果生产者的 rand() 值高于 10，他们将永远没有机会生产任何东西。

Answer 2

您应该从使用线程数组开始...

pthread_t tid[argarray[1] + argarray[2]];

for(c1 = 0; c1 < argarray[1]; c1++)
{
    pthread_create(&tid[c1], &attr, producer, NULL);
    printf("Creating producer #%d\n", c1); 
}
for(c1 = 0; c1 < argarray[2]; c1++)
{
    pthread_create(&tid[c1 + argarray[1]], &attr, consumer, NULL);
    printf("Creating consumer #%d\n", c1); 
}

可能还有其他问题，但这是我看到的第一个...

Answer 3

您不应该在主函数末尾直接调用 exit。您应该首先为最后创建的线程调用 pthread_join ，以使主线程保持活动状态，直到其他线程死亡。