I have added a Mutex class to adhere to RAII. I am not sure if the way I am using it is correct.
After the queue is locked by a producer, the program unexpectedly finishes.
MutexClass.h
#ifndef MUTEXCLASS
#define MUTEXCLASS
#include <pthread.h>
class MutexClass
{
private:
pthread_mutex_t & _mutexVariable;
public:
MutexClass (pthread_mutex_t &);
~MutexClass ();
};
#endif // MUTEXCLASS
MutexClass.cpp
#include "mutexClass.h"
#include <stdexcept>
MutexClass::MutexClass (pthread_mutex_t & arg) : _mutexVariable (arg)
{
_mutexVariable = PTHREAD_MUTEX_INITIALIZER;
int returnValue = pthread_mutex_lock (&_mutexVariable);
if (returnValue > 0)
{
throw std::logic_error ("Mutex couldn't be locked!");
}
}
MutexClass::~MutexClass()
{
pthread_mutex_unlock (&_mutexVariable);
}
This is the main.cpp where I am using the object of the above defined mutex class.
Qt's classes are here for the namesake because I am Qt Creator. Please ignore them.
#include "mainwindow.h"
#include <QApplication>
#include <stdexcept>
#include <stdio.h>
#include <unistd.h>
#include <sys/syscall.h>
#include <iostream>
#include <QDebug>
#include "mutexClass.h"
pthread_mutex_t mutexVariable;
pthread_cond_t conditionVariable;
int numberOfActiveProducers;
int numberOfActiveConsumers;
QList <int> sharedQueueA;
QList <int> sharedQueueB;
/*
* Shared queues are supposed to be shared among four threads. Two producer, and two consumer threads.
* Producer threads will put the 1's in it, and Consumer threads will remove the 1's.
* Assumption: `sharedQueue` can contain only 10 elements at a time.
*/
int sizeOfSharedQueue;
void checkForSpaceAndPush (QList <int> & argList, int listId, pthread_t argTId)
{
std::cerr << "\nQueue " << listId << ", First check by Producer: " << argTId;
if (argList.length () < sizeOfSharedQueue + 1)
{
{
MutexClass mutex1 (mutexVariable);
std::cerr << "\n\nQueue " << listId << ", Locked by Producer: " << argTId;
if (argList.length () < sizeOfSharedQueue + 1)
{
argList.push_back (1); std::cerr << "\nPushed by Producer " << argTId << ": " << "Length of queue " << listId << " is: " << argList.length ();
}
else
{
std::cerr << "\nProducer " << argTId << ". Queue " << listId << " is full. Length of queue is: " << argList.length ();
pthread_cond_wait (&conditionVariable, &mutexVariable);
}
}
std::cerr << "\n\nQueue " << listId << ", UnLocked by Producer: " << argTId;
}
}
void checkForSpaceAndPop (QList <int> & argList, int listId, pthread_t argTId)
{
std::cerr << "\nQueue " << listId << ", First check by Consumer: " << argTId;
if (argList.length () > 0)
{
{
MutexClass mutex1 (mutexVariable);
std::cerr << "\n\nQueue " << listId << ", Locked by Consumer: " << argTId;
if (argList.length () > 0)
{
argList.pop_front (); std::cerr << "\nRemoved by Consumer: " << argTId << ", Length of queue " << listId << " is: " << argList.length ();
}
else
{
pthread_cond_signal (&conditionVariable); std::cerr << "\nSignal issued by Consumer: " << argTId << ", Length of queue " << listId << " is: " << argList.length ();
}
}
std::cerr << "\n\nQueue " << listId << ", UnLocked by Consumer: " << argTId;
}
}
// This function is run by the `Producer` threads.
void *producerThreadFunction (void *arg)
{
Q_UNUSED (arg);
while (1)
{
pthread_t tId = pthread_self(); std::cerr << "\nProducers: " << tId; std::cerr.flush();
checkForSpaceAndPush (sharedQueueA, 1, tId);
checkForSpaceAndPush (sharedQueueB, 2, tId);
}
return NULL;
}
// This function is run by the `Consumer` threads.
void *consumerThreadFunction (void *arg)
{
Q_UNUSED (arg);
while (1)
{
pthread_t tId = pthread_self (); std::cerr << "\nConsumer: " << tId; std::cerr.flush();
checkForSpaceAndPop (sharedQueueA, 1, tId);
checkForSpaceAndPop (sharedQueueB, 2, tId);
}
return NULL;
}
int main (int argc, char *argv[])
{
numberOfActiveProducers = 2;
numberOfActiveConsumers = 2;
sizeOfSharedQueue = 10;
// Producer threads creation
pthread_t producerA;
pthread_t producerB;
if (pthread_create (&producerA, NULL, producerThreadFunction, NULL)) {
fprintf (stderr, "Error creating thread Producer A\n");
return 1;
}
if (pthread_create (&producerB, NULL, producerThreadFunction, NULL)) {
fprintf (stderr, "Error creating thread Producer B\n");
return 1;
}
// Consumer threads creation
pthread_t consumerA;
pthread_t consumerB;
if (pthread_create (&consumerA, NULL, consumerThreadFunction, NULL)) {
fprintf (stderr, "Error creating thread Consumer A\n");
return 1;
}
if (pthread_create (&consumerB, NULL, consumerThreadFunction, NULL)) {
fprintf (stderr, "Error creating thread Consumer B\n");
return 1;
}
// Joining every thread
if (pthread_join (producerA, NULL)) {
fprintf (stderr, "Error joining thread Producer A\n");
return 2;
}
if (pthread_join (producerB, NULL)) {
fprintf (stderr, "Error joining thread Producer B\n");
return 2;
}
if (pthread_join (consumerB, NULL)) {
fprintf (stderr, "Error joining thread Consumer B\n");
return 2;
}
if (pthread_join (consumerA, NULL)) {
fprintf (stderr, "Error joining thread Consumer A\n");
return 2;
}
QApplication a (argc, argv);
MainWindow w;
w.show ();
return a.exec ();
}
Output is here:
...
...
Queue 140388157085440
Removed by Consumer: 1403881570854401403881654781442
Queue , Locked by Consumer: 1140388148692736
Removed by Consumer: , Length of queue 1 is: , First check by Producer: 140388148692736, Length of queue 2 is: 1403881654781449
Queue
Queue 2, UnLocked by Consumer: 140388148692736
Consumer: 9
Queue 1, UnLocked by Consumer: 140388157085440
Queue 2, First check by Consumer: 1403881570854401140388148692736
Queue 1, First check by Consumer: 140388148692736
Queue , Locked by Producer:
Queue 2, Locked by Consumer: 140388157085440
Removed by Consumer: 1403881654781441, Locked by Consumer: 140388148692736
Pushed by Producer 140388165478144: Length of queue 1 is: 10
Queue 1, UnLocked by Producer: 140388165478144
Queue 2, First check by Producer: 140388165478144
Queue 2, Locked by Producer: 140388165478144The program has unexpectedly finished.
Also, notice the following part of output:
Queue 2, Locked by Consumer: 140388157085440 Removed by Consumer: 1403881654781441, Locked by Consumer: 140388148692736
I have created only 2 consumers but the pids shown here are 3. Why is that so?
Currently, you initialize mutexVariable to PTHREAD_MUTEX_INITIALIZER every time you create a new construct a new MutexClass. Consider what happens if thread A holds the mutexVariable and thread B wants to acquire the mutex:
thread A tries to lock mutexVariable and succeeds
mutexVariable = PTHRAED_MUTEX_INITIALIZER
pthread_mutex_lock(mutexVariable)
thread B tries to lock mutexVariable and succeeds
// this assignment overwrites the locked state thread A has stored
mutexVariable = PTHRAED_MUTEX_INITIALIZER
// mutex is default-initialized (not locked) - so lock it
pthread_mutex_lock(mutexVariable)
// both threads now believe they have the mutex
// and all syncronization is lost
thread B unlocks mutexVariable
// succeeds
thread A unlocks mutexVariable
// uh.. it is not even locked any more?!
You should initialize mutexVariable only once, in main.cpp:13:
pthread_mutex_t mutexVariable = PTHRAED_MUTEX_INITIALIZER;
and remove the initialization from MutexClass.
Not sure if this will fix all of your issues, but that's the thing I'd do first.