thread_queue.c

Go to the documentation of this file.

// SPDX-FileCopyrightText: 2022 deroad <wargio@libero.it>
// SPDX-License-Identifier: LGPL-3.0-only
 
#include <rz_th.h>
#include "thread.h"
 
struct rz_th_queue_t {
    RzThreadLock *lock;
    RzThreadCond *cond;
    size_t max_size;
    RzList *list;
};
 
RZ_API RZ_OWN RzThreadQueue *rz_th_queue_new(size_t max_size, RZ_NULLABLE RzListFree qfree) {
    RzThreadQueue *queue = RZ_NEW0(RzThreadQueue);
    if (!queue) {
        return NULL;
    }
 
    queue->max_size = max_size;
    queue->list = rz_list_newf(qfree);
    queue->lock = rz_th_lock_new(false);
    queue->cond = rz_th_cond_new();
    if (!queue->list || !queue->lock || !queue->cond) {
        rz_th_queue_free(queue);
        return NULL;
    }
 
    return queue;
}
 
RZ_API void rz_th_queue_free(RZ_NULLABLE RzThreadQueue *queue) {
    if (!queue) {
        return;
    }
 
    rz_list_free(queue->list);
    rz_th_lock_free(queue->lock);
    rz_th_cond_free(queue->cond);
    free(queue);
}
 
RZ_API bool rz_th_queue_push(RZ_NONNULL RzThreadQueue *queue, RZ_NONNULL void *user, bool tail) {
    rz_return_val_if_fail(queue && user, false);
 
    bool added = false;
    rz_th_lock_enter(queue->lock);
    if (!queue->max_size || rz_list_length(queue->list) < queue->max_size) {
        if (tail) {
            added = rz_list_append(queue->list, user) != NULL;
        } else {
            added = rz_list_prepend(queue->list, user) != NULL;
        }
    }
    if (added) {
        rz_th_cond_signal(queue->cond);
    }
    rz_th_lock_leave(queue->lock);
    return added;
}
 
RZ_API RZ_OWN void *rz_th_queue_pop(RZ_NONNULL RzThreadQueue *queue, bool tail) {
    rz_return_val_if_fail(queue, NULL);
 
    void *user = NULL;
    rz_th_lock_enter(queue->lock);
    if (tail) {
        user = rz_list_pop(queue->list);
    } else {
        user = rz_list_pop_head(queue->list);
    }
    rz_th_lock_leave(queue->lock);
    return user;
}
 
RZ_API RZ_OWN void *rz_th_queue_wait_pop(RZ_NONNULL RzThreadQueue *queue, bool tail) {
    rz_return_val_if_fail(queue, NULL);
 
    void *user = NULL;
    rz_th_lock_enter(queue->lock);
    if (rz_list_empty(queue->list)) {
        rz_th_cond_wait(queue->cond, queue->lock);
    }
    if (tail) {
        user = rz_list_pop(queue->list);
    } else {
        user = rz_list_pop_head(queue->list);
    }
    rz_th_lock_leave(queue->lock);
    return user;
}
 
RZ_API bool rz_th_queue_is_empty(RZ_NONNULL RzThreadQueue *queue) {
    rz_return_val_if_fail(queue, false);
 
    rz_th_lock_enter(queue->lock);
    bool is_empty = rz_list_empty(queue->list);
    rz_th_lock_leave(queue->lock);
    return is_empty;
}
 
RZ_API bool rz_th_queue_is_full(RZ_NONNULL RzThreadQueue *queue) {
    rz_return_val_if_fail(queue, false);
 
    rz_th_lock_enter(queue->lock);
    bool is_full = queue->max_size != RZ_THREAD_QUEUE_UNLIMITED && rz_list_length(queue->list) >= queue->max_size;
    rz_th_lock_leave(queue->lock);
    return is_full;
}