Производитель потребитель застрял в тупике с потребителем

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
}

Аналогично, вам нужночтобы включить флаг eos внутри мьютекса в producer_f().

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) для большей читаемости.