Compare commits

...

7 Commits

Author SHA1 Message Date
TapTap 38b59ad53c 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
2026-07-18 16:47:39 +02:00
TapTap 5cde1a832d fix: address review #69 - STATUS_ERROR sends, metadata_receive error handling, extract incremental_check helper 2026-07-18 16:38:33 +02:00
TapTap 9ca056c4cf Fix incremental-sync review: sendfile_no_path, reject -s+incremental, no data mutation, auto -M, STATUS_ERROR handling, PROTOCOL_VERSION bump 2026-07-18 16:38:33 +02:00
TapTap 109064441a Incremental sync: --incremental flag to skip unchanged files
- New STATUS_CHECK protocol status (value 6)
- Client sends path + size + mtime; server replies OK (skip) or NEXT (send)
- config.h/c: use_incremental field sent/received over wire
- file.c/h: file_send_single_calls_no_path helper for incremental path
- client_cli.c: --incremental flag
- client_send.c: per-file check before send in both single and chunk paths
- server.c: STATUS_CHECK handling in receive_files()
- multiprocessing.c: STATUS_CHECK handling in receive_thread()
- test.py: incremental sync test case
2026-07-18 16:38:33 +02:00
TapTap ecd486a8d0 fix: bw_tokens initialized to io_bwlimit, overflow check for --bwlimit 2026-07-18 16:38:33 +02:00
TapTap 115e492a54 Fix bwlimit review: remove __thread (breaks -m), validate >0, handle nanosleep EINTR 2026-07-18 16:38:33 +02:00
TapTap 34510ac921 Bandwidth throttling: --bwlimit <KB/s> with token bucket
- protocol.h/c: io_set_bwlimit() + token bucket in send_n_data
  (64KB chunks, nanosleep-based deficit compensation)
