diff --git a/src/shared/queue.c b/src/shared/queue.c index 1d7a0e4..b9930dd 100644 --- a/src/shared/queue.c +++ b/src/shared/queue.c @@ -1,154 +1,154 @@ -#include -#include -#include -#include -#include -#include - -#include "queue.h" - -Queue* queue_create(int capacity, void (*destroyer)(void* item)) { - if (capacity <= 0) - return NULL; - - Queue* queue = (Queue*)malloc(sizeof(Queue)); - if (queue == NULL) { - perror("ERROR: Could not allocate memory for queue structure"); - return NULL; - } - - queue->items = malloc(capacity * sizeof(void*)); - if (queue->items == NULL) { - free(queue); - return NULL; - } - - for (int i = 0; i < capacity; ++i) { - queue->items[i] = NULL; - } - - queue->capacity = capacity; - queue->front = 0; - queue->rear = 0; - queue->size = 0; - queue->item_destroyer = destroyer; - - return queue; -} - -void queue_destroy(Queue* queue) { - if (queue == NULL) - return; - - if (queue->item_destroyer != NULL) { - for (int i = 0; i < queue->size; ++i) { - int index = (queue->front + i) % queue->capacity; - queue->item_destroyer(queue->items[index]); - } - } - free(queue->items); - free(queue); -} - -bool queue_is_empty(const Queue* queue) { - if (queue == NULL) - return true; - return queue->size == 0; -} - -bool queue_is_full(const Queue* queue) { - if (queue == NULL) - return false; - return queue->size == queue->capacity; -} - -static bool queue_double_capacity(Queue* queue) { - if (queue == NULL) - return false; - if (queue->capacity > INT_MAX / 2) - return false; - int new_capacity = queue->capacity * 2; - if (new_capacity <= 1) - new_capacity = 100; - void** new_items = malloc(new_capacity * sizeof(void*)); - if (new_items == NULL) { - perror("ERROR: Could not allocate memory for doubling capacity of queue."); - return false; - } - for (int i = 0; i < queue->size; i++) - new_items[i] = queue->items[(i + queue->front) % queue->capacity]; - free(queue->items); - queue->items = new_items; - queue->front = 0; - queue->rear = queue->size; - queue->capacity = new_capacity; - return true; -} - -bool queue_enqueue(Queue* queue, void* item) { - if (queue == NULL || item == NULL) - return false; - if (queue_is_full(queue)) { - if (!queue_double_capacity(queue)) - return false; - } - queue->items[queue->rear] = item; - queue->rear = (queue->rear + 1) % queue->capacity; - queue->size++; - return true; -} - -bool queue_enqueue_multithreaded(Queue* queue, void* item, mtx_t* mutex, cnd_t* condition_not_empty, - cnd_t* condition_not_full) { - mtx_lock(mutex); - while (queue_is_full(queue)) - cnd_wait(condition_not_full, mutex); - bool ok = queue_enqueue(queue, item); - cnd_signal(condition_not_empty); - mtx_unlock(mutex); - return ok; -} - +#include +#include +#include +#include +#include +#include + +#include "queue.h" + +Queue* queue_create(int capacity, void (*destroyer)(void* item)) { + if (capacity <= 0) + return NULL; + + Queue* queue = (Queue*)malloc(sizeof(Queue)); + if (queue == NULL) { + perror("ERROR: Could not allocate memory for queue structure"); + return NULL; + } + + queue->items = malloc(capacity * sizeof(void*)); + if (queue->items == NULL) { + free(queue); + return NULL; + } + + for (int i = 0; i < capacity; ++i) { + queue->items[i] = NULL; + } + + queue->capacity = capacity; + queue->front = 0; + queue->rear = 0; + queue->size = 0; + queue->item_destroyer = destroyer; + + return queue; +} + +void queue_destroy(Queue* queue) { + if (queue == NULL) + return; + + if (queue->item_destroyer != NULL) { + for (int i = 0; i < queue->size; ++i) { + int index = (queue->front + i) % queue->capacity; + queue->item_destroyer(queue->items[index]); + } + } + free(queue->items); + free(queue); +} + +bool queue_is_empty(const Queue* queue) { + if (queue == NULL) + return true; + return queue->size == 0; +} + +bool queue_is_full(const Queue* queue) { + if (queue == NULL) + return false; + return queue->size == queue->capacity; +} + +static bool queue_double_capacity(Queue* queue) { + if (queue == NULL) + return false; + if (queue->capacity > INT_MAX / 2) + return false; + int new_capacity = queue->capacity * 2; + if (new_capacity <= 1) + new_capacity = 100; + void** new_items = malloc(new_capacity * sizeof(void*)); + if (new_items == NULL) { + perror("ERROR: Could not allocate memory for doubling capacity of queue."); + return false; + } + for (int i = 0; i < queue->size; i++) + new_items[i] = queue->items[(i + queue->front) % queue->capacity]; + free(queue->items); + queue->items = new_items; + queue->front = 0; + queue->rear = queue->size; + queue->capacity = new_capacity; + return true; +} + +bool queue_enqueue(Queue* queue, void* item) { + if (queue == NULL || item == NULL) + return false; + if (queue_is_full(queue)) { + if (!queue_double_capacity(queue)) + return false; + } + queue->items[queue->rear] = item; + queue->rear = (queue->rear + 1) % queue->capacity; + queue->size++; + return true; +} + +bool queue_enqueue_multithreaded(Queue* queue, void* item, mtx_t* mutex, cnd_t* condition_not_empty, + cnd_t* condition_not_full) { + mtx_lock(mutex); + while (queue_is_full(queue)) + cnd_wait(condition_not_full, mutex); + bool ok = queue_enqueue(queue, item); + cnd_signal(condition_not_empty); + mtx_unlock(mutex); + return ok; +} + bool queue_enqueue_multithreaded_cancel(Queue* queue, void* item, mtx_t* mutex, cnd_t* condition_not_empty, cnd_t* condition_not_full, const atomic_bool* cancelled) { - mtx_lock(mutex); + mtx_lock(mutex); while (queue_is_full(queue) && (cancelled == NULL || !atomic_load(cancelled))) - cnd_wait(condition_not_full, mutex); + cnd_wait(condition_not_full, mutex); if (cancelled != NULL && atomic_load(cancelled)) { - mtx_unlock(mutex); - return false; - } - bool ok = queue_enqueue(queue, item); - cnd_signal(condition_not_empty); - mtx_unlock(mutex); - return ok; -} - -void* queue_dequeue(Queue* queue) { - if (queue == NULL || queue_is_empty(queue)) { - perror("ERROR: Could not dequeue from null or empty queue."); - return NULL; - } - - void* item = queue->items[queue->front]; - queue->items[queue->front] = NULL; - queue->front = (queue->front + 1) % queue->capacity; - queue->size--; - return item; -} - -void* queue_dequeue_multithreaded(Queue* queue, mtx_t* mutex, cnd_t* condition_not_empty, - cnd_t* condition_not_full, const bool* other_thread_done) { - mtx_lock(mutex); - while (queue_is_empty(queue) && !*other_thread_done) - cnd_wait(condition_not_empty, mutex); - if (queue_is_empty(queue) && *other_thread_done) { - mtx_unlock(mutex); - return NULL; - } - void* item = queue_dequeue(queue); - cnd_signal(condition_not_full); - mtx_unlock(mutex); - return item; -} + mtx_unlock(mutex); + return false; + } + bool ok = queue_enqueue(queue, item); + cnd_signal(condition_not_empty); + mtx_unlock(mutex); + return ok; +} + +void* queue_dequeue(Queue* queue) { + if (queue == NULL || queue_is_empty(queue)) { + perror("ERROR: Could not dequeue from null or empty queue."); + return NULL; + } + + void* item = queue->items[queue->front]; + queue->items[queue->front] = NULL; + queue->front = (queue->front + 1) % queue->capacity; + queue->size--; + return item; +} + +void* queue_dequeue_multithreaded(Queue* queue, mtx_t* mutex, cnd_t* condition_not_empty, + cnd_t* condition_not_full, const bool* other_thread_done) { + mtx_lock(mutex); + while (queue_is_empty(queue) && !*other_thread_done) + cnd_wait(condition_not_empty, mutex); + if (queue_is_empty(queue) && *other_thread_done) { + mtx_unlock(mutex); + return NULL; + } + void* item = queue_dequeue(queue); + cnd_signal(condition_not_full); + mtx_unlock(mutex); + return item; +}