I have made a multi-threaded producer/consumer application which I have been struggling with for a couple of days now. The producer(s) place(s) Fibonacci numbers into a circular buffer and the consumer(s) take numbers from the buffer until the specified limit is reached.
I have protected the circular buffer with a mutex (Mutual Exclusion) which should prevent multiple threads from accessing the same data. I have also set up events which should prevent the producer(s) from overflowing the buffer, and the consumer(s) from accessing the buffer while it is empty.
While I say this, I still notice that the buffer is being accessed by consumers while it is empty. This is why I have added a break to the consumer thread (I do not quite understand why this would be necessary).
I'm also occasionally receiving "Access violation reading location" errors, which I just can't comprehend. I have noticed those occur more often up to almost always when creating more threads. I thought these might occur because the consumer was trying to read the buffer at locations that do not exist, but I have seen that this is not the case.
What might be causing my issues? Is it possible that multiple threads pass the WaitForSingleObject on the Mutex?
This is the Fibonacci.c
#include "Fibonacci.h"
#define MINIMUM 1
#define MAXIMUM 5
HANDLE eBufferNotFull;
HANDLE eBufferNotEmpty;
HANDLE fiboMutex;
HANDLE bufferMutex;
CircularBuffer *buffer;
Fibonumbers numbers;
int main(void) {
uint8_t amountOfProducers, amountOfConsumers, size;
ThreadStruct consumerInfo, producerInfo;
setValue("The amount of producers", &amountOfProducers, MINIMUM, MAXIMUM);
setValue("The amount of consumers", &amountOfConsumers, MINIMUM, MAXIMUM);
setValue("The size of the buffer", &size, 1, 80);
resetFibo(&numbers);
setValue("The sleeping time for producers", &producerInfo.sleep, 0, 10000);
setValue("The sleeping time for consumers", &consumerInfo.sleep, 0, 10000);
setValue("The limit for the fibonumber", &producerInfo.limit, 0, 35000000000000000);
consumerInfo.limit = producerInfo.limit;
HANDLE hProducer[MAXIMUM];
DWORD dwProducer[MAXIMUM];
HANDLE hConsumer[MAXIMUM];
DWORD dwConsumer[MAXIMUM];
buffer = createBuffer(size);
/* Create the Mutexes */
fiboMutex = CreateMutex(NULL, FALSE, NULL);
bufferMutex = CreateMutex(NULL, FALSE, NULL);
/* Create the Events */
eBufferNotFull = CreateEvent(NULL, FALSE, TRUE, TEXT("buffer_niet_vol"));
eBufferNotEmpty = CreateEvent(NULL, FALSE, FALSE, TEXT("buffer_niet_leeg"));
/* Create the producer threads*/
for (int i = 0; i < amountOfProducers; ++i) {
hProducer[i] = CreateThread(NULL, // No security
0, // Use default stack size
(LPTHREAD_START_ROUTINE)producer,
&producerInfo, // Thread argument
0, // Child became running
(LPDWORD)&dwProducer[i]); // Child id
}
/* Create the consumer threads*/
for (int i = 0; i < amountOfConsumers; ++i) {
hConsumer[i] = CreateThread(NULL, // No security
0, // Use default stack size
(LPTHREAD_START_ROUTINE)consumer,
&consumerInfo, // Thread argument
0, // Child became running
(LPDWORD)&dwConsumer[i]); // Child id
}
WaitForMultipleObjects(amountOfProducers, hProducer, true, INFINITE);
WaitForMultipleObjects(amountOfConsumers, hConsumer, true, INFINITE);
deleteBuffer(buffer);
return (EXIT_SUCCESS);
}
DWORD WINAPI producer(LPVOID lpParameter) {
ThreadStruct *info = (ThreadStruct *)lpParameter;
while (true) {
Sleep(info->sleep);
WaitForSingleObject(fiboMutex, INFINITE); // Lock the fibonumber struct
createNewFibonumber();
if (numbers.currentFibo > info->limit) {
ReleaseMutex(fiboMutex); // Release the fibonumber struct
ExitThread(EXIT_SUCCESS);
}
WaitForSingleObject(eBufferNotFull, INFINITE);
WaitForSingleObject(bufferMutex, INFINITE);
putElement(buffer, numbers.currentFibo);
ReleaseMutex(fiboMutex); // Release the fibonumber struct
ReleaseMutex(bufferMutex);
SetEvent(eBufferNotEmpty);
}
}
DWORD WINAPI consumer(LPVOID lpParameter) {
ThreadStruct *info = (ThreadStruct *)lpParameter;
while (true) {
Sleep(info->sleep);
WaitForSingleObject(eBufferNotEmpty, INFINITE);
WaitForSingleObject(bufferMutex, INFINITE);
printf(" fibogetal: %i \n", getElement(buffer));
ReleaseMutex(bufferMutex);
SetEvent(eBufferNotFull);
}
ExitThread(EXIT_SUCCESS);
}
void createNewFibonumber() {
uint64_t i = numbers.currentFibo;
numbers.currentFibo += numbers.lastFibo;
numbers.lastFibo = i;
}
void resetFibo(Fibonumbers *numbers) {
numbers->lastFibo = 0;
numbers->currentFibo = 1;
}
void setValue(char *text, void *intpointer, uint64_t minimum, uint64_t maximum) {
printf("%s\n", text);
do {
*(uint64_t *)intpointer = 0;
printf("Enter a value from %lli up to %lli : ", minimum, maximum);
scanf_s("%lli", intpointer);
} while (*(uint64_t *)intpointer < minimum || *(uint64_t *)intpointer > maximum);
}
Fibonacci.h
#include <stdio.h>
#include <stdint.h>
#include <conio.h>
#include <Windows.h>
#include "Buffer.h"
typedef struct {
uint64_t currentFibo;
uint64_t lastFibo;
} Fibonumbers;
typedef struct {
uint64_t limit;
uint16_t sleep;
} ThreadStruct;
/*
*
*/
DWORD WINAPI producer(LPVOID lpParameter);
/*
*
*/
DWORD WINAPI consumer(LPVOID lpParameter);
/*
*
*/
void createNewFibonumber();
/*
*
*/
void resetFibo(Fibonumbers *numbers);
/*
*
*/
void setValue(char *text, void *intpointer, uint64_t minimum, uint64_t maximum);
And the Buffer.c
#include "Buffer.h"
CircularBuffer *createBuffer(uint8_t size) {
CircularBuffer *buffer = (CircularBuffer *)calloc(1, sizeof(CircularBuffer));
buffer->size = size;
buffer->count = 0;
buffer->start = 0;
buffer->end = 0;
buffer->buffer = (uint64_t *)calloc(buffer->size, sizeof(uint64_t));
return buffer;
}
void deleteBuffer(CircularBuffer *buffer) {
if (buffer) {
free(buffer->buffer);
free(buffer);
}
}
void putElement(CircularBuffer *buffer, uint64_t element) {
buffer->count++;
buffer->buffer[buffer->start] = element;
buffer->start++;
if (buffer->start == buffer->size) {
buffer->start = 0;
}
printf("put: %i items in buffer.\n", buffer->count);
}
uint64_t getElement(CircularBuffer *buffer) {
buffer->count--;
uint64_t value = buffer->buffer[buffer->end];
buffer->end++;
if (buffer->end == buffer->size) {
buffer->end = 0;
}
printf(" get: %i items in buffer.\n", buffer->count);
return value;
}
bool isBufferFull(CircularBuffer *buffer) {
return (buffer->count == buffer->size);
}
bool isBufferEmpty(CircularBuffer *buffer) {
return (buffer->count == 0);
}
Buffer.h
#include <stdint.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
typedef struct {
uint64_t *buffer;
uint8_t size;
uint8_t count;
uint8_t start;
uint8_t end;
} CircularBuffer;
CircularBuffer *createBuffer(uint8_t size);
void deleteBuffer(CircularBuffer *buffer);
void putElement(CircularBuffer *buffer, uint64_t element);
uint64_t getElement(CircularBuffer *buffer);
bool isBufferFull(CircularBuffer *buffer);
bool isBufferEmpty(CircularBuffer *buffer);
If someone wishes to also check out the header files, please say so.
edit: I have updated the code, it is now fully functional. edit2: The program works when I build it under debug mode, but when build under release mode it seems to not start the threads.
I find this line to be highly suspicious:
buffer->buffer = (uint64_t *)calloc(buffer->size, sizeof(unsigned int));
If buffer->buffer is an array of uint64_t, why are you using sizeof(unsigned int) to allocate it? I don't know if that is your problem, but it's at least one thing that should be fixed up.