From 02619cca5ceeb7b6eee52af5f6b9112b9d510ec0 Mon Sep 17 00:00:00 2001 From: TapTap Date: Wed, 15 Jul 2026 19:36:26 +0200 Subject: [PATCH] refactor: split monolithic modules and extract shared components - Remove dead socket.c/socket.h (accumulated duplicate symbols) - Extract compression.h/c (data_compress/decompress from data.c) - Extract io.h/c (low-level I/O from protocol.c) - Extract metadata.h/c (unified metadata wire format from file.c, chunk.c, utils.c) - Split client.c into client_cli.c (CLI parsing) + client_send.c (send logic) - Move receiver pipeline threads from server.c to multiprocessing.c - Move to_disk/file_restore_metadata from utils.c to file.c/metadata.c - Fix const qualifiers on to_disk and str_dup signatures - Suppress chown unused-result warning --- src/client/client_cli.c | 150 +++++++++++ src/client/{client.c => client_send.c} | 256 ++++-------------- src/client/client_send.h | 15 ++ src/server/server.c | 98 +------ src/shared/chunk.c | 36 +-- src/shared/compression.c | 46 ++++ src/shared/compression.h | 9 + src/shared/config.c | 2 +- src/shared/data.c | 43 --- src/shared/data.h | 2 - src/shared/file.c | 53 ++-- src/shared/file.h | 2 +- src/shared/io.c | 49 ++++ src/shared/io.h | 10 + src/shared/metadata.c | 82 ++++++ src/shared/metadata.h | 15 ++ src/shared/multiprocessing.c | 100 +++++++ src/shared/multiprocessing.h | 4 + src/shared/protocol.c | 79 ++++++ src/shared/protocol.h | 20 ++ src/shared/socket.c | 346 ------------------------- src/shared/socket.h | 45 ---- src/shared/transport_ssh.c | 115 ++++++++ src/shared/transport_ssh.h | 8 + src/shared/transport_tcp.c | 119 +++++++++ src/shared/transport_tcp.h | 28 ++ src/shared/utils.c | 35 +-- src/shared/utils.h | 6 +- tests/test_compression.c | 1 + 29 files changed, 942 insertions(+), 832 deletions(-) create mode 100644 src/client/client_cli.c rename src/client/{client.c => client_send.c} (52%) create mode 100644 src/client/client_send.h create mode 100644 src/shared/compression.c create mode 100644 src/shared/compression.h create mode 100644 src/shared/io.c create mode 100644 src/shared/io.h create mode 100644 src/shared/metadata.c create mode 100644 src/shared/metadata.h create mode 100644 src/shared/protocol.c create mode 100644 src/shared/protocol.h delete mode 100644 src/shared/socket.c delete mode 100644 src/shared/socket.h create mode 100644 src/shared/transport_ssh.c create mode 100644 src/shared/transport_ssh.h create mode 100644 src/shared/transport_tcp.c create mode 100644 src/shared/transport_tcp.h diff --git a/src/client/client_cli.c b/src/client/client_cli.c new file mode 100644 index 0000000..f26fd62 --- /dev/null +++ b/src/client/client_cli.c @@ -0,0 +1,150 @@ +#include "client_send.h" +#include "config.h" +#include "log.h" +#include "utils.h" +#include +#include +#include + +char *server_host = "127.0.0.1"; +int server_port = 8080; + +static void print_usage(void) { + printf("Usage:\n"); + printf(" fastsync [options] \n"); + printf(" fastsync [options] --source-dir --dest-dir \n"); + printf("\n"); + printf("Destination formats:\n"); + printf(" user@host:/path SSH transport (rsync-style)\n"); + printf(" host:/path SSH transport (current user)\n"); + printf(" /local/path TCP transport (requires server on localhost:8080)\n"); + printf("\n"); + printf("Options:\n"); + printf(" -c [level] Enable compression (level 1-22, default 5)\n"); + printf(" -m Enable multithreading\n"); + printf(" -s Enable chunk serialization\n"); + printf(" -f Enable sendfile (TCP only, not with -c or -s)\n"); + printf(" -v, --verbose Enable debug logging\n"); + printf(" -M, --preserve Preserve file metadata\n"); + printf(" --source-dir Source directory\n"); + printf(" --dest-dir Destination directory\n"); + printf(" --save-to-disk Write received files to disk\n"); + printf(" --server-host Server IP address (default: 127.0.0.1)\n"); + printf(" --server-port Server port (default: 8080)\n"); + printf(" --help Show this help\n"); +} + +int main(int argc, char *argv[]) { + const char *env_source = getenv("FASTSYNC_SOURCE_DIR"); + const char *env_dest = getenv("FASTSYNC_DEST_DIR"); + const char *env_save = getenv("FASTSYNC_SAVE_TO_DISK"); + + bool save_to_disk = false; + if (env_save && + (strcmp(env_save, "true") == 0 || strcmp(env_save, "1") == 0)) { + save_to_disk = true; + } + + Config *config = config_create(str_dup("1.0.0"), NULL, NULL, + save_to_disk, false, false, false, false, 5, 20, false); + + int positional_args[2]; + int positional_count = 0; + + for (int i = 1; i < argc; i++) { + if (strcmp(argv[i], "--help") == 0) { + print_usage(); + return 0; + } else if (strcmp(argv[i], "-c") == 0) { + config->use_compression = true; + log_message(LOG_LEVEL_INFO, "Enabled Compression"); + if (i + 1 < argc) { + char *end_ptr; + int level = strtol(argv[i + 1], &end_ptr, 10); + if (*end_ptr == '\0') { + config->compression_level = level; + log_message(LOG_LEVEL_INFO, "Set Compression level to %d", + config->compression_level); + i++; + } + } + } else if (strcmp(argv[i], "--source-dir") == 0 && i + 1 < argc) { + free(config->send_directory); + config->send_directory = str_dup(argv[++i]); + } else if (strcmp(argv[i], "--dest-dir") == 0 && i + 1 < argc) { + free(config->receive_root_directory); + config->receive_root_directory = str_dup(argv[++i]); + } else if (strcmp(argv[i], "--save-to-disk") == 0) { + config->save_to_disk = true; + } else if (strcmp(argv[i], "-M") == 0 || strcmp(argv[i], "--preserve") == 0) { + config->use_metadata = true; + log_message(LOG_LEVEL_INFO, "Enabled metadata preservation"); + } else if (strcmp(argv[i], "-f") == 0 || strcmp(argv[i], "--sendfile") == 0) { + config->use_sendfile = true; + log_message(LOG_LEVEL_INFO, "Enabled sendfile"); + } else if (strcmp(argv[i], "-m") == 0) { + config->use_multithreading = true; + log_message(LOG_LEVEL_INFO, "Enabled Multithreading"); + } else if (strcmp(argv[i], "-s") == 0) { + config->use_chunk_serialization = true; + log_message(LOG_LEVEL_INFO, "Enabled Chunk Serialization"); + } else if (strcmp(argv[i], "--server-host") == 0 && i + 1 < argc) { + free(server_host); + server_host = str_dup(argv[++i]); + } else if (strcmp(argv[i], "--server-port") == 0 && i + 1 < argc) { + server_port = atoi(argv[++i]); + } else if (strcmp(argv[i], "-v") == 0 || strcmp(argv[i], "--verbose") == 0) { + set_log_level(LOG_LEVEL_DEBUG); + } else if (argv[i][0] == '-') { + fprintf(stderr, "Unknown option: %s\n", argv[i]); + print_usage(); + return 1; + } else { + if (positional_count < 2) + positional_args[positional_count++] = i; + else { + fprintf(stderr, "Unexpected argument: %s\n", argv[i]); + print_usage(); + return 1; + } + } + } + + if (positional_count == 2) { + free(config->send_directory); + free(config->receive_root_directory); + config->send_directory = str_dup(argv[positional_args[0]]); + config->receive_root_directory = str_dup(argv[positional_args[1]]); + config->save_to_disk = true; + + config_parse_ssh_dest(config); + } else if (positional_count == 1) { + fprintf(stderr, "Error: missing destination argument\n"); + print_usage(); + return 1; + } else { + if (!config->send_directory && env_source) + config->send_directory = str_dup((char *)env_source); + if (!config->receive_root_directory && env_dest) + config->receive_root_directory = str_dup((char *)env_dest); + } + + if (!config->send_directory || !config->receive_root_directory) { + fprintf(stderr, "Error: source and destination directories are required\n"); + print_usage(); + return 1; + } + if (config->use_sendfile && (config->use_chunk_serialization || config->use_compression)) { + fprintf(stderr, "Error: -f/--sendfile cannot be combined with -c (compression) or -s (chunk serialization)\n"); + return 1; + } + + if (config->transport == TRANSPORT_SSH && config->use_sendfile) { + fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n"); + return 1; + } + + if (config->use_multithreading) + return send_files_multithreaded(config); + return send_files(config); +} diff --git a/src/client/client.c b/src/client/client_send.c similarity index 52% rename from src/client/client.c rename to src/client/client_send.c index a003743..97cf409 100644 --- a/src/client/client.c +++ b/src/client/client_send.c @@ -1,23 +1,21 @@ -#include -#include -#include -#include -#include - +#include "client_send.h" #include "chunk.h" +#include "compression.h" #include "config.h" #include "data.h" #include "file.h" #include "log.h" #include "multiprocessing.h" +#include "protocol.h" #include "queue.h" #include "scanner.h" -#include "socket.h" +#include "transport_tcp.h" +#include "transport_ssh.h" #include "utils.h" -#include - -static char *server_host = "127.0.0.1"; -static int server_port = 8080; +#include +#include +#include +#include int send_chunk(Client *client, Chunk *chunk, Config *config) { if (config->use_chunk_serialization) { @@ -53,54 +51,7 @@ int send_chunk(Client *client, Chunk *chunk, Config *config) { return 0; } -int scan_directory_multithreaded(void *pipeline_context) { - PipelineContextSender *context = (PipelineContextSender *)pipeline_context; - mtx_lock(&context->mutex_scanner); - DirectoryScanner *scanner = - directory_scanner_create(context->config->send_directory, context->config->use_metadata); - mtx_unlock(&context->mutex_scanner); - - Chunk *current_chunk; - while ((current_chunk = directory_scanner_next(scanner)) != NULL) - queue_enqueue_multithreaded(context->queue_scanner, current_chunk, - &context->mutex_scanner, - &context->condition_not_empty_scanner, - &context->condition_not_full_scanner); - mtx_lock(&context->mutex_scanner); - context->scanner_done = true; - cnd_signal(&context->condition_not_empty_scanner); - mtx_unlock(&context->mutex_scanner); - - directory_scanner_destroy(scanner); - return thrd_success; -} - -int load_files_multithreaded(void *pipeline_context) { - PipelineContextSender *context = (PipelineContextSender *)pipeline_context; - while (true) { - Chunk *chunk = queue_dequeue_multithreaded( - context->queue_scanner, &context->mutex_scanner, - &context->condition_not_empty_scanner, - &context->condition_not_full_scanner, &context->scanner_done); - if (chunk == NULL) { - mtx_lock(&context->mutex_loader); - context->loader_done = true; - cnd_signal(&context->condition_not_empty_loader); - mtx_unlock(&context->mutex_loader); - return thrd_success; - } - if (!context->config->use_sendfile) { - for (int i = 0; i < chunk->element_count; i++) - file_load_data(chunk->items[i]); - } - queue_enqueue_multithreaded(context->queue_loader, chunk, - &context->mutex_loader, - &context->condition_not_empty_loader, - &context->condition_not_full_loader); - } -} - -int send_chunks_multithreaded(void *pipeline_context) { +static int send_chunks_multithreaded(void *pipeline_context) { PipelineContextSender *context = (PipelineContextSender *)pipeline_context; Client *client; if (context->config->transport == TRANSPORT_SSH) { @@ -135,6 +86,53 @@ int send_chunks_multithreaded(void *pipeline_context) { } } +static int scan_directory_multithreaded(void *pipeline_context) { + PipelineContextSender *context = (PipelineContextSender *)pipeline_context; + mtx_lock(&context->mutex_scanner); + DirectoryScanner *scanner = + directory_scanner_create(context->config->send_directory, context->config->use_metadata); + mtx_unlock(&context->mutex_scanner); + + Chunk *current_chunk; + while ((current_chunk = directory_scanner_next(scanner)) != NULL) + queue_enqueue_multithreaded(context->queue_scanner, current_chunk, + &context->mutex_scanner, + &context->condition_not_empty_scanner, + &context->condition_not_full_scanner); + mtx_lock(&context->mutex_scanner); + context->scanner_done = true; + cnd_signal(&context->condition_not_empty_scanner); + mtx_unlock(&context->mutex_scanner); + + directory_scanner_destroy(scanner); + return thrd_success; +} + +static int load_files_multithreaded(void *pipeline_context) { + PipelineContextSender *context = (PipelineContextSender *)pipeline_context; + while (true) { + Chunk *chunk = queue_dequeue_multithreaded( + context->queue_scanner, &context->mutex_scanner, + &context->condition_not_empty_scanner, + &context->condition_not_full_scanner, &context->scanner_done); + if (chunk == NULL) { + mtx_lock(&context->mutex_loader); + context->loader_done = true; + cnd_signal(&context->condition_not_empty_loader); + mtx_unlock(&context->mutex_loader); + return thrd_success; + } + if (!context->config->use_sendfile) { + for (int i = 0; i < chunk->element_count; i++) + file_load_data(chunk->items[i]); + } + queue_enqueue_multithreaded(context->queue_loader, chunk, + &context->mutex_loader, + &context->condition_not_empty_loader, + &context->condition_not_full_loader); + } +} + int send_files(Config *config) { Client *client; if (config->transport == TRANSPORT_SSH) { @@ -188,143 +186,3 @@ int send_files_multithreaded(Config *config) { pipeline_context_sender_destroy(context); return 0; } - -static void print_usage(void) { - printf("Usage:\n"); - printf(" fastsync [options] \n"); - printf(" fastsync [options] --source-dir --dest-dir \n"); - printf("\n"); - printf("Destination formats:\n"); - printf(" user@host:/path SSH transport (rsync-style)\n"); - printf(" host:/path SSH transport (current user)\n"); - printf(" /local/path TCP transport (requires server on localhost:8080)\n"); - printf("\n"); - printf("Options:\n"); - printf(" -c [level] Enable compression (level 1-22, default 5)\n"); - printf(" -m Enable multithreading\n"); - printf(" -s Enable chunk serialization\n"); - printf(" -f Enable sendfile (TCP only, not with -c or -s)\n"); - printf(" -v, --verbose Enable debug logging\n"); - printf(" -M, --preserve Preserve file metadata\n"); - printf(" --source-dir Source directory\n"); - printf(" --dest-dir Destination directory\n"); - printf(" --save-to-disk Write received files to disk\n"); - printf(" --server-host Server IP address (default: 127.0.0.1)\n"); - printf(" --server-port Server port (default: 8080)\n"); - printf(" --help Show this help\n"); -} - -int main(int argc, char *argv[]) { - const char *env_source = getenv("FASTSYNC_SOURCE_DIR"); - const char *env_dest = getenv("FASTSYNC_DEST_DIR"); - const char *env_save = getenv("FASTSYNC_SAVE_TO_DISK"); - - bool save_to_disk = false; - if (env_save && - (strcmp(env_save, "true") == 0 || strcmp(env_save, "1") == 0)) { - save_to_disk = true; - } - - Config *config = config_create(str_dup("1.0.0"), NULL, NULL, - save_to_disk, false, false, false, false, 5, 20, false); - - int positional_args[2]; - int positional_count = 0; - - for (int i = 1; i < argc; i++) { - if (strcmp(argv[i], "--help") == 0) { - print_usage(); - return 0; - } else if (strcmp(argv[i], "-c") == 0) { - config->use_compression = true; - log_message(LOG_LEVEL_INFO, "Enabled Compression"); - if (i + 1 < argc) { - char *end_ptr; - int level = strtol(argv[i + 1], &end_ptr, 10); - if (*end_ptr == '\0') { - config->compression_level = level; - log_message(LOG_LEVEL_INFO, "Set Compression level to %d", - config->compression_level); - i++; - } - } - } else if (strcmp(argv[i], "--source-dir") == 0 && i + 1 < argc) { - free(config->send_directory); - config->send_directory = str_dup(argv[++i]); - } else if (strcmp(argv[i], "--dest-dir") == 0 && i + 1 < argc) { - free(config->receive_root_directory); - config->receive_root_directory = str_dup(argv[++i]); - } else if (strcmp(argv[i], "--save-to-disk") == 0) { - config->save_to_disk = true; - } else if (strcmp(argv[i], "-M") == 0 || strcmp(argv[i], "--preserve") == 0) { - config->use_metadata = true; - log_message(LOG_LEVEL_INFO, "Enabled metadata preservation"); - } else if (strcmp(argv[i], "-f") == 0 || strcmp(argv[i], "--sendfile") == 0) { - config->use_sendfile = true; - log_message(LOG_LEVEL_INFO, "Enabled sendfile"); - } else if (strcmp(argv[i], "-m") == 0) { - config->use_multithreading = true; - log_message(LOG_LEVEL_INFO, "Enabled Multithreading"); - } else if (strcmp(argv[i], "-s") == 0) { - config->use_chunk_serialization = true; - log_message(LOG_LEVEL_INFO, "Enabled Chunk Serialization"); - } else if (strcmp(argv[i], "--server-host") == 0 && i + 1 < argc) { - free(server_host); - server_host = str_dup(argv[++i]); - } else if (strcmp(argv[i], "--server-port") == 0 && i + 1 < argc) { - server_port = atoi(argv[++i]); - } else if (strcmp(argv[i], "-v") == 0 || strcmp(argv[i], "--verbose") == 0) { - set_log_level(LOG_LEVEL_DEBUG); - } else if (argv[i][0] == '-') { - fprintf(stderr, "Unknown option: %s\n", argv[i]); - print_usage(); - return 1; - } else { - if (positional_count < 2) - positional_args[positional_count++] = i; - else { - fprintf(stderr, "Unexpected argument: %s\n", argv[i]); - print_usage(); - return 1; - } - } - } - - if (positional_count == 2) { - free(config->send_directory); - free(config->receive_root_directory); - config->send_directory = str_dup(argv[positional_args[0]]); - config->receive_root_directory = str_dup(argv[positional_args[1]]); - config->save_to_disk = true; - - config_parse_ssh_dest(config); - } else if (positional_count == 1) { - fprintf(stderr, "Error: missing destination argument\n"); - print_usage(); - return 1; - } else { - if (!config->send_directory && env_source) - config->send_directory = str_dup((char *)env_source); - if (!config->receive_root_directory && env_dest) - config->receive_root_directory = str_dup((char *)env_dest); - } - - if (!config->send_directory || !config->receive_root_directory) { - fprintf(stderr, "Error: source and destination directories are required\n"); - print_usage(); - return 1; - } - if (config->use_sendfile && (config->use_chunk_serialization || config->use_compression)) { - fprintf(stderr, "Error: -f/--sendfile cannot be combined with -c (compression) or -s (chunk serialization)\n"); - return 1; - } - - if (config->transport == TRANSPORT_SSH && config->use_sendfile) { - fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n"); - return 1; - } - - if (config->use_multithreading) - return send_files_multithreaded(config); - return send_files(config); -} diff --git a/src/client/client_send.h b/src/client/client_send.h new file mode 100644 index 0000000..132cb15 --- /dev/null +++ b/src/client/client_send.h @@ -0,0 +1,15 @@ +#ifndef CLIENT_SEND_H +#define CLIENT_SEND_H + +#include "chunk.h" +#include "config.h" +#include "transport_tcp.h" + +extern char *server_host; +extern int server_port; + +int send_chunk(Client *client, Chunk *chunk, Config *config); +int send_files(Config *config); +int send_files_multithreaded(Config *config); + +#endif diff --git a/src/server/server.c b/src/server/server.c index 6d1ef64..b6a7af7 100644 --- a/src/server/server.c +++ b/src/server/server.c @@ -1,108 +1,20 @@ #include "chunk.h" +#include "compression.h" #include "config.h" #include "data.h" #include "file.h" +#include "io.h" #include "log.h" +#include "metadata.h" #include "multiprocessing.h" +#include "protocol.h" #include "queue.h" -#include "socket.h" +#include "transport_tcp.h" #include "unistd.h" #include "utils.h" #include #include #include -#include - -File *file_receive(Config *config, int file_descriptor) { - char *path = (char *)receive_str(file_descriptor); - File *file = file_create(path); - free(path); - if (config->use_metadata) - file->metadata = file_receive_metadata(file_descriptor); - Data *file_data = receive_data(file_descriptor); - if (config->use_compression) { - Data *file_data_uncompressed = data_decompress(file_data); - data_destroy(file_data); - file_data = file_data_uncompressed; - } - data_destroy(file->data); - file->data = file_data; - return file; -} - -static void receive_chunk_enqueue(int file_descriptor, - PipelineContextReceiver *context) { - Data *chunk_data = receive_data(file_descriptor); - Data *data_to_process = chunk_data; - if (context->config->use_compression) { - data_to_process = data_decompress(chunk_data); - data_destroy(chunk_data); - } - Chunk *chunk = chunk_deserialize(data_to_process, context->config->use_metadata); - data_destroy(data_to_process); - - for (int i = 0; i < chunk->element_count; i++) { - File *file = chunk->items[i]; - chunk->items[i] = NULL; - queue_enqueue_multithreaded(context->queue, file, &context->mutex, - &context->condition_not_empty, - &context->condition_not_full); - } - chunk_destroy(chunk); -} - -int receive_thread(void *pipeline_context) { - PipelineContextReceiver *context = - (PipelineContextReceiver *)pipeline_context; - mtx_lock(&context->mutex); - int file_descriptor = context->file_descriptor; - Config *config = context->config; - mtx_unlock(&context->mutex); - - Status status = receive_status(file_descriptor); - 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); - queue_enqueue_multithreaded(context->queue, file, &context->mutex, - &context->condition_not_empty, - &context->condition_not_full); - } - status = receive_status(file_descriptor); - } - mtx_lock(&context->mutex); - context->receiver_done = true; - cnd_signal(&context->condition_not_empty); - mtx_unlock(&context->mutex); - return thrd_success; -} - -int write_thread(void *pipeline_context) { - PipelineContextReceiver *context = - (PipelineContextReceiver *)pipeline_context; - mtx_lock(&context->mutex); - bool save_to_disk = context->config->save_to_disk; - char *root_directory = str_dup(context->config->receive_root_directory); - mtx_unlock(&context->mutex); - - while (true) { - File *file = queue_dequeue_multithreaded( - context->queue, &context->mutex, &context->condition_not_empty, - &context->condition_not_full, &context->receiver_done); - if (file == NULL) { - free(root_directory); - return thrd_success; - } - if (save_to_disk) { - char *disk_path = path_cat(root_directory, file->path); - to_disk(disk_path, file->data->data, file->data->size); - file_restore_metadata(disk_path, file->metadata); - free(disk_path); - } - file_destroy(file); - } -} int receive_files(Config *config, int file_descriptor) { Status status = receive_status(file_descriptor); diff --git a/src/shared/chunk.c b/src/shared/chunk.c index e405ce6..9740643 100644 --- a/src/shared/chunk.c +++ b/src/shared/chunk.c @@ -1,17 +1,15 @@ -#include -#include #include #include #include #include -#include "chunk.h" #include "array_list.h" +#include "chunk.h" +#include "compression.h" #include "data.h" #include "file.h" #include "log.h" - -#define FILE_METADATA_WIRE_SIZE (sizeof(mode_t) + sizeof(uid_t) + sizeof(gid_t) + sizeof(time_t) + sizeof(long)) +#include "metadata.h" Chunk *chunk_create(File **items, int element_count) { Chunk *chunk = (Chunk *)malloc(sizeof(Chunk)); @@ -49,34 +47,6 @@ void chunk_destroy(void *item) { free(chunk); } -static void metadata_to_buf(char **buf, FileMetadata *m) { - int present = (m != NULL) ? 1 : 0; - memcpy(*buf, &present, sizeof(int)); - *buf += sizeof(int); - if (m == NULL) - return; - memcpy(*buf, &m->mode, sizeof(mode_t)); *buf += sizeof(mode_t); - memcpy(*buf, &m->uid, sizeof(uid_t)); *buf += sizeof(uid_t); - 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); -} - -static FileMetadata *metadata_from_buf(char **buf) { - int present; - memcpy(&present, *buf, sizeof(int)); - *buf += sizeof(int); - if (!present) - return NULL; - FileMetadata *m = malloc(sizeof(FileMetadata)); - memcpy(&m->mode, *buf, sizeof(mode_t)); *buf += sizeof(mode_t); - memcpy(&m->uid, *buf, sizeof(uid_t)); *buf += sizeof(uid_t); - 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; -} - static unsigned long long per_file_serialize_size(File *file, bool use_metadata) { return sizeof(size_t) + strlen(file->path) + (use_metadata ? sizeof(int) + (file->metadata ? FILE_METADATA_WIRE_SIZE : 0) : 0) + diff --git a/src/shared/compression.c b/src/shared/compression.c new file mode 100644 index 0000000..627659c --- /dev/null +++ b/src/shared/compression.c @@ -0,0 +1,46 @@ +#include "compression.h" +#include "data.h" +#include "log.h" +#include "stdlib.h" +#include "zstd.h" + +Data *data_compress(Data *data_to_compress, int compression_level) { + log_message(LOG_LEVEL_DEBUG, "Starting to compress data"); + Data *compressed_data = + data_create_empty(ZSTD_compressBound(data_to_compress->size)); + + compressed_data->size = ZSTD_compress( + compressed_data->data, compressed_data->size, data_to_compress->data, + data_to_compress->size, compression_level); + if (ZSTD_isError(compressed_data->size)) { + log_message(LOG_LEVEL_ERROR, "Compression failed: %s", + ZSTD_getErrorName(compressed_data->size)); + exit(EXIT_FAILURE); + } + + log_message(LOG_LEVEL_DEBUG, "Data succesfully compressed from %zu to %zu", + data_to_compress->size, compressed_data->size); + return compressed_data; +} + +Data *data_decompress(Data *compressed_data) { + log_message(LOG_LEVEL_DEBUG, "Start to decompress data"); + Data *uncompressed_data = data_create_empty( + ZSTD_getFrameContentSize(compressed_data->data, compressed_data->size)); + if (ZSTD_isError(uncompressed_data->size)) { + log_message(LOG_LEVEL_ERROR, "Decompression failed: %s", + ZSTD_getErrorName(uncompressed_data->size)); + exit(EXIT_FAILURE); + } + + uncompressed_data->size = + ZSTD_decompress(uncompressed_data->data, uncompressed_data->size, + compressed_data->data, compressed_data->size); + if (ZSTD_isError(uncompressed_data->size)) { + log_message(LOG_LEVEL_ERROR, "Decompression failed: %s", + ZSTD_getErrorName(uncompressed_data->size)); + exit(EXIT_FAILURE); + } + log_message(LOG_LEVEL_DEBUG, "Decompressed data successfully"); + return uncompressed_data; +} diff --git a/src/shared/compression.h b/src/shared/compression.h new file mode 100644 index 0000000..d902cfa --- /dev/null +++ b/src/shared/compression.h @@ -0,0 +1,9 @@ +#ifndef COMPRESSION_H +#define COMPRESSION_H + +#include "data.h" + +Data *data_compress(Data *data_to_compress, int compression_level); +Data *data_decompress(Data *compressed_data); + +#endif diff --git a/src/shared/config.c b/src/shared/config.c index da31c4f..6267c1a 100644 --- a/src/shared/config.c +++ b/src/shared/config.c @@ -1,5 +1,5 @@ #include "config.h" -#include "socket.h" +#include "protocol.h" #include "utils.h" #include #include diff --git a/src/shared/data.c b/src/shared/data.c index 7528230..648eb1d 100644 --- a/src/shared/data.c +++ b/src/shared/data.c @@ -1,8 +1,6 @@ - #include "data.h" #include "log.h" #include "stdlib.h" -#include "zstd.h" Data *data_create_empty(size_t data_size) { void *data = malloc(data_size); @@ -40,44 +38,3 @@ void data_destroy(Data *data) { free(data->data); free(data); } - -Data *data_compress(Data *data_to_compress, int compression_level) { - log_message(LOG_LEVEL_DEBUG, "Starting to compress data"); - Data *compressed_data = - data_create_empty(ZSTD_compressBound(data_to_compress->size)); - - compressed_data->size = ZSTD_compress( - compressed_data->data, compressed_data->size, data_to_compress->data, - data_to_compress->size, compression_level); - if (ZSTD_isError(compressed_data->size)) { - log_message(LOG_LEVEL_ERROR, "Compression failed: %s", - ZSTD_getErrorName(compressed_data->size)); - exit(EXIT_FAILURE); - } - - log_message(LOG_LEVEL_DEBUG, "Data succesfully compressed from %zu to %zu", - data_to_compress->size, compressed_data->size); - return compressed_data; -} - -Data *data_decompress(Data *compressed_data) { - log_message(LOG_LEVEL_DEBUG, "Start to decompress data"); - Data *uncompressed_data = data_create_empty( - ZSTD_getFrameContentSize(compressed_data->data, compressed_data->size)); - if (ZSTD_isError(uncompressed_data->size)) { - log_message(LOG_LEVEL_ERROR, "Decompression failed: %s", - ZSTD_getErrorName(uncompressed_data->size)); - exit(EXIT_FAILURE); - } - - uncompressed_data->size = - ZSTD_decompress(uncompressed_data->data, uncompressed_data->size, - compressed_data->data, compressed_data->size); - if (ZSTD_isError(uncompressed_data->size)) { - log_message(LOG_LEVEL_ERROR, "Decompression failed: %s", - ZSTD_getErrorName(uncompressed_data->size)); - exit(EXIT_FAILURE); - } - log_message(LOG_LEVEL_DEBUG, "Decompressed data successfully"); - return uncompressed_data; -} diff --git a/src/shared/data.h b/src/shared/data.h index 0fa33da..53ed762 100644 --- a/src/shared/data.h +++ b/src/shared/data.h @@ -12,7 +12,5 @@ Data *data_create_empty(size_t data_size); Data *data_create_reserve(size_t size); Data *data_create(void *data, size_t data_size); void data_destroy(Data *data); -Data *data_compress(Data *data_to_compress, int compression_level); -Data *data_decompress(Data *compressed_data); #endif diff --git a/src/shared/file.c b/src/shared/file.c index 3ee0511..3273b46 100644 --- a/src/shared/file.c +++ b/src/shared/file.c @@ -6,13 +6,16 @@ #include #include #include +#include #include -#include #include "data.h" #include "file.h" +#include "io.h" #include "log.h" -#include "socket.h" +#include "metadata.h" +#include "protocol.h" +#include "utils.h" File *file_create(const char *path) { File *file = (File *)malloc(sizeof(File)); @@ -87,35 +90,6 @@ void file_load_data(File *file) { } } -static void metadata_send(int file_descriptor, FileMetadata *m) { - if (m == NULL) { - int zero = 0; - send_n_data(file_descriptor, &zero, sizeof(int)); - return; - } - 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)); - send_n_data(file_descriptor, &m->mtime_nsec, sizeof(long)); -} - -FileMetadata *file_receive_metadata(int file_descriptor) { - int present; - receive_n_data(file_descriptor, &present, sizeof(int)); - 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)); - return m; -} - void file_send_single_calls(File *file, int file_descriptor, bool use_metadata) { send_str(file_descriptor, file->path); if (use_metadata) @@ -123,6 +97,21 @@ void file_send_single_calls(File *file, int file_descriptor, bool use_metadata) send_data(file_descriptor, file->data->data, file->data->size); } +void 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); + FILE *file_pointer = fopen(path, "wb"); + if (file_pointer == NULL) { + perror("Could not open File"); + exit(EXIT_FAILURE); + } + fwrite(data, 1, data_size, file_pointer); + fclose(file_pointer); + free(dir_to_free); +} + void file_send_sendfile(File *file, int file_descriptor, bool use_metadata) { send_str(file_descriptor, file->path); if (use_metadata) @@ -165,3 +154,5 @@ size_t file_content_to_buffer(File *file) { fclose(file_pointer); return bytes_read; } + + diff --git a/src/shared/file.h b/src/shared/file.h index 1b03197..535e693 100644 --- a/src/shared/file.h +++ b/src/shared/file.h @@ -27,6 +27,6 @@ void 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); -FileMetadata *file_receive_metadata(int file_descriptor); +void to_disk(const char *path, const void *data, unsigned long long data_size); #endif diff --git a/src/shared/io.c b/src/shared/io.c new file mode 100644 index 0000000..d2defa0 --- /dev/null +++ b/src/shared/io.c @@ -0,0 +1,49 @@ +#include "io.h" +#include "log.h" +#include +#include +#include + +static __thread int io_read_fd = -1; +static __thread int io_write_fd = -1; + +void io_set_fds(int read_fd, int write_fd) { + io_read_fd = read_fd; + io_write_fd = write_fd; +} + +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) { + 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; + while (total_bytes_send < 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); + } + total_bytes_send += bytes_send; + } + log_message(LOG_LEVEL_DEBUG, " Send n Data: %zu", total_bytes_send); +} + +void 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); + } + total_bytes_received += bytes_received; + } + log_message(LOG_LEVEL_DEBUG, " Received n Data: %zu", total_bytes_received); +} diff --git a/src/shared/io.h b/src/shared/io.h new file mode 100644 index 0000000..9da1fe1 --- /dev/null +++ b/src/shared/io.h @@ -0,0 +1,10 @@ +#ifndef IO_H +#define IO_H + +#include + +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); + +#endif diff --git a/src/shared/metadata.c b/src/shared/metadata.c new file mode 100644 index 0000000..6a0c283 --- /dev/null +++ b/src/shared/metadata.c @@ -0,0 +1,82 @@ +#include "metadata.h" +#include "file.h" +#include "io.h" +#include +#include +#include +#include +#include +#include + +#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) { + int present = (m != NULL) ? 1 : 0; + memcpy(*buf, &present, sizeof(int)); + *buf += sizeof(int); + if (m == NULL) + return; + memcpy(*buf, &m->mode, sizeof(mode_t)); *buf += sizeof(mode_t); + memcpy(*buf, &m->uid, sizeof(uid_t)); *buf += sizeof(uid_t); + 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) { + int present; + memcpy(&present, *buf, sizeof(int)); + *buf += sizeof(int); + if (!present) + return NULL; + FileMetadata *m = malloc(sizeof(FileMetadata)); + memcpy(&m->mode, *buf, sizeof(mode_t)); *buf += sizeof(mode_t); + memcpy(&m->uid, *buf, sizeof(uid_t)); *buf += sizeof(uid_t); + 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; +} + +void metadata_send(int file_descriptor, FileMetadata *m) { + if (m == NULL) { + int zero = 0; + send_n_data(file_descriptor, &zero, sizeof(int)); + return; + } + 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)); + 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 (!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)); + return m; +} + +void file_restore_metadata(const char *path, FileMetadata *metadata) { + if (metadata == NULL) + return; + chmod(path, metadata->mode & 07777); + int chown_ret = chown(path, metadata->uid, metadata->gid); + (void)chown_ret; + struct timespec times[2]; + times[0].tv_sec = 0; + times[0].tv_nsec = UTIME_OMIT; + times[1].tv_sec = metadata->mtime_sec; + times[1].tv_nsec = metadata->mtime_nsec; + utimensat(AT_FDCWD, path, times, 0); +} diff --git a/src/shared/metadata.h b/src/shared/metadata.h new file mode 100644 index 0000000..0edcdcf --- /dev/null +++ b/src/shared/metadata.h @@ -0,0 +1,15 @@ +#ifndef METADATA_H +#define METADATA_H + +#include "file.h" +#include + +#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); +FileMetadata *metadata_receive(int file_descriptor); +void file_restore_metadata(const char *path, FileMetadata *metadata); + +#endif diff --git a/src/shared/multiprocessing.c b/src/shared/multiprocessing.c index 7b52f3d..3be78a7 100644 --- a/src/shared/multiprocessing.c +++ b/src/shared/multiprocessing.c @@ -1,8 +1,17 @@ #include "multiprocessing.h" +#include "chunk.h" +#include "compression.h" #include "config.h" +#include "data.h" +#include "file.h" +#include "log.h" +#include "metadata.h" +#include "protocol.h" #include "queue.h" +#include "utils.h" #include #include +#include #include PipelineContextSender *pipeline_context_sender_create(Config *config, @@ -64,3 +73,94 @@ void pipeline_context_receiver_destroy(PipelineContextReceiver *context) { cnd_destroy(&context->condition_not_empty); free(context); } + +File *file_receive(Config *config, int file_descriptor) { + char *path = (char *)receive_str(file_descriptor); + File *file = file_create(path); + free(path); + if (config->use_metadata) + file->metadata = metadata_receive(file_descriptor); + Data *file_data = receive_data(file_descriptor); + if (config->use_compression) { + Data *file_data_uncompressed = data_decompress(file_data); + data_destroy(file_data); + file_data = file_data_uncompressed; + } + data_destroy(file->data); + file->data = file_data; + return file; +} + +static void receive_chunk_enqueue(int file_descriptor, + PipelineContextReceiver *context) { + Data *chunk_data = receive_data(file_descriptor); + Data *data_to_process = chunk_data; + if (context->config->use_compression) { + data_to_process = data_decompress(chunk_data); + data_destroy(chunk_data); + } + Chunk *chunk = chunk_deserialize(data_to_process, context->config->use_metadata); + data_destroy(data_to_process); + + for (int i = 0; i < chunk->element_count; i++) { + File *file = chunk->items[i]; + chunk->items[i] = NULL; + queue_enqueue_multithreaded(context->queue, file, &context->mutex, + &context->condition_not_empty, + &context->condition_not_full); + } + chunk_destroy(chunk); +} + +int receive_thread(void *pipeline_context) { + PipelineContextReceiver *context = + (PipelineContextReceiver *)pipeline_context; + mtx_lock(&context->mutex); + int file_descriptor = context->file_descriptor; + Config *config = context->config; + mtx_unlock(&context->mutex); + + Status status = receive_status(file_descriptor); + 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); + queue_enqueue_multithreaded(context->queue, file, &context->mutex, + &context->condition_not_empty, + &context->condition_not_full); + } + status = receive_status(file_descriptor); + } + mtx_lock(&context->mutex); + context->receiver_done = true; + cnd_signal(&context->condition_not_empty); + mtx_unlock(&context->mutex); + return thrd_success; +} + +int write_thread(void *pipeline_context) { + PipelineContextReceiver *context = + (PipelineContextReceiver *)pipeline_context; + mtx_lock(&context->mutex); + bool save_to_disk = context->config->save_to_disk; + char *root_directory = str_dup(context->config->receive_root_directory); + mtx_unlock(&context->mutex); + + while (true) { + File *file = queue_dequeue_multithreaded( + context->queue, &context->mutex, &context->condition_not_empty, + &context->condition_not_full, &context->receiver_done); + if (file == NULL) { + free(root_directory); + return thrd_success; + } + if (save_to_disk) { + char *disk_path = path_cat(root_directory, file->path); + to_disk(disk_path, file->data->data, file->data->size); + file_restore_metadata(disk_path, file->metadata); + free(disk_path); + } + file_destroy(file); + } +} diff --git a/src/shared/multiprocessing.h b/src/shared/multiprocessing.h index 909987c..22fad81 100644 --- a/src/shared/multiprocessing.h +++ b/src/shared/multiprocessing.h @@ -4,6 +4,7 @@ #include #include "config.h" +#include "file.h" #include "queue.h" typedef struct { @@ -38,4 +39,7 @@ PipelineContextReceiver *pipeline_context_receiver_create(Config *config, Queue *queue_receiver, int file_descriptor); void pipeline_context_receiver_destroy(PipelineContextReceiver *context); +File *file_receive(Config *config, int file_descriptor); +int receive_thread(void *pipeline_context); +int write_thread(void *pipeline_context); #endif diff --git a/src/shared/protocol.c b/src/shared/protocol.c new file mode 100644 index 0000000..2e86dba --- /dev/null +++ b/src/shared/protocol.c @@ -0,0 +1,79 @@ +#include "protocol.h" +#include "io.h" +#include "log.h" +#include +#include +#include + +static const char *status_to_string(Status status) { + switch (status) { + case STATUS_OK: + return "OK"; + case STATUS_ERROR: + return "ERROR"; + case STATUS_FINISHED: + return "FINISHED"; + case STATUS_NEXT: + return "NEXT"; + case STATUS_CHUNK: + return "CHUNK"; + default: + return "UNKNOWN"; + } +} + +void 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); + log_message(LOG_LEVEL_DEBUG, "Send String: %s", data); +} + +char *receive_str(int file_descriptor) { + size_t size; + receive_n_data(file_descriptor, &size, sizeof(size_t)); + char *data = (char *)malloc(size + 1); + receive_n_data(file_descriptor, data, size); + data[size] = '\0'; + log_message(LOG_LEVEL_DEBUG, "Received String: %s", data); + return data; +} + +void send_data(int file_descriptor, void *data, unsigned long long data_size) { + send_n_data(file_descriptor, &data_size, sizeof(unsigned long long)); + send_n_data(file_descriptor, data, data_size); + log_message(LOG_LEVEL_DEBUG, "Send %lld data", data_size); +} + +Data *receive_data(int file_descriptor) { + size_t size = 0; + receive_n_data(file_descriptor, &size, sizeof(unsigned long long)); + void *data = malloc(size); + receive_n_data(file_descriptor, data, size); + log_message(LOG_LEVEL_DEBUG, "Received %lld data", size); + return data_create(data, size); +} + +void send_int(int file_descriptor, int data) { + send_n_data(file_descriptor, &data, sizeof(int)); + log_message(LOG_LEVEL_DEBUG, "Send Int: %d", data); +} + +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; +} + +void send_status(int file_descriptor, Status status) { + send_n_data(file_descriptor, &status, sizeof(Status)); + log_message(LOG_LEVEL_DEBUG, "Send Status: %s", status_to_string(status)); +} + +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; +} diff --git a/src/shared/protocol.h b/src/shared/protocol.h new file mode 100644 index 0000000..c13e3ac --- /dev/null +++ b/src/shared/protocol.h @@ -0,0 +1,20 @@ +#ifndef PROTOCOL_H +#define PROTOCOL_H + +#include "data.h" +#include "io.h" +#include + +typedef int Status; +enum NET_STATUS { STATUS_OK, STATUS_ERROR, STATUS_FINISHED, STATUS_NEXT, STATUS_CHUNK }; + +void send_str(int file_descriptor, char *data); +char *receive_str(int file_descriptor); +void send_data(int file_descriptor, void *data, unsigned long long data_size); +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); + +#endif diff --git a/src/shared/socket.c b/src/shared/socket.c deleted file mode 100644 index fffada4..0000000 --- a/src/shared/socket.c +++ /dev/null @@ -1,346 +0,0 @@ -#include "socket.h" -#include "log.h" -#include -#include -#include -#include -#include -#include -#include -#include -#include - -static __thread int io_read_fd = -1; -static __thread int io_write_fd = -1; - -void io_set_fds(int read_fd, int write_fd) { - io_read_fd = read_fd; - io_write_fd = write_fd; -} - -static int io_fd(int dir_fd, int file_descriptor) { - return (dir_fd != -1) ? dir_fd : file_descriptor; -} - -typedef struct { - char user[256]; - char host[256]; - char remote_path[4096]; -} RemoteDest; - -static int parse_remote_dest(const char *dest, RemoteDest *r) { - const char *colon = strchr(dest, ':'); - if (!colon) return -1; - - size_t remote_path_len = strlen(colon + 1); - if (remote_path_len >= sizeof(r->remote_path)) return -1; - memcpy(r->remote_path, colon + 1, remote_path_len + 1); - - const char *at = memchr(dest, '@', colon - dest); - if (at) { - size_t user_len = at - dest; - if (user_len >= sizeof(r->user)) return -1; - memcpy(r->user, dest, user_len); - r->user[user_len] = '\0'; - - size_t host_len = colon - at - 1; - if (host_len >= sizeof(r->host)) return -1; - memcpy(r->host, at + 1, host_len); - r->host[host_len] = '\0'; - } else { - r->user[0] = '\0'; - size_t host_len = colon - dest; - if (host_len >= sizeof(r->host)) return -1; - memcpy(r->host, dest, host_len); - r->host[host_len] = '\0'; - } - return 0; -} - -Client *client_connect_ssh(char *destination) { - RemoteDest r; - if (parse_remote_dest(destination, &r) != 0) { - fprintf(stderr, "Invalid remote destination: %s\n", destination); - exit(EXIT_FAILURE); - } - - int sv[2]; - if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv) < 0) { - perror("socketpair failed"); - exit(EXIT_FAILURE); - } - - int exec_pipe[2]; - if (pipe(exec_pipe) < 0) { - perror("pipe failed"); - exit(EXIT_FAILURE); - } - - pid_t pid = fork(); - if (pid < 0) { - perror("fork failed"); - exit(EXIT_FAILURE); - } - - if (pid == 0) { - close(sv[0]); - close(exec_pipe[0]); - fcntl(exec_pipe[1], F_SETFD, FD_CLOEXEC); - - if (sv[1] != STDIN_FILENO) - dup2(sv[1], STDIN_FILENO); - if (sv[1] != STDOUT_FILENO) - dup2(sv[1], STDOUT_FILENO); - if (sv[1] > 1) close(sv[1]); - - char ssh_user[512]; - if (r.user[0] != '\0') - snprintf(ssh_user, sizeof(ssh_user), "%s@%s", r.user, r.host); - else - snprintf(ssh_user, sizeof(ssh_user), "%s", r.host); - - execlp("ssh", "ssh", "-o", "Compression=no", "-o", - "ControlMaster=no", ssh_user, "fastsync-server", "--stdio", - (char *)NULL); - perror("exec of ssh failed"); - (void)write(exec_pipe[1], "x", 1); - _exit(1); - } - - close(sv[1]); - close(exec_pipe[1]); - - char exec_status; - ssize_t n = read(exec_pipe[0], &exec_status, 1); - close(exec_pipe[0]); - - if (n > 0) { - close(sv[0]); - waitpid(pid, NULL, 0); - fprintf(stderr, "Error: could not launch 'fastsync-server --stdio' on remote\n"); - exit(EXIT_FAILURE); - } - - Client *client = malloc(sizeof(Client)); - client->file_descriptor = sv[0]; - client->address.sin_family = AF_UNIX; - client->address_length = 0; - client->ssh_child_pid = pid; - return client; -} - -Server *server_create(int port) { - Server *server = (Server *)malloc(sizeof(Server)); - if (server == NULL) { - perror("Could not allocate space for Server"); - exit(EXIT_FAILURE); - } - - int file_descriptor = socket(AF_INET, SOCK_STREAM, 0); - if (file_descriptor < 0) { - perror("Could not create Socket!"); - exit(EXIT_FAILURE); - } - server->file_descriptor = file_descriptor; - int opt = 1; - if (setsockopt(server->file_descriptor, SOL_SOCKET, SO_REUSEADDR, &opt, - sizeof(opt))) { - perror("Error setting a socket option!"); - close(server->file_descriptor); - free(server); - exit(EXIT_FAILURE); - } - - server->address.sin_family = AF_INET; - server->address.sin_addr.s_addr = INADDR_ANY; - server->address.sin_port = htons(port); - server->address_length = sizeof(server->address); - - if (bind(server->file_descriptor, (struct sockaddr *)&server->address, - server->address_length) < 0) { - perror("Could not bind server"); - close(server->file_descriptor); - free(server); - exit(EXIT_FAILURE); - } - - return server; -} - -void server_delete(Server **server) { - if (server == NULL || *server == NULL) return; - free(*server); - *server = NULL; -} - -void 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, 3) < 0) { - perror("Could not listen on port!"); - exit(EXIT_FAILURE); - } - - int file_descriptor = - accept(server->file_descriptor, (struct sockaddr *)&server->address, - &server->address_length); - if (file_descriptor < 0) { - perror("Could not accept the connection"); - exit(EXIT_FAILURE); - } - log_message(LOG_LEVEL_INFO, "Received Connection"); - handler(file_descriptor); - close(server->file_descriptor); - close(file_descriptor); -} - -Client *client_create() { - int file_descriptor = socket(AF_INET, SOCK_STREAM, 0); - if (file_descriptor < 0) { - perror("Could not create Socket!"); - exit(EXIT_FAILURE); - }; - - Client *client = (Client *)malloc(sizeof(Client)); - client->file_descriptor = file_descriptor; - client->address.sin_family = AF_INET; - client->address_length = sizeof(client->address); - client->ssh_child_pid = -1; - return client; -} - -void 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); - } - - if (connect(client->file_descriptor, (struct sockaddr *)&client->address, - client->address_length) < 0) { - perror("Could not connect to Server!"); - exit(EXIT_FAILURE); - } -} - -void client_disconnect(Client *client) { - close(client->file_descriptor); - if (client->ssh_child_pid > 0) { - int status; - waitpid(client->ssh_child_pid, &status, 0); - client->ssh_child_pid = -1; - } -} - -void client_delete(Client *client) { - if (client == NULL) - return; - free(client); -} - -void 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; - while (total_bytes_send < 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); - } - total_bytes_send += bytes_send; - } - log_message(LOG_LEVEL_DEBUG, " Send n Data: %zu", total_bytes_send); -} - -void 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); - } - total_bytes_received += bytes_received; - } - log_message(LOG_LEVEL_DEBUG, " Received n Data: %zu", total_bytes_received); -} - -void 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); - log_message(LOG_LEVEL_DEBUG, "Send String: %s", data); -} - -char *receive_str(int file_descriptor) { - size_t size; - receive_n_data(file_descriptor, &size, sizeof(size_t)); - char *data = (char *)malloc(size + 1); - receive_n_data(file_descriptor, data, size); - data[size] = '\0'; - log_message(LOG_LEVEL_DEBUG, "Received String: %s", data); - return data; -} - -void send_data(int file_descriptor, void *data, unsigned long long data_size) { - send_n_data(file_descriptor, &data_size, sizeof(unsigned long long)); - send_n_data(file_descriptor, data, data_size); - log_message(LOG_LEVEL_DEBUG, "Send %lld data", data_size); -} - -Data *receive_data(int file_descriptor) { - size_t size = 0; - receive_n_data(file_descriptor, &size, sizeof(unsigned long long)); - void *data = malloc(size); - receive_n_data(file_descriptor, data, size); - log_message(LOG_LEVEL_DEBUG, "Received %lld data", size); - return data_create(data, size); -} - -void send_int(int file_descriptor, int data) { - send_n_data(file_descriptor, &data, sizeof(int)); - log_message(LOG_LEVEL_DEBUG, "Send Int: %d", data); -} - -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; -} - -const char *status_to_string(Status status) { - switch (status) { - case STATUS_OK: - return "OK"; - case STATUS_ERROR: - return "ERROR"; - case STATUS_FINISHED: - return "FINISHED"; - case STATUS_NEXT: - return "NEXT"; - case STATUS_CHUNK: - return "CHUNK"; - default: - return "UNKNOWN"; - } -} - -void send_status(int file_descriptor, Status status) { - send_n_data(file_descriptor, &status, sizeof(Status)); - log_message(LOG_LEVEL_DEBUG, "Send Status: %s", status_to_string(status)); -} - -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; -} diff --git a/src/shared/socket.h b/src/shared/socket.h deleted file mode 100644 index 9a22e10..0000000 --- a/src/shared/socket.h +++ /dev/null @@ -1,45 +0,0 @@ -#ifndef SOCKET_H -#define SOCKET_H - -#include "data.h" -#include - -typedef int Status; -enum NET_STATUS { STATUS_OK, STATUS_ERROR, STATUS_FINISHED, STATUS_NEXT, STATUS_CHUNK }; - -typedef struct Server { - struct sockaddr_in address; - unsigned int address_length; - int file_descriptor; -} Server; - -Server *server_create(int port); -void server_listen(Server *server, void (*handler)(int file_descriptor)); -void server_delete(Server **server); - -typedef struct Client { - struct sockaddr_in address; - unsigned int address_length; - int file_descriptor; - pid_t ssh_child_pid; -} Client; - -Client *client_create(); -void client_disconnect(Client *client); -void client_delete(Client *client); -void client_connect(Client *client, char *host, int port); -Client *client_connect_ssh(char *destination); - -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); -void send_str(int file_descriptor, char *data); -char *receive_str(int file_descriptor); -void send_data(int file_descriptor, void *data, unsigned long long data_size); -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); - -#endif diff --git a/src/shared/transport_ssh.c b/src/shared/transport_ssh.c new file mode 100644 index 0000000..dd46a00 --- /dev/null +++ b/src/shared/transport_ssh.c @@ -0,0 +1,115 @@ +#include "transport_ssh.h" +#include +#include +#include +#include +#include +#include +#include + +typedef struct { + char user[256]; + char host[256]; + char remote_path[4096]; +} RemoteDest; + +static int parse_remote_dest(const char *dest, RemoteDest *r) { + const char *colon = strchr(dest, ':'); + if (!colon) return -1; + + size_t remote_path_len = strlen(colon + 1); + if (remote_path_len >= sizeof(r->remote_path)) return -1; + memcpy(r->remote_path, colon + 1, remote_path_len + 1); + + const char *at = memchr(dest, '@', colon - dest); + if (at) { + size_t user_len = at - dest; + if (user_len >= sizeof(r->user)) return -1; + memcpy(r->user, dest, user_len); + r->user[user_len] = '\0'; + + size_t host_len = colon - at - 1; + if (host_len >= sizeof(r->host)) return -1; + memcpy(r->host, at + 1, host_len); + r->host[host_len] = '\0'; + } else { + r->user[0] = '\0'; + size_t host_len = colon - dest; + if (host_len >= sizeof(r->host)) return -1; + memcpy(r->host, dest, host_len); + r->host[host_len] = '\0'; + } + return 0; +} + +Client *client_connect_ssh(char *destination) { + RemoteDest r; + if (parse_remote_dest(destination, &r) != 0) { + fprintf(stderr, "Invalid remote destination: %s\n", destination); + exit(EXIT_FAILURE); + } + + int sv[2]; + if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv) < 0) { + perror("socketpair failed"); + exit(EXIT_FAILURE); + } + + int exec_pipe[2]; + if (pipe(exec_pipe) < 0) { + perror("pipe failed"); + exit(EXIT_FAILURE); + } + + pid_t pid = fork(); + if (pid < 0) { + perror("fork failed"); + exit(EXIT_FAILURE); + } + + if (pid == 0) { + close(sv[0]); + close(exec_pipe[0]); + fcntl(exec_pipe[1], F_SETFD, FD_CLOEXEC); + + if (sv[1] != STDIN_FILENO) + dup2(sv[1], STDIN_FILENO); + if (sv[1] != STDOUT_FILENO) + dup2(sv[1], STDOUT_FILENO); + if (sv[1] > 1) close(sv[1]); + + char ssh_user[512]; + if (r.user[0] != '\0') + snprintf(ssh_user, sizeof(ssh_user), "%s@%s", r.user, r.host); + else + snprintf(ssh_user, sizeof(ssh_user), "%s", r.host); + + execlp("ssh", "ssh", "-o", "Compression=no", "-o", + "ControlMaster=no", ssh_user, "fastsync-server", "--stdio", + (char *)NULL); + perror("exec of ssh failed"); + (void)write(exec_pipe[1], "x", 1); + _exit(1); + } + + close(sv[1]); + close(exec_pipe[1]); + + char exec_status; + ssize_t n = read(exec_pipe[0], &exec_status, 1); + close(exec_pipe[0]); + + if (n > 0) { + close(sv[0]); + waitpid(pid, NULL, 0); + fprintf(stderr, "Error: could not launch 'fastsync-server --stdio' on remote\n"); + exit(EXIT_FAILURE); + } + + Client *client = malloc(sizeof(Client)); + client->file_descriptor = sv[0]; + client->address.sin_family = AF_UNIX; + client->address_length = 0; + client->ssh_child_pid = pid; + return client; +} diff --git a/src/shared/transport_ssh.h b/src/shared/transport_ssh.h new file mode 100644 index 0000000..3bad43c --- /dev/null +++ b/src/shared/transport_ssh.h @@ -0,0 +1,8 @@ +#ifndef TRANSPORT_SSH_H +#define TRANSPORT_SSH_H + +#include "transport_tcp.h" + +Client *client_connect_ssh(char *destination); + +#endif diff --git a/src/shared/transport_tcp.c b/src/shared/transport_tcp.c new file mode 100644 index 0000000..1349a9a --- /dev/null +++ b/src/shared/transport_tcp.c @@ -0,0 +1,119 @@ +#include "transport_tcp.h" +#include "log.h" +#include +#include +#include +#include +#include +#include +#include + +Server *server_create(int port) { + Server *server = (Server *)malloc(sizeof(Server)); + if (server == NULL) { + perror("Could not allocate space for Server"); + exit(EXIT_FAILURE); + } + + int file_descriptor = socket(AF_INET, SOCK_STREAM, 0); + if (file_descriptor < 0) { + perror("Could not create Socket!"); + exit(EXIT_FAILURE); + } + server->file_descriptor = file_descriptor; + int opt = 1; + if (setsockopt(server->file_descriptor, SOL_SOCKET, SO_REUSEADDR, &opt, + sizeof(opt))) { + perror("Error setting a socket option!"); + close(server->file_descriptor); + free(server); + exit(EXIT_FAILURE); + } + + server->address.sin_family = AF_INET; + server->address.sin_addr.s_addr = INADDR_ANY; + server->address.sin_port = htons(port); + server->address_length = sizeof(server->address); + + if (bind(server->file_descriptor, (struct sockaddr *)&server->address, + server->address_length) < 0) { + perror("Could not bind server"); + close(server->file_descriptor); + free(server); + exit(EXIT_FAILURE); + } + + return server; +} + +void server_delete(Server **server) { + if (server == NULL || *server == NULL) return; + free(*server); + *server = NULL; +} + +void 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, 3) < 0) { + perror("Could not listen on port!"); + exit(EXIT_FAILURE); + } + + int file_descriptor = + accept(server->file_descriptor, (struct sockaddr *)&server->address, + &server->address_length); + if (file_descriptor < 0) { + perror("Could not accept the connection"); + exit(EXIT_FAILURE); + } + log_message(LOG_LEVEL_INFO, "Received Connection"); + handler(file_descriptor); + close(server->file_descriptor); + close(file_descriptor); +} + +Client *client_create() { + int file_descriptor = socket(AF_INET, SOCK_STREAM, 0); + if (file_descriptor < 0) { + perror("Could not create Socket!"); + exit(EXIT_FAILURE); + } + + Client *client = (Client *)malloc(sizeof(Client)); + client->file_descriptor = file_descriptor; + client->address.sin_family = AF_INET; + client->address_length = sizeof(client->address); + client->ssh_child_pid = -1; + return client; +} + +void 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); + } + + if (connect(client->file_descriptor, (struct sockaddr *)&client->address, + client->address_length) < 0) { + perror("Could not connect to Server!"); + exit(EXIT_FAILURE); + } +} + +void client_disconnect(Client *client) { + close(client->file_descriptor); + if (client->ssh_child_pid > 0) { + int status; + waitpid(client->ssh_child_pid, &status, 0); + client->ssh_child_pid = -1; + } +} + +void client_delete(Client *client) { + if (client == NULL) + return; + free(client); +} diff --git a/src/shared/transport_tcp.h b/src/shared/transport_tcp.h new file mode 100644 index 0000000..7c394ac --- /dev/null +++ b/src/shared/transport_tcp.h @@ -0,0 +1,28 @@ +#ifndef TRANSPORT_TCP_H +#define TRANSPORT_TCP_H + +#include +#include + +typedef struct Server { + struct sockaddr_in address; + unsigned int address_length; + int file_descriptor; +} Server; + +typedef struct Client { + struct sockaddr_in address; + unsigned int address_length; + int file_descriptor; + pid_t ssh_child_pid; +} Client; + +Server *server_create(int port); +void 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); +void client_disconnect(Client *client); +void client_delete(Client *client); + +#endif diff --git a/src/shared/utils.c b/src/shared/utils.c index e848019..7054d44 100644 --- a/src/shared/utils.c +++ b/src/shared/utils.c @@ -1,12 +1,9 @@ #include "utils.h" #include "libgen.h" -#include "sys/stat.h" -#include #include #include #include -#include -#include +#include void mkdir_r(char *path) { char *path_duplicate = malloc(strlen(path) + 1); @@ -39,7 +36,7 @@ void mkdir_r(char *path) { free(path_current); } -char *str_dup(char *string) { +char *str_dup(const char *string) { if (string == NULL) return NULL; char *new_string = (char *)malloc(strlen(string) + 1); @@ -47,34 +44,6 @@ char *str_dup(char *string) { return new_string; } -void to_disk(char *path, void *data, unsigned long long data_size) { - char *directory = str_dup(path); - char *dir_to_free = directory; - directory = dirname(directory); - mkdir_r(directory); - FILE *file_pointer = fopen(path, "wb"); - if (file_pointer == NULL) { - perror("Could not open File"); - exit(EXIT_FAILURE); - } - fwrite(data, 1, data_size, file_pointer); - fclose(file_pointer); - free(dir_to_free); -} - -void file_restore_metadata(const char *path, FileMetadata *metadata) { - if (metadata == NULL) - return; - chmod(path, metadata->mode & 07777); - chown(path, metadata->uid, metadata->gid); - struct timespec times[2]; - times[0].tv_sec = 0; - times[0].tv_nsec = UTIME_OMIT; - times[1].tv_sec = metadata->mtime_sec; - times[1].tv_nsec = metadata->mtime_nsec; - utimensat(AT_FDCWD, path, times, 0); -} - char *path_cat(char *path1, char *path2) { if (path1 == NULL || *path1 == '\0') return str_dup(path2); diff --git a/src/shared/utils.h b/src/shared/utils.h index a3d8838..f54c98d 100644 --- a/src/shared/utils.h +++ b/src/shared/utils.h @@ -1,12 +1,8 @@ #ifndef UTILS_H #define UTILS_H -#include "file.h" - void mkdir_r(char *path); -char *str_dup(char *string); -void to_disk(char *path, void *data, unsigned long long data_size); -void file_restore_metadata(const char *path, FileMetadata *metadata); +char *str_dup(const char *string); char *path_cat(char *path1, char *path2); #endif diff --git a/tests/test_compression.c b/tests/test_compression.c index 869b163..e793941 100644 --- a/tests/test_compression.c +++ b/tests/test_compression.c @@ -1,5 +1,6 @@ #include "test_utils.h" #include "chunk.h" +#include "compression.h" #include "data.h" #include "file.h" #include "utils.h"