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
This commit is contained in:
+83
-133
@@ -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 <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/stat.h>
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/stat.h>
|
||||
#include <threads.h>
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user