TapTap/FastSync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: resolve all bug issues (#158, #163, #164, #165, #166, #167, #168, #169)

Browse Source
This commit is contained in:
TapTap
2026-07-29 19:02:09 +02:00
parent 269ce0749b
commit 4f383cc5a5
21 changed files with 1498 additions and 385 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client/client_cli.c
+386 -267
View File
@@ -12,6 +12,341 @@
#include <stdlib.h>
#include <string.h>
/* Parse environment variables for source/destination directories and save-to-disk flag. */
static void parse_environment(const char** out_env_source, const char** out_env_dest,
bool* out_save_to_disk) {
*out_env_source = getenv("FASTSYNC_SOURCE_DIR");
*out_env_dest = getenv("FASTSYNC_DEST_DIR");
const char* env_save = getenv("FASTSYNC_SAVE_TO_DISK");
*out_save_to_disk = false;
if (env_save && (strcmp(env_save, "true") == 0 || strcmp(env_save, "1") == 0))
*out_save_to_disk = true;
}
/* Parse a string as a positive integer, returning true on success. */
static bool parse_positive_int(const char* s, int* out_val) {
if (!s || *s == '\0')
return false;
char* endptr;
errno = 0;
long val = strtol(s, &endptr, 10);
if (errno != 0 || *endptr != '\0' || val <= 0 || val > INT_MAX)
return false;
*out_val = (int)val;
return true;
}
/* Parse a string as a non-negative integer, returning true on success. */
static bool parse_nonneg_int(const char* s, int* out_val) {
if (!s || *s == '\0')
return false;
char* endptr;
errno = 0;
long val = strtol(s, &endptr, 10);
if (errno != 0 || *endptr != '\0' || val < 0 || val > INT_MAX)
return false;
*out_val = (int)val;
return true;
}
static void print_usage(void);
static int read_patterns_from_file(const char* filepath, char*** patterns, int* count);
/* Parse CLI arguments into config. Returns 0 on success, -1 on error, 1 for help/clean-exit. */
static int parse_args(Config* config, int argc, char* argv[], int* positional_args,
int* positional_count) {
for (int i = 1; i < argc; i++) {
if (strcmp(argv[i], "--help") == 0) {
print_usage();
return 1;
} else if (strcmp(argv[i], "-a") == 0 || strcmp(argv[i], "--archive") == 0) {
config->use_compression = true;
config->use_multithreading = true;
config->use_metadata = true;
log_message(LOG_LEVEL_INFO, "Enabled archive mode (-c -m -M)");
} else if (strcmp(argv[i], "-n") == 0 || strcmp(argv[i], "--dry-run") == 0) {
config->dry_run = true;
} else if (strcmp(argv[i], "-p") == 0 && i + 1 < argc) {
if (!parse_positive_int(argv[++i], &config->ssh_port)) {
fprintf(stderr, "Error: invalid --port/-p value: %s\n", argv[i]);
return -1;
}
} else if (strcmp(argv[i], "--delete") == 0) {
config->use_delete = true;
} else if (strcmp(argv[i], "--exclude") == 0 && i + 1 < argc) {
char** tmp = realloc(config->exclude_patterns, (config->exclude_count + 1) * sizeof(char*));
if (!tmp) {
fprintf(stderr, "Error: memory allocation failed for --exclude\n");
return -1;
}
config->exclude_patterns = tmp;
char* dup = str_dup(argv[++i]);
if (!dup) {
fprintf(stderr, "Error: memory allocation failed for --exclude\n");
return -1;
}
config->exclude_patterns[config->exclude_count++] = dup;
} else if (strcmp(argv[i], "--include") == 0 && i + 1 < argc) {
char** tmp = realloc(config->include_patterns, (config->include_count + 1) * sizeof(char*));
if (!tmp) {
fprintf(stderr, "Error: memory allocation failed for --include\n");
return -1;
}
config->include_patterns = tmp;
char* dup = str_dup(argv[++i]);
if (!dup) {
fprintf(stderr, "Error: memory allocation failed for --include\n");
return -1;
}
config->include_patterns[config->include_count++] = dup;
} else if (strcmp(argv[i], "--max-size") == 0 && i + 1 < argc) {
config->max_size = strtoull(argv[++i], NULL, 10);
} else if (strcmp(argv[i], "--min-size") == 0 && i + 1 < argc) {
config->min_size = strtoull(argv[++i], NULL, 10);
} else if (strcmp(argv[i], "--incremental") == 0) {
config->use_incremental = true;
} else if (strcmp(argv[i], "--delta") == 0) {
config->use_delta = true;
} else if (strcmp(argv[i], "--delta-block") == 0 && i + 1 < argc) {
unsigned long long val = strtoull(argv[++i], NULL, 10);
if (val >= DELTA_BLOCK_SIZE_MIN && val <= DELTA_BLOCK_SIZE_MAX)
config->delta_block_size = (uint32_t)val;
else
fprintf(stderr, "Warning: --delta-block value %llu out of range, using default\n", val);
} else if (strcmp(argv[i], "--delta-max") == 0 && i + 1 < argc) {
unsigned long long val = strtoull(argv[++i], NULL, 10);
if (val >= DELTA_MIN_FILE_SIZE)
config->delta_max_file_size = val;
else
fprintf(stderr, "Warning: --delta-max value %llu too small, using default\n", val);
} else if (strcmp(argv[i], "-c") == 0 || strcmp(argv[i], "-z") == 0) {
config->use_compression = true;
log_message(LOG_LEVEL_INFO, "Enabled Compression");
if (i + 1 < argc) {
char* end_ptr;
long level = strtol(argv[i + 1], &end_ptr, 10);
if (*end_ptr == '\0') {
config->compression_level = (int)level;
log_message(LOG_LEVEL_INFO, "Set Compression level to %ld", level);
i++;
}
}
} else if (strcmp(argv[i], "--source-dir") == 0 && i + 1 < argc) {
char* dup = str_dup(argv[++i]);
if (!dup) {
fprintf(stderr, "Error: memory allocation failed for --source-dir\n");
return -1;
}
free(config->send_directory);
config->send_directory = dup;
} else if (strcmp(argv[i], "--dest-dir") == 0 && i + 1 < argc) {
char* dup = str_dup(argv[++i]);
if (!dup) {
fprintf(stderr, "Error: memory allocation failed for --dest-dir\n");
return -1;
}
free(config->receive_root_directory);
config->receive_root_directory = dup;
} else if (strcmp(argv[i], "--save-to-disk") == 0) {
config->save_to_disk = true;
} else if (strcmp(argv[i], "-M") == 0 || strcmp(argv[i], "--preserve") == 0) {
config->use_metadata = true;
log_message(LOG_LEVEL_INFO, "Enabled metadata preservation");
} else if (strcmp(argv[i], "-f") == 0 || strcmp(argv[i], "--sendfile") == 0) {
config->use_sendfile = true;
log_message(LOG_LEVEL_INFO, "Enabled sendfile");
} else if (strcmp(argv[i], "-m") == 0) {
config->use_multithreading = true;
log_message(LOG_LEVEL_INFO, "Enabled Multithreading");
} else if (strcmp(argv[i], "-s") == 0) {
config->use_chunk_serialization = true;
log_message(LOG_LEVEL_INFO, "Enabled Chunk Serialization");
} else if (strcmp(argv[i], "--server-host") == 0 && i + 1 < argc) {
char* dup = str_dup(argv[++i]);
if (!dup) {
fprintf(stderr, "Error: memory allocation failed for --server-host\n");
return -1;
}
free(config->server_host);
config->server_host = dup;
} else if (strcmp(argv[i], "--server-port") == 0 && i + 1 < argc) {
if (!parse_positive_int(argv[++i], &config->server_port)) {
fprintf(stderr, "Error: invalid --server-port value: %s\n", argv[i]);
return -1;
}
} 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) {
config->show_progress = true;
} else if (strcmp(argv[i], "--chunk-size") == 0 && i + 1 < argc) {
unsigned long long val = strtoull(argv[++i], NULL, 10);
if (val > 0)
config->chunk_size = val;
} else if (strcmp(argv[i], "--tls") == 0) {
config->use_tls = true;
} else if (strcmp(argv[i], "--cert") == 0 && i + 1 < argc) {
char* dup = str_dup(argv[++i]);
if (!dup) {
fprintf(stderr, "Error: memory allocation failed for --cert\n");
return -1;
}
free(config->tls_cert);
config->tls_cert = dup;
} else if (strcmp(argv[i], "--key") == 0 && i + 1 < argc) {
char* dup = str_dup(argv[++i]);
if (!dup) {
fprintf(stderr, "Error: memory allocation failed for --key\n");
return -1;
}
free(config->tls_key);
config->tls_key = dup;
} else if (strcmp(argv[i], "--ca") == 0 && i + 1 < argc) {
char* dup = str_dup(argv[++i]);
if (!dup) {
fprintf(stderr, "Error: memory allocation failed for --ca\n");
return -1;
}
free(config->tls_ca);
config->tls_ca = dup;
} else if (strcmp(argv[i], "--timeout") == 0 && i + 1 < argc) {
int val;
if (!parse_positive_int(argv[++i], &val)) {
fprintf(stderr, "Error: --timeout must be a positive integer\n");
return -1;
}
config->timeout = val;
} else if (strcmp(argv[i], "--contimeout") == 0 && i + 1 < argc) {
int val;
if (!parse_positive_int(argv[++i], &val)) {
fprintf(stderr, "Error: --contimeout must be a positive integer\n");
return -1;
}
config->contimeout = val;
} else if (strcmp(argv[i], "-q") == 0 || strcmp(argv[i], "--quiet") == 0 ||
strcmp(argv[i], "--silent") == 0) {
config->quiet = true;
} else if (strcmp(argv[i], "--backup") == 0) {
config->backup = true;
} else if (strcmp(argv[i], "--backup-dir") == 0 && i + 1 < argc) {
char* dup = str_dup(argv[++i]);
if (!dup) {
fprintf(stderr, "Error: memory allocation failed for --backup-dir\n");
return -1;
}
free(config->backup_dir);
config->backup_dir = dup;
} else if (strcmp(argv[i], "--stats") == 0) {
config->stats = true;
} else if (strcmp(argv[i], "--max-depth") == 0 && i + 1 < argc) {
if (!parse_nonneg_int(argv[++i], &config->max_depth)) {
fprintf(stderr, "Error: --max-depth must be a non-negative integer\n");
return -1;
}
} else if (strcmp(argv[i], "--log-file") == 0 && i + 1 < argc) {
FILE* lf = fopen(argv[++i], "a");
if (!lf) {
fprintf(stderr, "Error: could not open log file '%s': %s\n", argv[i], strerror(errno));
return -1;
}
config->log_file = lf;
log_set_file(lf);
} else if (strcmp(argv[i], "--queue-size") == 0 && i + 1 < argc) {
int val;
if (!parse_positive_int(argv[++i], &val)) {
fprintf(stderr, "Error: --queue-size must be a positive integer\n");
return -1;
}
config->queue_size = val;
} else if (strcmp(argv[i], "--exclude-from") == 0 && i + 1 < argc) {
if (read_patterns_from_file(argv[++i], &config->exclude_patterns, &config->exclude_count) !=
0)
return -1;
} else if (strcmp(argv[i], "--include-from") == 0 && i + 1 < argc) {
if (read_patterns_from_file(argv[++i], &config->include_patterns, &config->include_count) !=
0)
return -1;
} else if (strcmp(argv[i], "--partial") == 0) {
config->partial = true;
} else if (strcmp(argv[i], "--fastsync-server-path") == 0 && i + 1 < argc) {
char* dup = str_dup(argv[++i]);
if (!dup) {
fprintf(stderr, "Error: memory allocation failed for --fastsync-server-path\n");
return -1;
}
free(config->fastsync_server_path);
config->fastsync_server_path = dup;
} else if (strcmp(argv[i], "-v") == 0 || strcmp(argv[i], "--verbose") == 0) {
set_log_level(LOG_LEVEL_DEBUG);
} else if (argv[i][0] == '-') {
fprintf(stderr, "Unknown option: %s\n", argv[i]);
print_usage();
return -1;
} else {
if (*positional_count < 2)
positional_args[(*positional_count)++] = i;
else {
fprintf(stderr, "Unexpected argument: %s\n", argv[i]);
print_usage();
return -1;
}
}
}
return 0;
}
/* Validate config after parsing. Returns true if valid. */
static bool validate_config(const Config* config) {
if (!config->send_directory || !config->receive_root_directory) {
fprintf(stderr, "Error: source and destination directories are required\n");
print_usage();
return false;
}
if (config->use_sendfile && (config->use_chunk_serialization || config->use_compression)) {
fprintf(stderr, "Error: -f/--sendfile cannot be combined with -c (compression) or -s (chunk "
"serialization)\n");
return false;
}
if (config->transport == TRANSPORT_SSH && config->use_sendfile) {
fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n");
return false;
}
if (config->use_incremental && config->use_chunk_serialization) {
fprintf(stderr, "Error: --incremental is not supported with -s (chunk serialization)\n");
return false;
}
if (config->use_delta && !config->use_incremental) {
fprintf(stderr, "Error: --delta requires --incremental\n");
return false;
}
if (config->use_delta && config->use_chunk_serialization) {
fprintf(stderr, "Error: --delta cannot be combined with -s (chunk serialization)\n");
return false;
}
if (config->use_delta && config->use_sendfile) {
fprintf(stderr, "Error: --delta cannot be combined with -f (sendfile)\n");
return false;
}
if (config->use_tls) {
if (!config->tls_cert || !config->tls_key) {
fprintf(stderr, "Error: --tls requires --cert and --key\n");
return false;
}
}
return true;
}
static void print_usage(void) {
printf("Usage:\n");
printf(" fastsync [options] <source> <destination>\n");
@@ -99,249 +434,60 @@ static int read_patterns_from_file(const char* filepath, char*** patterns, int*
return -1;
}
*patterns = tmp;
(*patterns)[(*count)++] = str_dup(p);
char* dup = str_dup(p);
if (!dup) {
fprintf(stderr, "Error: memory allocation failed for pattern file\n");
fclose(fp);
return -1;
}
(*patterns)[(*count)++] = dup;
}
fclose(fp);
return 0;
}
int main(int argc, char* argv[]) {
const char* env_source = getenv("FASTSYNC_SOURCE_DIR");
const char* env_dest = getenv("FASTSYNC_DEST_DIR");
const char* env_save = getenv("FASTSYNC_SAVE_TO_DISK");
const char* env_source = NULL;
const char* env_dest = NULL;
bool save_to_disk = false;
if (env_save && (strcmp(env_save, "true") == 0 || strcmp(env_save, "1") == 0)) {
save_to_disk = true;
}
parse_environment(&env_source, &env_dest, &save_to_disk);
int exit_code = 0;
Config* config = NULL;
bool config_owned_by_pipeline = false;
char* config_version = str_dup(PROTOCOL_VERSION);
if (!config_version) {
exit_code = 1;
goto cleanup;
Config* config = config_create();
if (!config) {
fprintf(stderr, "Error: failed to allocate config\n");
return 1;
}
config = config_create(config_version, NULL, NULL, save_to_disk, false, false, false, false, 5,
false, 0);
config->save_to_disk = save_to_disk;
int positional_args[2];
int positional_count = 0;
for (int i = 1; i < argc; i++) {
if (strcmp(argv[i], "--help") == 0) {
print_usage();
goto cleanup;
} else if (strcmp(argv[i], "-a") == 0 || strcmp(argv[i], "--archive") == 0) {
config->use_compression = true;
config->use_multithreading = true;
config->use_metadata = true;
log_message(LOG_LEVEL_INFO, "Enabled archive mode (-c -m -M)");
} else if (strcmp(argv[i], "-n") == 0 || strcmp(argv[i], "--dry-run") == 0) {
config->dry_run = true;
} else if (strcmp(argv[i], "-p") == 0 && i + 1 < argc) {
config->ssh_port = atoi(argv[++i]);
} else if (strcmp(argv[i], "--delete") == 0) {
config->use_delete = true;
} else if (strcmp(argv[i], "--exclude") == 0 && i + 1 < argc) {
char** tmp = realloc(config->exclude_patterns, (config->exclude_count + 1) * sizeof(char*));
if (!tmp) {
fprintf(stderr, "Error: memory allocation failed for --exclude\n");
exit_code = 1;
goto cleanup;
}
config->exclude_patterns = tmp;
config->exclude_patterns[config->exclude_count++] = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--include") == 0 && i + 1 < argc) {
char** tmp = realloc(config->include_patterns, (config->include_count + 1) * sizeof(char*));
if (!tmp) {
fprintf(stderr, "Error: memory allocation failed for --include\n");
exit_code = 1;
goto cleanup;
}
config->include_patterns = tmp;
config->include_patterns[config->include_count++] = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--max-size") == 0 && i + 1 < argc) {
config->max_size = strtoull(argv[++i], NULL, 10);
} else if (strcmp(argv[i], "--min-size") == 0 && i + 1 < argc) {
config->min_size = strtoull(argv[++i], NULL, 10);
} else if (strcmp(argv[i], "--incremental") == 0) {
config->use_incremental = true;
} else if (strcmp(argv[i], "--delta") == 0) {
config->use_delta = true;
} else if (strcmp(argv[i], "--delta-block") == 0 && i + 1 < argc) {
unsigned long long val = strtoull(argv[++i], NULL, 10);
if (val >= DELTA_BLOCK_SIZE_MIN && val <= DELTA_BLOCK_SIZE_MAX)
config->delta_block_size = (uint32_t)val;
else
fprintf(stderr, "Warning: --delta-block value %llu out of range, using default\n", val);
} else if (strcmp(argv[i], "--delta-max") == 0 && i + 1 < argc) {
unsigned long long val = strtoull(argv[++i], NULL, 10);
if (val >= DELTA_MIN_FILE_SIZE)
config->delta_max_file_size = val;
else
fprintf(stderr, "Warning: --delta-max value %llu too small, using default\n", val);
} else if (strcmp(argv[i], "-c") == 0 || strcmp(argv[i], "-z") == 0) {
config->use_compression = true;
log_message(LOG_LEVEL_INFO, "Enabled Compression");
if (i + 1 < argc) {
char* end_ptr;
int level = strtol(argv[i + 1], &end_ptr, 10);
if (*end_ptr == '\0') {
config->compression_level = level;
log_message(LOG_LEVEL_INFO, "Set Compression level to %d", config->compression_level);
i++;
}
}
} else if (strcmp(argv[i], "--source-dir") == 0 && i + 1 < argc) {
free(config->send_directory);
config->send_directory = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--dest-dir") == 0 && i + 1 < argc) {
free(config->receive_root_directory);
config->receive_root_directory = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--save-to-disk") == 0) {
config->save_to_disk = true;
} else if (strcmp(argv[i], "-M") == 0 || strcmp(argv[i], "--preserve") == 0) {
config->use_metadata = true;
log_message(LOG_LEVEL_INFO, "Enabled metadata preservation");
} else if (strcmp(argv[i], "-f") == 0 || strcmp(argv[i], "--sendfile") == 0) {
config->use_sendfile = true;
log_message(LOG_LEVEL_INFO, "Enabled sendfile");
} else if (strcmp(argv[i], "-m") == 0) {
config->use_multithreading = true;
log_message(LOG_LEVEL_INFO, "Enabled Multithreading");
} else if (strcmp(argv[i], "-s") == 0) {
config->use_chunk_serialization = true;
log_message(LOG_LEVEL_INFO, "Enabled Chunk Serialization");
} else if (strcmp(argv[i], "--server-host") == 0 && i + 1 < argc) {
free(config->server_host);
config->server_host = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--server-port") == 0 && i + 1 < argc) {
config->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");
exit_code = 1;
goto cleanup;
}
if (kbps > ULLONG_MAX / 1024) {
fprintf(stderr, "Error: --bwlimit value too large\n");
exit_code = 1;
goto cleanup;
}
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) {
config->show_progress = true;
} else if (strcmp(argv[i], "--chunk-size") == 0 && i + 1 < argc) {
unsigned long long val = strtoull(argv[++i], NULL, 10);
if (val > 0)
config->chunk_size = val;
} else if (strcmp(argv[i], "--tls") == 0) {
config->use_tls = true;
} else if (strcmp(argv[i], "--cert") == 0 && i + 1 < argc) {
free(config->tls_cert);
config->tls_cert = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--key") == 0 && i + 1 < argc) {
free(config->tls_key);
config->tls_key = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--ca") == 0 && i + 1 < argc) {
free(config->tls_ca);
config->tls_ca = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--timeout") == 0 && i + 1 < argc) {
config->timeout = atoi(argv[++i]);
if (config->timeout <= 0) {
fprintf(stderr, "Error: --timeout must be a positive integer\n");
exit_code = 1;
goto cleanup;
}
} else if (strcmp(argv[i], "--contimeout") == 0 && i + 1 < argc) {
config->contimeout = atoi(argv[++i]);
if (config->contimeout <= 0) {
fprintf(stderr, "Error: --contimeout must be a positive integer\n");
exit_code = 1;
goto cleanup;
}
} else if (strcmp(argv[i], "-q") == 0 || strcmp(argv[i], "--quiet") == 0 ||
strcmp(argv[i], "--silent") == 0) {
config->quiet = true;
} else if (strcmp(argv[i], "--backup") == 0) {
config->backup = true;
} else if (strcmp(argv[i], "--backup-dir") == 0 && i + 1 < argc) {
config->backup_dir = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--stats") == 0) {
config->stats = true;
} else if (strcmp(argv[i], "--max-depth") == 0 && i + 1 < argc) {
config->max_depth = atoi(argv[++i]);
if (config->max_depth < 0) {
fprintf(stderr, "Error: --max-depth must be a non-negative integer\n");
exit_code = 1;
goto cleanup;
}
} else if (strcmp(argv[i], "--log-file") == 0 && i + 1 < argc) {
FILE* lf = fopen(argv[++i], "a");
if (!lf) {
fprintf(stderr, "Error: could not open log file '%s': %s\n", argv[i], strerror(errno));
exit_code = 1;
goto cleanup;
}
config->log_file = lf;
log_set_file(lf);
} else if (strcmp(argv[i], "--queue-size") == 0 && i + 1 < argc) {
config->queue_size = atoi(argv[++i]);
if (config->queue_size <= 0) {
fprintf(stderr, "Error: --queue-size must be a positive integer\n");
exit_code = 1;
goto cleanup;
}
} else if (strcmp(argv[i], "--exclude-from") == 0 && i + 1 < argc) {
if (read_patterns_from_file(argv[++i], &config->exclude_patterns, &config->exclude_count) !=
0) {
exit_code = 1;
goto cleanup;
}
} else if (strcmp(argv[i], "--include-from") == 0 && i + 1 < argc) {
if (read_patterns_from_file(argv[++i], &config->include_patterns, &config->include_count) !=
0) {
exit_code = 1;
goto cleanup;
}
} else if (strcmp(argv[i], "--partial") == 0) {
config->partial = true;
} else if (strcmp(argv[i], "--fastsync-server-path") == 0 && i + 1 < argc) {
free(config->fastsync_server_path);
config->fastsync_server_path = str_dup(argv[++i]);
} else if (strcmp(argv[i], "-v") == 0 || strcmp(argv[i], "--verbose") == 0) {
set_log_level(LOG_LEVEL_DEBUG);
} else if (argv[i][0] == '-') {
fprintf(stderr, "Unknown option: %s\n", argv[i]);
print_usage();
int parse_ret = parse_args(config, argc, argv, positional_args, &positional_count);
if (parse_ret != 0) {
if (parse_ret < 0)
exit_code = 1;
goto cleanup;
} else {
if (positional_count < 2)
positional_args[positional_count++] = i;
else {
fprintf(stderr, "Unexpected argument: %s\n", argv[i]);
print_usage();
exit_code = 1;
goto cleanup;
}
}
goto cleanup;
}
/* Handle positional arguments or fall back to environment variables */
if (positional_count == 2) {
free(config->send_directory);
free(config->receive_root_directory);
config->send_directory = str_dup(argv[positional_args[0]]);
if (!config->send_directory) {
fprintf(stderr, "Error: memory allocation failed\n");
exit_code = 1;
goto cleanup;
}
config->receive_root_directory = str_dup(argv[positional_args[1]]);
if (!config->receive_root_directory) {
fprintf(stderr, "Error: memory allocation failed\n");
exit_code = 1;
goto cleanup;
}
config->save_to_disk = true;
config_parse_ssh_dest(config);
} else if (positional_count == 1) {
fprintf(stderr, "Error: missing destination argument\n");
@@ -349,73 +495,46 @@ int main(int argc, char* argv[]) {
exit_code = 1;
goto cleanup;
} else {
if (!config->send_directory && env_source)
config->send_directory = str_dup((char*)env_source);
if (!config->receive_root_directory && env_dest)
config->receive_root_directory = str_dup((char*)env_dest);
if (!config->send_directory && env_source) {
config->send_directory = str_dup(env_source);
if (!config->send_directory) {
fprintf(stderr, "Error: memory allocation failed\n");
exit_code = 1;
goto cleanup;
}
}
if (!config->receive_root_directory && env_dest) {
config->receive_root_directory = str_dup(env_dest);
if (!config->receive_root_directory) {
fprintf(stderr, "Error: memory allocation failed\n");
exit_code = 1;
goto cleanup;
}
}
}
if (!config->send_directory || !config->receive_root_directory) {
fprintf(stderr, "Error: source and destination directories are required\n");
print_usage();
exit_code = 1;
goto cleanup;
}
if (config->use_sendfile && (config->use_chunk_serialization || config->use_compression)) {
fprintf(stderr, "Error: -f/--sendfile cannot be combined with -c (compression) or -s (chunk "
"serialization)\n");
exit_code = 1;
goto cleanup;
}
if (config->transport == TRANSPORT_SSH && config->use_sendfile) {
fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n");
exit_code = 1;
goto cleanup;
}
if (config->use_incremental && config->use_chunk_serialization) {
fprintf(stderr, "Error: --incremental is not supported with -s (chunk serialization)\n");
if (!validate_config(config)) {
exit_code = 1;
goto cleanup;
}
/* Enable implicit flags */
if (config->use_incremental && !config->use_metadata) {
log_message(LOG_LEVEL_INFO, "Enabling metadata preservation for --incremental");
config->use_metadata = true;
}
if (config->use_delta && !config->use_incremental) {
fprintf(stderr, "Error: --delta requires --incremental\n");
exit_code = 1;
goto cleanup;
}
if (config->use_delta && config->use_chunk_serialization) {
fprintf(stderr, "Error: --delta cannot be combined with -s (chunk serialization)\n");
exit_code = 1;
goto cleanup;
}
if (config->use_delta && config->use_sendfile) {
fprintf(stderr, "Error: --delta cannot be combined with -f (sendfile)\n");
exit_code = 1;
goto cleanup;
}
if (config->use_delta && !config->use_metadata) {
log_message(LOG_LEVEL_INFO, "Enabling metadata preservation for --delta");
config->use_metadata = true;
}
if (config->use_tls) {
if (!config->tls_cert || !config->tls_key) {
fprintf(stderr, "Error: --tls requires --cert and --key\n");
exit_code = 1;
goto cleanup;
}
/* Initialize TLS if needed */
if (config->use_tls)
tls_global_init();
}
tcp_set_timeouts(config->timeout, config->contimeout);
/* Execute transfer */
if (config->use_multithreading) {
config_owned_by_pipeline = true;
exit_code = send_files_multithreaded(config);
src/client/client_send.c
+47 -63
View File
@@ -25,6 +25,44 @@
#define STREAM_THRESHOLD (64ULL * 1024 * 1024)
/* Print dry-run manifest showing files that would be transferred. Returns 0 on success. */
static int send_dry_run_manifest(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);
if (!scanner)
return -1;
Chunk* chunk;
int file_count = 0;
unsigned long long total_bytes = 0;
printf("Dry run: files to be transferred\n");
while ((chunk = directory_scanner_next(scanner)) != NULL) {
for (int i = 0; i < chunk->element_count; i++) {
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);
printf("Total: %d files, %.1f MB\n", file_count, total_bytes / 1048576.0);
return 0;
}
/* Send the delete manifest (list of files) to the server. Returns 0 on success, -1 on failure. */
static int send_delete_manifest(int fd, ArrayList* manifest) {
if (!send_status(fd, STATUS_MANIFEST))
return -1;
if (!send_int(fd, manifest->size))
return -1;
for (int i = 0; i < manifest->size; i++) {
if (!send_str(fd, (char*)manifest->items[i]))
return -1;
}
return 0;
}
static int incremental_check(Client* client, File* file, DeltaSignature** out_sig) {
*out_sig = NULL;
if (!send_status(client->file_descriptor, STATUS_CHECK))
@@ -268,14 +306,8 @@ static int send_chunks_multithreaded(void* pipeline_context) {
&context->condition_not_full_loader, &context->loader_done);
if (current_chunk == NULL) {
if (context->config->use_delete) {
if (!send_status(client->file_descriptor, STATUS_MANIFEST))
if (send_delete_manifest(client->file_descriptor, context->manifest) != 0)
goto send_fail;
if (!send_int(client->file_descriptor, context->manifest->size))
goto send_fail;
for (int i = 0; i < context->manifest->size; i++) {
if (!send_str(client->file_descriptor, (char*)context->manifest->items[i]))
goto send_fail;
}
}
if (!send_status(client->file_descriptor, STATUS_FINISHED))
goto send_fail;
@@ -365,27 +397,8 @@ static int load_files_multithreaded(void* pipeline_context) {
}
int send_files(Config* config) {
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);
Chunk* chunk;
int file_count = 0;
unsigned long long total_bytes = 0;
printf("Dry run: files to be transferred\n");
while ((chunk = directory_scanner_next(scanner)) != NULL) {
for (int i = 0; i < chunk->element_count; i++) {
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);
printf("Total: %d files, %.1f MB\n", file_count, total_bytes / 1048576.0);
return 0;
}
if (config->dry_run)
return send_dry_run_manifest(config);
Client* client;
if (config->transport == TRANSPORT_SSH) {
@@ -470,20 +483,10 @@ int send_files(Config* config) {
chunk_destroy(current_chunk);
}
if (config->use_delete) {
if (!send_status(client->file_descriptor, STATUS_MANIFEST)) {
if (send_delete_manifest(client->file_descriptor, manifest) != 0) {
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);
}
if (!send_status(client->file_descriptor, STATUS_FINISHED))
@@ -498,37 +501,18 @@ int send_files(Config* config) {
directory_scanner_destroy(scanner);
client_disconnect(client);
client_delete(client);
return ok ? 0 : -1;
return ok ? 0 : 1;
send_fail:
directory_scanner_destroy(scanner);
client_disconnect(client);
client_delete(client);
return -1;
return 1;
}
int send_files_multithreaded(Config* config) {
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);
Chunk* chunk;
int file_count = 0;
unsigned long long total_bytes = 0;
printf("Dry run: files to be transferred\n");
while ((chunk = directory_scanner_next(scanner)) != NULL) {
for (int i = 0; i < chunk->element_count; i++) {
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);
printf("Total: %d files, %.1f MB\n", file_count, total_bytes / 1048576.0);
return 0;
}
if (config->dry_run)
return send_dry_run_manifest(config);
long pages = sysconf(_SC_AVPHYS_PAGES);
long page_size = sysconf(_SC_PAGE_SIZE);
@@ -575,5 +559,5 @@ int send_files_multithreaded(Config* config) {
thrd_join(sender, &sender_result);
pipeline_context_sender_destroy(context);
return sender_result == thrd_success ? 0 : -1;
return sender_result == thrd_success ? 0 : 1;
}
src/server/server.c
+5 -1
View File
@@ -144,8 +144,10 @@ void handler(int file_descriptor) {
thrd_join(writer, NULL);
send_status(file_descriptor, STATUS_OK);
pipeline_context_receiver_destroy(context);
} else
} else {
receive_files(config, file_descriptor);
config_delete(config);
}
close(file_descriptor);
}
@@ -174,6 +176,7 @@ static void print_server_usage(void) {
printf(" --help Show this help\n");
}
#ifndef FASTSYNC_SERVER_AS_LIB
int main(int argc, char* argv[]) {
bool use_tls = false;
char* tls_cert = NULL;
@@ -244,3 +247,4 @@ int main(int argc, char* argv[]) {
}
return 0;
}
#endif /* !FASTSYNC_SERVER_AS_LIB */
src/shared/config.c
+14 -16
View File
@@ -8,28 +8,24 @@
#include <stdlib.h>
#include <string.h>
Config* config_create(char* version, char* send_directory, char* receive_directory,
bool save_to_disk, bool use_multithreading, bool use_chunk_serialization,
bool use_compression, bool use_metadata, int compression_level,
bool use_sendfile, unsigned long long chunk_size) {
Config* config_create(void) {
Config* config = malloc(sizeof(Config));
if (!config)
return NULL;
config->version = version;
config->send_directory = send_directory;
config->receive_root_directory = receive_directory;
config->save_to_disk = save_to_disk;
config->use_multithreading = use_multithreading;
config->use_chunk_serialization = use_chunk_serialization;
config->use_compression = use_compression;
config->use_metadata = use_metadata;
config->version = str_dup(PROTOCOL_VERSION);
config->send_directory = NULL;
config->receive_root_directory = NULL;
config->save_to_disk = false;
config->use_multithreading = false;
config->use_chunk_serialization = false;
config->use_compression = false;
config->use_metadata = false;
config->show_progress = false;
config->dry_run = false;
config->use_delete = false;
config->compression_level = compression_level;
config->use_sendfile = use_sendfile;
config->chunk_size = chunk_size > 0 ? chunk_size : DEFAULT_CHUNK_SIZE;
config->compression_level = 5;
config->use_sendfile = false;
config->chunk_size = DEFAULT_CHUNK_SIZE;
config->ssh_port = 22;
config->transport = TRANSPORT_TCP;
config->ssh_destination = NULL;
@@ -59,6 +55,8 @@ Config* config_create(char* version, char* send_directory, char* receive_directo
config->max_depth = 0;
config->log_file = NULL;
config->queue_size = 100;
config->follow_symlinks = false;
config->partial = false;
return config;
}
src/shared/config.h
+1 -4
View File
@@ -58,10 +58,7 @@ typedef struct Config {
#define PROTOCOL_VERSION "1.3.0"
#define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024)
Config* config_create(char* version, char* send_directory, char* receive_directory,
bool save_to_disk, bool use_multithreading, bool use_chunk_serialization,
bool use_compression, bool use_metadata, int compression_level,
bool use_sendfile, unsigned long long chunk_size);
Config* config_create(void);
void config_delete(Config* config);
bool config_send(int file_descriptor, const Config* config);
Config* config_receive(int file_descriptor);
src/shared/file.c
+2
View File
@@ -510,6 +510,8 @@ bool file_send_sendfile(File* file, int file_descriptor, bool use_metadata, int
while ((unsigned long long)offset < file_size) {
ssize_t sent = sendfile(file_descriptor, fd, &offset, file_size - offset);
if (sent == -1) {
if (errno == EAGAIN || errno == EINTR)
continue;
perror("sendfile failed");
close(fd);
return false;
src/shared/multiprocessing.c
+6 -3
View File
@@ -81,10 +81,10 @@ void pipeline_context_receiver_destroy(PipelineContextReceiver* context) {
free(context);
}
static void receive_chunk_enqueue(int file_descriptor, PipelineContextReceiver* context) {
static bool receive_chunk_enqueue(int file_descriptor, PipelineContextReceiver* context) {
Chunk* chunk = receive_chunk_data(file_descriptor, context->config);
if (chunk == NULL)
return;
return false;
for (int i = 0; i < chunk->element_count; i++) {
File* file = chunk->items[i];
@@ -93,6 +93,7 @@ static void receive_chunk_enqueue(int file_descriptor, PipelineContextReceiver*
&context->condition_not_empty, &context->condition_not_full);
}
chunk_destroy(chunk);
return true;
}
int receive_thread(void* pipeline_context) {
@@ -125,7 +126,8 @@ int receive_thread(void* pipeline_context) {
&context->condition_not_empty, &context->condition_not_full);
}
} else if (status == STATUS_CHUNK) {
receive_chunk_enqueue(file_descriptor, context);
if (!receive_chunk_enqueue(file_descriptor, context))
return thrd_error;
} else if (status == STATUS_CHECK_BATCH) {
int count;
if (!receive_int(file_descriptor, &count))
@@ -161,6 +163,7 @@ int receive_thread(void* pipeline_context) {
&context->condition_not_empty, &context->condition_not_full);
} else {
log_message(LOG_LEVEL_ERROR, "Failed to receive file");
return thrd_error;
}
}
next:
src/shared/protocol.c
+1 -1
View File
@@ -15,7 +15,7 @@
static __thread int io_read_fd = -1;
static __thread int io_write_fd = -1;
static SSL* io_ssl;
static __thread SSL* io_ssl;
static unsigned long long io_bwlimit = 0;
static long long bw_tokens = 0;