portugalfuturista/rizin
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor thread code and add RzThreadQueue (#2683)

Browse source 
* Kill the thread only if is alive.
* Remove rz_th_lock_guard.
* Use rz_sys_usleep since pthread_yield/sched_yield are not portable.
* Kill threads only on error.
* Avoid killing already-dead threads.
* RzThreadFunction now returns void* and added test_thread_queue.
* Removed pthread_exit
This commit is contained in:
Giovanni 2022-06-19 15:07:13 +02:00 committed by GitHub
parent d8dd2c5097
commit a988941bf2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
17 changed files with 952 additions and 570 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
binrz/rz-test/rz-test.c
View file

@ -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) {

115
librz/core/basefind.c
View file

@ -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);
}

56
librz/core/rtr.c
View file

@ -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;
}

34
librz/core/rtr_http.c
View file

@ -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;

19
librz/core/task.c
View file

@ -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 */

92
librz/include/rz_th.h
View file

@ -1,3 +1,7 @@
// SPDX-FileCopyrightText: 2009-2017 pancake <pancake@nopcode.org>
// SPDX-FileCopyrightText: 2021-2022 deroad <wargio@libero.it>
// 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 <rz_types.h>
#include <rz_list.h>
#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

9
librz/socket/run.c
View file

@ -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");
}

2
librz/util/meson.build
View file

@ -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',

6
librz/util/subprocess.c
View file

@ -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]);

518
librz/util/thread.c
View file

@ -1,66 +1,25 @@
// SPDX-FileCopyrightText: 2009-2018 pancake <pancake@nopcode.org>
// SPDX-FileCopyrightText: 2020-2021 ret2libc <sirmy15@gmail.com>
// SPDX-FileCopyrightText: 2020-2022 deroad <wargio@libero.it>
// SPDX-FileCopyrightText: 2022 GustavoLCR <gugulcr@gmail.com>
// SPDX-License-Identifier: LGPL-3.0-only
#include <rz_util.h>
#include "thread.h"
#if __APPLE__
// Here to avoid polluting mach types macro redefinitions...
#include <mach/thread_act.h>
#include <mach/thread_policy.h>
#endif
#if __APPLE__ || __NetBSD__ || __FreeBSD__ || __OpenBSD__ || __DragonFly__ || __sun
#include <sys/param.h>
#include <sys/sysctl.h>
#endif
#if __sun
#include <sys/pset.h>
#endif
#if __HAIKU__
#include <kernel/scheduler.h>
#include <OS.h>
#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
}

43
librz/util/thread.h
View file

@ -1,4 +1,5 @@
// SPDX-FileCopyrightText: 2022 GustavoLCR <gugulcr@gmail.com>
// SPDX-FileCopyrightText: 2022 deroad <wargio@libero.it>
// SPDX-License-Identifier: LGPL-3.0-only
#ifndef RZ_THREAD_INTERNAL_H
@ -8,7 +9,8 @@
#endif
#define _GNU_SOURCE
#include <rz_th.h>
#include "rz_types.h"
#include <rz_types.h>
#include <rz_util/rz_assert.h>
#if __WINDOWS__
#include <rz_windows.h>
@ -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 <semaphore.h>
@ -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 <mach/thread_act.h>
#include <mach/thread_policy.h>
#endif
#if __APPLE__ || __NetBSD__ || __FreeBSD__ || __OpenBSD__ || __DragonFly__ || __sun
#include <sys/param.h>
#include <sys/sysctl.h>
#endif
#if __sun
#include <sys/pset.h>
#endif
#if __HAIKU__
#include <kernel/scheduler.h>
#include <OS.h>
#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 */

41
librz/util/thread_cond.c
View file

@ -1,9 +1,16 @@
// SPDX-FileCopyrightText: 2009-2020 thestr4ng3r <info@florianmaerkl.de>
// SPDX-FileCopyrightText: 2022 deroad <wargio@libero.it>
// 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;
}

111
librz/util/thread_lock.c
View file

@ -1,50 +1,65 @@
// SPDX-FileCopyrightText: 2009-2017 pancake <pancake@nopcode.org>
// SPDX-FileCopyrightText: 2022 deroad <wargio@libero.it>
// 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);
}

201
librz/util/thread_pool.c Normal file
View file

@ -0,0 +1,201 @@
// SPDX-FileCopyrightText: 2022 deroad <wargio@libero.it>
// SPDX-License-Identifier: LGPL-3.0-only
#include <rz_th.h>
#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;
}

171
librz/util/thread_queue.c Normal file
View file

@ -0,0 +1,171 @@
// SPDX-FileCopyrightText: 2022 deroad <wargio@libero.it>
// SPDX-License-Identifier: LGPL-3.0-only
#include <rz_th.h>
#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;
}

33
librz/util/thread_sem.c
View file

@ -1,4 +1,5 @@
// SPDX-FileCopyrightText: 2018 thestr4ng3r <info@florianmaerkl.de>
// SPDX-FileCopyrightText: 2022 deroad <wargio@libero.it>
// SPDX-License-Identifier: LGPL-3.0-only
#include "thread.h"
@ -15,7 +16,14 @@
#include <limits.h>
#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__

62
test/unit/test_threads.c
View file

@ -3,6 +3,8 @@
// SPDX-License-Identifier: LGPL-3.0-only
#include <rz_th.h>
#include <rz_util/rz_time.h>
#include <rz_util/rz_sys.h>
#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;
}