fix: reformat codebase and fix const-correctness for CI lint
CI / lint (push) Failing after 16s
CI / build-and-test (push) Has been skipped
CI / sanitizers (address) (push) Has been skipped
CI / sanitizers (thread) (push) Has been skipped
CI / lint (pull_request) Failing after 44s
CI / build-and-test (pull_request) Has been skipped
CI / sanitizers (address) (pull_request) Has been skipped
CI / sanitizers (thread) (pull_request) Has been skipped

- Reformat all C/H files to match .clang-format (LLVM style)
- Fix 26 cppcheck const-correctness warnings (constParameterPointer,
  constVariablePointer, constVariable)
- Update function declarations in headers to match const parameters
This commit is contained in:
2026-07-19 15:57:31 +02:00
parent 02266710fb
commit 23b1d6660c
44 changed files with 1053 additions and 923 deletions
+10 -9
View File
@@ -61,13 +61,12 @@ int main(int argc, char *argv[]) {
const char* env_save = getenv("FASTSYNC_SAVE_TO_DISK"); const char* env_save = getenv("FASTSYNC_SAVE_TO_DISK");
bool save_to_disk = false; bool save_to_disk = false;
if (env_save && if (env_save && (strcmp(env_save, "true") == 0 || strcmp(env_save, "1") == 0)) {
(strcmp(env_save, "true") == 0 || strcmp(env_save, "1") == 0)) {
save_to_disk = true; save_to_disk = true;
} }
Config *config = config_create(str_dup(PROTOCOL_VERSION), NULL, NULL, Config* config = config_create(str_dup(PROTOCOL_VERSION), NULL, NULL, save_to_disk, false, false,
save_to_disk, false, false, false, false, 5, false, 0); false, false, 5, false, 0);
int positional_args[2]; int positional_args[2];
int positional_count = 0; int positional_count = 0;
@@ -89,11 +88,13 @@ int main(int argc, char *argv[]) {
config->use_delete = true; config->use_delete = true;
} else if (strcmp(argv[i], "--exclude") == 0 && i + 1 < argc) { } else if (strcmp(argv[i], "--exclude") == 0 && i + 1 < argc) {
int idx = config->exclude_count++; int idx = config->exclude_count++;
config->exclude_patterns = realloc(config->exclude_patterns, config->exclude_count * sizeof(char *)); config->exclude_patterns =
realloc(config->exclude_patterns, config->exclude_count * sizeof(char*));
config->exclude_patterns[idx] = str_dup(argv[++i]); config->exclude_patterns[idx] = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--include") == 0 && i + 1 < argc) { } else if (strcmp(argv[i], "--include") == 0 && i + 1 < argc) {
int idx = config->include_count++; int idx = config->include_count++;
config->include_patterns = realloc(config->include_patterns, config->include_count * sizeof(char *)); config->include_patterns =
realloc(config->include_patterns, config->include_count * sizeof(char*));
config->include_patterns[idx] = str_dup(argv[++i]); config->include_patterns[idx] = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--max-size") == 0 && i + 1 < argc) { } else if (strcmp(argv[i], "--max-size") == 0 && i + 1 < argc) {
config->max_size = strtoull(argv[++i], NULL, 10); config->max_size = strtoull(argv[++i], NULL, 10);
@@ -109,8 +110,7 @@ int main(int argc, char *argv[]) {
int level = strtol(argv[i + 1], &end_ptr, 10); int level = strtol(argv[i + 1], &end_ptr, 10);
if (*end_ptr == '\0') { if (*end_ptr == '\0') {
config->compression_level = level; config->compression_level = level;
log_message(LOG_LEVEL_INFO, "Set Compression level to %d", log_message(LOG_LEVEL_INFO, "Set Compression level to %d", config->compression_level);
config->compression_level);
i++; i++;
} }
} }
@@ -212,7 +212,8 @@ int main(int argc, char *argv[]) {
return 1; return 1;
} }
if (config->use_sendfile && (config->use_chunk_serialization || config->use_compression)) { 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"); fprintf(stderr, "Error: -f/--sendfile cannot be combined with -c (compression) or -s (chunk "
"serialization)\n");
return 1; return 1;
} }
+86 -81
View File
@@ -20,19 +20,25 @@
#include <time.h> #include <time.h>
static int incremental_check(Client* client, File* file) { static int incremental_check(Client* client, File* file) {
if (!send_status(client->file_descriptor, STATUS_CHECK)) return -1; if (!send_status(client->file_descriptor, STATUS_CHECK))
if (!send_str(client->file_descriptor, file->path)) return -1; return -1;
if (!send_str(client->file_descriptor, file->path))
return -1;
unsigned long long fsize = file->data->size; unsigned long long fsize = file->data->size;
long long mtime = file->metadata ? file->metadata->mtime_sec : 0; 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, &fsize, sizeof(fsize)))
if (!send_n_data(client->file_descriptor, &mtime, sizeof(mtime))) return -1; return -1;
if (!send_n_data(client->file_descriptor, &mtime, sizeof(mtime)))
return -1;
Status s; Status s;
if (!receive_status(client->file_descriptor, &s)) return -1; if (!receive_status(client->file_descriptor, &s))
return -1;
if (s == STATUS_ERROR) { if (s == STATUS_ERROR) {
log_message(LOG_LEVEL_ERROR, "Server reported error for file"); log_message(LOG_LEVEL_ERROR, "Server reported error for file");
return -1; return -1;
} }
if (s == STATUS_OK) return 1; if (s == STATUS_OK)
return 1;
if (s != STATUS_NEXT) { if (s != STATUS_NEXT) {
log_message(LOG_LEVEL_ERROR, "Unexpected server status"); log_message(LOG_LEVEL_ERROR, "Unexpected server status");
return -1; return -1;
@@ -42,27 +48,37 @@ static int incremental_check(Client *client, File *file) {
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;
Data* data; Data* data;
if (config->use_compression) { if (config->use_compression) {
data = chunk_compress(chunk, config->compression_level, config->use_metadata); data = chunk_compress(chunk, config->compression_level, config->use_metadata);
} else { } else {
data = chunk_serialize(chunk, config->use_metadata); data = chunk_serialize(chunk, config->use_metadata);
} }
if (data == NULL) return -1; if (data == NULL)
if (!send_data(client->file_descriptor, data)) { data_destroy(data); return -1; } return -1;
if (!send_data(client->file_descriptor, data)) {
data_destroy(data);
return -1;
}
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) { if (config->use_incremental) {
int rc = incremental_check(client, chunk->items[i]); int rc = incremental_check(client, chunk->items[i]);
if (rc < 0) return -1; if (rc < 0)
if (rc > 0) continue; return -1;
if (!file_send_sendfile(chunk->items[i], client->file_descriptor, config->use_metadata, false)) if (rc > 0)
continue;
if (!file_send_sendfile(chunk->items[i], client->file_descriptor, config->use_metadata,
false))
return -1; return -1;
} else { } else {
if (!send_status(client->file_descriptor, STATUS_NEXT)) return -1; if (!send_status(client->file_descriptor, STATUS_NEXT))
if (!file_send_sendfile(chunk->items[i], client->file_descriptor, config->use_metadata, true)) return -1;
if (!file_send_sendfile(chunk->items[i], client->file_descriptor, config->use_metadata,
true))
return -1; return -1;
} }
} }
@@ -70,19 +86,18 @@ int send_chunk(Client *client, Chunk *chunk, Config *config) {
for (int i = 0; i < chunk->element_count; i++) { for (int i = 0; i < chunk->element_count; i++) {
if (config->use_incremental) { if (config->use_incremental) {
int rc = incremental_check(client, chunk->items[i]); int rc = incremental_check(client, chunk->items[i]);
if (rc < 0) return -1; if (rc < 0)
if (rc > 0) continue; return -1;
if (!file_send_single_calls(chunk->items[i], client->file_descriptor, if (rc > 0)
config->use_metadata, continue;
config->use_compression ? config->compression_level : 0, if (!file_send_single_calls(chunk->items[i], client->file_descriptor, config->use_metadata,
false)) config->use_compression ? config->compression_level : 0, false))
return -1; return -1;
} else { } else {
if (!send_status(client->file_descriptor, STATUS_NEXT)) return -1; if (!send_status(client->file_descriptor, STATUS_NEXT))
if (!file_send_single_calls(chunk->items[i], client->file_descriptor, return -1;
config->use_metadata, if (!file_send_single_calls(chunk->items[i], client->file_descriptor, config->use_metadata,
config->use_compression ? config->compression_level : 0, config->use_compression ? config->compression_level : 0, true))
true))
return -1; return -1;
} }
} }
@@ -101,18 +116,18 @@ static int send_chunks_multithreaded(void *pipeline_context) {
client = client_connect_ssh(context->config->ssh_destination, context->config->ssh_port); client = client_connect_ssh(context->config->ssh_destination, context->config->ssh_port);
} else if (context->config->use_tls) { } else if (context->config->use_tls) {
client = client_create(); client = client_create();
if (!client || !client_connect_tls(client, server_host, server_port, if (!client || !client_connect_tls(client, server_host, server_port, context->config->tls_cert,
context->config->tls_cert, context->config->tls_key, context->config->tls_ca)) {
context->config->tls_key, if (client)
context->config->tls_ca)) { client_delete(client);
if (client) client_delete(client);
fprintf(stderr, "Error: could not connect to server via TLS\n"); fprintf(stderr, "Error: could not connect to server via TLS\n");
return thrd_error; return thrd_error;
} }
} else { } else {
client = client_create(); client = client_create();
if (!client || !client_connect(client, server_host, server_port)) { if (!client || !client_connect(client, server_host, server_port)) {
if (client) client_delete(client); if (client)
client_delete(client);
fprintf(stderr, "Error: could not connect to server\n"); fprintf(stderr, "Error: could not connect to server\n");
return thrd_error; return thrd_error;
} }
@@ -125,16 +140,14 @@ static int send_chunks_multithreaded(void *pipeline_context) {
while (true) { while (true) {
Chunk* current_chunk = queue_dequeue_multithreaded( Chunk* current_chunk = queue_dequeue_multithreaded(
context->queue_loader, &context->mutex_loader, context->queue_loader, &context->mutex_loader, &context->condition_not_empty_loader,
&context->condition_not_empty_loader,
&context->condition_not_full_loader, &context->loader_done); &context->condition_not_full_loader, &context->loader_done);
if (current_chunk == NULL) { if (current_chunk == NULL) {
if (context->config->use_delete) { if (context->config->use_delete) {
send_status(client->file_descriptor, STATUS_MANIFEST); send_status(client->file_descriptor, STATUS_MANIFEST);
send_int(client->file_descriptor, context->manifest->size); send_int(client->file_descriptor, context->manifest->size);
for (int i = 0; i < context->manifest->size; i++) for (int i = 0; i < context->manifest->size; i++)
send_str(client->file_descriptor, send_str(client->file_descriptor, (char*)context->manifest->items[i]);
(char *)context->manifest->items[i]);
} }
send_status(client->file_descriptor, STATUS_FINISHED); send_status(client->file_descriptor, STATUS_FINISHED);
Status s; Status s;
@@ -157,10 +170,9 @@ static int scan_directory_multithreaded(void *pipeline_context) {
PipelineContextSender* context = (PipelineContextSender*)pipeline_context; PipelineContextSender* context = (PipelineContextSender*)pipeline_context;
mtx_lock(&context->mutex_scanner); mtx_lock(&context->mutex_scanner);
DirectoryScanner* scanner = directory_scanner_create( DirectoryScanner* scanner = directory_scanner_create(
context->config->send_directory, context->config->use_metadata, context->config->send_directory, context->config->use_metadata, context->config->chunk_size,
context->config->chunk_size, context->config->exclude_patterns, context->config->exclude_patterns, context->config->exclude_count,
context->config->exclude_count, context->config->include_patterns, context->config->include_patterns, context->config->include_count, context->config->max_size,
context->config->include_count, context->config->max_size,
context->config->min_size); context->config->min_size);
mtx_unlock(&context->mutex_scanner); mtx_unlock(&context->mutex_scanner);
@@ -170,13 +182,13 @@ static int scan_directory_multithreaded(void *pipeline_context) {
mtx_lock(&context->mutex_scanner); mtx_lock(&context->mutex_scanner);
for (int i = 0; i < current_chunk->element_count; i++) { for (int i = 0; i < current_chunk->element_count; i++) {
const char* p = current_chunk->items[i]->path; const char* p = current_chunk->items[i]->path;
if (*p == '/') p++; if (*p == '/')
p++;
array_list_add(context->manifest, str_dup(p)); array_list_add(context->manifest, str_dup(p));
} }
mtx_unlock(&context->mutex_scanner); mtx_unlock(&context->mutex_scanner);
} }
queue_enqueue_multithreaded(context->queue_scanner, current_chunk, queue_enqueue_multithreaded(context->queue_scanner, current_chunk, &context->mutex_scanner,
&context->mutex_scanner,
&context->condition_not_empty_scanner, &context->condition_not_empty_scanner,
&context->condition_not_full_scanner); &context->condition_not_full_scanner);
} }
@@ -193,8 +205,7 @@ static int load_files_multithreaded(void *pipeline_context) {
PipelineContextSender* context = (PipelineContextSender*)pipeline_context; PipelineContextSender* context = (PipelineContextSender*)pipeline_context;
while (true) { while (true) {
Chunk* chunk = queue_dequeue_multithreaded( Chunk* chunk = queue_dequeue_multithreaded(
context->queue_scanner, &context->mutex_scanner, context->queue_scanner, &context->mutex_scanner, &context->condition_not_empty_scanner,
&context->condition_not_empty_scanner,
&context->condition_not_full_scanner, &context->scanner_done); &context->condition_not_full_scanner, &context->scanner_done);
if (chunk == NULL) { if (chunk == NULL) {
mtx_lock(&context->mutex_loader); mtx_lock(&context->mutex_loader);
@@ -212,8 +223,7 @@ static int load_files_multithreaded(void *pipeline_context) {
} }
} }
} }
queue_enqueue_multithreaded(context->queue_loader, chunk, queue_enqueue_multithreaded(context->queue_loader, chunk, &context->mutex_loader,
&context->mutex_loader,
&context->condition_not_empty_loader, &context->condition_not_empty_loader,
&context->condition_not_full_loader); &context->condition_not_full_loader);
} }
@@ -222,26 +232,23 @@ static int load_files_multithreaded(void *pipeline_context) {
int send_files(Config* config) { int send_files(Config* config) {
if (config->dry_run) { if (config->dry_run) {
DirectoryScanner* scanner = directory_scanner_create( DirectoryScanner* scanner = directory_scanner_create(
config->send_directory, config->use_metadata, config->chunk_size, config->send_directory, config->use_metadata, config->chunk_size, config->exclude_patterns,
config->exclude_patterns, config->exclude_count, config->exclude_count, config->include_patterns, config->include_count, config->max_size,
config->include_patterns, config->include_count, config->min_size);
config->max_size, config->min_size);
Chunk* chunk; Chunk* chunk;
int file_count = 0; int file_count = 0;
unsigned long long total_bytes = 0; unsigned long long total_bytes = 0;
printf("Dry run: files to be transferred\n"); printf("Dry run: files to be transferred\n");
while ((chunk = directory_scanner_next(scanner)) != NULL) { while ((chunk = directory_scanner_next(scanner)) != NULL) {
for (int i = 0; i < chunk->element_count; i++) { for (int i = 0; i < chunk->element_count; i++) {
printf(" %s (%zu bytes)\n", chunk->items[i]->path, printf(" %s (%zu bytes)\n", chunk->items[i]->path, chunk->items[i]->data->size);
chunk->items[i]->data->size);
total_bytes += chunk->items[i]->data->size; total_bytes += chunk->items[i]->data->size;
file_count++; file_count++;
} }
chunk_destroy(chunk); chunk_destroy(chunk);
} }
directory_scanner_destroy(scanner); directory_scanner_destroy(scanner);
printf("Total: %d files, %.1f MB\n", file_count, printf("Total: %d files, %.1f MB\n", file_count, total_bytes / 1048576.0);
total_bytes / 1048576.0);
return 0; return 0;
} }
@@ -252,20 +259,22 @@ int send_files(Config *config) {
return 1; return 1;
} }
client = client_connect_ssh(config->ssh_destination, config->ssh_port); client = client_connect_ssh(config->ssh_destination, config->ssh_port);
if (!client) return 1; if (!client)
return 1;
} else if (config->use_tls) { } else if (config->use_tls) {
client = client_create(); client = client_create();
if (!client || !client_connect_tls(client, server_host, server_port, if (!client || !client_connect_tls(client, server_host, server_port, config->tls_cert,
config->tls_cert, config->tls_key, config->tls_key, config->tls_ca)) {
config->tls_ca)) { if (client)
if (client) client_delete(client); client_delete(client);
fprintf(stderr, "Error: could not connect to server via TLS\n"); fprintf(stderr, "Error: could not connect to server via TLS\n");
return 1; return 1;
} }
} else { } else {
client = client_create(); client = client_create();
if (!client || !client_connect(client, server_host, server_port)) { if (!client || !client_connect(client, server_host, server_port)) {
if (client) client_delete(client); if (client)
client_delete(client);
fprintf(stderr, "Error: could not connect to server\n"); fprintf(stderr, "Error: could not connect to server\n");
return 1; return 1;
} }
@@ -276,10 +285,9 @@ int send_files(Config *config) {
return 1; return 1;
} }
DirectoryScanner* scanner = directory_scanner_create( DirectoryScanner* scanner = directory_scanner_create(
config->send_directory, config->use_metadata, config->chunk_size, config->send_directory, config->use_metadata, config->chunk_size, config->exclude_patterns,
config->exclude_patterns, config->exclude_count, config->exclude_count, config->include_patterns, config->include_count, config->max_size,
config->include_patterns, config->include_count, config->min_size);
config->max_size, config->min_size);
Chunk* current_chunk; Chunk* current_chunk;
unsigned long long total_bytes = 0; unsigned long long total_bytes = 0;
time_t last_progress = 0; time_t last_progress = 0;
@@ -291,7 +299,8 @@ int send_files(Config *config) {
chunk_bytes += current_chunk->items[i]->data->size; chunk_bytes += current_chunk->items[i]->data->size;
if (manifest) { if (manifest) {
const char* p = current_chunk->items[i]->path; const char* p = current_chunk->items[i]->path;
if (*p == '/') p++; if (*p == '/')
p++;
array_list_add(manifest, str_dup(p)); array_list_add(manifest, str_dup(p));
} }
} }
@@ -345,38 +354,36 @@ int send_files(Config *config) {
int send_files_multithreaded(Config* config) { int send_files_multithreaded(Config* config) {
if (config->dry_run) { if (config->dry_run) {
DirectoryScanner* scanner = directory_scanner_create( DirectoryScanner* scanner = directory_scanner_create(
config->send_directory, config->use_metadata, config->chunk_size, config->send_directory, config->use_metadata, config->chunk_size, config->exclude_patterns,
config->exclude_patterns, config->exclude_count, config->exclude_count, config->include_patterns, config->include_count, config->max_size,
config->include_patterns, config->include_count, config->min_size);
config->max_size, config->min_size);
Chunk* chunk; Chunk* chunk;
int file_count = 0; int file_count = 0;
unsigned long long total_bytes = 0; unsigned long long total_bytes = 0;
printf("Dry run: files to be transferred\n"); printf("Dry run: files to be transferred\n");
while ((chunk = directory_scanner_next(scanner)) != NULL) { while ((chunk = directory_scanner_next(scanner)) != NULL) {
for (int i = 0; i < chunk->element_count; i++) { for (int i = 0; i < chunk->element_count; i++) {
printf(" %s (%zu bytes)\n", chunk->items[i]->path, printf(" %s (%zu bytes)\n", chunk->items[i]->path, chunk->items[i]->data->size);
chunk->items[i]->data->size);
total_bytes += chunk->items[i]->data->size; total_bytes += chunk->items[i]->data->size;
file_count++; file_count++;
} }
chunk_destroy(chunk); chunk_destroy(chunk);
} }
directory_scanner_destroy(scanner); directory_scanner_destroy(scanner);
printf("Total: %d files, %.1f MB\n", file_count, printf("Total: %d files, %.1f MB\n", file_count, total_bytes / 1048576.0);
total_bytes / 1048576.0);
return 0; return 0;
} }
Queue* q1 = queue_create(100, chunk_destroy); Queue* q1 = queue_create(100, chunk_destroy);
Queue* q2 = queue_create(100, chunk_destroy); Queue* q2 = queue_create(100, chunk_destroy);
if (!q1 || !q2) { if (!q1 || !q2) {
if (q1) queue_destroy(q1); if (q1)
if (q2) queue_destroy(q2); queue_destroy(q1);
if (q2)
queue_destroy(q2);
return 1; return 1;
} }
PipelineContextSender *context = PipelineContextSender* context = pipeline_context_sender_create(config, q1, q2);
pipeline_context_sender_create(config, q1, q2);
if (!context) { if (!context) {
queue_destroy(q1); queue_destroy(q1);
queue_destroy(q2); queue_destroy(q2);
@@ -386,11 +393,9 @@ int send_files_multithreaded(Config *config) {
context->manifest = array_list_create(free); context->manifest = array_list_create(free);
thrd_t scanner, loader, sender; thrd_t scanner, loader, sender;
if (thrd_create(&scanner, scan_directory_multithreaded, context) != if (thrd_create(&scanner, scan_directory_multithreaded, context) != thrd_success ||
thrd_success ||
thrd_create(&loader, load_files_multithreaded, context) != thrd_success || thrd_create(&loader, load_files_multithreaded, context) != thrd_success ||
thrd_create(&sender, send_chunks_multithreaded, context) != thrd_create(&sender, send_chunks_multithreaded, context) != thrd_success) {
thrd_success) {
perror("Error creating threads.\n"); perror("Error creating threads.\n");
pipeline_context_sender_destroy(context); pipeline_context_sender_destroy(context);
return 1; return 1;
+5 -1
View File
@@ -11,7 +11,11 @@
#include <sys/stat.h> #include <sys/stat.h>
#include <unistd.h> #include <unistd.h>
DirectoryScanner *directory_scanner_create(char *root_directory, bool use_metadata, unsigned long long chunk_size, char **exclude_patterns, int exclude_count, char **include_patterns, int include_count, unsigned long long max_size, unsigned long long min_size) { DirectoryScanner* directory_scanner_create(char* root_directory, bool use_metadata,
unsigned long long chunk_size, char** exclude_patterns,
int exclude_count, char** include_patterns,
int include_count, unsigned long long max_size,
unsigned long long min_size) {
DirectoryScanner* scanner = malloc(sizeof(DirectoryScanner)); DirectoryScanner* scanner = malloc(sizeof(DirectoryScanner));
scanner->directories = queue_create(100, free); scanner->directories = queue_create(100, free);
scanner->current_dir = NULL; scanner->current_dir = NULL;
+5 -1
View File
@@ -20,7 +20,11 @@ typedef struct {
unsigned long long min_size; unsigned long long min_size;
} DirectoryScanner; } DirectoryScanner;
DirectoryScanner *directory_scanner_create(char *root_directory, bool use_metadata, unsigned long long chunk_size, char **exclude_patterns, int exclude_count, char **include_patterns, int include_count, unsigned long long max_size, unsigned long long min_size); DirectoryScanner* directory_scanner_create(char* root_directory, bool use_metadata,
unsigned long long chunk_size, char** exclude_patterns,
int exclude_count, char** include_patterns,
int include_count, unsigned long long max_size,
unsigned long long min_size);
Chunk* directory_scanner_next(DirectoryScanner* scanner); Chunk* directory_scanner_next(DirectoryScanner* scanner);
void directory_scanner_destroy(DirectoryScanner* scanner); void directory_scanner_destroy(DirectoryScanner* scanner);
+9 -6
View File
@@ -18,14 +18,17 @@
int receive_files(Config* config, int fd) { int receive_files(Config* config, int fd) {
Status status; Status status;
if (!receive_status(fd, &status)) return -1; if (!receive_status(fd, &status))
return -1;
while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK) { while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK) {
if (status == STATUS_CHECK) { if (status == STATUS_CHECK) {
bool skipped; bool skipped;
File* file = receive_incremental_check(fd, config, &skipped); File* file = receive_incremental_check(fd, config, &skipped);
if (skipped) goto next; if (skipped)
if (file == NULL && !skipped) return -1; goto next;
if (file == NULL && !skipped)
return -1;
if (config->save_to_disk) if (config->save_to_disk)
file_save_to_disk(config->receive_root_directory, file); file_save_to_disk(config->receive_root_directory, file);
file_destroy(file); file_destroy(file);
@@ -59,7 +62,8 @@ int receive_files(Config *config, int fd) {
} }
if (status == STATUS_MANIFEST) { if (status == STATUS_MANIFEST) {
if (receive_manifest(fd, config, &status) != 0) return -1; 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");
@@ -84,8 +88,7 @@ void handler(int file_descriptor) {
close(file_descriptor); close(file_descriptor);
return; return;
} }
PipelineContextReceiver *context = pipeline_context_receiver_create( PipelineContextReceiver* context = pipeline_context_receiver_create(config, q, file_descriptor);
config, q, file_descriptor);
if (context == NULL) { if (context == NULL) {
queue_destroy(q); queue_destroy(q);
config_delete(config); config_delete(config);
+5 -3
View File
@@ -35,7 +35,8 @@ void array_list_delete(ArrayList *array_list) {
} }
bool array_list_extend(ArrayList* array_list) { bool array_list_extend(ArrayList* array_list) {
if (array_list == NULL) return false; if (array_list == NULL)
return false;
int new_capacity = array_list->capacity * 2; int new_capacity = array_list->capacity * 2;
if (new_capacity == 0) if (new_capacity == 0)
new_capacity = INITIAL_ARRAY_SIZE; new_capacity = INITIAL_ARRAY_SIZE;
@@ -50,7 +51,8 @@ bool array_list_extend(ArrayList *array_list) {
} }
bool array_list_add(ArrayList* array_list, void* item) { bool array_list_add(ArrayList* array_list, void* item) {
if (array_list == NULL) return false; if (array_list == NULL)
return false;
if (array_list->capacity == array_list->size) { if (array_list->capacity == array_list->size) {
if (!array_list_extend(array_list)) if (!array_list_extend(array_list))
return false; return false;
@@ -60,7 +62,7 @@ bool array_list_add(ArrayList *array_list, void *item) {
return true; return true;
} }
void **array_list_to_array(ArrayList *array_list) { void** array_list_to_array(const ArrayList* array_list) {
if (array_list == NULL) { if (array_list == NULL) {
return NULL; return NULL;
} }
+1 -1
View File
@@ -16,6 +16,6 @@ ArrayList *array_list_create(void (*item_destroyer)(void *item));
void array_list_delete(ArrayList* array_list); void array_list_delete(ArrayList* array_list);
bool array_list_extend(ArrayList* array_list); bool array_list_extend(ArrayList* array_list);
bool array_list_add(ArrayList* array_list, void* item); bool array_list_add(ArrayList* array_list, void* item);
void **array_list_to_array(ArrayList *array_list); void** array_list_to_array(const ArrayList* array_list);
#endif #endif
+6 -8
View File
@@ -59,8 +59,7 @@ Data *chunk_serialize(Chunk *chunk, bool use_metadata) {
} }
Data* data = data_create_empty(data_size); Data* data = data_create_empty(data_size);
if (data == NULL) { if (data == NULL) {
log_message(LOG_LEVEL_ERROR, log_message(LOG_LEVEL_ERROR, "Could not allocate memory for chunk serialization");
"Could not allocate memory for chunk serialization");
return NULL; return NULL;
} }
char* data_pointer = data->data; char* data_pointer = data->data;
@@ -171,15 +170,17 @@ Chunk *chunk_deserialize(Data *data, bool use_metadata) {
Data* chunk_compress(Chunk* chunk, int compression_level, bool use_metadata) { Data* chunk_compress(Chunk* chunk, int compression_level, bool use_metadata) {
log_message(LOG_LEVEL_DEBUG, "Starting to compress chunk"); log_message(LOG_LEVEL_DEBUG, "Starting to compress chunk");
Data* serialized = chunk_serialize(chunk, use_metadata); Data* serialized = chunk_serialize(chunk, use_metadata);
if (serialized == NULL) return NULL; if (serialized == NULL)
return NULL;
Data* compressed = data_compress(serialized, compression_level); Data* compressed = data_compress(serialized, compression_level);
data_destroy(serialized); data_destroy(serialized);
if (compressed == NULL) return NULL; if (compressed == NULL)
return NULL;
log_message(LOG_LEVEL_DEBUG, "Chunk successfully compressed"); log_message(LOG_LEVEL_DEBUG, "Chunk successfully compressed");
return compressed; return compressed;
} }
Chunk *receive_chunk_data(int fd, Config *config) { Chunk* receive_chunk_data(int fd, const Config* config) {
Data* chunk_data = receive_data(fd); Data* chunk_data = receive_data(fd);
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");
@@ -200,6 +201,3 @@ Chunk *receive_chunk_data(int fd, Config *config) {
log_message(LOG_LEVEL_ERROR, "Failed to deserialize chunk, skipping"); log_message(LOG_LEVEL_ERROR, "Failed to deserialize chunk, skipping");
return chunk; return chunk;
} }
+1 -1
View File
@@ -19,6 +19,6 @@ void chunk_destroy(void *chunk);
Data* chunk_serialize(Chunk* chunk, bool use_metadata); Data* chunk_serialize(Chunk* chunk, bool use_metadata);
Chunk* chunk_deserialize(Data* data, bool use_metadata); Chunk* chunk_deserialize(Data* data, bool use_metadata);
Data* chunk_compress(Chunk* chunk, int compression_level, bool use_metadata); Data* chunk_compress(Chunk* chunk, int compression_level, bool use_metadata);
Chunk *receive_chunk_data(int fd, Config *config); Chunk* receive_chunk_data(int fd, const Config* config);
#endif #endif
+9 -12
View File
@@ -11,7 +11,8 @@ Data *data_compress(Data *data_to_compress, int compression_level) {
log_message(LOG_LEVEL_DEBUG, "Starting to compress data"); log_message(LOG_LEVEL_DEBUG, "Starting to compress data");
size_t dst_size = ZSTD_compressBound(data_to_compress->size); size_t dst_size = ZSTD_compressBound(data_to_compress->size);
Data* compressed_data = data_create_empty(dst_size); Data* compressed_data = data_create_empty(dst_size);
if (compressed_data == NULL) return NULL; if (compressed_data == NULL)
return NULL;
ZSTD_CCtx* cctx = ZSTD_createCCtx(); ZSTD_CCtx* cctx = ZSTD_createCCtx();
if (!cctx) { if (!cctx) {
@@ -27,8 +28,7 @@ Data *data_compress(Data *data_to_compress, int compression_level) {
do { do {
ret = ZSTD_compressStream2(cctx, &output, &input, ZSTD_e_end); ret = ZSTD_compressStream2(cctx, &output, &input, ZSTD_e_end);
if (ZSTD_isError(ret)) { if (ZSTD_isError(ret)) {
log_message(LOG_LEVEL_ERROR, "Compression failed: %s", log_message(LOG_LEVEL_ERROR, "Compression failed: %s", ZSTD_getErrorName(ret));
ZSTD_getErrorName(ret));
ZSTD_freeCCtx(cctx); ZSTD_freeCCtx(cctx);
data_destroy(compressed_data); data_destroy(compressed_data);
return NULL; return NULL;
@@ -45,8 +45,8 @@ Data *data_compress(Data *data_to_compress, int compression_level) {
Data* data_decompress(Data* compressed_data) { Data* data_decompress(Data* compressed_data) {
log_message(LOG_LEVEL_DEBUG, "Start to decompress data"); log_message(LOG_LEVEL_DEBUG, "Start to decompress data");
unsigned long long dst_size = ZSTD_getFrameContentSize( unsigned long long dst_size =
compressed_data->data, compressed_data->size); ZSTD_getFrameContentSize(compressed_data->data, compressed_data->size);
if (ZSTD_isError(dst_size)) { if (ZSTD_isError(dst_size)) {
log_message(LOG_LEVEL_ERROR, "Failed to get decompressed size: %s", log_message(LOG_LEVEL_ERROR, "Failed to get decompressed size: %s",
ZSTD_getErrorName(dst_size)); ZSTD_getErrorName(dst_size));
@@ -55,14 +55,12 @@ Data *data_decompress(Data *compressed_data) {
ZSTD_DCtx* dctx = ZSTD_createDCtx(); ZSTD_DCtx* dctx = ZSTD_createDCtx();
if (!dctx) { if (!dctx) {
log_message(LOG_LEVEL_ERROR, log_message(LOG_LEVEL_ERROR, "Failed to create ZSTD decompression context");
"Failed to create ZSTD decompression context");
return NULL; return NULL;
} }
size_t buf_size = (!ZSTD_isError(dst_size) && dst_size > 0) size_t buf_size =
? (size_t)dst_size (!ZSTD_isError(dst_size) && dst_size > 0) ? (size_t)dst_size : INITIAL_DECOMPRESS_BUF_SIZE;
: INITIAL_DECOMPRESS_BUF_SIZE;
Data* uncompressed_data = data_create_empty(buf_size); Data* uncompressed_data = data_create_empty(buf_size);
if (!uncompressed_data) { if (!uncompressed_data) {
log_message(LOG_LEVEL_ERROR, "Failed to allocate decompression buffer"); log_message(LOG_LEVEL_ERROR, "Failed to allocate decompression buffer");
@@ -77,8 +75,7 @@ Data *data_decompress(Data *compressed_data) {
do { do {
ret = ZSTD_decompressStream(dctx, &output, &input); ret = ZSTD_decompressStream(dctx, &output, &input);
if (ZSTD_isError(ret)) { if (ZSTD_isError(ret)) {
log_message(LOG_LEVEL_ERROR, "Decompression failed: %s", log_message(LOG_LEVEL_ERROR, "Decompression failed: %s", ZSTD_getErrorName(ret));
ZSTD_getErrorName(ret));
ZSTD_freeDCtx(dctx); ZSTD_freeDCtx(dctx);
data_destroy(uncompressed_data); data_destroy(uncompressed_data);
return NULL; return NULL;
+84 -43
View File
@@ -7,12 +7,10 @@
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
Config *config_create(char *version, char *send_directory, Config* config_create(char* version, char* send_directory, char* receive_directory,
char *receive_directory, bool save_to_disk, bool save_to_disk, bool use_multithreading, bool use_chunk_serialization,
bool use_multithreading, bool use_chunk_serialization, bool use_compression, bool use_metadata, int compression_level,
bool use_compression, bool use_metadata, bool use_sendfile, unsigned long long chunk_size) {
int compression_level, bool use_sendfile,
unsigned long long chunk_size) {
Config* config = malloc(sizeof(Config)); Config* config = malloc(sizeof(Config));
config->version = version; config->version = version;
@@ -47,21 +45,26 @@ Config *config_create(char *version, char *send_directory,
} }
bool is_remote_dest(const char* s) { bool is_remote_dest(const char* s) {
if (s == NULL) return false; if (s == NULL)
return false;
const char* colon = strchr(s, ':'); const char* colon = strchr(s, ':');
if (colon == NULL) return false; if (colon == NULL)
if (colon == s) return false; return false;
if (colon == s)
return false;
for (const char* p = s; p < colon; p++) { for (const char* p = s; p < colon; p++) {
if (*p == '/') return false; if (*p == '/')
return false;
} }
return true; return true;
} }
void config_parse_ssh_dest(Config* config) { void config_parse_ssh_dest(Config* config) {
if (!is_remote_dest(config->receive_root_directory)) return; if (!is_remote_dest(config->receive_root_directory))
return;
config->transport = TRANSPORT_SSH; config->transport = TRANSPORT_SSH;
config->ssh_destination = str_dup(config->receive_root_directory); config->ssh_destination = str_dup(config->receive_root_directory);
char *colon = strchr(config->receive_root_directory, ':'); const char* colon = strchr(config->receive_root_directory, ':');
char* path = str_dup(colon + 1); char* path = str_dup(colon + 1);
free(config->receive_root_directory); free(config->receive_root_directory);
config->receive_root_directory = path; config->receive_root_directory = path;
@@ -85,21 +88,35 @@ void config_delete(Config *config) {
} }
bool config_send(int file_descriptor, Config* config) { bool config_send(int file_descriptor, Config* config) {
if (!send_str(file_descriptor, config->version)) return false; if (!send_str(file_descriptor, config->version))
if (!send_str(file_descriptor, config->send_directory)) return false; return false;
if (!send_str(file_descriptor, config->receive_root_directory)) return false; if (!send_str(file_descriptor, config->send_directory))
if (!send_int(file_descriptor, config->save_to_disk)) return false; return false;
if (!send_int(file_descriptor, config->use_multithreading)) return false; if (!send_str(file_descriptor, config->receive_root_directory))
if (!send_int(file_descriptor, config->use_chunk_serialization)) return false; return false;
if (!send_int(file_descriptor, config->use_compression)) return false; if (!send_int(file_descriptor, config->save_to_disk))
if (!send_int(file_descriptor, config->use_metadata)) return false; return false;
if (!send_int(file_descriptor, config->compression_level)) return false; if (!send_int(file_descriptor, config->use_multithreading))
if (!send_int(file_descriptor, (int)config->chunk_size)) return false; return false;
if (!send_int(file_descriptor, config->use_sendfile)) return false; if (!send_int(file_descriptor, config->use_chunk_serialization))
if (!send_int(file_descriptor, config->use_delete)) return false; return false;
if (!send_int(file_descriptor, config->use_incremental)) return false; if (!send_int(file_descriptor, config->use_compression))
return false;
if (!send_int(file_descriptor, config->use_metadata))
return false;
if (!send_int(file_descriptor, config->compression_level))
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_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) {
log_message(LOG_LEVEL_ERROR, "Error transmitting config"); log_message(LOG_LEVEL_ERROR, "Error transmitting config");
return false; return false;
@@ -109,41 +126,64 @@ bool config_send(int file_descriptor, Config *config) {
Config* config_receive(int file_descriptor) { Config* config_receive(int file_descriptor) {
Config* config = (Config*)malloc(sizeof(Config)); Config* config = (Config*)malloc(sizeof(Config));
if (config == NULL) return NULL; if (config == NULL)
return NULL;
config->version = receive_str(file_descriptor); config->version = receive_str(file_descriptor);
if (!config->version) { free(config); return NULL; } if (!config->version) {
free(config);
return NULL;
}
if (strcmp(config->version, PROTOCOL_VERSION) != 0) { if (strcmp(config->version, PROTOCOL_VERSION) != 0) {
fprintf(stderr, "Protocol version mismatch: client=%s, server=%s\n", fprintf(stderr, "Protocol version mismatch: client=%s, server=%s\n", config->version,
config->version, PROTOCOL_VERSION); PROTOCOL_VERSION);
free(config->version); free(config->version);
free(config); free(config);
send_status(file_descriptor, STATUS_ERROR); send_status(file_descriptor, STATUS_ERROR);
return NULL; return NULL;
} }
config->send_directory = receive_str(file_descriptor); config->send_directory = receive_str(file_descriptor);
if (!config->send_directory) { free(config->version); free(config); return NULL; } if (!config->send_directory) {
free(config->version);
free(config);
return NULL;
}
config->receive_root_directory = receive_str(file_descriptor); config->receive_root_directory = receive_str(file_descriptor);
if (!config->receive_root_directory) { free(config->version); free(config->send_directory); free(config); return NULL; } if (!config->receive_root_directory) {
free(config->version);
free(config->send_directory);
free(config);
return NULL;
}
int tmp; int tmp;
if (!receive_int(file_descriptor, &tmp)) goto error; if (!receive_int(file_descriptor, &tmp))
goto error;
config->save_to_disk = tmp; config->save_to_disk = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error; if (!receive_int(file_descriptor, &tmp))
goto error;
config->use_multithreading = tmp; config->use_multithreading = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error; if (!receive_int(file_descriptor, &tmp))
goto error;
config->use_chunk_serialization = tmp; config->use_chunk_serialization = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error; if (!receive_int(file_descriptor, &tmp))
goto error;
config->use_compression = tmp; config->use_compression = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error; if (!receive_int(file_descriptor, &tmp))
goto error;
config->use_metadata = tmp; config->use_metadata = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error; if (!receive_int(file_descriptor, &tmp))
goto error;
config->compression_level = tmp; config->compression_level = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error; if (!receive_int(file_descriptor, &tmp))
goto error;
config->chunk_size = (unsigned long long)tmp; config->chunk_size = (unsigned long long)tmp;
if (!receive_int(file_descriptor, &tmp)) goto error; if (!receive_int(file_descriptor, &tmp))
goto error;
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; if (!receive_int(file_descriptor, &tmp))
goto error;
config->use_incremental = tmp; config->use_incremental = tmp;
config->show_progress = false; config->show_progress = false;
config->dry_run = false; config->dry_run = false;
@@ -160,7 +200,8 @@ Config *config_receive(int file_descriptor) {
config->tls_cert = NULL; config->tls_cert = NULL;
config->tls_key = NULL; config->tls_key = NULL;
config->tls_ca = NULL; config->tls_ca = NULL;
if (!send_status(file_descriptor, STATUS_OK)) goto error; if (!send_status(file_descriptor, STATUS_OK))
goto error;
return config; return config;
error: error:
+5 -10
View File
@@ -3,10 +3,7 @@
#include <stdbool.h> #include <stdbool.h>
typedef enum { typedef enum { TRANSPORT_TCP, TRANSPORT_SSH } TransportType;
TRANSPORT_TCP,
TRANSPORT_SSH
} TransportType;
typedef struct Config { typedef struct Config {
char* version; char* version;
@@ -42,12 +39,10 @@ typedef struct Config {
#define PROTOCOL_VERSION "1.1.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, char* receive_directory,
char *receive_directory, bool save_to_disk, bool save_to_disk, bool use_multithreading, bool use_chunk_serialization,
bool use_multithreading, bool use_chunk_serialization, bool use_compression, bool use_metadata, int compression_level,
bool use_compression, bool use_metadata, bool use_sendfile, unsigned long long chunk_size);
int compression_level, bool use_sendfile,
unsigned long long chunk_size);
void config_delete(Config* config); void config_delete(Config* config);
bool config_send(int file_descriptor, Config* config); bool config_send(int file_descriptor, Config* config);
Config* config_receive(int file_descriptor); Config* config_receive(int file_descriptor);
+2 -1
View File
@@ -35,7 +35,8 @@ Data *data_create(void *data, size_t data_size) {
} }
void data_destroy(Data* data) { void data_destroy(Data* data) {
if (data == NULL) return; if (data == NULL)
return;
free(data->data); free(data->data);
free(data); free(data);
} }
+59 -28
View File
@@ -57,7 +57,7 @@ void file_destroy(void *item) {
free(file); free(file);
} }
FileMetadata *file_metadata_create(struct stat *stats) { FileMetadata* file_metadata_create(const struct stat* stats) {
FileMetadata* m = malloc(sizeof(FileMetadata)); FileMetadata* m = malloc(sizeof(FileMetadata));
if (m == NULL) { if (m == NULL) {
perror("ERROR: Could not allocate memory for file metadata"); perror("ERROR: Could not allocate memory for file metadata");
@@ -80,7 +80,8 @@ void file_metadata_destroy(void *metadata) {
} }
bool file_load_data(File* file) { bool file_load_data(File* file) {
if (file == NULL) return false; if (file == NULL)
return false;
if (file->data->data == NULL) { if (file->data->data == NULL) {
file->data->data = malloc(file->data->size); file->data->data = malloc(file->data->size);
if (file->data->data == NULL) { if (file->data->data == NULL) {
@@ -96,7 +97,8 @@ 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 send_path) { bool file_send_single_calls(File* file, int file_descriptor, bool use_metadata,
int compression_level, bool send_path) {
Data* data_to_send = file->data; Data* data_to_send = file->data;
Data* compressed_data = NULL; Data* compressed_data = NULL;
if (compression_level > 0) { if (compression_level > 0) {
@@ -125,16 +127,22 @@ bool file_send_single_calls(File *file, int file_descriptor, bool use_metadata,
bool file_save_to_disk(const char* root_directory, File* file) { bool file_save_to_disk(const char* root_directory, File* file) {
char* disk_path = path_cat((char*)root_directory, file->path); char* disk_path = path_cat((char*)root_directory, file->path);
if (disk_path == NULL) return false; if (disk_path == NULL)
return false;
bool ok = to_disk(disk_path, file->data->data, file->data->size); bool ok = to_disk(disk_path, file->data->data, file->data->size);
if (ok) file_restore_metadata(disk_path, file->metadata); if (ok)
file_restore_metadata(disk_path, file->metadata);
free(disk_path); free(disk_path);
return ok; return ok;
} }
File *receive_incremental_check(int fd, Config *config, bool *skipped) { *skipped = false; File* receive_incremental_check(int fd, Config* config, bool* skipped) {
*skipped = false;
char* check_path = receive_str(fd); char* check_path = receive_str(fd);
if (check_path == NULL) { send_status(fd, STATUS_ERROR); return NULL; } if (check_path == NULL) {
send_status(fd, STATUS_ERROR);
return NULL;
}
unsigned long long check_size; unsigned long long check_size;
long long check_mtime; long long check_mtime;
@@ -148,30 +156,42 @@ File *receive_incremental_check(int fd, Config *config, bool *skipped) { *skipp
char* full_path = path_cat(config->receive_root_directory, check_path); char* full_path = path_cat(config->receive_root_directory, check_path);
struct stat st; struct stat st;
bool match = false; bool match = false;
if (full_path && stat(full_path, &st) == 0 && if (full_path && stat(full_path, &st) == 0 && (unsigned long long)st.st_size == check_size &&
(unsigned long long)st.st_size == check_size &&
(long long)st.st_mtime == check_mtime) { (long long)st.st_mtime == check_mtime) {
match = true; match = true;
} }
free(full_path); free(full_path);
if (match) { if (match) {
if (!send_status(fd, STATUS_OK)) { free(check_path); return NULL; } if (!send_status(fd, STATUS_OK)) {
free(check_path);
return NULL;
}
free(check_path); free(check_path);
*skipped = true; *skipped = true;
return NULL; return NULL;
} }
if (!send_status(fd, STATUS_NEXT)) { free(check_path); return NULL; } if (!send_status(fd, STATUS_NEXT)) {
free(check_path);
return NULL;
}
File* file = file_create(check_path); File* file = file_create(check_path);
free(check_path); free(check_path);
if (file == NULL) { send_status(fd, STATUS_ERROR); return NULL; } if (file == NULL) {
send_status(fd, STATUS_ERROR);
return NULL;
}
if (config->use_metadata) { if (config->use_metadata) {
int meta_ok = 1; int meta_ok = 1;
file->metadata = metadata_receive(fd, &meta_ok); file->metadata = metadata_receive(fd, &meta_ok);
if (!meta_ok) { file_destroy(file); send_status(fd, STATUS_ERROR); return NULL; } if (!meta_ok) {
file_destroy(file);
send_status(fd, STATUS_ERROR);
return NULL;
}
} }
Data* file_data = receive_data(fd); Data* file_data = receive_data(fd);
@@ -184,7 +204,11 @@ File *receive_incremental_check(int fd, Config *config, bool *skipped) { *skipp
if (config->use_compression) { if (config->use_compression) {
Data* uncompressed = data_decompress(file_data); Data* uncompressed = data_decompress(file_data);
data_destroy(file_data); data_destroy(file_data);
if (uncompressed == NULL) { file_destroy(file); send_status(fd, STATUS_ERROR); return NULL; } if (uncompressed == NULL) {
file_destroy(file);
send_status(fd, STATUS_ERROR);
return NULL;
}
file_data = uncompressed; file_data = uncompressed;
} }
@@ -219,8 +243,10 @@ bool to_disk(const char *path, const void *data, unsigned long long data_size) {
} }
bool file_send_sendfile(File* file, int file_descriptor, bool use_metadata, bool send_path) { bool file_send_sendfile(File* file, int file_descriptor, bool use_metadata, bool send_path) {
if (send_path && !send_str(file_descriptor, file->path)) return false; if (send_path && !send_str(file_descriptor, file->path))
if (use_metadata && !metadata_send(file_descriptor, file->metadata)) return false; return false;
if (use_metadata && !metadata_send(file_descriptor, file->metadata))
return false;
int fd = open(file->path, O_RDONLY); int fd = open(file->path, O_RDONLY);
if (fd == -1) { if (fd == -1) {
@@ -248,16 +274,21 @@ bool file_send_sendfile(File *file, int file_descriptor, bool use_metadata, bool
return true; return true;
} }
File *file_receive(Config *config, int file_descriptor) { File* file_receive(const Config* config, int file_descriptor) {
char* path = receive_str(file_descriptor); char* path = receive_str(file_descriptor);
if (path == NULL) return NULL; if (path == NULL)
return NULL;
File* file = file_create(path); File* file = file_create(path);
free(path); free(path);
if (file == NULL) return NULL; if (file == NULL)
return NULL;
if (config->use_metadata) { if (config->use_metadata) {
int meta_ok = 1; int meta_ok = 1;
file->metadata = metadata_receive(file_descriptor, &meta_ok); file->metadata = metadata_receive(file_descriptor, &meta_ok);
if (!meta_ok) { file_destroy(file); return NULL; } 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) {
@@ -284,8 +315,7 @@ size_t file_content_to_buffer(File *file) {
perror("Could not open the file!"); perror("Could not open the file!");
return 0; return 0;
} }
size_t bytes_read = size_t bytes_read = fread(file->data->data, 1, file->data->size, file_pointer);
fread(file->data->data, 1, file->data->size, file_pointer);
if (bytes_read != (size_t)file->data->size) { if (bytes_read != (size_t)file->data->size) {
fclose(file_pointer); fclose(file_pointer);
perror("Read unexpected number of bytes from File!"); perror("Read unexpected number of bytes from File!");
@@ -295,21 +325,22 @@ size_t file_content_to_buffer(File *file) {
return bytes_read; return bytes_read;
} }
int receive_manifest(int fd, Config *config, int *next_status) { int receive_manifest(int fd, const Config* config, int* next_status) {
int count; int count;
if (!receive_int(fd, &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(fd); 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(fd, next_status)) return -1; if (!receive_status(fd, next_status))
return -1;
return 0; return 0;
} }
+5 -4
View File
@@ -23,15 +23,16 @@ typedef struct {
File* file_create(const char* path); File* file_create(const char* path);
void file_destroy(void* item); 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(const Config* config, int file_descriptor);
bool file_send_single_calls(File *file, int file_descriptor, bool use_metadata, int compression_level, bool send_path); bool file_send_single_calls(File* file, int file_descriptor, bool use_metadata,
int compression_level, bool send_path);
bool file_send_sendfile(File* file, int file_descriptor, bool use_metadata, bool send_path); bool file_send_sendfile(File* file, int file_descriptor, bool use_metadata, bool send_path);
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(const 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); bool file_save_to_disk(const char* root_directory, File* file);
File* receive_incremental_check(int fd, Config* config, bool* skipped); File* receive_incremental_check(int fd, Config* config, bool* skipped);
int receive_manifest(int fd, Config *config, int *next_status); int receive_manifest(int fd, const Config* config, int* next_status);
#endif #endif
+3 -4
View File
@@ -14,11 +14,10 @@ void log_message(LogLevel log_level, char *format, ...) {
if (log_level < current_log_level) if (log_level < current_log_level)
return; return;
time_t now = time(NULL); time_t now = time(NULL);
struct tm *t = localtime(&now); const struct tm* t = localtime(&now);
fprintf(stderr, "%04d-%02d-%02d %02d:%02d:%02d [%s]: ", t->tm_year + 1900, fprintf(stderr, "%04d-%02d-%02d %02d:%02d:%02d [%s]: ", t->tm_year + 1900, t->tm_mon + 1,
t->tm_mon + 1, t->tm_mday, t->tm_hour, t->tm_min, t->tm_sec, t->tm_mday, t->tm_hour, t->tm_min, t->tm_sec, log_level_strings[log_level]);
log_level_strings[log_level]);
va_list args; va_list args;
va_start(args, format); va_start(args, format);
+1 -6
View File
@@ -1,12 +1,7 @@
#ifndef LOG_H #ifndef LOG_H
#define LOG_H #define LOG_H
typedef enum { typedef enum { LOG_LEVEL_DEBUG, LOG_LEVEL_INFO, LOG_LEVEL_WARNING, LOG_LEVEL_ERROR } LogLevel;
LOG_LEVEL_DEBUG,
LOG_LEVEL_INFO,
LOG_LEVEL_WARNING,
LOG_LEVEL_ERROR
} LogLevel;
void log_message(LogLevel log_level, char* message, ...); void log_message(LogLevel log_level, char* message, ...);
void set_log_level(LogLevel level); void set_log_level(LogLevel level);
+34 -16
View File
@@ -8,17 +8,22 @@
#include <time.h> #include <time.h>
#include <unistd.h> #include <unistd.h>
void metadata_to_buf(char **buf, FileMetadata *m) { void metadata_to_buf(char** buf, const FileMetadata* m) {
int present = (m != NULL) ? 1 : 0; int present = (m != NULL) ? 1 : 0;
memcpy(*buf, &present, sizeof(int)); memcpy(*buf, &present, sizeof(int));
*buf += sizeof(int); *buf += sizeof(int);
if (m == NULL) if (m == NULL)
return; return;
memcpy(*buf, &m->mode, sizeof(mode_t)); *buf += sizeof(mode_t); memcpy(*buf, &m->mode, sizeof(mode_t));
memcpy(*buf, &m->uid, sizeof(uid_t)); *buf += sizeof(uid_t); *buf += sizeof(mode_t);
memcpy(*buf, &m->gid, sizeof(gid_t)); *buf += sizeof(gid_t); memcpy(*buf, &m->uid, sizeof(uid_t));
memcpy(*buf, &m->mtime_sec, sizeof(time_t)); *buf += sizeof(time_t); *buf += sizeof(uid_t);
memcpy(*buf, &m->mtime_nsec, sizeof(long)); *buf += sizeof(long); memcpy(*buf, &m->gid, sizeof(gid_t));
*buf += sizeof(gid_t);
memcpy(*buf, &m->mtime_sec, sizeof(time_t));
*buf += sizeof(time_t);
memcpy(*buf, &m->mtime_nsec, sizeof(long));
*buf += sizeof(long);
} }
FileMetadata* metadata_from_buf(char** buf) { FileMetadata* metadata_from_buf(char** buf) {
@@ -28,11 +33,16 @@ FileMetadata *metadata_from_buf(char **buf) {
if (!present) if (!present)
return NULL; return NULL;
FileMetadata* m = malloc(sizeof(FileMetadata)); FileMetadata* m = malloc(sizeof(FileMetadata));
memcpy(&m->mode, *buf, sizeof(mode_t)); *buf += sizeof(mode_t); memcpy(&m->mode, *buf, sizeof(mode_t));
memcpy(&m->uid, *buf, sizeof(uid_t)); *buf += sizeof(uid_t); *buf += sizeof(mode_t);
memcpy(&m->gid, *buf, sizeof(gid_t)); *buf += sizeof(gid_t); memcpy(&m->uid, *buf, sizeof(uid_t));
memcpy(&m->mtime_sec, *buf, sizeof(time_t)); *buf += sizeof(time_t); *buf += sizeof(uid_t);
memcpy(&m->mtime_nsec, *buf, sizeof(long)); *buf += sizeof(long); memcpy(&m->gid, *buf, sizeof(gid_t));
*buf += sizeof(gid_t);
memcpy(&m->mtime_sec, *buf, sizeof(time_t));
*buf += sizeof(time_t);
memcpy(&m->mtime_nsec, *buf, sizeof(long));
*buf += sizeof(long);
return m; return m;
} }
@@ -53,25 +63,33 @@ bool metadata_send(int file_descriptor, FileMetadata *m) {
FileMetadata* metadata_receive(int file_descriptor, int* ok) { 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; if (ok)
*ok = 0;
return NULL; return NULL;
} }
if (!present) { if (!present) {
if (ok) *ok = 1; if (ok)
*ok = 1;
return NULL; return NULL;
} }
FileMetadata* m = malloc(sizeof(FileMetadata)); FileMetadata* m = malloc(sizeof(FileMetadata));
if (m == NULL) { if (ok) *ok = 0; 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; if (ok)
*ok = 0;
return NULL; return NULL;
} }
if (ok) *ok = 1; if (ok)
*ok = 1;
return m; return m;
} }
+3 -2
View File
@@ -5,9 +5,10 @@
#include <stdbool.h> #include <stdbool.h>
#include <sys/stat.h> #include <sys/stat.h>
#define FILE_METADATA_WIRE_SIZE (sizeof(mode_t) + sizeof(uid_t) + sizeof(gid_t) + sizeof(time_t) + sizeof(long)) #define FILE_METADATA_WIRE_SIZE \
(sizeof(mode_t) + sizeof(uid_t) + sizeof(gid_t) + sizeof(time_t) + sizeof(long))
void metadata_to_buf(char **buf, FileMetadata *m); void metadata_to_buf(char** buf, const 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, int* ok); FileMetadata* metadata_receive(int file_descriptor, int* ok);
+24 -25
View File
@@ -13,11 +13,11 @@
#include <string.h> #include <string.h>
#include <threads.h> #include <threads.h>
PipelineContextSender *pipeline_context_sender_create(Config *config, PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* queue_scanner,
Queue *queue_scanner,
Queue* queue_loader) { Queue* queue_loader) {
PipelineContextSender* context = malloc(sizeof(PipelineContextSender)); PipelineContextSender* context = malloc(sizeof(PipelineContextSender));
if (context == NULL) return NULL; if (context == NULL)
return NULL;
context->config = config; context->config = config;
context->queue_scanner = queue_scanner; context->queue_scanner = queue_scanner;
context->queue_loader = queue_loader; context->queue_loader = queue_loader;
@@ -53,11 +53,11 @@ void pipeline_context_sender_destroy(PipelineContextSender *context) {
free(context); free(context);
} }
PipelineContextReceiver *pipeline_context_receiver_create(Config *config, PipelineContextReceiver* pipeline_context_receiver_create(Config* config, Queue* queue,
Queue *queue,
int file_descriptor) { int file_descriptor) {
PipelineContextReceiver* context = malloc(sizeof(PipelineContextReceiver)); PipelineContextReceiver* context = malloc(sizeof(PipelineContextReceiver));
if (context == NULL) return NULL; if (context == NULL)
return NULL;
context->config = config; context->config = config;
context->queue = queue; context->queue = queue;
context->file_descriptor = file_descriptor; context->file_descriptor = file_descriptor;
@@ -81,40 +81,39 @@ void pipeline_context_receiver_destroy(PipelineContextReceiver *context) {
free(context); free(context);
} }
static void receive_chunk_enqueue(int file_descriptor, static void receive_chunk_enqueue(int file_descriptor, PipelineContextReceiver* context) {
PipelineContextReceiver *context) {
Chunk* chunk = receive_chunk_data(file_descriptor, context->config); Chunk* chunk = receive_chunk_data(file_descriptor, context->config);
if (chunk == NULL) return; if (chunk == NULL)
return;
for (int i = 0; i < chunk->element_count; i++) { for (int i = 0; i < chunk->element_count; i++) {
File* file = chunk->items[i]; File* file = chunk->items[i];
chunk->items[i] = NULL; chunk->items[i] = NULL;
queue_enqueue_multithreaded(context->queue, file, &context->mutex, queue_enqueue_multithreaded(context->queue, file, &context->mutex,
&context->condition_not_empty, &context->condition_not_empty, &context->condition_not_full);
&context->condition_not_full);
} }
chunk_destroy(chunk); chunk_destroy(chunk);
} }
int receive_thread(void* pipeline_context) { int receive_thread(void* pipeline_context) {
PipelineContextReceiver *context = PipelineContextReceiver* context = (PipelineContextReceiver*)pipeline_context;
(PipelineContextReceiver *)pipeline_context;
mtx_lock(&context->mutex); mtx_lock(&context->mutex);
int file_descriptor = context->file_descriptor; int file_descriptor = context->file_descriptor;
Config* config = context->config; Config* config = context->config;
mtx_unlock(&context->mutex); mtx_unlock(&context->mutex);
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 || status == STATUS_CHECK) { while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK) {
if (status == STATUS_CHECK) { if (status == STATUS_CHECK) {
bool skipped; bool skipped;
File* file = receive_incremental_check(file_descriptor, config, &skipped); File* file = receive_incremental_check(file_descriptor, config, &skipped);
if (!skipped) { if (!skipped) {
if (file == NULL) return thrd_error; if (file == NULL)
return thrd_error;
queue_enqueue_multithreaded(context->queue, file, &context->mutex, queue_enqueue_multithreaded(context->queue, file, &context->mutex,
&context->condition_not_empty, &context->condition_not_empty, &context->condition_not_full);
&context->condition_not_full);
} }
} else if (status == STATUS_CHUNK) { } else if (status == STATUS_CHUNK) {
receive_chunk_enqueue(file_descriptor, context); receive_chunk_enqueue(file_descriptor, context);
@@ -122,16 +121,17 @@ int receive_thread(void *pipeline_context) {
File* file = file_receive(config, file_descriptor); File* file = file_receive(config, file_descriptor);
if (file) { if (file) {
queue_enqueue_multithreaded(context->queue, file, &context->mutex, queue_enqueue_multithreaded(context->queue, file, &context->mutex,
&context->condition_not_empty, &context->condition_not_empty, &context->condition_not_full);
&context->condition_not_full);
} else { } else {
log_message(LOG_LEVEL_ERROR, "Failed to receive file"); log_message(LOG_LEVEL_ERROR, "Failed to receive file");
} }
} }
if (!receive_status(file_descriptor, &status)) return thrd_error; if (!receive_status(file_descriptor, &status))
return thrd_error;
} }
if (status == STATUS_MANIFEST) { if (status == STATUS_MANIFEST) {
if (receive_manifest(file_descriptor, config, &status) != 0) return thrd_error; if (receive_manifest(file_descriptor, config, &status) != 0)
return thrd_error;
} }
mtx_lock(&context->mutex); mtx_lock(&context->mutex);
context->receiver_done = true; context->receiver_done = true;
@@ -141,16 +141,15 @@ int receive_thread(void *pipeline_context) {
} }
int write_thread(void* pipeline_context) { int write_thread(void* pipeline_context) {
PipelineContextReceiver *context = PipelineContextReceiver* context = (PipelineContextReceiver*)pipeline_context;
(PipelineContextReceiver *)pipeline_context;
mtx_lock(&context->mutex); mtx_lock(&context->mutex);
bool save_to_disk = context->config->save_to_disk; bool save_to_disk = context->config->save_to_disk;
char* root_directory = str_dup(context->config->receive_root_directory); char* root_directory = str_dup(context->config->receive_root_directory);
mtx_unlock(&context->mutex); mtx_unlock(&context->mutex);
while (true) { while (true) {
File *file = queue_dequeue_multithreaded( File* file =
context->queue, &context->mutex, &context->condition_not_empty, queue_dequeue_multithreaded(context->queue, &context->mutex, &context->condition_not_empty,
&context->condition_not_full, &context->receiver_done); &context->condition_not_full, &context->receiver_done);
if (file == NULL) { if (file == NULL) {
free(root_directory); free(root_directory);
+2 -4
View File
@@ -33,12 +33,10 @@ typedef struct PipelineContextReceiver {
bool receiver_done; bool receiver_done;
} PipelineContextReceiver; } PipelineContextReceiver;
PipelineContextSender *pipeline_context_sender_create(Config *config, PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* queue_scanner,
Queue *queue_scanner,
Queue* queue_loader); Queue* queue_loader);
void pipeline_context_sender_destroy(PipelineContextSender* context); void pipeline_context_sender_destroy(PipelineContextSender* context);
PipelineContextReceiver *pipeline_context_receiver_create(Config *config, PipelineContextReceiver* pipeline_context_receiver_create(Config* config, Queue* queue_receiver,
Queue *queue_receiver,
int file_descriptor); int file_descriptor);
void pipeline_context_receiver_destroy(PipelineContextReceiver* context); void pipeline_context_receiver_destroy(PipelineContextReceiver* context);
int receive_thread(void* pipeline_context); int receive_thread(void* pipeline_context);
+26 -16
View File
@@ -28,13 +28,14 @@ void io_set_bwlimit(unsigned long long bytes_per_sec) {
} }
static void bw_throttle(size_t bytes_written) { static void bw_throttle(size_t bytes_written) {
if (io_bwlimit == 0) return; if (io_bwlimit == 0)
return;
struct timespec now; struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now); clock_gettime(CLOCK_MONOTONIC, &now);
long long elapsed_ns = (now.tv_sec - bw_last_refill.tv_sec) * 1000000000LL + long long elapsed_ns =
(now.tv_nsec - bw_last_refill.tv_nsec); (now.tv_sec - bw_last_refill.tv_sec) * 1000000000LL + (now.tv_nsec - bw_last_refill.tv_nsec);
bw_last_refill = now; bw_last_refill = now;
long long tokens_to_add = (long long)((double)io_bwlimit * elapsed_ns / 1000000000.0); long long tokens_to_add = (long long)((double)io_bwlimit * elapsed_ns / 1000000000.0);
@@ -95,11 +96,11 @@ bool receive_n_data(int file_descriptor, void *data, size_t data_size) {
while (total_bytes_received < data_size) { while (total_bytes_received < data_size) {
ssize_t bytes_received; ssize_t bytes_received;
if (io_ssl) if (io_ssl)
bytes_received = SSL_read(io_ssl, (char *)data + total_bytes_received, bytes_received =
data_size - total_bytes_received); SSL_read(io_ssl, (char*)data + total_bytes_received, data_size - total_bytes_received);
else else
bytes_received = read(fd, (char *)data + total_bytes_received, bytes_received =
data_size - total_bytes_received); read(fd, (char*)data + total_bytes_received, data_size - total_bytes_received);
if (bytes_received <= 0) { if (bytes_received <= 0) {
if (bytes_received == 0) if (bytes_received == 0)
log_message(LOG_LEVEL_ERROR, "Connection closed while receiving data"); log_message(LOG_LEVEL_ERROR, "Connection closed while receiving data");
@@ -134,17 +135,21 @@ static const char *status_to_string(Status status) {
bool send_str(int file_descriptor, char* data) { bool send_str(int file_descriptor, char* data) {
size_t size = strlen(data); size_t size = strlen(data);
if (!send_n_data(file_descriptor, &size, sizeof(size_t))) return false; if (!send_n_data(file_descriptor, &size, sizeof(size_t)))
if (!send_n_data(file_descriptor, data, size)) return false; return false;
if (!send_n_data(file_descriptor, data, size))
return false;
log_message(LOG_LEVEL_DEBUG, "Send String: %s", data); log_message(LOG_LEVEL_DEBUG, "Send String: %s", data);
return true; return true;
} }
char* receive_str(int file_descriptor) { char* receive_str(int file_descriptor) {
size_t size; size_t size;
if (!receive_n_data(file_descriptor, &size, sizeof(size_t))) return NULL; if (!receive_n_data(file_descriptor, &size, sizeof(size_t)))
return NULL;
char* data = (char*)malloc(size + 1); char* data = (char*)malloc(size + 1);
if (data == NULL) return NULL; if (data == NULL)
return NULL;
if (!receive_n_data(file_descriptor, data, size)) { if (!receive_n_data(file_descriptor, data, size)) {
free(data); free(data);
return NULL; return NULL;
@@ -169,7 +174,8 @@ Data *receive_data(int file_descriptor) {
if (!receive_n_data(file_descriptor, &size, sizeof(unsigned long long))) if (!receive_n_data(file_descriptor, &size, sizeof(unsigned long long)))
return NULL; return NULL;
void* data = malloc((size_t)size); void* data = malloc((size_t)size);
if (data == NULL) return NULL; if (data == NULL)
return NULL;
if (!receive_n_data(file_descriptor, data, (size_t)size)) { if (!receive_n_data(file_descriptor, data, (size_t)size)) {
free(data); free(data);
return NULL; return NULL;
@@ -179,25 +185,29 @@ Data *receive_data(int file_descriptor) {
} }
bool send_int(int file_descriptor, int data) { bool send_int(int file_descriptor, int data) {
if (!send_n_data(file_descriptor, &data, sizeof(int))) return false; if (!send_n_data(file_descriptor, &data, sizeof(int)))
return false;
log_message(LOG_LEVEL_DEBUG, "Send Int: %d", data); log_message(LOG_LEVEL_DEBUG, "Send Int: %d", data);
return true; return true;
} }
bool receive_int(int file_descriptor, int* data) { bool receive_int(int file_descriptor, int* data) {
if (!receive_n_data(file_descriptor, data, sizeof(int))) return false; if (!receive_n_data(file_descriptor, data, sizeof(int)))
return false;
log_message(LOG_LEVEL_DEBUG, "Received Int: %d", *data); log_message(LOG_LEVEL_DEBUG, "Received Int: %d", *data);
return true; return true;
} }
bool send_status(int file_descriptor, Status status) { bool send_status(int file_descriptor, Status status) {
if (!send_n_data(file_descriptor, &status, sizeof(Status))) return false; if (!send_n_data(file_descriptor, &status, sizeof(Status)))
return false;
log_message(LOG_LEVEL_DEBUG, "Send Status: %s", status_to_string(status)); log_message(LOG_LEVEL_DEBUG, "Send Status: %s", status_to_string(status));
return true; return true;
} }
bool receive_status(int file_descriptor, Status* status) { bool receive_status(int file_descriptor, Status* status) {
if (!receive_n_data(file_descriptor, status, sizeof(Status))) return false; if (!receive_n_data(file_descriptor, status, sizeof(Status)))
return false;
log_message(LOG_LEVEL_DEBUG, "Received Status: %s", status_to_string(*status)); log_message(LOG_LEVEL_DEBUG, "Received Status: %s", status_to_string(*status));
return true; return true;
} }
+9 -1
View File
@@ -8,7 +8,15 @@
typedef struct ssl_st SSL; typedef struct ssl_st SSL;
typedef int Status; typedef int Status;
enum NET_STATUS { STATUS_OK, STATUS_ERROR, STATUS_FINISHED, STATUS_NEXT, STATUS_CHUNK, STATUS_MANIFEST, STATUS_CHECK }; 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); void io_set_bwlimit(unsigned long long bytes_per_sec);
+11 -11
View File
@@ -46,20 +46,21 @@ void queue_destroy(Queue *queue) {
free(queue); free(queue);
} }
bool queue_is_empty(Queue *queue) { bool queue_is_empty(const Queue* queue) {
if (queue == NULL) if (queue == NULL)
return true; return true;
return queue->size == 0; return queue->size == 0;
} }
bool queue_is_full(Queue *queue) { bool queue_is_full(const Queue* queue) {
if (queue == NULL) if (queue == NULL)
return false; return false;
return queue->size == queue->capacity; return queue->size == queue->capacity;
} }
static bool queue_double_capacity(Queue* queue) { static bool queue_double_capacity(Queue* queue) {
if (queue == NULL) return false; if (queue == NULL)
return false;
unsigned int new_capacity = queue->capacity * 2; unsigned int new_capacity = queue->capacity * 2;
if (new_capacity <= 1) if (new_capacity <= 1)
new_capacity = 100; new_capacity = 100;
@@ -79,9 +80,11 @@ static bool queue_double_capacity(Queue *queue) {
} }
bool queue_enqueue(Queue* queue, void* item) { bool queue_enqueue(Queue* queue, void* item) {
if (queue == NULL || item == NULL) return false; if (queue == NULL || item == NULL)
return false;
if (queue_is_full(queue)) { if (queue_is_full(queue)) {
if (!queue_double_capacity(queue)) return false; if (!queue_double_capacity(queue))
return false;
} }
queue->items[queue->rear] = item; queue->items[queue->rear] = item;
queue->rear = (queue->rear + 1) % queue->capacity; queue->rear = (queue->rear + 1) % queue->capacity;
@@ -89,8 +92,7 @@ bool queue_enqueue(Queue *queue, void *item) {
return true; return true;
} }
bool queue_enqueue_multithreaded(Queue *queue, void *item, mtx_t *mutex, bool queue_enqueue_multithreaded(Queue* queue, void* item, mtx_t* mutex, cnd_t* condition_not_empty,
cnd_t *condition_not_empty,
cnd_t* condition_not_full) { cnd_t* condition_not_full) {
mtx_lock(mutex); mtx_lock(mutex);
while (queue_is_full(queue)) while (queue_is_full(queue))
@@ -114,10 +116,8 @@ void *queue_dequeue(Queue *queue) {
return item; return item;
} }
void *queue_dequeue_multithreaded(Queue *queue, mtx_t *mutex, void* queue_dequeue_multithreaded(Queue* queue, mtx_t* mutex, cnd_t* condition_not_empty,
cnd_t *condition_not_empty, cnd_t* condition_not_full, const bool* other_thread_done) {
cnd_t *condition_not_full,
bool *other_thread_done) {
mtx_lock(mutex); mtx_lock(mutex);
while (queue_is_empty(queue) && !*other_thread_done) while (queue_is_empty(queue) && !*other_thread_done)
cnd_wait(condition_not_empty, mutex); cnd_wait(condition_not_empty, mutex);
+5 -8
View File
@@ -15,16 +15,13 @@ typedef struct Queue {
Queue* queue_create(int capacity, void (*destroyer)(void* item)); Queue* queue_create(int capacity, void (*destroyer)(void* item));
void queue_destroy(Queue* queue); void queue_destroy(Queue* queue);
bool queue_is_empty(Queue *queue); bool queue_is_empty(const Queue* queue);
bool queue_is_full(Queue *queue); bool queue_is_full(const Queue* queue);
bool queue_enqueue(Queue* queue, void* item); bool queue_enqueue(Queue* queue, void* item);
bool queue_enqueue_multithreaded(Queue *queue, void *item, mtx_t *mutex, bool queue_enqueue_multithreaded(Queue* queue, void* item, mtx_t* mutex, cnd_t* condition_not_empty,
cnd_t *condition_not_empty,
cnd_t* condition_not_full); cnd_t* condition_not_full);
void* queue_dequeue(Queue* queue); void* queue_dequeue(Queue* queue);
void *queue_dequeue_multithreaded(Queue *queue, mtx_t *mutex, void* queue_dequeue_multithreaded(Queue* queue, mtx_t* mutex, cnd_t* condition_not_empty,
cnd_t *condition_not_empty, cnd_t* condition_not_full, const bool* other_thread_done);
cnd_t *condition_not_full,
bool *other_thread_done);
#endif #endif
+19 -10
View File
@@ -15,34 +15,39 @@ typedef struct {
static int parse_remote_dest(const char* dest, RemoteDest* r) { static int parse_remote_dest(const char* dest, RemoteDest* r) {
const char* colon = strchr(dest, ':'); const char* colon = strchr(dest, ':');
if (!colon) return -1; if (!colon)
return -1;
size_t remote_path_len = strlen(colon + 1); size_t remote_path_len = strlen(colon + 1);
if (remote_path_len >= sizeof(r->remote_path)) return -1; if (remote_path_len >= sizeof(r->remote_path))
return -1;
memcpy(r->remote_path, colon + 1, remote_path_len + 1); memcpy(r->remote_path, colon + 1, remote_path_len + 1);
const char* at = memchr(dest, '@', colon - dest); const char* at = memchr(dest, '@', colon - dest);
if (at) { if (at) {
size_t user_len = at - dest; size_t user_len = at - dest;
if (user_len >= sizeof(r->user)) return -1; if (user_len >= sizeof(r->user))
return -1;
memcpy(r->user, dest, user_len); memcpy(r->user, dest, user_len);
r->user[user_len] = '\0'; r->user[user_len] = '\0';
size_t host_len = colon - at - 1; size_t host_len = colon - at - 1;
if (host_len >= sizeof(r->host)) return -1; if (host_len >= sizeof(r->host))
return -1;
memcpy(r->host, at + 1, host_len); memcpy(r->host, at + 1, host_len);
r->host[host_len] = '\0'; r->host[host_len] = '\0';
} else { } else {
r->user[0] = '\0'; r->user[0] = '\0';
size_t host_len = colon - dest; size_t host_len = colon - dest;
if (host_len >= sizeof(r->host)) return -1; if (host_len >= sizeof(r->host))
return -1;
memcpy(r->host, dest, host_len); memcpy(r->host, dest, host_len);
r->host[host_len] = '\0'; r->host[host_len] = '\0';
} }
return 0; return 0;
} }
Client *client_connect_ssh(char *destination, int port) { Client* client_connect_ssh(const char* destination, int port) {
RemoteDest r; RemoteDest r;
if (parse_remote_dest(destination, &r) != 0) { if (parse_remote_dest(destination, &r) != 0) {
fprintf(stderr, "Invalid remote destination: %s\n", destination); fprintf(stderr, "Invalid remote destination: %s\n", destination);
@@ -64,15 +69,18 @@ Client *client_connect_ssh(char *destination, int port) {
int exec_pipe[2]; int exec_pipe[2];
if (pipe(exec_pipe) < 0) { if (pipe(exec_pipe) < 0) {
perror("pipe failed"); perror("pipe failed");
close(sv[0]); close(sv[1]); close(sv[0]);
close(sv[1]);
return NULL; return NULL;
} }
pid_t pid = fork(); pid_t pid = fork();
if (pid < 0) { if (pid < 0) {
perror("fork failed"); perror("fork failed");
close(sv[0]); close(sv[1]); close(sv[0]);
close(exec_pipe[0]); close(exec_pipe[1]); close(sv[1]);
close(exec_pipe[0]);
close(exec_pipe[1]);
return NULL; return NULL;
} }
@@ -85,7 +93,8 @@ Client *client_connect_ssh(char *destination, int port) {
dup2(sv[1], STDIN_FILENO); dup2(sv[1], STDIN_FILENO);
if (sv[1] != STDOUT_FILENO) if (sv[1] != STDOUT_FILENO)
dup2(sv[1], STDOUT_FILENO); dup2(sv[1], STDOUT_FILENO);
if (sv[1] > 1) close(sv[1]); if (sv[1] > 1)
close(sv[1]);
char ssh_user[512]; char ssh_user[512];
if (r.user[0] != '\0') if (r.user[0] != '\0')
+1 -1
View File
@@ -3,6 +3,6 @@
#include "transport_tcp.h" #include "transport_tcp.h"
Client *client_connect_ssh(char *destination, int port); Client* client_connect_ssh(const char* destination, int port);
#endif #endif
+19 -19
View File
@@ -26,8 +26,7 @@ Server *server_create(int port) {
} }
server->file_descriptor = file_descriptor; server->file_descriptor = file_descriptor;
int opt = 1; int opt = 1;
if (setsockopt(server->file_descriptor, SOL_SOCKET, SO_REUSEADDR, &opt, if (setsockopt(server->file_descriptor, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt))) {
sizeof(opt))) {
perror("Error setting a socket option!"); perror("Error setting a socket option!");
close(server->file_descriptor); close(server->file_descriptor);
free(server); free(server);
@@ -40,8 +39,8 @@ Server *server_create(int port) {
server->address_length = sizeof(server->address); server->address_length = sizeof(server->address);
server->ssl_ctx = NULL; server->ssl_ctx = NULL;
if (bind(server->file_descriptor, (struct sockaddr *)&server->address, if (bind(server->file_descriptor, (struct sockaddr*)&server->address, server->address_length) <
server->address_length) < 0) { 0) {
perror("Could not bind server"); perror("Could not bind server");
close(server->file_descriptor); close(server->file_descriptor);
free(server); free(server);
@@ -52,7 +51,8 @@ Server *server_create(int port) {
} }
void server_delete(Server** server) { void server_delete(Server** server) {
if (server == NULL || *server == NULL) return; if (server == NULL || *server == NULL)
return;
close((*server)->file_descriptor); close((*server)->file_descriptor);
if ((*server)->ssl_ctx) { if ((*server)->ssl_ctx) {
SSL_CTX_free((*server)->ssl_ctx); SSL_CTX_free((*server)->ssl_ctx);
@@ -62,8 +62,8 @@ void server_delete(Server **server) {
*server = NULL; *server = NULL;
} }
static void accept_loop(Server *server, void (*child_fn)(int, void *), static void accept_loop(Server* server, void (*child_fn)(int, void*), void* child_ctx,
void *child_ctx, const char *log_fmt) { const char* log_fmt) {
if (listen(server->file_descriptor, SOMAXCONN) < 0) { if (listen(server->file_descriptor, SOMAXCONN) < 0) {
perror("Could not listen on port!"); perror("Could not listen on port!");
return; return;
@@ -72,8 +72,7 @@ static void accept_loop(Server *server, void (*child_fn)(int, void *),
while (1) { while (1) {
struct sockaddr_in client_addr; struct sockaddr_in client_addr;
socklen_t client_len = sizeof(client_addr); socklen_t client_len = sizeof(client_addr);
int fd = accept(server->file_descriptor, (struct sockaddr *)&client_addr, int fd = accept(server->file_descriptor, (struct sockaddr*)&client_addr, &client_len);
&client_len);
if (fd < 0) { if (fd < 0) {
perror("Could not accept the connection"); perror("Could not accept the connection");
continue; continue;
@@ -90,24 +89,24 @@ static void accept_loop(Server *server, void (*child_fn)(int, void *),
} }
} }
struct plain_ctx { void (*handler)(int); }; struct plain_ctx {
void (*handler)(int);
};
static void plain_child_fn(int fd, void* ctx) { static void plain_child_fn(int fd, void* ctx) {
((struct plain_ctx*)ctx)->handler(fd); ((struct plain_ctx*)ctx)->handler(fd);
} }
bool server_listen(Server* server, void (*handler)(int file_descriptor)) { bool server_listen(Server* server, void (*handler)(int file_descriptor)) {
log_message(LOG_LEVEL_INFO, "Start Listening on Port: %d", log_message(LOG_LEVEL_INFO, "Start Listening on Port: %d", ntohs(server->address.sin_port));
ntohs(server->address.sin_port));
struct plain_ctx ctx = {handler}; struct plain_ctx ctx = {handler};
accept_loop(server, plain_child_fn, &ctx, "Received Connection"); accept_loop(server, plain_child_fn, &ctx, "Received Connection");
return true; return true;
} }
void server_accept_loop(Server *server, void (*child_fn)(int, void *), void server_accept_loop(Server* server, void (*child_fn)(int, void*), void* child_ctx,
void *child_ctx, const char *log_fmt) { const char* log_fmt) {
log_message(LOG_LEVEL_INFO, "Start TLS Listening on Port: %d", log_message(LOG_LEVEL_INFO, "Start TLS Listening on Port: %d", ntohs(server->address.sin_port));
ntohs(server->address.sin_port));
accept_loop(server, child_fn, child_ctx, log_fmt); accept_loop(server, child_fn, child_ctx, log_fmt);
} }
@@ -140,8 +139,8 @@ bool client_connect(Client *client, char *host, int port) {
return false; return false;
} }
if (connect(client->file_descriptor, (struct sockaddr *)&client->address, if (connect(client->file_descriptor, (struct sockaddr*)&client->address, client->address_length) <
client->address_length) < 0) { 0) {
perror("Could not connect to Server!"); perror("Could not connect to Server!");
return false; return false;
} }
@@ -164,7 +163,8 @@ void client_disconnect(Client *client) {
} }
void client_delete(Client* client) { void client_delete(Client* client) {
if (client == NULL) return; if (client == NULL)
return;
if (client->ssl_ctx) { if (client->ssl_ctx) {
SSL_CTX_free(client->ssl_ctx); SSL_CTX_free(client->ssl_ctx);
client->ssl_ctx = NULL; client->ssl_ctx = NULL;
+2 -2
View File
@@ -23,8 +23,8 @@ typedef struct Client {
Server* server_create(int port); Server* server_create(int port);
bool server_listen(Server* server, void (*handler)(int file_descriptor)); bool server_listen(Server* server, void (*handler)(int file_descriptor));
void server_accept_loop(Server *server, void (*child_fn)(int, void *), void server_accept_loop(Server* server, void (*child_fn)(int, void*), void* child_ctx,
void *child_ctx, const char *log_fmt); const char* log_fmt);
void server_delete(Server** server); void server_delete(Server** server);
Client* client_create(); Client* client_create();
bool client_connect(Client* client, char* host, int port); bool client_connect(Client* client, char* host, int port);
+17 -18
View File
@@ -31,10 +31,9 @@ static void log_ssl_errors(void) {
} }
} }
static SSL_CTX *create_ssl_ctx(bool is_server, const char *cert, static SSL_CTX* create_ssl_ctx(bool is_server, const char* cert, const char* key,
const char *key, const char *ca_path) { const char* ca_path) {
const SSL_METHOD *method = const SSL_METHOD* method = is_server ? TLS_server_method() : TLS_client_method();
is_server ? TLS_server_method() : TLS_client_method();
SSL_CTX* ctx = SSL_CTX_new(method); SSL_CTX* ctx = SSL_CTX_new(method);
if (!ctx) { if (!ctx) {
log_message(LOG_LEVEL_ERROR, "Unable to create SSL context"); log_message(LOG_LEVEL_ERROR, "Unable to create SSL context");
@@ -58,8 +57,7 @@ static SSL_CTX *create_ssl_ctx(bool is_server, const char *cert,
return NULL; return NULL;
} }
if (!SSL_CTX_check_private_key(ctx)) { if (!SSL_CTX_check_private_key(ctx)) {
log_message(LOG_LEVEL_ERROR, log_message(LOG_LEVEL_ERROR, "Private key does not match certificate");
"Private key does not match certificate");
SSL_CTX_free(ctx); SSL_CTX_free(ctx);
return NULL; return NULL;
} }
@@ -93,8 +91,7 @@ static SSL *wrap_fd_with_ssl(int fd, SSL_CTX *ctx, bool is_server) {
ret = SSL_connect(ssl); ret = SSL_connect(ssl);
if (ret <= 0) { if (ret <= 0) {
log_message(LOG_LEVEL_ERROR, "SSL %s failed", log_message(LOG_LEVEL_ERROR, "SSL %s failed", is_server ? "accept" : "connect");
is_server ? "accept" : "connect");
log_ssl_errors(); log_ssl_errors();
SSL_free(ssl); SSL_free(ssl);
return NULL; return NULL;
@@ -102,10 +99,11 @@ static SSL *wrap_fd_with_ssl(int fd, SSL_CTX *ctx, bool is_server) {
return ssl; return ssl;
} }
bool server_create_tls(Server *server, const char *cert_path, bool server_create_tls(Server* server, const char* cert_path, const char* key_path,
const char *key_path, const char *ca_path) { const char* ca_path) {
SSL_CTX* ctx = create_ssl_ctx(true, cert_path, key_path, ca_path); SSL_CTX* ctx = create_ssl_ctx(true, cert_path, key_path, ca_path);
if (!ctx) return false; if (!ctx)
return false;
server->ssl_ctx = ctx; server->ssl_ctx = ctx;
return true; return true;
} }
@@ -118,7 +116,8 @@ struct tls_child_ctx {
static void tls_child_fn(int fd, void* arg) { static void tls_child_fn(int fd, void* arg) {
struct tls_child_ctx* ctx = (struct tls_child_ctx*)arg; struct tls_child_ctx* ctx = (struct tls_child_ctx*)arg;
SSL* ssl = wrap_fd_with_ssl(fd, ctx->ssl_ctx, true); SSL* ssl = wrap_fd_with_ssl(fd, ctx->ssl_ctx, true);
if (!ssl) return; if (!ssl)
return;
io_set_ssl(ssl); io_set_ssl(ssl);
ctx->handler(fd); ctx->handler(fd);
SSL_shutdown(ssl); SSL_shutdown(ssl);
@@ -132,22 +131,22 @@ bool server_listen_tls(Server *server, void (*handler)(int file_descriptor)) {
return true; return true;
} }
bool client_connect_tls(Client *client, char *host, int port, bool client_connect_tls(Client* client, char* host, int port, const char* cert_path,
const char *cert_path, const char *key_path, const char* key_path, const char* ca_path) {
const char *ca_path) {
client->address.sin_port = htons(port); client->address.sin_port = htons(port);
if (inet_pton(AF_INET, host, &client->address.sin_addr) <= 0) { if (inet_pton(AF_INET, host, &client->address.sin_addr) <= 0) {
perror("Could not convert host address!"); perror("Could not convert host address!");
return false; return false;
} }
if (connect(client->file_descriptor, (struct sockaddr *)&client->address, if (connect(client->file_descriptor, (struct sockaddr*)&client->address, client->address_length) <
client->address_length) < 0) { 0) {
perror("Could not connect to Server!"); perror("Could not connect to Server!");
return false; return false;
} }
SSL_CTX* ctx = create_ssl_ctx(false, cert_path, key_path, ca_path); SSL_CTX* ctx = create_ssl_ctx(false, cert_path, key_path, ca_path);
if (!ctx) return false; if (!ctx)
return false;
client->ssl_ctx = ctx; client->ssl_ctx = ctx;
SSL* ssl = wrap_fd_with_ssl(client->file_descriptor, ctx, false); SSL* ssl = wrap_fd_with_ssl(client->file_descriptor, ctx, false);
+4 -5
View File
@@ -6,11 +6,10 @@
bool tls_global_init(void); bool tls_global_init(void);
bool server_create_tls(Server *server, const char *cert_path, bool server_create_tls(Server* server, const char* cert_path, const char* key_path,
const char *key_path, const char *ca_path);
bool server_listen_tls(Server *server, void (*handler)(int file_descriptor));
bool client_connect_tls(Client *client, char *host, int port,
const char *cert_path, const char *key_path,
const char* ca_path); const char* ca_path);
bool server_listen_tls(Server* server, void (*handler)(int file_descriptor));
bool client_connect_tls(Client* client, char* host, int port, const char* cert_path,
const char* key_path, const char* ca_path);
#endif #endif
+12 -8
View File
@@ -8,12 +8,16 @@
#include <sys/stat.h> #include <sys/stat.h>
#include <unistd.h> #include <unistd.h>
bool mkdir_r(char *path) { bool mkdir_r(const char* path) {
char* path_duplicate = malloc(strlen(path) + 1); char* path_duplicate = malloc(strlen(path) + 1);
if (!path_duplicate) return false; if (!path_duplicate)
return false;
strcpy(path_duplicate, path); strcpy(path_duplicate, path);
char* path_current = (char*)malloc((strlen(path) + 2) * sizeof(char)); char* path_current = (char*)malloc((strlen(path) + 2) * sizeof(char));
if (!path_current) { free(path_duplicate); return false; } if (!path_current) {
free(path_duplicate);
return false;
}
char* path_current_position = path_current; char* path_current_position = path_current;
if (path[0] == '/') { if (path[0] == '/') {
strcpy(path_current, "/"); strcpy(path_current, "/");
@@ -22,7 +26,7 @@ bool mkdir_r(char *path) {
path_current[0] = '\0'; path_current[0] = '\0';
} }
const char* delimiter = "/"; const char* delimiter = "/";
char *part = strtok(path_duplicate, delimiter); const char* part = strtok(path_duplicate, delimiter);
bool ok = true; bool ok = true;
while (part != NULL) { while (part != NULL) {
strcpy(path_current_position, part); strcpy(path_current_position, part);
@@ -77,8 +81,7 @@ bool glob_match(const char *pattern, const char *str) {
return *str == '\0'; return *str == '\0';
} }
static void delete_extras_walk(const char *abs_path, const char *rel_path, static void delete_extras_walk(const char* abs_path, const char* rel_path, ArrayList* manifest) {
ArrayList *manifest) {
DIR* dir = opendir(abs_path); DIR* dir = opendir(abs_path);
if (!dir) if (!dir)
return; return;
@@ -121,7 +124,7 @@ void delete_extras(const char *dest_root, ArrayList *manifest) {
delete_extras_walk(dest_root, "", manifest); delete_extras_walk(dest_root, "", manifest);
} }
char *path_cat(char *path1, char *path2) { char* path_cat(const char* path1, char* path2) {
if (path1 == NULL || *path1 == '\0') if (path1 == NULL || *path1 == '\0')
return str_dup(path2); return str_dup(path2);
if (path2 == NULL || *path2 == '\0') if (path2 == NULL || *path2 == '\0')
@@ -136,7 +139,8 @@ char *path_cat(char *path1, char *path2) {
path2_len -= 1; path2_len -= 1;
} }
char* new_path = malloc(path1_len + path2_len + 2); char* new_path = malloc(path1_len + path2_len + 2);
if (new_path == NULL) return NULL; if (new_path == NULL)
return NULL;
memcpy(new_path, path1, path1_len); memcpy(new_path, path1, path1_len);
new_path[path1_len] = '/'; new_path[path1_len] = '/';
memcpy(new_path + path1_len + 1, path2_pointer, path2_len); memcpy(new_path + path1_len + 1, path2_pointer, path2_len);
+2 -2
View File
@@ -4,9 +4,9 @@
#include "array_list.h" #include "array_list.h"
#include <stdbool.h> #include <stdbool.h>
bool mkdir_r(char *path); bool mkdir_r(const char* path);
char* str_dup(const char* string); char* str_dup(const char* string);
char *path_cat(char *path1, char *path2); char* path_cat(const char* path1, char* path2);
bool glob_match(const char* pattern, const char* str); bool glob_match(const char* pattern, const char* str);
void delete_extras(const char* dest_root, ArrayList* manifest); void delete_extras(const char* dest_root, ArrayList* manifest);
+1 -1
View File
@@ -9,7 +9,7 @@
#include <unistd.h> #include <unistd.h>
static void test_data_compress_decompress_roundtrip() { static void test_data_compress_decompress_roundtrip() {
char original[] = "Hello, World! This is test data for compression round-trip!"; const char original[] = "Hello, World! This is test data for compression round-trip!";
size_t len = strlen(original); size_t len = strlen(original);
char* buf = malloc(len); char* buf = malloc(len);
+12 -12
View File
@@ -7,8 +7,8 @@
#include <stdlib.h> #include <stdlib.h>
static void test_config_lifecycle() { static void test_config_lifecycle() {
Config *cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("/dst"), Config* cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("/dst"), true, true, false,
true, true, false, false, false, 1, false, 0); false, false, 1, false, 0);
EXPECT_NOT_NULL(cfg); EXPECT_NOT_NULL(cfg);
EXPECT_EQ_STR(cfg->version, "1.0"); EXPECT_EQ_STR(cfg->version, "1.0");
EXPECT_EQ_STR(cfg->send_directory, "/src"); EXPECT_EQ_STR(cfg->send_directory, "/src");
@@ -23,8 +23,8 @@ static void test_config_lifecycle() {
} }
static void test_config_ssh_dest() { static void test_config_ssh_dest() {
Config *cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("user@host:/dst"), Config* cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("user@host:/dst"), true,
true, false, false, false, false, 1, false, 0); false, false, false, false, 1, false, 0);
EXPECT_NOT_NULL(cfg); EXPECT_NOT_NULL(cfg);
EXPECT_EQ_INT(cfg->transport, TRANSPORT_TCP); EXPECT_EQ_INT(cfg->transport, TRANSPORT_TCP);
EXPECT_NULL(cfg->ssh_destination); EXPECT_NULL(cfg->ssh_destination);
@@ -38,8 +38,8 @@ static void test_config_ssh_dest() {
} }
static void test_config_ssh_dest_local_path() { static void test_config_ssh_dest_local_path() {
Config *cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("/local/path"), Config* cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("/local/path"), true, false,
true, false, false, false, false, 1, false, 0); false, false, false, 1, false, 0);
config_parse_ssh_dest(cfg); config_parse_ssh_dest(cfg);
EXPECT_EQ_INT(cfg->transport, TRANSPORT_TCP); EXPECT_EQ_INT(cfg->transport, TRANSPORT_TCP);
EXPECT_NULL(cfg->ssh_destination); EXPECT_NULL(cfg->ssh_destination);
@@ -48,8 +48,8 @@ static void test_config_ssh_dest_local_path() {
} }
static void test_config_ssh_dest_no_user() { static void test_config_ssh_dest_no_user() {
Config *cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("host:/remote"), Config* cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("host:/remote"), true, false,
true, false, false, false, false, 1, false, 0); false, false, false, 1, false, 0);
config_parse_ssh_dest(cfg); config_parse_ssh_dest(cfg);
EXPECT_EQ_INT(cfg->transport, TRANSPORT_SSH); EXPECT_EQ_INT(cfg->transport, TRANSPORT_SSH);
EXPECT_EQ_STR(cfg->ssh_destination, "host:/remote"); EXPECT_EQ_STR(cfg->ssh_destination, "host:/remote");
@@ -58,8 +58,8 @@ static void test_config_ssh_dest_no_user() {
} }
static void test_pipeline_sender_lifecycle() { static void test_pipeline_sender_lifecycle() {
Config *cfg = config_create(str_dup("2.0"), str_dup("/src2"), Config* cfg = config_create(str_dup("2.0"), str_dup("/src2"), str_dup("/dst2"), false, false,
str_dup("/dst2"), false, false, true, true, false, 1, false, 0); true, true, false, 1, false, 0);
Queue* q1 = queue_create(5, NULL); Queue* q1 = queue_create(5, NULL);
Queue* q2 = queue_create(15, NULL); Queue* q2 = queue_create(15, NULL);
@@ -75,8 +75,8 @@ static void test_pipeline_sender_lifecycle() {
} }
static void test_pipeline_receiver_lifecycle() { static void test_pipeline_receiver_lifecycle() {
Config *cfg = config_create(str_dup("3.0"), str_dup("/src3"), Config* cfg = config_create(str_dup("3.0"), str_dup("/src3"), str_dup("/dst3"), true, true, true,
str_dup("/dst3"), true, true, true, true, false, 1, false, 0); true, false, 1, false, 0);
Queue* q = queue_create(20, NULL); Queue* q = queue_create(20, NULL);
PipelineContextReceiver* pcr = pipeline_context_receiver_create(cfg, q, 42); PipelineContextReceiver* pcr = pipeline_context_receiver_create(cfg, q, 42);
+10 -11
View File
@@ -44,7 +44,7 @@ static void test_queue_basic() {
EXPECT_EQ_INT(q->size, 6); EXPECT_EQ_INT(q->size, 6);
int expected_vals[] = {30, 40, 50, 60, 70, 80}; const int expected_vals[] = {30, 40, 50, 60, 70, 80};
for (int i = 0; i < 6; i++) { for (int i = 0; i < 6; i++) {
int* v = (int*)queue_dequeue(q); int* v = (int*)queue_dequeue(q);
EXPECT_NOT_NULL(v); EXPECT_NOT_NULL(v);
@@ -69,7 +69,7 @@ static void test_queue_resize() {
EXPECT_TRUE(queue_is_full(q)); EXPECT_TRUE(queue_is_full(q));
int *v1 = (int *)queue_dequeue(q); const int* v1 = (const int*)queue_dequeue(q);
EXPECT_NOT_NULL(v1); EXPECT_NOT_NULL(v1);
EXPECT_EQ_INT(*v1, 1); EXPECT_EQ_INT(*v1, 1);
@@ -84,19 +84,19 @@ static void test_queue_resize() {
EXPECT_EQ_INT(q->size, 4); EXPECT_EQ_INT(q->size, 4);
// Dequeue all and check order: B, C, D, E // Dequeue all and check order: B, C, D, E
int *v2 = (int *)queue_dequeue(q); const int* v2 = (const int*)queue_dequeue(q);
EXPECT_NOT_NULL(v2); EXPECT_NOT_NULL(v2);
EXPECT_EQ_INT(*v2, 2); EXPECT_EQ_INT(*v2, 2);
int *v3 = (int *)queue_dequeue(q); const int* v3 = (const int*)queue_dequeue(q);
EXPECT_NOT_NULL(v3); EXPECT_NOT_NULL(v3);
EXPECT_EQ_INT(*v3, 3); EXPECT_EQ_INT(*v3, 3);
int *v4 = (int *)queue_dequeue(q); const int* v4 = (const int*)queue_dequeue(q);
EXPECT_NOT_NULL(v4); EXPECT_NOT_NULL(v4);
EXPECT_EQ_INT(*v4, 4); EXPECT_EQ_INT(*v4, 4);
int *v5 = (int *)queue_dequeue(q); const int* v5 = (const int*)queue_dequeue(q);
EXPECT_NOT_NULL(v5); EXPECT_NOT_NULL(v5);
EXPECT_EQ_INT(*v5, 5); EXPECT_EQ_INT(*v5, 5);
@@ -142,7 +142,8 @@ typedef struct {
static int consumer_func(void* arg) { static int consumer_func(void* arg) {
ThreadContext* ctx = (ThreadContext*)arg; ThreadContext* ctx = (ThreadContext*)arg;
while (true) { while (true) {
int *val = (int *)queue_dequeue_multithreaded(ctx->q, ctx->mutex, ctx->cnd_empty, ctx->cnd_full, &ctx->done); int* val = (int*)queue_dequeue_multithreaded(ctx->q, ctx->mutex, ctx->cnd_empty, ctx->cnd_full,
&ctx->done);
if (val == NULL) { if (val == NULL) {
break; break;
} }
@@ -162,14 +163,12 @@ static void test_queue_multithreaded() {
cnd_init(&cnd_empty); cnd_init(&cnd_empty);
cnd_init(&cnd_full); cnd_init(&cnd_full);
ThreadContext ctx = { ThreadContext ctx = {.q = q,
.q = q,
.mutex = &mutex, .mutex = &mutex,
.cnd_empty = &cnd_empty, .cnd_empty = &cnd_empty,
.cnd_full = &cnd_full, .cnd_full = &cnd_full,
.done = false, .done = false,
.sum = 0 .sum = 0};
};
thrd_t consumer; thrd_t consumer;
int res = thrd_create(&consumer, consumer_func, &ctx); int res = thrd_create(&consumer, consumer_func, &ctx);
+17 -11
View File
@@ -18,7 +18,8 @@ static void test_scanner_single_file() {
mkdir(dir, 0755); mkdir(dir, 0755);
create_test_file(file1, content1); create_test_file(file1, content1);
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0, NULL, 0, NULL, 0, 0, 0); DirectoryScanner* scanner =
directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 0, 0);
EXPECT_NOT_NULL(scanner); EXPECT_NOT_NULL(scanner);
Chunk* chunk = directory_scanner_next(scanner); Chunk* chunk = directory_scanner_next(scanner);
@@ -26,7 +27,7 @@ static void test_scanner_single_file() {
EXPECT_EQ_INT(chunk->element_count, 1); EXPECT_EQ_INT(chunk->element_count, 1);
EXPECT_EQ_STR(chunk->items[0]->path, file1); EXPECT_EQ_STR(chunk->items[0]->path, file1);
Chunk *next = directory_scanner_next(scanner); const Chunk* next = directory_scanner_next(scanner);
EXPECT_NULL(next); EXPECT_NULL(next);
chunk_destroy(chunk); chunk_destroy(chunk);
@@ -46,25 +47,28 @@ static void test_scanner_multiple_files() {
create_test_file(file1, content1); create_test_file(file1, content1);
create_test_file(file2, content2); create_test_file(file2, content2);
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0, NULL, 0, NULL, 0, 0, 0); DirectoryScanner* scanner =
directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 0, 0);
EXPECT_NOT_NULL(scanner); EXPECT_NOT_NULL(scanner);
Chunk *chunk = directory_scanner_next(scanner); const Chunk* chunk = directory_scanner_next(scanner);
EXPECT_NOT_NULL(chunk); EXPECT_NOT_NULL(chunk);
EXPECT_EQ_INT(chunk->element_count, 2); EXPECT_EQ_INT(chunk->element_count, 2);
int found1 = 0, found2 = 0; int found1 = 0, found2 = 0;
for (int i = 0; i < chunk->element_count; i++) { for (int i = 0; i < chunk->element_count; i++) {
if (strcmp(chunk->items[i]->path, file1) == 0) found1 = 1; if (strcmp(chunk->items[i]->path, file1) == 0)
if (strcmp(chunk->items[i]->path, file2) == 0) found2 = 1; found1 = 1;
if (strcmp(chunk->items[i]->path, file2) == 0)
found2 = 1;
} }
EXPECT_TRUE(found1); EXPECT_TRUE(found1);
EXPECT_TRUE(found2); EXPECT_TRUE(found2);
Chunk *next = directory_scanner_next(scanner); const Chunk* next = directory_scanner_next(scanner);
EXPECT_NULL(next); EXPECT_NULL(next);
chunk_destroy(chunk); chunk_destroy((void*)chunk);
directory_scanner_destroy(scanner); directory_scanner_destroy(scanner);
unlink(file1); unlink(file1);
unlink(file2); unlink(file2);
@@ -83,7 +87,8 @@ static void test_scanner_subdirectory() {
create_test_file(root_file, content); create_test_file(root_file, content);
create_test_file(sub_file, content); create_test_file(sub_file, content);
DirectoryScanner *scanner = directory_scanner_create((char *)root, false, 0, NULL, 0, NULL, 0, 0, 0); DirectoryScanner* scanner =
directory_scanner_create((char*)root, false, 0, NULL, 0, NULL, 0, 0, 0);
EXPECT_NOT_NULL(scanner); EXPECT_NOT_NULL(scanner);
int total_files = 0; int total_files = 0;
@@ -106,10 +111,11 @@ static void test_scanner_empty_directory() {
mkdir(dir, 0755); mkdir(dir, 0755);
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0, NULL, 0, NULL, 0, 0, 0); DirectoryScanner* scanner =
directory_scanner_create((char*)dir, false, 0, NULL, 0, NULL, 0, 0, 0);
EXPECT_NOT_NULL(scanner); EXPECT_NOT_NULL(scanner);
Chunk *chunk = directory_scanner_next(scanner); const Chunk* chunk = directory_scanner_next(scanner);
EXPECT_NULL(chunk); EXPECT_NULL(chunk);
directory_scanner_destroy(scanner); directory_scanner_destroy(scanner);
+1 -1
View File
@@ -6,7 +6,7 @@
void test_shared_utils() { void test_shared_utils() {
// Test str_dup // Test str_dup
char *dup_null = str_dup(NULL); const char* dup_null = str_dup(NULL);
EXPECT_NULL(dup_null); EXPECT_NULL(dup_null);
char* dup_empty = str_dup(""); char* dup_empty = str_dup("");
+12 -6
View File
@@ -29,7 +29,8 @@ extern bool current_test_failed;
#define EXPECT_TRUE(condition) \ #define EXPECT_TRUE(condition) \
do { \ do { \
if (!(condition)) { \ if (!(condition)) { \
printf(" \033[1;31m[FAIL]\033[0m %s:%d: Assertion failed: %s is false\n", __FILE__, __LINE__, #condition); \ printf(" \033[1;31m[FAIL]\033[0m %s:%d: Assertion failed: %s is false\n", __FILE__, \
__LINE__, #condition); \
current_test_failed = true; \ current_test_failed = true; \
return; \ return; \
} \ } \
@@ -38,7 +39,8 @@ extern bool current_test_failed;
#define EXPECT_FALSE(condition) \ #define EXPECT_FALSE(condition) \
do { \ do { \
if (condition) { \ if (condition) { \
printf(" \033[1;31m[FAIL]\033[0m %s:%d: Assertion failed: %s is true\n", __FILE__, __LINE__, #condition); \ printf(" \033[1;31m[FAIL]\033[0m %s:%d: Assertion failed: %s is true\n", __FILE__, \
__LINE__, #condition); \
current_test_failed = true; \ current_test_failed = true; \
return; \ return; \
} \ } \
@@ -49,7 +51,8 @@ extern bool current_test_failed;
int act = (actual); \ int act = (actual); \
int exp = (expected); \ int exp = (expected); \
if (act != exp) { \ if (act != exp) { \
printf(" \033[1;31m[FAIL]\033[0m %s:%d: Expected %d, got %d\n", __FILE__, __LINE__, exp, act); \ printf(" \033[1;31m[FAIL]\033[0m %s:%d: Expected %d, got %d\n", __FILE__, __LINE__, exp, \
act); \
current_test_failed = true; \ current_test_failed = true; \
return; \ return; \
} \ } \
@@ -67,7 +70,8 @@ extern bool current_test_failed;
return; \ return; \
} \ } \
} else if (strcmp(act, exp) != 0) { \ } else if (strcmp(act, exp) != 0) { \
printf(" \033[1;31m[FAIL]\033[0m %s:%d: Expected \"%s\", got \"%s\"\n", __FILE__, __LINE__, exp, act); \ printf(" \033[1;31m[FAIL]\033[0m %s:%d: Expected \"%s\", got \"%s\"\n", __FILE__, \
__LINE__, exp, act); \
current_test_failed = true; \ current_test_failed = true; \
return; \ return; \
} \ } \
@@ -76,7 +80,8 @@ extern bool current_test_failed;
#define EXPECT_NOT_NULL(ptr) \ #define EXPECT_NOT_NULL(ptr) \
do { \ do { \
if ((ptr) == NULL) { \ if ((ptr) == NULL) { \
printf(" \033[1;31m[FAIL]\033[0m %s:%d: Expected non-null pointer, got NULL\n", __FILE__, __LINE__); \ printf(" \033[1;31m[FAIL]\033[0m %s:%d: Expected non-null pointer, got NULL\n", __FILE__, \
__LINE__); \
current_test_failed = true; \ current_test_failed = true; \
return; \ return; \
} \ } \
@@ -85,7 +90,8 @@ extern bool current_test_failed;
#define EXPECT_NULL(ptr) \ #define EXPECT_NULL(ptr) \
do { \ do { \
if ((ptr) != NULL) { \ if ((ptr) != NULL) { \
printf(" \033[1;31m[FAIL]\033[0m %s:%d: Expected NULL, got %p\n", __FILE__, __LINE__, (void*)(ptr)); \ printf(" \033[1;31m[FAIL]\033[0m %s:%d: Expected NULL, got %p\n", __FILE__, __LINE__, \
(void*)(ptr)); \
current_test_failed = true; \ current_test_failed = true; \
return; \ return; \
} \ } \