From e8e9436879ceedadd68779d383322d85069de4f7 Mon Sep 17 00:00:00 2001 From: TapTap Date: Sat, 8 Aug 2026 20:25:58 +0200 Subject: [PATCH] fix: harden queue and receiver error handling --- src/shared/multiprocessing.c | 45 +++++++++++++++++++++++------------ src/shared/queue.c | 24 ++++++++++++++++++- tests/test_multiprocessing.c | 46 +++++++++++++++++++++++++++++++----- tests/test_queue.c | 6 +++++ 4 files changed, 99 insertions(+), 22 deletions(-) diff --git a/src/shared/multiprocessing.c b/src/shared/multiprocessing.c index df03388..f5019a9 100644 --- a/src/shared/multiprocessing.c +++ b/src/shared/multiprocessing.c @@ -101,7 +101,20 @@ static bool receive_chunk_enqueue(int file_descriptor, PipelineContextReceiver* return true; } +static void receiver_thread_fail(PipelineContextReceiver* context) { + mtx_lock(&context->mutex); + context->receiver_done = true; + cnd_broadcast(&context->condition_not_empty); + cnd_broadcast(&context->condition_not_full); + mtx_unlock(&context->mutex); +} + int receive_thread(void* pipeline_context) { +#define RECEIVE_THREAD_FAIL() \ + do { \ + receiver_thread_fail(context); \ + return thrd_error; \ + } while (0) PipelineContextReceiver* context = (PipelineContextReceiver*)pipeline_context; if (context->ssl) io_set_ssl(context->ssl); @@ -112,53 +125,52 @@ int receive_thread(void* pipeline_context) { Status status; if (!receive_status(file_descriptor, &status)) - return thrd_error; + RECEIVE_THREAD_FAIL(); while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK || status == STATUS_KEEPALIVE || status == STATUS_ABORT || status == STATUS_CHECK_BATCH) { if (status == STATUS_KEEPALIVE) { - send_status(file_descriptor, STATUS_KEEPALIVE); + if (!send_status(file_descriptor, STATUS_KEEPALIVE)) + RECEIVE_THREAD_FAIL(); goto next; } if (status == STATUS_ABORT) { log_message(LOG_LEVEL_INFO, "Received abort from client, cleaning up"); - return thrd_error; + RECEIVE_THREAD_FAIL(); } if (status == STATUS_CHECK) { bool skipped; File* file = receive_incremental_check(file_descriptor, config, &skipped); if (!skipped) { if (file == NULL) - return thrd_error; + RECEIVE_THREAD_FAIL(); queue_enqueue_multithreaded(context->queue, file, &context->mutex, &context->condition_not_empty, &context->condition_not_full); } } else if (status == STATUS_CHUNK) { if (!receive_chunk_enqueue(file_descriptor, context)) - return thrd_error; + RECEIVE_THREAD_FAIL(); } else if (status == STATUS_CHECK_BATCH) { int count; if (!receive_int(file_descriptor, &count)) - return thrd_error; + RECEIVE_THREAD_FAIL(); for (int i = 0; i < count; i++) { char* check_path = receive_str(file_descriptor); if (!check_path) - return thrd_error; + RECEIVE_THREAD_FAIL(); unsigned long long check_size; long long check_mtime; if (!receive_n_data(file_descriptor, &check_size, sizeof(check_size)) || !receive_n_data(file_descriptor, &check_mtime, sizeof(check_mtime))) { free(check_path); - return thrd_error; + RECEIVE_THREAD_FAIL(); } char* full_path = path_cat(config->receive_root_directory, check_path); struct stat st; bool has_old = full_path && lstat(full_path, &st) == 0; bool match = has_old && (unsigned long long)st.st_size == check_size && (long long)st.st_mtime == check_mtime; - if (match) - send_status(file_descriptor, STATUS_OK); - else - send_status(file_descriptor, STATUS_NEXT); + if (!send_status(file_descriptor, match ? STATUS_OK : STATUS_NEXT)) + RECEIVE_THREAD_FAIL(); free(full_path); free(check_path); } @@ -170,21 +182,24 @@ int receive_thread(void* pipeline_context) { &context->condition_not_empty, &context->condition_not_full); } else { log_message(LOG_LEVEL_ERROR, "Failed to receive file"); - return thrd_error; + RECEIVE_THREAD_FAIL(); } } next: if (!receive_status(file_descriptor, &status)) - return thrd_error; + RECEIVE_THREAD_FAIL(); } if (status == STATUS_MANIFEST) { if (receive_manifest(file_descriptor, config, &status) != 0) - return thrd_error; + RECEIVE_THREAD_FAIL(); } + if (status != STATUS_FINISHED) + RECEIVE_THREAD_FAIL(); mtx_lock(&context->mutex); context->receiver_done = true; cnd_signal(&context->condition_not_empty); mtx_unlock(&context->mutex); +#undef RECEIVE_THREAD_FAIL return thrd_success; } diff --git a/src/shared/queue.c b/src/shared/queue.c index 14d901d..5b763f2 100644 --- a/src/shared/queue.c +++ b/src/shared/queue.c @@ -1,4 +1,5 @@ #include +#include #include #include #include @@ -7,6 +8,9 @@ #include "queue.h" Queue* queue_create(int capacity, void (*destroyer)(void* item)) { + if (capacity <= 0) + return NULL; + Queue* queue = (Queue*)malloc(sizeof(Queue)); if (queue == NULL) { perror("ERROR: Could not allocate memory for queue structure"); @@ -61,7 +65,9 @@ bool queue_is_full(const Queue* queue) { static bool queue_double_capacity(Queue* queue) { if (queue == NULL) return false; - unsigned int new_capacity = queue->capacity * 2; + if (queue->capacity > INT_MAX / 2) + return false; + int new_capacity = queue->capacity * 2; if (new_capacity <= 1) new_capacity = 100; void** new_items = malloc(new_capacity * sizeof(void*)); @@ -103,6 +109,22 @@ bool queue_enqueue_multithreaded(Queue* queue, void* item, mtx_t* mutex, cnd_t* return ok; } +bool queue_enqueue_multithreaded_cancel(Queue* queue, void* item, mtx_t* mutex, + cnd_t* condition_not_empty, cnd_t* condition_not_full, + const bool* cancelled) { + mtx_lock(mutex); + while (queue_is_full(queue) && (cancelled == NULL || !*cancelled)) + cnd_wait(condition_not_full, mutex); + if (cancelled != NULL && *cancelled) { + mtx_unlock(mutex); + return false; + } + bool ok = queue_enqueue(queue, item); + cnd_signal(condition_not_empty); + mtx_unlock(mutex); + return ok; +} + void* queue_dequeue(Queue* queue) { if (queue == NULL || queue_is_empty(queue)) { perror("ERROR: Could not dequeue from null or empty queue."); diff --git a/tests/test_multiprocessing.c b/tests/test_multiprocessing.c index 54ed5d6..b023d90 100644 --- a/tests/test_multiprocessing.c +++ b/tests/test_multiprocessing.c @@ -82,7 +82,7 @@ static void test_sender_queue_capacities() { pipeline_context_sender_destroy(ctx); } -/* Test that create handles zero-capacity queues */ +/* Invalid queue capacities must not create unusable pipeline queues. */ static void test_sender_zero_capacity() { Config* cfg = config_create(); EXPECT_NOT_NULL(cfg); @@ -93,11 +93,9 @@ static void test_sender_zero_capacity() { Queue* q1 = queue_create(0, NULL); Queue* q2 = queue_create(0, NULL); - PipelineContextSender* ctx = pipeline_context_sender_create(cfg, q1, q2); - EXPECT_NOT_NULL(ctx); - EXPECT_EQ_INT(ctx->queue_scanner->capacity, 0); - EXPECT_EQ_INT(ctx->queue_loader->capacity, 0); - pipeline_context_sender_destroy(ctx); + EXPECT_NULL(q1); + EXPECT_NULL(q2); + config_delete(cfg); } /* Test receiver with zero file_descriptor */ @@ -168,6 +166,41 @@ static void test_receive_thread_finished() { } } +/* A malformed terminal status must wake a writer waiting on an empty queue. */ +static void test_receive_thread_failure_wakes_writer() { + Config* cfg = config_create(); + EXPECT_NOT_NULL(cfg); + free(cfg->version); + cfg->version = str_dup(PROTOCOL_VERSION); + cfg->send_directory = str_dup("/src"); + cfg->receive_root_directory = str_dup("/tmp/dst"); + + int p[2]; + EXPECT_EQ_INT(pipe(p), 0); + Queue* q = queue_create(1, file_destroy); + EXPECT_NOT_NULL(q); + PipelineContextReceiver* ctx = pipeline_context_receiver_create(cfg, q, p[0], NULL); + EXPECT_NOT_NULL(ctx); + + thrd_t receiver; + thrd_t writer; + EXPECT_EQ_INT(thrd_create(&writer, write_thread, ctx), thrd_success); + EXPECT_EQ_INT(thrd_create(&receiver, receive_thread, ctx), thrd_success); + EXPECT_TRUE(send_status(p[1], STATUS_OK)); + close(p[1]); + + int receiver_result; + int writer_result; + EXPECT_EQ_INT(thrd_join(receiver, &receiver_result), thrd_success); + EXPECT_EQ_INT(thrd_join(writer, &writer_result), thrd_success); + EXPECT_EQ_INT(receiver_result, thrd_error); + EXPECT_EQ_INT(writer_result, thrd_success); + EXPECT_TRUE(ctx->receiver_done); + + close(p[0]); + pipeline_context_receiver_destroy(ctx); +} + /* Test that write_thread completes cleanly when queue signals done */ static void test_write_thread_done() { Config* cfg = config_create(); @@ -223,6 +256,7 @@ void test_multiprocessing() { test_receiver_fd_zero(); if (!is_running_under_valgrind()) { test_receive_thread_finished(); + test_receive_thread_failure_wakes_writer(); } test_write_thread_done(); } diff --git a/tests/test_queue.c b/tests/test_queue.c index c392fcb..3ab6d3c 100644 --- a/tests/test_queue.c +++ b/tests/test_queue.c @@ -56,6 +56,11 @@ static void test_queue_basic() { queue_destroy(q); } +static void test_queue_rejects_invalid_capacity() { + EXPECT_NULL(queue_create(0, NULL)); + EXPECT_NULL(queue_create(-1, NULL)); +} + static void test_queue_resize() { Queue* q = queue_create(3, NULL); EXPECT_NOT_NULL(q); @@ -199,6 +204,7 @@ static void test_queue_multithreaded() { void test_queue() { test_queue_basic(); + test_queue_rejects_invalid_capacity(); test_queue_resize(); test_queue_destroyer(); test_queue_multithreaded();