生产者 消费者与消费者陷入僵局

Question

我正在尝试使用条件变量来实现生产者消费者，以便了解同步。我正在使用github 来指导我并解决了一些段错误，但现在看来我的消费者从未被执行或陷入僵局。我不确定可能是什么错误。我在生产者中包含了 printf 语句，以在每次运行时执行，并且它完成了在 messages.txt 中生成小于 5 的任何字符串。消费者虽然没有并且陷入死锁。

 #define max 5

int par = 0;

// Data structure for queue
struct queue
{
    char *items;        // array to store queue elements
    int maxsize;    // maximum capacity of the queue
    int front;      // front points to front element in the queue (if any)
    int rear;       // rear points to last element in the queue
    int size;       // current capacity of the queue
    pthread_mutex_t mutex; // needed to add/remove data from the buffer
    pthread_cond_t can_produce; // signaled when items are removed
    pthread_cond_t can_consume; // signaled when items are added
};

// Utility function to initialize queue
struct queue* newQueue(int size)
{
    struct queue *pt = NULL;
    pt = (struct queue*)malloc(sizeof(struct queue));

    pt->items = (char*)malloc(size * sizeof(char));
    pt->maxsize = size;
    pt->front = 0;
    pt->rear = -1;
    pt->size = 0;
    pthread_mutex_init(&pt->mutex, NULL);
    pthread_cond_init(&pt->can_produce, NULL);
     pthread_cond_init(&pt->can_consume, NULL);
    return pt;
}

// Utility function to return the size of the queue
int size(struct queue *pt)
{
    return pt->size;
}

// Utility function to check if the queue is empty or not
int isEmpty(struct queue *pt)
{
    return !size(pt);
}

// Utility function to return front element in queue
char front(struct queue *pt)
{
    if (isEmpty(pt))
    {
        //printf("UnderFlow\nProgram Terminated\n");

    }

    return pt->items[pt->front];
}

// Utility function to add an element x in the queue
void enqueue(struct queue *pt, char x)
{
    if (size(pt) == pt->maxsize)
    {
        //printf("OverFlow\nProgram Terminated\n");

    }

    //printf("Inserting %c\t", x);

    pt->rear = (pt->rear + 1) % pt->maxsize;    // circular queue
    pt->items[pt->rear] = x;
    pt->size++;

    //printf("front = %c, rear = %c\n", pt->front, pt->rear);
}

// Utility function to remove element from the queue
void dequeue(struct queue *pt)
{
    if (isEmpty(pt)) // front == rear
    {
        //printf("UnderFlow\nProgram Terminated\n");

    }

    //printf("Removing  %c\t", front(pt));

    pt->front = (pt->front + 1) % pt->maxsize;  // circular queue
    pt->size--;

    //printf("front = %d, rear = %c\n", pt->front, pt->rear);
}

void consumer_f(void *arg)
{
    struct queue *pt = (struct queue*)arg;
     while (par==0 && !isEmpty(pt))
    {
        pthread_mutex_lock(&pt->mutex);
        if (pt->size == 0)
        { // empty
            // wait for new items to be appended to the buffer
            pthread_cond_wait(&pt->can_consume, &pt->mutex);
        }

        printf("%c", pt->front);
        dequeue(pt);
        pthread_cond_signal(&pt->can_produce);
        pthread_mutex_unlock(&pt->mutex);
    } 
}
void producer_f(void *arg)
{
    struct queue *pt = (struct queue*)arg;
     char tmp;
    FILE *fp;
    fp = fopen("messages.txt", "r");
    if (fp == NULL)
    {
        //fprintf(stderr, "error opening messages.txt");
        return -1;
    }
    while ((tmp = fgetc(fp)) != EOF)
    {
        pthread_mutex_lock(&pt->mutex);
        if (pt->size == max)
            pthread_cond_wait(&pt->can_produce, &pt->mutex);
        enqueue(pt, tmp);
        printf("sent");
        pthread_cond_signal(&pt->can_consume);
        pthread_mutex_unlock(&pt->mutex);
    }
    par = 1; //denotes EOF for consumer */
}
int main()
{
    printf("nop");
    struct queue *pt = newQueue(5);


    pthread_t producer;
    pthread_t consumer;
    printf("got here");
    if (pthread_create(&producer, NULL, &producer_f, pt))
    {
        //fprintf(stderr, "Error creating producer thread\n");
        return -1;
    }
      if (pthread_create(&consumer, NULL, &consumer_f, pt))
    {
        //fprintf(stderr, "Error creating consumer thread\n");
        return -1;
    } 
     if (pthread_join(producer_f, NULL))
    {
        //fprintf(stderr, "Error joining proucer thread\n");
        return -1;
    } 
     if (pthread_join(consumer_f, NULL))
    {
        //fprintf(stderr, "Error joinging consumer thread\n");
        return -1;
    }  

    return 0;
}

Answer 1

消费者线程不会进入死锁状态，但它会在不消费的情况下退出，因为在消费者开始消费之前已达到 EOS。

如您所知，线程可以由操作系统以随机方式调度（我的意思是，至少您可以假设它们以随机方式调度）。有了这个假设，生产者可能已经启动并读取了所有字节并启用了 eos 标志，即par=1。如果消费者线程在par=1 之后启动，则根本不消费。

要处理这种情况，您需要更新consumer_f() 函数。

while (1) //Run the loop always
{
    pthread_mutex_lock(&pt->mutex);
    if ((pt->size == 0) && (par == 0)) //Make sure that EOS is not already reached.
    { // empty
        // wait for new items to be appended to the buffer
        pthread_cond_wait(&pt->can_consume, &pt->mutex);
    }

    if(par && isEmpty(pt))
    {
         //If Array is empty and eos is reached, unlock and exit loop
         pthread_mutex_lock(&pt->mutex);
         break;
    }

    //Other code
}

同样，您需要在 producer_f() 的 mutex 中启用 eos 标志。

while ((tmp = fgetc(fp)) != EOF)
{
    //Your code
}

pthread_mutex_lock(&pt->mutex);
par = 1;
pthread_cond_signal(&pt->can_consume); //To make sure that consumer thread wakesup
pthread_mutex_unlock(&pt->mutex);

PS：pt->size == 0 可以替换为isEmpty(pt) 以提高可读性。