src/client/client_cli.c

@@ -46,7 +46,7 @@ int main(int argc, char *argv[]) {
   }
 
   Config *config = config_create(str_dup("1.0.0"), NULL, NULL,
-                                 save_to_disk, false, false, false, false, 5, 20, false);
+                                 save_to_disk, false, false, false, false, 5, false);
 
   int positional_args[2];
   int positional_count = 0;

src/client/client_send.c

@@ -171,10 +171,11 @@ int send_files_multithreaded(Config *config) {
     return 1;
   }
 
+  int sender_result;
   thrd_join(scanner, NULL);
   thrd_join(loader, NULL);
-  thrd_join(sender, NULL);
+  thrd_join(sender, &sender_result);
 
   pipeline_context_sender_destroy(context);
-  return 0;
+  return sender_result == thrd_success ? 0 : -1;
 }

src/client/scanner.c

@@ -90,7 +90,7 @@ Chunk *directory_scanner_next(DirectoryScanner *scanner) {
       continue;
     }
 
-    if (!S_ISREG(stats.st_mode)) {
+    if (S_ISDIR(stats.st_mode)) {
       queue_enqueue(scanner->directories, (void *)cur_path);
     } else {
       File *file = file_create(cur_path);
@@ -99,8 +99,10 @@ Chunk *directory_scanner_next(DirectoryScanner *scanner) {
         file->metadata = file_metadata_create(&stats);
       array_list_add(chunk_data, file);
       chunk_data_size += file->data->size;
-      if (chunk_data_size > DESIRED_CHUNK_SIZE)
+      if (chunk_data_size > DESIRED_CHUNK_SIZE) {
+        free(cur_path);
         return chunk_data_to_chunk(chunk_data);
+      }
       free(cur_path);
     }
   }

src/server/server.c

@@ -11,6 +11,7 @@
 #include "transport_tcp.h"
 #include "unistd.h"
 #include "utils.h"
+#include <signal.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -27,6 +28,11 @@ int receive_files(Config *config, int file_descriptor) {
       }
       Chunk *chunk = chunk_deserialize(data_to_process, config->use_metadata);
       data_destroy(data_to_process);
