diff --git a/AGENTS.md b/AGENTS.md index 042ab34..9fc41ec 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -206,7 +206,7 @@ tea pr close --repo TapTap/FastSync ## Common pitfalls -- **`__thread` on shared SSL context**: io_ssl must NOT be thread-local — worker threads inherit the SSL context from the main thread. Use regular `static SSL* io_ssl`. +- **Per-thread SSL context**: `io_ssl` is stored per-thread (`static __thread SSL* io_ssl`). Each thread that performs protocol I/O must call `io_set_ssl()` to install its own SSL object before using `send_*` / `receive_*` primitives. The main thread's SSL context is not automatically inherited by worker threads. - **SSL WANT_READ/WANT_WRITE retry**: Always retry on `SSL_ERROR_WANT_READ` and `SSL_ERROR_WANT_WRITE` in `send_n_data`/`receive_n_data`. Removing these breaks TLS multithreaded transfers. - **clang-format version**: The CI image uses clang-format 18. Always format inside the CI Docker container for exact match. - **Merge order matters**: Merge the most comprehensive branch first, then smaller ones, to minimize conflicts when creating a combined branch. diff --git a/src/client/client_cli.c b/src/client/client_cli.c index 5a15301..5ed58bb 100644 --- a/src/client/client_cli.c +++ b/src/client/client_cli.c @@ -276,13 +276,16 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar if (set_nonneg_int_option(&config->max_depth, argv[++i], "--max-depth") != 0) return -1; } else if (strcmp(argv[i], "--log-file") == 0 && i + 1 < argc) { + if (config->log_file) { + fclose(config->log_file); + config->log_file = NULL; + log_set_file(NULL); + } FILE* lf = fopen(argv[++i], "a"); if (!lf) { fprintf(stderr, "Error: could not open log file '%s': %s\n", argv[i], strerror(errno)); return -1; } - if (config->log_file) - fclose(config->log_file); config->log_file = lf; log_set_file(lf); } else if (strcmp(argv[i], "--queue-size") == 0 && i + 1 < argc) { @@ -427,6 +430,10 @@ static int parse_args(Config* config, int argc, char* argv[], int* positional_ar } else if (strcmp(argv[i], "--compress-level") == 0 && i + 1 < argc) { if (set_positive_int_option(&config->compression_level, argv[++i], "--compress-level") != 0) return -1; + if (config->compression_level < 1 || config->compression_level > 22) { + fprintf(stderr, "Error: --compress-level must be between 1 and 22\n"); + return -1; + } } else if (argv[i][0] == '-') { fprintf(stderr, "Unknown option: %s\n", argv[i]); print_usage(); @@ -600,8 +607,10 @@ static int read_patterns_from_file(const char* filepath, char*** patterns, int* fprintf(stderr, "Error: could not open pattern file '%s': %s\n", filepath, strerror(errno)); return -1; } - char line[4096]; - while (fgets(line, sizeof(line), fp)) { + char* line = NULL; + size_t line_size = 0; + ssize_t n; + while ((n = getline(&line, &line_size, fp)) != -1) { char* p = line; while (*p == ' ' || *p == '\t') p++; @@ -615,6 +624,7 @@ static int read_patterns_from_file(const char* filepath, char*** patterns, int* char** tmp = realloc(*patterns, (*count + 1) * sizeof(char*)); if (!tmp) { fprintf(stderr, "Error: memory allocation failed for pattern file\n"); + free(line); fclose(fp); return -1; } @@ -622,11 +632,13 @@ static int read_patterns_from_file(const char* filepath, char*** patterns, int* char* dup = str_dup(p); if (!dup) { fprintf(stderr, "Error: memory allocation failed for pattern file\n"); + free(line); fclose(fp); return -1; } (*patterns)[(*count)++] = dup; } + free(line); fclose(fp); return 0; } diff --git a/src/client/client_send.c b/src/client/client_send.c index 620cb57..16e0e66 100644 --- a/src/client/client_send.c +++ b/src/client/client_send.c @@ -346,6 +346,7 @@ static int send_chunks_multithreaded(void* pipeline_context) { } if (send_chunk(client, current_chunk, context->config) != 0) { fprintf(stderr, "Error: unexpected error while sending chunk\n"); + chunk_destroy(current_chunk); client_disconnect(client); client_delete(client); mtx_lock(&context->mutex_progress); @@ -385,13 +386,31 @@ static int scan_directory_multithreaded(void* pipeline_context) { const char* p = current_chunk->items[i]->path; if (*p == '/') p++; - array_list_add(context->manifest, str_dup(p)); + char* manifest_entry = str_dup(p); + if (!manifest_entry) { + log_message(LOG_LEVEL_ERROR, "Failed to allocate manifest entry"); + mtx_unlock(&context->mutex_scanner); + context->cancelled = true; + cnd_broadcast(&context->condition_not_full_scanner); + cnd_broadcast(&context->condition_not_empty_scanner); + parallel_scanner_destroy(scanner); + return thrd_error; + } + array_list_add(context->manifest, manifest_entry); } mtx_unlock(&context->mutex_scanner); } - queue_enqueue_multithreaded(context->queue_scanner, current_chunk, &context->mutex_scanner, - &context->condition_not_empty_scanner, - &context->condition_not_full_scanner); + if (!queue_enqueue_multithreaded_cancel( + context->queue_scanner, current_chunk, &context->mutex_scanner, + &context->condition_not_empty_scanner, &context->condition_not_full_scanner, + &context->cancelled)) { + chunk_destroy(current_chunk); + context->cancelled = true; + cnd_broadcast(&context->condition_not_full_scanner); + cnd_broadcast(&context->condition_not_empty_scanner); + parallel_scanner_destroy(scanner); + return thrd_error; + } } mtx_lock(&context->mutex_scanner); context->scanner_done = true; @@ -427,9 +446,16 @@ static int load_files_multithreaded(void* pipeline_context) { } } } - queue_enqueue_multithreaded(context->queue_loader, chunk, &context->mutex_loader, - &context->condition_not_empty_loader, - &context->condition_not_full_loader); + if (!queue_enqueue_multithreaded_cancel(context->queue_loader, chunk, &context->mutex_loader, + &context->condition_not_empty_loader, + &context->condition_not_full_loader, + &context->cancelled)) { + chunk_destroy(chunk); + context->cancelled = true; + cnd_broadcast(&context->condition_not_full_loader); + cnd_broadcast(&context->condition_not_empty_loader); + return thrd_error; + } } } @@ -487,16 +513,20 @@ int send_files(Config* config) { client = client_create(); if (!client || !client_connect_tls(client, config->server_host, config->server_port, config->tls_cert, config->tls_key, config->tls_ca)) { - if (client) + if (client) { + client_disconnect(client); client_delete(client); + } fprintf(stderr, "Error: could not connect to server via TLS\n"); return 1; } } else { client = client_create(); if (!client || !client_connect(client, config->server_host, config->server_port)) { - if (client) + if (client) { + client_disconnect(client); client_delete(client); + } fprintf(stderr, "Error: could not connect to server\n"); return 1; } @@ -526,7 +556,17 @@ int send_files(Config* config) { const char* p = current_chunk->items[i]->path; if (*p == '/') p++; - array_list_add(manifest, str_dup(p)); + char* manifest_entry = str_dup(p); + if (!manifest_entry) { + log_message(LOG_LEVEL_ERROR, "Failed to allocate manifest entry"); + chunk_destroy(current_chunk); + array_list_delete(manifest); + directory_scanner_destroy(scanner); + client_disconnect(client); + client_delete(client); + return 1; + } + array_list_add(manifest, manifest_entry); } } if (!config->use_sendfile) { @@ -625,17 +665,42 @@ int send_files_multithreaded(Config* config) { if (config->use_delete) context->manifest = array_list_create(free); - thrd_t scanner, loader, sender, progress; - if (thrd_create(&scanner, scan_directory_multithreaded, context) != thrd_success || - thrd_create(&loader, load_files_multithreaded, context) != thrd_success || - thrd_create(&sender, send_chunks_multithreaded, context) != thrd_success) { + thrd_t scanner, loader, sender; + bool scanner_created = false; + bool loader_created = false; + bool sender_created = false; + + scanner_created = (thrd_create(&scanner, scan_directory_multithreaded, context) == thrd_success); + if (scanner_created) + loader_created = (thrd_create(&loader, load_files_multithreaded, context) == thrd_success); + if (scanner_created && loader_created) + sender_created = (thrd_create(&sender, send_chunks_multithreaded, context) == thrd_success); + + if (!scanner_created || !loader_created || !sender_created) { perror("Error creating threads.\n"); + context->cancelled = true; + context->scanner_done = true; + context->loader_done = true; + context->sender_done = true; + cnd_broadcast(&context->condition_not_full_scanner); + cnd_broadcast(&context->condition_not_empty_scanner); + cnd_broadcast(&context->condition_not_full_loader); + cnd_broadcast(&context->condition_not_empty_loader); + if (sender_created) + thrd_join(sender, NULL); + if (loader_created) + thrd_join(loader, NULL); + if (scanner_created) + thrd_join(scanner, NULL); pipeline_context_sender_destroy(context); return 1; } + thrd_t progress; + bool progress_created = false; if (config->show_progress) { - if (thrd_create(&progress, progress_thread_fn, context) != thrd_success) { + progress_created = (thrd_create(&progress, progress_thread_fn, context) == thrd_success); + if (!progress_created) { perror("Error creating progress thread.\n"); /* Non-fatal; continue without progress reporting */ } @@ -646,7 +711,7 @@ int send_files_multithreaded(Config* config) { thrd_join(loader, NULL); thrd_join(sender, &sender_result); - if (config->show_progress) { + if (progress_created) { /* Signal progress thread to exit if it hasn't already */ mtx_lock(&context->mutex_progress); context->sender_done = true; diff --git a/src/client/scanner.c b/src/client/scanner.c index dfaa50e..848f66d 100644 --- a/src/client/scanner.c +++ b/src/client/scanner.c @@ -27,10 +27,14 @@ static void dir_entry_destroy(void* item) { static DirEntry* dir_entry_create(const char* path, int depth) { DirEntry* de = malloc(sizeof(DirEntry)); - if (de) { - de->path = str_dup(path); - de->depth = depth; + if (!de) + return NULL; + de->path = str_dup(path); + if (!de->path) { + free(de); + return NULL; } + de->depth = depth; return de; } @@ -62,7 +66,18 @@ DirectoryScanner* directory_scanner_create(const char* root_directory, bool use_ scanner->safe_links = safe_links; scanner->copy_unsafe_links = copy_unsafe_links; scanner->checksum = checksum; - queue_enqueue(scanner->directories, dir_entry_create(root_directory, 0)); + DirEntry* root = dir_entry_create(root_directory, 0); + if (!root) { + queue_destroy(scanner->directories); + free(scanner); + return NULL; + } + if (!queue_enqueue(scanner->directories, root)) { + dir_entry_destroy(root); + queue_destroy(scanner->directories); + free(scanner); + return NULL; + } return scanner; } @@ -323,9 +338,28 @@ ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata free(ps); return NULL; } - if (mtx_init(&ps->result_mutex, mtx_plain) != thrd_success || - cnd_init(&ps->result_not_empty) != thrd_success || - cnd_init(&ps->result_not_full) != thrd_success) { + int init = 0; + bool ok = true; + if (mtx_init(&ps->result_mutex, mtx_plain) != thrd_success) + ok = false; + if (ok) { + init++; + if (cnd_init(&ps->result_not_empty) != thrd_success) + ok = false; + } + if (ok) { + // cppcheck-suppress unreadVariable + init++; + if (cnd_init(&ps->result_not_full) != thrd_success) + ok = false; + } + if (!ok) { + if (init >= 3) + cnd_destroy(&ps->result_not_full); + if (init >= 2) + cnd_destroy(&ps->result_not_empty); + if (init >= 1) + mtx_destroy(&ps->result_mutex); queue_destroy(ps->result_queue); free(ps); return NULL; @@ -564,6 +598,7 @@ void parallel_scanner_destroy(ParallelScanner* ps) { return; ps->done = true; cnd_signal(&ps->result_not_empty); + cnd_broadcast(&ps->result_not_full); for (int i = 0; i < ps->num_threads; i++) thrd_join(ps->threads[i], NULL); free(ps->threads); diff --git a/src/server/server.c b/src/server/server.c index f8fdb54..1827d6c 100644 --- a/src/server/server.c +++ b/src/server/server.c @@ -135,11 +135,27 @@ void handler(int file_descriptor) { return; } thrd_t receiver, writer; - if (thrd_create(&receiver, receive_thread, context) != thrd_success || - thrd_create(&writer, write_thread, context) != thrd_success) { + bool receiver_created = false; + bool writer_created = false; + receiver_created = (thrd_create(&receiver, receive_thread, context) == thrd_success); + if (receiver_created) + writer_created = (thrd_create(&writer, write_thread, context) == thrd_success); + if (!receiver_created || !writer_created) { perror("Error creating Threads"); + if (receiver_created) { + mtx_lock(&context->mutex); + context->cancelled = true; + cnd_broadcast(&context->condition_not_full); + cnd_broadcast(&context->condition_not_empty); + mtx_unlock(&context->mutex); + close(file_descriptor); + thrd_join(receiver, NULL); + } else { + close(file_descriptor); + } + if (writer_created) + thrd_join(writer, NULL); pipeline_context_receiver_destroy(context); - close(file_descriptor); return; } thrd_join(receiver, NULL); diff --git a/src/shared/compression.c b/src/shared/compression.c index c1eb1f1..27b6da4 100644 --- a/src/shared/compression.c +++ b/src/shared/compression.c @@ -1,12 +1,13 @@ #include "compression.h" #include "data.h" #include "log.h" -#include "stdlib.h" -#include "string.h" +#include +#include #include -#include "zstd.h" +#include #define INITIAL_DECOMPRESS_BUF_SIZE (1024 * 1024) +#define MAX_DECOMPRESSED_SIZE (100ULL * 1024 * 1024) /* 100 MB hard ceiling */ static const char* SKIP_COMPRESSION_EXTENSIONS[] = {".jpg", ".jpeg", ".png", ".gif", ".mp4", ".mkv", ".zip", ".gz", ".xz", ".zst", NULL}; @@ -84,6 +85,13 @@ Data* data_decompress(Data* compressed_data) { dst_size = compressed_data->size * 3; if (dst_size < INITIAL_DECOMPRESS_BUF_SIZE) dst_size = INITIAL_DECOMPRESS_BUF_SIZE; + if (dst_size > MAX_DECOMPRESSED_SIZE) + dst_size = MAX_DECOMPRESSED_SIZE; + } + if (dst_size > MAX_DECOMPRESSED_SIZE) { + log_message(LOG_LEVEL_ERROR, "Declared decompressed size exceeds %llu bytes", + (unsigned long long)MAX_DECOMPRESSED_SIZE); + return NULL; } ZSTD_DCtx* dctx = ZSTD_createDCtx(); @@ -113,7 +121,16 @@ Data* data_decompress(Data* compressed_data) { return NULL; } if (ret > 0 && output.pos == output.size) { + if (buf_size >= MAX_DECOMPRESSED_SIZE) { + log_message(LOG_LEVEL_ERROR, "Decompressed data exceeds %llu bytes", + (unsigned long long)MAX_DECOMPRESSED_SIZE); + ZSTD_freeDCtx(dctx); + data_destroy(uncompressed_data); + return NULL; + } buf_size *= 2; + if (buf_size > MAX_DECOMPRESSED_SIZE) + buf_size = MAX_DECOMPRESSED_SIZE; void* new_data = realloc(uncompressed_data->data, buf_size); if (!new_data) { log_message(LOG_LEVEL_ERROR, "Failed to grow decompression buffer"); diff --git a/src/shared/config.c b/src/shared/config.c index 6d52cc4..c7900cc 100644 --- a/src/shared/config.c +++ b/src/shared/config.c @@ -131,6 +131,8 @@ void config_parse_ssh_dest(Config* config) { } void config_delete(Config* config) { + if (config == NULL) + return; free(config->version); free(config->send_directory); free(config->receive_root_directory); @@ -167,6 +169,16 @@ void config_delete(Config* config) { free(config); } +/* Wire format order (must match config_receive and be updated when PROTOCOL_VERSION bumps): + * version, send_directory, receive_root_directory, save_to_disk, use_multithreading, + * use_chunk_serialization, use_compression, use_metadata, compression_level, chunk_size, + * use_sendfile, use_delete, use_incremental, use_delta, delta_block_size, delta_max_file_size, + * backup, backup_dir, follow_symlinks, copy_links, safe_links, copy_unsafe_links, + * preserve_hard_links, preserve_acls, preserve_xattrs, preserve_devices, preserve_sparse, + * update, inplace, append, append_verify, delete_excluded, delete_after, max_delete, relative, + * prune_empty_dirs, temp_dir, partial, partial_dir, suffix, delete_before, checksum, + * compress_choice, status + */ bool config_send(int file_descriptor, const Config* config) { if (!send_str(file_descriptor, config->version)) return false; @@ -264,6 +276,7 @@ bool config_send(int file_descriptor, const Config* config) { return true; } +/* Wire format order: see the comment above config_send. */ Config* config_receive(int file_descriptor) { Config* config = (Config*)malloc(sizeof(Config)); if (config == NULL) diff --git a/src/shared/config.h b/src/shared/config.h index 9c78505..22359e4 100644 --- a/src/shared/config.h +++ b/src/shared/config.h @@ -128,7 +128,8 @@ typedef struct Config { char* compress_choice; } Config; -#define PROTOCOL_VERSION "1.3.0" +/* This version must be bumped whenever config_send / config_receive wire format changes. */ +#define PROTOCOL_VERSION "2.1.0" #define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024) Config* config_create(void); diff --git a/src/shared/file.c b/src/shared/file.c index 2b730f8..488321c 100644 --- a/src/shared/file.c +++ b/src/shared/file.c @@ -16,7 +16,6 @@ #include "config.h" #include "data.h" #include "file.h" -#include "log.h" #include "metadata.h" #include "protocol.h" #include "utils.h" @@ -96,6 +95,9 @@ bool file_load_data(File* file) { size_t bytes_read = file_content_to_buffer(file); if (bytes_read != file->data->size) { log_message(LOG_LEVEL_ERROR, "Did not read expected amount of bytes from file"); + free(file->data->data); + file->data->data = NULL; + file->data->size = 0; return false; } return true; diff --git a/src/shared/log.c b/src/shared/log.c index 07a5d27..fbd9400 100644 --- a/src/shared/log.c +++ b/src/shared/log.c @@ -18,11 +18,15 @@ void log_set_file(FILE* fp) { void log_message(LogLevel log_level, const char* format, ...) { if (log_level < current_log_level) return; + if (log_level < 0 || log_level >= (int)(sizeof(log_level_strings) / sizeof(log_level_strings[0]))) + return; time_t now = time(NULL); - const struct tm* t = localtime(&now); + struct tm t; + if (!localtime_r(&now, &t)) + return; - fprintf(stderr, "%04d-%02d-%02d %02d:%02d:%02d [%s]: ", t->tm_year + 1900, t->tm_mon + 1, - t->tm_mday, t->tm_hour, t->tm_min, t->tm_sec, log_level_strings[log_level]); + fprintf(stderr, "%04d-%02d-%02d %02d:%02d:%02d [%s]: ", t.tm_year + 1900, t.tm_mon + 1, t.tm_mday, + t.tm_hour, t.tm_min, t.tm_sec, log_level_strings[log_level]); va_list args; va_start(args, format); @@ -31,8 +35,8 @@ void log_message(LogLevel log_level, const char* format, ...) { fprintf(stderr, "\n"); if (log_fp) { - fprintf(log_fp, "%04d-%02d-%02d %02d:%02d:%02d [%s]: ", t->tm_year + 1900, t->tm_mon + 1, - t->tm_mday, t->tm_hour, t->tm_min, t->tm_sec, log_level_strings[log_level]); + fprintf(log_fp, "%04d-%02d-%02d %02d:%02d:%02d [%s]: ", t.tm_year + 1900, t.tm_mon + 1, + t.tm_mday, t.tm_hour, t.tm_min, t.tm_sec, log_level_strings[log_level]); va_start(args, format); vfprintf(log_fp, format, args); va_end(args); diff --git a/src/shared/multiprocessing.c b/src/shared/multiprocessing.c index df03388..24a4a5c 100644 --- a/src/shared/multiprocessing.c +++ b/src/shared/multiprocessing.c @@ -26,18 +26,48 @@ PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* que context->manifest = NULL; context->progress_bytes = 0; context->sender_done = false; - if (mtx_init(&context->mutex_scanner, mtx_plain) != thrd_success || - cnd_init(&context->condition_not_full_scanner) != thrd_success || - cnd_init(&context->condition_not_empty_scanner) != thrd_success || - mtx_init(&context->mutex_loader, mtx_plain) != thrd_success || - cnd_init(&context->condition_not_full_loader) != thrd_success || - mtx_init(&context->mutex_progress, mtx_plain) != thrd_success || - cnd_init(&context->condition_not_empty_loader) != thrd_success) { - perror("Error initializing synchronization objects"); - free(context); - return NULL; - } + context->cancelled = false; + int init = 0; + if (mtx_init(&context->mutex_scanner, mtx_plain) != thrd_success) + goto fail; + init++; + if (cnd_init(&context->condition_not_full_scanner) != thrd_success) + goto fail; + init++; + if (cnd_init(&context->condition_not_empty_scanner) != thrd_success) + goto fail; + init++; + if (mtx_init(&context->mutex_loader, mtx_plain) != thrd_success) + goto fail; + init++; + if (cnd_init(&context->condition_not_full_loader) != thrd_success) + goto fail; + init++; + if (cnd_init(&context->condition_not_empty_loader) != thrd_success) + goto fail; + init++; + if (mtx_init(&context->mutex_progress, mtx_plain) != thrd_success) + goto fail; + // cppcheck-suppress unreadVariable + init++; return context; + +fail: + perror("Error initializing synchronization objects"); + if (init >= 6) + cnd_destroy(&context->condition_not_empty_loader); + if (init >= 5) + cnd_destroy(&context->condition_not_full_loader); + if (init >= 4) + mtx_destroy(&context->mutex_loader); + if (init >= 3) + cnd_destroy(&context->condition_not_empty_scanner); + if (init >= 2) + cnd_destroy(&context->condition_not_full_scanner); + if (init >= 1) + mtx_destroy(&context->mutex_scanner); + free(context); + return NULL; } void pipeline_context_sender_destroy(PipelineContextSender* context) { @@ -67,14 +97,30 @@ PipelineContextReceiver* pipeline_context_receiver_create(Config* config, Queue* context->file_descriptor = file_descriptor; context->ssl = ssl; context->receiver_done = false; - if (mtx_init(&context->mutex, mtx_plain) != thrd_success || - cnd_init(&context->condition_not_full) != thrd_success || - cnd_init(&context->condition_not_empty) != thrd_success) { - perror("Error initializing synchronization objects"); - free(context); - return NULL; - } + context->cancelled = false; + int init = 0; + if (mtx_init(&context->mutex, mtx_plain) != thrd_success) + goto fail; + init++; + if (cnd_init(&context->condition_not_full) != thrd_success) + goto fail; + init++; + if (cnd_init(&context->condition_not_empty) != thrd_success) + goto fail; + // cppcheck-suppress unreadVariable + init++; return context; + +fail: + perror("Error initializing synchronization objects"); + if (init >= 3) + cnd_destroy(&context->condition_not_empty); + if (init >= 2) + cnd_destroy(&context->condition_not_full); + if (init >= 1) + mtx_destroy(&context->mutex); + free(context); + return NULL; } void pipeline_context_receiver_destroy(PipelineContextReceiver* context) { @@ -94,8 +140,13 @@ static bool receive_chunk_enqueue(int file_descriptor, PipelineContextReceiver* for (int i = 0; i < chunk->element_count; i++) { File* file = chunk->items[i]; chunk->items[i] = NULL; - queue_enqueue_multithreaded(context->queue, file, &context->mutex, - &context->condition_not_empty, &context->condition_not_full); + if (!queue_enqueue_multithreaded_cancel(context->queue, file, &context->mutex, + &context->condition_not_empty, + &context->condition_not_full, &context->cancelled)) { + file_destroy(file); + chunk_destroy(chunk); + return false; + } } chunk_destroy(chunk); return true; @@ -129,8 +180,12 @@ int receive_thread(void* pipeline_context) { if (!skipped) { if (file == NULL) return thrd_error; - queue_enqueue_multithreaded(context->queue, file, &context->mutex, - &context->condition_not_empty, &context->condition_not_full); + if (!queue_enqueue_multithreaded_cancel( + context->queue, file, &context->mutex, &context->condition_not_empty, + &context->condition_not_full, &context->cancelled)) { + file_destroy(file); + return thrd_error; + } } } else if (status == STATUS_CHUNK) { if (!receive_chunk_enqueue(file_descriptor, context)) @@ -166,8 +221,12 @@ int receive_thread(void* pipeline_context) { } else { File* file = file_receive(config, file_descriptor); if (file) { - queue_enqueue_multithreaded(context->queue, file, &context->mutex, - &context->condition_not_empty, &context->condition_not_full); + if (!queue_enqueue_multithreaded_cancel( + context->queue, file, &context->mutex, &context->condition_not_empty, + &context->condition_not_full, &context->cancelled)) { + file_destroy(file); + return thrd_error; + } } else { log_message(LOG_LEVEL_ERROR, "Failed to receive file"); return thrd_error; diff --git a/src/shared/multiprocessing.h b/src/shared/multiprocessing.h index aa53a13..443720a 100644 --- a/src/shared/multiprocessing.h +++ b/src/shared/multiprocessing.h @@ -26,6 +26,7 @@ typedef struct { mtx_t mutex_progress; unsigned long long progress_bytes; bool sender_done; + bool cancelled; } PipelineContextSender; typedef struct PipelineContextReceiver { @@ -37,6 +38,7 @@ typedef struct PipelineContextReceiver { cnd_t condition_not_full; cnd_t condition_not_empty; bool receiver_done; + bool cancelled; } PipelineContextReceiver; PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* queue_scanner, diff --git a/src/shared/protocol.c b/src/shared/protocol.c index 0b32f96..0d41808 100644 --- a/src/shared/protocol.c +++ b/src/shared/protocol.c @@ -6,6 +6,7 @@ #include #include #include +#include #include #include @@ -20,6 +21,8 @@ static __thread SSL* io_ssl; static unsigned long long io_bwlimit = 0; static long long bw_tokens = 0; static struct timespec bw_last_refill = {0, 0}; +static mtx_t bw_mutex; +static once_flag bw_mutex_once = ONCE_FLAG_INIT; static __thread unsigned long long total_allocated_bytes = 0; @@ -28,15 +31,24 @@ void io_set_fds(int read_fd, int write_fd) { io_write_fd = write_fd; } +static void bw_mutex_init(void) { + mtx_init(&bw_mutex, mtx_plain); +} + void io_set_bwlimit(unsigned long long bytes_per_sec) { + call_once(&bw_mutex_once, bw_mutex_init); + mtx_lock(&bw_mutex); io_bwlimit = bytes_per_sec; bw_tokens = (long long)io_bwlimit; clock_gettime(CLOCK_MONOTONIC, &bw_last_refill); + mtx_unlock(&bw_mutex); } static void bw_throttle(size_t bytes_written) { if (io_bwlimit == 0) return; + call_once(&bw_mutex_once, bw_mutex_init); + mtx_lock(&bw_mutex); struct timespec now; clock_gettime(CLOCK_MONOTONIC, &now); @@ -61,6 +73,7 @@ static void bw_throttle(size_t bytes_written) { bw_tokens = 0; clock_gettime(CLOCK_MONOTONIC, &bw_last_refill); } + mtx_unlock(&bw_mutex); } void io_set_ssl(SSL* ssl) { @@ -177,6 +190,8 @@ static const char* status_to_string(Status status) { } bool send_str(int file_descriptor, const char* data) { + if (data == NULL) + return false; size_t size = strlen(data); if (!send_n_data(file_descriptor, &size, sizeof(size_t))) return false; diff --git a/src/shared/queue.c b/src/shared/queue.c index 14d901d..1213746 100644 --- a/src/shared/queue.c +++ b/src/shared/queue.c @@ -103,6 +103,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/src/shared/queue.h b/src/shared/queue.h index 8bd6ba4..f482d86 100644 --- a/src/shared/queue.h +++ b/src/shared/queue.h @@ -20,6 +20,9 @@ bool queue_is_full(const Queue* queue); bool queue_enqueue(Queue* queue, void* item); bool queue_enqueue_multithreaded(Queue* queue, void* item, mtx_t* mutex, cnd_t* condition_not_empty, cnd_t* condition_not_full); +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); void* queue_dequeue(Queue* queue); void* queue_dequeue_multithreaded(Queue* queue, mtx_t* mutex, cnd_t* condition_not_empty, cnd_t* condition_not_full, const bool* other_thread_done); diff --git a/src/shared/utils.c b/src/shared/utils.c index dc7f0f9..43ed1e2 100644 --- a/src/shared/utils.c +++ b/src/shared/utils.c @@ -10,11 +10,13 @@ #include bool mkdir_r(const char* path) { - char* path_duplicate = malloc(strlen(path) + 1); + size_t path_len = strlen(path); + char* path_duplicate = malloc(path_len + 1); if (!path_duplicate) return false; - memcpy(path_duplicate, path, strlen(path) + 1); - char* path_current = (char*)malloc((strlen(path) + 2) * sizeof(char)); + memcpy(path_duplicate, path, path_len + 1); + size_t capacity = path_len + 2; + char* path_current = (char*)malloc(capacity * sizeof(char)); if (!path_current) { free(path_duplicate); return false; @@ -33,6 +35,10 @@ bool mkdir_r(const char* path) { bool ok = true; while (part != NULL) { size_t part_len = strlen(part); + if ((size_t)(path_current_position - path_current) + part_len + 2 > capacity) { + ok = false; + break; + } memcpy(path_current_position, part, part_len); path_current_position += part_len; path_current_position[0] = '/'; @@ -64,10 +70,18 @@ char* str_dup(const char* string) { return new_string; } +/* Match a glob pattern against a string. Supported wildcards: + * ? matches any single character except '/'. + * * matches any sequence of characters within one path component (no '/'). + * ** matches any sequence of characters, including '/' (cross-directory). + * slash-star-star-slash is treated as a cross-directory wildcard when it appears between + * literals. + */ bool glob_match(const char* pattern, const char* str) { while (*pattern) { if (*pattern == '*') { if (*(pattern + 1) == '*') { + /* globstar: match across directories */ pattern += 2; if (*pattern == '\0') return true; @@ -80,6 +94,7 @@ bool glob_match(const char* pattern, const char* str) { } return glob_match(pattern, str); } + /* single *: match within one path component */ pattern++; while (*str && *str != '/') { if (glob_match(pattern, str)) @@ -94,6 +109,7 @@ bool glob_match(const char* pattern, const char* str) { str++; } else { if (*pattern != *str) { + /* allow literal / ** / rest to match any number of directories */ if (*pattern == '/' && *(pattern + 1) == '*' && *(pattern + 2) == '*') { const char* rest = pattern + 3; if (*rest == '/')