Commit 45dbb8130 for clamav.net
commit 45dbb8130706990c4d6de0660adb58f116e4197c
Author: Valerie Snyder <valsnyde@cisco.com>
Date: Tue Aug 4 15:59:39 2026 -0400
clamd: Synchronize STATS pool snapshots
Protect thread-pool counters and queue traversal with the pool
mutex while STATS captures its response. Preserve the
pools_lock-to-pool_mutex lock order during worker initialization
and pool destruction.
Make worker retirement recheck queued work before decrementing
the live-thread count so dispatch cannot leave work stranded in
the retirement window. Retain engine references while collecting
memory-pool statistics instead of racing on the reference count.
Add regression coverage for queue churn and work dispatched while
a worker is retiring.
CLAM-3052
diff --git a/clamd/thrmgr.c b/clamd/thrmgr.c
index f6775a486..ec2c0859e 100644
--- a/clamd/thrmgr.c
+++ b/clamd/thrmgr.c
@@ -126,6 +126,15 @@ struct stats_buffer {
size_t capacity;
};
+struct queue_stats {
+ long min_wait;
+ long max_wait;
+ long total_wait;
+ unsigned valid;
+ unsigned invalid;
+ unsigned item_count;
+};
+
/**
* @brief Free a buffered STATS response.
*
@@ -263,54 +272,69 @@ static void remove_frompools(threadpool_t *t)
}
/**
- * @brief Append queue timing statistics to a buffered STATS response.
+ * @brief Copy timing statistics from a work queue.
*
- * The caller must ensure that the queue remains valid while it is read.
+ * The caller must hold the owning thread pool's mutex.
*
- * @param buffer Response buffer to extend.
* @param queue Queue to summarize.
* @param tv_now Time used to calculate queue wait durations.
- * @return CL_SUCCESS on success, or an error status.
+ * @param stats Queue statistics to populate.
*/
-static cl_error_t stats_buffer_append_queue(struct stats_buffer *buffer, work_queue_t *queue, struct timeval *tv_now)
+static void stats_snapshot_queue(const work_queue_t *queue, const struct timeval *tv_now, struct queue_stats *stats)
{
- cl_error_t status;
- long umin = LONG_MAX, umax = 0, usum = 0;
- unsigned invalids = 0, cnt = 0;
- work_item_t *q;
+ const work_item_t *q;
+
+ memset(stats, 0, sizeof(*stats));
+ stats->min_wait = LONG_MAX;
+ stats->item_count = (unsigned)queue->item_count;
- if (!queue->head)
- return CL_SUCCESS;
for (q = queue->head; q; q = q->next) {
long delta;
delta = tv_now->tv_usec - q->time_queued.tv_usec;
delta += (tv_now->tv_sec - q->time_queued.tv_sec) * 1000000;
if (delta < 0) {
- invalids++;
+ stats->invalid++;
continue;
}
- if (delta > umax)
- umax = delta;
- if (delta < umin)
- umin = delta;
- usum += delta;
- ++cnt;
+ if (delta > stats->max_wait)
+ stats->max_wait = delta;
+ if (delta < stats->min_wait)
+ stats->min_wait = delta;
+ stats->total_wait += delta;
+ stats->valid++;
}
+}
+
+/**
+ * @brief Append a work queue snapshot to a buffered STATS response.
+ *
+ * @param buffer Response buffer to extend.
+ * @param stats Queue statistics to append.
+ * @return CL_SUCCESS on success, or an error status.
+ */
+static cl_error_t stats_buffer_append_queue(struct stats_buffer *buffer, const struct queue_stats *stats)
+{
+ cl_error_t status;
+
+ if (!stats->item_count)
+ return CL_SUCCESS;
status = stats_buffer_append(buffer, " min_wait: %.6f max_wait: %.6f avg_wait: %.6f",
- umin / 1e6, umax / 1e6, usum / (1e6 * cnt));
+ stats->valid ? stats->min_wait / 1e6 : 0.0,
+ stats->valid ? stats->max_wait / 1e6 : 0.0,
+ stats->valid ? stats->total_wait / (1e6 * stats->valid) : 0.0);
if (CL_SUCCESS != status)
return status;
- if (invalids) {
- status = stats_buffer_append(buffer, " (INVALID timestamps: %u)", invalids);
+ if (stats->invalid) {
+ status = stats_buffer_append(buffer, " (INVALID timestamps: %u)", stats->invalid);
if (CL_SUCCESS != status)
return status;
}
- if (cnt + invalids != (unsigned)queue->item_count) {
- status = stats_buffer_append(buffer, " (ERROR: %u != %u)", cnt + invalids,
- (unsigned)queue->item_count);
+ if (stats->valid + stats->invalid != stats->item_count) {
+ status = stats_buffer_append(buffer, " (ERROR: %u != %u)", stats->valid + stats->invalid,
+ stats->item_count);
if (CL_SUCCESS != status)
return status;
}
@@ -318,6 +342,94 @@ static cl_error_t stats_buffer_append_queue(struct stats_buffer *buffer, work_qu
return CL_SUCCESS;
}
+/**
+ * @brief Append a synchronized thread pool snapshot.
+ *
+ * The caller must hold pools_lock, which pins the pool while its mutex is
+ * acquired. No code may acquire pools_lock while holding a pool mutex.
+ *
+ * @param buffer Response buffer to extend.
+ * @param pool Thread pool to summarize.
+ * @param primary Whether this is the primary pool.
+ * @param tv_now Time captured for the subsequent task snapshot.
+ * @return CL_SUCCESS on success, or an error status.
+ */
+static cl_error_t stats_buffer_append_pool(struct stats_buffer *buffer, threadpool_t *pool, int primary, struct timeval *tv_now)
+{
+ struct queue_stats bulk_stats;
+ struct queue_stats single_stats;
+ cl_error_t status = CL_SUCCESS;
+ pool_state_t pool_state;
+ unsigned thr_alive;
+ unsigned thr_idle;
+ unsigned thr_max;
+ unsigned idle_timeout;
+ unsigned queue_items;
+ const char *state;
+
+ if (pthread_mutex_lock(&pool->pool_mutex) != 0) {
+ logg(LOGG_ERROR, "Unable to lock thread pool mutex for statistics\n");
+ return CL_ELOCK;
+ }
+
+ pool_state = pool->state;
+ thr_alive = (unsigned)pool->thr_alive;
+ thr_idle = (unsigned)pool->thr_idle;
+ thr_max = (unsigned)pool->thr_max;
+ idle_timeout = (unsigned)pool->idle_timeout;
+ queue_items = (unsigned)pool->single_queue->item_count + (unsigned)pool->bulk_queue->item_count;
+ gettimeofday(tv_now, NULL);
+ stats_snapshot_queue(pool->bulk_queue, tv_now, &bulk_stats);
+ stats_snapshot_queue(pool->single_queue, tv_now, &single_stats);
+
+ if (pthread_mutex_unlock(&pool->pool_mutex) != 0) {
+ logg(LOGG_ERROR, "Unable to unlock thread pool mutex for statistics\n");
+ return CL_ELOCK;
+ }
+
+ switch (pool_state) {
+ case POOL_INVALID:
+ state = "INVALID";
+ break;
+ case POOL_VALID:
+ state = "VALID";
+ break;
+ case POOL_EXIT:
+ state = "EXIT";
+ break;
+ default:
+ state = "??";
+ break;
+ }
+
+ status = stats_buffer_append(buffer, "STATE: %s %s\n", state, primary ? "PRIMARY" : "");
+ if (CL_SUCCESS != status)
+ goto done;
+
+ status = stats_buffer_append(buffer, "THREADS: live %u idle %u max %u idle-timeout %u\n",
+ thr_alive, thr_idle, thr_max, idle_timeout);
+ if (CL_SUCCESS != status)
+ goto done;
+
+ /* TODO: show both queues */
+ status = stats_buffer_append(buffer, "QUEUE: %u items", queue_items);
+ if (CL_SUCCESS != status)
+ goto done;
+
+ status = stats_buffer_append_queue(buffer, &bulk_stats);
+ if (CL_SUCCESS != status)
+ goto done;
+
+ status = stats_buffer_append_queue(buffer, &single_stats);
+ if (CL_SUCCESS != status)
+ goto done;
+
+ status = stats_buffer_append(buffer, "\n");
+
+done:
+ return status;
+}
+
/**
* @brief Create an immutable snapshot of the STATS response.
*
@@ -353,10 +465,8 @@ static cl_error_t stats_response_create(struct stats_buffer *response)
for (l = pools; l; l = l->nxt) {
threadpool_t *pool = l->pool;
- const char *state;
struct timeval tv_now;
struct task_desc *task;
- cnt = 0;
if (!pool) {
status = stats_buffer_append(response, "NULL\n\n");
@@ -365,43 +475,7 @@ static cl_error_t stats_response_create(struct stats_buffer *response)
continue;
}
- switch (pool->state) {
- case POOL_INVALID:
- state = "INVALID";
- break;
- case POOL_VALID:
- state = "VALID";
- break;
- case POOL_EXIT:
- state = "EXIT";
- break;
- default:
- state = "??";
- break;
- }
-
- status = stats_buffer_append(response, "STATE: %s %s\n", state, l->nxt ? "" : "PRIMARY");
- if (CL_SUCCESS != status)
- goto unlock;
- status = stats_buffer_append(response, "THREADS: live %u idle %u max %u idle-timeout %u\n",
- pool->thr_alive, pool->thr_idle, pool->thr_max, pool->idle_timeout);
- if (CL_SUCCESS != status)
- goto unlock;
-
- /* TODO: show both queues */
- status = stats_buffer_append(response, "QUEUE: %u items",
- pool->single_queue->item_count + pool->bulk_queue->item_count);
- if (CL_SUCCESS != status)
- goto unlock;
-
- gettimeofday(&tv_now, NULL);
- status = stats_buffer_append_queue(response, pool->bulk_queue, &tv_now);
- if (CL_SUCCESS != status)
- goto unlock;
- status = stats_buffer_append_queue(response, pool->single_queue, &tv_now);
- if (CL_SUCCESS != status)
- goto unlock;
- status = stats_buffer_append(response, "\n");
+ status = stats_buffer_append_pool(response, pool, !l->nxt, &tv_now);
if (CL_SUCCESS != status)
goto unlock;
@@ -560,12 +634,13 @@ void thrmgr_destroy(threadpool_t *threadpool)
return;
}
}
- remove_frompools(threadpool);
if (pthread_mutex_unlock(&threadpool->pool_mutex) != 0) {
logg(LOGG_ERROR, "Mutex unlock failed\n");
exit(-1);
}
+ remove_frompools(threadpool);
+
pthread_mutex_destroy(&(threadpool->pool_mutex));
pthread_cond_destroy(&(threadpool->idle_cond));
pthread_cond_destroy(&(threadpool->queueable_single_cond));
@@ -813,7 +888,8 @@ void thrmgr_setactiveengine(const struct cl_engine *engine)
logg(LOGG_ERROR, "Unable to unlock task statistics mutex\n");
}
-/* thread pool mutex must be held on entry */
+/* Must be called without pool_mutex to preserve the pools_lock -> pool_mutex
+ * lock order used when collecting statistics. */
static void stats_init(threadpool_t *pool)
{
struct task_desc *desc = calloc(1, sizeof(*desc));
@@ -844,13 +920,22 @@ static void stats_init(threadpool_t *pool)
pthread_mutex_unlock(&pools_lock);
}
-/* thread pool mutex must be held on entry */
-static void stats_destroy(threadpool_t *pool)
+/**
+ * @brief Unlink the calling worker's task descriptor.
+ *
+ * The caller must hold pools_lock. The returned descriptor is no longer
+ * visible to STATS and may be freed after releasing pools_lock.
+ *
+ * @param pool Thread pool owning the calling worker.
+ * @return The unlinked descriptor, or NULL if statistics were not initialized.
+ */
+static struct task_desc *stats_unlink_locked(threadpool_t *pool)
{
struct task_desc *desc = pthread_getspecific(stats_tls_key);
+
if (!desc)
- return;
- pthread_mutex_lock(&pools_lock);
+ return NULL;
+
if (desc->prv)
desc->prv->nxt = desc->nxt;
if (desc->nxt)
@@ -858,8 +943,8 @@ static void stats_destroy(threadpool_t *pool)
if (pool->tasks == desc)
pool->tasks = desc->nxt;
pthread_setspecific(stats_tls_key, NULL);
- pthread_mutex_unlock(&pools_lock);
- task_desc_free(desc);
+
+ return desc;
}
static inline int thrmgr_contended(threadpool_t *pool, int bulk)
@@ -921,20 +1006,21 @@ static void *thrmgr_pop(threadpool_t *pool)
static void *thrmgr_worker(void *arg)
{
threadpool_t *threadpool = (threadpool_t *)arg;
+ struct task_desc *desc;
void *job_data;
- int retval, must_exit = FALSE, stats_inited = FALSE;
+ int retval, must_exit;
struct timespec timeout;
+ stats_init(threadpool);
+
/* loop looking for work */
for (;;) {
+ must_exit = FALSE;
+
if (pthread_mutex_lock(&(threadpool->pool_mutex)) != 0) {
logg(LOGG_ERROR, "Fatal: mutex lock failed\n");
exit(-2);
}
- if (!stats_inited) {
- stats_init(threadpool);
- stats_inited = TRUE;
- }
thrmgr_setactiveengine(NULL);
thrmgr_setactivetask(NULL, IDLE_TASK);
timeout.tv_sec = time(NULL) + threadpool->idle_timeout;
@@ -962,26 +1048,54 @@ static void *thrmgr_worker(void *arg)
if (job_data) {
threadpool->handler(job_data);
} else if (must_exit) {
- break;
+ /* A dispatcher can add work after the timed wait releases the
+ * pool mutex but before this worker retires. Make the final
+ * decision while holding both locks in the same order used by
+ * STATS so dispatch either observes this worker alive or starts
+ * a replacement after it has retired. */
+ if (pthread_mutex_lock(&pools_lock) != 0) {
+ logg(LOGG_ERROR, "Fatal: pools mutex lock failed\n");
+ exit(-2);
+ }
+ if (pthread_mutex_lock(&(threadpool->pool_mutex)) != 0) {
+ logg(LOGG_ERROR, "Fatal: mutex lock failed\n");
+ exit(-2);
+ }
+
+ if (threadpool->state == POOL_VALID &&
+ (threadpool->single_queue->item_count != 0 ||
+ threadpool->bulk_queue->item_count != 0)) {
+ if (pthread_mutex_unlock(&(threadpool->pool_mutex)) != 0) {
+ logg(LOGG_ERROR, "Fatal: mutex unlock failed\n");
+ exit(-2);
+ }
+ if (pthread_mutex_unlock(&pools_lock) != 0) {
+ logg(LOGG_ERROR, "Fatal: pools mutex unlock failed\n");
+ exit(-2);
+ }
+ continue;
+ }
+
+ desc = stats_unlink_locked(threadpool);
+ threadpool->thr_alive--;
+ if (threadpool->thr_alive == 0) {
+ /* signal that all threads are finished */
+ pthread_cond_broadcast(&threadpool->pool_cond);
+ }
+
+ if (pthread_mutex_unlock(&(threadpool->pool_mutex)) != 0) {
+ logg(LOGG_ERROR, "Fatal: mutex unlock failed\n");
+ exit(-2);
+ }
+ if (pthread_mutex_unlock(&pools_lock) != 0) {
+ logg(LOGG_ERROR, "Fatal: pools mutex unlock failed\n");
+ exit(-2);
+ }
+
+ task_desc_free(desc);
+ return NULL;
}
}
- if (pthread_mutex_lock(&(threadpool->pool_mutex)) != 0) {
- /* Fatal error */
- logg(LOGG_ERROR, "Fatal: mutex lock failed\n");
- exit(-2);
- }
- threadpool->thr_alive--;
- if (threadpool->thr_alive == 0) {
- /* signal that all threads are finished */
- pthread_cond_broadcast(&threadpool->pool_cond);
- }
- stats_destroy(threadpool);
- if (pthread_mutex_unlock(&(threadpool->pool_mutex)) != 0) {
- /* Fatal error */
- logg(LOGG_ERROR, "Fatal: mutex unlock failed\n");
- exit(-2);
- }
- return NULL;
}
static int thrmgr_dispatch_internal(threadpool_t *threadpool, void *user_data, int bulk)
diff --git a/libclamav/mpool.c b/libclamav/mpool.c
index cd4f7f99e..14d2fab06 100644
--- a/libclamav/mpool.c
+++ b/libclamav/mpool.c
@@ -562,8 +562,8 @@ int mpool_getstats(const struct cl_engine *eng, size_t *used, size_t *total)
const struct MPMAP *mpm;
const mpool_t *mp;
- /* checking refcount is not necessary, but safer */
- if (!eng || !eng->refcount)
+ /* The caller must retain a reference while statistics are collected. */
+ if (!eng)
return -1;
mp = eng->mempool;
if (!mp)
diff --git a/libclamav/mpool.h b/libclamav/mpool.h
index e36b3b1ee..7210151db 100644
--- a/libclamav/mpool.h
+++ b/libclamav/mpool.h
@@ -47,6 +47,13 @@ char *cli_mpool_strndup(mpool_t *mpool, const char *s, size_t n);
char *cli_mpool_virname(mpool_t *mpool, const char *virname, unsigned int official);
uint16_t *cli_mpool_hex2ui(mpool_t *mpool, const char *hex);
void mpool_flush(mpool_t *mpool);
+
+/**
+ * @brief Get memory usage statistics for an engine's memory pool.
+ *
+ * The caller must retain a reference to the engine for the duration of the
+ * call.
+ */
int mpool_getstats(const struct cl_engine *engine, size_t *used, size_t *total);
#define MPOOL_MALLOC(a, b) mpool_malloc(a, b)
diff --git a/unit_tests/check_clamd.c b/unit_tests/check_clamd.c
index ba500cff0..548138686 100644
--- a/unit_tests/check_clamd.c
+++ b/unit_tests/check_clamd.c
@@ -1055,6 +1055,7 @@ struct stats_print_thread_state {
pthread_cond_t cond;
int socket_fd;
int result;
+ int started;
int done;
};
@@ -1096,6 +1097,11 @@ static void *stats_print_thread(void *data)
{
struct stats_print_thread_state *state = data;
+ pthread_mutex_lock(&state->mutex);
+ state->started = 1;
+ pthread_cond_broadcast(&state->cond);
+ pthread_mutex_unlock(&state->mutex);
+
state->result = thrmgr_printstats(state->socket_fd, '\n');
close_socket(state->socket_fd);
@@ -1279,6 +1285,203 @@ START_TEST(test_stats_write_does_not_hold_pool_lock)
}
END_TEST
+struct stats_worker_retirement_test_state {
+ pthread_mutex_t mutex;
+ pthread_cond_t cond;
+ unsigned handled;
+};
+
+static void stats_worker_retirement_test_handler(void *data)
+{
+ struct stats_worker_retirement_test_state *state = data;
+
+ pthread_mutex_lock(&state->mutex);
+ state->handled++;
+ pthread_cond_broadcast(&state->cond);
+ pthread_mutex_unlock(&state->mutex);
+}
+
+static void stats_test_sleep_millisecond(void)
+{
+#ifdef _WIN32
+ Sleep(1);
+#else
+ struct timespec delay = { .tv_sec = 0, .tv_nsec = 1000000L };
+ nanosleep(&delay, NULL);
+#endif
+}
+
+START_TEST(test_stats_does_not_strand_work_during_worker_retirement)
+{
+ struct stats_worker_retirement_test_state state;
+ struct stats_print_thread_state print_state;
+ struct task_desc *desc;
+ threadpool_t *threadpool;
+ pthread_t print_thread;
+ char *stats;
+ size_t stats_len;
+ unsigned i;
+ int sockets[2];
+ int worker_is_retiring = 0;
+ int second_job_handled = 0;
+
+ memset(&state, 0, sizeof(state));
+ ck_assert_int_eq(pthread_mutex_init(&state.mutex, NULL), 0);
+ ck_assert_int_eq(pthread_cond_init(&state.cond, NULL), 0);
+
+ threadpool = thrmgr_new(1, 1, 2, stats_worker_retirement_test_handler);
+ ck_assert_ptr_nonnull(threadpool);
+ ck_assert_int_ne(thrmgr_dispatch(threadpool, &state), 0);
+
+ pthread_mutex_lock(&state.mutex);
+ while (state.handled < 1)
+ pthread_cond_wait(&state.cond, &state.mutex);
+ pthread_mutex_unlock(&state.mutex);
+
+ /* Wait until the worker has returned to its timed idle wait, then hold
+ * its task descriptor so STATS retains pools_lock while snapshotting it. */
+ pthread_mutex_lock(&threadpool->pool_mutex);
+ while (threadpool->thr_idle < 1)
+ pthread_cond_wait(&threadpool->idle_cond, &threadpool->pool_mutex);
+ desc = threadpool->tasks;
+ ck_assert_ptr_nonnull(desc);
+ ck_assert_int_eq(pthread_mutex_lock(&desc->mutex), 0);
+ pthread_mutex_unlock(&threadpool->pool_mutex);
+
+ create_tcp_socket_pair(sockets);
+ memset(&print_state, 0, sizeof(print_state));
+ print_state.socket_fd = sockets[1];
+ ck_assert_int_eq(pthread_mutex_init(&print_state.mutex, NULL), 0);
+ ck_assert_int_eq(pthread_cond_init(&print_state.cond, NULL), 0);
+ ck_assert_int_eq(pthread_create(&print_thread, NULL, stats_print_thread, &print_state), 0);
+
+ pthread_mutex_lock(&print_state.mutex);
+ while (!print_state.started)
+ pthread_cond_wait(&print_state.cond, &print_state.mutex);
+ pthread_mutex_unlock(&print_state.mutex);
+
+ /* Once idle is zero but the sole worker is still alive, its timeout has
+ * fired and it is waiting for the global STATS lock before retiring. */
+ for (i = 0; i < 5000; i++) {
+ pthread_mutex_lock(&threadpool->pool_mutex);
+ worker_is_retiring = threadpool->thr_alive == 1 && threadpool->thr_idle == 0;
+ pthread_mutex_unlock(&threadpool->pool_mutex);
+ if (worker_is_retiring)
+ break;
+ stats_test_sleep_millisecond();
+ }
+ ck_assert_msg(worker_is_retiring,
+ "Worker did not reach the retirement checkpoint while STATS held the pool list lock");
+
+ /* The dispatcher sees the retiring worker as alive and therefore does
+ * not create a replacement. The worker must recheck the queue before it
+ * commits to exit. */
+ ck_assert_int_ne(thrmgr_dispatch(threadpool, &state), 0);
+ ck_assert_int_eq(pthread_mutex_unlock(&desc->mutex), 0);
+
+ for (i = 0; i < 5000; i++) {
+ pthread_mutex_lock(&state.mutex);
+ second_job_handled = state.handled == 2;
+ pthread_mutex_unlock(&state.mutex);
+ if (second_job_handled)
+ break;
+ stats_test_sleep_millisecond();
+ }
+
+ stats = recvfull(sockets[0], &stats_len);
+ close_socket(sockets[0]);
+ ck_assert_int_eq(pthread_join(print_thread, NULL), 0);
+
+ ck_assert_msg(second_job_handled,
+ "Work dispatched during worker retirement remained queued without a worker");
+ ck_assert_int_eq(print_state.result, 0);
+ ck_assert_ptr_nonnull(stats);
+ ck_assert_msg(stats_len >= 4 && memcmp(stats + stats_len - 4, "END\n", 4) == 0,
+ "STATS response was truncated during worker retirement:\n%s", stats);
+ free(stats);
+
+ thrmgr_destroy(threadpool);
+ pthread_cond_destroy(&print_state.cond);
+ pthread_mutex_destroy(&print_state.mutex);
+ pthread_cond_destroy(&state.cond);
+ pthread_mutex_destroy(&state.mutex);
+}
+END_TEST
+
+struct stats_queue_churn_test_state {
+ pthread_mutex_t mutex;
+ pthread_cond_t cond;
+ unsigned handled;
+};
+
+static void stats_queue_churn_test_handler(void *data)
+{
+ struct stats_queue_churn_test_state *state = data;
+
+ /* Keep the queue active long enough for repeated snapshots to overlap
+ * with work-item removal. */
+#ifdef _WIN32
+ Sleep(1);
+#else
+ struct timespec delay = { .tv_sec = 0, .tv_nsec = 1000000L };
+ nanosleep(&delay, NULL);
+#endif
+
+ pthread_mutex_lock(&state->mutex);
+ state->handled++;
+ pthread_cond_broadcast(&state->cond);
+ pthread_mutex_unlock(&state->mutex);
+}
+
+START_TEST(test_stats_while_queue_is_changing)
+{
+ enum {
+ STATS_QUEUE_CHURN_JOBS = 256,
+ STATS_QUEUE_CHURN_SNAPSHOTS = 64
+ };
+ struct stats_queue_churn_test_state state;
+ threadpool_t *threadpool;
+ unsigned i;
+
+ memset(&state, 0, sizeof(state));
+ ck_assert_int_eq(pthread_mutex_init(&state.mutex, NULL), 0);
+ ck_assert_int_eq(pthread_cond_init(&state.cond, NULL), 0);
+
+ threadpool = thrmgr_new(1, 60, STATS_QUEUE_CHURN_JOBS * 2,
+ stats_queue_churn_test_handler);
+ ck_assert_ptr_nonnull(threadpool);
+
+ for (i = 0; i < STATS_QUEUE_CHURN_JOBS; i++)
+ ck_assert_int_ne(thrmgr_dispatch(threadpool, &state), 0);
+
+ for (i = 0; i < STATS_QUEUE_CHURN_SNAPSHOTS; i++) {
+ char *stats;
+ size_t stats_len;
+ int sockets[2];
+
+ create_tcp_socket_pair(sockets);
+ ck_assert_int_eq(thrmgr_printstats(sockets[1], '\n'), 0);
+ close_socket(sockets[1]);
+ stats = recvfull(sockets[0], &stats_len);
+ close_socket(sockets[0]);
+
+ ck_assert_ptr_nonnull(stats);
+ ck_assert_msg(stats_len >= 4 && memcmp(stats + stats_len - 4, "END\n", 4) == 0,
+ "STATS response was truncated while the work queue changed:\n%s", stats);
+ free(stats);
+ }
+
+ pthread_mutex_lock(&state.mutex);
+ while (state.handled < STATS_QUEUE_CHURN_JOBS)
+ pthread_cond_wait(&state.cond, &state.mutex);
+ pthread_mutex_unlock(&state.mutex);
+
+ thrmgr_destroy(threadpool);
+ pthread_cond_destroy(&state.cond);
+ pthread_mutex_destroy(&state.mutex);
+}
+END_TEST
+
static Suite *test_clamd_suite(void)
{
Suite *s = suite_create("clamd");
@@ -1317,6 +1520,8 @@ static Suite *test_clamd_suite(void)
suite_add_tcase(s, tc_thrmgr);
tcase_add_test(tc_thrmgr, test_stats_owns_task_filename);
tcase_add_test(tc_thrmgr, test_stats_write_does_not_hold_pool_lock);
+ tcase_add_test(tc_thrmgr, test_stats_does_not_strand_work_during_worker_retirement);
+ tcase_add_test(tc_thrmgr, test_stats_while_queue_is_changing);
return s;
}