diff --git a/src/client/client_send.c b/src/client/client_send.c index 8dfb2af..a610c0d 100644 --- a/src/client/client_send.c +++ b/src/client/client_send.c @@ -16,22 +16,14 @@ #include "transport_ssh.h" #include "transport_tls.h" #include "utils.h" -#include #include #include #include #include #include +#include -static volatile sig_atomic_t g_abort_requested = 0; -static int g_abort_fd = -1; - -static void handle_sigint(int sig) { - (void)sig; - g_abort_requested = 1; -} - -#define KEEPALIVE_INTERVAL 30 +#define STREAM_THRESHOLD (64ULL * 1024 * 1024) static int incremental_check(Client* client, File* file, DeltaSignature** out_sig) { *out_sig = NULL; @@ -107,35 +99,6 @@ static int send_delta(Client* client, File* file, DeltaSignature* sig, Config* c return ok ? 0 : -1; } -static bool batch_incremental_check(Client* client, ArrayList* files) { - if (!send_status(client->file_descriptor, STATUS_CHECK_BATCH)) - return false; - if (!send_int(client->file_descriptor, files->size)) - return false; - for (int i = 0; i < files->size; i++) { - File* file = (File*)files->items[i]; - if (!send_str(client->file_descriptor, file->path)) - return false; - unsigned long long fsize = file->data ? file->data->size : 0; - long long mtime = file->metadata ? file->metadata->mtime_sec : 0; - if (!send_n_data(client->file_descriptor, &fsize, sizeof(fsize))) - return false; - if (!send_n_data(client->file_descriptor, &mtime, sizeof(mtime))) - return false; - } - for (int i = 0; i < files->size; i++) { - Status s; - if (!receive_status(client->file_descriptor, &s)) - return false; - File* file = (File*)files->items[i]; - if (s == STATUS_OK) - file->skip = true; - else if (s == STATUS_ERROR) - return false; - } - return true; -} - typedef bool (*file_send_fn)(File*, int, bool, int, bool); // Send a single file directly (non-incremental path). @@ -158,9 +121,6 @@ static int send_single_file(Client* client, File* file, Config* config, bool use bool use_sendfile) { int compression_level = config->use_compression ? config->compression_level : 0; - if (file->skip) - return 1; - if (!use_incremental) { if (use_sendfile) { return send_file_direct_sendfile(file, client->file_descriptor, config->use_metadata) ? 0 @@ -251,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) @@ -298,30 +261,7 @@ static int send_chunks_multithreaded(void* pipeline_context) { return thrd_error; } - g_abort_fd = client->file_descriptor; - time_t last_activity = time(NULL); while (true) { - if (g_abort_requested) { - send_status(client->file_descriptor, STATUS_ABORT); - client_disconnect(client); - client_delete(client); - return thrd_error; - } - time_t now = time(NULL); - if (now - last_activity >= KEEPALIVE_INTERVAL) { - if (!send_status(client->file_descriptor, STATUS_KEEPALIVE)) { - client_disconnect(client); - client_delete(client); - return thrd_error; - } - Status s; - if (!receive_status(client->file_descriptor, &s)) { - client_disconnect(client); - client_delete(client); - return thrd_error; - } - last_activity = now; - } Chunk* current_chunk = queue_dequeue_multithreaded( context->queue_loader, &context->mutex_loader, &context->condition_not_empty_loader, &context->condition_not_full_loader, &context->loader_done); @@ -361,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, context->config->max_depth); - 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++) { @@ -390,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; } @@ -409,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; } } @@ -427,24 +368,21 @@ int send_files(Config* config) { DirectoryScanner* scanner = directory_scanner_create( config->send_directory, config->use_metadata, config->chunk_size, config->exclude_patterns, config->exclude_count, config->include_patterns, config->include_count, config->max_size, - config->min_size, config->max_depth); + config->min_size); Chunk* chunk; int file_count = 0; unsigned long long total_bytes = 0; - if (!config->quiet) - printf("Dry run: files to be transferred\n"); + printf("Dry run: files to be transferred\n"); while ((chunk = directory_scanner_next(scanner)) != NULL) { for (int i = 0; i < chunk->element_count; i++) { - if (!config->quiet) - printf(" %s (%zu bytes)\n", chunk->items[i]->path, chunk->items[i]->data->size); + printf(" %s (%zu bytes)\n", chunk->items[i]->path, chunk->items[i]->data->size); total_bytes += chunk->items[i]->data->size; file_count++; } chunk_destroy(chunk); } directory_scanner_destroy(scanner); - if (!config->quiet) - printf("Total: %d files, %.1f MB\n", file_count, total_bytes / 1048576.0); + printf("Total: %d files, %.1f MB\n", file_count, total_bytes / 1048576.0); return 0; } @@ -480,98 +418,45 @@ int send_files(Config* config) { client_delete(client); return 1; } - - g_abort_fd = client->file_descriptor; - struct sigaction sa; - memset(&sa, 0, sizeof(sa)); - sa.sa_handler = handle_sigint; - sigaction(SIGINT, &sa, NULL); - sigaction(SIGTERM, &sa, NULL); - DirectoryScanner* scanner = directory_scanner_create( config->send_directory, config->use_metadata, config->chunk_size, config->exclude_patterns, config->exclude_count, config->include_patterns, config->include_count, config->max_size, - config->min_size, config->max_depth); - ArrayList* all_files = array_list_create(NULL); - ArrayList* manifest = config->use_delete ? array_list_create(free) : NULL; + config->min_size); Chunk* current_chunk; + unsigned long long total_bytes = 0; + time_t last_progress = 0; + time_t start = time(NULL); + ArrayList* manifest = config->use_delete ? array_list_create(free) : NULL; while ((current_chunk = directory_scanner_next(scanner)) != NULL) { + unsigned long long chunk_bytes = 0; for (int i = 0; i < current_chunk->element_count; i++) { - File* f = current_chunk->items[i]; - array_list_add(all_files, f); - current_chunk->items[i] = NULL; + chunk_bytes += current_chunk->items[i]->data->size; if (manifest) { - const char* p = f->path; + const char* p = current_chunk->items[i]->path; if (*p == '/') p++; array_list_add(manifest, str_dup(p)); } } - chunk_destroy(current_chunk); - } - directory_scanner_destroy(scanner); - scanner = NULL; - - bool batch_ok = true; - if (config->use_incremental && all_files->size > 0) { - if (!batch_incremental_check(client, all_files)) { - log_message(LOG_LEVEL_ERROR, "Batch incremental check failed"); - batch_ok = false; + if (!config->use_sendfile) { + for (int i = 0; i < current_chunk->element_count; 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; + } + } } - } - - unsigned long long total_bytes = 0; - time_t last_progress = 0; - time_t last_activity = 0; - time_t start = time(NULL); - bool use_sendfile = config->use_sendfile && !config->use_compression; - for (int i = 0; i < all_files->size; i++) { - File* file = (File*)all_files->items[i]; - if (file->skip) - continue; - if (g_abort_requested) { - send_status(client->file_descriptor, STATUS_ABORT); - batch_ok = false; + if (send_chunk(client, current_chunk, config) != 0) { + log_message(LOG_LEVEL_ERROR, "Failed to send chunk"); + chunk_destroy(current_chunk); break; } - time_t now = time(NULL); - if (now - last_activity >= KEEPALIVE_INTERVAL) { - if (!send_status(client->file_descriptor, STATUS_KEEPALIVE)) { - batch_ok = false; - break; - } - Status s; - if (!receive_status(client->file_descriptor, &s)) { - batch_ok = false; - break; - } - last_activity = now; - } - int compression_level = config->use_compression ? config->compression_level : 0; - if (!send_status(client->file_descriptor, STATUS_NEXT)) { - batch_ok = false; - break; - } - if (use_sendfile) { - if (!file_send_sendfile(file, client->file_descriptor, config->use_metadata, 0, true)) { - log_message(LOG_LEVEL_ERROR, "Failed to send file via sendfile"); - batch_ok = false; - break; - } - } else { - if (!file_load_data(file)) { - log_message(LOG_LEVEL_ERROR, "Failed to load file data"); - continue; - } - if (!file_send_single_calls(file, client->file_descriptor, config->use_metadata, - compression_level, true)) { - log_message(LOG_LEVEL_ERROR, "Failed to send file"); - batch_ok = false; - break; - } - } - total_bytes += file->data ? file->data->size : 0; if (config->show_progress) { + total_bytes += chunk_bytes; + time_t now = time(NULL); if (now - last_progress >= 1) { last_progress = now; double elapsed = difftime(now, start); @@ -580,69 +465,81 @@ int send_files(Config* config) { fflush(stderr); } } + chunk_destroy(current_chunk); } - - if (batch_ok && config->use_delete && manifest) { - if (!send_status(client->file_descriptor, STATUS_MANIFEST)) - batch_ok = false; - else if (!send_int(client->file_descriptor, manifest->size)) - batch_ok = false; - else { - for (int i = 0; i < manifest->size && batch_ok; i++) { - if (!send_str(client->file_descriptor, (char*)manifest->items[i])) - batch_ok = false; + if (config->use_delete) { + if (!send_status(client->file_descriptor, STATUS_MANIFEST)) { + array_list_delete(manifest); + goto send_fail; + } + if (!send_int(client->file_descriptor, manifest->size)) { + array_list_delete(manifest); + goto send_fail; + } + for (int i = 0; i < manifest->size; i++) { + if (!send_str(client->file_descriptor, (char*)manifest->items[i])) { + array_list_delete(manifest); + goto send_fail; } } + array_list_delete(manifest); } - array_list_delete(manifest); - - if (batch_ok && !send_status(client->file_descriptor, STATUS_FINISHED)) - batch_ok = false; + if (!send_status(client->file_descriptor, STATUS_FINISHED)) + goto send_fail; Status s; - int ok = 0; - if (batch_ok) - ok = receive_status(client->file_descriptor, &s) && s == STATUS_OK; + int ok = receive_status(client->file_descriptor, &s) && s == STATUS_OK; if (config->show_progress) { double elapsed = difftime(time(NULL), start); double rate = elapsed > 0 ? total_bytes / (1048576.0 * elapsed) : 0; fprintf(stderr, "\rSent %.1f MB (%.1f MB/s) Done.\n", total_bytes / 1048576.0, rate); } - for (int i = 0; i < all_files->size; i++) - file_destroy(all_files->items[i]); - array_list_delete(all_files); + directory_scanner_destroy(scanner); client_disconnect(client); client_delete(client); - return (batch_ok && ok) ? 0 : -1; + return ok ? 0 : -1; + +send_fail: + directory_scanner_destroy(scanner); + client_disconnect(client); + client_delete(client); + return -1; } int send_files_multithreaded(Config* config) { - time_t start_time = time(NULL); if (config->dry_run) { DirectoryScanner* scanner = directory_scanner_create( config->send_directory, config->use_metadata, config->chunk_size, config->exclude_patterns, config->exclude_count, config->include_patterns, config->include_count, config->max_size, - config->min_size, config->max_depth); + config->min_size); Chunk* chunk; int file_count = 0; unsigned long long total_bytes = 0; - if (!config->quiet) - printf("Dry run: files to be transferred\n"); + printf("Dry run: files to be transferred\n"); while ((chunk = directory_scanner_next(scanner)) != NULL) { for (int i = 0; i < chunk->element_count; i++) { - if (!config->quiet) - printf(" %s (%zu bytes)\n", chunk->items[i]->path, chunk->items[i]->data->size); + printf(" %s (%zu bytes)\n", chunk->items[i]->path, chunk->items[i]->data->size); total_bytes += chunk->items[i]->data->size; file_count++; } chunk_destroy(chunk); } directory_scanner_destroy(scanner); - if (!config->quiet) - printf("Total: %d files, %.1f MB\n", file_count, total_bytes / 1048576.0); + printf("Total: %d files, %.1f MB\n", file_count, total_bytes / 1048576.0); return 0; } - int qsize = config->queue_size > 0 ? config->queue_size : 100; + 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) { @@ -675,12 +572,6 @@ int send_files_multithreaded(Config* config) { thrd_join(loader, NULL); thrd_join(sender, &sender_result); - if (config->stats && !config->quiet) { - double elapsed = difftime(time(NULL), start_time); - printf("\nTransfer statistics:\n"); - printf(" Elapsed time: %.1f sec\n", elapsed); - } - pipeline_context_sender_destroy(context); return sender_result == thrd_success ? 0 : -1; } diff --git a/src/client/scanner.c b/src/client/scanner.c index d5f2d18..ee194c3 100644 --- a/src/client/scanner.c +++ b/src/client/scanner.c @@ -9,6 +9,7 @@ #include #include #include +#include #include typedef struct { @@ -79,7 +80,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); @@ -202,3 +202,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 7a77a25..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; @@ -18,16 +19,34 @@ typedef struct { int include_count; unsigned long long max_size; unsigned long long min_size; - int max_depth; - int current_depth; } DirectoryScanner; -DirectoryScanner* directory_scanner_create(const char* root_directory, bool use_metadata, +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, int include_count, unsigned long long max_size, - unsigned long long min_size, int max_depth); + unsigned long long min_size); 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 7c60686..97ae61f 100644 --- a/src/shared/compression.c +++ b/src/shared/compression.c @@ -1,26 +1,31 @@ #include "compression.h" #include "data.h" #include "log.h" -#include -#include +#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"); - - /* Clamp compression level to valid zstd range [1, 22] */ - if (compression_level < 1) { - log_message(LOG_LEVEL_WARNING, "compression_level %d out of range [1,22], using 1", - compression_level); - compression_level = 1; - } else if (compression_level > 22) { - log_message(LOG_LEVEL_WARNING, "compression_level %d out of range [1,22], using 22", - compression_level); - compression_level = 22; - } - size_t dst_size = ZSTD_compressBound(data_to_compress->size); Data* compressed_data = data_create_empty(dst_size); if (compressed_data == NULL) @@ -79,16 +84,8 @@ Data* data_decompress(Data* compressed_data) { return NULL; } - size_t buf_size = INITIAL_DECOMPRESS_BUF_SIZE; - if (!ZSTD_isError(dst_size) && dst_size > 0) { - if (dst_size > SIZE_MAX) { - log_message(LOG_LEVEL_ERROR, - "Decompressed size %llu exceeds addressable memory, using fallback buffer", - dst_size); - } else { - buf_size = (size_t)dst_size; - } - } + size_t buf_size = + (!ZSTD_isError(dst_size) && dst_size > 0) ? (size_t)dst_size : INITIAL_DECOMPRESS_BUF_SIZE; Data* uncompressed_data = data_create_empty(buf_size); if (!uncompressed_data) { log_message(LOG_LEVEL_ERROR, "Failed to allocate decompression buffer"); 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 6e65a06..a531e3d 100644 --- a/src/shared/file.c +++ b/src/shared/file.c @@ -104,7 +104,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 32c2539..14192c9 100644 --- a/src/shared/protocol.c +++ b/src/shared/protocol.c @@ -2,6 +2,7 @@ #include "log.h" #include #include +#include #include #include #include @@ -14,7 +15,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; @@ -52,12 +53,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); }