Reputation: 1012
I am implementing a producer-consumer problem using pthreads and semaphore. I have 1 Producer and 2 Consumers. My Producer reads characters one by one from a file and enqueues them into a circular queue. I want the consumers to read from the queue and store into separate arrays. I want the reading in such a way that the first consumer reads 2 characters and the second consumer reads every 3rd character. I am trying to do this using pthread_cond_wait()
but it is not working out. This is my code:
#include<iostream>
#include<pthread.h>
#include<fstream>
#include<unistd.h>
#include<semaphore.h>
#include<queue>
#include "circular_queue"
// define queue size
#define QUEUE_SIZE 5
// declare and initialize semaphore and read/write counter
static sem_t mutex,queueEmptyMutex;
//static int counter = 0;
// Queue for saving characters
static Queue charQueue(QUEUE_SIZE);
//static std::queue<char> charQueue;
// indicator for end of file
static bool endOfFile = false;
// save arrays
static char consumerArray1[100];
static char consumerArray2[100];
static pthread_cond_t cond;
static pthread_mutex_t cond_mutex;
static bool thirdCharToRead = false;
void *Producer(void *ptr)
{
int i=0;
std::ifstream input("string.txt");
char temp;
while(input>>temp)
{
std::cout<<"reached here a"<<std::endl;
sem_wait(&mutex);
std::cout<<"reached here b"<<std::endl;
if(!charQueue.full())
{
charQueue.enQueue(temp);
}
sem_post(&queueEmptyMutex);
sem_post(&mutex);
i++;
sleep(4);
}
endOfFile = true;
sem_post(&queueEmptyMutex);
pthread_exit(NULL);
}
void *Consumer1(void *ptr)
{
int i = 0;
sem_wait(&queueEmptyMutex);
bool loopCond = endOfFile;
while(!loopCond)
{
std::cout<<"consumer 1 loop"<<std::endl;
if(endOfFile)
{
loopCond = charQueue.empty();
std::cout<<loopCond<<std::endl;
sem_post(&queueEmptyMutex);
}
sem_wait(&queueEmptyMutex);
sem_wait(&mutex);
if(!charQueue.empty())
{
consumerArray1[i] = charQueue.deQueue();
i++;
if(i%2==0)
{
pthread_mutex_lock(&cond_mutex);
std::cout<<"Signal cond. i = "<<i<<std::endl;
thirdCharToRead = true;
pthread_mutex_unlock(&cond_mutex);
pthread_cond_signal(&cond);
}
}
if(charQueue.empty()&&endOfFile)
{
sem_post(&mutex);
sem_post(&queueEmptyMutex);
break;
}
sem_post(&mutex);
sleep(2);
std::cout<<"consumer 1 loop end"<<std::endl;
}
consumerArray1[i] = '\0';
pthread_exit(NULL);
}
void *Consumer2(void *ptr)
{
int i = 0;
sem_wait(&queueEmptyMutex);
bool loopCond = endOfFile;
while(!loopCond)
{
std::cout<<"consumer 2 loop"<<std::endl;
if(endOfFile)
{
loopCond = charQueue.empty();
std::cout<<loopCond<<std::endl;
sem_post(&queueEmptyMutex);
}
sem_wait(&queueEmptyMutex);
sem_wait(&mutex);
if(!charQueue.empty())
{
pthread_mutex_lock(&cond_mutex);
while(!thirdCharToRead)
{
std::cout<<"Waiting for condition"<<std::endl;
pthread_cond_wait(&cond,&cond_mutex);
}
std::cout<<"Wait over"<<std::endl;
thirdCharToRead = false;
pthread_mutex_unlock(&cond_mutex);
consumerArray2[i] = charQueue.deQueue();
i++;
}
if(charQueue.empty()&& endOfFile)
{
sem_post(&mutex);
sem_post(&queueEmptyMutex);
break;
}
sem_post(&mutex);
std::cout<<"consumer 2 loop end"<<std::endl;
sleep(2);
}
consumerArray2[i] = '\0';
pthread_exit(NULL);
}
int main()
{
pthread_t thread[3];
sem_init(&mutex,0,1);
sem_init(&queueEmptyMutex,0,1);
pthread_mutex_init(&cond_mutex,NULL);
pthread_cond_init(&cond,NULL);
pthread_create(&thread[0],NULL,Producer,NULL);
int rc = pthread_create(&thread[1],NULL,Consumer1,NULL);
if(rc)
{
std::cout<<"Thread not created"<<std::endl;
}
pthread_create(&thread[2],NULL,Consumer2,NULL);
pthread_join(thread[0],NULL);pthread_join(thread[1],NULL);pthread_join(thread[2],NULL);
std::cout<<"First array: "<<consumerArray1<<std::endl;
std::cout<<"Second array: "<<consumerArray2<<std::endl;
sem_destroy(&mutex);
sem_destroy(&queueEmptyMutex);
pthread_exit(NULL);
}
The problem I am having is after one read, consumer 2 goes into infinite loop in the while(!thirdCharToRead)
. Is there any better way to implement this?
Upvotes: 0
Views: 189
Reputation: 182753
Okay, let's start with this code:
std::cout<<"Wait over"<<std::endl;
pthread_mutex_unlock(&cond_mutex);
thirdCharToRead = false;
This code says that cond_mutex
does not protect thirdCharToRead
from concurrent access. Why? Because it modifies thirdCharToRead
without holding that mutex.
Now look at this code:
pthread_mutex_lock(&cond_mutex);
while(!thirdCharToRead)
{
std::cout<<"Waiting for condition"<<std::endl;
pthread_cond_wait(&cond,&cond_mutex);
}
Now, the while
loop checks thirdCharToRead
, so we must hold whatever lock protects thirdCharToRead
from concurrent access when we test it. But the while
loop will loop forever if thirdCharToRead
stays locked for the whole loop since no other thread could ever change it. Thus, this code only makes sense if somewhere in the loop we release the lock that protects thirdCharToRead
, and the only lock we release in the loop is cond_mutex
in the call to pthread_cond_wait
.
So this code only makes sense if cond_mutex
protects thirdCharToRead
.
Houston, we have a problem. One chunk of code says cond_mutex
does not protect thirdCharToRead
and one chunk of code says cond_mutex
does protect thirdCharToRead
.
Upvotes: 2