Thread: Pthread Circular buffer not looping, need help!

I'm designing a circular buffer that reads a input text file and writes to output.

It works by creating a pthread named Lift-R and calls request_t() to load values onto the buffer. When the buffer is full, it goes to the three other created pthreads lift-1 to lift-3 which all call lift(). The problem is that once lift-3 is done, the code suddenly becomes stuck a waiting deadlock for another lift thread (which there isn't).

The code should return to lift-1 once lift-3 is done and continue until text file is finished.

How can I get the code to cycle through all the pthreads that call lift() until the input text file is empty.

Attempts I've made was to add a second mutex lock to prevent collisions. I've also tried arranging pthread_join() to no avail.

I've tried asking stack overflow, but you know what they're like.

Code:
header file:

Code:

    #include <stdbool.h>
    
    typedef struct Buffer
    {
        int from;
        int to;
    }buffer_t; //buffer arrary to store from and to values from sim_input
    
    typedef struct Output
    {
        int p; //previous floor
        int rf; //request from
        int rt; //request to
        int total_m[3]; //total movement
        int c; // current position
        int m; //movement
        int total_r[3]; //total requests
    }write_t;
    typedef struct shared_memory
    {
        void* in;
        void* out;
        int id; //name of thread (more relieable than pthread_self())
    }share_t; //Shared memory to be used across functions (Allow multiple arguements in pthread_create)

main file:

Code:

    #include <stdio.h>
    #include <stdlib.h>
    #include <pthread.h> 
    #include <unistd.h>
    #include <assert.h>
    #include "list.h"
    pthread_cond_t full, empty; //declare thread conditions
    pthread_mutex_t lock, lock2; //declare mutex to lock functions
    int counter; //number of requests made
    static int size; //buffer size (must only change once at beginning)
    static int sec; //Time in seconds, needed for lift to execute operation
    int length; //Number of values in buffer
    unsigned int in, out_i; //tail and head of buffer
    buffer_t *A; //the circular queue, is initialized in main due to size declared by user input
    write_t write1; //Used for printing the information from buffer.
    int id; //Lift id to be printed in output
    void *lift(void *vargp) //Consumer
    {
            pthread_mutex_lock(&lock2);
            while (length <= 0)
            {
                printf("Waiting for request\n");
                pthread_cond_wait(&empty, &lock2); //buffer is empty
            }
            sleep(sec); //Simulate time of elevator movement
            printf("lifting information\n");
            id++;
            //gather information to print
            if (write1.p == 0) //only for when system begins (Assign Previous to the first record `from` value)
            {
                write1.p = A[out_i].from;
            }
            write1.rf = A[out_i].from;
            write1.rt = A[out_i].to;
            write1.m = (write1.p - A[out_i].from) + (A[out_i].to - A[out_i].from);
            write1.c = A[out_i].to;
            //Now write the information
            fprintf(((share_t*)vargp)->out, "Lift-%d Operation\n", id);
            fprintf(((share_t*)vargp)->out, "Previous position: Floor %d\n", write1.p);
            fprintf(((share_t*)vargp)->out, "Request: Floor %d to Floor %d\n", write1.rf, write1.rt);
            fprintf(((share_t*)vargp)->out, "Detail operations:\n");
            fprintf(((share_t*)vargp)->out, "    Go from Floor %d to Floor %d\n", write1.p, write1.rf);
            fprintf(((share_t*)vargp)->out, "    Go from Floor %d to Floor %d\n", write1.rf, write1.rt);
            fprintf(((share_t*)vargp)->out, "    #movement for this request: %d\n", write1.m);
            write1.total_r[id] = write1.total_r[id] + 1;
            fprintf(((share_t*)vargp)->out, "    #request: %d\n", write1.total_r[id]);
            write1.total_m[id] = write1.total_m[id] + 1;
            fprintf(((share_t*)vargp)->out, "    Total #movement: %d\n", write1.total_m[id]);
            fprintf(((share_t*)vargp)->out, "Current Position: Floor %d\n", write1.c);
            fprintf(((share_t*)vargp)->out, "\n");
            fprintf(((share_t*)vargp)->out, "----------------------------\n");
            write1.p = write1.c; //for next statement
            length--;
            out_i = (out_i + 1) % size;
            if (id < 3)
            {
                id = 0;
            }
            if (length == size-1)
            {
                printf("Signalling Request\n");
                pthread_cond_signal(&full);
            }
        pthread_mutex_unlock(&lock2);
        return NULL;
    }
    void *request_t(void *vargp) //producer
    {
        pthread_mutex_lock(&lock); //Now only request can operate (mutual exclusion)
        //read the input line by line and into the buffer
        while (fscanf(((share_t*)vargp)->in, "%d %d\n", &A[in].from, &A[in].to) != EOF) //Stores values in buffer
        {
            while (length >= size) //buffer is full
            {
                printf("Waiting for lift\n");
                pthread_cond_wait(&full, &lock);
            }
            printf("Requesting information\n");
            //If requests go out of bounds, they will be automatically registered as either
            //maximum level or minimum level
            if (A[in].to > 20)
            {
                A[in].to = 20;
            }
            else if (A[in].to < 1)
                {
                    A[in].to = 1;
                }
            else if (A[in].from > 20)
                {
                    A[in].from = 20;
                }
            else if (A[in].from < 1)
                {
                    A[in].from = 1;
                }
            //Print buffer information to sim_output
            fprintf(((share_t*)vargp)->out, "New Lift Request from Floor %d to Floor %d \n", A[in].from, A[in].to);
            fprintf(((share_t*)vargp)->out, "Request No %d \n", counter++);
            fprintf(((share_t*)vargp)->out, "----------------------------\n");
            length++;
            in = (in + 1) % size;
            if (length == 1)
            {
                printf("Signalling Lift\n");
                pthread_cond_signal(&empty);
            }
        }
        pthread_mutex_unlock(&lock);
        pthread_exit(NULL);
    }
    void main(int argc, char *argv[]) // to avoid segmentation fault
    {
        //establish values for use (convert argv to int values)
        counter = 1;
        share_t *share = (share_t *)malloc(sizeof(share_t)); //Allow multiple arguments in pthread
        in = 0, out_i = 0, id = 0, length = 0; //All slots are empty
        long arg = strtol(argv[1], NULL, 10);
        size = arg;
        if (size <= 1)
        {
            printf("buffer size too small\n");
            exit(0);
        }
        else
        {
            //initialize Buffer as empty array.
            A = calloc(size, sizeof *A);
        }
        arg = strtol(argv[2], NULL, 10);
        sec = arg; //time required by each lift to serve a request
        if (sec <= 0)
        {
            printf("insufficient time\n");
            exit(0);
        }
        FILE *out = fopen("sim_output.txt", "w");
        FILE *in = fopen("sim_input.txt", "r");
        if (in == NULL && out == NULL) //Check if files exists before we can use them
        {
            printf("Input file empty\n");
            exit(0);
        }
        else //Load files into share for transporting.
        {
            share->out = out;
            share->in = in;
        }
        //Create threads
        printf("Creating threads\n");
        pthread_t Lift_R, lift_1, lift_2, lift_3;
        assert(pthread_cond_init(&full, NULL) == 0);
        assert(pthread_cond_init(&empty, NULL) == 0);
        assert(pthread_mutex_init(&lock, NULL) == 0);
        assert(pthread_mutex_init(&lock2, NULL) == 0);
        assert(pthread_create(&Lift_R, NULL, request_t, share) == 0);
        share->id = 1;
        assert(pthread_create(&lift_1, NULL, lift, share) == 0);
        share->id = 2;
        assert(pthread_create(&lift_2, NULL, lift, share) == 0);
        share->id = 3;
        assert(pthread_create(&lift_3, NULL, lift, share) == 0);
        assert(pthread_join(Lift_R, NULL) == 0);
        assert(pthread_join(lift_1, NULL) == 0);
        assert(pthread_join(lift_2, NULL) == 0);
        assert(pthread_join(lift_3, NULL) == 0);
        //pthread_mutex_destroy(NULL);
        int sum_r = write1.total_r[1] + write1.total_r[2] + write1.total_r[3],
            sum_t = write1.total_m[1] + write1.total_m[2] + write1.total_m[3];
        fprintf(out, "\n");
        fprintf(out, "Total number of requests %d \n", sum_r);
        fprintf(out, "Total number of movememts %d \n", sum_t);
        fclose(in);
        fclose(out);
    }

input for the code looks like this:
1 10
4 20
5 14
6 7
8 1
4 5
10 14
16 13

It is compiled with -lpthread and to run the code, use: lift_sim_A m t.(m is size and t is seconds).

when I test this code, I usually use lift_sim_A 3 1. I'll try to cut the code down later to make it easier to read.

I created share so that I can have multiple arguments inside the pthreads which is important for the FILE* in and out. I may try to create a local pointer for id.

I've changed mutex to only one lock. Both now have the same mutex. I'm now getting errors for fscanf on line 71. It's reporting too many arguements.

This is what I've got so far:

Code:

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h> 
#include <unistd.h>
#include <assert.h>
#include "list.h"
pthread_cond_t full, empty; //declare thread conditions
pthread_mutex_t lock; //declare mutex to lock functions
int counter; //number of requests made
static int size; //buffer size (must only change once at beginning)
static int sec; //Time in seconds, needed for lift to execute operation
int length; //Number of values in buffer
unsigned int in_i, out_i; //tail and head of buffer
buffer_t *A; //the circular queue, is initialized in main due to size declared by user input
write_t write1; //Used for printing the information from buffer.
void *lift(void *vargp) //Consumer
{
    pthread_mutex_lock(&lock);
    while (length <= 0)
    {
        printf("Waiting for request\n");
        pthread_cond_wait(&empty, &lock); //buffer is empty
    }
    sleep(sec); //Simulate time of elevator movement
    printf("lifting information\n");
    //gather information to print
    if (write1.p == 0) //only for when system begins (Assign Previous to the first record `from` value)
    {
        write1.p = A[out_i].from;
    }
    write1.rf = A[out_i].from;
    write1.rt = A[out_i].to;
    write1.m = (write1.p - A[out_i].from) + (A[out_i].to - A[out_i].from);
    write1.c = A[out_i].to;
    int id = ((share_t*)vargp)->id;
    
    //Now write the information
    fprintf(((share_t*)vargp)->out, "Lift-%d Operation\n", id);
    fprintf(((share_t*)vargp)->out, "Previous position: Floor %d\n", write1.p);
    fprintf(((share_t*)vargp)->out, "Request: Floor %d to Floor %d\n", write1.rf, write1.rt);
    fprintf(((share_t*)vargp)->out, "Detail operations:\n");
    fprintf(((share_t*)vargp)->out, "    Go from Floor %d to Floor %d\n", write1.p, write1.rf);
    fprintf(((share_t*)vargp)->out, "    Go from Floor %d to Floor %d\n", write1.rf, write1.rt);
    fprintf(((share_t*)vargp)->out, "    #movement for this request: %d\n", write1.m);
    write1.total_r[id] = write1.total_r[id] + 1;
    fprintf(((share_t*)vargp)->out, "    #request: %d\n", write1.total_r[id]);
    write1.total_m[id] = write1.total_m[id] + 1;
    fprintf(((share_t*)vargp)->out, "    Total #movement: %d\n", write1.total_m[id]);
    fprintf(((share_t*)vargp)->out, "Current Position: Floor %d\n", write1.c);
    fprintf(((share_t*)vargp)->out, "\n");
    fprintf(((share_t*)vargp)->out, "----------------------------\n");
    write1.p = write1.c; //for next statement
    length--;
    out_i = (out_i + 1) % size;
    if (length == size-1)
    {
        printf("Signalling Request\n");
        pthread_cond_signal(&full);
    }
    pthread_mutex_unlock(&lock);
    return NULL;
}
void *request_t(void *vargp) //producer
{
    pthread_mutex_lock(&lock); //Now only request can operate (mutual exclusion)
    //read the input line by line and into the buffer
    while (fscanf(((share_t*)vargp)->in, "%d %d\n", &A[in_i].from, &A[in_i].to) != EOF) //Stores values in buffer
    {
        while (length >= size) //buffer is full
        {
            printf("Waiting for lift\n");
            pthread_cond_wait(&full, &lock);
        }
        printf("Requesting information\n");
        //If requests go out of bounds, they will be automatically registered as either
        //maximum level or minimum level
        if (A[in_i].to > 20)
        {
            A[in_i].to = 20;
        }
        else if (A[in_i].to < 1)
        {
            A[in_i].to = 1;
        }
        else if (A[in_i].from > 20)
        {
            A[in_i].from = 20;
        }
        else if (A[in_i].from < 1)
        {
            A[in_i].from = 1;
        }
        //Print buffer information to sim_output
        fprintf(((share_t*)vargp)->out, "New Lift Request from Floor %d to Floor %d \n", A[in_i].from, A[in_i].to);
        fprintf(((share_t*)vargp)->out, "Request No %d \n", counter++);
        fprintf(((share_t*)vargp)->out, "----------------------------\n");
        length++;
        in_i = (in_i + 1) % size;
        if (length == 1)
        {
            printf("Signalling Lift\n");
            pthread_cond_signal(&empty);
        }
    }
    pthread_mutex_unlock(&lock);
    pthread_exit(NULL);
}
void main(int argc, char *argv[]) // to avoid segmentation fault
{
    //establish values for use (convert argv to int values)
    counter = 1;
    share_t share[4]; //Allow multiple arguments in pthread
    in_i = 0, out_i = 0, length = 0; //All slots are empty
    long arg = strtol(argv[1], NULL, 10);
    size = arg;
    if (size <= 1)
    {
        printf("buffer size too small\n");
        exit(0);
    }
    else
    {
        //initialize Buffer as empty array.
        A = calloc(size, sizeof *A);
    }
    arg = strtol(argv[2], NULL, 10);
    sec = arg; //time required by each lift to serve a request
    if (sec <= 0)
    {
        printf("insufficient time\n");
        exit(0);
    }
    FILE *out_t = fopen("sim_output.txt", "w");
    FILE *in_t = fopen("sim_input.txt", "r");
    if (in_t == NULL && out_t == NULL) //Check if files exists before we can use them
    {
        printf("Input file empty\n");
        exit(0);
    }
    for (int i = 0; i < 3; i++)//Load files into share for transporting.
    {
        share[i].in = in_t;
        share[i].out = out_t;
        share[i].id = i;
    }
    //Create threads
    printf("Creating threads\n");
    pthread_t Lift_R, lift_1, lift_2, lift_3;
    assert(pthread_cond_init(&full, NULL) == 0);
    assert(pthread_cond_init(&empty, NULL) == 0);
    assert(pthread_mutex_init(&lock, NULL) == 0);
    assert(pthread_create(&Lift_R, NULL, request_t, &share[0]) == 0);
    assert(pthread_create(&lift_1, NULL, lift, &share[1]) == 0);
    assert(pthread_create(&lift_2, NULL, lift, &share[2]) == 0);
    assert(pthread_create(&lift_3, NULL, lift, &share[3]) == 0);
    assert(pthread_join(Lift_R, NULL) == 0);
    assert(pthread_join(lift_1, NULL) == 0);
    assert(pthread_join(lift_2, NULL) == 0);
    assert(pthread_join(lift_3, NULL) == 0);
    pthread_mutex_destroy(NULL);
    int sum_r = write1.total_r[1] + write1.total_r[2] + write1.total_r[3],
        sum_t = write1.total_m[1] + write1.total_m[2] + write1.total_m[3];
    fprintf(out_t, "\n");
    fprintf(out_t, "Total number of requests %d \n", sum_r);
    fprintf(out_t, "Total number of movememts %d \n", sum_t);
    fclose(in_t);
    fclose(out_t);
}