Fix deadlock in rz_th_queue_close_when_empty() (#6383)

empty_cond was never signalled and the termination depended only on
the timeout in rz_th_queue_close_when_empty() causing a re-check of
emptiness.
However, the rz_th_cond_timed_wait() implementation, which was used
there, was flawed because it expected a relative timeout but passed that
directly to pthread_cond_timedwait() which expected an absolute time
value, practically causing it to time out immediately. Depending on the
pthread_cond implementation, this possibly created a situation where the
mutex could never be acquired by another thread, effectively causing a
deadlock. This behavior was observed on Mac OS X 10.5 (ppc) when running
the test_core_bin test.
We solve this by not using a timeout at all and signalling the condition
variable for all waiting threads at the appropriate time.

librz/include/rz_th.h

@@ -80,7 +80,6 @@ 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();

librz/util/thread_cond.c

@@ -69,25 +69,6 @@ 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
  *

librz/util/thread_queue.c

@@ -19,7 +19,7 @@ struct rz_th_queue_t {
 	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
+	RzThreadCond *empty_cond; ///< Cond for predicate `closed || rz_list_empty(list)`
 
 	RzThreadLock *data_lock; ///< Lock used to modify the RzThreadQueue data
 	RzThreadQueueSize max_size; ///< Max queue size or unlimited
@@ -123,27 +123,23 @@ RZ_API RZ_OWN RzThreadQueue *rz_th_queue_from_pvector(RZ_NONNULL RZ_BORROW RzPVe
 /**
  * \brief  Closes a RzThreadQueue but only when empty (once closed you cannot read/write data).
  *
+ * This function will block until the queue is empty and then close it.
+ * If the queue is closed from somewhere else, it returns early, even if the queue is not empty.
+ *
  * \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)) {
+	while (!queue->closed && !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);
+		rz_th_cond_wait(queue->empty_cond, queue->data_lock);
+	}
+	if (!queue->closed) {
+		// we finally close the queue & notify all awating readers
+		queue->closed = true;
+		rz_th_cond_signal_all(queue->reader_cond);
 	}
-
-	// 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);
 }
 
@@ -159,6 +155,7 @@ RZ_API void rz_th_queue_close(RZ_NONNULL RzThreadQueue *queue) {
 	if (!queue->closed) {
 		queue->closed = true;
 		rz_th_cond_signal_all(queue->reader_cond);
+		rz_th_cond_signal_all(queue->empty_cond);
 	}
 	rz_th_lock_leave(queue->data_lock);
 }
@@ -248,10 +245,10 @@ end:
  * \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
+ * \return True if an element was popped from the queue, false if the queue was closed
  */
 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);
+	rz_return_val_if_fail(queue && data, false);
 
 	bool ret = false;
 	rz_th_lock_enter(queue->reader_lock);
@@ -274,6 +271,9 @@ RZ_API bool rz_th_queue_pop(RZ_NONNULL RzThreadQueue *queue, bool tail, RZ_NONNU
 	} else {
 		*data = rz_list_pop_head(queue->list);
 	}
+	if (rz_list_empty(queue->list)) {
+		rz_th_cond_signal_all(queue->empty_cond);
+	}
 	ret = true;
 
 end:

test/unit/test_threads.c

@@ -127,6 +127,69 @@ bool test_thread_queue(void) {
 	mu_end;
 }
 
+void *thread_queue_consumer(RzThreadQueue *queue) {
+	ut64 *data;
+	while (rz_th_queue_pop(queue, true, (void **)&data)) {
+		*data = *data * 2;
+	}
+	return NULL;
+}
+
+void *thread_queue_waiter(RzThreadQueue *queue) {
+	rz_th_queue_close_when_empty(queue);
+	return NULL;
+}
+
+bool test_thread_queue_multi_wait(void) {
+	// Test for correct behavior with multiple consumers and empty-waiters,
+	// specifically to discover bugs in condition variable handling.
+	RzThreadQueue *queue = rz_th_queue_new(RZ_THREAD_QUEUE_UNLIMITED, NULL);
+	ut64 items[1000];
+	for (size_t i = 0; i < RZ_ARRAY_SIZE(items); i++) {
+		items[i] = (ut64)i;
+		rz_th_queue_push(queue, &items[i], true);
+	}
+	RzThread *consumer[10];
+	RzThread *waiter[RZ_ARRAY_SIZE(consumer)];
+	for (size_t i = 0; i < RZ_ARRAY_SIZE(consumer); i++) {
+		consumer[i] = rz_th_new((RzThreadFunction)thread_queue_consumer, queue);
+		waiter[i] = rz_th_new((RzThreadFunction)thread_queue_waiter, queue);
+	}
+	for (size_t i = 0; i < RZ_ARRAY_SIZE(consumer); i++) {
+		rz_th_wait(consumer[i]);
+		rz_th_free(consumer[i]);
+		rz_th_wait(waiter[i]);
+		rz_th_free(waiter[i]);
+	}
+	for (size_t i = 0; i < RZ_ARRAY_SIZE(items); i++) {
+		mu_assert_eq(items[i], (ut64)i * 2, "computed result");
+	}
+	rz_th_queue_free(queue);
+	mu_end;
+}
+
+void *thread_queue_closer(RzThreadQueue *queue) {
+	// raise probability that we close while rz_th_queue_close_when_empty() is already waiting on the queue's empty_cond
+	rz_sys_usleep(1000);
+
+	rz_th_queue_close(queue);
+	return NULL;
+}
+
+bool test_thread_queue_nonempty_close(void) {
+	// Test for correct behavior when a queue is closed before being fully empty.
+	RzThreadQueue *queue = rz_th_queue_new(RZ_THREAD_QUEUE_UNLIMITED, NULL);
+	rz_th_queue_push(queue, (void *)(size_t)42, true);
+	RzThread *closer = rz_th_new((RzThreadFunction)thread_queue_closer, queue);
+	rz_th_queue_close_when_empty(queue);
+	rz_th_wait(closer);
+	rz_th_free(closer);
+	mu_assert_eq(rz_th_queue_size(queue), 1, "queue size");
+	mu_assert_true(rz_th_queue_is_closed(queue), "queue closed");
+	rz_th_queue_free(queue);
+	mu_end;
+}
+
 bool test_thread_ht(void) {
 	bool v_boolean = false;
 	const char *element = NULL;
@@ -652,6 +715,8 @@ int all_tests() {
 	mu_run_test(test_thread_limit);
 	mu_run_test(test_thread_pool_cores);
 	mu_run_test(test_thread_queue);
+	mu_run_test(test_thread_queue_multi_wait);
+	mu_run_test(test_thread_queue_nonempty_close);
 	mu_run_test(test_thread_ht);
 	mu_run_test(test_thread_iterator_list);
 	mu_run_test(test_thread_iterator_pvec);