From c4aeed9f4dbee64a834876952803f166dea9d16a Mon Sep 17 00:00:00 2001 From: TapTap Date: Sat, 4 Jul 2026 21:38:46 +0200 Subject: [PATCH] feat: implement chunk serialization protocol with -s flag - Extract chunk_serialize/chunk_deserialize from chunk_compress/chunk_decompress - Add STATUS_CHUNK to wire protocol for chunk-mode transfers - Client: -s sends chunk via STATUS_CHUNK (serialized or compressed+serialized) - Server: handle STATUS_CHUNK in both single-threaded and multithreaded paths - Fix memory leak in receive_file_receive (free->data_destroy) - Remove dead declarations: file_receive_from_buffer, file_receive_decompress, file_compress, chunk_data_to_disk, unused #defines --- src/client/client.c | 13 ++- src/server/server.c | 75 ++++++++++++--- src/shared/chunk.c | 216 ++++++++++++++++++++++---------------------- src/shared/chunk.h | 5 +- src/shared/file.c | 2 +- src/shared/file.h | 3 - src/shared/socket.c | 2 + src/shared/socket.h | 2 +- 8 files changed, 188 insertions(+), 130 deletions(-) diff --git a/src/client/client.c b/src/client/client.c index 79d588e..2cfc027 100644 --- a/src/client/client.c +++ b/src/client/client.c @@ -17,9 +17,18 @@ #include int send_chunk(Client *client, Chunk *chunk, Config *config) { - if (config->use_compression && config->use_chunk_serialization) { - Data *data = chunk_compress(chunk, config->compression_level); + if (config->use_chunk_serialization) { + send_status(client->file_descriptor, STATUS_CHUNK); + Data *data; + if (config->use_compression) { + data = chunk_compress(chunk, config->compression_level); + } else { + for (int i = 0; i < chunk->element_count; i++) + file_load_data(chunk->items[i]); + data = chunk_serialize(chunk); + } send_data(client->file_descriptor, data->data, data->size); + data_destroy(data); } else { for (int i = 0; i < chunk->element_count; i++) { send_status(client->file_descriptor, STATUS_NEXT); diff --git a/src/server/server.c b/src/server/server.c index 3891c51..3debf97 100644 --- a/src/server/server.c +++ b/src/server/server.c @@ -1,3 +1,4 @@ +#include "chunk.h" #include "config.h" #include "data.h" #include "file.h" @@ -16,13 +17,36 @@ FileReceive *receive_file_receive(Config *config, int file_descriptor) { Data *file_data = receive_data(file_descriptor); if (config->use_compression) { Data *file_data_uncompressed = data_decompress(file_data); - free(file_data); + data_destroy(file_data); file_data = file_data_uncompressed; } FileReceive *file = file_receive_create(path, file_data); return file; } +static void receive_chunk_enqueue(int file_descriptor, Config *config, + PipelineContextReceiver *context) { + Data *chunk_data = receive_data(file_descriptor); + Data *data_to_process = chunk_data; + if (config->use_compression) { + data_to_process = data_decompress(chunk_data); + data_destroy(chunk_data); + } + Chunk *chunk = chunk_deserialize(data_to_process); + data_destroy(data_to_process); + + for (int i = 0; i < chunk->element_count; i++) { + FileReceive *file = file_receive_create(chunk->items[i]->path, + chunk->items[i]->data); + chunk->items[i]->path = NULL; + chunk->items[i]->data = 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; @@ -31,11 +55,17 @@ int receive_thread(void *pipeline_context) { Config *config = context->config; mtx_unlock(&context->mutex); - while (receive_status(file_descriptor) == STATUS_NEXT) { - FileReceive *file = receive_file_receive(config, file_descriptor); - queue_enqueue_multithreaded(context->queue, file, &context->mutex, - &context->condition_not_empty, - &context->condition_not_full); + Status status = receive_status(file_descriptor); + while (status == STATUS_NEXT || status == STATUS_CHUNK) { + if (status == STATUS_CHUNK) { + receive_chunk_enqueue(file_descriptor, config, context); + } else { + FileReceive *file = receive_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; @@ -68,17 +98,34 @@ int write_thread(void *pipeline_context) { int receive_files(Config *config, int file_descriptor) { Status status = receive_status(file_descriptor); - while (status == STATUS_NEXT) { - FileReceive *file = receive_file_receive(config, file_descriptor); - if (config->save_to_disk) - to_disk(path_cat(config->receive_root_directory, file->path), - file->data->data, file->data->size); - file_receive_destroy(file); - // send_status(file_descriptor, STATUS_OK); + while (status == STATUS_NEXT || status == STATUS_CHUNK) { + if (status == STATUS_CHUNK) { + Data *chunk_data = receive_data(file_descriptor); + Data *data_to_process = chunk_data; + if (config->use_compression) { + data_to_process = data_decompress(chunk_data); + data_destroy(chunk_data); + } + Chunk *chunk = chunk_deserialize(data_to_process); + data_destroy(data_to_process); + + for (int i = 0; i < chunk->element_count; i++) { + if (config->save_to_disk) + to_disk(path_cat(config->receive_root_directory, chunk->items[i]->path), + chunk->items[i]->data->data, chunk->items[i]->data->size); + } + chunk_destroy(chunk); + } else { + FileReceive *file = receive_file_receive(config, file_descriptor); + if (config->save_to_disk) + to_disk(path_cat(config->receive_root_directory, file->path), + file->data->data, file->data->size); + file_receive_destroy(file); + } status = receive_status(file_descriptor); } if (status != STATUS_FINISHED) { - log_message(LOG_LEVEL_ERROR, "Did not receive FINISHED or NEXT Status"); + log_message(LOG_LEVEL_ERROR, "Did not receive FINISHED Status"); send_status(file_descriptor, STATUS_ERROR); return -1; } diff --git a/src/shared/chunk.c b/src/shared/chunk.c index 87d5f1e..27433a8 100644 --- a/src/shared/chunk.c +++ b/src/shared/chunk.c @@ -4,7 +4,6 @@ #include #include #include -#include #include "chunk.h" #include "array_list.h" @@ -99,40 +98,129 @@ Data *chunk_format(Chunk *chunk) { return chunk_data_create(data, buffer_size); } -Data *chunk_compress(Chunk *chunk, int compression_level) { - log_message(LOG_LEVEL_DEBUG, "Starting to gather data for chunk compression"); +Data *chunk_serialize(Chunk *chunk) { unsigned long long data_size = 0; for (int i = 0; i < chunk->element_count; i++) { - data_size += sizeof(unsigned long long); + data_size += sizeof(size_t); data_size += strlen(chunk->items[i]->path); - data_size += sizeof(unsigned long long); + data_size += sizeof(size_t); data_size += chunk->items[i]->stats.st_size; } Data *data = data_create_empty(data_size); if (data == NULL) { log_message(LOG_LEVEL_ERROR, - "Could not allocate memory for chunk compression"); + "Could not allocate memory for chunk serialization"); exit(EXIT_FAILURE); } char *data_pointer = data->data; for (int i = 0; i < chunk->element_count; i++) { - // path length size_t path_len = strlen(chunk->items[i]->path); memcpy(data_pointer, &path_len, sizeof(size_t)); data_pointer += sizeof(size_t); memcpy(data_pointer, chunk->items[i]->path, path_len); data_pointer += path_len; - // file data - unsigned long long data_size = chunk->items[i]->stats.st_size; - memcpy(data_pointer, &data_size, sizeof(size_t)); + + size_t file_data_size = chunk->items[i]->stats.st_size; + memcpy(data_pointer, &file_data_size, sizeof(size_t)); data_pointer += sizeof(size_t); - memcpy(data_pointer, chunk->items[i]->data->data, data_size); - data_pointer += data_size; + memcpy(data_pointer, chunk->items[i]->data->data, file_data_size); + data_pointer += file_data_size; + } + return data; +} + +Chunk *chunk_deserialize(Data *data) { + ArrayList *files = array_list_create(file_destroy); + char *data_pointer = data->data; + size_t remaining_size = data->size; + + while (remaining_size > 0) { + if (remaining_size < sizeof(size_t)) { + log_message(LOG_LEVEL_ERROR, "Invalid chunk format: not enough data for path length"); + array_list_delete(files); + return NULL; + } + + size_t path_len = *(size_t *)data_pointer; + data_pointer += sizeof(size_t); + remaining_size -= sizeof(size_t); + + if (remaining_size < path_len) { + log_message(LOG_LEVEL_ERROR, "Invalid chunk format: not enough data for path"); + array_list_delete(files); + return NULL; + } + + char *path = malloc(path_len + 1); + if (path == NULL) { + perror("Could not allocate memory for file path"); + array_list_delete(files); + return NULL; + } + memcpy(path, data_pointer, path_len); + path[path_len] = '\0'; + data_pointer += path_len; + remaining_size -= path_len; + + if (remaining_size < sizeof(size_t)) { + log_message(LOG_LEVEL_ERROR, "Invalid chunk format: not enough data for data size"); + free(path); + array_list_delete(files); + return NULL; + } + + size_t file_data_size = *(size_t *)data_pointer; + data_pointer += sizeof(size_t); + remaining_size -= sizeof(size_t); + + if (remaining_size < file_data_size) { + log_message(LOG_LEVEL_ERROR, "Invalid chunk format: not enough data for file content"); + free(path); + array_list_delete(files); + return NULL; + } + + struct stat st = {0}; + st.st_size = file_data_size; + File *file = file_create(path, &st); + if (file == NULL) { + free(path); + array_list_delete(files); + return NULL; + } + + void *file_data = malloc(file_data_size); + if (file_data == NULL) { + perror("Could not allocate memory for file data"); + free(path); + array_list_delete(files); + return NULL; + } + memcpy(file_data, data_pointer, file_data_size); + file->data = data_create(file_data, file_data_size); + data_pointer += file_data_size; + remaining_size -= file_data_size; + + array_list_add(files, file); + free(path); } - log_message(LOG_LEVEL_DEBUG, "Chunk succesfully compressed"); - Data *compressed = data_compress(data, compression_level); - data_destroy(data); + File **file_array = (File **)array_list_to_array(files); + Chunk *chunk = chunk_create(file_array, files->size); + + free(file_array); + files->item_destroyer = NULL; + array_list_delete(files); + + return chunk; +} + +Data *chunk_compress(Chunk *chunk, int compression_level) { + log_message(LOG_LEVEL_DEBUG, "Starting to compress chunk"); + Data *serialized = chunk_serialize(chunk); + Data *compressed = data_compress(serialized, compression_level); + data_destroy(serialized); + log_message(LOG_LEVEL_DEBUG, "Chunk successfully compressed"); return compressed; } @@ -143,99 +231,15 @@ Chunk *chunk_decompress(Data *compressed_data) { log_message(LOG_LEVEL_ERROR, "Failed to decompress chunk data"); return NULL; } - - ArrayList *files = array_list_create(file_destroy); - char *data_pointer = uncompressed_data->data; - size_t remaining_size = uncompressed_data->size; - - while (remaining_size > 0) { - if (remaining_size < sizeof(size_t)) { - log_message(LOG_LEVEL_ERROR, "Invalid chunk format: not enough data for path length"); - array_list_delete(files); - data_destroy(uncompressed_data); - return NULL; - } - - size_t path_len = *(size_t *)data_pointer; - data_pointer += sizeof(size_t); - remaining_size -= sizeof(size_t); - - if (remaining_size < path_len) { - log_message(LOG_LEVEL_ERROR, "Invalid chunk format: not enough data for path"); - array_list_delete(files); - data_destroy(uncompressed_data); - return NULL; - } - - char *path = malloc(path_len + 1); - if (path == NULL) { - perror("Could not allocate memory for file path"); - array_list_delete(files); - data_destroy(uncompressed_data); - return NULL; - } - memcpy(path, data_pointer, path_len); - path[path_len] = '\0'; - data_pointer += path_len; - remaining_size -= path_len; - - if (remaining_size < sizeof(size_t)) { - log_message(LOG_LEVEL_ERROR, "Invalid chunk format: not enough data for data size"); - free(path); - array_list_delete(files); - data_destroy(uncompressed_data); - return NULL; - } - - size_t data_size = *(size_t *)data_pointer; - data_pointer += sizeof(size_t); - remaining_size -= sizeof(size_t); - - if (remaining_size < data_size) { - log_message(LOG_LEVEL_ERROR, "Invalid chunk format: not enough data for file content"); - free(path); - array_list_delete(files); - data_destroy(uncompressed_data); - return NULL; - } - - struct stat st = {0}; - st.st_size = data_size; - File *file = file_create(path, &st); - if (file == NULL) { - free(path); - array_list_delete(files); - data_destroy(uncompressed_data); - return NULL; - } - - void *file_data = malloc(data_size); - if (file_data == NULL) { - perror("Could not allocate memory for file data"); - free(path); - array_list_delete(files); - data_destroy(uncompressed_data); - return NULL; - } - memcpy(file_data, data_pointer, data_size); - file->data = data_create(file_data, data_size); - data_pointer += data_size; - remaining_size -= data_size; - - array_list_add(files, file); - free(path); + + Chunk *chunk = chunk_deserialize(uncompressed_data); + if (chunk == NULL) { + log_message(LOG_LEVEL_ERROR, "Failed to deserialize chunk data"); + data_destroy(uncompressed_data); + return NULL; } - - // Create the chunk from the files - File **file_array = (File **)array_list_to_array(files); - Chunk *chunk = chunk_create(file_array, files->size); - - // Clean up - files are now owned by the chunk - free(file_array); - files->item_destroyer = NULL; - array_list_delete(files); + data_destroy(uncompressed_data); - log_message(LOG_LEVEL_DEBUG, "Chunk successfully decompressed"); return chunk; } diff --git a/src/shared/chunk.h b/src/shared/chunk.h index 59feeed..41b62b8 100644 --- a/src/shared/chunk.h +++ b/src/shared/chunk.h @@ -6,8 +6,6 @@ #include #define DESIRED_CHUNK_SIZE 10 * 1024 * 1024 -#define FILE_PATH_SEPERATOR "#&&SEPP&&#" -#define FILE_PATH_DATA_SEPERATOR "#&&SEPD&&#" typedef struct { File **items; @@ -18,10 +16,11 @@ 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); +Chunk *chunk_deserialize(Data *data); Data *chunk_compress(Chunk *chunk, int compression_level); Chunk *chunk_decompress(Data *compressed_data); Data *chunk_data_create(void *data, unsigned long long data_size); void chunk_data_delete(void *chunk); -void chunk_data_to_disk(Data *chunk, char *root_directory); #endif diff --git a/src/shared/file.c b/src/shared/file.c index e51f724..9343349 100644 --- a/src/shared/file.c +++ b/src/shared/file.c @@ -100,4 +100,4 @@ void file_receive_destroy(void *file_receive) { free(file); } -FileReceive *file_receive_from_buffer(void *buffer) {} + diff --git a/src/shared/file.h b/src/shared/file.h index d73bcb7..2bf85ea 100644 --- a/src/shared/file.h +++ b/src/shared/file.h @@ -21,11 +21,8 @@ void file_load_data(File *file); void file_print(void *item); void file_send_single_calls(File *file, int file_descriptor); size_t file_content_to_buffer(File *file); -Data *file_compress(File *file); FileReceive *file_receive_create(char *path, Data *data); void file_receive_destroy(void *file_receive); -FileReceive *file_receive_from_buffer(void *buffer); -FileReceive *file_receive_decompress(void *FileReceive); #endif diff --git a/src/shared/socket.c b/src/shared/socket.c index fd5d91e..0d6f31e 100644 --- a/src/shared/socket.c +++ b/src/shared/socket.c @@ -195,6 +195,8 @@ const char *status_to_string(Status status) { return "FINISHED"; case STATUS_NEXT: return "NEXT"; + case STATUS_CHUNK: + return "CHUNK"; default: return "UNKNOWN"; } diff --git a/src/shared/socket.h b/src/shared/socket.h index 5f0cc4f..db694f1 100644 --- a/src/shared/socket.h +++ b/src/shared/socket.h @@ -5,7 +5,7 @@ #include typedef int Status; -enum NET_STATUS { STATUS_OK, STATUS_ERROR, STATUS_FINISHED, STATUS_NEXT }; +enum NET_STATUS { STATUS_OK, STATUS_ERROR, STATUS_FINISHED, STATUS_NEXT, STATUS_CHUNK }; typedef struct Server { struct sockaddr_in address;