Merge pull request 'feat: implement chunk serialization protocol with -s flag' (#5) from chunk-serialization into main

Reviewed-on: #5

src/client/client.c

@@ -17,9 +17,18 @@
 #include <dirent.h>
 
 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);

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;
   }

src/shared/chunk.c

@@ -4,7 +4,6 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
-#include <zstd.h>
 
 #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;
 }

src/shared/chunk.h

@@ -6,8 +6,6 @@
 #include <sys/stat.h>
 
 #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

src/shared/file.c

@@ -100,4 +100,4 @@ void file_receive_destroy(void *file_receive) {
   free(file);
 }
 
-FileReceive *file_receive_from_buffer(void *buffer) {}
+

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

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";
   }

src/shared/socket.h

@@ -5,7 +5,7 @@
 #include <netinet/in.h>
 
 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;

test.py

@@ -36,18 +36,15 @@ CLIENT_CMD_PREFIX = [
 TEST_CASES = [
     {"name": "Standard (Single-threaded)", "flags": []},
     {"name": "Multithreading (-m)", "flags": ["-m"]},
-    # {"name": "Compression (-c -5)", "flags": ["-c -5"]},
     {"name": "Compression (-c 0)", "flags": ["-c 0"]},
-    # {"name": "Compression (-c 10)", "flags": ["-c 10"]},
-    # {"name": "Compression (-c 20)", "flags": ["-c 20"]},
-    # {"name": "Chunk Serialization (-s)", "flags": ["-s"]},
+    {"name": "Chunk Serialization (-s)", "flags": ["-s"]},
+    {"name": "Compression + Chunk Serialization (-c -s)", "flags": ["-c", "-s"]},
     {"name": "Multithreading + Compression (-m -c)", "flags": ["-m", "-c"]},
-    # {"name": "Multithreading + Chunk Serialization (-m -s)", "flags": ["-m", "-s"]},
-    # {"name": "Compression + Chunk Serialization (-c -s)", "flags": ["-c", "-s"]},
-    # {
-    #     "name": "Multithreading + Compression + Chunk Serialization (-m -c -s)",
-    #     "flags": ["-m", "-c", "-s"],
-    # },
+    {"name": "Multithreading + Chunk Serialization (-m -s)", "flags": ["-m", "-s"]},
+    {
+        "name": "Multithreading + Compression + Chunk Serialization (-m -c -s)",
+        "flags": ["-m", "-c", "-s"],
+    },
 ]