diff --git a/binrz/rz-test/rz-test.c b/binrz/rz-test/rz-test.c index bae314f746..1ded570ed7 100644 --- a/binrz/rz-test/rz-test.c +++ b/binrz/rz-test/rz-test.c @@ -51,7 +51,7 @@ typedef struct rz_test_state_t { RzPVector results; } RzTestState; -static RzThreadFunctionRet worker_th(RzThread *th); +static void *worker_th(RzTestState *state); static void print_state(RzTestState *state, ut64 prev_completed); static void print_log(RzTestState *state, ut64 prev_completed, ut64 prev_paths_completed); static void interact(RzTestState *state); @@ -509,7 +509,7 @@ int rz_test_main(int argc, const char **argv) { rz_pvector_init(&workers, NULL); int i; for (i = 0; i < workers_count; i++) { - RzThread *th = rz_th_new(worker_th, &state, 0); + RzThread *th = rz_th_new((RzThreadFunction)worker_th, &state); if (!th) { eprintf("Failed to start thread.\n"); rz_th_lock_leave(state.lock); @@ -650,8 +650,7 @@ static void test_result_to_json(PJ *pj, RzTestResultInfo *result) { pj_end(pj); } -static RzThreadFunctionRet worker_th(RzThread *th) { - RzTestState *state = rz_th_get_user(th); +static void *worker_th(RzTestState *state) { rz_th_lock_enter(state->lock); while (true) { if (rz_pvector_empty(&state->queue)) { @@ -706,7 +705,7 @@ static RzThreadFunctionRet worker_th(RzThread *th) { rz_th_cond_signal(state->cond); } rz_th_lock_leave(state->lock); - return RZ_TH_STOP; + return NULL; } static void print_diff(const char *actual, const char *expected, const char *regexp) { diff --git a/librz/core/basefind.c b/librz/core/basefind.c index cf1cfed7bd..cccf527864 100644 --- a/librz/core/basefind.c +++ b/librz/core/basefind.c @@ -37,6 +37,11 @@ typedef struct basefind_thread_data_t { BaseFindArray *array; } BaseFindThreadData; +typedef struct basefind_thread_cons_t { + bool progress; + RzThreadPool *pool; +} BaseFindThreadCons; + static RzBinFile *basefind_new_bin_file(RzCore *core) { // Copied from cbin.c -> rz_core_bin_whole_strings_print // TODO: manually creating an RzBinFile like this is a hack and abuse of RzBin API @@ -200,8 +205,7 @@ static int basefind_score_compare(const RzBaseFindScore *a, const RzBaseFindScor return 1; } -static RzThreadFunctionRet basefind_thread_runner(RzThread *th) { - BaseFindThreadData *bftd = (BaseFindThreadData *)rz_th_get_user(th); +static void *basefind_thread_runner(BaseFindThreadData *bftd) { RzBaseFindScore *pair = NULL; BaseFindData bfd; ut64 base; @@ -240,17 +244,57 @@ static RzThreadFunctionRet basefind_thread_runner(RzThread *th) { RZ_LOG_DEBUG("basefind: possible candidate at 0x%016" PFMT64x " with score of %u\n", base, bfd.score); rz_th_lock_leave(bftd->lock); } + bftd->current = base; - return RZ_TH_STOP; + return NULL; +} + +// this thread does not care about thread-safety since it only prints +// data that will always be available during its lifetime. +static void *basefind_thread_cons(BaseFindThreadCons *th_cons) { + bool progress = th_cons->progress; + RzThreadPool *pool = th_cons->pool; + size_t pool_size = rz_th_pool_size(pool); + rz_cons_flush(); + int begin_line = rz_cons_get_cur_line(); + do { + if (progress) { + rz_cons_gotoxy(1, begin_line); + for (ut32 i = 0; i < pool_size; ++i) { + RzThread *th = rz_th_pool_get_thread(pool, i); + if (!th) { + continue; + } + BaseFindThreadData *bftd = rz_th_get_user(th); + ut32 perc = ((bftd->current - bftd->base_start) * 100) / (bftd->base_end - bftd->base_start); + if (perc > 100) { + perc = 100; + } + rz_cons_printf("basefind: thread %u: 0x%08" PFMT64x " / 0x%08" PFMT64x " %u%%\n", i, bftd->current, bftd->base_end, perc); + } + rz_cons_flush(); + begin_line = rz_cons_get_cur_line() - pool_size; + } + rz_sys_usleep(100000); + if (rz_cons_is_breaked()) { + rz_th_pool_kill(pool); + break; + } + } while (1); + return NULL; } static inline bool create_thread_interval(RzThreadPool *pool, BaseFindThreadData *bfd) { - RzThread *thread = rz_th_new(basefind_thread_runner, bfd, 0); + RzThread *thread = rz_th_new((RzThreadFunction)basefind_thread_runner, bfd); if (!thread) { - RZ_LOG_ERROR("basefind: cannot allocate BaseFindData\n"); + RZ_LOG_ERROR("basefind: cannot allocate RzThread\n"); + return false; + } else if (!rz_th_pool_add_thread(pool, thread)) { + RZ_LOG_ERROR("basefind: cannot add thread to pool\n"); + rz_th_free(thread); return false; } - return rz_th_pool_add_thread(pool, thread); + return true; } /** @@ -275,7 +319,7 @@ RZ_API RZ_OWN RzList *rz_basefind(RZ_NONNULL RzCore *core, ut32 pointer_size) { HtUU *pointers = NULL; ut64 base_start = 0, base_end = 0, base_inc = 0; ut32 score_min = 0; - size_t max_threads = 0; + size_t max_threads = 0, pool_size = 1; RzThreadPool *pool = NULL; RzThreadLock *lock = NULL; bool progress = false; @@ -330,9 +374,10 @@ RZ_API RZ_OWN RzList *rz_basefind(RZ_NONNULL RzCore *core, ut32 pointer_size) { pool = rz_th_pool_new(max_threads); if (!pool) { - RZ_LOG_ERROR("basefind: cannot thread pool.\n"); + RZ_LOG_ERROR("basefind: cannot allocate thread pool.\n"); goto rz_basefind_end; } + pool_size = rz_th_pool_size(pool); lock = rz_th_lock_new(false); if (!lock) { @@ -340,14 +385,15 @@ RZ_API RZ_OWN RzList *rz_basefind(RZ_NONNULL RzCore *core, ut32 pointer_size) { goto rz_basefind_end; } - RZ_LOG_VERBOSE("basefind: using %u threads\n", (ut32)pool->size); + RZ_LOG_VERBOSE("basefind: using %u threads\n", (ut32)pool_size); ut64 io_size = rz_io_size(core->io); - ut64 sector_size = (((base_end - base_start) + pool->size - 1) / pool->size); - for (size_t i = 0; i < pool->size; ++i) { + ut64 sector_size = (((base_end - base_start) + pool_size - 1) / pool_size); + for (size_t i = 0; i < pool_size; ++i) { BaseFindThreadData *bftd = RZ_NEW(BaseFindThreadData); if (!bftd) { RZ_LOG_ERROR("basefind: cannot allocate BaseFindThreadData.\n"); + rz_th_pool_kill(pool); goto rz_basefind_end; } bftd->base_inc = base_inc; @@ -362,41 +408,38 @@ RZ_API RZ_OWN RzList *rz_basefind(RZ_NONNULL RzCore *core, ut32 pointer_size) { bftd->array = array; if (!create_thread_interval(pool, bftd)) { free(bftd); + rz_th_pool_kill(pool); goto rz_basefind_end; } } - rz_sys_sleep(1); - int line = rz_cons_get_cur_line(); - do { - if (progress) { - rz_cons_gotoxy(1, line); - for (ut32 i = 0; i < pool->size; ++i) { - BaseFindThreadData *bftd = rz_th_get_user(pool->threads[i]); - ut32 perc = ((bftd->current - bftd->base_start) * 100) / (bftd->base_end - bftd->base_start); - if (perc > 100) { - perc = 100; - } - rz_cons_printf("basefind: thread %u: 0x%08" PFMT64x " / 0x%08" PFMT64x " %u%%\n", i, bftd->current, bftd->base_end, perc); - } - rz_cons_flush(); - } - rz_sys_sleep(1); - if (rz_cons_is_breaked()) { - RZ_LOG_WARN("basefind: catched CTRL-C. returning scores\n"); - rz_th_pool_kill(pool, true); - break; - } - } while (!rz_th_pool_wait_async(pool)); + BaseFindThreadCons th_cons; + th_cons.progress = progress; + th_cons.pool = pool; + + RzThread *cons_thread = rz_th_new((RzThreadFunction)basefind_thread_cons, &th_cons); + if (!cons_thread) { + rz_th_pool_kill(pool); + goto rz_basefind_end; + } + rz_th_pool_wait(pool); + if (progress) { + // ensure to print the 100% + rz_sys_usleep(100000); + } + rz_th_kill(cons_thread); + rz_th_free(cons_thread); rz_list_sort(scores, (RzListComparator)basefind_score_compare); rz_basefind_end: if (pool) { - for (ut32 i = 0; i < pool->size; ++i) { - if (pool->threads[i]) { - free(rz_th_get_user(pool->threads[i])); + for (ut32 i = 0; i < pool_size; ++i) { + RzThread *th = rz_th_pool_get_thread(pool, i); + if (!th) { + continue; } + free(rz_th_get_user(th)); } rz_th_pool_free(pool); } diff --git a/librz/core/rtr.c b/librz/core/rtr.c index 5ecc95f1d3..82508fab3b 100644 --- a/librz/core/rtr.c +++ b/librz/core/rtr.c @@ -58,10 +58,18 @@ typedef struct { RZ_API void rz_core_wait(RzCore *core) { rz_cons_singleton()->context->breaked = true; - rz_th_kill(httpthread, true); - rz_th_kill(rapthread, true); - rz_th_wait(httpthread); - rz_th_wait(rapthread); + if (httpthread) { + rz_th_kill(httpthread); + } + if (rapthread) { + rz_th_kill(rapthread); + } + if (httpthread) { + rz_th_wait(httpthread); + } + if (rapthread) { + rz_th_wait(rapthread); + } } static void http_logf(RzCore *core, const char *fmt, ...) { @@ -841,15 +849,13 @@ static bool rz_core_rtr_rap_run(RzCore *core, const char *input) { // rz_core_cmdf (core, "o rap://%s", input); } -static RzThreadFunctionRet rz_core_rtr_rap_thread(RzThread *th) { - if (!th) { - return false; - } - RapThread *rt = rz_th_get_user(th); +static void *rz_core_rtr_rap_thread(RapThread *rt) { if (!rt || !rt->core) { return false; } - return rz_core_rtr_rap_run(rt->core, rt->input) ? RZ_TH_REPEAT : RZ_TH_STOP; + while (rz_core_rtr_rap_run(rt->core, rt->input)) + ; + return NULL; } RZ_API void rz_core_rtr_cmd(RzCore *core, const char *input) { @@ -877,18 +883,26 @@ RZ_API void rz_core_rtr_cmd(RzCore *core, const char *input) { eprintf("This is experimental and probably buggy. Use at your own risk\n"); } else { // TODO: use tasks - RapThread *RT = RZ_NEW0(RapThread); - if (RT) { - RT->core = core; - RT->input = strdup(input + 1); - // RapThread rt = { core, strdup (input + 1) }; - rapthread = rz_th_new(rz_core_rtr_rap_thread, RT, false); - int cpuaff = (int)rz_config_get_i(core->config, "cfg.cpuaffinity"); - rz_th_setaffinity(rapthread, cpuaff); - rz_th_setname(rapthread, "rapthread"); - rz_th_start(rapthread, true); - eprintf("Background rap server started.\n"); + RapThread *rap_th = RZ_NEW0(RapThread); + if (!rap_th) { + RZ_LOG_ERROR("cannot allocate RapThread\n"); + return; } + rap_th->core = core; + rap_th->input = strdup(input + 1); + // RapThread rt = { core, strdup (input + 1) }; + rapthread = rz_th_new((RzThreadFunction)rz_core_rtr_rap_thread, rap_th); + if (!rap_th) { + RZ_LOG_ERROR("cannot spawn the RzThread\n"); + return; + } + int cpuaff = (int)rz_config_get_i(core->config, "cfg.cpuaffinity"); + if (cpuaff) { + // modify the affinity only when the flag is actually set. + rz_th_set_affinity(rapthread, cpuaff); + } + rz_th_set_name(rapthread, "rapthread"); + RZ_LOG_WARN("Background rap server started.\n"); } return; } diff --git a/librz/core/rtr_http.c b/librz/core/rtr_http.c index d2f585fc19..f176544380 100644 --- a/librz/core/rtr_http.c +++ b/librz/core/rtr_http.c @@ -489,24 +489,23 @@ the_end : { } #if 0 -static RzThreadFunctionRet rz_core_rtr_http_thread (RzThread *th) { - if (!th) { - return false; - } - HttpThread *ht = th->user; +static void *rz_core_rtr_http_thread (HttpThread *ht) { if (!ht || !ht->core) { return false; } - int ret = rz_core_rtr_http_run (ht->core, ht->launch, ht->browse, ht->path); - RZ_FREE (ht->path); - if (ret) { - int p = rz_config_get_i (ht->core->config, "http.port"); - rz_config_set_i (ht->core->config, "http.port", p + 1); - if (p >= rz_config_get_i (ht->core->config, "http.maxport")) { - return RZ_TH_STOP; + int ret = 0; + do { + ret = rz_core_rtr_http_run (ht->core, ht->launch, ht->browse, ht->path); + RZ_FREE (ht->path); + if (ret) { + int p = rz_config_get_i (ht->core->config, "http.port"); + rz_config_set_i (ht->core->config, "http.port", p + 1); + if (p >= rz_config_get_i (ht->core->config, "http.maxport")) { + break; + } } - } - return ret ? RZ_TH_REPEAT : RZ_TH_STOP; + } while(ret); + return NULL; } #endif @@ -546,11 +545,8 @@ RZ_API int rz_core_rtr_http(RzCore *core, int launch, int browse, const char *pa ht->launch = launch; ht->browse = browse; ht->path = strdup (tpath); - httpthread = rz_th_new (rz_core_rtr_http_thread, ht, false); - if (httpthread) { - rz_th_setname (httpthread, "httpthread"); - } - rz_th_start (httpthread, true); + httpthread = rz_th_new ((RzThreadFunction)rz_core_rtr_http_thread, ht); + rz_th_setname (httpthread, "httpthread"); eprintf ("Background http server started.\n"); } return 0; diff --git a/librz/core/task.c b/librz/core/task.c index 1a4513c199..cd91025c1f 100644 --- a/librz/core/task.c +++ b/librz/core/task.c @@ -148,8 +148,10 @@ static void task_free(RzCoreTask *task) { if (task->runner_free) { task->runner_free(task->runner_user); } - rz_th_wait(task->thread); - rz_th_free(task->thread); + if (task->thread) { + rz_th_wait(task->thread); + rz_th_free(task->thread); + } rz_th_sem_free(task->running_sem); rz_th_cond_free(task->dispatch_cond); rz_th_lock_free(task->dispatch_lock); @@ -343,7 +345,7 @@ static void task_end(RzCoreTask *t) { rz_core_task_schedule(t, RZ_CORE_TASK_STATE_DONE); } -static RzThreadFunctionRet task_run(RzCoreTask *task) { +static void *task_run_thread(RzCoreTask *task) { RzCoreTaskScheduler *sched = task->sched; task_wakeup(task); @@ -366,12 +368,7 @@ nonstart: } tasks_lock_leave(sched, &old_sigset); - return RZ_TH_STOP; -} - -static RzThreadFunctionRet task_run_thread(RzThread *th) { - RzCoreTask *task = (RzCoreTask *)rz_th_get_user(th); - return task_run(task); + return NULL; } RZ_API void rz_core_task_enqueue(RzCoreTaskScheduler *scheduler, RzCoreTask *task) { @@ -387,7 +384,7 @@ RZ_API void rz_core_task_enqueue(RzCoreTaskScheduler *scheduler, RzCoreTask *tas rz_th_sem_wait(task->running_sem); } rz_list_append(scheduler->tasks, task); - task->thread = rz_th_new(task_run_thread, task, 0); + task->thread = rz_th_new((RzThreadFunction)task_run_thread, task); tasks_lock_leave(scheduler, &old_sigset); } @@ -417,7 +414,7 @@ RZ_API void rz_core_task_enqueue_oneshot(RzCoreTaskScheduler *scheduler, RzCoreT RZ_API int rz_core_task_run_sync(RzCoreTaskScheduler *scheduler, RzCoreTask *task) { task->thread = NULL; - return task_run(task); + return task_run_thread(task) != NULL; } /* begin running stuff synchronously on the main task */ diff --git a/librz/include/rz_th.h b/librz/include/rz_th.h index 3fe1e6c5ae..bab8fed853 100644 --- a/librz/include/rz_th.h +++ b/librz/include/rz_th.h @@ -1,3 +1,7 @@ +// SPDX-FileCopyrightText: 2009-2017 pancake +// SPDX-FileCopyrightText: 2021-2022 deroad +// SPDX-License-Identifier: LGPL-3.0-only + #ifndef RZ_TH_H #define RZ_TH_H @@ -5,72 +9,70 @@ #undef _GNU_SOURCE #endif #define _GNU_SOURCE -#include "rz_types.h" +#include +#include #ifdef __cplusplus extern "C" { #endif -typedef enum { - RZ_TH_FREED = -1, - RZ_TH_STOP = 0, - RZ_TH_REPEAT = 1 -} RzThreadFunctionRet; -#define RZ_TH_FUNCTION(x) RzThreadFunctionRet (*x)(struct rz_th_t *) +#define RZ_THREAD_POOL_ALL_CORES (0) +#define RZ_THREAD_QUEUE_UNLIMITED (0) typedef struct rz_th_sem_t RzThreadSemaphore; typedef struct rz_th_lock_t RzThreadLock; typedef struct rz_th_cond_t RzThreadCond; typedef struct rz_th_t RzThread; - -#define RZ_THREAD_POOL_ALL_CORES (0) - -typedef struct rz_th_pool_t { - size_t size; - RzThread **threads; -} RzThreadPool; +typedef struct rz_th_pool_t RzThreadPool; +typedef struct rz_th_queue_t RzThreadQueue; +typedef void *(*RzThreadFunction)(void *user); #ifdef RZ_API -RZ_API RzThread *rz_th_new(RZ_TH_FUNCTION(fun), void *user, int delay); -RZ_API void *rz_th_get_user(RzThread *th); -RZ_API bool rz_th_start(RzThread *th, int enable); -RZ_API bool rz_th_wait(RzThread *th); -RZ_API bool rz_th_wait_async(RzThread *th); -RZ_API void rz_th_break(RzThread *th); -RZ_API void rz_th_free(RzThread *th); -RZ_API void rz_th_kill_free(RzThread *th); -RZ_API bool rz_th_kill(RzThread *th, bool force); -RZ_API bool rz_th_setname(RzThread *th, const char *name); -RZ_API bool rz_th_getname(RzThread *th, char *name, size_t len); -RZ_API bool rz_th_setaffinity(RzThread *th, int cpuid); +RZ_API RZ_OWN RzThread *rz_th_new(RZ_NONNULL RzThreadFunction function, RZ_NULLABLE void *user); +RZ_API RZ_OWN void *rz_th_get_user(RZ_NONNULL RzThread *th); +RZ_API RZ_OWN void *rz_th_get_retv(RZ_NONNULL RzThread *th); +RZ_API bool rz_th_wait(RZ_NONNULL RzThread *th); +RZ_API void rz_th_free(RZ_NULLABLE RzThread *th); +RZ_API void rz_th_kill(RZ_NONNULL RzThread *th); +RZ_API void rz_th_kill_free(RZ_NONNULL RzThread *th); +RZ_API bool rz_th_set_name(RZ_NONNULL RzThread *th, RZ_NONNULL const char *name); +RZ_API bool rz_th_get_name(RZ_NONNULL RzThread *th, RZ_NONNULL RZ_OUT char *name, size_t len); +RZ_API bool rz_th_set_affinity(RZ_NONNULL RzThread *th, int cpuid); +RZ_API bool rz_th_yield(void); -RZ_API RzThreadSemaphore *rz_th_sem_new(unsigned int initial); -RZ_API void rz_th_sem_free(RzThreadSemaphore *sem); -RZ_API void rz_th_sem_post(RzThreadSemaphore *sem); -RZ_API void rz_th_sem_wait(RzThreadSemaphore *sem); +RZ_API RZ_OWN RzThreadSemaphore *rz_th_sem_new(unsigned int initial); +RZ_API void rz_th_sem_free(RZ_NULLABLE RzThreadSemaphore *sem); +RZ_API void rz_th_sem_post(RZ_NONNULL RzThreadSemaphore *sem); +RZ_API void rz_th_sem_wait(RZ_NONNULL RzThreadSemaphore *sem); -RZ_API RzThreadLock *rz_th_lock_new(bool recursive); -RZ_API int rz_th_lock_wait(RzThreadLock *th); -RZ_API int rz_th_lock_tryenter(RzThreadLock *thl); -RZ_API int rz_th_lock_enter(RzThreadLock *thl); -RZ_API int rz_th_lock_leave(RzThreadLock *thl); -RZ_API void *rz_th_lock_free(RzThreadLock *thl); +RZ_API RZ_OWN RzThreadLock *rz_th_lock_new(bool recursive); +RZ_API bool rz_th_lock_tryenter(RZ_NONNULL RzThreadLock *thl); +RZ_API void rz_th_lock_enter(RZ_NONNULL RzThreadLock *thl); +RZ_API void rz_th_lock_leave(RZ_NONNULL RzThreadLock *thl); +RZ_API void rz_th_lock_free(RZ_NULLABLE RzThreadLock *thl); -RZ_API RzThreadCond *rz_th_cond_new(void); -RZ_API void rz_th_cond_signal(RzThreadCond *cond); -RZ_API void rz_th_cond_signal_all(RzThreadCond *cond); -RZ_API void rz_th_cond_wait(RzThreadCond *cond, RzThreadLock *lock); -RZ_API void rz_th_cond_free(RzThreadCond *cond); +RZ_API RZ_OWN RzThreadCond *rz_th_cond_new(void); +RZ_API void rz_th_cond_signal(RZ_NONNULL RzThreadCond *cond); +RZ_API void rz_th_cond_signal_all(RZ_NONNULL RzThreadCond *cond); +RZ_API void rz_th_cond_wait(RZ_NONNULL RzThreadCond *cond, RZ_NONNULL RzThreadLock *lock); +RZ_API void rz_th_cond_free(RZ_NULLABLE RzThreadCond *cond); RZ_API size_t rz_th_physical_core_number(); RZ_API RZ_OWN RzThreadPool *rz_th_pool_new(size_t max_threads); RZ_API void rz_th_pool_free(RZ_NULLABLE RzThreadPool *pool); RZ_API bool rz_th_pool_add_thread(RZ_NONNULL RzThreadPool *pool, RZ_NONNULL RzThread *thread); -RZ_API bool rz_th_pool_start(RZ_NONNULL RzThreadPool *pool, bool enable); +RZ_API RZ_BORROW RzThread *rz_th_pool_get_thread(RZ_NONNULL RzThreadPool *pool, size_t index); RZ_API bool rz_th_pool_wait(RZ_NONNULL RzThreadPool *pool); -RZ_API bool rz_th_pool_wait_async(RZ_NONNULL RzThreadPool *pool); -RZ_API bool rz_th_pool_kill(RZ_NONNULL RzThreadPool *pool, bool force); -RZ_API bool rz_th_pool_kill_free(RZ_NONNULL RzThreadPool *pool); +RZ_API bool rz_th_pool_kill(RZ_NONNULL RzThreadPool *pool); +RZ_API size_t rz_th_pool_size(RZ_NULLABLE RzThreadPool *pool); + +RZ_API RZ_OWN RzThreadQueue *rz_th_queue_new(size_t max_size, RZ_NULLABLE RzListFree qfree); +RZ_API void rz_th_queue_free(RZ_NULLABLE RzThreadQueue *queue); +RZ_API bool rz_th_queue_push(RZ_NONNULL RzThreadQueue *queue, RZ_NONNULL void *user, bool tail); +RZ_API RZ_OWN void *rz_th_queue_pop(RZ_NONNULL RzThreadQueue *queue, bool tail); +RZ_API RZ_OWN void *rz_th_queue_wait_pop(RZ_NONNULL RzThreadQueue *queue, bool tail); +RZ_API bool rz_th_queue_is_empty(RZ_NULLABLE RzThreadQueue *queue); +RZ_API bool rz_th_queue_is_full(RZ_NULLABLE RzThreadQueue *queue); #endif diff --git a/librz/socket/run.c b/librz/socket/run.c index a99978d720..99d6566305 100644 --- a/librz/socket/run.c +++ b/librz/socket/run.c @@ -719,9 +719,12 @@ static int redirect_socket_to_stdio(RzSocket *sock) { } #if __WINDOWS__ -static RzThreadFunctionRet exit_process(RzThread *th) { - // eprintf ("\nrz_run: Interrupted by timeout\n"); +static void *exit_process(void *user) { + int timeout = (int)(void *)user; + rz_sys_sleep(timeout); + RZ_LOG_DEBUG("rz_run: Interrupted by timeout\n"); exit(0); + return NULL; } #endif @@ -1065,7 +1068,7 @@ RZ_API int rz_run_config_env(RzRunProfile *p) { } #else if (p->_timeout_sig < 1 || p->_timeout_sig == 9) { - rz_th_new(exit_process, NULL, p->_timeout); + rz_th_new(exit_process, (void *)p->_timeout); } else { eprintf("timeout with signal not supported for this platform\n"); } diff --git a/librz/util/meson.build b/librz/util/meson.build index f9e0b478b3..3bb29e96bd 100644 --- a/librz/util/meson.build +++ b/librz/util/meson.build @@ -65,6 +65,8 @@ rz_util_sources = [ 'thread_cond.c', 'thread_lock.c', 'thread_sem.c', + 'thread_pool.c', + 'thread_queue.c', 'time.c', 'tree.c', 'ubase64.c', diff --git a/librz/util/subprocess.c b/librz/util/subprocess.c index 3c1970fdc5..0847a0f72a 100644 --- a/librz/util/subprocess.c +++ b/librz/util/subprocess.c @@ -554,7 +554,7 @@ static void handle_sigchld(int sig) { rz_xwrite(sigchld_pipe[1], &b, 1); } -static RzThreadFunctionRet sigchld_th(RzThread *th) { +static void *sigchld_th(void *th) { while (true) { ut8 b; ssize_t rd = read(sigchld_pipe[0], &b, 1); @@ -601,7 +601,7 @@ static RzThreadFunctionRet sigchld_th(RzThread *th) { subprocess_unlock(); } } - return RZ_TH_STOP; + return NULL; } RZ_API bool rz_subprocess_init(void) { @@ -615,7 +615,7 @@ RZ_API bool rz_subprocess_init(void) { rz_th_lock_free(subprocs_mutex); return false; } - sigchld_thread = rz_th_new(sigchld_th, NULL, 0); + sigchld_thread = rz_th_new(sigchld_th, NULL); if (!sigchld_thread) { rz_sys_pipe_close(sigchld_pipe[0]); rz_sys_pipe_close(sigchld_pipe[1]); diff --git a/librz/util/thread.c b/librz/util/thread.c index 15852b3b2f..a65e0427b1 100644 --- a/librz/util/thread.c +++ b/librz/util/thread.c @@ -1,66 +1,25 @@ -// SPDX-FileCopyrightText: 2009-2018 pancake +// SPDX-FileCopyrightText: 2020-2021 ret2libc +// SPDX-FileCopyrightText: 2020-2022 deroad +// SPDX-FileCopyrightText: 2022 GustavoLCR // SPDX-License-Identifier: LGPL-3.0-only #include #include "thread.h" -#if __APPLE__ -// Here to avoid polluting mach types macro redefinitions... -#include -#include -#endif - -#if __APPLE__ || __NetBSD__ || __FreeBSD__ || __OpenBSD__ || __DragonFly__ || __sun -#include -#include -#endif - -#if __sun -#include -#endif - -#if __HAIKU__ -#include -#include -#endif - -#if __WINDOWS__ -static DWORD WINAPI _rz_th_launcher(void *_th) { -#else -static void *_rz_th_launcher(void *_th) { -#endif - int ret; - RzThread *th = _th; - th->ready = true; - if (th->delay > 0) { - rz_sys_sleep(th->delay); - } else if (th->delay < 0) { - rz_th_lock_wait(th->lock); - } - rz_th_lock_enter(th->lock); - do { - rz_th_lock_leave(th->lock); - th->running = true; - ret = th->fun(th); - if (ret < 0) { - // th has been freed - return 0; - } - th->running = false; - rz_th_lock_enter(th->lock); - } while (ret); - rz_th_lock_leave(th->lock); +/* + * Main thread function, this function is meant to be + * hidden from the user which is using the C APIs. + */ +static RZ_TH_RET_T thread_main_function(void *_th) { #if HAVE_PTHREAD - pthread_exit(&ret); +#ifndef __ANDROID__ + pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL); + pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL); #endif - return 0; -} - -RZ_API int rz_th_push_task(struct rz_th_t *th, void *user) { - int ret = true; - th->user = user; - rz_th_lock_leave(th->lock); - return ret; +#endif + RzThread *th = (RzThread *)_th; + th->retv = th->function(th->user); + return NULL; } RZ_IPI RZ_TH_TID rz_th_self(void) { @@ -74,28 +33,38 @@ RZ_IPI RZ_TH_TID rz_th_self(void) { #endif } -RZ_API bool rz_th_setname(RzThread *th, const char *name) { +/** + * \brief Sets the name of the thread + * + * \param th The thread to rename + * \param name The name to assign to the thread + * + * \return On success returns true, otherwise false + */ +RZ_API bool rz_th_set_name(RZ_NONNULL RzThread *th, RZ_NONNULL const char *name) { + rz_return_val_if_fail(th && name, false); + #if defined(HAVE_PTHREAD_NP) && HAVE_PTHREAD_NP #if __linux__ || __sun if (pthread_setname_np(th->tid, name) != 0) { - eprintf("Failed to set thread name\n"); + RZ_LOG_ERROR("thread: Failed to set thread name\n"); return false; } #elif __APPLE__ && defined(MAC_OS_X_VERSION_10_6) if (pthread_setname_np(name) != 0) { - eprintf("Failed to set thread name\n"); + RZ_LOG_ERROR("thread: Failed to set thread name\n"); return false; } #elif __FreeBSD__ || __OpenBSD__ || __DragonFly__ || __sun pthread_set_name_np(th->tid, name); #elif __NetBSD__ if (pthread_setname_np(th->tid, "%s", (void *)name) != 0) { - eprintf("Failed to set thread name\n"); + RZ_LOG_ERROR("thread: Failed to set thread name\n"); return false; } #elif __HAIKU__ if (rename_thread((thread_id)th->tid, name) != B_OK) { - eprintf("Failed to set thread name\n"); + RZ_LOG_ERROR("thread: Failed to set thread name\n"); return false; } #else @@ -105,11 +74,22 @@ RZ_API bool rz_th_setname(RzThread *th, const char *name) { return true; } -RZ_API bool rz_th_getname(RzThread *th, char *name, size_t len) { +/** + * \brief Gets the name of the thread and writes it into the output buffer + * + * \param th The thread from which the name is taken + * \param name The output buffer name to use to copy the name + * \param len The output buffer length + * + * \return On success returns true, otherwise false + */ +RZ_API bool rz_th_get_name(RZ_NONNULL RzThread *th, RZ_NONNULL RZ_OUT char *name, size_t len) { + rz_return_val_if_fail(th && name && len > 0, false); + #if defined(HAVE_PTHREAD_NP) && HAVE_PTHREAD_NP #if __linux__ || __NetBSD__ || (__APPLE__ && defined(MAC_OS_X_VERSION_10_6)) || __sun if (pthread_getname_np(th->tid, name, len) != 0) { - eprintf("Failed to get thread name\n"); + RZ_LOG_ERROR("thread: Failed to get thread name\n"); return false; } #elif (__FreeBSD__ && __FreeBSD_version >= 1200000) || __DragonFly__ || (__OpenBSD__ && OpenBSD >= 201905) @@ -119,7 +99,7 @@ RZ_API bool rz_th_getname(RzThread *th, char *name, size_t len) { size_t flen = len < B_OS_NAME_LENGTH ? len : B_OS_NAME_LENGTH; if (get_thread_info((thread_id)th->tid, &ti) != B_OK) { - eprintf("Failed to get thread name\n"); + RZ_LOG_ERROR("thread: Failed to get thread name\n"); return false; } @@ -131,7 +111,17 @@ RZ_API bool rz_th_getname(RzThread *th, char *name, size_t len) { return true; } -RZ_API bool rz_th_setaffinity(RzThread *th, int cpuid) { +/** + * \brief Sets the thread cpu affinity + * + * \param th The thread to change the cpu affinity + * \param cpuid The cpuid to set to the thread. + * + * \return On success returns true, otherwise false. + */ +RZ_API bool rz_th_set_affinity(RZ_NONNULL RzThread *th, int cpuid) { + rz_return_val_if_fail(th, false); + #if __linux__ #if defined(__GLIBC__) && defined(__GLIBC_MINOR__) && (__GLIBC__ <= 2) && (__GLIBC_MINOR__ <= 2) // Old versions of GNU libc don't have this feature @@ -142,7 +132,7 @@ RZ_API bool rz_th_setaffinity(RzThread *th, int cpuid) { CPU_SET(cpuid, &c); if (sched_setaffinity((pid_t)(ut64)th->tid, sizeof(c), &c) != 0) { - eprintf("Failed to set cpu affinity\n"); + RZ_LOG_ERROR("thread: Failed to set cpu affinity\n"); return false; } #endif @@ -152,7 +142,7 @@ RZ_API bool rz_th_setaffinity(RzThread *th, int cpuid) { CPU_SET(cpuid, &c); if (pthread_setaffinity_np(th->tid, sizeof(c), &c) != 0) { - eprintf("Failed to set cpu affinity\n"); + RZ_LOG_ERROR("thread: Failed to set cpu affinity\n"); return false; } #elif __NetBSD__ @@ -161,7 +151,7 @@ RZ_API bool rz_th_setaffinity(RzThread *th, int cpuid) { if (pthread_setaffinity_np(th->tid, cpuset_size(c), c) != 0) { cpuset_destroy(c); - eprintf("Failed to set cpu affinity\n"); + RZ_LOG_ERROR("thread: Failed to set cpu affinity\n"); return false; } @@ -170,12 +160,12 @@ RZ_API bool rz_th_setaffinity(RzThread *th, int cpuid) { thread_affinity_policy_data_t c = { cpuid }; if (thread_policy_set(pthread_mach_thread_np(th->tid), THREAD_AFFINITY_POLICY, (thread_policy_t)&c, 1) != KERN_SUCCESS) { - eprintf("Failed to set cpu affinity\n"); + RZ_LOG_ERROR("thread: Failed to set cpu affinity\n"); return false; } #elif __WINDOWS__ if (SetThreadAffinityMask(th->tid, (DWORD_PTR)1 << cpuid) == 0) { - eprintf("Failed to set cpu affinity\n"); + RZ_LOG_ERROR("thread: Failed to set cpu affinity\n"); return false; } #elif __sun @@ -186,7 +176,7 @@ RZ_API bool rz_th_setaffinity(RzThread *th, int cpuid) { if (pset_bind(c, P_PID, getpid(), NULL)) { pset_destroy(c); - eprintf("Failed to set cpu affinity\n"); + RZ_LOG_ERROR("thread: Failed to set cpu affinity\n"); return false; } @@ -197,323 +187,106 @@ RZ_API bool rz_th_setaffinity(RzThread *th, int cpuid) { return true; } -RZ_API RzThread *rz_th_new(RZ_TH_FUNCTION(fun), void *user, int delay) { +/** + * \brief Creates and starts a new thread. + * + * \param function The callback to call when the thread starts. + * \param user A pointer to a user structure to pass to the callback function + * + * \return On success returns a valid pointer, otherwise NULL. + */ +RZ_API RZ_OWN RzThread *rz_th_new(RZ_NONNULL RzThreadFunction function, RZ_NULLABLE void *user) { + rz_return_val_if_fail(function, NULL); + RzThread *th = RZ_NEW0(RzThread); - if (th) { - th->lock = rz_th_lock_new(false); - th->running = false; - th->fun = fun; - th->user = user; - th->delay = delay; - th->breaked = false; - th->ready = false; + if (!th) { + RZ_LOG_ERROR("thread: Failed to allocate RzThread\n"); + return NULL; + } + + th->function = function; + th->user = user; + #if HAVE_PTHREAD - pthread_create(&th->tid, NULL, _rz_th_launcher, th); + if (!pthread_create(&th->tid, NULL, thread_main_function, th)) { + return th; + } #elif __WINDOWS__ - th->tid = CreateThread(NULL, 0, _rz_th_launcher, th, 0, 0); + if ((th->tid = CreateThread(NULL, 0, thread_main_function, th, 0, 0))) { + return th; + } #endif - } - return th; + RZ_LOG_ERROR("thread: Failed to start the RzThread\n"); + free(th); + return NULL; } -RZ_API void rz_th_break(RzThread *th) { - th->breaked = true; -} +/** + * \brief Force-stops a thread + * + * \param RzThread The thread to stop + */ +RZ_API void rz_th_kill(RZ_NONNULL RzThread *th) { + rz_return_if_fail(th); -RZ_API bool rz_th_kill(RzThread *th, bool force) { - if (!th || !th->tid || !th->running) { - return false; - } - th->breaked = true; - th->running = false; - rz_th_break(th); - rz_th_wait(th); #if HAVE_PTHREAD + if (!pthread_kill(th->tid, 0)) { #ifdef __ANDROID__ - pthread_kill(th->tid, 9); + pthread_kill(th->tid, 9); #else - pthread_cancel(th->tid); + pthread_cancel(th->tid); #endif -#elif __WINDOWS__ - TerminateThread(th->tid, -1); -#endif - return 0; -} - -RZ_API bool rz_th_start(RzThread *th, int enable) { - bool ret = true; - if (enable) { - if (!th->running) { - // start thread - while (!th->ready) { - /* spinlock */ - } - rz_th_lock_leave(th->lock); - } - } else { - if (th->running) { - // stop thread - // rz_th_kill (th, 0); - rz_th_lock_enter(th->lock); // deadlock? - } } - th->running = enable; - return ret; +#elif __WINDOWS__ + if (WaitForSingleObject(th->tid, 0)) { + TerminateThread(th->tid, -1); + } +#endif } -RZ_API bool rz_th_wait(RzThread *th) { - bool ret = false; - if (th) { +/** + * \brief Awaits indefinetely for a thread to join + * + * \param[in] th The thread to await for. + * + * \return On graceful stop returns true, otherwise false + */ +RZ_API bool rz_th_wait(RZ_NONNULL RzThread *th) { + rz_return_val_if_fail(th, false); #if HAVE_PTHREAD - void *thret = NULL; - ret = pthread_join(th->tid, &thret); + void *thret = NULL; + return pthread_join(th->tid, &thret) == 0; #elif __WINDOWS__ - ret = WaitForSingleObject(th->tid, INFINITE); + return WaitForSingleObject(th->tid, INFINITE) == 0; // WAIT_OBJECT_0 #endif - th->running = false; - } - return ret; } -RZ_API bool rz_th_wait_async(RzThread *th) { - return th->running; -} - -RZ_API void rz_th_free(RzThread *th) { +/** + * \brief Frees a RzThread structure + * + * \param th The RzThread to free + */ +RZ_API void rz_th_free(RZ_NULLABLE RzThread *th) { if (!th) { return; } #if __WINDOWS__ CloseHandle(th->tid); #endif - rz_th_lock_free(th->lock); free(th); } -RZ_API void rz_th_kill_free(RzThread *th) { - if (!th) { - return; - } - rz_th_kill(th, true); +/** + * \brief Stops the thread and frees the RzThread structure + * + * \param th The thread to stop and free. + */ +RZ_API void rz_th_kill_free(RZ_NONNULL RzThread *th) { + rz_return_if_fail(th); + rz_th_kill(th); rz_th_free(th); } -RZ_API size_t rz_th_physical_core_number() { -#ifdef __WINDOWS__ - SYSTEM_INFO sysinfo; - GetSystemInfo(&sysinfo); - return sysinfo.dwNumberOfProcessors; -#elif __APPLE__ || __FreeBSD__ || __OpenBSD__ || __DragonFly__ || __NetBSD__ - int os_status = 0; - int mib[4]; - unsigned long n_cpus = 1; - size_t n_cpus_length = sizeof(n_cpus); - - /* set the mib for hw.ncpu */ - mib[0] = CTL_HW; -#if __NetBSD__ - mib[1] = HW_NCPUONLINE; -#elif __OpenBSD__ || __FreeBSD__ || __DragonFly__ - mib[1] = HW_NCPU; -#else - mib[1] = HW_AVAILCPU; -#endif - - os_status = sysctl(mib, 2, &n_cpus, &n_cpus_length, NULL, 0); - - if (os_status != 0) { -#if __OpenBSD__ || __FreeBSD__ - n_cpus = 1; -#else - // HW_AVAILCPU does not exist. - mib[1] = HW_NCPU; - os_status = sysctl(mib, 2, &n_cpus, &n_cpus_length, NULL, 0); - if (os_status != 0) { - n_cpus = 1; - } -#endif - } - // this is needed because the upper bits are set on bsd platforms - n_cpus &= UT32_MAX; - - return n_cpus; -#elif __HAIKU__ - system_info info; - get_system_info(&info); - return info.cpu_count; -#else - return sysconf(_SC_NPROCESSORS_ONLN); -#endif -} - -/** - * \brief returns a new RzThreadPool structure with a pool of thread - * - * Returns a new RzThreadPool structure with a pool of thread limited - * by either the physical core number count or by the value specified - * by the user (if set to 0, it will be the max physical cores number) - * - * \param max_threads The maximum number of threads needed in the pool - * \return RzThreadPool The RzThreadPool structure - */ -RZ_API RZ_OWN RzThreadPool *rz_th_pool_new(size_t max_threads) { - RzThreadPool *pool = RZ_NEW0(RzThreadPool); - if (!pool) { - return NULL; - } - - size_t cores = rz_th_physical_core_number(); - if (max_threads) { - cores = RZ_MIN(cores, max_threads); - } - - pool->size = cores; - pool->threads = RZ_NEWS0(RzThread *, cores); - if (!pool->threads) { - free(pool); - return NULL; - } - - return pool; -} - -/** - * \brief Kills (and frees) the threads and frees the RzThreadPool struct - * - * \param RzThreadPool *The thread pool to free - */ -RZ_API void rz_th_pool_free(RZ_NULLABLE RzThreadPool *pool) { - if (!pool) { - return; - } - rz_th_pool_kill_free(pool); - free(pool->threads); - free(pool); -} - -/** - * \brief Adds a thread to the thread pool - * - * \param RzThreadPool The thread pool where to add the thread - * \param RzThread The thread to add to the pool - * \return true if a slot is found, false otherwise - */ -RZ_API bool rz_th_pool_add_thread(RZ_NONNULL RzThreadPool *pool, RZ_NONNULL RzThread *thread) { - rz_return_val_if_fail(pool && thread, false); - for (ut32 i = 0; i < pool->size; ++i) { - if (!pool->threads[i]) { - RZ_LOG_DEBUG("thread: thread %u added\n", i); - pool->threads[i] = thread; - return true; - } - } - return false; -} - -/** - * @brief Starts all the threads in the thread pool - * - * @param RzThreadPool The thread pool to start - * @param enable Enable the thread or disables them (see rz_th_start) - * - * @return returns true if starts any thread from the pool, otherwise false - */ -RZ_API bool rz_th_pool_start(RZ_NONNULL RzThreadPool *pool, bool enable) { - rz_return_val_if_fail(pool, false); - bool started = false; - for (ut32 i = 0; i < pool->size; ++i) { - if (pool->threads[i]) { - RZ_LOG_DEBUG("thread: started thread %u\n", i); - rz_th_start(pool->threads[i], enable); - started = true; - } - } - if (!started) { - RZ_LOG_ERROR("thread: cannot start thread pool when there are no threads in it\n"); - } - return started; -} - -/** - * \brief Waits the end of all the threads in the thread pool - * - * \param RzThreadPool The thread pool to wait for - * - * \return true if managed to wait all threads, otherwise false - */ -RZ_API bool rz_th_pool_wait(RZ_NONNULL RzThreadPool *pool) { - rz_return_val_if_fail(pool, false); - bool has_exited = true; - for (ut32 i = 0; i < pool->size; ++i) { - if (pool->threads[i]) { - RZ_LOG_DEBUG("thread: waiting for thread %u\n", i); - has_exited &= !rz_th_wait(pool->threads[i]); - } - } - return has_exited; -} - -/** - * \brief Waits asynchronously the end of all the threads in the thread pool - * - * \param RzThreadPool The thread pool to wait for - * - * \return true if managed to wait all threads, otherwise false - */ -RZ_API bool rz_th_pool_wait_async(RZ_NONNULL RzThreadPool *pool) { - rz_return_val_if_fail(pool, false); - bool has_exited = true; - for (ut32 i = 0; i < pool->size; ++i) { - if (pool->threads[i]) { - RZ_LOG_DEBUG("thread: waiting for thread %u (async)\n", i); - has_exited &= !rz_th_wait_async(pool->threads[i]); - } - } - return has_exited; -} - -/** - * \brief Kills all threads in the thread pool - * - * \param pool The thread pool to kill - * \param force Set to true if force killing the threads - * - * \return true if managed to kill all threads, otherwise false - */ -RZ_API bool rz_th_pool_kill(RZ_NONNULL RzThreadPool *pool, bool force) { - rz_return_val_if_fail(pool, false); - bool has_exited = false; - for (ut32 i = 0; i < pool->size; ++i) { - if (pool->threads[i]) { - RZ_LOG_DEBUG("thread: killing thread %u\n", i); - rz_th_kill(pool->threads[i], force); - has_exited = true; - } - } - return has_exited; -} - -/** - * \brief Force kills all threads in the thread pool and frees them - * - * \param pool The thread pool to kill - * - * \return true if managed to kill all threads, otherwise false - */ -RZ_API bool rz_th_pool_kill_free(RZ_NONNULL RzThreadPool *pool) { - rz_return_val_if_fail(pool, false); - bool has_exited = false; - for (ut32 i = 0; i < pool->size; ++i) { - if (pool->threads[i]) { - RZ_LOG_DEBUG("thread: killing thread %u\n", i); - rz_th_kill_free(pool->threads[i]); - has_exited = true; - pool->threads[i] = NULL; - } - } - return has_exited; -} - /** * \brief Returns user pointer of thread * @@ -521,6 +294,35 @@ RZ_API bool rz_th_pool_kill_free(RZ_NONNULL RzThreadPool *pool) { * * \return user pointer set by the rz_th_new user parameter */ -RZ_API void *rz_th_get_user(RzThread *th) { +RZ_API RZ_OWN void *rz_th_get_user(RZ_NONNULL RzThread *th) { + rz_return_val_if_fail(th, NULL); return th->user; } + +/** + * \brief Returns return value of the thread + * + * \param th The thread to get the return value from + * + * \return returns a pointer set when the thread returns + */ +RZ_API RZ_OWN void *rz_th_get_retv(RZ_NONNULL RzThread *th) { + rz_return_val_if_fail(th, NULL); + return th->retv; +} + +/** + * \brief Yield the processor + * + * \return On success returns true, otherwise false + */ +RZ_API bool rz_th_yield(void) { +#if __WINDOWS__ + return SwitchToThread() != 0; +#else + // sched_yield is not available everywhere. + // usleep is more portable. + rz_sys_usleep(1); + return true; +#endif +} diff --git a/librz/util/thread.h b/librz/util/thread.h index 19e3ccffdc..4a49f6a8a4 100644 --- a/librz/util/thread.h +++ b/librz/util/thread.h @@ -1,4 +1,5 @@ // SPDX-FileCopyrightText: 2022 GustavoLCR +// SPDX-FileCopyrightText: 2022 deroad // SPDX-License-Identifier: LGPL-3.0-only #ifndef RZ_THREAD_INTERNAL_H @@ -8,7 +9,8 @@ #endif #define _GNU_SOURCE #include -#include "rz_types.h" +#include +#include #if __WINDOWS__ #include @@ -16,8 +18,7 @@ #define RZ_TH_LOCK_T CRITICAL_SECTION #define RZ_TH_COND_T CONDITION_VARIABLE #define RZ_TH_SEM_T HANDLE -// HANDLE - +#define RZ_TH_RET_T DWORD WINAPI #elif HAVE_PTHREAD #define __GNU #include @@ -43,13 +44,29 @@ #define RZ_TH_LOCK_T pthread_mutex_t #define RZ_TH_COND_T pthread_cond_t #define RZ_TH_SEM_T sem_t * - +#define RZ_TH_RET_T void * #else #error Threading library only supported for pthread and w32 #endif -#ifdef __cplusplus -extern "C" { +#if __APPLE__ +// Here to avoid polluting mach types macro redefinitions... +#include +#include +#endif + +#if __APPLE__ || __NetBSD__ || __FreeBSD__ || __OpenBSD__ || __DragonFly__ || __sun +#include +#include +#endif + +#if __sun +#include +#endif + +#if __HAIKU__ +#include +#include #endif struct rz_th_sem_t { @@ -65,16 +82,12 @@ struct rz_th_cond_t { }; struct rz_th_t { - RZ_TH_TID tid; - RzThreadLock *lock; - RZ_TH_FUNCTION(fun); - void *user; // user pointer - bool running; - bool breaked; // thread aims to be interrupted - int delay; // delay the startup of the thread N seconds - bool ready; // thread is properly setup + RZ_TH_TID tid; ///< Thread identifier. + RzThreadFunction function; ///< User defined thread function. + void *user; ///< User defined thread data to pass (can be NULL). + void *retv; ///< Thread return value. }; RZ_IPI RZ_TH_TID rz_th_self(void); -#endif +#endif /* RZ_THREAD_INTERNAL_H */ diff --git a/librz/util/thread_cond.c b/librz/util/thread_cond.c index 7d86108a98..673ee94d52 100644 --- a/librz/util/thread_cond.c +++ b/librz/util/thread_cond.c @@ -1,9 +1,16 @@ // SPDX-FileCopyrightText: 2009-2020 thestr4ng3r +// SPDX-FileCopyrightText: 2022 deroad // SPDX-License-Identifier: LGPL-3.0-only #include "thread.h" -RZ_API RzThreadCond *rz_th_cond_new(void) { +/** + * \brief Condition variables are intended to be used to communicate changes in the state of data shared between threads. + * Condition variables are always associated with a mutex to provide synchronized access to the shared data. + * + * \return On success returns a valid pointer to a RzThreadCond structure. + */ +RZ_API RZ_OWN RzThreadCond *rz_th_cond_new(void) { RzThreadCond *cond = RZ_NEW0(RzThreadCond); if (!cond) { return NULL; @@ -19,7 +26,13 @@ RZ_API RzThreadCond *rz_th_cond_new(void) { return cond; } -RZ_API void rz_th_cond_signal(RzThreadCond *cond) { +/** + * \brief This function shall unblock at least one of the threads that are blocked on the specified condition + * + * \param cond The RzThreadCond to use for signalling a waiting thread + */ +RZ_API void rz_th_cond_signal(RZ_NONNULL RzThreadCond *cond) { + rz_return_if_fail(cond); #if HAVE_PTHREAD pthread_cond_signal(&cond->cond); #elif __WINDOWS__ @@ -27,7 +40,13 @@ RZ_API void rz_th_cond_signal(RzThreadCond *cond) { #endif } -RZ_API void rz_th_cond_signal_all(RzThreadCond *cond) { +/** + * \brief This function shall unblock all threads currently blocked on the specified condition + * + * \param cond The RzThreadCond to use for signalling all waiting threads + */ +RZ_API void rz_th_cond_signal_all(RZ_NONNULL RzThreadCond *cond) { + rz_return_if_fail(cond); #if HAVE_PTHREAD pthread_cond_broadcast(&cond->cond); #elif __WINDOWS__ @@ -35,7 +54,14 @@ RZ_API void rz_th_cond_signal_all(RzThreadCond *cond) { #endif } -RZ_API void rz_th_cond_wait(RzThreadCond *cond, RzThreadLock *lock) { +/** + * \brief The function shall block on a condition variable and shall be called with RzThreadLock locked by the calling thread. + * + * \param cond The RzThreadCond to use for waiting the signal + * \param lock The RzThreadLock lock to use (the lock must be already taken by the thread) + */ +RZ_API void rz_th_cond_wait(RZ_NONNULL RzThreadCond *cond, RZ_NONNULL RzThreadLock *lock) { + rz_return_if_fail(cond); #if HAVE_PTHREAD pthread_cond_wait(&cond->cond, &lock->lock); #elif __WINDOWS__ @@ -43,7 +69,12 @@ RZ_API void rz_th_cond_wait(RzThreadCond *cond, RzThreadLock *lock) { #endif } -RZ_API void rz_th_cond_free(RzThreadCond *cond) { +/** + * \brief Frees a RzThreadCond struct + * + * \param cond The RzThreadCond to free + */ +RZ_API void rz_th_cond_free(RZ_NULLABLE RzThreadCond *cond) { if (!cond) { return; } diff --git a/librz/util/thread_lock.c b/librz/util/thread_lock.c index 864f16fc4f..cb436ccf91 100644 --- a/librz/util/thread_lock.c +++ b/librz/util/thread_lock.c @@ -1,50 +1,65 @@ // SPDX-FileCopyrightText: 2009-2017 pancake +// SPDX-FileCopyrightText: 2022 deroad // SPDX-License-Identifier: LGPL-3.0-only #include "thread.h" -/* locks/mutex/sems */ - -RZ_API RzThreadLock *rz_th_lock_new(bool recursive) { +/** + * \brief Allocates and initialize a RzThreadLock structure. + * + * \param recursive Set it to true for recursive locking (on windows, all the locks are always recursive). + * + * \return On success returns a valid pointer, otherwise NULL + */ +RZ_API RZ_OWN RzThreadLock *rz_th_lock_new(bool recursive) { RzThreadLock *thl = RZ_NEW0(RzThreadLock); - if (thl) { -#if HAVE_PTHREAD - if (recursive) { - pthread_mutexattr_t attr; - pthread_mutexattr_init(&attr); -#if !defined(__GLIBC__) || __USE_UNIX98__ - pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE); -#else - pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE_NP); -#endif - pthread_mutex_init(&thl->lock, &attr); - } else { - pthread_mutex_init(&thl->lock, NULL); - } -#elif __WINDOWS__ - // TODO: obey `recursive` (currently it is always recursive) - InitializeCriticalSection(&thl->lock); -#endif + if (!thl) { + return NULL; } +#if HAVE_PTHREAD + if (recursive) { + pthread_mutexattr_t attr; + pthread_mutexattr_init(&attr); +#if !defined(__GLIBC__) || __USE_UNIX98__ + pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE); +#else + pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE_NP); +#endif + pthread_mutex_init(&thl->lock, &attr); + } else { + pthread_mutex_init(&thl->lock, NULL); + } +#elif __WINDOWS__ + // Windows critical sections always accept recursive + // access and it cannot be configured in any other way. + InitializeCriticalSection(&thl->lock); +#endif return thl; } -RZ_API int rz_th_lock_wait(RzThreadLock *thl) { - rz_th_lock_enter(thl); // locks here - rz_th_lock_leave(thl); // releases previous mutex - return 0; -} - -RZ_API int rz_th_lock_enter(RzThreadLock *thl) { +/** + * \brief Acquires a RzThreadLock structure + * + * \param thl The RzThreadLock to acquire + */ +RZ_API void rz_th_lock_enter(RZ_NONNULL RzThreadLock *thl) { + rz_return_if_fail(thl); #if HAVE_PTHREAD - return pthread_mutex_lock(&thl->lock); + pthread_mutex_lock(&thl->lock); #elif __WINDOWS__ EnterCriticalSection(&thl->lock); - return 0; #endif } -RZ_API int rz_th_lock_tryenter(RzThreadLock *thl) { +/** + * \brief Tries to acquire a RzThreadLock structure + * + * \param thl The RzThreadLock to try to acquire + * + * \return On success returns true, otherwise false + */ +RZ_API bool rz_th_lock_tryenter(RZ_NONNULL RzThreadLock *thl) { + rz_return_val_if_fail(thl, false); #if HAVE_PTHREAD return !pthread_mutex_trylock(&thl->lock); #elif __WINDOWS__ @@ -52,23 +67,33 @@ RZ_API int rz_th_lock_tryenter(RzThreadLock *thl) { #endif } -RZ_API int rz_th_lock_leave(RzThreadLock *thl) { +/** + * \brief Releases a RzThreadLock structure + * + * \param thl The RzThreadLock to release + */ +RZ_API void rz_th_lock_leave(RZ_NONNULL RzThreadLock *thl) { + rz_return_if_fail(thl); #if HAVE_PTHREAD - return pthread_mutex_unlock(&thl->lock); + pthread_mutex_unlock(&thl->lock); #elif __WINDOWS__ LeaveCriticalSection(&thl->lock); - return 0; #endif } -RZ_API void *rz_th_lock_free(RzThreadLock *thl) { - if (thl) { -#if HAVE_PTHREAD - pthread_mutex_destroy(&thl->lock); -#elif __WINDOWS__ - DeleteCriticalSection(&thl->lock); -#endif - free(thl); +/** + * \brief Frees a RzThreadLock structure + * + * \param thl The RzThreadLock to free + */ +RZ_API void rz_th_lock_free(RZ_NULLABLE RzThreadLock *thl) { + if (!thl) { + return; } - return NULL; +#if HAVE_PTHREAD + pthread_mutex_destroy(&thl->lock); +#elif __WINDOWS__ + DeleteCriticalSection(&thl->lock); +#endif + free(thl); } diff --git a/librz/util/thread_pool.c b/librz/util/thread_pool.c new file mode 100644 index 0000000000..eabc90d557 --- /dev/null +++ b/librz/util/thread_pool.c @@ -0,0 +1,201 @@ +// SPDX-FileCopyrightText: 2022 deroad +// SPDX-License-Identifier: LGPL-3.0-only + +#include +#include "thread.h" + +/** + * \brief RzThreadPool is a structure which handles n-threads threads + * + * This structure provides methods to handle multiple threads, like they were one. + */ +struct rz_th_pool_t { + size_t size; + RzThread **threads; +}; + +/** + * \brief Returns the number of available physical cores of the host machine + * + * \return The number of available physical cores (always >= 1) + */ +RZ_API size_t rz_th_physical_core_number() { +#ifdef __WINDOWS__ + SYSTEM_INFO sysinfo; + GetSystemInfo(&sysinfo); + return sysinfo.dwNumberOfProcessors; +#elif __APPLE__ || __FreeBSD__ || __OpenBSD__ || __DragonFly__ || __NetBSD__ + int os_status = 0; + int mib[4]; + unsigned long n_cpus = 1; + size_t n_cpus_length = sizeof(n_cpus); + + /* set the mib for hw.ncpu */ + mib[0] = CTL_HW; +#if __NetBSD__ + mib[1] = HW_NCPUONLINE; +#elif __OpenBSD__ || __FreeBSD__ || __DragonFly__ + mib[1] = HW_NCPU; +#else + mib[1] = HW_AVAILCPU; +#endif + + os_status = sysctl(mib, 2, &n_cpus, &n_cpus_length, NULL, 0); + + if (os_status != 0) { +#if __OpenBSD__ || __FreeBSD__ + n_cpus = 1; +#else + // HW_AVAILCPU does not exist. + mib[1] = HW_NCPU; + os_status = sysctl(mib, 2, &n_cpus, &n_cpus_length, NULL, 0); + if (os_status != 0) { + n_cpus = 1; + } +#endif + } + // this is needed because the upper bits are set on bsd platforms + n_cpus &= UT32_MAX; + + return n_cpus; +#elif __HAIKU__ + system_info info; + get_system_info(&info); + return info.cpu_count; +#else + return sysconf(_SC_NPROCESSORS_ONLN); +#endif +} + +/** + * \brief returns a new RzThreadPool structure with a pool of thread + * + * Returns a new RzThreadPool structure with a pool of thread limited + * by either the physical core number count or by the value specified + * by the user (if set to 0, it will be the max physical cores number) + * + * \param max_threads The maximum number of threads needed in the pool + * \return RzThreadPool The RzThreadPool structure + */ +RZ_API RZ_OWN RzThreadPool *rz_th_pool_new(size_t max_threads) { + RzThreadPool *pool = RZ_NEW0(RzThreadPool); + if (!pool) { + return NULL; + } + + size_t cores = rz_th_physical_core_number(); + if (max_threads) { + cores = RZ_MIN(cores, max_threads); + } + + pool->size = cores; + pool->threads = RZ_NEWS0(RzThread *, cores); + if (!pool->threads) { + free(pool); + return NULL; + } + + return pool; +} + +/** + * \brief Kills (and frees) the threads and frees the RzThreadPool struct + * + * \param RzThreadPool *The thread pool to free + */ +RZ_API void rz_th_pool_free(RZ_NULLABLE RzThreadPool *pool) { + if (!pool) { + return; + } + for (ut32 i = 0; i < pool->size; ++i) { + if (pool->threads[i]) { + rz_th_free(pool->threads[i]); + pool->threads[i] = NULL; + } + } + free(pool->threads); + free(pool); +} + +/** + * \brief Adds a thread to the thread pool + * + * \param RzThreadPool The thread pool where to add the thread + * \param RzThread The thread to add to the pool + * \return true if a slot is found, false otherwise + */ +RZ_API bool rz_th_pool_add_thread(RZ_NONNULL RzThreadPool *pool, RZ_NONNULL RzThread *thread) { + rz_return_val_if_fail(pool && thread, false); + for (ut32 i = 0; i < pool->size; ++i) { + if (!pool->threads[i]) { + RZ_LOG_DEBUG("thread: thread %u added\n", i); + pool->threads[i] = thread; + return true; + } + } + return false; +} + +/** + * \brief Returns the n-th thread in the thread pool. + * + * \param pool The thread pool to use + * \param index The index of the thread to get + * + * \return Returns the pointer of the n-th thread in the thread pool. + */ +RZ_API RZ_BORROW RzThread *rz_th_pool_get_thread(RZ_NONNULL RzThreadPool *pool, size_t index) { + rz_return_val_if_fail(pool && index < pool->size, NULL); + return pool->threads[index]; +} + +/** + * \brief Waits the end of all the threads in the thread pool + * + * \param RzThreadPool The thread pool to wait for + * + * \return true if managed to wait all threads, otherwise false + */ +RZ_API bool rz_th_pool_wait(RZ_NONNULL RzThreadPool *pool) { + rz_return_val_if_fail(pool, false); + bool has_exited = true; + for (ut32 i = 0; i < pool->size; ++i) { + if (pool->threads[i]) { + RZ_LOG_DEBUG("thread: waiting for thread %u\n", i); + has_exited = has_exited && rz_th_wait(pool->threads[i]); + } + } + return has_exited; +} + +/** + * \brief Force-stops all threads in the thread pool + * + * \param pool The thread pool to kill + * + * \return true if managed to kill all threads, otherwise false + */ +RZ_API bool rz_th_pool_kill(RZ_NONNULL RzThreadPool *pool) { + rz_return_val_if_fail(pool, false); + bool has_exited = false; + for (ut32 i = 0; i < pool->size; ++i) { + if (pool->threads[i]) { + RZ_LOG_DEBUG("thread: killing thread %u\n", i); + rz_th_kill(pool->threads[i]); + has_exited = true; + } + } + return has_exited; +} + +/** + * \brief Returns the thread pool size + * + * \param pool The RzThreadPool to use + * + * \return The size of the thread pool (always >= 1). + */ +RZ_API size_t rz_th_pool_size(RZ_NONNULL RzThreadPool *pool) { + rz_return_val_if_fail(pool, 1); + return pool->size; +} \ No newline at end of file diff --git a/librz/util/thread_queue.c b/librz/util/thread_queue.c new file mode 100644 index 0000000000..d3d9030e6f --- /dev/null +++ b/librz/util/thread_queue.c @@ -0,0 +1,171 @@ +// SPDX-FileCopyrightText: 2022 deroad +// SPDX-License-Identifier: LGPL-3.0-only + +#include +#include "thread.h" + +/** + * \brief RzThreadQueue is a thread-safe queue that can be listened on from multiple threads. + * + * This Queue is thread-safe and allows to perform LIFO/FIFO operations. + * rz_th_queue_new Allocates a RzThreadQueue structure and allows to limit the size of the queue. + * rz_th_queue_push Pushes an element to the queue unless the limit is reached. + * rz_th_queue_pop Pops an element from the queue, but returns NULL when is empty. + * rz_th_queue_wait_pop Pops an element from the queue, but awaits for new elements when is empty. + * rz_th_queue_free Frees a RzThreadQueue structure, if the queue is not empty, it frees the elements with the provided qfree function. + */ +struct rz_th_queue_t { + RzThreadLock *lock; + RzThreadCond *cond; + size_t max_size; + RzList *list; +}; + +/** + * \brief Allocates and initializes a new fifo queue + * + * \param max_size The maximum size of the queue, use RZ_THREAD_QUEUE_UNLIMITED for an unlimited size + * \param qfree Pointer to a custom free function to free the queue if not empty. + * + * \return On success returns a valid pointer, otherwise NULL + */ +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; +} + +/** + * \brief Frees a RzThreadQueue structure + * + * \param queue The RzThreadQueue to free + */ +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); +} + +/** + * \brief Pushes a new element into the queue + * + * \param queue The RzThreadQueue to push to + * \param user The non-null pointer to push to the queue + * \param tail When true, appends the element to the tail, otherwise to the head + * + * \return On success returns true, otherwise false + */ +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; +} + +/** + * \brief Removes an element from the queue, but does not awaits when empty. + * + * \param queue The RzThreadQueue to push to + * \param tail When true, pops the element from the tail, otherwise from the head + * + * \return On success returns a valid pointer, otherwise NULL + */ +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; +} + +/** + * \brief Removes an element from the queue, but yields the thread till not empty. + * + * \param queue The RzThreadQueue to push to + * \param tail When true, pops the element from the tail, otherwise from the head + * + * \return On success returns a valid pointer, otherwise NULL + */ +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; +} + +/** + * \brief Returns true if the queue is empty (thread-safe) + * + * \param queue The RzThreadQueue to check + * + * \return When empty returns true, otherwise false + */ +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; +} + +/** + * \brief Returns true if the queue is full and when the size is not RZ_THREAD_QUEUE_UNLIMITED (thread-safe) + * + * \param queue The RzThreadQueue to check + * + * \return When full returns true, otherwise false + */ +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; +} diff --git a/librz/util/thread_sem.c b/librz/util/thread_sem.c index 307120c56d..85a71ee441 100644 --- a/librz/util/thread_sem.c +++ b/librz/util/thread_sem.c @@ -1,4 +1,5 @@ // SPDX-FileCopyrightText: 2018 thestr4ng3r +// SPDX-FileCopyrightText: 2022 deroad // SPDX-License-Identifier: LGPL-3.0-only #include "thread.h" @@ -15,7 +16,14 @@ #include #endif -RZ_API RzThreadSemaphore *rz_th_sem_new(unsigned int initial) { +/** + * \brief Allocates and initialize a RzThreadSemaphore structure + * + * \param initial The initial status of the semaphore + * + * \return On success returns a valid RzThreadSemaphore pointer, otherwise NULL + */ +RZ_API RZ_OWN RzThreadSemaphore *rz_th_sem_new(unsigned int initial) { RzThreadSemaphore *sem = RZ_NEW(RzThreadSemaphore); if (!sem) { return NULL; @@ -57,7 +65,12 @@ RZ_API RzThreadSemaphore *rz_th_sem_new(unsigned int initial) { return sem; } -RZ_API void rz_th_sem_free(RzThreadSemaphore *sem) { +/** + * \brief Frees a RzThreadSemaphore struct + * + * \param sem The RzThreadSemaphore to free + */ +RZ_API void rz_th_sem_free(RZ_NULLABLE RzThreadSemaphore *sem) { if (!sem) { return; } @@ -76,7 +89,13 @@ RZ_API void rz_th_sem_free(RzThreadSemaphore *sem) { free(sem); } -RZ_API void rz_th_sem_post(RzThreadSemaphore *sem) { +/** + * \brief increments (releases) a semaphore + * + * \param sem The RzThreadSemaphore to increment (release) + */ +RZ_API void rz_th_sem_post(RZ_NONNULL RzThreadSemaphore *sem) { + rz_return_if_fail(sem); #if HAVE_PTHREAD sem_post(sem->sem); #elif __WINDOWS__ @@ -84,7 +103,13 @@ RZ_API void rz_th_sem_post(RzThreadSemaphore *sem) { #endif } -RZ_API void rz_th_sem_wait(RzThreadSemaphore *sem) { +/** + * \brief Decrements (acquires) the semaphore (waits indefinetely) + * + * \param sem The RzThreadSemaphore to decrement (acquire) + */ +RZ_API void rz_th_sem_wait(RZ_NONNULL RzThreadSemaphore *sem) { + rz_return_if_fail(sem); #if HAVE_PTHREAD sem_wait(sem->sem); #elif __WINDOWS__ diff --git a/test/unit/test_threads.c b/test/unit/test_threads.c index fbc386eb62..d6683da740 100644 --- a/test/unit/test_threads.c +++ b/test/unit/test_threads.c @@ -3,6 +3,8 @@ // SPDX-License-Identifier: LGPL-3.0-only #include +#include +#include #include "minunit.h" bool test_thread_pool_cores(void) { @@ -10,22 +12,78 @@ bool test_thread_pool_cores(void) { RzThreadPool *pool = rz_th_pool_new(RZ_THREAD_POOL_ALL_CORES); mu_assert_notnull(pool, "rz_th_pool_new(RZ_THREAD_POOL_ALL_CORES) null check"); - mu_assert_eq(pool->size, cores, "rz_th_pool_new(RZ_THREAD_POOL_ALL_CORES) core count check"); + size_t pool_size = rz_th_pool_size(pool); + mu_assert_eq(pool_size, cores, "rz_th_pool_new(RZ_THREAD_POOL_ALL_CORES) core count check"); rz_th_pool_free(pool); if (cores > 1) { /* this can be tested only when cores are more than 1 */ pool = rz_th_pool_new(cores - 1); mu_assert_notnull(pool, "rz_th_pool_new(cores - 1) null check"); - mu_assert_eq(pool->size, cores - 1, "rz_th_pool_new(cores - 1) core count check"); + pool_size = rz_th_pool_size(pool); + mu_assert_eq(pool_size, cores - 1, "rz_th_pool_new(cores - 1) core count check"); rz_th_pool_free(pool); } mu_end; } +void *thread_queue_push_timed(RzThreadQueue *queue) { + rz_sys_sleep(2); + return rz_th_queue_push(queue, queue, true) ? queue : NULL; +} + +bool test_thread_queue(void) { + // test limited queue + void *head = (void *)"aaaaaa"; + void *tail = (void *)"bbbbbb"; + RzThreadQueue *queue = rz_th_queue_new(3, NULL); + mu_assert_notnull(queue, "rz_th_queue_new(3) null check"); + mu_assert_true(rz_th_queue_is_empty(queue), "queue is empty"); + mu_assert_true(rz_th_queue_push(queue, "cccccc", true), "queue pushed new element"); + mu_assert_true(rz_th_queue_push(queue, head, false), "queue pushed head new element"); + mu_assert_true(rz_th_queue_push(queue, tail, true), "queue pushed tail new element"); + mu_assert_true(rz_th_queue_is_full(queue), "queue is full"); + mu_assert_false(rz_th_queue_push(queue, "kkkkkk", true), "queue cannot push a new element"); + mu_assert_ptreq(rz_th_queue_pop(queue, false), head, "queue can pop head and is that element"); + mu_assert_ptreq(rz_th_queue_pop(queue, true), tail, "queue can pop tail and is that element"); + mu_assert_false(rz_th_queue_is_empty(queue), "queue is empty"); + mu_assert_false(rz_th_queue_is_full(queue), "queue is not full"); + rz_th_queue_free(queue); + + // test unlimited queue + queue = rz_th_queue_new(RZ_THREAD_QUEUE_UNLIMITED, NULL); + mu_assert_notnull(queue, "rz_th_queue_new(RZ_THREAD_QUEUE_UNLIMITED) null check"); + mu_assert_true(rz_th_queue_push(queue, "aaaaa", false), "queue can push a new element"); + mu_assert_true(rz_th_queue_push(queue, "aaaaa", true), "queue can push a new element"); + mu_assert_true(rz_th_queue_push(queue, "aaaaa", false), "queue can push a new element"); + mu_assert_true(rz_th_queue_push(queue, "aaaaa", true), "queue can push a new element"); + mu_assert_true(rz_th_queue_push(queue, "aaaaa", false), "queue can push a new element"); + mu_assert_true(rz_th_queue_push(queue, "aaaaa", true), "queue can push a new element"); + mu_assert_false(rz_th_queue_is_empty(queue), "queue is not empty"); + mu_assert_false(rz_th_queue_is_full(queue), "queue is not full"); + rz_th_queue_free(queue); + + // test queue + queue = rz_th_queue_new(RZ_THREAD_QUEUE_UNLIMITED, NULL); + RzThread *th = rz_th_new((RzThreadFunction)thread_queue_push_timed, queue); + mu_assert_notnull(th, "rz_th_new(thread_queue_push_timed, queue) null check"); + ut64 start = rz_time_now(); + tail = rz_th_queue_wait_pop(queue, true); + ut64 diff = rz_time_now() - start; + rz_th_wait(th); + mu_assert_ptreq(tail, queue, "rz_th_queue_wait_pop(queue, true) is queue"); + mu_assert_true(diff >= 1500000, "queue did wait for value."); + mu_assert_ptreq(rz_th_get_retv(th), queue, "verify it returned queue"); + rz_th_free(th); + rz_th_queue_free(queue); + + mu_end; +} + int all_tests() { mu_run_test(test_thread_pool_cores); + mu_run_test(test_thread_queue); return tests_passed != tests_run; }