Commit ae91fba29 for clamav.net
commit ae91fba29d2064af055e91cc269f9c2a6bd0d9d1
Author: Valerie Snyder <valsnyde@cisco.com>
Date: Tue Aug 4 10:48:17 2026 -0400
clamd: Own task filenames reported by STATS
The thread manager stores active task filenames supplied by scan
handlers. Some callers pass stack-backed strings, while the STATS
command reads task descriptors from another worker. A delayed STATS
response can therefore read caller storage after it has been reused
and disclose process address bytes.
Give each task descriptor an owned filename and protect its mutable
state with a per-task mutex. Build an immutable STATS response while
pool and task state is protected, then release all daemon bookkeeping
locks before writing to the client. Retain engine references needed by
the response so database reloads remain safe.
Add deterministic tests for caller-storage reuse and for a
non-reading STATS client that must not block pool registration.
CLAM-3052
diff --git a/clamd/thrmgr.c b/clamd/thrmgr.c
index fda0f7576..f6775a486 100644
--- a/clamd/thrmgr.c
+++ b/clamd/thrmgr.c
@@ -28,6 +28,8 @@
#include <pthread.h>
#include <time.h>
#include <errno.h>
+#include <stdarg.h>
+#include <stdint.h>
#include <string.h>
// libclamav
@@ -118,6 +120,104 @@ static struct threadpool_list {
} *pools = NULL;
static pthread_mutex_t pools_lock = PTHREAD_MUTEX_INITIALIZER;
+struct stats_buffer {
+ char *data;
+ size_t length;
+ size_t capacity;
+};
+
+/**
+ * @brief Free a buffered STATS response.
+ *
+ * @param buffer Response buffer to free.
+ */
+static void stats_buffer_cleanup(struct stats_buffer *buffer)
+{
+ if (!buffer)
+ return;
+
+ free(buffer->data);
+ memset(buffer, 0, sizeof(*buffer));
+}
+
+/**
+ * @brief Append formatted text to a buffered STATS response.
+ *
+ * @param buffer Response buffer to extend.
+ * @param format printf-style format string.
+ * @return CL_SUCCESS on success, or an error status.
+ */
+static cl_error_t stats_buffer_append(struct stats_buffer *buffer, const char *format, ...)
+{
+ cl_error_t status = CL_SUCCESS;
+ size_t required;
+ va_list args;
+ int needed;
+ int written;
+
+ va_start(args, format);
+ needed = vsnprintf(NULL, 0, format, args);
+ va_end(args);
+ if (needed < 0)
+ return CL_EFORMAT;
+
+ if ((size_t)needed >= SIZE_MAX - buffer->length)
+ return CL_EMEM;
+ required = buffer->length + (size_t)needed + 1;
+
+ if (required > buffer->capacity) {
+ size_t capacity = buffer->capacity ? buffer->capacity : 1024;
+ char *data;
+
+ while (capacity < required) {
+ if (capacity > SIZE_MAX / 2) {
+ capacity = required;
+ break;
+ }
+ capacity *= 2;
+ }
+
+ data = realloc(buffer->data, capacity);
+ if (!data)
+ return CL_EMEM;
+
+ buffer->data = data;
+ buffer->capacity = capacity;
+ }
+
+ va_start(args, format);
+ written = vsnprintf(buffer->data + buffer->length,
+ buffer->capacity - buffer->length, format, args);
+ va_end(args);
+ if (written != needed) {
+ status = CL_EFORMAT;
+ goto done;
+ }
+
+ buffer->length += (size_t)written;
+
+done:
+ return status;
+}
+
+/**
+ * @brief Free a task descriptor and its owned resources.
+ *
+ * The caller must ensure that the descriptor is no longer visible to STATS
+ * and that its worker thread has stopped using it.
+ *
+ * @param desc Task descriptor to free.
+ */
+static void task_desc_free(struct task_desc *desc)
+{
+ if (!desc)
+ return;
+
+ free(desc->filename);
+ pthread_mutex_destroy(&desc->mutex);
+ free(desc);
+}
+
static void add_topools(threadpool_t *t)
{
struct threadpool_list *new = malloc(sizeof(*new));
@@ -156,20 +256,31 @@ static void remove_frompools(threadpool_t *t)
while (desc) {
struct task_desc *q = desc;
desc = desc->nxt;
- free(q);
+ task_desc_free(q);
}
t->tasks = NULL;
pthread_mutex_unlock(&pools_lock);
}
-static void print_queue(int f, work_queue_t *queue, struct timeval *tv_now)
+/**
+ * @brief Append queue timing statistics to a buffered STATS response.
+ *
+ * The caller must ensure that the queue remains valid while it is read.
+ *
+ * @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.
+ */
+static cl_error_t stats_buffer_append_queue(struct stats_buffer *buffer, work_queue_t *queue, struct timeval *tv_now)
{
+ cl_error_t status;
long umin = LONG_MAX, umax = 0, usum = 0;
unsigned invalids = 0, cnt = 0;
work_item_t *q;
if (!queue->head)
- return;
+ return CL_SUCCESS;
for (q = queue->head; q; q = q->next) {
long delta;
delta = tv_now->tv_usec - q->time_queued.tv_usec;
@@ -185,28 +296,62 @@ static void print_queue(int f, work_queue_t *queue, struct timeval *tv_now)
usum += delta;
++cnt;
}
- mdprintf(f, " min_wait: %.6f max_wait: %.6f avg_wait: %.6f",
- umin / 1e6, umax / 1e6, usum / (1e6 * cnt));
- if (invalids)
- mdprintf(f, " (INVALID timestamps: %u)", invalids);
- if (cnt + invalids != (unsigned)queue->item_count)
- mdprintf(f, " (ERROR: %u != %u)", cnt + invalids,
- (unsigned)queue->item_count);
+
+ status = stats_buffer_append(buffer, " min_wait: %.6f max_wait: %.6f avg_wait: %.6f",
+ umin / 1e6, umax / 1e6, usum / (1e6 * cnt));
+ if (CL_SUCCESS != status)
+ return status;
+
+ if (invalids) {
+ status = stats_buffer_append(buffer, " (INVALID timestamps: %u)", invalids);
+ 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 (CL_SUCCESS != status)
+ return status;
+ }
+
+ return CL_SUCCESS;
}
-int thrmgr_printstats(int f, char term)
+/**
+ * @brief Create an immutable snapshot of the STATS response.
+ *
+ * Pool and task state is formatted into an owned memory buffer while the
+ * corresponding descriptors are protected. The global pool lock is released
+ * before any response bytes are written to the client, so a slow or
+ * non-reading client cannot block scan-worker bookkeeping.
+ *
+ * @param response Response snapshot to populate.
+ * @return CL_SUCCESS on success, or an error status.
+ */
+static cl_error_t stats_response_create(struct stats_buffer *response)
{
struct threadpool_list *l;
+ cl_error_t status = CL_SUCCESS;
unsigned cnt, pool_cnt = 0;
- size_t pool_used = 0, pool_total = 0, seen_cnt = 0, error_flag = 0;
+ size_t pool_used = 0, pool_total = 0, seen_cnt = 0;
float mem_heap = 0, mem_mmap = 0, mem_used = 0, mem_free = 0, mem_releasable = 0;
- const struct cl_engine **seen = NULL;
- int has_libc_memstats = 0;
+ struct cl_engine **seen = NULL;
+ int has_libc_memstats = 0;
+
+ memset(response, 0, sizeof(*response));
+
+ if (pthread_mutex_lock(&pools_lock) != 0) {
+ logg(LOGG_ERROR, "Unable to lock thread pool statistics mutex\n");
+ return CL_ELOCK;
+ }
- pthread_mutex_lock(&pools_lock);
for (cnt = 0, l = pools; l; l = l->nxt) cnt++;
- mdprintf(f, "POOLS: %u\n\n", cnt);
- for (l = pools; l && !error_flag; l = l->nxt) {
+ status = stats_buffer_append(response, "POOLS: %u\n\n", cnt);
+ if (CL_SUCCESS != status)
+ goto unlock;
+
+ for (l = pools; l; l = l->nxt) {
threadpool_t *pool = l->pool;
const char *state;
struct timeval tv_now;
@@ -214,12 +359,12 @@ int thrmgr_printstats(int f, char term)
cnt = 0;
if (!pool) {
- mdprintf(f, "NULL\n\n");
+ status = stats_buffer_append(response, "NULL\n\n");
+ if (CL_SUCCESS != status)
+ goto unlock;
continue;
}
- /* now we can access desc->, knowing that they won't get freed
- * because the other tasks can't quit while pool_mutex is taken
- */
+
switch (pool->state) {
case POOL_INVALID:
state = "INVALID";
@@ -234,26 +379,50 @@ int thrmgr_printstats(int f, char term)
state = "??";
break;
}
- mdprintf(f, "STATE: %s %s\n", state, l->nxt ? "" : "PRIMARY");
- mdprintf(f, "THREADS: live %u idle %u max %u idle-timeout %u\n", pool->thr_alive, pool->thr_idle, pool->thr_max,
- pool->idle_timeout);
+
+ 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 */
- mdprintf(f, "QUEUE: %u items", pool->single_queue->item_count + pool->bulk_queue->item_count);
+ 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);
- print_queue(f, pool->bulk_queue, &tv_now);
- print_queue(f, pool->single_queue, &tv_now);
- mdprintf(f, "\n");
+ 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");
+ if (CL_SUCCESS != status)
+ goto unlock;
+
for (task = pool->tasks; task; task = task->nxt) {
double delta;
- size_t used, total;
+
+ if (pthread_mutex_lock(&task->mutex) != 0) {
+ logg(LOGG_ERROR, "Unable to lock task statistics mutex\n");
+ status = CL_ELOCK;
+ goto unlock;
+ }
delta = tv_now.tv_usec - task->tv.tv_usec;
delta += (tv_now.tv_sec - task->tv.tv_sec) * 1000000.0;
- mdprintf(f, "\t%s %f %s\n",
- task->command ? task->command : "N/A",
- delta / 1e6,
- task->filename ? task->filename : "");
- if (task->engine) {
+
+ status = stats_buffer_append(response, "\t%s %f %s\n",
+ task->command ? task->command : "N/A",
+ delta / 1e6,
+ task->filename ? task->filename : "");
+
+ if (CL_SUCCESS == status && task->engine) {
/* we usually have at most 2 engines so a linear
* search is good enough */
size_t i;
@@ -264,28 +433,59 @@ int thrmgr_printstats(int f, char term)
/* we need to count the memusage from the same
* engine only once */
if (i == seen_cnt) {
- const struct cl_engine **s;
+ struct cl_engine **s;
/* new engine */
- ++seen_cnt;
- s = realloc((void *)seen, seen_cnt * sizeof(*seen));
+ s = realloc(seen, (seen_cnt + 1) * sizeof(*seen));
if (!s) {
- error_flag = 1;
- break;
- }
- seen = s;
- seen[seen_cnt - 1] = task->engine;
-
- if (MPOOL_GETSTATS(task->engine, &used, &total) != -1) {
- pool_used += used;
- pool_total += total;
- pool_cnt++;
+ status = CL_EMEM;
+ } else {
+ seen = s;
+ status = cl_engine_addref((struct cl_engine *)task->engine);
+ if (CL_SUCCESS == status)
+ seen[seen_cnt++] = (struct cl_engine *)task->engine;
}
}
}
+
+ if (pthread_mutex_unlock(&task->mutex) != 0) {
+ logg(LOGG_ERROR, "Unable to unlock task statistics mutex\n");
+ status = CL_ELOCK;
+ }
+
+ if (CL_SUCCESS != status)
+ goto unlock;
+ }
+
+ status = stats_buffer_append(response, "\n");
+ if (CL_SUCCESS != status)
+ goto unlock;
+ }
+
+unlock:
+ if (pthread_mutex_unlock(&pools_lock) != 0) {
+ logg(LOGG_ERROR, "Unable to unlock thread pool statistics mutex\n");
+ status = CL_ELOCK;
+ }
+
+ if (CL_SUCCESS == status) {
+ for (cnt = 0; cnt < seen_cnt; cnt++) {
+ size_t used, total;
+
+ if (MPOOL_GETSTATS(seen[cnt], &used, &total) != -1) {
+ pool_used += used;
+ pool_total += total;
+ pool_cnt++;
+ }
}
- mdprintf(f, "\n");
}
- free((void *)seen);
+
+ for (cnt = 0; cnt < seen_cnt; cnt++)
+ cl_engine_free(seen[cnt]);
+ free(seen);
+
+ if (CL_SUCCESS != status)
+ goto done;
+
#ifdef HAVE_MALLINFO
{
struct mallinfo inf = mallinfo();
@@ -297,19 +497,35 @@ int thrmgr_printstats(int f, char term)
has_libc_memstats = 1;
}
#endif
- if (error_flag) {
- mdprintf(f, "ERROR: error encountered while formatting statistics\n");
+
+ if (has_libc_memstats)
+ status = stats_buffer_append(response,
+ "MEMSTATS: heap %.3fM mmap %.3fM used %.3fM free %.3fM releasable %.3fM pools %u pools_used %.3fM pools_total %.3fM\n",
+ mem_heap, mem_mmap, mem_used, mem_free, mem_releasable, pool_cnt,
+ pool_used / (1024 * 1024.0), pool_total / (1024 * 1024.0));
+ else
+ status = stats_buffer_append(response,
+ "MEMSTATS: heap N/A mmap N/A used N/A free N/A releasable N/A pools %u pools_used %.3fM pools_total %.3fM\n",
+ pool_cnt, pool_used / (1024 * 1024.0), pool_total / (1024 * 1024.0));
+
+done:
+ if (CL_SUCCESS != status)
+ stats_buffer_cleanup(response);
+
+ return status;
+}
+
+int thrmgr_printstats(int f, char term)
+{
+ struct stats_buffer response;
+
+ if (CL_SUCCESS == stats_response_create(&response)) {
+ mdprintf(f, "%sEND%c", response.data, term);
+ stats_buffer_cleanup(&response);
} else {
- if (has_libc_memstats)
- mdprintf(f, "MEMSTATS: heap %.3fM mmap %.3fM used %.3fM free %.3fM releasable %.3fM pools %u pools_used %.3fM pools_total %.3fM\n",
- mem_heap, mem_mmap, mem_used, mem_free, mem_releasable, pool_cnt,
- pool_used / (1024 * 1024.0), pool_total / (1024 * 1024.0));
- else
- mdprintf(f, "MEMSTATS: heap N/A mmap N/A used N/A free N/A releasable N/A pools %u pools_used %.3fM pools_total %.3fM\n",
- pool_cnt, pool_used / (1024 * 1024.0), pool_total / (1024 * 1024.0));
+ mdprintf(f, "ERROR: error encountered while formatting statistics\nEND%c", term);
}
- mdprintf(f, "END%c", term);
- pthread_mutex_unlock(&pools_lock);
+
return 0;
}
@@ -540,21 +756,42 @@ static void stats_tls_key_alloc(void)
static const char *IDLE_TASK = "IDLE";
-/* no mutex is needed, we are using thread local variable */
void thrmgr_setactivetask(const char *filename, const char *cmd)
{
struct task_desc *desc;
+ char *filename_copy = NULL;
+ char *old_filename;
+
pthread_once(&stats_tls_key_once, stats_tls_key_alloc);
desc = pthread_getspecific(stats_tls_key);
if (!desc)
return;
- desc->filename = filename;
+
+ if (filename) {
+ filename_copy = strdup(filename);
+ if (!filename_copy)
+ logg(LOGG_ERROR, "Unable to copy active task filename\n");
+ }
+
+ if (pthread_mutex_lock(&desc->mutex) != 0) {
+ logg(LOGG_ERROR, "Unable to lock task statistics mutex\n");
+ free(filename_copy);
+ return;
+ }
+
+ old_filename = desc->filename;
+ desc->filename = filename_copy;
if (cmd) {
- if (cmd == IDLE_TASK && desc->command == cmd)
- return;
- desc->command = cmd;
- gettimeofday(&desc->tv, NULL);
+ if (!(cmd == IDLE_TASK && desc->command == cmd)) {
+ desc->command = cmd;
+ gettimeofday(&desc->tv, NULL);
+ }
}
+
+ if (pthread_mutex_unlock(&desc->mutex) != 0)
+ logg(LOGG_ERROR, "Unable to unlock task statistics mutex\n");
+
+ free(old_filename);
}
void thrmgr_setactiveengine(const struct cl_engine *engine)
@@ -564,7 +801,16 @@ void thrmgr_setactiveengine(const struct cl_engine *engine)
desc = pthread_getspecific(stats_tls_key);
if (!desc)
return;
+
+ if (pthread_mutex_lock(&desc->mutex) != 0) {
+ logg(LOGG_ERROR, "Unable to lock task statistics mutex\n");
+ return;
+ }
+
desc->engine = engine;
+
+ if (pthread_mutex_unlock(&desc->mutex) != 0)
+ logg(LOGG_ERROR, "Unable to unlock task statistics mutex\n");
}
/* thread pool mutex must be held on entry */
@@ -573,8 +819,21 @@ static void stats_init(threadpool_t *pool)
struct task_desc *desc = calloc(1, sizeof(*desc));
if (!desc)
return;
+
+ if (pthread_mutex_init(&desc->mutex, NULL) != 0) {
+ logg(LOGG_ERROR, "Unable to initialize task statistics mutex\n");
+ free(desc);
+ return;
+ }
+
pthread_once(&stats_tls_key_once, stats_tls_key_alloc);
- pthread_setspecific(stats_tls_key, desc);
+ if (pthread_setspecific(stats_tls_key, desc) != 0) {
+ logg(LOGG_ERROR, "Unable to initialize task statistics state\n");
+ task_desc_free(desc);
+ return;
+ }
+
+ pthread_mutex_lock(&pools_lock);
if (!pool->tasks)
pool->tasks = desc;
else {
@@ -582,6 +841,7 @@ static void stats_init(threadpool_t *pool)
pool->tasks->prv = desc;
pool->tasks = desc;
}
+ pthread_mutex_unlock(&pools_lock);
}
/* thread pool mutex must be held on entry */
@@ -597,9 +857,9 @@ static void stats_destroy(threadpool_t *pool)
desc->nxt->prv = desc->prv;
if (pool->tasks == desc)
pool->tasks = desc->nxt;
- free(desc);
pthread_setspecific(stats_tls_key, NULL);
pthread_mutex_unlock(&pools_lock);
+ task_desc_free(desc);
}
static inline int thrmgr_contended(threadpool_t *pool, int bulk)
diff --git a/clamd/thrmgr.h b/clamd/thrmgr.h
index 3d40dbe33..4595251ff 100644
--- a/clamd/thrmgr.h
+++ b/clamd/thrmgr.h
@@ -49,7 +49,8 @@ typedef enum {
} pool_state_t;
struct task_desc {
- const char *filename;
+ pthread_mutex_t mutex;
+ char *filename;
const char *command;
struct timeval tv;
struct task_desc *prv;
diff --git a/unit_tests/CMakeLists.txt b/unit_tests/CMakeLists.txt
index 00503777c..e23fc92a6 100644
--- a/unit_tests/CMakeLists.txt
+++ b/unit_tests/CMakeLists.txt
@@ -70,7 +70,10 @@ if(ENABLE_APP)
# check_clamd is used by the clamd tests
add_executable(check_clamd)
target_sources(check_clamd
- PRIVATE check_clamd.c checks.h)
+ PRIVATE
+ check_clamd.c
+ checks.h
+ ../clamd/thrmgr.c)
target_link_libraries(check_clamd
PRIVATE
ClamAV::libclamav
diff --git a/unit_tests/check_clamd.c b/unit_tests/check_clamd.c
index de23b76e7..f25ecc98e 100644
--- a/unit_tests/check_clamd.c
+++ b/unit_tests/check_clamd.c
@@ -63,6 +63,13 @@
// common
#include "fdpassing.h"
+// clamd
+#include "clamd/thrmgr.h"
+
+/* Globals used by the thread manager's job-group shutdown checks. */
+pthread_mutex_t exit_mutex = PTHREAD_MUTEX_INITIALIZER;
+int progexit = 0;
+
static int conn_tcp(int port)
{
struct sockaddr_in server;
@@ -132,6 +139,45 @@ static void conn_teardown(void)
#endif
}
+static void close_socket(int socket_fd)
+{
+#ifndef _WIN32
+ close(socket_fd);
+#else
+ closesocket(socket_fd);
+#endif
+}
+
+static void create_tcp_socket_pair(int sockets[2])
+{
+ struct sockaddr_in address;
+ socklen_t address_len = sizeof(address);
+ int listener;
+
+ listener = socket(AF_INET, SOCK_STREAM, 0);
+ ck_assert_msg(listener != -1, "Unable to create listener socket: %s\n", strerror(errno));
+
+ memset(&address, 0, sizeof(address));
+ address.sin_family = AF_INET;
+ address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+ address.sin_port = 0;
+
+ ck_assert_msg(bind(listener, (struct sockaddr *)&address, sizeof(address)) != -1,
+ "Unable to bind listener socket: %s\n", strerror(errno));
+ ck_assert_msg(getsockname(listener, (struct sockaddr *)&address, &address_len) != -1,
+ "Unable to get listener socket address: %s\n", strerror(errno));
+ ck_assert_msg(listen(listener, 1) != -1, "Unable to listen on socket: %s\n", strerror(errno));
+
+ sockets[0] = socket(AF_INET, SOCK_STREAM, 0);
+ ck_assert_msg(sockets[0] != -1, "Unable to create client socket: %s\n", strerror(errno));
+ ck_assert_msg(connect(sockets[0], (struct sockaddr *)&address, address_len) != -1,
+ "Unable to connect client socket: %s\n", strerror(errno));
+
+ sockets[1] = accept(listener, NULL, NULL);
+ ck_assert_msg(sockets[1] != -1, "Unable to accept socket connection: %s\n", strerror(errno));
+ close_socket(listener);
+}
+
#ifndef REPO_VERSION
#define REPO_VERSION VERSION
#endif
@@ -912,10 +958,256 @@ START_TEST(test_idsession)
}
END_TEST
+struct stats_filename_test_state {
+ pthread_mutex_t mutex;
+ pthread_cond_t cond;
+ int stage;
+};
+
+static void stats_filename_test_handler(void *data)
+{
+ struct stats_filename_test_state *state = data;
+ char filename[64] = "task-filename-before-stack-reuse";
+
+ thrmgr_setactivetask(filename, "TEST");
+
+ pthread_mutex_lock(&state->mutex);
+ state->stage = 1;
+ pthread_cond_broadcast(&state->cond);
+ while (state->stage < 2)
+ pthread_cond_wait(&state->cond, &state->mutex);
+
+ strcpy(filename, "task-filename-after-stack-reuse");
+ state->stage = 3;
+ pthread_cond_broadcast(&state->cond);
+ while (state->stage < 4)
+ pthread_cond_wait(&state->cond, &state->mutex);
+ pthread_mutex_unlock(&state->mutex);
+
+ thrmgr_setactivetask(NULL, NULL);
+
+ pthread_mutex_lock(&state->mutex);
+ state->stage = 5;
+ pthread_cond_broadcast(&state->cond);
+ pthread_mutex_unlock(&state->mutex);
+}
+
+START_TEST(test_stats_owns_task_filename)
+{
+ struct stats_filename_test_state state;
+ threadpool_t *threadpool;
+ char *stats;
+ size_t stats_len;
+ int sockets[2];
+
+ 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, 1, stats_filename_test_handler);
+ ck_assert_ptr_nonnull(threadpool);
+ ck_assert_int_ne(thrmgr_dispatch(threadpool, &state), 0);
+
+ pthread_mutex_lock(&state.mutex);
+ while (state.stage < 1)
+ pthread_cond_wait(&state.cond, &state.mutex);
+ state.stage = 2;
+ pthread_cond_broadcast(&state.cond);
+ while (state.stage < 3)
+ pthread_cond_wait(&state.cond, &state.mutex);
+ pthread_mutex_unlock(&state.mutex);
+
+ 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_msg(strstr(stats, "task-filename-before-stack-reuse") != NULL,
+ "STATS did not retain its owned task filename:\n%s", stats);
+ ck_assert_msg(strstr(stats, "task-filename-after-stack-reuse") == NULL,
+ "STATS read the task filename from reused caller storage:\n%s", stats);
+ free(stats);
+
+ pthread_mutex_lock(&state.mutex);
+ state.stage = 4;
+ pthread_cond_broadcast(&state.cond);
+ while (state.stage < 5)
+ 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
+
+struct stats_slow_client_test_state {
+ pthread_mutex_t mutex;
+ pthread_cond_t cond;
+ const char *filename;
+ int stage;
+};
+
+struct stats_print_thread_state {
+ int socket_fd;
+ int result;
+};
+
+struct stats_pool_create_state {
+ pthread_mutex_t mutex;
+ pthread_cond_t cond;
+ threadpool_t *threadpool;
+ int done;
+};
+
+static void stats_slow_client_test_handler(void *data)
+{
+ struct stats_slow_client_test_state *state = data;
+
+ thrmgr_setactivetask(state->filename, "TEST");
+
+ pthread_mutex_lock(&state->mutex);
+ state->stage = 1;
+ pthread_cond_broadcast(&state->cond);
+ while (state->stage < 2)
+ pthread_cond_wait(&state->cond, &state->mutex);
+ pthread_mutex_unlock(&state->mutex);
+
+ thrmgr_setactivetask(NULL, NULL);
+
+ pthread_mutex_lock(&state->mutex);
+ state->stage = 3;
+ pthread_cond_broadcast(&state->cond);
+ pthread_mutex_unlock(&state->mutex);
+}
+
+static void stats_noop_handler(void *data)
+{
+ UNUSEDPARAM(data);
+}
+
+static void *stats_print_thread(void *data)
+{
+ struct stats_print_thread_state *state = data;
+
+ state->result = thrmgr_printstats(state->socket_fd, '\n');
+ close_socket(state->socket_fd);
+ return NULL;
+}
+
+static void *stats_pool_create_thread(void *data)
+{
+ struct stats_pool_create_state *state = data;
+
+ state->threadpool = thrmgr_new(1, 60, 1, stats_noop_handler);
+
+ pthread_mutex_lock(&state->mutex);
+ state->done = 1;
+ pthread_cond_broadcast(&state->cond);
+ pthread_mutex_unlock(&state->mutex);
+ return NULL;
+}
+
+START_TEST(test_stats_write_does_not_hold_pool_lock)
+{
+ struct stats_slow_client_test_state scan_state;
+ struct stats_print_thread_state print_state;
+ struct stats_pool_create_state pool_state;
+ threadpool_t *threadpool;
+ pthread_t print_thread;
+ pthread_t pool_thread;
+ struct timeval now;
+ struct timespec deadline;
+ char *filename;
+ char drain_buffer[8192];
+ int send_buffer_size = 4096;
+ int sockets[2];
+ int pool_created_before_drain;
+ int recv_result;
+
+ filename = malloc(1024 * 1024);
+ ck_assert_ptr_nonnull(filename);
+ memset(filename, 'A', (1024 * 1024) - 1);
+ filename[(1024 * 1024) - 1] = '\0';
+
+ memset(&scan_state, 0, sizeof(scan_state));
+ scan_state.filename = filename;
+ ck_assert_int_eq(pthread_mutex_init(&scan_state.mutex, NULL), 0);
+ ck_assert_int_eq(pthread_cond_init(&scan_state.cond, NULL), 0);
+
+ threadpool = thrmgr_new(1, 60, 1, stats_slow_client_test_handler);
+ ck_assert_ptr_nonnull(threadpool);
+ ck_assert_int_ne(thrmgr_dispatch(threadpool, &scan_state), 0);
+
+ pthread_mutex_lock(&scan_state.mutex);
+ while (scan_state.stage < 1)
+ pthread_cond_wait(&scan_state.cond, &scan_state.mutex);
+ pthread_mutex_unlock(&scan_state.mutex);
+
+ create_tcp_socket_pair(sockets);
+ ck_assert_int_eq(setsockopt(sockets[1], SOL_SOCKET, SO_SNDBUF,
+ (const char *)&send_buffer_size, sizeof(send_buffer_size)),
+ 0);
+
+ memset(&print_state, 0, sizeof(print_state));
+ print_state.socket_fd = sockets[1];
+ ck_assert_int_eq(pthread_create(&print_thread, NULL, stats_print_thread, &print_state), 0);
+
+ recv_result = recv(sockets[0], drain_buffer, 1, 0);
+ ck_assert_int_eq(recv_result, 1);
+
+ memset(&pool_state, 0, sizeof(pool_state));
+ ck_assert_int_eq(pthread_mutex_init(&pool_state.mutex, NULL), 0);
+ ck_assert_int_eq(pthread_cond_init(&pool_state.cond, NULL), 0);
+ ck_assert_int_eq(pthread_create(&pool_thread, NULL, stats_pool_create_thread, &pool_state), 0);
+
+ gettimeofday(&now, NULL);
+ deadline.tv_sec = now.tv_sec + 2;
+ deadline.tv_nsec = now.tv_usec * 1000;
+
+ pthread_mutex_lock(&pool_state.mutex);
+ while (!pool_state.done) {
+ if (ETIMEDOUT == pthread_cond_timedwait(&pool_state.cond, &pool_state.mutex, &deadline))
+ break;
+ }
+ pool_created_before_drain = pool_state.done;
+ pthread_mutex_unlock(&pool_state.mutex);
+
+ while ((recv_result = recv(sockets[0], drain_buffer, sizeof(drain_buffer), 0)) > 0) {
+ }
+ close_socket(sockets[0]);
+
+ ck_assert_int_eq(pthread_join(print_thread, NULL), 0);
+ ck_assert_int_eq(pthread_join(pool_thread, NULL), 0);
+ ck_assert_int_eq(print_state.result, 0);
+ ck_assert_ptr_nonnull(pool_state.threadpool);
+
+ thrmgr_destroy(pool_state.threadpool);
+ pthread_cond_destroy(&pool_state.cond);
+ pthread_mutex_destroy(&pool_state.mutex);
+
+ pthread_mutex_lock(&scan_state.mutex);
+ scan_state.stage = 2;
+ pthread_cond_broadcast(&scan_state.cond);
+ while (scan_state.stage < 3)
+ pthread_cond_wait(&scan_state.cond, &scan_state.mutex);
+ pthread_mutex_unlock(&scan_state.mutex);
+
+ thrmgr_destroy(threadpool);
+ pthread_cond_destroy(&scan_state.cond);
+ pthread_mutex_destroy(&scan_state.mutex);
+ free(filename);
+
+ ck_assert_msg(pool_created_before_drain,
+ "STATS held the global pool lock while writing to a slow client");
+}
+END_TEST
+
static Suite *test_clamd_suite(void)
{
Suite *s = suite_create("clamd");
- TCase *tc_commands, *tc_stress;
+ TCase *tc_commands, *tc_stress, *tc_thrmgr;
tc_commands = tcase_create("clamd commands");
suite_add_tcase(s, tc_commands);
tcase_add_unchecked_fixture(tc_commands, commands_setup, commands_teardown);
@@ -946,6 +1238,11 @@ static Suite *test_clamd_suite(void)
tcase_add_test(tc_stress, test_connections); // Disabled on Windows because test uses fork() instead of threads, and needs to be rewritten.
#endif
#endif
+ tc_thrmgr = tcase_create("thread manager");
+ 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);
+
return s;
}