From b3f69208ce0f1271b588843b4730ec49227761d1 Mon Sep 17 00:00:00 2001 From: TapTap Date: Tue, 7 Jul 2026 19:07:24 +0200 Subject: [PATCH 1/3] cleanup: remove dead code, fix bugs, refactor SSH dest parsing - Remove debug printf calls from file.c - Remove dead code: chunk_print, chunk_format, chunk_data_create/delete - Remove unused config.use_single_send_per_file field - Fix server_delete() no-op (double-pointer) - Fix double cast (long)(long) -> ptrdiff_t in chunk.c - Fix thread return value: thrd_error instead of 1 - Remove unused config param from receive_chunk_enqueue() - Make queue_double_capacity() static - Move SSH dest parsing into config.c as config_parse_ssh_dest() - Fix test_config_ssh_dest to test parsing round-trip - Remove chunk_format test (obsolete format) - Replace chunk_data_delete with data_destroy in tests --- src/client/client.c | 21 ++------------ src/server/server.c | 11 ++++--- src/shared/chunk.c | 71 --------------------------------------------- src/shared/chunk.h | 5 +--- src/shared/config.c | 23 +++++++++++++++ src/shared/config.h | 3 +- src/shared/file.c | 8 ----- src/shared/file.h | 1 - src/shared/queue.c | 2 +- src/shared/queue.h | 1 - src/shared/socket.c | 9 +++--- src/shared/socket.h | 2 +- tests/test_chunk.c | 70 +------------------------------------------- tests/test_config.c | 31 ++++++++++++++++++-- 14 files changed, 69 insertions(+), 189 deletions(-) diff --git a/src/client/client.c b/src/client/client.c index beb2253..a003743 100644 --- a/src/client/client.c +++ b/src/client/client.c @@ -125,7 +125,7 @@ int send_chunks_multithreaded(void *pipeline_context) { int ok = receive_status(client->file_descriptor) == STATUS_OK; client_disconnect(client); client_delete(client); - return ok ? thrd_success : 1; + return ok ? thrd_success : thrd_error; } if (send_chunk(client, current_chunk, context->config) != 0) { perror("Something unexpected happend while sending the chunk"); @@ -189,16 +189,6 @@ int send_files_multithreaded(Config *config) { return 0; } -static int is_remote_dest(const char *s) { - const char *colon = strchr(s, ':'); - if (!colon) return 0; - if (colon == s) return 0; - for (const char *p = s; p < colon; p++) { - if (*p == '/') return 0; - } - return 1; -} - static void print_usage(void) { printf("Usage:\n"); printf(" fastsync [options] \n"); @@ -307,14 +297,7 @@ int main(int argc, char *argv[]) { config->receive_root_directory = str_dup(argv[positional_args[1]]); config->save_to_disk = true; - if (is_remote_dest(config->receive_root_directory)) { - config->transport = TRANSPORT_SSH; - config->ssh_destination = str_dup(config->receive_root_directory); - char *colon = strchr(config->receive_root_directory, ':'); - char *path = str_dup(colon + 1); - free(config->receive_root_directory); - config->receive_root_directory = path; - } + config_parse_ssh_dest(config); } else if (positional_count == 1) { fprintf(stderr, "Error: missing destination argument\n"); print_usage(); diff --git a/src/server/server.c b/src/server/server.c index a0f731f..6d1ef64 100644 --- a/src/server/server.c +++ b/src/server/server.c @@ -30,16 +30,15 @@ File *file_receive(Config *config, int file_descriptor) { return file; } -static void receive_chunk_enqueue(int file_descriptor, Config *config, +static void receive_chunk_enqueue(int file_descriptor, PipelineContextReceiver *context) { - (void)config; Data *chunk_data = receive_data(file_descriptor); Data *data_to_process = chunk_data; - if (config->use_compression) { + if (context->config->use_compression) { data_to_process = data_decompress(chunk_data); data_destroy(chunk_data); } - Chunk *chunk = chunk_deserialize(data_to_process, config->use_metadata); + 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++) { @@ -63,7 +62,7 @@ int receive_thread(void *pipeline_context) { Status status = receive_status(file_descriptor); while (status == STATUS_NEXT || status == STATUS_CHUNK) { if (status == STATUS_CHUNK) { - receive_chunk_enqueue(file_descriptor, config, context); + receive_chunk_enqueue(file_descriptor, context); } else { File *file = file_receive(config, file_descriptor); queue_enqueue_multithreaded(context->queue, file, &context->mutex, @@ -180,6 +179,6 @@ int main(int argc, char *argv[]) { } Server *server = server_create(8080); server_listen(server, handler); - server_delete(server); + server_delete(&server); return 0; } diff --git a/src/shared/chunk.c b/src/shared/chunk.c index 6157ece..e405ce6 100644 --- a/src/shared/chunk.c +++ b/src/shared/chunk.c @@ -49,15 +49,6 @@ void chunk_destroy(void *item) { free(chunk); } -void chunk_print(void *item) { - if (item == NULL) - return; - Chunk *chunk = (Chunk *)item; - for (int i = 0; i < chunk->element_count; ++i) - if (chunk->items[i] != NULL) - file_print(chunk->items[i]); -} - static void metadata_to_buf(char **buf, FileMetadata *m) { int present = (m != NULL) ? 1 : 0; memcpy(*buf, &present, sizeof(int)); @@ -86,53 +77,6 @@ static FileMetadata *metadata_from_buf(char **buf) { return m; } -static unsigned long long per_file_chunk_format_size(File *file) { - return sizeof(int) + strlen(file->path) + sizeof(int) + - (file->metadata ? FILE_METADATA_WIRE_SIZE : 0) + - sizeof(unsigned long long) + file->data->size; -} - -Data *chunk_format(Chunk *chunk) { - unsigned long long buffer_size = 0; - for (int i = 0; i < chunk->element_count; ++i) { - buffer_size += per_file_chunk_format_size(chunk->items[i]); - } - - char *data = malloc(buffer_size); - if (data == NULL) { - perror("Could not allocate data for ChunkFormated!"); - exit(EXIT_FAILURE); - } - char *current_data_pointer = data; - for (int i = 0; i < chunk->element_count; ++i) { - File *file = chunk->items[i]; - // add path len - int path_length = (int)strlen(file->path); - memcpy(current_data_pointer, &path_length, sizeof(int)); - current_data_pointer += sizeof(int); - // add path - memcpy(current_data_pointer, file->path, path_length); - current_data_pointer += path_length; - // add metadata - metadata_to_buf(¤t_data_pointer, file->metadata); - // add file data len - unsigned long long file_length = file->data->size; - memcpy(current_data_pointer, &file_length, sizeof(unsigned long long)); - current_data_pointer += sizeof(unsigned long long); - // add file data - if (file->data->data == NULL) { - file_load_data(file); - } - memcpy(current_data_pointer, file->data->data, file_length); - current_data_pointer += file_length; - } - if (current_data_pointer - data != (long)(long)buffer_size) { - perror("Buffer of Chunk wasn't filled enough!"); - exit(EXIT_FAILURE); - } - return chunk_data_create(data, buffer_size); -} - 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) + @@ -284,19 +228,4 @@ Chunk *chunk_decompress(Data *compressed_data, bool use_metadata) { return chunk; } -Data *chunk_data_create(void *data, unsigned long long data_size) { - Data *chunk_formated = malloc(sizeof(Data)); - if (chunk_formated == NULL) { - perror("Could not allocate memory for ChunkFormated"); - exit(EXIT_FAILURE); - } - chunk_formated->data = data; - chunk_formated->size = data_size; - return chunk_formated; -} -void chunk_data_delete(void *chunk) { - Data *chunk_data = (Data *)chunk; - free(chunk_data->data); - free(chunk_data); -} diff --git a/src/shared/chunk.h b/src/shared/chunk.h index b89896d..1d5c80d 100644 --- a/src/shared/chunk.h +++ b/src/shared/chunk.h @@ -15,13 +15,10 @@ typedef struct { Chunk *chunk_create(File **items, int element_count); void chunk_destroy(void *chunk); -void chunk_print(void *chunk); -Data *chunk_format(Chunk *chunk); Data *chunk_serialize(Chunk *chunk, bool use_metadata); Chunk *chunk_deserialize(Data *data, bool use_metadata); Data *chunk_compress(Chunk *chunk, int compression_level, bool use_metadata); Chunk *chunk_decompress(Data *compressed_data, bool use_metadata); -Data *chunk_data_create(void *data, unsigned long long data_size); -void chunk_data_delete(void *chunk); + #endif diff --git a/src/shared/config.c b/src/shared/config.c index 4db3409..da31c4f 100644 --- a/src/shared/config.c +++ b/src/shared/config.c @@ -1,8 +1,10 @@ #include "config.h" #include "socket.h" +#include "utils.h" #include #include #include +#include Config *config_create(char *version, char *send_directory, char *receive_directory, bool save_to_disk, @@ -27,6 +29,27 @@ Config *config_create(char *version, char *send_directory, return config; } +bool is_remote_dest(const char *s) { + if (s == NULL) return false; + const char *colon = strchr(s, ':'); + if (colon == NULL) return false; + if (colon == s) return false; + for (const char *p = s; p < colon; p++) { + if (*p == '/') return false; + } + return true; +} + +void config_parse_ssh_dest(Config *config) { + if (!is_remote_dest(config->receive_root_directory)) return; + config->transport = TRANSPORT_SSH; + config->ssh_destination = str_dup(config->receive_root_directory); + char *colon = strchr(config->receive_root_directory, ':'); + char *path = str_dup(colon + 1); + free(config->receive_root_directory); + config->receive_root_directory = path; +} + void config_delete(Config *config) { free(config->version); free(config->send_directory); diff --git a/src/shared/config.h b/src/shared/config.h index 72e0524..db12612 100644 --- a/src/shared/config.h +++ b/src/shared/config.h @@ -17,7 +17,6 @@ typedef struct Config { bool use_chunk_serialization; bool use_compression; bool use_sendfile; - bool use_single_send_per_file; bool use_metadata; int compression_level; int num_connections; @@ -33,5 +32,7 @@ Config *config_create(char *version, char *send_directory, void config_delete(Config *config); void 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); #endif diff --git a/src/shared/file.c b/src/shared/file.c index c6190e6..3ee0511 100644 --- a/src/shared/file.c +++ b/src/shared/file.c @@ -80,7 +80,6 @@ void file_load_data(File *file) { exit(EXIT_FAILURE); } } - printf("%ld is file big", file->data->size); 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"); @@ -88,12 +87,6 @@ void file_load_data(File *file) { } } -void file_print(void *item) { - if (item == NULL) - return; - printf("%s\n", ((File *)item)->path); -} - static void metadata_send(int file_descriptor, FileMetadata *m) { if (m == NULL) { int zero = 0; @@ -127,7 +120,6 @@ void file_send_single_calls(File *file, int file_descriptor, bool use_metadata) send_str(file_descriptor, file->path); if (use_metadata) metadata_send(file_descriptor, file->metadata); - printf("Sending File: %ld", file->data->size); send_data(file_descriptor, file->data->data, file->data->size); } diff --git a/src/shared/file.h b/src/shared/file.h index f6d5cb3..1b03197 100644 --- a/src/shared/file.h +++ b/src/shared/file.h @@ -22,7 +22,6 @@ typedef struct { File *file_create(const char *path); void file_destroy(void *item); void file_load_data(File *file); -void file_print(void *item); void file_send_single_calls(File *file, int file_descriptor, bool use_metadata); void file_send_sendfile(File *file, int file_descriptor, bool use_metadata); size_t file_content_to_buffer(File *file); diff --git a/src/shared/queue.c b/src/shared/queue.c index 5a131e5..6267e51 100644 --- a/src/shared/queue.c +++ b/src/shared/queue.c @@ -59,7 +59,7 @@ bool queue_is_full(Queue *queue) { return queue->size == queue->capacity; } -void queue_double_capacity(Queue *queue) { +static void queue_double_capacity(Queue *queue) { if (queue == NULL) return; unsigned int new_capacity = queue->capacity * 2; diff --git a/src/shared/queue.h b/src/shared/queue.h index de58442..9666b5a 100644 --- a/src/shared/queue.h +++ b/src/shared/queue.h @@ -17,7 +17,6 @@ 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_double_capacity(Queue *queue); void queue_enqueue(Queue *queue, void *item); void queue_enqueue_multithreaded(Queue *queue, void *item, mtx_t *mutex, cnd_t *condition_not_empty, diff --git a/src/shared/socket.c b/src/shared/socket.c index 6e3b8a7..fffada4 100644 --- a/src/shared/socket.c +++ b/src/shared/socket.c @@ -167,10 +167,11 @@ Server *server_create(int port) { return server; } -void server_delete(Server *server) { - free(server); - server = NULL; -}; +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", diff --git a/src/shared/socket.h b/src/shared/socket.h index 7973edf..9a22e10 100644 --- a/src/shared/socket.h +++ b/src/shared/socket.h @@ -15,7 +15,7 @@ typedef struct Server { Server *server_create(int port); void server_listen(Server *server, void (*handler)(int file_descriptor)); -void server_delete(Server *server); +void server_delete(Server **server); typedef struct Client { struct sockaddr_in address; diff --git a/tests/test_chunk.c b/tests/test_chunk.c index 2d7ab8d..024e729 100644 --- a/tests/test_chunk.c +++ b/tests/test_chunk.c @@ -80,77 +80,9 @@ static void test_chunk_operations() { EXPECT_EQ_INT(memcmp(deserialized->items[0]->data->data, content1, len1), 0); EXPECT_EQ_INT(memcmp(deserialized->items[1]->data->data, content2, len2), 0); - chunk_data_delete(serialized); + data_destroy(serialized); chunk_destroy(deserialized); - // Test chunk_format layout (old format) - Data *formatted = chunk_format(chunk); - EXPECT_NOT_NULL(formatted); - - unsigned long long expected_size = - (sizeof(int) + strlen(path1) + sizeof(int) + sizeof(unsigned long long) + len1) + - (sizeof(int) + strlen(path2) + sizeof(int) + sizeof(unsigned long long) + len2); - EXPECT_EQ_INT((int)formatted->size, (int)expected_size); - - char *ptr = (char *)formatted->data; - - // File 1 - int p_len1; - memcpy(&p_len1, ptr, sizeof(int)); - ptr += sizeof(int); - EXPECT_EQ_INT(p_len1, (int)strlen(path1)); - - char read_path1[256]; - memcpy(read_path1, ptr, p_len1); - read_path1[p_len1] = '\0'; - ptr += p_len1; - EXPECT_EQ_STR(read_path1, path1); - - int meta_present1; - memcpy(&meta_present1, ptr, sizeof(int)); - ptr += sizeof(int); - EXPECT_EQ_INT(meta_present1, 0); - - unsigned long long d_len1; - memcpy(&d_len1, ptr, sizeof(unsigned long long)); - ptr += sizeof(unsigned long long); - EXPECT_EQ_INT((int)d_len1, (int)len1); - - char read_content1[256]; - memcpy(read_content1, ptr, d_len1); - read_content1[d_len1] = '\0'; - ptr += d_len1; - EXPECT_EQ_STR(read_content1, content1); - - // File 2 - int p_len2; - memcpy(&p_len2, ptr, sizeof(int)); - ptr += sizeof(int); - EXPECT_EQ_INT(p_len2, (int)strlen(path2)); - - char read_path2[256]; - memcpy(read_path2, ptr, p_len2); - read_path2[p_len2] = '\0'; - ptr += p_len2; - EXPECT_EQ_STR(read_path2, path2); - - int meta_present2; - memcpy(&meta_present2, ptr, sizeof(int)); - ptr += sizeof(int); - EXPECT_EQ_INT(meta_present2, 0); - - unsigned long long d_len2; - memcpy(&d_len2, ptr, sizeof(unsigned long long)); - ptr += sizeof(unsigned long long); - EXPECT_EQ_INT((int)d_len2, (int)len2); - - char read_content2[256]; - memcpy(read_content2, ptr, d_len2); - read_content2[d_len2] = '\0'; - ptr += d_len2; - EXPECT_EQ_STR(read_content2, content2); - - chunk_data_delete(formatted); chunk_destroy(chunk); unlink(path1); diff --git a/tests/test_config.c b/tests/test_config.c index 973a330..2b22e1e 100644 --- a/tests/test_config.c +++ b/tests/test_config.c @@ -26,12 +26,35 @@ static void test_config_lifecycle() { static void test_config_ssh_dest() { Config *cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("user@host:/dst"), true, false, false, false, false, 1, 4, false); - cfg->transport = TRANSPORT_SSH; - cfg->ssh_destination = str_dup("user@host:/dst"); EXPECT_NOT_NULL(cfg); + EXPECT_EQ_INT(cfg->transport, TRANSPORT_TCP); + EXPECT_NULL(cfg->ssh_destination); + EXPECT_EQ_STR(cfg->receive_root_directory, "user@host:/dst"); + + config_parse_ssh_dest(cfg); EXPECT_EQ_INT(cfg->transport, TRANSPORT_SSH); EXPECT_EQ_STR(cfg->ssh_destination, "user@host:/dst"); - EXPECT_EQ_STR(cfg->receive_root_directory, "user@host:/dst"); + EXPECT_EQ_STR(cfg->receive_root_directory, "/dst"); + config_delete(cfg); +} + +static void test_config_ssh_dest_local_path() { + Config *cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("/local/path"), + true, false, false, false, false, 1, 4, false); + config_parse_ssh_dest(cfg); + EXPECT_EQ_INT(cfg->transport, TRANSPORT_TCP); + EXPECT_NULL(cfg->ssh_destination); + EXPECT_EQ_STR(cfg->receive_root_directory, "/local/path"); + config_delete(cfg); +} + +static void test_config_ssh_dest_no_user() { + Config *cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("host:/remote"), + true, false, false, false, false, 1, 4, false); + config_parse_ssh_dest(cfg); + EXPECT_EQ_INT(cfg->transport, TRANSPORT_SSH); + EXPECT_EQ_STR(cfg->ssh_destination, "host:/remote"); + EXPECT_EQ_STR(cfg->receive_root_directory, "/remote"); config_delete(cfg); } @@ -70,6 +93,8 @@ static void test_pipeline_receiver_lifecycle() { void test_config() { test_config_lifecycle(); test_config_ssh_dest(); + test_config_ssh_dest_local_path(); + test_config_ssh_dest_no_user(); test_pipeline_sender_lifecycle(); test_pipeline_receiver_lifecycle(); } -- 2.52.0 From 02619cca5ceeb7b6eee52af5f6b9112b9d510ec0 Mon Sep 17 00:00:00 2001 From: TapTap Date: Wed, 15 Jul 2026 19:36:26 +0200 Subject: [PATCH 2/3] 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" -- 2.52.0 From d36768792a0d4a7b322a5ebc73da17b4476d042d Mon Sep 17 00:00:00 2001 From: TapTap Date: Thu, 16 Jul 2026 11:54:20 +0200 Subject: [PATCH 3/3] fixup: address PR review comments - Merge io.h/c back into protocol.h/c - Change send_data to take Data* argument - Remove redundant file_load_data from send_chunk - Move compression into file_send_single_calls - Move file_receive from multiprocessing.c to file.c --- src/client/client_send.c | 16 +++-------- src/server/server.c | 1 - src/shared/file.c | 29 +++++++++++++++++--- src/shared/file.h | 4 ++- src/shared/io.c | 49 ---------------------------------- src/shared/io.h | 10 ------- src/shared/metadata.c | 2 +- src/shared/multiprocessing.c | 17 ------------ src/shared/multiprocessing.h | 1 - src/shared/protocol.c | 51 +++++++++++++++++++++++++++++++++--- src/shared/protocol.h | 7 +++-- 11 files changed, 87 insertions(+), 100 deletions(-) delete mode 100644 src/shared/io.c delete mode 100644 src/shared/io.h diff --git a/src/client/client_send.c b/src/client/client_send.c index 97cf409..c89f9b0 100644 --- a/src/client/client_send.c +++ b/src/client/client_send.c @@ -1,6 +1,5 @@ #include "client_send.h" #include "chunk.h" -#include "compression.h" #include "config.h" #include "data.h" #include "file.h" @@ -24,11 +23,9 @@ int send_chunk(Client *client, Chunk *chunk, Config *config) { if (config->use_compression) { data = chunk_compress(chunk, config->compression_level, config->use_metadata); } else { - for (int i = 0; i < chunk->element_count; i++) - file_load_data(chunk->items[i]); data = chunk_serialize(chunk, config->use_metadata); } - send_data(client->file_descriptor, data->data, data->size); + send_data(client->file_descriptor, data); data_destroy(data); } else if (config->use_sendfile && !config->use_compression) { for (int i = 0; i < chunk->element_count; i++) { @@ -38,14 +35,9 @@ int send_chunk(Client *client, Chunk *chunk, Config *config) { } else { for (int i = 0; i < chunk->element_count; i++) { send_status(client->file_descriptor, STATUS_NEXT); - File *file = chunk->items[i]; - if (config->use_compression) { - Data *compressed_data = - data_compress(file->data, config->compression_level); - data_destroy(file->data); - file->data = compressed_data; - } - file_send_single_calls(file, client->file_descriptor, config->use_metadata); + file_send_single_calls(chunk->items[i], client->file_descriptor, + config->use_metadata, + config->use_compression ? config->compression_level : 0); } } return 0; diff --git a/src/server/server.c b/src/server/server.c index b6a7af7..3af90ae 100644 --- a/src/server/server.c +++ b/src/server/server.c @@ -3,7 +3,6 @@ #include "config.h" #include "data.h" #include "file.h" -#include "io.h" #include "log.h" #include "metadata.h" #include "multiprocessing.h" diff --git a/src/shared/file.c b/src/shared/file.c index 3273b46..604ce98 100644 --- a/src/shared/file.c +++ b/src/shared/file.c @@ -9,9 +9,10 @@ #include #include +#include "compression.h" +#include "config.h" #include "data.h" #include "file.h" -#include "io.h" #include "log.h" #include "metadata.h" #include "protocol.h" @@ -90,11 +91,16 @@ void file_load_data(File *file) { } } -void file_send_single_calls(File *file, int file_descriptor, bool use_metadata) { +void 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); + data_destroy(file->data); + 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->data, file->data->size); + send_data(file_descriptor, file->data); } void to_disk(const char *path, const void *data, unsigned long long data_size) { @@ -139,6 +145,23 @@ void file_send_sendfile(File *file, int file_descriptor, bool use_metadata) { close(fd); } +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; +} + size_t file_content_to_buffer(File *file) { FILE *file_pointer = fopen(file->path, "rb"); if (file_pointer == NULL) { diff --git a/src/shared/file.h b/src/shared/file.h index 535e693..cf2bd52 100644 --- a/src/shared/file.h +++ b/src/shared/file.h @@ -1,6 +1,7 @@ #ifndef FILE_H #define FILE_H +#include "config.h" #include "data.h" #include #include @@ -22,7 +23,8 @@ typedef struct { File *file_create(const char *path); void file_destroy(void *item); void file_load_data(File *file); -void file_send_single_calls(File *file, int file_descriptor, bool use_metadata); +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); size_t file_content_to_buffer(File *file); FileMetadata *file_metadata_create(struct stat *stats); diff --git a/src/shared/io.c b/src/shared/io.c deleted file mode 100644 index d2defa0..0000000 --- a/src/shared/io.c +++ /dev/null @@ -1,49 +0,0 @@ -#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 deleted file mode 100644 index 9da1fe1..0000000 --- a/src/shared/io.h +++ /dev/null @@ -1,10 +0,0 @@ -#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 index 6a0c283..02c453b 100644 --- a/src/shared/metadata.c +++ b/src/shared/metadata.c @@ -1,6 +1,6 @@ #include "metadata.h" #include "file.h" -#include "io.h" +#include "protocol.h" #include #include #include diff --git a/src/shared/multiprocessing.c b/src/shared/multiprocessing.c index 3be78a7..9d57c8f 100644 --- a/src/shared/multiprocessing.c +++ b/src/shared/multiprocessing.c @@ -74,23 +74,6 @@ void pipeline_context_receiver_destroy(PipelineContextReceiver *context) { 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); diff --git a/src/shared/multiprocessing.h b/src/shared/multiprocessing.h index 22fad81..81ab589 100644 --- a/src/shared/multiprocessing.h +++ b/src/shared/multiprocessing.h @@ -39,7 +39,6 @@ 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 index 2e86dba..734387a 100644 --- a/src/shared/protocol.c +++ b/src/shared/protocol.c @@ -1,9 +1,53 @@ #include "protocol.h" -#include "io.h" #include "log.h" #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; +} + +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); +} static const char *status_to_string(Status status) { switch (status) { @@ -39,9 +83,10 @@ char *receive_str(int file_descriptor) { return data; } -void send_data(int file_descriptor, void *data, unsigned long long data_size) { +void 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_size); + send_n_data(file_descriptor, data->data, data_size); log_message(LOG_LEVEL_DEBUG, "Send %lld data", data_size); } diff --git a/src/shared/protocol.h b/src/shared/protocol.h index c13e3ac..e2961e6 100644 --- a/src/shared/protocol.h +++ b/src/shared/protocol.h @@ -2,15 +2,18 @@ #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 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); +void 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); -- 2.52.0