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

Multi-client server, protocol versioning, streaming decompression #11

Merged
TapTap merged 2 commits from server-improvements into main 2026-07-16 15:15:20 +02:00
Conversation 0 Commits 2 Files Changed 7
+552 -263
27 changed files with 552 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit a62894be05 - Show all commits
src/client/client_cli.c
+11
View File
@@ -28,6 +28,9 @@ static void print_usage(void) {
printf(" --progress Show transfer progress\n");
printf(" --delete Delete files on receiver not in source\n");
printf(" --exclude <pattern> Exclude files matching pattern\n");
printf(" --include <pattern> Only include files matching pattern\n");
printf(" --max-size <n> Skip files larger than n bytes\n");
printf(" --min-size <n> Skip files smaller than n bytes\n");
printf(" -m Enable multithreading\n");
printf(" -s Enable chunk serialization\n");
printf(" -f Enable sendfile (TCP only, not with -c or -s)\n");
@@ -78,6 +81,14 @@ int main(int argc, char *argv[]) {
int idx = config->exclude_count++;
config->exclude_patterns = realloc(config->exclude_patterns, config->exclude_count * sizeof(char *));
config->exclude_patterns[idx] = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--include") == 0 && i + 1 < argc) {
int idx = config->include_count++;
config->include_patterns = realloc(config->include_patterns, config->include_count * sizeof(char *));
config->include_patterns[idx] = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--max-size") == 0 && i + 1 < argc) {
config->max_size = strtoull(argv[++i], NULL, 10);
} else if (strcmp(argv[i], "--min-size") == 0 && i + 1 < argc) {
config->min_size = strtoull(argv[++i], NULL, 10);
} else if (strcmp(argv[i], "-c") == 0 || strcmp(argv[i], "-z") == 0) {
config->use_compression = true;
log_message(LOG_LEVEL_INFO, "Enabled Compression");
src/client/client_send.c
+83 -26
View File
@@ -20,26 +20,29 @@
int send_chunk(Client *client, Chunk *chunk, Config *config) {
if (config->use_chunk_serialization) {
send_status(client->file_descriptor, STATUS_CHUNK);
if (!send_status(client->file_descriptor, STATUS_CHUNK)) return -1;
Data *data;
if (config->use_compression) {
data = chunk_compress(chunk, config->compression_level, config->use_metadata);
} else {
data = chunk_serialize(chunk, config->use_metadata);
}
send_data(client->file_descriptor, data);
if (data == NULL) return -1;
if (!send_data(client->file_descriptor, data)) { data_destroy(data); return -1; }
data_destroy(data);
} else if (config->use_sendfile && !config->use_compression) {
for (int i = 0; i < chunk->element_count; i++) {
send_status(client->file_descriptor, STATUS_NEXT);
file_send_sendfile(chunk->items[i], client->file_descriptor, config->use_metadata);
if (!send_status(client->file_descriptor, STATUS_NEXT)) return -1;
if (!file_send_sendfile(chunk->items[i], client->file_descriptor, config->use_metadata))
return -1;
}
} else {
for (int i = 0; i < chunk->element_count; i++) {
send_status(client->file_descriptor, STATUS_NEXT);
file_send_single_calls(chunk->items[i], client->file_descriptor,
if (!send_status(client->file_descriptor, STATUS_NEXT)) return -1;
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))
return -1;
}
}
return 0;
@@ -56,9 +59,17 @@ static int send_chunks_multithreaded(void *pipeline_context) {
client = client_connect_ssh(context->config->ssh_destination, context->config->ssh_port);
} else {
client = client_create();
client_connect(client, server_host, server_port);
if (!client || !client_connect(client, server_host, server_port)) {
if (client) client_delete(client);
fprintf(stderr, "Error: could not connect to server\n");
return thrd_error;
}
}
if (!config_send(client->file_descriptor, context->config)) {
client_disconnect(client);
client_delete(client);
return thrd_error;
}
config_send(client->file_descriptor, context->config);
while (true) {
Chunk *current_chunk = queue_dequeue_multithreaded(
@@ -74,14 +85,17 @@ static int send_chunks_multithreaded(void *pipeline_context) {
(char *)context->manifest->items[i]);
}
send_status(client->file_descriptor, STATUS_FINISHED);
int ok = receive_status(client->file_descriptor) == STATUS_OK;
Status s;
int ok = receive_status(client->file_descriptor, &s) && s == STATUS_OK;
client_disconnect(client);
client_delete(client);
return ok ? thrd_success : thrd_error;
}
if (send_chunk(client, current_chunk, context->config) != 0) {
perror("Something unexpected happend while sending the chunk");
exit(EXIT_FAILURE);
fprintf(stderr, "Error: unexpected error while sending chunk\n");
client_disconnect(client);
client_delete(client);
return thrd_error;
}
chunk_destroy(current_chunk);
}
@@ -93,7 +107,9 @@ static int scan_directory_multithreaded(void *pipeline_context) {
DirectoryScanner *scanner = directory_scanner_create(
context->config->send_directory, context->config->use_metadata,
context->config->chunk_size, context->config->exclude_patterns,
context->config->exclude_count);
context->config->exclude_count, context->config->include_patterns,
context->config->include_count, context->config->max_size,
context->config->min_size);
mtx_unlock(&context->mutex_scanner);
Chunk *current_chunk;
@@ -136,8 +152,13 @@ static int load_files_multithreaded(void *pipeline_context) {
return thrd_success;
}
if (!context->config->use_sendfile) {
for (int i = 0; i < chunk->element_count; i++)
file_load_data(chunk->items[i]);
for (int i = 0; i < chunk->element_count; i++) {
if (!file_load_data(chunk->items[i])) {
log_message(LOG_LEVEL_ERROR, "Failed to load file data, skipping");
file_destroy(chunk->items[i]);
chunk->items[i] = NULL;
}
}
}
queue_enqueue_multithreaded(context->queue_loader, chunk,
&context->mutex_loader,
@@ -150,7 +171,9 @@ int send_files(Config *config) {
if (config->dry_run) {
DirectoryScanner *scanner = directory_scanner_create(
config->send_directory, config->use_metadata, config->chunk_size,
config->exclude_patterns, config->exclude_count);
config->exclude_patterns, config->exclude_count,
config->include_patterns, config->include_count,
config->max_size, config->min_size);
Chunk *chunk;
int file_count = 0;
unsigned long long total_bytes = 0;
@@ -177,14 +200,25 @@ int send_files(Config *config) {
return 1;
}
client = client_connect_ssh(config->ssh_destination, config->ssh_port);
if (!client) return 1;
} else {
client = client_create();
client_connect(client, server_host, server_port);
if (!client || !client_connect(client, server_host, server_port)) {
if (client) client_delete(client);
fprintf(stderr, "Error: could not connect to server\n");
return 1;
}
}
if (!config_send(client->file_descriptor, config)) {
client_disconnect(client);
client_delete(client);
return 1;
}
config_send(client->file_descriptor, config);
DirectoryScanner *scanner = directory_scanner_create(
config->send_directory, config->use_metadata, config->chunk_size,
config->exclude_patterns, config->exclude_count);
config->exclude_patterns, config->exclude_count,
config->include_patterns, config->include_count,
config->max_size, config->min_size);
Chunk *current_chunk;
unsigned long long total_bytes = 0;
time_t last_progress = 0;
@@ -201,10 +235,18 @@ int send_files(Config *config) {
}
}
if (!config->use_sendfile) {
for (int i = 0; i < current_chunk->element_count; i++)
file_load_data(current_chunk->items[i]);
for (int i = 0; i < current_chunk->element_count; i++) {
if (!file_load_data(current_chunk->items[i])) {
log_message(LOG_LEVEL_ERROR, "Failed to load file data");
continue;
}
}
}
if (send_chunk(client, current_chunk, config) != 0) {
log_message(LOG_LEVEL_ERROR, "Failed to send chunk");
chunk_destroy(current_chunk);
break;
}
send_chunk(client, current_chunk, config);
if (config->show_progress) {
total_bytes += chunk_bytes;
time_t now = time(NULL);
@@ -226,7 +268,8 @@ int send_files(Config *config) {
array_list_delete(manifest);
}
send_status(client->file_descriptor, STATUS_FINISHED);
int ok = receive_status(client->file_descriptor) == STATUS_OK;
Status s;
int ok = receive_status(client->file_descriptor, &s) && s == STATUS_OK;
if (config->show_progress) {
double elapsed = difftime(time(NULL), start);
double rate = elapsed > 0 ? total_bytes / (1048576.0 * elapsed) : 0;
@@ -242,7 +285,9 @@ int send_files_multithreaded(Config *config) {
if (config->dry_run) {
DirectoryScanner *scanner = directory_scanner_create(
config->send_directory, config->use_metadata, config->chunk_size,
config->exclude_patterns, config->exclude_count);
config->exclude_patterns, config->exclude_count,
config->include_patterns, config->include_count,
config->max_size, config->min_size);
Chunk *chunk;
int file_count = 0;
unsigned long long total_bytes = 0;
@@ -262,9 +307,20 @@ int send_files_multithreaded(Config *config) {
return 0;
}
Queue *q1 = queue_create(100, chunk_destroy);
Queue *q2 = queue_create(100, chunk_destroy);
if (!q1 || !q2) {
if (q1) queue_destroy(q1);
if (q2) queue_destroy(q2);
return 1;
}
PipelineContextSender *context =
pipeline_context_sender_create(config, queue_create(100, chunk_destroy),
queue_create(100, chunk_destroy));
pipeline_context_sender_create(config, q1, q2);
if (!context) {
queue_destroy(q1);
queue_destroy(q2);
return 1;
}
if (config->use_delete)
context->manifest = array_list_create(free);
@@ -275,6 +331,7 @@ int send_files_multithreaded(Config *config) {
thrd_create(&sender, send_chunks_multithreaded, context) !=
thrd_success) {
perror("Error creating threads.\n");
pipeline_context_sender_destroy(context);
return 1;
}
src/client/scanner.c
+34 -3
View File
@@ -11,7 +11,7 @@
#include <sys/stat.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) {
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));
scanner->directories = queue_create(100, free);
scanner->current_dir = NULL;
@@ -20,6 +20,10 @@ DirectoryScanner *directory_scanner_create(char *root_directory, bool use_metada
scanner->chunk_size = chunk_size > 0 ? chunk_size : DESIRED_CHUNK_SIZE;
scanner->exclude_patterns = exclude_patterns;
scanner->exclude_count = exclude_count;
scanner->include_patterns = include_patterns;
scanner->include_count = include_count;
scanner->max_size = max_size;
scanner->min_size = min_size;
queue_enqueue(scanner->directories, str_dup(root_directory));
return scanner;
}
@@ -58,8 +62,10 @@ static int open_next_directory(DirectoryScanner *scanner) {
scanner->current_path = (char *)queue_dequeue(scanner->directories);
scanner->current_dir = opendir(scanner->current_path);
if (scanner->current_dir == NULL) {
perror("Could not open directory!");
exit(EXIT_FAILURE);
perror("Could not open directory");
free(scanner->current_path);
scanner->current_path = NULL;
return 0;
}
return 1;
}
@@ -107,7 +113,32 @@ Chunk *directory_scanner_next(DirectoryScanner *scanner) {
free(cur_path);
continue;
}
if (scanner->include_count > 0) {
bool included = false;
for (int i = 0; i < scanner->include_count; i++) {
if (glob_match(scanner->include_patterns[i], entry->d_name)) {
included = true;
break;
}
}
if (!included) {
free(cur_path);
continue;
}
}
if ((scanner->max_size > 0 && (unsigned long long)stats.st_size > scanner->max_size) ||
(scanner->min_size > 0 && (unsigned long long)stats.st_size < scanner->min_size)) {
free(cur_path);
continue;
}
File *file = file_create(cur_path);
if (file == NULL) {
free(cur_path);
continue;
}
file->data->size = stats.st_size;
if (scanner->use_metadata)
file->metadata = file_metadata_create(&stats);
src/client/scanner.h
+5 -1
View File
@@ -14,9 +14,13 @@ typedef struct {
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;
DirectoryScanner *directory_scanner_create(char *root_directory, bool use_metadata, unsigned long long chunk_size, char **exclude_patterns, int exclude_count);
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);
void directory_scanner_destroy(DirectoryScanner *scanner);
src/server/server.c
+56 -10
View File
@@ -18,10 +18,16 @@
#include <string.h>
int receive_files(Config *config, int file_descriptor) {
Status status = receive_status(file_descriptor);
Status status;
if (!receive_status(file_descriptor, &status)) return -1;
while (status == STATUS_NEXT || status == STATUS_CHUNK) {
if (status == STATUS_CHUNK) {
Data *chunk_data = receive_data(file_descriptor);
if (chunk_data == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to receive chunk data");
send_status(file_descriptor, STATUS_ERROR);
return -1;
}
Data *data_to_process = chunk_data;
if (config->use_compression) {
data_to_process = data_decompress(chunk_data);
@@ -43,33 +49,50 @@ int receive_files(Config *config, int file_descriptor) {
for (int i = 0; i < chunk->element_count; i++) {
if (config->save_to_disk) {
char *disk_path = path_cat(config->receive_root_directory, chunk->items[i]->path);
if (disk_path) {
to_disk(disk_path, chunk->items[i]->data->data, chunk->items[i]->data->size);
file_restore_metadata(disk_path, chunk->items[i]->metadata);
free(disk_path);
}
}
}
chunk_destroy(chunk);
} else {
File *file = file_receive(config, file_descriptor);
if (file == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to receive file");
send_status(file_descriptor, STATUS_ERROR);
return -1;
}
if (config->save_to_disk) {
char *disk_path = path_cat(config->receive_root_directory, file->path);
if (disk_path) {
to_disk(disk_path, file->data->data, file->data->size);
file_restore_metadata(disk_path, file->metadata);
free(disk_path);
}
}
file_destroy(file);
}
status = receive_status(file_descriptor);
if (!receive_status(file_descriptor, &status)) {
send_status(file_descriptor, STATUS_ERROR);
return -1;
}
}
if (status == STATUS_MANIFEST) {
int count = receive_int(file_descriptor);
int count;
if (!receive_int(file_descriptor, &count)) return -1;
ArrayList *manifest = array_list_create(free);
for (int i = 0; i < count; i++)
array_list_add(manifest, receive_str(file_descriptor));
if (manifest) {
for (int i = 0; i < count; i++) {
char *s = receive_str(file_descriptor);
if (s) array_list_add(manifest, s);
}
fprintf(stderr, "Deleting files not in manifest...\n");
delete_extras(config->receive_root_directory, manifest);
array_list_delete(manifest);
status = receive_status(file_descriptor);
}
if (!receive_status(file_descriptor, &status)) return -1;
}
if (status != STATUS_FINISHED) {
log_message(LOG_LEVEL_ERROR, "Did not receive FINISHED Status");
@@ -82,19 +105,38 @@ int receive_files(Config *config, int file_descriptor) {
void handler(int file_descriptor) {
Config *config = config_receive(file_descriptor);
if (config == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to receive config");
close(file_descriptor);
return;
}
if (config->use_multithreading) {
Queue *q = queue_create(100, file_destroy);
if (q == NULL) {
config_delete(config);
close(file_descriptor);
return;
}
PipelineContextReceiver *context = pipeline_context_receiver_create(
config, queue_create(100, file_destroy), file_descriptor);
config, q, file_descriptor);
if (context == NULL) {
queue_destroy(q);
config_delete(config);
close(file_descriptor);
return;
}
thrd_t receiver, writer;
if (thrd_create(&receiver, receive_thread, context) != thrd_success ||
thrd_create(&writer, write_thread, context) != thrd_success) {
perror("Error creating Threads!");
exit(EXIT_FAILURE);
perror("Error creating Threads");
pipeline_context_receiver_destroy(context);
close(file_descriptor);
return;
}
thrd_join(receiver, NULL);
thrd_join(writer, NULL);
pipeline_context_receiver_destroy(context);
send_status(file_descriptor, STATUS_OK);
pipeline_context_receiver_destroy(context);
} else
receive_files(config, file_descriptor);
close(file_descriptor);
@@ -124,6 +166,10 @@ int main(int argc, char *argv[]) {
signal(SIGINT, cleanup);
signal(SIGTERM, cleanup);
g_server = server_create(8080);
if (g_server == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to create server");
return 1;
}
server_listen(g_server, handler);
return 0;
}
src/shared/array_list.c
+16 -16
View File
@@ -6,15 +6,14 @@
ArrayList *array_list_create(void (*item_destroyer)(void *item)) {
ArrayList *list = (ArrayList *)malloc(sizeof(ArrayList));
if (list == NULL) {
perror("FATAL ERROR: Could not allocate memory for array list struct");
exit(EXIT_FAILURE);
perror("ERROR: Could not allocate memory for array list struct");
return NULL;
}
list->items = malloc(INITIAL_ARRAY_SIZE * sizeof(void *));
if (list->items == NULL) {
perror("FATAL ERROR: Could not allocate memory for list items");
free(list);
exit(EXIT_FAILURE);
return NULL;
}
list->size = 0;
list->capacity = INITIAL_ARRAY_SIZE;
@@ -35,29 +34,30 @@ void array_list_delete(ArrayList *array_list) {
free(array_list);
}
void array_list_extend(ArrayList *array_list) {
if (array_list == NULL)
return;
bool array_list_extend(ArrayList *array_list) {
if (array_list == NULL) return false;
int new_capacity = array_list->capacity * 2;
if (new_capacity == 0)
new_capacity = INITIAL_ARRAY_SIZE;
array_list->items = realloc(array_list->items, new_capacity * sizeof(void *));
if (array_list->items == NULL) {
perror("FATAL ERROR: Could not reallocate memory for array list struct");
exit(EXIT_FAILURE);
void *new_items = realloc(array_list->items, new_capacity * sizeof(void *));
if (new_items == NULL) {
perror("ERROR: Could not reallocate memory for array list items");
return false;
}
array_list->items = new_items;
array_list->capacity = new_capacity;
return true;
}
void array_list_add(ArrayList *array_list, void *item) {
if (array_list == NULL) {
return;
}
bool array_list_add(ArrayList *array_list, void *item) {
if (array_list == NULL) return false;
if (array_list->capacity == array_list->size) {
array_list_extend(array_list);
if (!array_list_extend(array_list))
return false;
}
array_list->items[array_list->size] = item;
array_list->size += 1;
return true;
}
void **array_list_to_array(ArrayList *array_list) {
src/shared/array_list.h
+4 -2
View File
@@ -1,6 +1,8 @@
#ifndef ARRAY_LIST_H
#define ARRAY_LIST_H
#include <stdbool.h>
#define INITIAL_ARRAY_SIZE 100
typedef struct ArrayList {
@@ -12,8 +14,8 @@ typedef struct ArrayList {
ArrayList *array_list_create(void (*item_destroyer)(void *item));
void array_list_delete(ArrayList *array_list);
void array_list_extend(ArrayList *array_list);
void array_list_add(ArrayList *array_list, void *item);
bool array_list_extend(ArrayList *array_list);
bool array_list_add(ArrayList *array_list, void *item);
void **array_list_to_array(ArrayList *array_list);
#endif
src/shared/chunk.c
+6 -6
View File
@@ -14,16 +14,14 @@
Chunk *chunk_create(File **items, int element_count) {
Chunk *chunk = (Chunk *)malloc(sizeof(Chunk));
if (chunk == NULL) {
perror("FATAL ERROR: Could not allocate memory for chunk structure");
exit(EXIT_FAILURE);
perror("ERROR: Could not allocate memory for chunk structure");
return NULL;
}
chunk->items = (File **)malloc(element_count * sizeof(File *));
if (chunk->items == NULL) {
perror("FATAL ERROR: Could not allocate memory for items of chunk "
"structure");
free(chunk);
exit(EXIT_FAILURE);
return NULL;
}
for (int i = 0; i < element_count; i++) {
@@ -62,7 +60,7 @@ Data *chunk_serialize(Chunk *chunk, bool use_metadata) {
if (data == NULL) {
log_message(LOG_LEVEL_ERROR,
"Could not allocate memory for chunk serialization");
exit(EXIT_FAILURE);
return NULL;
}
char *data_pointer = data->data;
for (int i = 0; i < chunk->element_count; i++) {
@@ -172,8 +170,10 @@ Chunk *chunk_deserialize(Data *data, bool use_metadata) {
Data *chunk_compress(Chunk *chunk, int compression_level, bool use_metadata) {
log_message(LOG_LEVEL_DEBUG, "Starting to compress chunk");
Data *serialized = chunk_serialize(chunk, use_metadata);
if (serialized == NULL) return NULL;
Data *compressed = data_compress(serialized, compression_level);
data_destroy(serialized);
if (compressed == NULL) return NULL;
log_message(LOG_LEVEL_DEBUG, "Chunk successfully compressed");
return compressed;
}
src/shared/compression.c
+6 -4
View File
@@ -10,10 +10,7 @@ Data *data_compress(Data *data_to_compress, int compression_level) {
log_message(LOG_LEVEL_DEBUG, "Starting to compress data");
size_t dst_size = ZSTD_compressBound(data_to_compress->size);
Data *compressed_data = data_create_empty(dst_size);
if (!compressed_data) {
log_message(LOG_LEVEL_ERROR, "Failed to allocate compression buffer");
return NULL;
}
if (compressed_data == NULL) return NULL;
ZSTD_CCtx *cctx = ZSTD_createCCtx();
if (!cctx) {
@@ -49,6 +46,11 @@ Data *data_decompress(Data *compressed_data) {
log_message(LOG_LEVEL_DEBUG, "Start to decompress data");
unsigned long long dst_size = ZSTD_getFrameContentSize(
compressed_data->data, compressed_data->size);
if (ZSTD_isError(dst_size)) {
log_message(LOG_LEVEL_ERROR, "Failed to get decompressed size: %s",
ZSTD_getErrorName(dst_size));
return NULL;
}
ZSTD_DCtx *dctx = ZSTD_createDCtx();
if (!dctx) {
src/shared/config.c
+63 -27
View File
@@ -1,4 +1,5 @@
#include "config.h"
#include "log.h"
#include "protocol.h"
#include "utils.h"
#include <stdbool.h>
@@ -33,6 +34,10 @@ Config *config_create(char *version, char *send_directory,
config->ssh_destination = NULL;
config->exclude_patterns = NULL;
config->exclude_count = 0;
config->include_patterns = NULL;
config->include_count = 0;
config->max_size = 0;
config->min_size = 0;
return config;
}
@@ -65,50 +70,70 @@ void config_delete(Config *config) {
for (int i = 0; i < config->exclude_count; i++)
free(config->exclude_patterns[i]);
free(config->exclude_patterns);
for (int i = 0; i < config->include_count; i++)
free(config->include_patterns[i]);
free(config->include_patterns);
free(config);
}
void config_send(int file_descriptor, Config *config) {
send_str(file_descriptor, config->version);
send_str(file_descriptor, config->send_directory);
send_str(file_descriptor, config->receive_root_directory);
send_int(file_descriptor, config->save_to_disk);
send_int(file_descriptor, config->use_multithreading);
send_int(file_descriptor, config->use_chunk_serialization);
send_int(file_descriptor, config->use_compression);
send_int(file_descriptor, config->use_metadata);
send_int(file_descriptor, config->compression_level);
send_int(file_descriptor, (int)config->chunk_size);
send_int(file_descriptor, config->use_sendfile);
send_int(file_descriptor, config->use_delete);
if (receive_status(file_descriptor) != STATUS_OK) {
perror("Error transmitting config!");
exit(EXIT_FAILURE);
bool config_send(int file_descriptor, Config *config) {
if (!send_str(file_descriptor, config->version)) return false;
if (!send_str(file_descriptor, config->send_directory)) return false;
if (!send_str(file_descriptor, config->receive_root_directory)) return false;
if (!send_int(file_descriptor, config->save_to_disk)) return false;
if (!send_int(file_descriptor, config->use_multithreading)) return false;
if (!send_int(file_descriptor, config->use_chunk_serialization)) 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;
Status status;
if (!receive_status(file_descriptor, &status)) return false;
if (status != STATUS_OK) {
log_message(LOG_LEVEL_ERROR, "Error transmitting config");
return false;
}
return true;
}
Config *config_receive(int file_descriptor) {
Config *config = (Config *)malloc(sizeof(Config));
if (config == NULL) return NULL;
config->version = receive_str(file_descriptor);
if (!config->version) { free(config); return NULL; }
if (strcmp(config->version, PROTOCOL_VERSION) != 0) {
fprintf(stderr, "Protocol version mismatch: client=%s, server=%s\n",
config->version, PROTOCOL_VERSION);
free(config->version);
free(config);
send_status(file_descriptor, STATUS_ERROR);
exit(EXIT_FAILURE);
return NULL;
}
config->send_directory = receive_str(file_descriptor);
if (!config->send_directory) { free(config->version); free(config); return NULL; }
config->receive_root_directory = receive_str(file_descriptor);
config->save_to_disk = receive_int(file_descriptor);
config->use_multithreading = receive_int(file_descriptor);
config->use_chunk_serialization = receive_int(file_descriptor);
config->use_compression = receive_int(file_descriptor);
config->use_metadata = receive_int(file_descriptor);
config->compression_level = receive_int(file_descriptor);
config->chunk_size = (unsigned long long)receive_int(file_descriptor);
config->use_sendfile = receive_int(file_descriptor);
config->use_delete = receive_int(file_descriptor);
if (!config->receive_root_directory) { free(config->version); free(config->send_directory); free(config); return NULL; }
int tmp;
if (!receive_int(file_descriptor, &tmp)) goto error;
config->save_to_disk = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error;
config->use_multithreading = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error;
config->use_chunk_serialization = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error;
config->use_compression = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error;
config->use_metadata = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error;
config->compression_level = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error;
config->chunk_size = (unsigned long long)tmp;
if (!receive_int(file_descriptor, &tmp)) goto error;
config->use_sendfile = tmp;
if (!receive_int(file_descriptor, &tmp)) goto error;
config->use_delete = tmp;
config->show_progress = false;
config->dry_run = false;
config->ssh_port = 22;
@@ -116,6 +141,17 @@ Config *config_receive(int file_descriptor) {
config->ssh_destination = NULL;
config->exclude_patterns = NULL;
config->exclude_count = 0;
send_status(file_descriptor, STATUS_OK);
config->include_patterns = NULL;
config->include_count = 0;
config->max_size = 0;
config->min_size = 0;
if (!send_status(file_descriptor, STATUS_OK)) goto error;
return config;
error:
free(config->version);
free(config->send_directory);
free(config->receive_root_directory);
free(config);
return NULL;
}
src/shared/config.h
+5 -1
View File
@@ -28,6 +28,10 @@ typedef struct Config {
char *ssh_destination;
char **exclude_patterns;
int exclude_count;
char **include_patterns;
int include_count;
unsigned long long max_size;
unsigned long long min_size;
} Config;
#define PROTOCOL_VERSION "1.0.0"
@@ -40,7 +44,7 @@ Config *config_create(char *version, char *send_directory,
int compression_level, bool use_sendfile,
unsigned long long chunk_size);
void config_delete(Config *config);
void config_send(int file_descriptor, Config *config);
bool config_send(int file_descriptor, Config *config);
Config *config_receive(int file_descriptor);
bool is_remote_dest(const char *s);
void config_parse_ssh_dest(Config *config);
src/shared/data.c
+4 -3
View File
@@ -6,7 +6,7 @@ Data *data_create_empty(size_t data_size) {
void *data = malloc(data_size);
if (data == NULL) {
log_message(LOG_LEVEL_ERROR, "Could not allocate memory for empty data");
exit(EXIT_FAILURE);
return NULL;
}
return data_create(data, data_size);
}
@@ -15,7 +15,7 @@ Data *data_create_reserve(size_t size) {
Data *d = malloc(sizeof(Data));
if (d == NULL) {
log_message(LOG_LEVEL_ERROR, "Could not allocate memory for data");
exit(EXIT_FAILURE);
return NULL;
}
d->data = NULL;
d->size = size;
@@ -26,7 +26,8 @@ Data *data_create(void *data, size_t data_size) {
Data *new_data = malloc(sizeof(Data));
if (new_data == NULL) {
log_message(LOG_LEVEL_ERROR, "Could not allocate memory for data");
exit(EXIT_FAILURE);
free(data);
return NULL;
}
new_data->data = data;
new_data->size = data_size;
src/shared/file.c
+58 -30
View File
@@ -22,20 +22,24 @@
File *file_create(const char *path) {
File *file = (File *)malloc(sizeof(File));
if (file == NULL) {
perror("FATAL ERROR: Could not allocate memory for file struct");
exit(EXIT_FAILURE);
perror("ERROR: Could not allocate memory for file struct");
return NULL;
}
int path_len = strlen(path);
file->path = (char *)malloc(path_len + 1);
if (file->path == NULL) {
perror("FATAL ERROR: Could not allocate memory for path file string");
free(file);
exit(EXIT_FAILURE);
return NULL;
}
strcpy(file->path, path);
file->data = data_create_reserve(0);
if (file->data == NULL) {
free(file->path);
free(file);
return NULL;
}
file->metadata = NULL;
return file;
}
@@ -56,8 +60,8 @@ void file_destroy(void *item) {
FileMetadata *file_metadata_create(struct stat *stats) {
FileMetadata *m = malloc(sizeof(FileMetadata));
if (m == NULL) {
perror("FATAL ERROR: Could not allocate memory for file metadata");
exit(EXIT_FAILURE);
perror("ERROR: Could not allocate memory for file metadata");
return NULL;
}
m->mode = stats->st_mode;
m->uid = stats->st_uid;
@@ -75,26 +79,30 @@ void file_metadata_destroy(void *metadata) {
free(metadata);
}
void file_load_data(File *file) {
if (file == NULL)
return;
bool file_load_data(File *file) {
if (file == NULL) return false;
if (file->data->data == NULL) {
file->data->data = malloc(file->data->size);
if (file->data->data == NULL) {
perror("Could not allocate memory for file data");
exit(EXIT_FAILURE);
return false;
}
}
size_t bytes_read = file_content_to_buffer(file);
if (bytes_read != file->data->size) {
log_message(STATUS_ERROR, "Didnt read expected amount of bytes from file");
exit(EXIT_FAILURE);
log_message(LOG_LEVEL_ERROR, "Did not read expected amount of bytes from file");
return false;
}
return true;
}
void file_send_single_calls(File *file, int file_descriptor, bool use_metadata, int compression_level) {
bool file_send_single_calls(File *file, int file_descriptor, bool use_metadata, int compression_level) {
if (compression_level > 0) {
Data *compressed_data = data_compress(file->data, compression_level);
if (compressed_data == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to compress file data");
return false;
}
data_destroy(file->data);
if (compressed_data == NULL) {
log_message(LOG_LEVEL_ERROR, "Compression failed in file_send_single_calls");
@@ -102,44 +110,52 @@ void file_send_single_calls(File *file, int file_descriptor, bool use_metadata,
}
file->data = compressed_data;
}
send_str(file_descriptor, file->path);
if (use_metadata)
metadata_send(file_descriptor, file->metadata);
send_data(file_descriptor, file->data);
if (!send_str(file_descriptor, file->path)) return false;
if (use_metadata && !metadata_send(file_descriptor, file->metadata)) return false;
if (!send_data(file_descriptor, file->data)) return false;
return true;
}
void to_disk(const char *path, const void *data, unsigned long long data_size) {
bool to_disk(const char *path, const void *data, unsigned long long data_size) {
char *directory = str_dup(path);
char *dir_to_free = directory;
directory = dirname(directory);
mkdir_r(directory);
if (!mkdir_r(directory)) {
free(dir_to_free);
return false;
}
FILE *file_pointer = fopen(path, "wb");
if (file_pointer == NULL) {
perror("Could not open File");
exit(EXIT_FAILURE);
free(dir_to_free);
return false;
}
if (fwrite(data, 1, data_size, file_pointer) != data_size) {
perror("Failed to write all data to disk");
fclose(file_pointer);
exit(EXIT_FAILURE);
free(dir_to_free);
return false;
}
fclose(file_pointer);
free(dir_to_free);
return true;
}
void file_send_sendfile(File *file, int file_descriptor, bool use_metadata) {
send_str(file_descriptor, file->path);
if (use_metadata)
metadata_send(file_descriptor, file->metadata);
bool file_send_sendfile(File *file, int file_descriptor, bool use_metadata) {
if (!send_str(file_descriptor, file->path)) return false;
if (use_metadata && !metadata_send(file_descriptor, file->metadata)) return false;
int fd = open(file->path, O_RDONLY);
if (fd == -1) {
perror("Could not open file for sendfile");
exit(EXIT_FAILURE);
return false;
}
unsigned long long file_size = file->data->size;
send_n_data(file_descriptor, &file_size, sizeof(unsigned long long));
if (!send_n_data(file_descriptor, &file_size, sizeof(unsigned long long))) {
close(fd);
return false;
}
off_t offset = 0;
while (offset < file_size) {
@@ -147,23 +163,35 @@ void file_send_sendfile(File *file, int file_descriptor, bool use_metadata) {
if (sent == -1) {
perror("sendfile failed");
close(fd);
exit(EXIT_FAILURE);
return false;
}
}
close(fd);
return true;
}
File *file_receive(Config *config, int file_descriptor) {
char *path = (char *)receive_str(file_descriptor);
char *path = receive_str(file_descriptor);
if (path == NULL) return NULL;
File *file = file_create(path);
free(path);
if (config->use_metadata)
if (file == NULL) return NULL;
if (config->use_metadata) {
file->metadata = metadata_receive(file_descriptor);
}
Data *file_data = receive_data(file_descriptor);
if (file_data == NULL) {
file_destroy(file);
return NULL;
}
if (config->use_compression) {
Data *file_data_uncompressed = data_decompress(file_data);
data_destroy(file_data);
if (file_data_uncompressed == NULL) {
file_destroy(file);
return NULL;
}
file_data = file_data_uncompressed;
}
data_destroy(file->data);
src/shared/file.h
+4 -4
View File
@@ -22,13 +22,13 @@ typedef struct {
File *file_create(const char *path);
void file_destroy(void *item);
void file_load_data(File *file);
bool file_load_data(File *file);
File *file_receive(Config *config, int file_descriptor);
void file_send_single_calls(File *file, int file_descriptor, bool use_metadata, int compression_level);
void file_send_sendfile(File *file, int file_descriptor, bool use_metadata);
bool file_send_single_calls(File *file, int file_descriptor, bool use_metadata, int compression_level);
bool file_send_sendfile(File *file, int file_descriptor, bool use_metadata);
size_t file_content_to_buffer(File *file);
FileMetadata *file_metadata_create(struct stat *stats);
void file_metadata_destroy(void *metadata);
void 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);
#endif
src/shared/metadata.c
+18 -14
View File
@@ -36,32 +36,36 @@ FileMetadata *metadata_from_buf(char **buf) {
return m;
}
void metadata_send(int file_descriptor, FileMetadata *m) {
bool metadata_send(int file_descriptor, FileMetadata *m) {
if (m == NULL) {
int zero = 0;
send_n_data(file_descriptor, &zero, sizeof(int));
return;
return send_n_data(file_descriptor, &zero, sizeof(int));
}
int present = 1;
send_n_data(file_descriptor, &present, sizeof(int));
send_n_data(file_descriptor, &m->mode, sizeof(mode_t));
send_n_data(file_descriptor, &m->uid, sizeof(uid_t));
send_n_data(file_descriptor, &m->gid, sizeof(gid_t));
send_n_data(file_descriptor, &m->mtime_sec, sizeof(time_t));
return send_n_data(file_descriptor, &present, sizeof(int)) &&
send_n_data(file_descriptor, &m->mode, sizeof(mode_t)) &&
send_n_data(file_descriptor, &m->uid, sizeof(uid_t)) &&
send_n_data(file_descriptor, &m->gid, sizeof(gid_t)) &&
send_n_data(file_descriptor, &m->mtime_sec, sizeof(time_t)) &&
send_n_data(file_descriptor, &m->mtime_nsec, sizeof(long));
}
FileMetadata *metadata_receive(int file_descriptor) {
int present;
receive_n_data(file_descriptor, &present, sizeof(int));
if (!receive_n_data(file_descriptor, &present, sizeof(int)))
return NULL;
if (!present)
return NULL;
FileMetadata *m = malloc(sizeof(FileMetadata));
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->gid, sizeof(gid_t));
receive_n_data(file_descriptor, &m->mtime_sec, sizeof(time_t));
receive_n_data(file_descriptor, &m->mtime_nsec, sizeof(long));
if (m == NULL) return NULL;
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->gid, sizeof(gid_t)) ||
!receive_n_data(file_descriptor, &m->mtime_sec, sizeof(time_t)) ||
!receive_n_data(file_descriptor, &m->mtime_nsec, sizeof(long))) {
free(m);
return NULL;
}
return m;
}
src/shared/metadata.h
+2 -1
View File
@@ -2,13 +2,14 @@
#define METADATA_H
#include "file.h"
#include <stdbool.h>
#include <sys/stat.h>
#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);
FileMetadata *metadata_from_buf(char **buf);
void metadata_send(int file_descriptor, FileMetadata *m);
bool metadata_send(int file_descriptor, FileMetadata *m);
FileMetadata *metadata_receive(int file_descriptor);
void file_restore_metadata(const char *path, FileMetadata *metadata);
src/shared/multiprocessing.c
+34 -11
View File
@@ -19,19 +19,22 @@ PipelineContextSender *pipeline_context_sender_create(Config *config,
Queue *queue_scanner,
Queue *queue_loader) {
PipelineContextSender *context = malloc(sizeof(PipelineContextSender));
if (context == NULL) return NULL;
context->config = config;
context->queue_scanner = queue_scanner;
context->queue_loader = queue_loader;
context->scanner_done = false;
context->loader_done = false;
context->manifest = NULL;
if (mtx_init(&context->mutex_scanner, mtx_plain) != thrd_success ||
cnd_init(&context->condition_not_full_scanner) != thrd_success ||
cnd_init(&context->condition_not_empty_scanner) != thrd_success ||
mtx_init(&context->mutex_loader, mtx_plain) != thrd_success ||
cnd_init(&context->condition_not_full_loader) != thrd_success ||
cnd_init(&context->condition_not_empty_loader) != thrd_success) {
perror("Error initializing synchronization objects!");
exit(EXIT_FAILURE);
perror("Error initializing synchronization objects");
free(context);
return NULL;
}
return context;
}
@@ -56,6 +59,7 @@ PipelineContextReceiver *pipeline_context_receiver_create(Config *config,
Queue *queue,
int file_descriptor) {
PipelineContextReceiver *context = malloc(sizeof(PipelineContextReceiver));
if (context == NULL) return NULL;
context->config = config;
context->queue = queue;
context->file_descriptor = file_descriptor;
@@ -63,8 +67,9 @@ PipelineContextReceiver *pipeline_context_receiver_create(Config *config,
if (mtx_init(&context->mutex, mtx_plain) != thrd_success ||
cnd_init(&context->condition_not_full) != thrd_success ||
cnd_init(&context->condition_not_empty) != thrd_success) {
perror("Error initializing synchronization objects!");
exit(EXIT_FAILURE);
perror("Error initializing synchronization objects");
free(context);
return NULL;
}
return context;
}
@@ -81,12 +86,16 @@ void pipeline_context_receiver_destroy(PipelineContextReceiver *context) {
static void receive_chunk_enqueue(int file_descriptor,
PipelineContextReceiver *context) {
Data *chunk_data = receive_data(file_descriptor);
if (chunk_data == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to receive chunk data");
return;
}
Data *data_to_process = chunk_data;
if (context->config->use_compression) {
data_to_process = data_decompress(chunk_data);
data_destroy(chunk_data);
if (data_to_process == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to decompress chunk, skipping");
log_message(LOG_LEVEL_ERROR, "Failed to decompress chunk");
return;
}
}
@@ -115,28 +124,40 @@ int receive_thread(void *pipeline_context) {
Config *config = context->config;
mtx_unlock(&context->mutex);
Status status = receive_status(file_descriptor);
Status status;
if (!receive_status(file_descriptor, &status)) return thrd_error;
while (status == STATUS_NEXT || status == STATUS_CHUNK) {
if (status == STATUS_CHUNK) {
receive_chunk_enqueue(file_descriptor, context);
} else {
File *file = file_receive(config, file_descriptor);
if (file) {
queue_enqueue_multithreaded(context->queue, file, &context->mutex,
&context->condition_not_empty,
&context->condition_not_full);
} else {
log_message(LOG_LEVEL_ERROR, "Failed to receive file");
}
status = receive_status(file_descriptor);
}
if (!receive_status(file_descriptor, &status)) return thrd_error;
}
if (status == STATUS_MANIFEST) {
int count = receive_int(file_descriptor);
int count;
if (!receive_int(file_descriptor, &count)) return thrd_error;
ArrayList *manifest = array_list_create(free);
for (int i = 0; i < count; i++)
array_list_add(manifest, receive_str(file_descriptor));
if (manifest) {
for (int i = 0; i < count; i++) {
char *s = receive_str(file_descriptor);
if (s) {
array_list_add(manifest, s);
}
}
delete_extras(context->config->receive_root_directory, manifest);
for (int i = 0; i < manifest->size; i++)
free(manifest->items[i]);
array_list_delete(manifest);
status = receive_status(file_descriptor);
}
if (!receive_status(file_descriptor, &status)) return thrd_error;
}
mtx_lock(&context->mutex);
context->receiver_done = true;
@@ -163,10 +184,12 @@ int write_thread(void *pipeline_context) {
}
if (save_to_disk) {
char *disk_path = path_cat(root_directory, file->path);
if (disk_path) {
to_disk(disk_path, file->data->data, file->data->size);
file_restore_metadata(disk_path, file->metadata);
free(disk_path);
}
}
file_destroy(file);
}
}
src/shared/protocol.c
+49 -31
View File
@@ -17,7 +17,7 @@ static int io_fd(int dir_fd, int file_descriptor) {
return (dir_fd != -1) ? dir_fd : file_descriptor;
}
void send_n_data(int file_descriptor, void *data, size_t data_size) {
bool send_n_data(int file_descriptor, void *data, size_t data_size) {
log_message(LOG_LEVEL_DEBUG, " Sending n Data: %zu", data_size);
int fd = io_fd(io_write_fd, file_descriptor);
ssize_t total_bytes_send = 0;
@@ -25,28 +25,33 @@ void send_n_data(int file_descriptor, void *data, size_t data_size) {
ssize_t bytes_send =
write(fd, (char *)data + total_bytes_send, data_size - total_bytes_send);
if (bytes_send <= 0) {
perror("Could not send data!");
exit(EXIT_FAILURE);
log_message(LOG_LEVEL_ERROR, "Could not send data");
return false;
}
total_bytes_send += bytes_send;
}
log_message(LOG_LEVEL_DEBUG, " Send n Data: %zu", total_bytes_send);
return true;
}
void receive_n_data(int file_descriptor, void *data, size_t data_size) {
bool receive_n_data(int file_descriptor, void *data, size_t data_size) {
log_message(LOG_LEVEL_DEBUG, " Receiving n Data: %zu", data_size);
int fd = io_fd(io_read_fd, file_descriptor);
size_t total_bytes_received = 0;
while (total_bytes_received < data_size) {
ssize_t bytes_received =
read(fd, (char *)data + total_bytes_received, data_size - total_bytes_received);
if (bytes_received == -1 || bytes_received == 0) {
perror("Could not receive bytes!");
exit(EXIT_FAILURE);
if (bytes_received <= 0) {
if (bytes_received == 0)
log_message(LOG_LEVEL_ERROR, "Connection closed while receiving data");
else
log_message(LOG_LEVEL_ERROR, "Could not receive bytes");
return false;
}
total_bytes_received += bytes_received;
}
log_message(LOG_LEVEL_DEBUG, " Received n Data: %zu", total_bytes_received);
return true;
}
static const char *status_to_string(Status status) {
@@ -66,59 +71,72 @@ static const char *status_to_string(Status status) {
}
}
void send_str(int file_descriptor, char *data) {
bool send_str(int file_descriptor, char *data) {
size_t size = strlen(data);
send_n_data(file_descriptor, &size, sizeof(size_t));
send_n_data(file_descriptor, data, size);
if (!send_n_data(file_descriptor, &size, sizeof(size_t))) return false;
if (!send_n_data(file_descriptor, data, size)) return false;
log_message(LOG_LEVEL_DEBUG, "Send String: %s", data);
return true;
}
char *receive_str(int file_descriptor) {
size_t size;
receive_n_data(file_descriptor, &size, sizeof(size_t));
if (!receive_n_data(file_descriptor, &size, sizeof(size_t))) return NULL;
char *data = (char *)malloc(size + 1);
receive_n_data(file_descriptor, data, size);
if (data == NULL) return NULL;
if (!receive_n_data(file_descriptor, data, size)) {
free(data);
return NULL;
}
data[size] = '\0';
log_message(LOG_LEVEL_DEBUG, "Received String: %s", data);
return data;
}
void send_data(int file_descriptor, Data *data) {
bool send_data(int file_descriptor, Data *data) {
unsigned long long data_size = data->size;
send_n_data(file_descriptor, &data_size, sizeof(unsigned long long));
send_n_data(file_descriptor, data->data, data_size);
if (!send_n_data(file_descriptor, &data_size, sizeof(unsigned long long)))
return false;
if (!send_n_data(file_descriptor, data->data, data_size))
return false;
log_message(LOG_LEVEL_DEBUG, "Send %lld data", data_size);
return true;
}
Data *receive_data(int file_descriptor) {
unsigned long long size = 0;
receive_n_data(file_descriptor, &size, sizeof(unsigned long long));
if (!receive_n_data(file_descriptor, &size, sizeof(unsigned long long)))
return NULL;
void *data = malloc((size_t)size);
receive_n_data(file_descriptor, data, (size_t)size);
if (data == NULL) return NULL;
if (!receive_n_data(file_descriptor, data, (size_t)size)) {
free(data);
return NULL;
}
log_message(LOG_LEVEL_DEBUG, "Received %lld data", size);
return data_create(data, (size_t)size);
}
void send_int(int file_descriptor, int data) {
send_n_data(file_descriptor, &data, sizeof(int));
bool send_int(int file_descriptor, int data) {
if (!send_n_data(file_descriptor, &data, sizeof(int))) return false;
log_message(LOG_LEVEL_DEBUG, "Send Int: %d", data);
return true;
}
int receive_int(int file_descriptor) {
int data;
receive_n_data(file_descriptor, &data, sizeof(int));
log_message(LOG_LEVEL_DEBUG, "Received Int: %d", data);
return data;
bool receive_int(int file_descriptor, int *data) {
if (!receive_n_data(file_descriptor, data, sizeof(int))) return false;
log_message(LOG_LEVEL_DEBUG, "Received Int: %d", *data);
return true;
}
void send_status(int file_descriptor, Status status) {
send_n_data(file_descriptor, &status, sizeof(Status));
bool send_status(int file_descriptor, Status status) {
if (!send_n_data(file_descriptor, &status, sizeof(Status))) return false;
log_message(LOG_LEVEL_DEBUG, "Send Status: %s", status_to_string(status));
return true;
}
Status receive_status(int file_descriptor) {
Status data;
receive_n_data(file_descriptor, &data, sizeof(Status));
log_message(LOG_LEVEL_DEBUG, "Received Status: %s", status_to_string(data));
return data;
bool receive_status(int file_descriptor, Status *status) {
if (!receive_n_data(file_descriptor, status, sizeof(Status))) return false;
log_message(LOG_LEVEL_DEBUG, "Received Status: %s", status_to_string(*status));
return true;
}
src/shared/protocol.h
+9 -8
View File
@@ -2,22 +2,23 @@
#define PROTOCOL_H
#include "data.h"
#include <stdbool.h>
#include <stddef.h>
typedef int Status;
enum NET_STATUS { STATUS_OK, STATUS_ERROR, STATUS_FINISHED, STATUS_NEXT, STATUS_CHUNK, STATUS_MANIFEST };
void io_set_fds(int read_fd, int write_fd);
void send_n_data(int file_descriptor, void *data, size_t data_size);
void receive_n_data(int file_descriptor, void *data, size_t data_size);
bool send_n_data(int file_descriptor, void *data, size_t data_size);
bool receive_n_data(int file_descriptor, void *data, size_t data_size);
void send_str(int file_descriptor, char *data);
bool send_str(int file_descriptor, char *data);
char *receive_str(int file_descriptor);
void send_data(int file_descriptor, Data *data);
bool send_data(int file_descriptor, Data *data);
Data *receive_data(int file_descriptor);
void send_int(int file_descriptor, int data);
int receive_int(int file_descriptor);
void send_status(int file_descriptor, Status status);
Status receive_status(int file_descriptor);
bool send_int(int file_descriptor, int data);
bool receive_int(int file_descriptor, int *data);
bool send_status(int file_descriptor, Status status);
bool receive_status(int file_descriptor, Status *status);
#endif
src/shared/queue.c
+16 -18
View File
@@ -9,15 +9,14 @@
Queue *queue_create(int capacity, void (*destroyer)(void *item)) {
Queue *queue = (Queue *)malloc(sizeof(Queue));
if (queue == NULL) {
perror("FATAL ERROR: Could not allocate memory for queue structure");
exit(EXIT_FAILURE);
perror("ERROR: Could not allocate memory for queue structure");
return NULL;
}
queue->items = malloc(capacity * sizeof(void *));
if (queue->items == NULL) {
perror("FATAL ERROR: Could not allocate memory for queue items");
free(queue);
exit(EXIT_FAILURE);
return NULL;
}
for (int i = 0; i < capacity; ++i) {
@@ -59,17 +58,15 @@ bool queue_is_full(Queue *queue) {
return queue->size == queue->capacity;
}
static void queue_double_capacity(Queue *queue) {
if (queue == NULL)
return;
static bool queue_double_capacity(Queue *queue) {
if (queue == NULL) return false;
unsigned int new_capacity = queue->capacity * 2;
if (new_capacity <= 1)
new_capacity = 100;
void **new_items = malloc(new_capacity * sizeof(void *));
if (new_items == NULL) {
perror("FATAL ERROR: Could not allocate memory for doubling capacity of "
"queue.");
exit(EXIT_FAILURE);
perror("ERROR: Could not allocate memory for doubling capacity of queue.");
return false;
}
for (int i = 0; i < queue->size; i++)
new_items[i] = queue->items[(i + queue->front) % queue->capacity];
@@ -78,29 +75,30 @@ static void queue_double_capacity(Queue *queue) {
queue->front = 0;
queue->rear = queue->size;
queue->capacity = new_capacity;
return true;
}
void queue_enqueue(Queue *queue, void *item) {
if (queue == NULL || item == NULL) {
perror("ERROR: Cannot enqueue with a null queue or item.\n");
exit(EXIT_FAILURE);
bool queue_enqueue(Queue *queue, void *item) {
if (queue == NULL || item == NULL) return false;
if (queue_is_full(queue)) {
if (!queue_double_capacity(queue)) return false;
}
if (queue_is_full(queue))
queue_double_capacity(queue);
queue->items[queue->rear] = item;
queue->rear = (queue->rear + 1) % queue->capacity;
queue->size++;
return true;
}
void 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_full) {
mtx_lock(mutex);
while (queue_is_full(queue))
cnd_wait(condition_not_full, mutex);
queue_enqueue(queue, item);
bool ok = queue_enqueue(queue, item);
cnd_signal(condition_not_empty);
mtx_unlock(mutex);
return ok;
}
void *queue_dequeue(Queue *queue) {
src/shared/queue.h
+2 -2
View File
@@ -17,8 +17,8 @@ Queue *queue_create(int capacity, void (*destroyer)(void *item));
void queue_destroy(Queue *queue);
bool queue_is_empty(Queue *queue);
bool queue_is_full(Queue *queue);
void queue_enqueue(Queue *queue, void *item);
void queue_enqueue_multithreaded(Queue *queue, void *item, mtx_t *mutex,
bool queue_enqueue(Queue *queue, void *item);
bool queue_enqueue_multithreaded(Queue *queue, void *item, mtx_t *mutex,
cnd_t *condition_not_empty,
cnd_t *condition_not_full);
void *queue_dequeue(Queue *queue);
src/shared/transport_ssh.c
+13 -5
View File
@@ -46,13 +46,13 @@ Client *client_connect_ssh(char *destination, int port) {
RemoteDest r;
if (parse_remote_dest(destination, &r) != 0) {
fprintf(stderr, "Invalid remote destination: %s\n", destination);
exit(EXIT_FAILURE);
return NULL;
}
int sv[2];
if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv) < 0) {
perror("socketpair failed");
exit(EXIT_FAILURE);
return NULL;
}
int buf_size = 1024 * 1024;
@@ -64,13 +64,16 @@ Client *client_connect_ssh(char *destination, int port) {
int exec_pipe[2];
if (pipe(exec_pipe) < 0) {
perror("pipe failed");
exit(EXIT_FAILURE);
close(sv[0]); close(sv[1]);
return NULL;
}
pid_t pid = fork();
if (pid < 0) {
perror("fork failed");
exit(EXIT_FAILURE);
close(sv[0]); close(sv[1]);
close(exec_pipe[0]); close(exec_pipe[1]);
return NULL;
}
if (pid == 0) {
@@ -127,10 +130,15 @@ Client *client_connect_ssh(char *destination, int port) {
close(sv[0]);
waitpid(pid, NULL, 0);
fprintf(stderr, "Error: could not launch 'fastsync-server --stdio' on remote\n");
exit(EXIT_FAILURE);
return NULL;
}
Client *client = malloc(sizeof(Client));
if (client == NULL) {
close(sv[0]);
waitpid(pid, NULL, 0);
return NULL;
}
client->file_descriptor = sv[0];
client->address.sin_family = AF_UNIX;
client->address_length = 0;
src/shared/transport_tcp.c
+17 -10
View File
@@ -13,13 +13,14 @@ Server *server_create(int port) {
Server *server = (Server *)malloc(sizeof(Server));
if (server == NULL) {
perror("Could not allocate space for Server");
exit(EXIT_FAILURE);
return NULL;
}
int file_descriptor = socket(AF_INET, SOCK_STREAM, 0);
if (file_descriptor < 0) {
perror("Could not create Socket!");
exit(EXIT_FAILURE);
free(server);
return NULL;
}
server->file_descriptor = file_descriptor;
int opt = 1;
@@ -28,7 +29,7 @@ Server *server_create(int port) {
perror("Error setting a socket option!");
close(server->file_descriptor);
free(server);
exit(EXIT_FAILURE);
return NULL;
}
server->address.sin_family = AF_INET;
@@ -41,7 +42,7 @@ Server *server_create(int port) {
perror("Could not bind server");
close(server->file_descriptor);
free(server);
exit(EXIT_FAILURE);
return NULL;
}
return server;
@@ -54,12 +55,12 @@ void server_delete(Server **server) {
*server = NULL;
}
void 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",
server->address.sin_port);
if (listen(server->file_descriptor, SOMAXCONN) < 0) {
perror("Could not listen on port!");
exit(EXIT_FAILURE);
return false;
}
signal(SIGCHLD, SIG_IGN);
@@ -84,16 +85,21 @@ void server_listen(Server *server, void (*handler)(int file_descriptor)) {
}
close(file_descriptor);
}
return true;
}
Client *client_create() {
int file_descriptor = socket(AF_INET, SOCK_STREAM, 0);
if (file_descriptor < 0) {
perror("Could not create Socket!");
exit(EXIT_FAILURE);
return NULL;
}
Client *client = (Client *)malloc(sizeof(Client));
if (client == NULL) {
close(file_descriptor);
return NULL;
}
client->file_descriptor = file_descriptor;
client->address.sin_family = AF_INET;
client->address_length = sizeof(client->address);
@@ -101,19 +107,20 @@ Client *client_create() {
return client;
}
void client_connect(Client *client, char *host, int port) {
bool client_connect(Client *client, char *host, int port) {
client->address.sin_port = htons(port);
if (inet_pton(AF_INET, host, &client->address.sin_addr) <= 0) {
perror("Could not convert host address!");
exit(EXIT_FAILURE);
return false;
}
if (connect(client->file_descriptor, (struct sockaddr *)&client->address,
client->address_length) < 0) {
perror("Could not connect to Server!");
exit(EXIT_FAILURE);
return false;
}
return true;
}
void client_disconnect(Client *client) {
src/shared/transport_tcp.h
+3 -2
View File
@@ -2,6 +2,7 @@
#define TRANSPORT_TCP_H
#include <netinet/in.h>
#include <stdbool.h>
#include <sys/types.h>
typedef struct Server {
@@ -18,10 +19,10 @@ typedef struct Client {
} Client;
Server *server_create(int port);
void server_listen(Server *server, void (*handler)(int file_descriptor));
bool server_listen(Server *server, void (*handler)(int file_descriptor));
void server_delete(Server **server);
Client *client_create();
void client_connect(Client *client, char *host, int port);
bool client_connect(Client *client, char *host, int port);
void client_disconnect(Client *client);
void client_delete(Client *client);
src/shared/utils.c
+8 -2
View File
@@ -8,10 +8,12 @@
#include <sys/stat.h>
#include <unistd.h>
void mkdir_r(char *path) {
bool mkdir_r(char *path) {
char *path_duplicate = malloc(strlen(path) + 1);
if (!path_duplicate) return false;
strcpy(path_duplicate, path);
char *path_current = (char *)malloc((strlen(path) + 2) * sizeof(char));
if (!path_current) { free(path_duplicate); return false; }
char *path_current_position = path_current;
if (path[0] == '/') {
strcpy(path_current, "/");
@@ -21,6 +23,7 @@ void mkdir_r(char *path) {
}
const char *delimiter = "/";
char *part = strtok(path_duplicate, delimiter);
bool ok = true;
while (part != NULL) {
strcpy(path_current_position, part);
path_current_position += strlen(part) * sizeof(char);
@@ -30,13 +33,15 @@ void mkdir_r(char *path) {
if (stat(path_current, &st) != 0) {
if (mkdir(path_current, 0755) != 0) {
perror("Could not create directory");
exit(EXIT_FAILURE);
ok = false;
break;
}
}
part = strtok(NULL, delimiter);
}
free(path_duplicate);
free(path_current);
return ok;
}
char *str_dup(const char *string) {
@@ -131,6 +136,7 @@ char *path_cat(char *path1, char *path2) {
path2_len -= 1;
}
char *new_path = malloc(path1_len + path2_len + 2);
if (new_path == NULL) return NULL;
memcpy(new_path, path1, path1_len);
new_path[path1_len] = '/';
memcpy(new_path + path1_len + 1, path2_pointer, path2_len);
src/shared/utils.h
+1 -1
View File
@@ -4,7 +4,7 @@
#include "array_list.h"
#include <stdbool.h>
void mkdir_r(char *path);
bool mkdir_r(char *path);
char *str_dup(const char *string);
char *path_cat(char *path1, char *path2);
bool glob_match(const char *pattern, const char *str);
tests/test_scanner.c
+5 -5
View File
@@ -7,7 +7,7 @@
#include <unistd.h>
static void create_test_file(const char *path, const char *content) {
to_disk(path, content, strlen(content));
(void)to_disk(path, content, strlen(content));
}
static void test_scanner_single_file() {
@@ -18,7 +18,7 @@ static void test_scanner_single_file() {
mkdir(dir, 0755);
create_test_file(file1, content1);
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0, NULL, 0);
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0, NULL, 0, NULL, 0, 0, 0);
EXPECT_NOT_NULL(scanner);
Chunk *chunk = directory_scanner_next(scanner);
@@ -46,7 +46,7 @@ static void test_scanner_multiple_files() {
create_test_file(file1, content1);
create_test_file(file2, content2);
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0, NULL, 0);
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0, NULL, 0, NULL, 0, 0, 0);
EXPECT_NOT_NULL(scanner);
Chunk *chunk = directory_scanner_next(scanner);
@@ -83,7 +83,7 @@ static void test_scanner_subdirectory() {
create_test_file(root_file, content);
create_test_file(sub_file, content);
DirectoryScanner *scanner = directory_scanner_create((char *)root, false, 0, NULL, 0);
DirectoryScanner *scanner = directory_scanner_create((char *)root, false, 0, NULL, 0, NULL, 0, 0, 0);
EXPECT_NOT_NULL(scanner);
int total_files = 0;
@@ -106,7 +106,7 @@ static void test_scanner_empty_directory() {
mkdir(dir, 0755);
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0, NULL, 0);
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0, NULL, 0, NULL, 0, 0, 0);
EXPECT_NOT_NULL(scanner);
Chunk *chunk = directory_scanner_next(scanner);