Circular Queue Producer Consumer with Mutex and Condition Variable vs. Sempahore

Question

How can I optimize the performance of the code below? Destruction of the condition variable at the end of main blocks, not sure why this behavior is occurring. Is it possible to remove some of the sleep statements and while loops? What are other general best practices for proper implementation of this type of program. Is there a more accurate way to measure the runtime of processing various amount of elements in the queue/buffer?

/* main.c - main */
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <pthread.h> 
#include <sched.h>
#include <sys/resource.h>
#include <stdbool.h>
//Declaring constant values for use throughout your code
#define ARR_SIZE 250
#define RESTRICT 1700
pthread_t producer_id;
pthread_t consumer_id;
pthread_t timer_id;
int consumed_count = 0;
const int CONSUMED_MAX = 100;
//Declaring circular buffer structure here
int myQueue[ARR_SIZE];
int rear=0, front=0, data=-1, d=1;
//Declaring thread sync variables here
pthread_mutex_t m;
pthread_cond_t cv;
bool ready = false;
//Defining void to insert integer element in queue
void insertIntoQ(int n){
 myQueue[rear] = n; //insert integer element at rear position
 rear = (rear + 1) % ARR_SIZE; //increment rear by 1 
 //Ensuring it points to 1st loc when it reaches buffer limit
}
//Defining void to remove integer element in queue
int removeFromQ(){
 data = myQueue[front]; //remove integer element from front position
 front = (front + 1) % ARR_SIZE; //increment front by 1 
 //Ensuring it points to 1st loc when it reaches buffer limit
 return data;
}
/* Code for the buffer producer*/
void* producer(void *args){
 while(d<=RESTRICT) { //Continue producing till certain limit
 pthread_mutex_lock(&m); //acquire the mutex
 insertIntoQ(d); //insert element
 printf("Produced: %d \n",d); //print to console
 d += 1; // increment 'd'
 ready = true;
 pthread_mutex_unlock(&m); //release the mutex
 pthread_cond_signal(&cv);
 usleep(1);
 } 
}
/* Place the code for the buffer consumer here */
void* consumer(void *args){
 int j=1;
 int x,y;
 while(j<=RESTRICT){ //Continue consuming till certain limit
 pthread_mutex_lock(&m); //acquire the mutex
 while(!ready){
 pthread_cond_wait(&cv, &m);
 // printf("consumer waiting\n");
 }
 x = removeFromQ(); //delete element
 y = front; // print loc of element dequeued
 printf("Is %d power of 2 at buffer location %d : %s \n",x,y,(x && (!(x&(x-1))))?"Yes":"No"); //print to console
 j += 1; // increment 'j'
 consumed_count += 1; // increment 'consumed_count'
 ready = false; 
 pthread_mutex_unlock(&m); //release the mutex
 }
}
/* Timing utility function - please ignore */
void* time_and_end(void *args){
 double times[5];
 int i;
 for (i = 0; i < 5; ++i){
 times[i] = clock();
 sched_yield();
 consumed_count = 0;
 while (consumed_count < CONSUMED_MAX*(i+1)){
 sched_yield();
 }
 times[i] = (double)(clock() - times[i])/CLOCKS_PER_SEC;
 }
 for (i = 0; i < 5; ++i){ 
 printf("TIME ELAPSED (%d): %f\n", (i+1)*CONSUMED_MAX, times[i]);
 }
}
int main(void){
 printf("\n\n This is mutex execution!! \n\n");
 /* Initialize thread sync variables */
 pthread_mutex_init(&m, NULL);
 pthread_cond_init(&cv, NULL);
 producer_id = pthread_create(&producer_id, NULL, &producer, NULL);
 consumer_id = pthread_create(&consumer_id, NULL, &consumer, NULL);
 timer_id = pthread_create(&timer_id, NULL, &time_and_end, NULL);
 sched_yield();
 // setpriority(PRIO_PROCESS, producer_id, -20);
 // setpriority(PRIO_PROCESS, consumer_id, -20);
 // setpriority(PRIO_PROCESS, timer_id, -20);
 int x = pthread_join(producer_id, NULL);
 if(x != 0){
 perror("Failed to join producer thread");
 printf("%d\n", x);
 }
 int y = pthread_join(consumer_id, NULL);
 if(y != 0){
 perror("Failed to join consumer thread");
 printf("%d\n", y);
 }
 int z = pthread_join(timer_id, NULL);
 if(z != 0){
 perror("Failed to join timer thread");
 printf("%d\n", z);
 }
 
 int b = pthread_mutex_destroy(&m);
 if(b =! 0){
 perror("Failed to destroy cv");
 printf("%d\n", b);
 }
 usleep(178500);
 printf("cond var: %d\n", cv);
 // int a = pthread_cond_destroy(&cv);
 // if(a =! 0){
 // perror("Failed to destroy cv");
 // printf("%d\n", a);
 // }
 return 0;
}