+      if (chunk == NULL) {
+        log_message(LOG_LEVEL_ERROR, "Failed to deserialize chunk, skipping");
+        send_status(file_descriptor, STATUS_ERROR);
+        return -1;
+      }
 
       for (int i = 0; i < chunk->element_count; i++) {
         if (config->save_to_disk) {
@@ -79,6 +85,7 @@ void handler(int file_descriptor) {
 }
 
 int main(int argc, char *argv[]) {
+  signal(SIGPIPE, SIG_IGN);
   for (int i = 1; i < argc; i++) {
     if (strcmp(argv[i], "--stdio") == 0) {
       io_set_fds(STDIN_FILENO, STDOUT_FILENO);

src/shared/array_list.c

@@ -35,14 +35,6 @@ void array_list_delete(ArrayList *array_list) {
   free(array_list);
 }
 
-void array_list_clear(ArrayList *array_list) {
-  if (array_list == NULL)
-    return;
-  for (int i = 0; i < array_list->size; i++)
-    array_list->items[i] = NULL;
-  array_list->size = 0;
-}
-
 void array_list_extend(ArrayList *array_list) {
   if (array_list == NULL)
     return;

src/shared/array_list.h

@@ -12,7 +12,6 @@ typedef struct ArrayList {
 
 ArrayList *array_list_create(void (*item_destroyer)(void *item));
 void array_list_delete(ArrayList *array_list);
-void array_list_clear(ArrayList *array_list);
 void array_list_extend(ArrayList *array_list);
 void array_list_add(ArrayList *array_list, void *item);
 void **array_list_to_array(ArrayList *array_list);

src/shared/chunk.c

@@ -178,24 +178,5 @@ Data *chunk_compress(Chunk *chunk, int compression_level, bool use_metadata) {
   return compressed;
 }
 
-Chunk *chunk_decompress(Data *compressed_data, bool use_metadata) {
-  log_message(LOG_LEVEL_DEBUG, "Starting to decompress chunk");
-  Data *uncompressed_data = data_decompress(compressed_data);
-  if (uncompressed_data == NULL) {
-    log_message(LOG_LEVEL_ERROR, "Failed to decompress chunk data");
-    return NULL;
-  }
-
-  Chunk *chunk = chunk_deserialize(uncompressed_data, use_metadata);
-  if (chunk == NULL) {
-    log_message(LOG_LEVEL_ERROR, "Failed to deserialize chunk data");
-    data_destroy(uncompressed_data);
-    return NULL;
-  }
-
-  data_destroy(uncompressed_data);
-  log_message(LOG_LEVEL_DEBUG, "Chunk successfully decompressed");
-  return chunk;
-}
 
 

src/shared/chunk.h

@@ -18,7 +18,5 @@ void chunk_destroy(void *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);
-
 
 #endif

src/shared/config.c

@@ -10,7 +10,7 @@ Config *config_create(char *version, char *send_directory,
                       char *receive_directory, bool save_to_disk,
                       bool use_multithreading, bool use_chunk_serialization,
                        bool use_compression, bool use_metadata,
-                       int compression_level, int num_connections, bool use_sendfile) {
+                       int compression_level, bool use_sendfile) {
 
   Config *config = malloc(sizeof(Config));
   config->version = version;
@@ -22,7 +22,6 @@ Config *config_create(char *version, char *send_directory,
   config->use_compression = use_compression;
   config->use_metadata = use_metadata;
   config->compression_level = compression_level;
-  config->num_connections = num_connections;
   config->use_sendfile = use_sendfile;
   config->transport = TRANSPORT_TCP;
   config->ssh_destination = NULL;
@@ -68,7 +67,6 @@ void config_send(int file_descriptor, Config *config) {
   send_int(file_descriptor, config->use_compression);
   send_int(file_descriptor, config->use_metadata);
   send_int(file_descriptor, config->compression_level);
-  send_int(file_descriptor, config->num_connections);
   send_int(file_descriptor, config->use_sendfile);
   if (receive_status(file_descriptor) != STATUS_OK) {
     perror("Error transmitting config!");
@@ -87,7 +85,6 @@ Config *config_receive(int file_descriptor) {
   config->use_compression = receive_int(file_descriptor);
   config->use_metadata = receive_int(file_descriptor);
   config->compression_level = receive_int(file_descriptor);
-  config->num_connections = receive_int(file_descriptor);
   config->use_sendfile = receive_int(file_descriptor);
   config->transport = TRANSPORT_TCP;
   config->ssh_destination = NULL;

src/shared/config.h

@@ -19,7 +19,6 @@ typedef struct Config {
   bool use_sendfile;
   bool use_metadata;
   int compression_level;
-  int num_connections;
   TransportType transport;
   char *ssh_destination;
 } Config;
@@ -28,7 +27,7 @@ Config *config_create(char *version, char *send_directory,
                       char *receive_directory, bool save_to_disk,
                       bool use_multithreading, bool use_chunk_serialization,
                        bool use_compression, bool use_metadata,
-                       int compression_level, int num_connections, bool use_sendfile);
+                       int compression_level, bool use_sendfile);
 void config_delete(Config *config);
 void config_send(int file_descriptor, Config *config);
 Config *config_receive(int file_descriptor);

src/shared/file.c

@@ -113,7 +113,11 @@ void to_disk(const char *path, const void *data, unsigned long long data_size) {
     perror("Could not open File");
     exit(EXIT_FAILURE);
   }
-  fwrite(data, 1, data_size, file_pointer);
+  if (fwrite(data, 1, data_size, file_pointer) != data_size) {
+    perror("Failed to write all data to disk");
+    fclose(file_pointer);
+    exit(EXIT_FAILURE);
+  }
   fclose(file_pointer);
   free(dir_to_free);
 }
@@ -171,7 +175,8 @@ size_t file_content_to_buffer(File *file) {
   size_t bytes_read =
       fread(file->data->data, 1, file->data->size, file_pointer);
   if (bytes_read != (size_t)file->data->size) {
-    perror("Read to many or to less bytes from File!");
+    fclose(file_pointer);
+    perror("Read unexpected number of bytes from File!");
     return 0;
   }
   fclose(file_pointer);

src/shared/metadata.c

@@ -8,8 +8,6 @@
 #include <time.h>
 #include <unistd.h>
 
-#define FILE_METADATA_WIRE_SIZE (sizeof(mode_t) + sizeof(uid_t) + sizeof(gid_t) + sizeof(time_t) + sizeof(long))
-
 void metadata_to_buf(char **buf, FileMetadata *m) {
   int present = (m != NULL) ? 1 : 0;
   memcpy(*buf, &present, sizeof(int));

src/shared/multiprocessing.c

@@ -84,6 +84,10 @@ static void receive_chunk_enqueue(int file_descriptor,
   }
   Chunk *chunk = chunk_deserialize(data_to_process, context->config->use_metadata);
   data_destroy(data_to_process);
+  if (chunk == NULL) {
+    log_message(LOG_LEVEL_ERROR, "Failed to deserialize chunk, skipping");
+    return;
+  }
 
   for (int i = 0; i < chunk->element_count; i++) {
     File *file = chunk->items[i];

src/shared/protocol.c

@@ -91,12 +91,12 @@ void send_data(int file_descriptor, Data *data) {
 }
 
 Data *receive_data(int file_descriptor) {
-  size_t size = 0;
+  unsigned long long size = 0;
   receive_n_data(file_descriptor, &size, sizeof(unsigned long long));
-  void *data = malloc(size);
-  receive_n_data(file_descriptor, data, size);
+  void *data = malloc((size_t)size);
+  receive_n_data(file_descriptor, data, (size_t)size);
   log_message(LOG_LEVEL_DEBUG, "Received %lld data", size);
-  return data_create(data, size);
+  return data_create(data, (size_t)size);
 }
 
 void send_int(int file_descriptor, int data) {

tests/test_array_list.c

@@ -49,16 +49,4 @@ void test_array_list() {
   list->item_destroyer = test_destroyer;
   array_list_delete(list);
   EXPECT_EQ_INT(destroyer_calls, 106);
-
-  // Test clear with NULL destroyer
-  list = array_list_create(NULL);
-  int a = 1, b = 2;
-  array_list_add(list, &a);
-  array_list_add(list, &b);
-  EXPECT_EQ_INT(list->size, 2);
-  array_list_clear(list);
-  EXPECT_EQ_INT(list->size, 0);
-  EXPECT_NULL(list->items[0]);
-  EXPECT_NULL(list->items[1]);
-  array_list_delete(list);
 }

tests/test_compression.c

@@ -86,7 +86,10 @@ static void test_chunk_compress_decompress_roundtrip() {
   Data *compressed = chunk_compress(chunk, 3, false);
   EXPECT_NOT_NULL(compressed);
 
-  Chunk *decompressed_chunk = chunk_decompress(compressed, false);
+  Data *decompressed_data = data_decompress(compressed);
+  EXPECT_NOT_NULL(decompressed_data);
+
+  Chunk *decompressed_chunk = chunk_deserialize(decompressed_data, false);
   EXPECT_NOT_NULL(decompressed_chunk);
   EXPECT_EQ_INT(decompressed_chunk->element_count, 2);
 
@@ -100,6 +103,7 @@ static void test_chunk_compress_decompress_roundtrip() {
 
   chunk_destroy(chunk);
   data_destroy(compressed);
+  data_destroy(decompressed_data);
   chunk_destroy(decompressed_chunk);
 
   unlink(path1);

tests/test_config.c

@@ -8,7 +8,7 @@
 
 static void test_config_lifecycle() {
   Config *cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("/dst"),
-                              true, true, false, false, false, 1, 4, false);
+                              true, true, false, false, false, 1, false);
   EXPECT_NOT_NULL(cfg);
   EXPECT_EQ_STR(cfg->version, "1.0");
   EXPECT_EQ_STR(cfg->send_directory, "/src");
@@ -17,7 +17,6 @@ static void test_config_lifecycle() {
   EXPECT_TRUE(cfg->use_multithreading);
   EXPECT_FALSE(cfg->use_chunk_serialization);
   EXPECT_FALSE(cfg->use_compression);
-  EXPECT_EQ_INT(cfg->num_connections, 4);
   EXPECT_EQ_INT(cfg->transport, TRANSPORT_TCP);
   EXPECT_NULL(cfg->ssh_destination);
   config_delete(cfg);
@@ -25,7 +24,7 @@ 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);
+                              true, false, false, false, false, 1, false);
   EXPECT_NOT_NULL(cfg);
   EXPECT_EQ_INT(cfg->transport, TRANSPORT_TCP);
   EXPECT_NULL(cfg->ssh_destination);
@@ -40,7 +39,7 @@ static void test_config_ssh_dest() {
 
 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);
+                              true, false, false, false, false, 1, false);
   config_parse_ssh_dest(cfg);
   EXPECT_EQ_INT(cfg->transport, TRANSPORT_TCP);
   EXPECT_NULL(cfg->ssh_destination);
@@ -50,7 +49,7 @@ static void test_config_ssh_dest_local_path() {
 
 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);
+                              true, false, false, false, false, 1, false);
   config_parse_ssh_dest(cfg);
   EXPECT_EQ_INT(cfg->transport, TRANSPORT_SSH);
   EXPECT_EQ_STR(cfg->ssh_destination, "host:/remote");
@@ -60,7 +59,7 @@ static void test_config_ssh_dest_no_user() {
 
 static void test_pipeline_sender_lifecycle() {
   Config *cfg = config_create(str_dup("2.0"), str_dup("/src2"),
-                              str_dup("/dst2"), false, false, true, true, false, 1, 8, false);
+                              str_dup("/dst2"), false, false, true, true, false, 1, false);
   Queue *q1 = queue_create(5, NULL);
   Queue *q2 = queue_create(15, NULL);
 
@@ -77,7 +76,7 @@ static void test_pipeline_sender_lifecycle() {
 
 static void test_pipeline_receiver_lifecycle() {
   Config *cfg = config_create(str_dup("3.0"), str_dup("/src3"),
-                              str_dup("/dst3"), true, true, true, true, false, 1, 2, false);
+                              str_dup("/dst3"), true, true, true, true, false, 1, false);
   Queue *q = queue_create(20, NULL);
 
   PipelineContextReceiver *pcr = pipeline_context_receiver_create(cfg, q, 42);