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

Optimize queue to act also as proper channel (#5627)

Browse source
This commit is contained in:
Giovanni 2025-12-20 00:40:13 +08:00 committed by GitHub
parent 77c63ce77c
commit 0a0e6c0b88
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 269 additions and 116 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
librz/arch/similarity.c
View file

@ -1,5 +1,5 @@
// SPDX-FileCopyrightText: 2022 RizinOrg <info@rizin.re>
// SPDX-FileCopyrightText: 2022 deroad <wargio@libero.it>
// SPDX-FileCopyrightText: 2022-2025 RizinOrg <info@rizin.re>
// SPDX-FileCopyrightText: 2022-2025 deroad <deroad@kumo.xn--q9jyb4c>
// SPDX-License-Identifier: LGPL-3.0-only
#include <rz_analysis.h>
@ -305,6 +305,7 @@ static RZ_OWN RzAnalysisMatchResult *analysis_match_result_new(RZ_NONNULL RzAnal
}
}
rz_th_queue_close_when_empty(shared.queue);
rz_th_pool_wait(pool);
if (!rz_atomic_bool_get(shared.loop)) {
@ -362,6 +363,15 @@ RZ_API void rz_analysis_match_result_free(RZ_NULLABLE RzAnalysisMatchResult *res
free(result);
}
static void *shared_queue_pop(SharedContext *shared) {
void *data = NULL;
if (!rz_atomic_bool_get(shared->loop) ||
!rz_th_queue_pop(shared->queue, false, &data)) {
return NULL;
}
return data;
}
static void *analysis_match_basic_blocks(SharedContext *shared) {
double max_similarity = 0.0, calc_similarity = 0.0;
const RzListIter *iter = NULL;
@ -370,7 +380,7 @@ static void *analysis_match_basic_blocks(SharedContext *shared) {
ut32 size_a = 0, size_b = 0;
ut8 *buf_a = NULL, *buf_b = NULL;
while (rz_atomic_bool_get(shared->loop) && (bb_a = rz_th_queue_pop(shared->queue, false))) {
while ((bb_a = shared_queue_pop(shared))) {
if (!shared_context_alloc_a(shared, bb_a, &buf_a, &size_a)) {
RZ_LOG_ERROR("analysis_match: cannot allocate buffer for block 0x%08" PFMT64x " (A)\n", bb_a->addr);
rz_th_queue_push(shared->unmatch, bb_a, true);
@ -468,7 +478,7 @@ static void *analysis_match_functions(SharedContext *shared) {
ut32 size_a = 0, size_b = 0;
ut8 *buf_a = NULL, *buf_b = NULL;
while (rz_atomic_bool_get(shared->loop) && (fcn_a = rz_th_queue_pop(shared->queue, false))) {
while ((fcn_a = shared_queue_pop(shared))) {
if (!shared_context_alloc_a(shared, fcn_a, &buf_a, &size_a)) {
RZ_LOG_ERROR("analysis_match: cannot allocate buffer for function %s (A)\n", fcn_a->name);
rz_th_queue_push(shared->unmatch, fcn_a, true);
@ -525,7 +535,7 @@ static void *analysis_match_one_function(SharedContext *shared) {
return NULL;
}
while (rz_atomic_bool_get(shared->loop) && (fcn_a = rz_th_queue_pop(shared->queue, false))) {
while ((fcn_a = shared_queue_pop(shared))) {
if (!shared_context_alloc_b(shared, fcn_a, &buf_a, &size_a)) {
RZ_LOG_ERROR("analysis_match: cannot allocate buffer for function %s (A)\n", fcn_a->name);
free(buf_b);

8
librz/bin/bfile_string.c
View file

@ -1,4 +1,4 @@
// SPDX-FileCopyrightText: 2022 deroad <wargio@libero.it>
// SPDX-FileCopyrightText: 2022-2025 deroad <deroad@kumo.xn--q9jyb4c>
// SPDX-License-Identifier: LGPL-3.0-only
#include <rz_bin.h>
@ -263,10 +263,11 @@ static void *search_string_thread_runner(SearchThreadData *std) {
const RzBinFile *bf = shared->bf; // this data is always RO
do {
itv = rz_th_queue_pop(std->intervals, false);
if (!itv) {
void *data = NULL;
if (!rz_th_queue_pop(std->intervals, false, &data) || !data) {
break;
}
itv = (SearchInterval *)data;
paddr = itv->paddr;
psize = itv->psize;
free(itv);
@ -669,6 +670,7 @@ RZ_API RZ_OWN RzPVector /*<RzBinString *>*/ *rz_bin_file_strings(RZ_NONNULL RzBi
}
}
rz_th_queue_close_when_empty(intervals);
rz_th_pool_wait(pool);
results = rz_pvector_new((RzPVectorFree)rz_bin_string_free);

7
librz/include/rz_th.h
View file

@ -69,6 +69,7 @@ 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_timed_wait(RZ_NONNULL RzThreadCond *cond, RZ_NONNULL RzThreadLock *lock, size_t timeout_ms);
RZ_API void rz_th_cond_free(RZ_NULLABLE RzThreadCond *cond);
RZ_API RzThreadNCores rz_th_physical_core_number();
@ -86,12 +87,14 @@ RZ_API RZ_OWN RzThreadQueue *rz_th_queue_from_list(RZ_NONNULL RZ_BORROW RzList /
RZ_API RZ_OWN RzThreadQueue *rz_th_queue_from_pvector(RZ_NONNULL RZ_BORROW RzPVector /*<void *>*/ *vector, 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_pop(RZ_NONNULL RzThreadQueue *queue, bool tail, RZ_NONNULL RZ_OUT void **data);
RZ_API RZ_OWN RzList /*<void *>*/ *rz_th_queue_pop_all(RZ_NONNULL RzThreadQueue *queue);
RZ_API bool rz_th_queue_is_empty(RZ_NONNULL RzThreadQueue *queue);
RZ_API bool rz_th_queue_is_full(RZ_NONNULL RzThreadQueue *queue);
RZ_API size_t rz_th_queue_size(RZ_NONNULL RzThreadQueue *queue);
RZ_API void rz_th_queue_close_when_empty(RZ_NONNULL RzThreadQueue *queue);
RZ_API void rz_th_queue_close(RZ_NONNULL RzThreadQueue *queue);
RZ_API bool rz_th_queue_is_closed(RZ_NONNULL RzThreadQueue *queue);
RZ_API RZ_OWN RzAtomicBool *rz_atomic_bool_new(bool value);
RZ_API void rz_atomic_bool_free(RZ_NULLABLE RzAtomicBool *tbool);

12
librz/search/search.c
View file

@ -1,5 +1,6 @@
// SPDX-FileCopyrightText: 2024 RizinOrg <info@rizin.re>
// SPDX-FileCopyrightText: 2024 deroad <wargio@libero.it>
// SPDX-FileCopyrightText: 2024-2025 RizinOrg <info@rizin.re>
// SPDX-FileCopyrightText: 2024-2025 Rot127 <rot127@posteo.com>
// SPDX-FileCopyrightText: 2024-2025 deroad <deroad@kumo.xn--q9jyb4c>
// SPDX-License-Identifier: LGPL-3.0-only
#include <rz_list.h>
@ -567,11 +568,13 @@ typedef struct search_ctx {
static void print_intervals(RZ_NONNULL RzThreadQueue *intervals) {
rz_return_if_fail(intervals);
RzSearchInterval *search_interval = NULL;
while ((search_interval = rz_th_queue_pop(intervals, false))) {
void *data = NULL;
while (rz_th_queue_pop(intervals, false, &data) && data) {
RzSearchInterval *search_interval = (RzSearchInterval *)data;
RzInterval *itv = &search_interval->interval;
eprintf("[0x%" PFMT64x ", 0x%" PFMT64x "): %" PFMTSZu "\n", itv->addr, itv->addr + itv->size,
search_interval->n_hits);
data = NULL;
}
}
@ -788,6 +791,7 @@ RZ_API RZ_OWN RzList /*<RzSearchHit *>*/ *rz_search_on_io(
if (cancel_th) {
// stop & free cancel thread.
rz_atomic_bool_set(ctx.loop, false);
rz_th_queue_close_when_empty(intervals);
rz_th_wait(cancel_th);
rz_th_free(cancel_th);
rz_atomic_bool_free(ctx.loop);

11
librz/search/string_search.c
View file

@ -1,5 +1,6 @@
// SPDX-FileCopyrightText: 2024 RizinOrg <info@rizin.re>
// SPDX-FileCopyrightText: 2024 deroad <wargio@libero.it>
// SPDX-FileCopyrightText: 2024-2025 RizinOrg <info@rizin.re>
// SPDX-FileCopyrightText: 2024-2025 Rot127 <rot127@posteo.com>
// SPDX-FileCopyrightText: 2024-2025 deroad <deroad@kumo.xn--q9jyb4c>
// SPDX-License-Identifier: LGPL-3.0-only
#include <rz_search.h>
@ -219,11 +220,13 @@ static bool string_find(RzSearchFindOpt *fopt, void *user, ut64 offset, const Rz
RZ_OUT RzThreadQueue *hits, RZ_OUT size_t *n_hits) {
rz_return_val_if_fail(fopt, false);
void *data = NULL;
StringSearch *ss = (StringSearch *)user;
size_t *thread_id = rz_th_queue_pop(ss->thread_ids, false);
if (!thread_id) {
if (!rz_th_queue_pop(ss->thread_ids, false, &data) || !data) {
return false;
}
size_t *thread_id = data;
RzPVector *strings = rz_pvector_at(ss->strings, *thread_id);
void **it_m = NULL;

21
librz/util/thread_cond.c
View file

@ -1,5 +1,5 @@
// SPDX-FileCopyrightText: 2009-2020 thestr4ng3r <info@florianmaerkl.de>
// SPDX-FileCopyrightText: 2022 deroad <wargio@libero.it>
// SPDX-FileCopyrightText: 2022-2025 deroad <deroad@kumo.xn--q9jyb4c>
// SPDX-License-Identifier: LGPL-3.0-only
#include "thread.h"
@ -69,6 +69,25 @@ RZ_API void rz_th_cond_wait(RZ_NONNULL RzThreadCond *cond, RZ_NONNULL RzThreadLo
#endif
}
/**
* \brief The function shall block up to a given timeout 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)
* \param timeout_ms How many millisecs it needs to wait before it is ok to continue.
*/
RZ_API void rz_th_cond_timed_wait(RZ_NONNULL RzThreadCond *cond, RZ_NONNULL RzThreadLock *lock, size_t timeout_ms) {
rz_return_if_fail(cond);
#if HAVE_PTHREAD
struct timespec timeout;
timeout.tv_sec = timeout_ms / 1000;
timeout.tv_nsec = (timeout_ms % 1000) * 1000000ull;
pthread_cond_timedwait(&cond->cond, &lock->lock, &timeout);
#elif __WINDOWS__
SleepConditionVariableCS(&cond->cond, &lock->lock, timeout_ms);
#endif
}
/**
* \brief Frees a RzThreadCond struct
*

256
librz/util/thread_queue.c
View file

@ -1,4 +1,4 @@
// SPDX-FileCopyrightText: 2022 deroad <wargio@libero.it>
// SPDX-FileCopyrightText: 2022-2025 deroad <deroad@kumo.xn--q9jyb4c>
// SPDX-License-Identifier: LGPL-3.0-only
#include <rz_th.h>
@ -15,12 +15,42 @@
* 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;
RzThreadQueueSize max_size;
RzList /*<void *>*/ *list;
RzThreadLock *reader_lock; ///< Lock for readers
RzThreadCond *reader_cond; ///< Cond for readers
size_t reader_awaiting; ///< Number of readers awaiting to read
RzThreadCond *empty_cond; ///< Cond any thread waiting for an empty queue
RzThreadLock *data_lock; ///< Lock used to modify the RzThreadQueue data
RzThreadQueueSize max_size; ///< Max queue size or unlimited
RzList /*<void *>*/ *list; ///< Stored data
bool closed; ///< Defines if the queue is closed (i.e. reads or writes cannot be performed).
};
static RZ_OWN RzThreadQueue *th_queue_new(RzThreadQueueSize max_size, RzList /*<void *>*/ *list) {
RzThreadQueue *queue = RZ_NEW0(RzThreadQueue);
if (!queue) {
return NULL;
}
queue->max_size = max_size;
queue->list = list;
queue->data_lock = rz_th_lock_new(false);
queue->reader_lock = rz_th_lock_new(false);
queue->reader_cond = rz_th_cond_new();
queue->empty_cond = rz_th_cond_new();
if (!queue->list ||
!queue->data_lock ||
!queue->reader_lock ||
!queue->reader_cond ||
!queue->empty_cond) {
rz_th_queue_free(queue);
return NULL;
}
return queue;
}
/**
* \brief Allocates and initializes a new fifo queue
*
@ -30,21 +60,12 @@ struct rz_th_queue_t {
* \return On success returns a valid pointer, otherwise NULL
*/
RZ_API RZ_OWN RzThreadQueue *rz_th_queue_new(RzThreadQueueSize max_size, RZ_NULLABLE RzListFree qfree) {
RzThreadQueue *queue = RZ_NEW0(RzThreadQueue);
if (!queue) {
RzList *list = rz_list_newf(qfree);
if (!list) {
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;
return th_queue_new(max_size, list);
}
/**
@ -56,27 +77,15 @@ RZ_API RZ_OWN RzThreadQueue *rz_th_queue_new(RzThreadQueueSize max_size, RZ_NULL
*/
RZ_API RZ_OWN RzThreadQueue *rz_th_queue_from_list(RZ_NONNULL RZ_BORROW RzList /*<void *>*/ *list, RZ_NULLABLE RzListFree qfree) {
rz_return_val_if_fail(list, NULL);
RzThreadQueue *queue = RZ_NEW0(RzThreadQueue);
if (!queue) {
size_t max_size = rz_list_length(list);
RzList *copy = rz_list_clone(list);
if (!list) {
return NULL;
}
queue->list = rz_list_clone(list);
if (!queue->list) {
free(queue);
return NULL;
}
queue->list->free = qfree;
queue->max_size = rz_list_length(list);
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;
copy->free = qfree;
return th_queue_new(max_size, copy);
}
/**
@ -89,8 +98,10 @@ RZ_API RZ_OWN RzThreadQueue *rz_th_queue_from_list(RZ_NONNULL RZ_BORROW RzList /
*/
RZ_API RZ_OWN RzThreadQueue *rz_th_queue_from_pvector(RZ_NONNULL RZ_BORROW RzPVector /*<void *>*/ *vector, RZ_NULLABLE RzListFree qfree) {
rz_return_val_if_fail(vector, NULL);
RzThreadQueue *queue = rz_th_queue_new(rz_pvector_len(vector), qfree);
if (!queue) {
size_t max_size = rz_pvector_len(vector);
RzList *list = rz_list_newf(qfree);
if (!list) {
return NULL;
}
@ -100,13 +111,69 @@ RZ_API RZ_OWN RzThreadQueue *rz_th_queue_from_pvector(RZ_NONNULL RZ_BORROW RzPVe
if (!value) {
continue;
}
if (!rz_list_append(queue->list, value)) {
rz_th_queue_free(queue);
if (!rz_list_append(list, value)) {
rz_list_free(list);
return NULL;
}
}
return queue;
return th_queue_new(max_size, list);
}
/**
* \brief Closes a RzThreadQueue but only when empty (once closed you cannot read/write data).
*
* \param queue The RzThreadQueue to close
*/
RZ_API void rz_th_queue_close_when_empty(RZ_NONNULL RzThreadQueue *queue) {
rz_return_if_fail(queue);
rz_th_lock_enter(queue->data_lock);
if (queue->closed) {
// already closed.
rz_th_lock_leave(queue->data_lock);
return;
}
while (!rz_list_empty(queue->list)) {
// the list is not empty, so we wait for it to be empty.
rz_th_cond_timed_wait(queue->empty_cond, queue->data_lock, 100);
}
// we finally close the queue & notify all awating readers
queue->closed = true;
rz_th_cond_signal_all(queue->reader_cond);
rz_th_lock_leave(queue->data_lock);
}
/**
* \brief Closes a RzThreadQueue (once closed you cannot read/write data).
*
* \param queue The RzThreadQueue to close
*/
RZ_API void rz_th_queue_close(RZ_NONNULL RzThreadQueue *queue) {
rz_return_if_fail(queue);
rz_th_lock_enter(queue->data_lock);
if (!queue->closed) {
queue->closed = true;
rz_th_cond_signal_all(queue->reader_cond);
}
rz_th_lock_leave(queue->data_lock);
}
/**
* \brief Returns true if a given RzThreadQueue is closed.
*
* \param queue The RzThreadQueue to check if is closed or not.
*/
RZ_API bool rz_th_queue_is_closed(RZ_NONNULL RzThreadQueue *queue) {
rz_return_val_if_fail(queue, false);
rz_th_lock_enter(queue->data_lock);
bool closed = queue->closed;
rz_th_lock_leave(queue->data_lock);
return closed;
}
/**
@ -119,9 +186,14 @@ RZ_API void rz_th_queue_free(RZ_NULLABLE RzThreadQueue *queue) {
return;
}
// always close before freeing it.
rz_th_queue_close(queue);
rz_list_free(queue->list);
rz_th_lock_free(queue->lock);
rz_th_cond_free(queue->cond);
rz_th_cond_free(queue->empty_cond);
rz_th_lock_free(queue->reader_lock);
rz_th_cond_free(queue->reader_cond);
rz_th_lock_free(queue->data_lock);
free(queue);
}
@ -138,7 +210,13 @@ RZ_API bool rz_th_queue_push(RZ_NONNULL RzThreadQueue *queue, RZ_NONNULL void *u
rz_return_val_if_fail(queue && user, false);
bool added = false;
rz_th_lock_enter(queue->lock);
rz_th_lock_enter(queue->data_lock);
if (queue->closed) {
// never write to a closed queue
goto end;
}
if (!queue->max_size || rz_list_length(queue->list) < queue->max_size) {
if (tail) {
added = rz_list_append(queue->list, user) != NULL;
@ -146,58 +224,62 @@ RZ_API bool rz_th_queue_push(RZ_NONNULL RzThreadQueue *queue, RZ_NONNULL void *u
added = rz_list_prepend(queue->list, user) != NULL;
}
}
if (added) {
rz_th_cond_signal(queue->cond);
if (!added) {
// we failed to add an element to the queue, so we return.
goto end;
}
rz_th_lock_leave(queue->lock);
// we notify a reader that there is new data
if (queue->reader_awaiting > 0) {
// we notify a reader that there is now data.
rz_th_cond_signal(queue->reader_cond);
}
end:
rz_th_lock_leave(queue->data_lock);
return added;
}
/**
* \brief Removes an element from the queue, but does not awaits when empty.
* \brief Removes an element from the queue. It blocks until it can read or the queue is closed.
*
* \param queue The RzThreadQueue to pop from
* \param tail When true, pops the element from the tail, otherwise from the head
* \param data The data removed from the queue
*
* \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);
RZ_API bool rz_th_queue_pop(RZ_NONNULL RzThreadQueue *queue, bool tail, RZ_NONNULL RZ_OUT void **data) {
rz_return_val_if_fail(queue && data, NULL);
bool ret = false;
rz_th_lock_enter(queue->reader_lock);
rz_th_lock_enter(queue->data_lock);
while (!queue->closed && rz_list_empty(queue->list)) {
// the queue is not closed and we need to wait till there is new data
queue->reader_awaiting++;
rz_th_cond_wait(queue->reader_cond, queue->data_lock);
queue->reader_awaiting--;
}
if (queue->closed) {
// the queue is closed, nothing to do.
goto end;
}
void *user = NULL;
rz_th_lock_enter(queue->lock);
if (tail) {
user = rz_list_pop(queue->list);
*data = rz_list_pop(queue->list);
} else {
user = rz_list_pop_head(queue->list);
*data = rz_list_pop_head(queue->list);
}
rz_th_lock_leave(queue->lock);
return user;
}
ret = true;
/**
* \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;
end:
rz_th_lock_leave(queue->data_lock);
rz_th_lock_leave(queue->reader_lock);
return ret;
}
/**
@ -210,9 +292,9 @@ RZ_API RZ_OWN void *rz_th_queue_wait_pop(RZ_NONNULL RzThreadQueue *queue, bool t
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);
rz_th_lock_enter(queue->data_lock);
bool is_empty = rz_list_empty(queue->list);
rz_th_lock_leave(queue->lock);
rz_th_lock_leave(queue->data_lock);
return is_empty;
}
@ -226,9 +308,9 @@ RZ_API bool rz_th_queue_is_empty(RZ_NONNULL RzThreadQueue *queue) {
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);
rz_th_lock_enter(queue->data_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);
rz_th_lock_leave(queue->data_lock);
return is_full;
}
@ -242,9 +324,9 @@ RZ_API bool rz_th_queue_is_full(RZ_NONNULL RzThreadQueue *queue) {
RZ_API size_t rz_th_queue_size(RZ_NONNULL RzThreadQueue *queue) {
rz_return_val_if_fail(queue, false);
rz_th_lock_enter(queue->lock);
rz_th_lock_enter(queue->data_lock);
size_t size = rz_list_length(queue->list);
rz_th_lock_leave(queue->lock);
rz_th_lock_leave(queue->data_lock);
return size;
}
@ -263,9 +345,9 @@ RZ_API RZ_OWN RzList /*<void *>*/ *rz_th_queue_pop_all(RZ_NONNULL RzThreadQueue
return NULL;
}
rz_th_lock_enter(queue->lock);
rz_th_lock_enter(queue->data_lock);
RzList *res = queue->list;
queue->list = list;
rz_th_lock_leave(queue->lock);
rz_th_lock_leave(queue->data_lock);
return res;
}

3
test/integration/test_str_search.c
View file

@ -121,6 +121,8 @@ int test_rz_str_search_single_simple(void) {
free(hit_str);
}
rz_search_collection_free(collection);
rz_list_free(hits);
mu_end;
}
@ -280,6 +282,7 @@ int test_rz_str_search_multiple_enc(void) {
mu_assert_eq(hit->size, 10, "Incorrect size");
rz_list_free(hits);
rz_search_collection_free(collection);
mu_end;
}

47
test/unit/test_threads.c
View file

@ -44,15 +44,28 @@ bool test_thread_pool_cores(void) {
mu_end;
}
#define THREAD_WAIT_AT_LEAST_MICROSEC 1500000
void *thread_queue_push_timed(RzThreadQueue *queue) {
rz_sys_sleep(2);
return rz_th_queue_push(queue, queue, true) ? queue : NULL;
void *data = NULL;
ut64 start = rz_time_now();
if (!rz_th_queue_pop(queue, true, &data)) {
return NULL;
}
ut64 diff = rz_time_now() - start;
if (diff < THREAD_WAIT_AT_LEAST_MICROSEC) {
return "did not wait for " RZ_STR(THREAD_WAIT_AT_LEAST_MICROSEC) " microsec";
} else if (!strcmp((const char *)data, "rizin")) {
return "OK";
}
return "did not receive 'rizin'";
}
bool test_thread_queue(void) {
// test limited queue
void *head = (void *)"aaaaaa";
void *tail = (void *)"bbbbbb";
void *pop_data = NULL;
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");
@ -61,10 +74,20 @@ bool test_thread_queue(void) {
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_true(rz_th_queue_pop(queue, false, &pop_data), "queue can pop head");
mu_assert_ptreq(pop_data, head, "queue popped head and is head");
pop_data = NULL;
mu_assert_true(rz_th_queue_pop(queue, true, &pop_data), "queue can pop tail");
mu_assert_ptreq(pop_data, tail, "queue popped tail and is tail");
mu_assert_false(rz_th_queue_is_empty(queue), "queue is empty");
mu_assert_false(rz_th_queue_is_closed(queue), "queue is not closed");
mu_assert_false(rz_th_queue_is_full(queue), "queue is not full");
// close queue, so no read/writes can happen
pop_data = NULL;
rz_th_queue_close(queue);
mu_assert_true(rz_th_queue_is_closed(queue), "queue is closed");
mu_assert_false(rz_th_queue_push(queue, "cccccc", true), "closed queue cannot push new data");
mu_assert_false(rz_th_queue_pop(queue, false, &pop_data), "closed queue cannot pop new data");
rz_th_queue_free(queue);
// test unlimited queue
@ -85,14 +108,18 @@ bool test_thread_queue(void) {
RzThread *th = rz_th_new((RzThreadFunction)thread_queue_push_timed, queue);
mu_assert_false(rz_th_terminated(th), "Thread should still sleep and count as running.");
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_sys_sleep(2);
mu_assert_true(rz_th_queue_push(queue, "rizin", true), "queue can push an element after 2 sec of waiting");
// we wait for the queue to be empty
rz_th_queue_close_when_empty(queue);
rz_th_wait(th);
const char *thread_string = rz_th_get_retv(th);
mu_assert_notnull(thread_string, "thread retuned non-null value");
mu_assert_streq(thread_string, "OK", "thread retuned the 'OK' string");
mu_assert_true(rz_th_terminated(th), "Thread should count as terminated.");
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");
mu_assert_true(rz_th_queue_is_closed(queue), "verify the queue is closed");
rz_th_free(th);
rz_th_queue_free(queue);