- client_cli.c: --bwlimit <KB/s> flag
- test.py: bandwidth limit test case
2026-07-18 16:38:33 +02:00
13 changed files with 416 additions and 114 deletions
+31
View File
@@ -1,7 +1,10 @@
#include "client_send.h" #include "client_send.h"
#include "config.h" #include "config.h"
#include "log.h" #include "log.h"
#include "protocol.h"
#include "utils.h" #include "utils.h"
#include <errno.h>
#include <limits.h>
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
@@ -31,6 +34,7 @@ static void print_usage(void) {
printf(" --include <pattern> Only include files matching pattern\n"); printf(" --include <pattern> Only include files matching pattern\n");
printf(" --max-size <n> Skip files larger than n bytes\n"); printf(" --max-size <n> Skip files larger than n bytes\n");
printf(" --min-size <n> Skip files smaller than n bytes\n"); printf(" --min-size <n> Skip files smaller than n bytes\n");
printf(" --incremental Skip files unchanged since last transfer\n");
printf(" -m Enable multithreading\n"); printf(" -m Enable multithreading\n");
printf(" -s Enable chunk serialization\n"); printf(" -s Enable chunk serialization\n");
printf(" -f Enable sendfile (TCP only, not with -c or -s)\n"); printf(" -f Enable sendfile (TCP only, not with -c or -s)\n");
@@ -42,6 +46,7 @@ static void print_usage(void) {
printf(" --save-to-disk Write received files to disk\n"); printf(" --save-to-disk Write received files to disk\n");
printf(" --server-host <ip> Server IP address (default: 127.0.0.1)\n"); printf(" --server-host <ip> Server IP address (default: 127.0.0.1)\n");
printf(" --server-port <n> Server port (default: 8080)\n"); printf(" --server-port <n> Server port (default: 8080)\n");
printf(" --bwlimit <KB/s> Bandwidth limit in kilobytes per second\n");
printf(" --help Show this help\n"); printf(" --help Show this help\n");
} }
@@ -89,6 +94,8 @@ int main(int argc, char *argv[]) {
config->max_size = strtoull(argv[++i], NULL, 10); config->max_size = strtoull(argv[++i], NULL, 10);
} else if (strcmp(argv[i], "--min-size") == 0 && i + 1 < argc) { } else if (strcmp(argv[i], "--min-size") == 0 && i + 1 < argc) {
config->min_size = strtoull(argv[++i], NULL, 10); config->min_size = strtoull(argv[++i], NULL, 10);
} else if (strcmp(argv[i], "--incremental") == 0) {
config->use_incremental = true;
} else if (strcmp(argv[i], "-c") == 0 || strcmp(argv[i], "-z") == 0) { } else if (strcmp(argv[i], "-c") == 0 || strcmp(argv[i], "-z") == 0) {
config->use_compression = true; config->use_compression = true;
log_message(LOG_LEVEL_INFO, "Enabled Compression"); log_message(LOG_LEVEL_INFO, "Enabled Compression");
@@ -127,6 +134,20 @@ int main(int argc, char *argv[]) {
server_host = str_dup(argv[++i]); server_host = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--server-port") == 0 && i + 1 < argc) { } else if (strcmp(argv[i], "--server-port") == 0 && i + 1 < argc) {
server_port = atoi(argv[++i]); server_port = atoi(argv[++i]);
} else if (strcmp(argv[i], "--bwlimit") == 0 && i + 1 < argc) {
char *end;
errno = 0;
unsigned long long kbps = strtoull(argv[++i], &end, 10);
if (errno != 0 || *end != '\0' || kbps == 0) {
fprintf(stderr, "Error: --bwlimit must be a positive integer\n");
return 1;
}
if (kbps > ULLONG_MAX / 1024) {
fprintf(stderr, "Error: --bwlimit value too large\n");
return 1;
}
io_set_bwlimit(kbps * 1024);
log_message(LOG_LEVEL_INFO, "Set bandwidth limit to %llu KB/s", kbps);
} else if (strcmp(argv[i], "--progress") == 0) { } else if (strcmp(argv[i], "--progress") == 0) {
config->show_progress = true; config->show_progress = true;
} else if (strcmp(argv[i], "--chunk-size") == 0 && i + 1 < argc) { } else if (strcmp(argv[i], "--chunk-size") == 0 && i + 1 < argc) {
@@ -184,6 +205,16 @@ int main(int argc, char *argv[]) {
return 1; return 1;
} }
if (config->use_incremental && config->use_chunk_serialization) {
fprintf(stderr, "Error: --incremental is not supported with -s (chunk serialization)\n");
return 1;
}
if (config->use_incremental && !config->use_metadata) {
log_message(LOG_LEVEL_INFO, "Enabling metadata preservation for --incremental");
config->use_metadata = true;
}
if (config->use_multithreading) if (config->use_multithreading)
return send_files_multithreaded(config); return send_files_multithreaded(config);
return send_files(config); return send_files(config);
+39
View File
@@ -18,6 +18,27 @@
#include <threads.h> #include <threads.h>
#include <time.h> #include <time.h>
static int incremental_check(Client *client, File *file) {
if (!send_status(client->file_descriptor, STATUS_CHECK)) return -1;
if (!send_str(client->file_descriptor, file->path)) return -1;
unsigned long long fsize = file->data->size;
long long mtime = file->metadata ? file->metadata->mtime_sec : 0;
if (!send_n_data(client->file_descriptor, &fsize, sizeof(fsize))) return -1;
if (!send_n_data(client->file_descriptor, &mtime, sizeof(mtime))) return -1;
Status s;
if (!receive_status(client->file_descriptor, &s)) return -1;
if (s == STATUS_ERROR) {
log_message(LOG_LEVEL_ERROR, "Server reported error for file");
return -1;
}
if (s == STATUS_OK) return 1;
if (s != STATUS_NEXT) {
log_message(LOG_LEVEL_ERROR, "Unexpected server status");
return -1;
}
return 0;
}
int send_chunk(Client *client, Chunk *chunk, Config *config) { int send_chunk(Client *client, Chunk *chunk, Config *config) {
if (config->use_chunk_serialization) { if (config->use_chunk_serialization) {
if (!send_status(client->file_descriptor, STATUS_CHUNK)) return -1; if (!send_status(client->file_descriptor, STATUS_CHUNK)) return -1;
@@ -32,12 +53,29 @@ int send_chunk(Client *client, Chunk *chunk, Config *config) {
data_destroy(data); data_destroy(data);
} else if (config->use_sendfile && !config->use_compression) { } else if (config->use_sendfile && !config->use_compression) {
for (int i = 0; i < chunk->element_count; i++) { for (int i = 0; i < chunk->element_count; i++) {
if (config->use_incremental) {
int rc = incremental_check(client, chunk->items[i]);
if (rc < 0) return -1;
if (rc > 0) continue;
if (!file_send_sendfile_no_path(chunk->items[i], client->file_descriptor, config->use_metadata))
return -1;
} else {
if (!send_status(client->file_descriptor, STATUS_NEXT)) return -1; if (!send_status(client->file_descriptor, STATUS_NEXT)) return -1;
if (!file_send_sendfile(chunk->items[i], client->file_descriptor, config->use_metadata)) if (!file_send_sendfile(chunk->items[i], client->file_descriptor, config->use_metadata))
return -1; return -1;
} }
}
} else { } else {
for (int i = 0; i < chunk->element_count; i++) { for (int i = 0; i < chunk->element_count; i++) {
if (config->use_incremental) {
int rc = incremental_check(client, chunk->items[i]);
if (rc < 0) return -1;
if (rc > 0) continue;
if (!file_send_single_calls_no_path(chunk->items[i], client->file_descriptor,
config->use_metadata,
config->use_compression ? config->compression_level : 0))
return -1;
} else {
if (!send_status(client->file_descriptor, STATUS_NEXT)) return -1; if (!send_status(client->file_descriptor, STATUS_NEXT)) return -1;
if (!file_send_single_calls(chunk->items[i], client->file_descriptor, if (!file_send_single_calls(chunk->items[i], client->file_descriptor,
config->use_metadata, config->use_metadata,
@@ -45,6 +83,7 @@ int send_chunk(Client *client, Chunk *chunk, Config *config) {
return -1; return -1;
} }
} }
}
return 0; return 0;
} }
+54 -46
View File
@@ -5,7 +5,6 @@
#include "data.h" #include "data.h"
#include "file.h" #include "file.h"
#include "log.h" #include "log.h"
#include "metadata.h"
#include "multiprocessing.h" #include "multiprocessing.h"
#include "protocol.h" #include "protocol.h"
#include "queue.h" #include "queue.h"
@@ -17,15 +16,10 @@
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
int receive_files(Config *config, int file_descriptor) { static int receive_chunk(int fd, Config *config) {
Status status; Data *chunk_data = receive_data(fd);
if (!receive_status(file_descriptor, &status)) return -1;
while (status == STATUS_NEXT || status == STATUS_CHUNK) {
if (status == STATUS_CHUNK) {
Data *chunk_data = receive_data(file_descriptor);
if (chunk_data == NULL) { if (chunk_data == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to receive chunk data"); log_message(LOG_LEVEL_ERROR, "Failed to receive chunk data");
send_status(file_descriptor, STATUS_ERROR);
return -1; return -1;
} }
Data *data_to_process = chunk_data; Data *data_to_process = chunk_data;
@@ -34,7 +28,6 @@ int receive_files(Config *config, int file_descriptor) {
data_destroy(chunk_data); data_destroy(chunk_data);
if (data_to_process == NULL) { if (data_to_process == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to decompress chunk"); log_message(LOG_LEVEL_ERROR, "Failed to decompress chunk");
send_status(file_descriptor, STATUS_ERROR);
return -1; return -1;
} }
} }
@@ -42,64 +35,79 @@ int receive_files(Config *config, int file_descriptor) {
data_destroy(data_to_process); data_destroy(data_to_process);
if (chunk == NULL) { if (chunk == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to deserialize chunk, skipping"); log_message(LOG_LEVEL_ERROR, "Failed to deserialize chunk, skipping");
send_status(file_descriptor, STATUS_ERROR);
return -1; return -1;
} }
for (int i = 0; i < chunk->element_count; i++) { for (int i = 0; i < chunk->element_count; i++) {
if (config->save_to_disk) { if (config->save_to_disk)
char *disk_path = path_cat(config->receive_root_directory, chunk->items[i]->path); file_save_to_disk(config->receive_root_directory, chunk->items[i]);
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); chunk_destroy(chunk);
} else { return 0;
File *file = file_receive(config, file_descriptor); }
if (file == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to receive file"); static int receive_manifest(int fd, Config *config, Status *next_status) {
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);
return -1;
}
}
if (status == STATUS_MANIFEST) {
int count; int count;
if (!receive_int(file_descriptor, &count)) return -1; if (!receive_int(fd, &count)) return -1;
ArrayList *manifest = array_list_create(free); ArrayList *manifest = array_list_create(free);
if (manifest) { if (manifest) {
for (int i = 0; i < count; i++) { for (int i = 0; i < count; i++) {
char *s = receive_str(file_descriptor); char *s = receive_str(fd);
if (s) array_list_add(manifest, s); if (s) array_list_add(manifest, s);
} }
fprintf(stderr, "Deleting files not in manifest...\n"); fprintf(stderr, "Deleting files not in manifest...\n");
delete_extras(config->receive_root_directory, manifest); delete_extras(config->receive_root_directory, manifest);
array_list_delete(manifest); array_list_delete(manifest);
} }
if (!receive_status(file_descriptor, &status)) return -1; 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) { if (status != STATUS_FINISHED) {
log_message(LOG_LEVEL_ERROR, "Did not receive FINISHED Status"); log_message(LOG_LEVEL_ERROR, "Did not receive FINISHED Status");
send_status(file_descriptor, STATUS_ERROR); send_status(fd, STATUS_ERROR);
return -1; return -1;
} }
send_status(file_descriptor, STATUS_OK); send_status(fd, STATUS_OK);
return 0; return 0;
} }
+4
View File
@@ -38,6 +38,7 @@ Config *config_create(char *version, char *send_directory,
config->include_count = 0; config->include_count = 0;
config->max_size = 0; config->max_size = 0;
config->min_size = 0; config->min_size = 0;
config->use_incremental = false;
return config; return config;
} }
@@ -89,6 +90,7 @@ bool config_send(int file_descriptor, Config *config) {
if (!send_int(file_descriptor, (int)config->chunk_size)) return false; if (!send_int(file_descriptor, (int)config->chunk_size)) return false;
if (!send_int(file_descriptor, config->use_sendfile)) return false; if (!send_int(file_descriptor, config->use_sendfile)) return false;
if (!send_int(file_descriptor, config->use_delete)) return false; if (!send_int(file_descriptor, config->use_delete)) return false;
if (!send_int(file_descriptor, config->use_incremental)) return false;
Status status; Status status;
if (!receive_status(file_descriptor, &status)) return false; if (!receive_status(file_descriptor, &status)) return false;
if (status != STATUS_OK) { if (status != STATUS_OK) {
@@ -134,6 +136,8 @@ Config *config_receive(int file_descriptor) {
config->use_sendfile = tmp; config->use_sendfile = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error; if (!receive_int(file_descriptor, &tmp)) goto error;
config->use_delete = tmp; config->use_delete = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error;
config->use_incremental = tmp;
config->show_progress = false; config->show_progress = false;
config->dry_run = false; config->dry_run = false;
config->ssh_port = 22; config->ssh_port = 22;
+2 -1
View File
@@ -32,9 +32,10 @@ typedef struct Config {
int include_count; int include_count;
unsigned long long max_size; unsigned long long max_size;
unsigned long long min_size; unsigned long long min_size;
bool use_incremental;
} Config; } Config;
#define PROTOCOL_VERSION "1.0.0" #define PROTOCOL_VERSION "1.1.0"
#define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024) #define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024)
Config *config_create(char *version, char *send_directory, Config *config_create(char *version, char *send_directory,
+143 -11
View File
@@ -96,26 +96,156 @@ bool file_load_data(File *file) {
return true; return true;
} }
bool file_send_single_calls(File *file, int file_descriptor, bool use_metadata, int compression_level) { bool file_send_single_calls_no_path(File *file, int file_descriptor, bool use_metadata, int compression_level) {
Data *data_to_send = file->data;
Data *compressed_data = NULL;
if (compression_level > 0) { if (compression_level > 0) {
Data *compressed_data = data_compress(file->data, compression_level); compressed_data = data_compress(file->data, compression_level);
if (compressed_data == NULL) { if (compressed_data == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to compress file data"); log_message(LOG_LEVEL_ERROR, "Failed to compress file data");
return false; return false;
} }
data_destroy(file->data); data_to_send = compressed_data;
if (compressed_data == NULL) {
log_message(LOG_LEVEL_ERROR, "Compression failed in file_send_single_calls");
exit(EXIT_FAILURE);
} }
file->data = compressed_data; if (use_metadata && !metadata_send(file_descriptor, file->metadata)) {
data_destroy(compressed_data);
return false;
} }
if (!send_str(file_descriptor, file->path)) return false; if (!send_data(file_descriptor, data_to_send)) {
if (use_metadata && !metadata_send(file_descriptor, file->metadata)) return false; data_destroy(compressed_data);
if (!send_data(file_descriptor, file->data)) return false; return false;
}
data_destroy(compressed_data);
return true; return true;
} }
bool file_send_sendfile_no_path(File *file, int file_descriptor, bool use_metadata) {
if (use_metadata && !metadata_send(file_descriptor, file->metadata)) return false;
int fd = open(file->path, O_RDONLY);
if (fd == -1) {
perror("Could not open file for sendfile");
return false;
}
unsigned long long file_size = file->data->size;
if (!send_n_data(file_descriptor, &file_size, sizeof(unsigned long long))) {
close(fd);
return false;
}
off_t offset = 0;
while (offset < file_size) {
ssize_t sent = sendfile(file_descriptor, fd, &offset, file_size - offset);
if (sent == -1) {
perror("sendfile failed");
close(fd);
return false;
}
}
close(fd);
return true;
}
bool file_send_single_calls(File *file, int file_descriptor, bool use_metadata, int compression_level) {
Data *data_to_send = file->data;
Data *compressed_data = NULL;
if (compression_level > 0) {
compressed_data = data_compress(file->data, compression_level);
if (compressed_data == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to compress file data");
return false;
}
data_to_send = compressed_data;
}
if (!send_str(file_descriptor, file->path)) {
data_destroy(compressed_data);
return false;
}
if (use_metadata && !metadata_send(file_descriptor, file->metadata)) {
data_destroy(compressed_data);
return false;
}
if (!send_data(file_descriptor, data_to_send)) {
data_destroy(compressed_data);
return false;
}
data_destroy(compressed_data);
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) { bool to_disk(const char *path, const void *data, unsigned long long data_size) {
char *directory = str_dup(path); char *directory = str_dup(path);
char *dir_to_free = directory; char *dir_to_free = directory;
@@ -178,7 +308,9 @@ File *file_receive(Config *config, int file_descriptor) {
free(path); free(path);
if (file == NULL) return NULL; if (file == NULL) return NULL;
if (config->use_metadata) { if (config->use_metadata) {
file->metadata = metadata_receive(file_descriptor); int meta_ok = 1;
file->metadata = metadata_receive(file_descriptor, &meta_ok);
if (!meta_ok) { file_destroy(file); return NULL; }
} }
Data *file_data = receive_data(file_descriptor); Data *file_data = receive_data(file_descriptor);
if (file_data == NULL) { if (file_data == NULL) {
+4
View File
@@ -25,10 +25,14 @@ void file_destroy(void *item);
bool file_load_data(File *file); bool file_load_data(File *file);
File *file_receive(Config *config, int file_descriptor); File *file_receive(Config *config, int file_descriptor);
bool file_send_single_calls(File *file, int file_descriptor, bool use_metadata, int compression_level); bool file_send_single_calls(File *file, int file_descriptor, bool use_metadata, int compression_level);
bool file_send_single_calls_no_path(File *file, int file_descriptor, bool use_metadata, int compression_level);
bool file_send_sendfile(File *file, int file_descriptor, bool use_metadata); bool file_send_sendfile(File *file, int file_descriptor, bool use_metadata);
bool file_send_sendfile_no_path(File *file, int file_descriptor, bool use_metadata);
size_t file_content_to_buffer(File *file); size_t file_content_to_buffer(File *file);
FileMetadata *file_metadata_create(struct stat *stats); FileMetadata *file_metadata_create(struct stat *stats);
void file_metadata_destroy(void *metadata); void file_metadata_destroy(void *metadata);
bool to_disk(const char *path, const void *data, unsigned long long data_size); 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 #endif
+10 -4
View File
@@ -50,22 +50,28 @@ bool metadata_send(int file_descriptor, FileMetadata *m) {
send_n_data(file_descriptor, &m->mtime_nsec, sizeof(long)); send_n_data(file_descriptor, &m->mtime_nsec, sizeof(long));
} }
FileMetadata *metadata_receive(int file_descriptor) { FileMetadata *metadata_receive(int file_descriptor, int *ok) {
int present; int present;
if (!receive_n_data(file_descriptor, &present, sizeof(int))) if (!receive_n_data(file_descriptor, &present, sizeof(int))) {
if (ok) *ok = 0;
return NULL; return NULL;
if (!present) }
if (!present) {
if (ok) *ok = 1;
return NULL; return NULL;
}
FileMetadata *m = malloc(sizeof(FileMetadata)); FileMetadata *m = malloc(sizeof(FileMetadata));
if (m == NULL) return NULL; if (m == NULL) { if (ok) *ok = 0; return NULL; }
if (!receive_n_data(file_descriptor, &m->mode, sizeof(mode_t)) || if (!receive_n_data(file_descriptor, &m->mode, sizeof(mode_t)) ||
!receive_n_data(file_descriptor, &m->uid, sizeof(uid_t)) || !receive_n_data(file_descriptor, &m->uid, sizeof(uid_t)) ||
!receive_n_data(file_descriptor, &m->gid, sizeof(gid_t)) || !receive_n_data(file_descriptor, &m->gid, sizeof(gid_t)) ||
!receive_n_data(file_descriptor, &m->mtime_sec, sizeof(time_t)) || !receive_n_data(file_descriptor, &m->mtime_sec, sizeof(time_t)) ||
!receive_n_data(file_descriptor, &m->mtime_nsec, sizeof(long))) { !receive_n_data(file_descriptor, &m->mtime_nsec, sizeof(long))) {
free(m); free(m);
if (ok) *ok = 0;
return NULL; return NULL;
} }
if (ok) *ok = 1;
return m; return m;
} }
+1 -1
View File
@@ -10,7 +10,7 @@
void metadata_to_buf(char **buf, FileMetadata *m); void metadata_to_buf(char **buf, FileMetadata *m);
FileMetadata *metadata_from_buf(char **buf); FileMetadata *metadata_from_buf(char **buf);
bool metadata_send(int file_descriptor, FileMetadata *m); bool metadata_send(int file_descriptor, FileMetadata *m);
FileMetadata *metadata_receive(int file_descriptor); FileMetadata *metadata_receive(int file_descriptor, int *ok);
void file_restore_metadata(const char *path, FileMetadata *metadata); void file_restore_metadata(const char *path, FileMetadata *metadata);
#endif #endif
+13 -11
View File
@@ -6,7 +6,6 @@
#include "data.h" #include "data.h"
#include "file.h" #include "file.h"
#include "log.h" #include "log.h"
#include "metadata.h"
#include "protocol.h" #include "protocol.h"
#include "queue.h" #include "queue.h"
#include "utils.h" #include "utils.h"
@@ -126,8 +125,17 @@ int receive_thread(void *pipeline_context) {
Status status; Status status;
if (!receive_status(file_descriptor, &status)) return thrd_error; if (!receive_status(file_descriptor, &status)) return thrd_error;
while (status == STATUS_NEXT || status == STATUS_CHUNK) { while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK) {
if (status == STATUS_CHUNK) { if (status == STATUS_CHECK) {
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);
}
} else if (status == STATUS_CHUNK) {
receive_chunk_enqueue(file_descriptor, context); receive_chunk_enqueue(file_descriptor, context);
} else { } else {
File *file = file_receive(config, file_descriptor); File *file = file_receive(config, file_descriptor);
@@ -182,14 +190,8 @@ int write_thread(void *pipeline_context) {
free(root_directory); free(root_directory);
return thrd_success; return thrd_success;
} }
if (save_to_disk) { if (save_to_disk)
char *disk_path = path_cat(root_directory, file->path); file_save_to_disk(root_directory, file);
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); file_destroy(file);
} }
} }
+48 -1
View File
@@ -1,18 +1,59 @@
#include "protocol.h" #include "protocol.h"
#include "log.h" #include "log.h"
#include <errno.h>
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <time.h>
#include <unistd.h> #include <unistd.h>
static __thread int io_read_fd = -1; static __thread int io_read_fd = -1;
static __thread int io_write_fd = -1; static __thread int io_write_fd = -1;
static unsigned long long io_bwlimit = 0;
static long long bw_tokens = 0;
static struct timespec bw_last_refill = {0, 0};
void io_set_fds(int read_fd, int write_fd) { void io_set_fds(int read_fd, int write_fd) {
io_read_fd = read_fd; io_read_fd = read_fd;
io_write_fd = write_fd; io_write_fd = write_fd;
} }
void io_set_bwlimit(unsigned long long bytes_per_sec) {
io_bwlimit = bytes_per_sec;
bw_tokens = (long long)io_bwlimit;
clock_gettime(CLOCK_MONOTONIC, &bw_last_refill);
}
static void bw_throttle(size_t bytes_written) {
if (io_bwlimit == 0) return;
struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
long long elapsed_ns = (now.tv_sec - bw_last_refill.tv_sec) * 1000000000LL +
(now.tv_nsec - bw_last_refill.tv_nsec);
bw_last_refill = now;
long long tokens_to_add = (long long)((double)io_bwlimit * elapsed_ns / 1000000000.0);
bw_tokens += tokens_to_add;
if (bw_tokens > (long long)io_bwlimit)
bw_tokens = (long long)io_bwlimit;
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;
bw_tokens = 0;
clock_gettime(CLOCK_MONOTONIC, &bw_last_refill);
}
}
static int io_fd(int dir_fd, int file_descriptor) { static int io_fd(int dir_fd, int file_descriptor) {
return (dir_fd != -1) ? dir_fd : file_descriptor; return (dir_fd != -1) ? dir_fd : file_descriptor;
} }
@@ -22,12 +63,16 @@ bool send_n_data(int file_descriptor, void *data, size_t data_size) {
int fd = io_fd(io_write_fd, file_descriptor); int fd = io_fd(io_write_fd, file_descriptor);
ssize_t total_bytes_send = 0; ssize_t total_bytes_send = 0;
while (total_bytes_send < data_size) { while (total_bytes_send < data_size) {
size_t chunk = data_size - total_bytes_send;
if (io_bwlimit > 0 && chunk > 65536)
chunk = 65536;
ssize_t bytes_send = ssize_t bytes_send =
write(fd, (char *)data + total_bytes_send, data_size - total_bytes_send); write(fd, (char *)data + total_bytes_send, chunk);
if (bytes_send <= 0) { if (bytes_send <= 0) {
log_message(LOG_LEVEL_ERROR, "Could not send data"); log_message(LOG_LEVEL_ERROR, "Could not send data");
return false; return false;
} }
bw_throttle((size_t)bytes_send);
total_bytes_send += bytes_send; total_bytes_send += bytes_send;
} }
log_message(LOG_LEVEL_DEBUG, " Send n Data: %zu", total_bytes_send); log_message(LOG_LEVEL_DEBUG, " Send n Data: %zu", total_bytes_send);
@@ -66,6 +111,8 @@ static const char *status_to_string(Status status) {
return "NEXT"; return "NEXT";
case STATUS_CHUNK: case STATUS_CHUNK:
return "CHUNK"; return "CHUNK";
case STATUS_CHECK:
return "CHECK";
default: default:
return "UNKNOWN"; return "UNKNOWN";
} }
+2 -1
View File
@@ -6,9 +6,10 @@
#include <stddef.h> #include <stddef.h>
typedef int Status; typedef int Status;
enum NET_STATUS { STATUS_OK, STATUS_ERROR, STATUS_FINISHED, STATUS_NEXT, STATUS_CHUNK, STATUS_MANIFEST }; enum NET_STATUS { STATUS_OK, STATUS_ERROR, STATUS_FINISHED, STATUS_NEXT, STATUS_CHUNK, STATUS_MANIFEST, STATUS_CHECK };
void io_set_fds(int read_fd, int write_fd); void io_set_fds(int read_fd, int write_fd);
void io_set_bwlimit(unsigned long long bytes_per_sec);
bool send_n_data(int file_descriptor, void *data, size_t data_size); bool send_n_data(int file_descriptor, void *data, size_t data_size);
bool receive_n_data(int file_descriptor, void *data, size_t data_size); bool receive_n_data(int file_descriptor, void *data, size_t data_size);
+27
View File
@@ -383,6 +383,33 @@ def run_profile(profile_name, source_dir, dest_dir):
except Exception as e: except Exception as e:
results.append({"name": "Progress (--progress)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)}) results.append({"name": "Progress (--progress)", "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:
flags = BASE_CLIENT_FLAGS + ["-M"]
srv = subprocess.Popen(SERVER_CMD, stdout=subprocess.DEVNULL, stderr=None)
time.sleep(0.5)
first_cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + flags
r1 = subprocess.run(first_cmd, text=True, capture_output=True)
wait_proc(srv)
if r1.returncode != 0:
raise RuntimeError(f"First sync failed: {r1.stderr[:100]}")
srv2 = subprocess.Popen(SERVER_CMD, stdout=subprocess.DEVNULL, stderr=None)
time.sleep(0.5)
second_cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + flags + ["--incremental"]
start = time.monotonic()
r2 = subprocess.run(second_cmd, text=True, capture_output=True, timeout=30)
duration = time.monotonic() - start
wait_proc(srv2)
r = {"name": "Incremental (--incremental)", "suite": profile_name,
"status": "Success" if r2.returncode == 0 else "Failed",
"time": f"{duration:.4f}s" if r2.returncode == 0 else "N/A",
"error": "" if r2.returncode == 0 else f"Exit {r2.returncode}: {(r2.stderr or r2.stdout)[:60]}"}
results.append(r)
except Exception as e:
results.append({"name": "Incremental (--incremental)", "suite": profile_name,
"status": "Error", "time": "N/A", "error": str(e)})
# Chunk size (--chunk-size 5242880) # Chunk size (--chunk-size 5242880)
feature_flags = BASE_CLIENT_FLAGS + ["--chunk-size", "5242880"] feature_flags = BASE_CLIENT_FLAGS + ["--chunk-size", "5242880"]
cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + feature_flags cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + feature_flags