Semaphore Version

/* main.c - main */
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <pthread.h> 
#include <sched.h>
#include <sys/resource.h>
#include <stdbool.h>
#include <semaphore.h>
//Declaring constant values for use throughout your code
#define ARR_SIZE 250
#define RESTRICT 1700
pthread_t producer_id;
pthread_t consumer_id;
pthread_t timer_id;
int consumed_count = 0;
const int CONSUMED_MAX = 100;
//Declaring circular buffer structure here
int myQueue[ARR_SIZE];
int rear=0, front=0, data=-1, d=1;
//Declaring thread sync variables here
sem_t sem;
//Defining void to insert integer element in queue
void insertIntoQ(int n){
 myQueue[rear] = n; //insert integer element at rear position
 rear = (rear + 1) % ARR_SIZE; //increment rear by 1 
 //Ensuring it points to 1st loc when it reaches buffer limit
}
//Defining void to remove integer element in queue
int removeFromQ(){
 data = myQueue[front]; //remove integer element from front position
 front = (front + 1) % ARR_SIZE; //increment front by 1 
 //Ensuring it points to 1st loc when it reaches buffer limit
 return data;
}
/* Code for the buffer producer*/
void* producer(void *args){
 while(d<=RESTRICT) { //Continue producing till certain limit
 sem_wait(&sem);
 insertIntoQ(d); //insert element
 printf("Produced: %d \n",d); //print to console
 d += 1; // increment 'd'
 sem_post(&sem);
 } 
}
/* Place the code for the buffer consumer here */
void* consumer(void *args){
 int j=1;
 int x,y;
 while(j<=RESTRICT){ //Continue consuming till certain limit
 sem_wait(&sem);
 x = removeFromQ(); //delete element
 y = front; // print loc of element dequeued
 printf("Is %d power of 2 at buffer location %d : %s \n",x,y,(x && (!(x&(x-1))))?"Yes":"No"); //print to console
 j += 1; // increment 'j'
 consumed_count += 1; // increment 'consumed_count'
 sem_post(&sem);
 }
}
/* Timing utility function - please ignore */
void* time_and_end(void *args){
 double times[5];
 int i;
 for (i = 0; i < 5; ++i){
 times[i] = clock();
 sched_yield();
 consumed_count = 0;
 while (consumed_count < CONSUMED_MAX*(i+1)){
 sched_yield();
 }
 times[i] = (double)(clock() - times[i])/CLOCKS_PER_SEC;
 }
 for (i = 0; i < 5; ++i){ 
 printf("TIME ELAPSED (%d): %f\n", (i+1)*CONSUMED_MAX, times[i]);
 }
}
int main(void){
 printf("\n\n This is mutex execution!! \n\n");
 /* Initialize thread sync variables */
 sem_init(&sem,0,1);
 producer_id = pthread_create(&producer_id, NULL, &producer, NULL);
 consumer_id = pthread_create(&consumer_id, NULL, &consumer, NULL);
 timer_id = pthread_create(&timer_id, NULL, &time_and_end, NULL);
 sched_yield();
 // setpriority(PRIO_PROCESS, producer_id, -20);
 // setpriority(PRIO_PROCESS, consumer_id, -20);
 // setpriority(PRIO_PROCESS, timer_id, -20);
 int x = pthread_join(producer_id, NULL);
 if(x != 0){
 perror("Failed to join producer thread");
 printf("%d\n", x);
 }
 int y = pthread_join(consumer_id, NULL);
 if(y != 0){
 perror("Failed to join consumer thread");
 printf("%d\n", y);
 }
 int z = pthread_join(timer_id, NULL);
 if(z != 0){
 perror("Failed to join timer thread");
 printf("%d\n", z);
 }
 
 int b = sem_destroy(&sem);
 if(b =! 0){
 perror("Failed to destroy cv");
 printf("%d\n", b);
 }
 usleep(365500);
 return 0;
}

