#include "threadpool.h" #include "arena.h" #include "shared.h" #include "ll.h" #include #include #include #include /* Mutex: Lock a shared resource. Used to prevent race conditions when accessing / modifying some shared resource. A lock must // always be followed by an unlock // Semaphore: Send / wait on a signal; solves the consumer/producer problem. A function that sends should never wait, and a // function that waits should never send */ // Pair some data with a mutex. Specifically a way to deal with mutices easier, not for data storage (mtxpair_free does not free the `(void*)data` member) typedef struct mtxp { void *data; mtx_t *mtx; } mtxpair; mtxpair * mtxpair_init(void * const data, int type) { mtxpair *mtxp = malloc(1 * sizeof(*mtxp)); if(!mtxp) return NULL; // Make room for the mutex mtxp->mtx = malloc(1 * sizeof(*mtxp->mtx)); if(!mtxp->mtx) { free(mtxp); return NULL; } // Init the mutex if(mtx_init(mtxp->mtx, type) == thrd_error) { free(mtxp->mtx); free(mtxp); RETURNWERR(errno, NULL); } mtxp->data = data; return mtxp; } void mtxpair_free(mtxpair *mp) { if(!mp) return; mtx_destroy(mp->mtx); free(mp->mtx); free(mp); return; } int mtxpair_setdata(mtxpair * const mp, void * const data) { if(!mp) RETURNWERR(EINVAL, -1); mp->data = data; return 0; } // thrd_create which calls mtx_lock/unlock on `arg` automatically int thrd_createwmx(thrd_t * const thr, thrd_start_t func, mtxpair * const mtxd) { if(!thr) RETURNWERR(EINVAL, thrd_error); if(!func) RETURNWERR(EINVAL, thrd_error); if(!mtxd) RETURNWERR(EINVAL, thrd_error); if(mtx_lock(mtxd->mtx) == thrd_error) {RETURNWERR(errno, thrd_error);} int retval = thrd_create(thr, func, mtxd->data); if(mtx_unlock(mtxd->mtx) == thrd_error) {RETURNWERR(errno, thrd_error);} return retval; } /* Ok, after doing a little more research, the best way to do this is probaby via a producer/consumer architecture. Spawn a bunch of // threads waiting on a queue (via semaphore) and when one is notified pop a task of the queue and execute it. In this case, the // producer would be the filesystem scanner funciton providing new files to encrypt, and the consumers would be threads waiting // to encrypt them */ // Threadpool: // Array of threads // Task Queue // Readiness semaphore // Linked List of Tasks // Task: // int (*callback)(void*) // void *arg // Here's a good reference of this implemented in C++ using Boost: https://gist.github.com/mikeando/482342 typedef int (*task_callback)(void*); typedef struct task { task_callback cb; void *arg; } task; typedef struct cq { dlinkedlist *list; mtx_t *mutex; cnd_t *conditional; } cqueue; typedef struct tp { thrd_t **threads; int nthreads; cqueue *taskqueue; } threadpool; task * task_init(task_callback cb, void *arg) { if(cb == NULL) RETURNWERR(EINVAL, NULL); task *task = VALLOC(1, sizeof(*task)); if(!task) return NULL; task->cb = cb; task->arg = arg; return task; } void task_free(task *ts) { if(!ts) return; free(ts); // Not making any assumptions about the data in the task return; } static void ___ucleanup_mtxd(void *mtx) { if(!mtx) return; mtx_destroy((mtx_t *)mtx); return; } static void ___ucleanup_cndd(void *cnd) { if(!cnd) return; cnd_destroy((cnd_t *)cnd); return; } static void ___ucleanup_dll(void *dll) { if(!dll) return; dlinkedlist_free((dlinkedlist *)dll); return; } cqueue * cqueue_init(int mtx_type) { unsigned char flag = 0; cleanup_CREATE(10); cqueue *cq = VALLOC(1, sizeof(*cq)); if(!cq) return NULL; cleanup_REGISTER(free, cq); cq->mutex = VALLOC(1, sizeof(*(cq->mutex))); if(!(cq->mutex)) flag++; cleanup_CNDREGISTER(flag, free, cq->mutex); if(!flag && mtx_init(cq->mutex, mtx_type) != thrd_success) flag++; cleanup_CNDREGISTER(flag, ___ucleanup_mtxd, cq->mutex); if(!flag && !(cq->conditional = VALLOC(1, sizeof(*(cq->conditional))))) flag++; cleanup_CNDREGISTER(flag, free, cq->conditional); if(!flag && cnd_init(cq->conditional) != thrd_success) flag++; cleanup_CNDREGISTER(flag, ___ucleanup_cndd, cq->conditional); cq->list = dlinkedlist_init(); if(!flag && !cq->list) flag++; cleanup_CNDREGISTER(flag, ___ucleanup_dll, cq->list); if(flag) cleanup_fire(&__CLEANUP); // This implementation is better and should be far less error prone than the thing I did earlier, but it would be nicer if C had anonymous functions return cq; } void cqueue_free(cqueue *cq) { if(!cq) return; dlinkedlist_free(cq->list); cnd_destroy(cq->conditional); mtx_destroy(cq->mutex); free(cq->conditional); free(cq->mutex); free(cq); return; } int cqueue_append(cqueue * const cq, task *tsk) { if(!cq || !tsk) RETURNWERR(EINVAL, -1); mtx_lock(cq->mutex); dlinkedlist_append(cq->list, tsk, free); mtx_unlock(cq->mutex); cnd_signal(cq->conditional); return 0; } int cqueue_prepend(cqueue * const cq, task *tsk) { if(!cq || !tsk) RETURNWERR(EINVAL, -1); mtx_lock(cq->mutex); dlinkedlist_prepend(cq->list, tsk, free); mtx_unlock(cq->mutex); cnd_signal(cq->conditional); return 0; } int cqueue_insert(cqueue * const cq, task *tsk, int index) { if(!cq || !tsk || index < 0) // Can't check to see if the index is too high without locking the mutex first RETURNWERR(EINVAL, -1); mtx_lock(cq->mutex); dlinkedlist_insert(cq->list, tsk, free, index); mtx_unlock(cq->mutex); cnd_signal(cq->conditional); return 0; } int cqueue_size(cqueue const * const cq) { if(!cq) RETURNWERR(EINVAL, -1); mtx_lock(cq->mutex); int retval = dlinkedlist_size(cq->list); mtx_unlock(cq->mutex); return retval; } int cqueue_isempty(cqueue const * const cq) { return (cqueue_size(cq) == 0); } int cqueue_trypop(cqueue * const cq, task **ret) { if(!cq || !ret || !*ret) RETURNWERR(EINVAL, -1); int retval = 0; mtx_lock(cq->mutex); if(!dlinkedlist_isempty(cq->list)) { *ret = (task*)dlinkedlist_poplast(cq->list); retval = 1; } mtx_unlock(cq->mutex); return retval; } int cqueue_waitpop(cqueue * const cq, task **ret) { if(!cq || !ret || !*ret) RETURNWERR(EINVAL, -1); mtx_lock(cq->mutex); while(!dlinkedlist_isempty(cq->list)) cnd_wait(cq->conditional, cq->mutex); // Unlocks mutex while waiting, acquires lock once waiting is done *ret = dlinkedlist_poplast(cq->list); mtx_unlock(cq->mutex); return 0; }