From 4bba1e8c02f7d37101cf31a1fd1d9eb5c7899903 Mon Sep 17 00:00:00 2001 From: TapTap Date: Tue, 21 Jul 2026 16:59:46 +0200 Subject: [PATCH] perf: performance improvements (#107-#111) --- src/client/client_send.c | 50 +++++--- src/client/scanner.c | 252 ++++++++++++++++++++++++++++++++++++++- src/client/scanner.h | 21 ++++ src/shared/compression.c | 18 +++ src/shared/compression.h | 2 + src/shared/file.c | 2 +- src/shared/protocol.c | 14 +-- src/shared/protocol.h | 5 +- 8 files changed, 340 insertions(+), 24 deletions(-) diff --git a/src/client/client_send.c b/src/client/client_send.c index 256d8e4..a610c0d 100644 --- a/src/client/client_send.c +++ b/src/client/client_send.c @@ -21,6 +21,9 @@ #include #include #include +#include + +#define STREAM_THRESHOLD (64ULL * 1024 * 1024) static int incremental_check(Client* client, File* file, DeltaSignature** out_sig) { *out_sig = NULL; @@ -208,10 +211,13 @@ int send_chunk(Client* client, Chunk* chunk, Config* config) { return 0; } - bool use_sendfile = config->use_sendfile && !config->use_compression; for (int i = 0; i < chunk->element_count; i++) { - int rc = - send_single_file(client, chunk->items[i], config, config->use_incremental, use_sendfile); + File* f = chunk->items[i]; + if (f == NULL) + continue; + bool stream = f->data->data == NULL && f->data->size > 0; + bool use_sendfile = (config->use_sendfile && !config->use_compression) || stream; + int rc = send_single_file(client, f, config, config->use_incremental, use_sendfile); if (rc == 1) continue; if (rc < 0) @@ -295,16 +301,14 @@ static int send_chunks_multithreaded(void* pipeline_context) { static int scan_directory_multithreaded(void* pipeline_context) { PipelineContextSender* context = (PipelineContextSender*)pipeline_context; - mtx_lock(&context->mutex_scanner); - DirectoryScanner* scanner = directory_scanner_create( + ParallelScanner* scanner = parallel_scanner_create( context->config->send_directory, context->config->use_metadata, context->config->chunk_size, context->config->exclude_patterns, context->config->exclude_count, context->config->include_patterns, context->config->include_count, context->config->max_size, - context->config->min_size); - mtx_unlock(&context->mutex_scanner); + context->config->min_size, 4); Chunk* current_chunk; - while ((current_chunk = directory_scanner_next(scanner)) != NULL) { + while ((current_chunk = parallel_scanner_next(scanner)) != NULL) { if (context->config->use_delete) { mtx_lock(&context->mutex_scanner); for (int i = 0; i < current_chunk->element_count; i++) { @@ -324,7 +328,7 @@ static int scan_directory_multithreaded(void* pipeline_context) { cnd_signal(&context->condition_not_empty_scanner); mtx_unlock(&context->mutex_scanner); - directory_scanner_destroy(scanner); + parallel_scanner_destroy(scanner); return thrd_success; } @@ -343,9 +347,12 @@ static int load_files_multithreaded(void* pipeline_context) { } if (!context->config->use_sendfile) { for (int i = 0; i < chunk->element_count; i++) { - if (!file_load_data(chunk->items[i])) { + File* f = chunk->items[i]; + if (f->data->size > STREAM_THRESHOLD) + continue; + if (!file_load_data(f)) { log_message(LOG_LEVEL_ERROR, "Failed to load file data, skipping"); - file_destroy(chunk->items[i]); + file_destroy(f); chunk->items[i] = NULL; } } @@ -433,7 +440,10 @@ int send_files(Config* config) { } if (!config->use_sendfile) { for (int i = 0; i < current_chunk->element_count; i++) { - if (!file_load_data(current_chunk->items[i])) { + File* f = current_chunk->items[i]; + if (f->data->size > STREAM_THRESHOLD) + continue; + if (!file_load_data(f)) { log_message(LOG_LEVEL_ERROR, "Failed to load file data"); continue; } @@ -518,8 +528,20 @@ int send_files_multithreaded(Config* config) { return 0; } - Queue* q1 = queue_create(100, chunk_destroy); - Queue* q2 = queue_create(100, chunk_destroy); + long pages = sysconf(_SC_AVPHYS_PAGES); + long page_size = sysconf(_SC_PAGE_SIZE); + unsigned long long available_memory = + pages > 0 && page_size > 0 ? (unsigned long long)pages * (unsigned long long)page_size + : 512ULL * 1024 * 1024; + unsigned long long avg_file_size = 1024 * 1024; + int qsize = (int)(available_memory / avg_file_size); + if (qsize < 10) + qsize = 10; + if (qsize > 1000) + qsize = 1000; + + Queue* q1 = queue_create(qsize, chunk_destroy); + Queue* q2 = queue_create(qsize, chunk_destroy); if (!q1 || !q2) { if (q1) queue_destroy(q1); diff --git a/src/client/scanner.c b/src/client/scanner.c index ad56fb2..556c34b 100644 --- a/src/client/scanner.c +++ b/src/client/scanner.c @@ -9,6 +9,7 @@ #include #include #include +#include #include DirectoryScanner* directory_scanner_create(char* root_directory, bool use_metadata, @@ -55,7 +56,6 @@ static Chunk* chunk_data_to_chunk(ArrayList* chunk_data) { return chunk; } -// Returns: 1 on success, 0 if no more directories in queue, -1 on opendir failure static int open_next_directory(DirectoryScanner* scanner) { if (scanner->current_dir) { closedir(scanner->current_dir); @@ -167,3 +167,253 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) { array_list_delete(chunk_data); return NULL; } + +typedef struct { + ParallelScanner* ps; + char** dirs; + int dir_count; + bool use_metadata; + unsigned long long chunk_size; + char** exclude_patterns; + int exclude_count; + char** include_patterns; + int include_count; + unsigned long long max_size; + unsigned long long min_size; +} ParallelWorkerArg; + +static int parallel_worker_thread(void* arg) { + ParallelWorkerArg* wa = (ParallelWorkerArg*)arg; + for (int i = 0; i < wa->dir_count; i++) { + DirectoryScanner* ds = directory_scanner_create( + wa->dirs[i], wa->use_metadata, wa->chunk_size, wa->exclude_patterns, wa->exclude_count, + wa->include_patterns, wa->include_count, wa->max_size, wa->min_size); + Chunk* chunk; + while ((chunk = directory_scanner_next(ds)) != NULL) { + queue_enqueue_multithreaded(wa->ps->result_queue, chunk, &wa->ps->result_mutex, + &wa->ps->result_not_empty, &wa->ps->result_not_full); + } + directory_scanner_destroy(ds); + free(wa->dirs[i]); + } + ParallelScanner* ps = wa->ps; + free(wa->dirs); + free(wa); + mtx_lock(&ps->result_mutex); + ps->completed++; + if (ps->completed >= ps->num_threads) { + ps->done = true; + cnd_signal(&ps->result_not_empty); + } + mtx_unlock(&ps->result_mutex); + return thrd_success; +} + +ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata, + unsigned long long chunk_size, char** exclude_patterns, + int exclude_count, char** include_patterns, + int include_count, unsigned long long max_size, + unsigned long long min_size, int num_threads) { + ParallelScanner* ps = calloc(1, sizeof(ParallelScanner)); + if (!ps) + return NULL; + ps->result_queue = queue_create(100, chunk_destroy); + if (!ps->result_queue) { + 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) { + queue_destroy(ps->result_queue); + free(ps); + return NULL; + } + + DIR* dir = opendir(root_directory); + if (!dir) { + perror("Could not open root directory for parallel scan"); + parallel_scanner_destroy(ps); + return NULL; + } + + ArrayList* root_files = array_list_create(file_destroy); + ArrayList* subdirs = array_list_create(free); + struct dirent* entry; + while ((entry = readdir(dir)) != NULL) { + if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0) + continue; + char* cur_path = path_cat(root_directory, entry->d_name); + if (!cur_path) + continue; + struct stat st; + if (stat(cur_path, &st) != 0) { + free(cur_path); + continue; + } + if (S_ISDIR(st.st_mode)) { + array_list_add(subdirs, cur_path); + } else { + bool excluded = false; + for (int i = 0; i < exclude_count; i++) { + if (glob_match(exclude_patterns[i], entry->d_name)) { + excluded = true; + break; + } + } + if (excluded) { + free(cur_path); + continue; + } + if (include_count > 0) { + bool included = false; + for (int i = 0; i < include_count; i++) { + if (glob_match(include_patterns[i], entry->d_name)) { + included = true; + break; + } + } + if (!included) { + free(cur_path); + continue; + } + } + if ((max_size > 0 && (unsigned long long)st.st_size > max_size) || + (min_size > 0 && (unsigned long long)st.st_size < min_size)) { + free(cur_path); + continue; + } + File* file = file_create(cur_path); + free(cur_path); + if (!file) + continue; + file->data->size = st.st_size; + if (use_metadata) + file->metadata = file_metadata_create(&st); + array_list_add(root_files, file); + } + } + closedir(dir); + + unsigned long long cs = chunk_size > 0 ? chunk_size : DESIRED_CHUNK_SIZE; + if (root_files->size > 0) { + ArrayList* batch = array_list_create(NULL); + unsigned long long batch_size = 0; + Chunk* first = NULL; + for (int i = 0; i < root_files->size; i++) { + File* f = (File*)root_files->items[i]; + array_list_add(batch, f); + batch_size += f->data->size; + if (batch_size >= cs || i == root_files->size - 1) { + void** items = array_list_to_array(batch); + Chunk* c = chunk_create((File**)items, batch->size); + free(items); + batch->item_destroyer = NULL; + array_list_delete(batch); + batch = NULL; + if (!first) { + first = c; + } else { + queue_enqueue_multithreaded(ps->result_queue, c, &ps->result_mutex, &ps->result_not_empty, + &ps->result_not_full); + } + if (i < root_files->size - 1) { + batch = array_list_create(NULL); + batch_size = 0; + } + } + } + if (batch) { + batch->item_destroyer = NULL; + array_list_delete(batch); + } + ps->initial_chunk = first; + root_files->item_destroyer = NULL; + } + array_list_delete(root_files); + + int n = num_threads > 0 ? num_threads : 4; + if (n > subdirs->size) + n = subdirs->size > 0 ? subdirs->size : 1; + + if (subdirs->size > 0) { + ps->num_threads = n; + ps->threads = calloc(n, sizeof(thrd_t)); + if (!ps->threads) { + array_list_delete(subdirs); + parallel_scanner_destroy(ps); + return NULL; + } + int dirs_per_thread = subdirs->size / n; + int remainder = subdirs->size % n; + int start = 0; + for (int t = 0; t < n; t++) { + int count = dirs_per_thread + (t < remainder ? 1 : 0); + if (count == 0) + break; + ParallelWorkerArg* wa = calloc(1, sizeof(ParallelWorkerArg)); + if (!wa) + break; + wa->ps = ps; + wa->dirs = calloc(count, sizeof(char*)); + if (!wa->dirs) { + free(wa); + break; + } + for (int j = 0; j < count; j++) + wa->dirs[j] = str_dup((char*)subdirs->items[start + j]); + wa->dir_count = count; + wa->use_metadata = use_metadata; + wa->chunk_size = cs; + wa->exclude_patterns = exclude_patterns; + wa->exclude_count = exclude_count; + wa->include_patterns = include_patterns; + wa->include_count = include_count; + wa->max_size = max_size; + wa->min_size = min_size; + start += count; + if (thrd_create(&ps->threads[t], parallel_worker_thread, wa) != thrd_success) { + for (int j = 0; j < count; j++) + free(wa->dirs[j]); + free(wa->dirs); + free(wa); + ps->num_threads = t; + break; + } + } + } + array_list_delete(subdirs); + return ps; +} + +Chunk* parallel_scanner_next(ParallelScanner* ps) { + if (ps->initial_chunk) { + Chunk* c = ps->initial_chunk; + ps->initial_chunk = NULL; + return c; + } + if (ps->num_threads == 0) { + ps->done = true; + return NULL; + } + Chunk* chunk = queue_dequeue_multithreaded( + ps->result_queue, &ps->result_mutex, &ps->result_not_empty, &ps->result_not_full, &ps->done); + return chunk; +} + +void parallel_scanner_destroy(ParallelScanner* ps) { + if (!ps) + return; + ps->done = true; + cnd_signal(&ps->result_not_empty); + for (int i = 0; i < ps->num_threads; i++) + thrd_join(ps->threads[i], NULL); + free(ps->threads); + if (ps->initial_chunk) + chunk_destroy(ps->initial_chunk); + queue_destroy(ps->result_queue); + mtx_destroy(&ps->result_mutex); + cnd_destroy(&ps->result_not_empty); + cnd_destroy(&ps->result_not_full); + free(ps); +} diff --git a/src/client/scanner.h b/src/client/scanner.h index 8202d61..c6958fe 100644 --- a/src/client/scanner.h +++ b/src/client/scanner.h @@ -5,6 +5,7 @@ #include "queue.h" #include #include +#include typedef struct { Queue* directories; @@ -20,6 +21,18 @@ typedef struct { unsigned long long min_size; } DirectoryScanner; +typedef struct { + Queue* result_queue; + mtx_t result_mutex; + cnd_t result_not_empty; + cnd_t result_not_full; + int num_threads; + thrd_t* threads; + bool done; + int completed; + Chunk* initial_chunk; +} ParallelScanner; + DirectoryScanner* directory_scanner_create(char* root_directory, bool use_metadata, unsigned long long chunk_size, char** exclude_patterns, int exclude_count, char** include_patterns, @@ -28,4 +41,12 @@ DirectoryScanner* directory_scanner_create(char* root_directory, bool use_metada Chunk* directory_scanner_next(DirectoryScanner* scanner); void directory_scanner_destroy(DirectoryScanner* scanner); +ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata, + unsigned long long chunk_size, char** exclude_patterns, + int exclude_count, char** include_patterns, + int include_count, unsigned long long max_size, + unsigned long long min_size, int num_threads); +Chunk* parallel_scanner_next(ParallelScanner* scanner); +void parallel_scanner_destroy(ParallelScanner* scanner); + #endif diff --git a/src/shared/compression.c b/src/shared/compression.c index 3641859..97ae61f 100644 --- a/src/shared/compression.c +++ b/src/shared/compression.c @@ -2,10 +2,28 @@ #include "data.h" #include "log.h" #include "stdlib.h" +#include "string.h" +#include #include "zstd.h" #define INITIAL_DECOMPRESS_BUF_SIZE (1024 * 1024) +static const char* SKIP_COMPRESSION_EXTENSIONS[] = {".jpg", ".jpeg", ".png", ".gif", ".mp4", ".mkv", + ".zip", ".gz", ".xz", ".zst", NULL}; + +bool compression_should_skip(const char* path) { + if (!path) + return false; + const char* dot = strrchr(path, '.'); + if (!dot) + return false; + for (int i = 0; SKIP_COMPRESSION_EXTENSIONS[i]; i++) { + if (strcasecmp(dot, SKIP_COMPRESSION_EXTENSIONS[i]) == 0) + return true; + } + return false; +} + Data* data_compress(Data* data_to_compress, int compression_level) { log_message(LOG_LEVEL_DEBUG, "Starting to compress data"); size_t dst_size = ZSTD_compressBound(data_to_compress->size); diff --git a/src/shared/compression.h b/src/shared/compression.h index dcf30dd..d392f25 100644 --- a/src/shared/compression.h +++ b/src/shared/compression.h @@ -2,8 +2,10 @@ #define COMPRESSION_H #include "data.h" +#include Data* data_compress(Data* data_to_compress, int compression_level); Data* data_decompress(Data* compressed_data); +bool compression_should_skip(const char* path); #endif diff --git a/src/shared/file.c b/src/shared/file.c index 58b0455..78d0a26 100644 --- a/src/shared/file.c +++ b/src/shared/file.c @@ -102,7 +102,7 @@ bool file_send_single_calls(File* file, int file_descriptor, bool use_metadata, int compression_level, bool send_path) { const Data* data_to_send = file->data; Data* compressed_data = NULL; - if (compression_level > 0) { + if (compression_level > 0 && !compression_should_skip(file->path)) { compressed_data = data_compress(file->data, compression_level); if (compressed_data == NULL) { log_message(LOG_LEVEL_ERROR, "Failed to compress file data"); diff --git a/src/shared/protocol.c b/src/shared/protocol.c index 4d91ce6..3076ba5 100644 --- a/src/shared/protocol.c +++ b/src/shared/protocol.c @@ -2,6 +2,7 @@ #include "log.h" #include #include +#include #include #include #include @@ -10,7 +11,7 @@ static __thread int io_read_fd = -1; static __thread int io_write_fd = -1; -static SSL* io_ssl = NULL; +static SSL* io_ssl; static unsigned long long io_bwlimit = 0; static long long bw_tokens = 0; @@ -46,12 +47,11 @@ static void bw_throttle(size_t bytes_written) { bw_tokens -= (long long)bytes_written; if (bw_tokens < 0) { - long long deficit_ns = (long long)((double)(-bw_tokens) / io_bwlimit * 1000000000.0); - struct timespec sleep_time, remaining; - sleep_time.tv_sec = deficit_ns / 1000000000LL; - sleep_time.tv_nsec = deficit_ns % 1000000000LL; - while (nanosleep(&sleep_time, &remaining) < 0 && errno == EINTR) - sleep_time = remaining; + long long deficit_us = (long long)((double)(-bw_tokens) / io_bwlimit * 1000000.0); + if (deficit_us >= 1000) + poll(NULL, 0, (int)(deficit_us / 1000)); + else + usleep((useconds_t)deficit_us); bw_tokens = 0; clock_gettime(CLOCK_MONOTONIC, &bw_last_refill); } diff --git a/src/shared/protocol.h b/src/shared/protocol.h index a7854f1..113f14f 100644 --- a/src/shared/protocol.h +++ b/src/shared/protocol.h @@ -17,7 +17,10 @@ enum NET_STATUS { STATUS_MANIFEST, STATUS_CHECK, STATUS_DELTA_SIGNATURE, - STATUS_DELTA_DATA + STATUS_DELTA_DATA, + STATUS_KEEPALIVE, + STATUS_ABORT, + STATUS_CHECK_BATCH }; void io_set_fds(int read_fd, int write_fd);