• Darnoc Eloc: There is one producer and one consumer, producer must not insert data when buffer is full, consumer must not remove data when buffer is empty, the producer and consumer should not insert and remove data simultaneously. I updated both implementations to satisfy these criteria.
• J_H: Please see the "Do not change the code ... after receiving an answer" part of the site guidelines. I have rolled back the relevant revisions.

Answer 1

manifest constants

//Declaring constant values for use throughout your code

Please elide this comment, as it does not further illuminate the source code, similar to other "section heading" comments we see further down. ARR_SIZE and RESTRICT are lovely identifiers. They're in all caps, so it's clear that they are manifest constants.

In modern source code it would be better to declare them with const int assignments, as you later do with CONSUMED_MAX. For that matter, you didn't tell us if you're targetting -std=c17 or some other variant of ANSI C.

comments

//Defining void to insert integer element in queue
void insertIntoQ( int ...

I can't imagine how the phrase "Defining void" would help a maintenance engineer. In general, please apply the following test to each comment you write:

Does this comment explain the "why"? Or the "how"?

The code already tells us "how" actions will be performed. Reserve your comments to tell us "why" a certain coding decision was made.

The interesting thing to say about this function is that it always ensures rear points at the next free index.

//increment rear by 1 
//ensuring it points to 1st loc when it reaches buffer limit

Limiting a description of the modulo operator to a single-word // wraparound comment would suffice. Personally I find the offered comment confusing or misleading, since it uses Fortran indexing terminology rather than a C "0-th location". Not unlike the subsequent "Defining void to remove ..." comment, which inexplicably describes a function that returns int rather than void.

It is well known that "Comments lie!"

Write fewer of them, and fewer maintenance engineers will be led astray.

meaningful identifiers

pthread_mutex_t m;

Single character identifiers are great at local scope. But for a global like this, consider at least typing mut or mutex.

Consider using static to scope this (and other variables) down to just the current compilation unit. As written, you're inviting the ld link editor to pick up m to satisfy references that another *.o object file might make. And then hilarity ensues.

Similarly for the rather vague d.

OTOH, front & rear are very nice identifiers.

volatile

I suspect that you want several variables marked volatile, including d, front, rear, and consumed_count.

Oh, wait! It turns out d pops up in fewer places than the initial usage suggested. Just make it a local. No need to rename it to cnt or something more descriptive.

unused arguments

I have no idea why time_and_end, producer, and consumer take a void *args parameter. Just elide it.

correctness

It's really weird that producer blindly overwrites unconsumed elements when consumer falls behind, and there's no commentary on that.

I would usually expect to see a datastructure like myQueue (along with its ancillary variables) documented in terms of what thread coordination (such as holding mutex) you're required to do before operating on it. You didn't really spell out the rules to play by, which is going to complicate the job of any future maintenance engineer who works on this code.

---

This was a nice learning exercise.

Now put it away and move on to something else, as anyone else will have a hard time trying to maintain this code or make it operate correctly.

Answer 2

This is my solution which satisfy these criteria:

• one producer and one consumer.
• producer must not insert data when buffer is full.
• consumer must not remove data when buffer is empty.
• the producer and consumer should not insert and remove data simultaneously.

Open to comments on the use of sched_yield(), setpriority(), and other critiques.

Mutex and CV version.

/* main.c - main */
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <pthread.h> 
#include <sched.h>
#include <sys/resource.h>
#include <stdbool.h>
//Declaring constant values for use throughout your code
#define ARR_SIZE 250
const int RESTRICT = 1700;
const int CONSUMED_MAX = 100;
pthread_t producer_id;
pthread_t consumer_id;
pthread_t timer_id;
static int consumed_count = 0;
//Declaring circular buffer structure here
static int myQueue[ARR_SIZE];
static int rear=0, front=0, data=-1;
//Declaring thread sync variables here
pthread_mutex_t mutQ;
pthread_cond_t cv;
//Function to insert integer element into queue
void insertIntoQ(int n){
 myQueue[rear] = n; //insert integer element at rear position
 rear = (rear + 1) % ARR_SIZE; //increment rear by 1 
 //Ensuring it points to 1st loc when it reaches buffer limit
}
//Function to read integer element from queue
int removeFromQ(){
 data = myQueue[front]; //remove integer element from front position
 front = (front + 1) % ARR_SIZE; //increment front by 1 
 //Ensuring it points to 1st loc when it reaches buffer limit
 return data;
}
/* Code for the buffer producer*/
void* producer(void *args){
 int d = 1;
 while(d<=RESTRICT) { //Continue producing till certain limit
 pthread_mutex_lock(&mutQ); //acquire the mutex
 if(rear == front){
 insertIntoQ(d); //insert element
 printf("Produced: %d \n",d); //print to console
 ++d; // increment 'd'
 }
 pthread_mutex_unlock(&mutQ); //release the mutex
 pthread_cond_signal(&cv);
 usleep(1);
 } 
}
/*Code for the buffer consumer*/
void* consumer(void *args){
 int j=1;
 int x,y;
 while(j<=RESTRICT){ //Continue consuming till certain limit
 pthread_mutex_lock(&mutQ); //acquire the mutex
 while(!true){
 pthread_cond_wait(&cv, &mutQ);
 }
 while(rear != front){
 x = removeFromQ(); //delete element
 y = front; // print loc of element dequeued
 printf("Is %d power of 2 at buffer location %d : %s \n",x,y,(x && (!(x&(x-1))))?"Yes":"No"); //print to console
 ++j; // increment 'j'
 consumed_count += 1; // increment 'consumed_count'
 }
 pthread_mutex_unlock(&mutQ); //release the mutex
 usleep(1);
 }
}
/* Timing utility function - please ignore */
void* time_and_end(void *args){
 double times[5];
 int i;
 for (i = 0; i < 5; ++i){
 times[i] = clock();
 sched_yield();
 consumed_count = 0;
 while (consumed_count < CONSUMED_MAX*(i+1)){
 sched_yield();
 }
 times[i] = (double)(clock() - times[i])/CLOCKS_PER_SEC;
 }
 for (i = 0; i < 5; ++i){ 
 printf("TIME ELAPSED (%d): %f\n", (i+1)*CONSUMED_MAX, times[i]);
 }
}
int main(void){
 printf("\n\n This is mutex execution!! \n\n");
 /* Initialize thread sync variables */
 pthread_mutex_init(&mutQ, NULL);
 // pthread_mutex_init(&cons, NULL);
 pthread_cond_init(&cv, NULL);
 producer_id = pthread_create(&producer_id, NULL, &producer, NULL);
 consumer_id = pthread_create(&consumer_id, NULL, &consumer, NULL);
 timer_id = pthread_create(&timer_id, NULL, &time_and_end, NULL);
 sched_yield();
 setpriority(PRIO_PROCESS, producer_id, -20);
 setpriority(PRIO_PROCESS, consumer_id, -20);
 setpriority(PRIO_PROCESS, timer_id, -20);
 int x = pthread_join(producer_id, NULL);
 if(x != 0){
 perror("Failed to join producer thread");
 printf("%d\n", x);
 }
 int y = pthread_join(consumer_id, NULL);
 if(y != 0){
 perror("Failed to join consumer thread");
 printf("%d\n", y);
 }
 int z = pthread_join(timer_id, NULL);
 if(z != 0){
 perror("Failed to join timer thread");
 printf("%d\n", z);
 }
 
 int a = pthread_mutex_destroy(&mutQ);
 if(a =! 0){
 perror("Failed to destroy mutex");
 printf("%d\n", a);
 }
 int b = pthread_cond_destroy(&cv);
 if(b =! 0){
 perror("Failed to destroy cond var");
 printf("%d\n", b);
 }
 usleep(109900); 
 return 0;
}

Semaphore Version

/* main.c - main */
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <pthread.h> 
#include <sched.h>
#include <sys/resource.h>
#include <stdbool.h>
#include <semaphore.h>
//Declaring constant values for use throughout your code
#define ARR_SIZE 250
#define RESTRICT 1700
pthread_t producer_id;
pthread_t consumer_id;
pthread_t timer_id;
static int consumed_count = 0;
const int CONSUMED_MAX = 100;
//Declaring circular buffer structure here
static int myQueue[ARR_SIZE];
static int rear=0, front=0, data=-1;
//Declaring thread sync variables here
pthread_mutex_t mutQ;
sem_t empty;
sem_t full;
//Function to insert integer element in queue
void insertIntoQ(int n){
 myQueue[rear] = n; //insert integer element at rear position
 rear = (rear + 1) % ARR_SIZE; //increment rear by 1 
 //Ensuring it points to 1st loc when it reaches buffer limit
}
//Function to read integer element from queue
int removeFromQ(){
 data = myQueue[front]; //remove integer element from front position
 front = (front + 1) % ARR_SIZE; //increment front by 1 
 //Ensuring it points to 1st loc when it reaches buffer limit
 return data;
}
/* Code for the buffer producer*/
void* producer(void *args){
 int d = 1;
 while(d<=RESTRICT) { //Continue producing till certain limit
 sem_wait(&empty);
 pthread_mutex_lock(&mutQ); //acquire the mutex
 insertIntoQ(d); //insert element
 printf("Produced: %d \n",d); //print to console
 d += 1; // increment 'd'
 pthread_mutex_unlock(&mutQ); //release the mutex
 sem_post(&full);
 usleep(1);
 } 
}
/* Place the code for the buffer consumer here */
void* consumer(void *args){
 int j=1;
 int x,y;
 while(j<=RESTRICT){ //Continue consuming till certain limit
 sem_wait(&full);
 pthread_mutex_lock(&mutQ); //acquire the mutex
 x = removeFromQ(); //delete element
 y = front; // print loc of element dequeued
 printf("Is %d power of 2 at buffer location %d : %s \n",x,y,(x && (!(x&(x-1))))?"Yes":"No"); //print to console
 j += 1; // increment 'j'
 consumed_count += 1; // increment 'consumed_count'
 pthread_mutex_unlock(&mutQ); //release the mutex
 sem_post(&empty);
 usleep(1);
 }
}
/* Timing utility function - please ignore */
void* time_and_end(void *args){
 double times[5];
 int i;
 for (i = 0; i < 5; ++i){
 times[i] = clock();
 sched_yield();
 consumed_count = 0;
 while (consumed_count < CONSUMED_MAX*(i+1)){
 sched_yield();
 }
 times[i] = (double)(clock() - times[i])/CLOCKS_PER_SEC;
 }
 for (i = 0; i < 5; ++i){ 
 printf("TIME ELAPSED (%d): %f\n", (i+1)*CONSUMED_MAX, times[i]);
 }
}
int main(void){
 printf("\n\n This is mutex execution!! \n\n");
 /* Initialize thread sync variables */
 pthread_mutex_init(&mutQ, NULL);
 sem_init(&empty,0,ARR_SIZE);
 sem_init(&full,0,0);
 producer_id = pthread_create(&producer_id, NULL, &producer, NULL);
 consumer_id = pthread_create(&consumer_id, NULL, &consumer, NULL);
 timer_id = pthread_create(&timer_id, NULL, &time_and_end, NULL);
 sched_yield();
 // setpriority(PRIO_PROCESS, producer_id, -20);
 // setpriority(PRIO_PROCESS, consumer_id, -20);
 // setpriority(PRIO_PROCESS, timer_id, -20);
 int x = pthread_join(producer_id, NULL);
 if(x != 0){
 perror("Failed to join producer thread");
 printf("%d\n", x);
 }
 int y = pthread_join(consumer_id, NULL);
 if(y != 0){
 perror("Failed to join consumer thread");
 printf("%d\n", y);
 }
 int z = pthread_join(timer_id, NULL);
 if(z != 0){
 perror("Failed to join timer thread");
 printf("%d\n", z);
 }
 
 int a = pthread_mutex_destroy(&mutQ);
 if(a =! 0){
 perror("Failed to destroy mutex");
 printf("%d\n", a);
 }
 int b = sem_destroy(&empty);
 if(b =! 0){
 perror("Failed to destroy empty semaphore");
 printf("%d\n", b);
 }
 int c = sem_destroy(&full);
 if(c =! 0){
 perror("Failed to destroy full semaphore");
 printf("%d\n", c);
 }
 usleep(135500);
 return 0;
}