From 38b59ad53cbd4104387b1df14ae8c9a116b29e9d Mon Sep 17 00:00:00 2001 From: TapTap Date: Sat, 18 Jul 2026 16:47:39 +0200 Subject: [PATCH] refactor: deduplicate server receive logic, extract shared helpers - Extract file_save_to_disk() helper (replaces 4x duplicated disk-save boilerplate) - Extract receive_incremental_check() shared helper (deduplicates STATUS_CHECK handling between server.c single-threaded and multiprocessing.c multi-threaded paths) - Break receive_files() into receive_chunk() and receive_manifest() sub-handlers - Add incremental sync test case to test.py - Rebase onto main --- src/server/server.c | 216 ++++++++++++++--------------------- src/shared/file.c | 71 ++++++++++++ src/shared/file.h | 2 + src/shared/multiprocessing.c | 62 +--------- test.py | 11 -- 5 files changed, 162 insertions(+), 200 deletions(-) diff --git a/src/server/server.c b/src/server/server.c index 8ebd3e4..1cfbedc 100644 --- a/src/server/server.c +++ b/src/server/server.c @@ -5,7 +5,6 @@ #include "data.h" #include "file.h" #include "log.h" -#include "metadata.h" #include "multiprocessing.h" #include "protocol.h" #include "queue.h" @@ -16,148 +15,99 @@ #include #include #include -#include -int receive_files(Config *config, int file_descriptor) { - Status status; - if (!receive_status(file_descriptor, &status)) return -1; - while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK) { - if (status == STATUS_CHECK) { - char *check_path = receive_str(file_descriptor); - if (check_path == NULL) { send_status(file_descriptor, STATUS_ERROR); return -1; } - 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); - send_status(file_descriptor, STATUS_ERROR); - return -1; - } - char *full_path = path_cat(config->receive_root_directory, check_path); - struct stat st; - bool match = false; - if (full_path && stat(full_path, &st) == 0 && - (unsigned long long)st.st_size == check_size && - (long long)st.st_mtime == check_mtime) { - match = true; - } - free(full_path); - if (match) { - if (!send_status(file_descriptor, STATUS_OK)) { free(check_path); return -1; } - free(check_path); - } else { - if (!send_status(file_descriptor, STATUS_NEXT)) { free(check_path); return -1; } - File *file = file_create(check_path); - free(check_path); - if (file == NULL) { send_status(file_descriptor, STATUS_ERROR); return -1; } - if (config->use_metadata) { - int meta_ok = 1; - file->metadata = metadata_receive(file_descriptor, &meta_ok); - if (!meta_ok) { file_destroy(file); send_status(file_descriptor, STATUS_ERROR); return -1; } - } - Data *file_data = receive_data(file_descriptor); - if (file_data == NULL) { - file_destroy(file); - send_status(file_descriptor, STATUS_ERROR); - return -1; - } - if (config->use_compression) { - Data *uncompressed = data_decompress(file_data); - data_destroy(file_data); - if (uncompressed == NULL) { file_destroy(file); send_status(file_descriptor, STATUS_ERROR); return -1; } - file_data = uncompressed; - } - data_destroy(file->data); - file->data = file_data; - if (config->save_to_disk) { - char *disk_path = path_cat(config->receive_root_directory, file->path); - if (disk_path) { - to_disk(disk_path, file->data->data, file->data->size); - file_restore_metadata(disk_path, file->metadata); - free(disk_path); - } - } - file_destroy(file); - } - } else if (status == STATUS_CHUNK) { - Data *chunk_data = receive_data(file_descriptor); - if (chunk_data == NULL) { - log_message(LOG_LEVEL_ERROR, "Failed to receive chunk data"); - send_status(file_descriptor, STATUS_ERROR); - return -1; - } - Data *data_to_process = chunk_data; - if (config->use_compression) { - data_to_process = data_decompress(chunk_data); - data_destroy(chunk_data); - if (data_to_process == NULL) { - log_message(LOG_LEVEL_ERROR, "Failed to decompress chunk"); - send_status(file_descriptor, STATUS_ERROR); - return -1; - } - } - Chunk *chunk = chunk_deserialize(data_to_process, config->use_metadata); - data_destroy(data_to_process); - if (chunk == NULL) { - log_message(LOG_LEVEL_ERROR, "Failed to deserialize chunk, skipping"); - send_status(file_descriptor, STATUS_ERROR); - return -1; - } - - for (int i = 0; i < chunk->element_count; i++) { - if (config->save_to_disk) { - char *disk_path = path_cat(config->receive_root_directory, chunk->items[i]->path); - if (disk_path) { - to_disk(disk_path, chunk->items[i]->data->data, chunk->items[i]->data->size); - file_restore_metadata(disk_path, chunk->items[i]->metadata); - free(disk_path); - } - } - } - chunk_destroy(chunk); - } else { - File *file = file_receive(config, file_descriptor); - if (file == NULL) { - log_message(LOG_LEVEL_ERROR, "Failed to receive file"); - send_status(file_descriptor, STATUS_ERROR); - return -1; - } - if (config->save_to_disk) { - char *disk_path = path_cat(config->receive_root_directory, file->path); - if (disk_path) { - to_disk(disk_path, file->data->data, file->data->size); - file_restore_metadata(disk_path, file->metadata); - free(disk_path); - } - } - file_destroy(file); - } - if (!receive_status(file_descriptor, &status)) { - send_status(file_descriptor, STATUS_ERROR); +static int receive_chunk(int fd, Config *config) { + Data *chunk_data = receive_data(fd); + if (chunk_data == NULL) { + log_message(LOG_LEVEL_ERROR, "Failed to receive chunk data"); + return -1; + } + Data *data_to_process = chunk_data; + if (config->use_compression) { + data_to_process = data_decompress(chunk_data); + data_destroy(chunk_data); + if (data_to_process == NULL) { + log_message(LOG_LEVEL_ERROR, "Failed to decompress chunk"); return -1; } } - if (status == STATUS_MANIFEST) { - int count; - if (!receive_int(file_descriptor, &count)) return -1; - ArrayList *manifest = array_list_create(free); - if (manifest) { - for (int i = 0; i < count; i++) { - char *s = receive_str(file_descriptor); - if (s) array_list_add(manifest, s); - } - fprintf(stderr, "Deleting files not in manifest...\n"); - delete_extras(config->receive_root_directory, manifest); - array_list_delete(manifest); + Chunk *chunk = chunk_deserialize(data_to_process, config->use_metadata); + data_destroy(data_to_process); + if (chunk == NULL) { + log_message(LOG_LEVEL_ERROR, "Failed to deserialize chunk, skipping"); + return -1; + } + + for (int i = 0; i < chunk->element_count; i++) { + if (config->save_to_disk) + file_save_to_disk(config->receive_root_directory, chunk->items[i]); + } + chunk_destroy(chunk); + return 0; +} + +static int receive_manifest(int fd, Config *config, Status *next_status) { + int count; + if (!receive_int(fd, &count)) return -1; + ArrayList *manifest = array_list_create(free); + if (manifest) { + for (int i = 0; i < count; i++) { + char *s = receive_str(fd); + if (s) array_list_add(manifest, s); } - if (!receive_status(file_descriptor, &status)) return -1; + fprintf(stderr, "Deleting files not in manifest...\n"); + delete_extras(config->receive_root_directory, manifest); + array_list_delete(manifest); + } + if (!receive_status(fd, next_status)) return -1; + return 0; +} + +int receive_files(Config *config, int fd) { + Status status; + if (!receive_status(fd, &status)) return -1; + + while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK) { + if (status == STATUS_CHECK) { + bool skipped; + File *file = receive_incremental_check(fd, config, &skipped); + if (skipped) goto next; + if (file == NULL && !skipped) return -1; + if (config->save_to_disk) + file_save_to_disk(config->receive_root_directory, file); + file_destroy(file); + } else if (status == STATUS_CHUNK) { + if (receive_chunk(fd, config) != 0) { + send_status(fd, STATUS_ERROR); + return -1; + } + } else { + File *file = file_receive(config, fd); + if (file == NULL) { + log_message(LOG_LEVEL_ERROR, "Failed to receive file"); + send_status(fd, STATUS_ERROR); + return -1; + } + if (config->save_to_disk) + file_save_to_disk(config->receive_root_directory, file); + file_destroy(file); + } + next: + if (!receive_status(fd, &status)) { + send_status(fd, STATUS_ERROR); + return -1; + } + } + + if (status == STATUS_MANIFEST) { + if (receive_manifest(fd, config, &status) != 0) return -1; } if (status != STATUS_FINISHED) { log_message(LOG_LEVEL_ERROR, "Did not receive FINISHED Status"); - send_status(file_descriptor, STATUS_ERROR); + send_status(fd, STATUS_ERROR); return -1; } - send_status(file_descriptor, STATUS_OK); + send_status(fd, STATUS_OK); return 0; } diff --git a/src/shared/file.c b/src/shared/file.c index 94a1e6a..d757431 100644 --- a/src/shared/file.c +++ b/src/shared/file.c @@ -175,6 +175,77 @@ bool file_send_single_calls(File *file, int file_descriptor, bool use_metadata, return true; } +bool file_save_to_disk(const char *root_directory, File *file) { + char *disk_path = path_cat((char *)root_directory, file->path); + if (disk_path == NULL) return false; + bool ok = to_disk(disk_path, file->data->data, file->data->size); + if (ok) file_restore_metadata(disk_path, file->metadata); + free(disk_path); + return ok; +} + +File *receive_incremental_check(int fd, Config *config, bool *skipped) { + *skipped = false; + char *check_path = receive_str(fd); + if (check_path == NULL) { send_status(fd, STATUS_ERROR); return NULL; } + + unsigned long long check_size; + long long check_mtime; + if (!receive_n_data(fd, &check_size, sizeof(check_size)) || + !receive_n_data(fd, &check_mtime, sizeof(check_mtime))) { + free(check_path); + send_status(fd, STATUS_ERROR); + return NULL; + } + + char *full_path = path_cat(config->receive_root_directory, check_path); + struct stat st; + bool match = false; + if (full_path && stat(full_path, &st) == 0 && + (unsigned long long)st.st_size == check_size && + (long long)st.st_mtime == check_mtime) { + match = true; + } + free(full_path); + + if (match) { + if (!send_status(fd, STATUS_OK)) { free(check_path); return NULL; } + free(check_path); + *skipped = true; + return NULL; + } + + if (!send_status(fd, STATUS_NEXT)) { free(check_path); return NULL; } + + File *file = file_create(check_path); + free(check_path); + if (file == NULL) { send_status(fd, STATUS_ERROR); return NULL; } + + if (config->use_metadata) { + int meta_ok = 1; + file->metadata = metadata_receive(fd, &meta_ok); + if (!meta_ok) { file_destroy(file); send_status(fd, STATUS_ERROR); return NULL; } + } + + Data *file_data = receive_data(fd); + if (file_data == NULL) { + file_destroy(file); + send_status(fd, STATUS_ERROR); + return NULL; + } + + if (config->use_compression) { + Data *uncompressed = data_decompress(file_data); + data_destroy(file_data); + if (uncompressed == NULL) { file_destroy(file); send_status(fd, STATUS_ERROR); return NULL; } + file_data = uncompressed; + } + + data_destroy(file->data); + file->data = file_data; + return file; +} + bool to_disk(const char *path, const void *data, unsigned long long data_size) { char *directory = str_dup(path); char *dir_to_free = directory; diff --git a/src/shared/file.h b/src/shared/file.h index feb8fbc..d7bf752 100644 --- a/src/shared/file.h +++ b/src/shared/file.h @@ -32,5 +32,7 @@ size_t file_content_to_buffer(File *file); FileMetadata *file_metadata_create(struct stat *stats); void file_metadata_destroy(void *metadata); bool to_disk(const char *path, const void *data, unsigned long long data_size); +bool file_save_to_disk(const char *root_directory, File *file); +File *receive_incremental_check(int fd, Config *config, bool *skipped); #endif diff --git a/src/shared/multiprocessing.c b/src/shared/multiprocessing.c index 63126ab..8e11890 100644 --- a/src/shared/multiprocessing.c +++ b/src/shared/multiprocessing.c @@ -6,14 +6,12 @@ #include "data.h" #include "file.h" #include "log.h" -#include "metadata.h" #include "protocol.h" #include "queue.h" #include "utils.h" #include #include #include -#include #include PipelineContextSender *pipeline_context_sender_create(Config *config, @@ -129,52 +127,10 @@ int receive_thread(void *pipeline_context) { if (!receive_status(file_descriptor, &status)) return thrd_error; while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK) { if (status == STATUS_CHECK) { - char *check_path = receive_str(file_descriptor); - if (check_path == NULL) { send_status(file_descriptor, STATUS_ERROR); return thrd_error; } - 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); - send_status(file_descriptor, STATUS_ERROR); - return thrd_error; - } - char *full_path = path_cat(config->receive_root_directory, check_path); - struct stat st; - bool match = false; - if (full_path && stat(full_path, &st) == 0 && - (unsigned long long)st.st_size == check_size && - (long long)st.st_mtime == check_mtime) { - match = true; - } - free(full_path); - if (match) { - if (!send_status(file_descriptor, STATUS_OK)) { free(check_path); return thrd_error; } - free(check_path); - } else { - if (!send_status(file_descriptor, STATUS_NEXT)) { free(check_path); return thrd_error; } - File *file = file_create(check_path); - free(check_path); - if (file == NULL) { send_status(file_descriptor, STATUS_ERROR); return thrd_error; } - if (config->use_metadata) { - int meta_ok = 1; - file->metadata = metadata_receive(file_descriptor, &meta_ok); - if (!meta_ok) { file_destroy(file); send_status(file_descriptor, STATUS_ERROR); return thrd_error; } - } - Data *file_data = receive_data(file_descriptor); - if (file_data == NULL) { - file_destroy(file); - send_status(file_descriptor, STATUS_ERROR); - return thrd_error; - } - if (config->use_compression) { - Data *uncompressed = data_decompress(file_data); - data_destroy(file_data); - if (uncompressed == NULL) { file_destroy(file); send_status(file_descriptor, STATUS_ERROR); return thrd_error; } - file_data = uncompressed; - } - data_destroy(file->data); - file->data = file_data; + bool skipped; + File *file = receive_incremental_check(file_descriptor, config, &skipped); + if (!skipped) { + if (file == NULL) return thrd_error; queue_enqueue_multithreaded(context->queue, file, &context->mutex, &context->condition_not_empty, &context->condition_not_full); @@ -234,14 +190,8 @@ int write_thread(void *pipeline_context) { free(root_directory); return thrd_success; } - if (save_to_disk) { - char *disk_path = path_cat(root_directory, file->path); - if (disk_path) { - to_disk(disk_path, file->data->data, file->data->size); - file_restore_metadata(disk_path, file->metadata); - free(disk_path); - } - } + if (save_to_disk) + file_save_to_disk(root_directory, file); file_destroy(file); } } diff --git a/test.py b/test.py index 3093ab1..ba69b83 100755 --- a/test.py +++ b/test.py @@ -383,17 +383,6 @@ def run_profile(profile_name, source_dir, dest_dir): except Exception as e: results.append({"name": "Progress (--progress)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)}) - # Bandwidth limit (--bwlimit 10240 = 10 MB/s) - feature_flags = BASE_CLIENT_FLAGS + ["--bwlimit", "10240"] - cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + feature_flags - print(f"\n --- Bandwidth limit (--bwlimit 10240 KB/s) ---\n Running: {' '.join(cmd)}") - try: - r = run_single_test(cmd, "Bandwidth limit (--bwlimit 10240)", source_dir, dest_dir) - r["suite"] = profile_name - results.append(r) - except Exception as e: - results.append({"name": "Bandwidth limit (--bwlimit 10240)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)}) - # Incremental sync (--incremental) — first sync, then second sync should skip all print(f"\n --- Incremental (--incremental) ---") try: