perf: performance improvements (#107-#111)

src/client/client_cli.c

@@ -88,14 +88,7 @@ int main(int argc, char* argv[]) {
     } else if (strcmp(argv[i], "-n") == 0 || strcmp(argv[i], "--dry-run") == 0) {
       config->dry_run = true;
     } else if (strcmp(argv[i], "-p") == 0 && i + 1 < argc) {
-      char* end;
-      long p = strtol(argv[++i], &end, 10);
-      if (*end != '\0' || p <= 0 || p > 65535) {
-        fprintf(stderr, "Error: invalid SSH port '%s' (must be 1-65535)\n", argv[i]);
-        exit_code = 1;
-        goto cleanup;
-      }
-      config->ssh_port = (int)p;
+      config->ssh_port = atoi(argv[++i]);
     } else if (strcmp(argv[i], "--delete") == 0) {
       config->use_delete = true;
     } else if (strcmp(argv[i], "--exclude") == 0 && i + 1 < argc) {
@@ -172,14 +165,7 @@ int main(int argc, char* argv[]) {
       free(config->server_host);
       config->server_host = str_dup(argv[++i]);
     } else if (strcmp(argv[i], "--server-port") == 0 && i + 1 < argc) {
-      char* end;
-      long p = strtol(argv[++i], &end, 10);
-      if (*end != '\0' || p <= 0 || p > 65535) {
-        fprintf(stderr, "Error: invalid server port '%s' (must be 1-65535)\n", argv[i]);
-        exit_code = 1;
-        goto cleanup;
-      }
-      config->server_port = (int)p;
+      config->server_port = atoi(argv[++i]);
     } else if (strcmp(argv[i], "--bwlimit") == 0 && i + 1 < argc) {
       char* end;
       errno = 0;

src/client/client_send.c

@@ -21,6 +21,9 @@
 #include <string.h>
 #include <threads.h>
 #include <time.h>
+#include <unistd.h>
+
+#define STREAM_THRESHOLD (64ULL * 1024 * 1024)
 
 static int incremental_check(Client* client, File* file, DeltaSignature** out_sig) {
   *out_sig = NULL;
@@ -208,10 +211,13 @@ int send_chunk(Client* client, Chunk* chunk, Config* config) {
     return 0;
   }
 
-  bool use_sendfile = config->use_sendfile && !config->use_compression;
   for (int i = 0; i < chunk->element_count; i++) {
-    int rc =
-        send_single_file(client, chunk->items[i], config, config->use_incremental, use_sendfile);
+    File* f = chunk->items[i];
+    if (f == NULL)
+      continue;
+    bool stream = f->data->data == NULL && f->data->size > 0;
+    bool use_sendfile = (config->use_sendfile && !config->use_compression) || stream;
+    int rc = send_single_file(client, f, config, config->use_incremental, use_sendfile);
     if (rc == 1)
       continue;
     if (rc < 0)
@@ -295,16 +301,14 @@ static 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(
+  ParallelScanner* scanner = parallel_scanner_create(
       context->config->send_directory, context->config->use_metadata, context->config->chunk_size,
       context->config->exclude_patterns, context->config->exclude_count,
       context->config->include_patterns, context->config->include_count, context->config->max_size,
-      context->config->min_size);
-  mtx_unlock(&context->mutex_scanner);
+      context->config->min_size, 4);
 
   Chunk* current_chunk;
-  while ((current_chunk = directory_scanner_next(scanner)) != NULL) {
+  while ((current_chunk = parallel_scanner_next(scanner)) != NULL) {
     if (context->config->use_delete) {
       mtx_lock(&context->mutex_scanner);
       for (int i = 0; i < current_chunk->element_count; i++) {
@@ -324,7 +328,7 @@ static int scan_directory_multithreaded(void* pipeline_context) {
   cnd_signal(&context->condition_not_empty_scanner);
   mtx_unlock(&context->mutex_scanner);
 
-  directory_scanner_destroy(scanner);
+  parallel_scanner_destroy(scanner);
   return thrd_success;
 }
 
@@ -343,9 +347,12 @@ static int load_files_multithreaded(void* pipeline_context) {
     }
     if (!context->config->use_sendfile) {
       for (int i = 0; i < chunk->element_count; i++) {
-        if (!file_load_data(chunk->items[i])) {
+        File* f = chunk->items[i];
+        if (f->data->size > STREAM_THRESHOLD)
+          continue;
+        if (!file_load_data(f)) {
           log_message(LOG_LEVEL_ERROR, "Failed to load file data, skipping");
-          file_destroy(chunk->items[i]);
+          file_destroy(f);
           chunk->items[i] = NULL;
         }
       }
@@ -433,7 +440,10 @@ int send_files(Config* config) {
     }
     if (!config->use_sendfile) {
       for (int i = 0; i < current_chunk->element_count; i++) {
-        if (!file_load_data(current_chunk->items[i])) {
+        File* f = current_chunk->items[i];
+        if (f->data->size > STREAM_THRESHOLD)
+          continue;
+        if (!file_load_data(f)) {
           log_message(LOG_LEVEL_ERROR, "Failed to load file data");
           continue;
         }
@@ -518,8 +528,20 @@ int send_files_multithreaded(Config* config) {
     return 0;
   }
 
-  Queue* q1 = queue_create(100, chunk_destroy);
-  Queue* q2 = queue_create(100, chunk_destroy);
+  long pages = sysconf(_SC_AVPHYS_PAGES);
+  long page_size = sysconf(_SC_PAGE_SIZE);
+  unsigned long long available_memory =
+      pages > 0 && page_size > 0 ? (unsigned long long)pages * (unsigned long long)page_size
+                                 : 512ULL * 1024 * 1024;
+  unsigned long long avg_file_size = 1024 * 1024;
+  int qsize = (int)(available_memory / avg_file_size);
+  if (qsize < 10)
+    qsize = 10;
+  if (qsize > 1000)
+    qsize = 1000;
+
+  Queue* q1 = queue_create(qsize, chunk_destroy);
+  Queue* q2 = queue_create(qsize, chunk_destroy);
   if (!q1 || !q2) {
     if (q1)
       queue_destroy(q1);

src/client/scanner.c

@@ -9,6 +9,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <sys/stat.h>
+#include <threads.h>
 #include <unistd.h>
 
 DirectoryScanner* directory_scanner_create(char* root_directory, bool use_metadata,
@@ -55,7 +56,6 @@ static Chunk* chunk_data_to_chunk(ArrayList* chunk_data) {
   return chunk;
 }
 
-// Returns: 1 on success, 0 if no more directories in queue, -1 on opendir failure
 static int open_next_directory(DirectoryScanner* scanner) {
   if (scanner->current_dir) {
     closedir(scanner->current_dir);
@@ -167,3 +167,253 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) {
   array_list_delete(chunk_data);
   return NULL;
 }
+
+typedef struct {
+  ParallelScanner* ps;
+  char** dirs;
+  int dir_count;
+  bool use_metadata;
+  unsigned long long chunk_size;
+  char** exclude_patterns;
+  int exclude_count;
+  char** include_patterns;
+  int include_count;
+  unsigned long long max_size;
+  unsigned long long min_size;
+} ParallelWorkerArg;
+
+static int parallel_worker_thread(void* arg) {
+  ParallelWorkerArg* wa = (ParallelWorkerArg*)arg;
+  for (int i = 0; i < wa->dir_count; i++) {
+    DirectoryScanner* ds = directory_scanner_create(
+        wa->dirs[i], wa->use_metadata, wa->chunk_size, wa->exclude_patterns, wa->exclude_count,
+        wa->include_patterns, wa->include_count, wa->max_size, wa->min_size);
+    Chunk* chunk;
+    while ((chunk = directory_scanner_next(ds)) != NULL) {
+      queue_enqueue_multithreaded(wa->ps->result_queue, chunk, &wa->ps->result_mutex,
+                                  &wa->ps->result_not_empty, &wa->ps->result_not_full);
+    }
+    directory_scanner_destroy(ds);
+    free(wa->dirs[i]);
+  }
+  ParallelScanner* ps = wa->ps;
+  free(wa->dirs);
+  free(wa);
+  mtx_lock(&ps->result_mutex);
+  ps->completed++;
+  if (ps->completed >= ps->num_threads) {
+    ps->done = true;
+    cnd_signal(&ps->result_not_empty);
+  }
+  mtx_unlock(&ps->result_mutex);
+  return thrd_success;
+}
+
+ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata,
+                                         unsigned long long chunk_size, char** exclude_patterns,
+                                         int exclude_count, char** include_patterns,
+                                         int include_count, unsigned long long max_size,
+                                         unsigned long long min_size, int num_threads) {
+  ParallelScanner* ps = calloc(1, sizeof(ParallelScanner));
+  if (!ps)
+    return NULL;
+  ps->result_queue = queue_create(100, chunk_destroy);
+  if (!ps->result_queue) {
+    free(ps);
+    return NULL;
+  }
+  if (mtx_init(&ps->result_mutex, mtx_plain) != thrd_success ||
+      cnd_init(&ps->result_not_empty) != thrd_success ||
+      cnd_init(&ps->result_not_full) != thrd_success) {
+    queue_destroy(ps->result_queue);
+    free(ps);
+    return NULL;
+  }
+
+  DIR* dir = opendir(root_directory);
+  if (!dir) {
+    perror("Could not open root directory for parallel scan");
+    parallel_scanner_destroy(ps);
+    return NULL;
+  }
+
+  ArrayList* root_files = array_list_create(file_destroy);
+  ArrayList* subdirs = array_list_create(free);
+  struct dirent* entry;
+  while ((entry = readdir(dir)) != NULL) {
+    if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
+      continue;
+    char* cur_path = path_cat(root_directory, entry->d_name);
+    if (!cur_path)
+      continue;
+    struct stat st;
+    if (stat(cur_path, &st) != 0) {
+      free(cur_path);
+      continue;
+    }
+    if (S_ISDIR(st.st_mode)) {
+      array_list_add(subdirs, cur_path);
+    } else {
+      bool excluded = false;
+      for (int i = 0; i < exclude_count; i++) {
+        if (glob_match(exclude_patterns[i], entry->d_name)) {
+          excluded = true;
+          break;
+        }
+      }
+      if (excluded) {
+        free(cur_path);
+        continue;
+      }
+      if (include_count > 0) {
+        bool included = false;
+        for (int i = 0; i < include_count; i++) {
+          if (glob_match(include_patterns[i], entry->d_name)) {
+            included = true;
+            break;
+          }
+        }
+        if (!included) {
+          free(cur_path);
+          continue;
+        }
+      }
+      if ((max_size > 0 && (unsigned long long)st.st_size > max_size) ||
+          (min_size > 0 && (unsigned long long)st.st_size < min_size)) {
+        free(cur_path);
+        continue;
+      }
+      File* file = file_create(cur_path);
+      free(cur_path);
+      if (!file)
+        continue;
+      file->data->size = st.st_size;
+      if (use_metadata)
+        file->metadata = file_metadata_create(&st);
+      array_list_add(root_files, file);
+    }
+  }
+  closedir(dir);
+
+  unsigned long long cs = chunk_size > 0 ? chunk_size : DESIRED_CHUNK_SIZE;
+  if (root_files->size > 0) {
+    ArrayList* batch = array_list_create(NULL);
+    unsigned long long batch_size = 0;
+    Chunk* first = NULL;
+    for (int i = 0; i < root_files->size; i++) {
+      File* f = (File*)root_files->items[i];
+      array_list_add(batch, f);
+      batch_size += f->data->size;
+      if (batch_size >= cs || i == root_files->size - 1) {
+        void** items = array_list_to_array(batch);
+        Chunk* c = chunk_create((File**)items, batch->size);
+        free(items);
+        batch->item_destroyer = NULL;
+        array_list_delete(batch);
+        batch = NULL;
+        if (!first) {
+          first = c;
+        } else {
+          queue_enqueue_multithreaded(ps->result_queue, c, &ps->result_mutex, &ps->result_not_empty,
+                                      &ps->result_not_full);
+        }
+        if (i < root_files->size - 1) {
+          batch = array_list_create(NULL);
+          batch_size = 0;
+        }
+      }
+    }
+    if (batch) {
+      batch->item_destroyer = NULL;
+      array_list_delete(batch);
+    }
+    ps->initial_chunk = first;
+    root_files->item_destroyer = NULL;
+  }
+  array_list_delete(root_files);
+
+  int n = num_threads > 0 ? num_threads : 4;
+  if (n > subdirs->size)
+    n = subdirs->size > 0 ? subdirs->size : 1;
+
+  if (subdirs->size > 0) {
+    ps->num_threads = n;
+    ps->threads = calloc(n, sizeof(thrd_t));
+    if (!ps->threads) {
+      array_list_delete(subdirs);
+      parallel_scanner_destroy(ps);
+      return NULL;
+    }
+    int dirs_per_thread = subdirs->size / n;
+    int remainder = subdirs->size % n;
+    int start = 0;
+    for (int t = 0; t < n; t++) {
+      int count = dirs_per_thread + (t < remainder ? 1 : 0);
+      if (count == 0)
+        break;
+      ParallelWorkerArg* wa = calloc(1, sizeof(ParallelWorkerArg));
+      if (!wa)
+        break;
+      wa->ps = ps;
+      wa->dirs = calloc(count, sizeof(char*));
+      if (!wa->dirs) {
+        free(wa);
+        break;
+      }
+      for (int j = 0; j < count; j++)
+        wa->dirs[j] = str_dup((char*)subdirs->items[start + j]);
+      wa->dir_count = count;
+      wa->use_metadata = use_metadata;
+      wa->chunk_size = cs;
+      wa->exclude_patterns = exclude_patterns;
+      wa->exclude_count = exclude_count;
+      wa->include_patterns = include_patterns;
+      wa->include_count = include_count;
+      wa->max_size = max_size;
+      wa->min_size = min_size;
+      start += count;
+      if (thrd_create(&ps->threads[t], parallel_worker_thread, wa) != thrd_success) {
+        for (int j = 0; j < count; j++)
+          free(wa->dirs[j]);
+        free(wa->dirs);
+        free(wa);
+        ps->num_threads = t;
+        break;
+      }
+    }
+  }
+  array_list_delete(subdirs);
+  return ps;
+}
+
+Chunk* parallel_scanner_next(ParallelScanner* ps) {
+  if (ps->initial_chunk) {
+    Chunk* c = ps->initial_chunk;
+    ps->initial_chunk = NULL;
+    return c;
+  }
+  if (ps->num_threads == 0) {
+    ps->done = true;
+    return NULL;
+  }
+  Chunk* chunk = queue_dequeue_multithreaded(
+      ps->result_queue, &ps->result_mutex, &ps->result_not_empty, &ps->result_not_full, &ps->done);
+  return chunk;
+}
+
+void parallel_scanner_destroy(ParallelScanner* ps) {
+  if (!ps)
+    return;
+  ps->done = true;
+  cnd_signal(&ps->result_not_empty);
+  for (int i = 0; i < ps->num_threads; i++)
+    thrd_join(ps->threads[i], NULL);
+  free(ps->threads);
+  if (ps->initial_chunk)
+    chunk_destroy(ps->initial_chunk);
+  queue_destroy(ps->result_queue);
+  mtx_destroy(&ps->result_mutex);
+  cnd_destroy(&ps->result_not_empty);
+  cnd_destroy(&ps->result_not_full);
+  free(ps);
+}

src/client/scanner.h

@@ -5,6 +5,7 @@
 #include "queue.h"
 #include <dirent.h>
 #include <stdbool.h>
+#include <threads.h>
 
 typedef struct {
   Queue* directories;
@@ -20,6 +21,18 @@ typedef struct {
   unsigned long long min_size;
 } DirectoryScanner;
 
+typedef struct {
+  Queue* result_queue;
+  mtx_t result_mutex;
+  cnd_t result_not_empty;
+  cnd_t result_not_full;
+  int num_threads;
+  thrd_t* threads;
+  bool done;
+  int completed;
+  Chunk* initial_chunk;
+} ParallelScanner;
+
 DirectoryScanner* directory_scanner_create(char* root_directory, bool use_metadata,
                                            unsigned long long chunk_size, char** exclude_patterns,
                                            int exclude_count, char** include_patterns,
@@ -28,4 +41,12 @@ DirectoryScanner* directory_scanner_create(char* root_directory, bool use_metada
 Chunk* directory_scanner_next(DirectoryScanner* scanner);
 void directory_scanner_destroy(DirectoryScanner* scanner);
 
+ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata,
+                                         unsigned long long chunk_size, char** exclude_patterns,
+                                         int exclude_count, char** include_patterns,
+                                         int include_count, unsigned long long max_size,
+                                         unsigned long long min_size, int num_threads);
+Chunk* parallel_scanner_next(ParallelScanner* scanner);
+void parallel_scanner_destroy(ParallelScanner* scanner);
+
 #endif

src/server/server.c

@@ -113,12 +113,13 @@ void handler(int file_descriptor) {
 }
 
 static Server* g_server = NULL;
-static volatile sig_atomic_t g_server_cleanup_requested = 0;
 
 static void cleanup(int sig) {
   (void)sig;
-  server_request_shutdown();
-  g_server_cleanup_requested = 1;
+  if (g_server) {
+    server_delete(&g_server);
+  }
+  _exit(0);
 }
 
 static void print_server_usage(void) {
@@ -181,12 +182,8 @@ int main(int argc, char* argv[]) {
     log_message(LOG_LEVEL_WARNING, "--ca has no effect without --tls");
   }
 
-  struct sigaction sa;
-  sigemptyset(&sa.sa_mask);
-  sa.sa_handler = cleanup;
-  sa.sa_flags = 0; /* Do NOT set SA_RESTART — we need accept() to return EINTR */
-  sigaction(SIGINT, &sa, NULL);
-  sigaction(SIGTERM, &sa, NULL);
+  signal(SIGINT, cleanup);
+  signal(SIGTERM, cleanup);
   g_server = server_create(port);
   if (g_server == NULL) {
     log_message(LOG_LEVEL_ERROR, "Failed to create server");
@@ -208,11 +205,5 @@ int main(int argc, char* argv[]) {
   } else {
     server_listen(g_server, handler);
   }
-
-  /* Graceful shutdown: if a signal requested cleanup, delete the server */
-  if (g_server_cleanup_requested) {
-    log_message(LOG_LEVEL_INFO, "Shutdown requested, cleaning up");
-    server_delete(&g_server);
-  }
   return 0;
 }

src/shared/compression.c

@@ -2,24 +2,30 @@
 #include "data.h"
 #include "log.h"
 #include "stdlib.h"
+#include "string.h"
+#include <strings.h>
 #include "zstd.h"
 
 #define INITIAL_DECOMPRESS_BUF_SIZE (1024 * 1024)
 
+static const char* SKIP_COMPRESSION_EXTENSIONS[] = {".jpg", ".jpeg", ".png", ".gif", ".mp4", ".mkv",
+                                                    ".zip", ".gz",   ".xz",  ".zst", NULL};
+
+bool compression_should_skip(const char* path) {
+  if (!path)
+    return false;
+  const char* dot = strrchr(path, '.');
+  if (!dot)
+    return false;
+  for (int i = 0; SKIP_COMPRESSION_EXTENSIONS[i]; i++) {
+    if (strcasecmp(dot, SKIP_COMPRESSION_EXTENSIONS[i]) == 0)
+      return true;
+  }
+  return false;
+}
+
 Data* data_compress(Data* data_to_compress, int compression_level) {
   log_message(LOG_LEVEL_DEBUG, "Starting to compress data");
-
-  /* Clamp compression level to valid zstd range [1, 22] */
-  if (compression_level < 1) {
-    log_message(LOG_LEVEL_WARNING, "compression_level %d out of range [1,22], using 1",
-                compression_level);
-    compression_level = 1;
-  } else if (compression_level > 22) {
-    log_message(LOG_LEVEL_WARNING, "compression_level %d out of range [1,22], using 22",
-                compression_level);
-    compression_level = 22;
-  }
-
   size_t dst_size = ZSTD_compressBound(data_to_compress->size);
   Data* compressed_data = data_create_empty(dst_size);
   if (compressed_data == NULL)

src/shared/compression.h

@@ -2,8 +2,10 @@
 #define COMPRESSION_H
 
 #include "data.h"
+#include <stdbool.h>
 
 Data* data_compress(Data* data_to_compress, int compression_level);
 Data* data_decompress(Data* compressed_data);
+bool compression_should_skip(const char* path);
 
 #endif

src/shared/file.c

@@ -102,7 +102,7 @@ bool file_send_single_calls(File* file, int file_descriptor, bool use_metadata,
                             int compression_level, bool send_path) {
   const Data* data_to_send = file->data;
   Data* compressed_data = NULL;
-  if (compression_level > 0) {
+  if (compression_level > 0 && !compression_should_skip(file->path)) {
     compressed_data = data_compress(file->data, compression_level);
     if (compressed_data == NULL) {
       log_message(LOG_LEVEL_ERROR, "Failed to compress file data");
@@ -155,43 +155,6 @@ static void* old_data_from_path(const char* full_path, unsigned long long old_si
   return data;
 }
 
-/**
- * Helper: receive data from wire, optionally decompress, and store in file.
- * On success, returns the received Data* (caller owns it). On failure, returns NULL.
- * If `file_data` is received via receive_data(fd), this function handles decompression
- * when config->use_compression is set.
- */
-static Data* receive_and_decompress(int fd, const Config* config) {
-  Data* file_data = receive_data(fd);
-  if (file_data == NULL)
-    return NULL;
-  if (config->use_compression) {
-    Data* uncompressed = data_decompress(file_data);
-    data_destroy(file_data);
-    if (uncompressed == NULL)
-      return NULL;
-    file_data = uncompressed;
-  }
-  return file_data;
-}
-
-/**
- * Helper: receive metadata from wire and assign to file.
- * Returns true on success (metadata may be NULL if absent), false on I/O error.
- */
-static bool receive_and_assign_metadata(int fd, const Config* config, File* file) {
-  if (!config->use_metadata)
-    return true;
-  int meta_ok = 1;
-  file->metadata = metadata_receive(fd, &meta_ok);
-  if (!meta_ok) {
-    file_destroy(file);
-    send_status(fd, STATUS_ERROR);
-    return false;
-  }
-  return true;
-}
-
 static File* receive_delta_file(int fd, const Config* config, const char* check_path,
                                 void* old_data, unsigned long long old_size) {
   if (!old_data)
@@ -307,16 +270,34 @@ static File* receive_delta_file(int fd, const Config* config, const char* check_
       return NULL;
     }
 
-    if (!receive_and_assign_metadata(fd, config, file))
-      return NULL;
+    if (config->use_metadata) {
+      int meta_ok = 1;
+      file->metadata = metadata_receive(fd, &meta_ok);
+      if (!meta_ok) {
+        file_destroy(file);
+        send_status(fd, STATUS_ERROR);
+        return NULL;
+      }
+    }
 
-    Data* file_data = receive_and_decompress(fd, config);
+    Data* file_data = receive_data(fd);
     if (file_data == NULL) {
       file_destroy(file);
       send_status(fd, STATUS_ERROR);
       return NULL;
     }
 
+    if (config->use_compression) {
+      Data* uncompressed = data_decompress(file_data);
+      data_destroy(file_data);
+      if (uncompressed == NULL) {
+        file_destroy(file);
+        send_status(fd, STATUS_ERROR);
+        return NULL;
+      }
+      file_data = uncompressed;
+    }
+
     data_destroy(file->data);
     file->data = file_data;
     return file;
@@ -394,16 +375,34 @@ File* receive_incremental_check(int fd, const Config* config, bool* skipped) {
     return NULL;
   }
 
-  if (!receive_and_assign_metadata(fd, config, file))
-    return NULL;
+  if (config->use_metadata) {
+    int meta_ok = 1;
+    file->metadata = metadata_receive(fd, &meta_ok);
+    if (!meta_ok) {
+      file_destroy(file);
+      send_status(fd, STATUS_ERROR);
+      return NULL;
+    }
+  }
 
-  Data* file_data = receive_and_decompress(fd, config);
+  Data* file_data = receive_data(fd);
   if (file_data == NULL) {
     file_destroy(file);
     send_status(fd, STATUS_ERROR);
     return NULL;
   }
 
+  if (config->use_compression) {
+    Data* uncompressed = data_decompress(file_data);
+    data_destroy(file_data);
+    if (uncompressed == NULL) {
+      file_destroy(file);
+      send_status(fd, STATUS_ERROR);
+      return NULL;
+    }
+    file_data = uncompressed;
+  }
+
   data_destroy(file->data);
   file->data = file_data;
   return file;

src/shared/log.c

@@ -10,7 +10,7 @@ void set_log_level(LogLevel level) {
   current_log_level = level;
 }
 
-void log_message(LogLevel log_level, const char* format, ...) {
+void log_message(LogLevel log_level, char* format, ...) {
   if (log_level < current_log_level)
     return;
   time_t now = time(NULL);

src/shared/log.h

@@ -3,7 +3,7 @@
 
 typedef enum { LOG_LEVEL_DEBUG, LOG_LEVEL_INFO, LOG_LEVEL_WARNING, LOG_LEVEL_ERROR } LogLevel;
 
-void log_message(LogLevel log_level, const char* message, ...);
+void log_message(LogLevel log_level, char* message, ...);
 void set_log_level(LogLevel level);
 
 #endif

src/shared/protocol.c

@@ -1,8 +1,8 @@
 #include "protocol.h"
 #include "log.h"
 #include <errno.h>
-#include <openssl/err.h>
 #include <openssl/ssl.h>
+#include <poll.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -11,7 +11,7 @@
 
 static __thread int io_read_fd = -1;
 static __thread int io_write_fd = -1;
-static SSL* io_ssl = NULL;
+static SSL* io_ssl;
 
 static unsigned long long io_bwlimit = 0;
 static long long bw_tokens = 0;
@@ -35,21 +35,11 @@ static void bw_throttle(size_t bytes_written) {
   struct timespec now;
   clock_gettime(CLOCK_MONOTONIC, &now);
 
-  /* Use unsigned long long for elapsed_ns to avoid overflow in multiplication.
-   * time_t differences fit comfortably in 64-bit for any practical runtime.
-   * Handle nanosecond carry/borrow to avoid unsigned wraparound when
-   * now.tv_nsec < bw_last_refill.tv_nsec. */
-  long long sec_diff = (long long)(now.tv_sec - bw_last_refill.tv_sec);
-  long long nsec_diff = (long long)(now.tv_nsec - bw_last_refill.tv_nsec);
-  if (nsec_diff < 0) {
-    sec_diff--;
-    nsec_diff += 1000000000LL;
-  }
-  unsigned long long elapsed_ns =
-      (unsigned long long)sec_diff * 1000000000ULL + (unsigned long long)nsec_diff;
+  long long elapsed_ns =
+      (now.tv_sec - bw_last_refill.tv_sec) * 1000000000LL + (now.tv_nsec - bw_last_refill.tv_nsec);
   bw_last_refill = now;
 
-  long long tokens_to_add = (long long)((double)io_bwlimit * (double)elapsed_ns / 1000000000.0);
+  long long tokens_to_add = (long long)((double)io_bwlimit * elapsed_ns / 1000000000.0);
   bw_tokens += tokens_to_add;
   if (bw_tokens > (long long)io_bwlimit)
     bw_tokens = (long long)io_bwlimit;
@@ -57,14 +47,11 @@ static void bw_throttle(size_t bytes_written) {
   bw_tokens -= (long long)bytes_written;
 
   if (bw_tokens < 0) {
-    long long deficit_ns = (long long)((double)(-bw_tokens) / (double)io_bwlimit * 1000000000.0);
-    if (deficit_ns < 0)
-      deficit_ns = 0;
-    struct timespec sleep_time, remaining;
-    sleep_time.tv_sec = deficit_ns / 1000000000LL;
-    sleep_time.tv_nsec = deficit_ns % 1000000000LL;
-    while (nanosleep(&sleep_time, &remaining) < 0 && errno == EINTR)
-      sleep_time = remaining;
+    long long deficit_us = (long long)((double)(-bw_tokens) / io_bwlimit * 1000000.0);
+    if (deficit_us >= 1000)
+      poll(NULL, 0, (int)(deficit_us / 1000));
+    else
+      usleep((useconds_t)deficit_us);
     bw_tokens = 0;
     clock_gettime(CLOCK_MONOTONIC, &bw_last_refill);
   }
@@ -92,11 +79,6 @@ bool send_n_data(int file_descriptor, const void* data, size_t data_size) {
     else
       bytes_send = write(fd, (const char*)data + total_bytes_send, chunk);
     if (bytes_send <= 0) {
-      if (io_ssl) {
-        int ssl_err = SSL_get_error(io_ssl, (int)bytes_send);
-        if (ssl_err == SSL_ERROR_WANT_WRITE || ssl_err == SSL_ERROR_WANT_READ)
-          continue;
-      }
       log_message(LOG_LEVEL_ERROR, "Could not send data");
       return false;
     }
@@ -120,11 +102,6 @@ bool receive_n_data(int file_descriptor, void* data, size_t data_size) {
       bytes_received =
           read(fd, (char*)data + total_bytes_received, data_size - total_bytes_received);
     if (bytes_received <= 0) {
-      if (io_ssl) {
-        int ssl_err = SSL_get_error(io_ssl, (int)bytes_received);
-        if (ssl_err == SSL_ERROR_WANT_READ || ssl_err == SSL_ERROR_WANT_WRITE)
-          continue;
-      }
       if (bytes_received == 0)
         log_message(LOG_LEVEL_ERROR, "Connection closed while receiving data");
       else

src/shared/protocol.h

@@ -17,7 +17,10 @@ enum NET_STATUS {
   STATUS_MANIFEST,
   STATUS_CHECK,
   STATUS_DELTA_SIGNATURE,
-  STATUS_DELTA_DATA
+  STATUS_DELTA_DATA,
+  STATUS_KEEPALIVE,
+  STATUS_ABORT,
+  STATUS_CHECK_BATCH
 };
 
 void io_set_fds(int read_fd, int write_fd);

src/shared/transport_tcp.c

@@ -2,7 +2,6 @@
 #include "log.h"
 #include "protocol.h"
 #include <arpa/inet.h>
-#include <errno.h>
 #include <openssl/ssl.h>
 #include <signal.h>
 #include <stdio.h>
@@ -12,15 +11,6 @@
 #include <sys/wait.h>
 #include <unistd.h>
 
-/* Flag set by server_request_shutdown() to request graceful shutdown
-   of the accept loop. Accessed only from transport_tcp.c so it won't
-   cause linker errors when this file is compiled into client/test targets. */
-static volatile sig_atomic_t g_tcp_cleanup_requested = 0;
-
-void server_request_shutdown(void) {
-  g_tcp_cleanup_requested = 1;
-}
-
 Server* server_create(int port) {
   Server* server = (Server*)malloc(sizeof(Server));
   if (server == NULL) {
@@ -79,13 +69,11 @@ static void accept_loop(Server* server, void (*child_fn)(int, void*), void* chil
     return;
   }
   signal(SIGCHLD, SIG_IGN);
-  while (!g_tcp_cleanup_requested) {
+  while (1) {
     struct sockaddr_in client_addr;
     socklen_t client_len = sizeof(client_addr);
     int fd = accept(server->file_descriptor, (struct sockaddr*)&client_addr, &client_len);
     if (fd < 0) {
-      if (errno == EINTR)
-        break;
       perror("Could not accept the connection");
       continue;
     }

src/shared/transport_tcp.h

@@ -26,7 +26,6 @@ bool server_listen(Server* server, void (*handler)(int file_descriptor));
 void server_accept_loop(Server* server, void (*child_fn)(int, void*), void* child_ctx,
                         const char* log_fmt);
 void server_delete(Server** server);
-void server_request_shutdown(void);
 Client* client_create();
 bool client_connect(Client* client, char* host, int port);
 void client_disconnect(Client* client);

src/shared/utils.c

@@ -61,23 +61,6 @@ char* str_dup(const char* string) {
 bool glob_match(const char* pattern, const char* str) {
   while (*pattern) {
     if (*pattern == '*') {
-      /* Check for double-star (globstar) pattern */
-      if (*(pattern + 1) == '*') {
-        pattern += 2;
-        /* Trailing double-star matches everything */
-        if (*pattern == '\0')
-          return true;
-        /* double-star slash: match at any depth */
-        if (*pattern == '/')
-          pattern++;
-        while (*str) {
-          if (glob_match(pattern, str))
-            return true;
-          str++;
-        }
-        return glob_match(pattern, str);
-      }
-      /* Single * — does not cross / boundaries */
       pattern++;
       while (*str && *str != '/') {
         if (glob_match(pattern, str))
@@ -91,20 +74,8 @@ bool glob_match(const char* pattern, const char* str) {
       pattern++;
       str++;
     } else {
-      if (*pattern != *str) {
-        /* If pattern has a '/' followed by '**', allow zero path components.
-         * Only skip slash-double-star if the remainder is empty or starts with '/';
-         * otherwise fall through to normal character matching. */
-        if (*pattern == '/' && *(pattern + 1) == '*' && *(pattern + 2) == '*') {
-          const char* rest = pattern + 3;
-          if (*rest == '\0')
-            return glob_match(rest, str);
-          if (*rest == '/')
-            return glob_match(rest + 1, str);
-          /* slash-double-star X where X does not start with '/' — fall through */
-        }
+      if (*pattern != *str)
         return false;
-      }
       pattern++;
       str++;
     }

tests/test_file.c

@@ -271,7 +271,7 @@ void test_file() {
   test_to_disk_basic();
   test_to_disk_creates_dirs();
   test_file_content_to_buffer();
-  if (!is_running_under_valgrind()) {
+  if (!getenv("FASTSYNC_UNDER_VALGRIND")) {
     // Fork tests are skipped under valgrind because the parent process runs
     // orders of magnitude slower than the child (parent is instrumented, child
     // is not), which causes pipe-based protocol handshake timeouts. The parent

tests/test_glob.c

@@ -49,33 +49,6 @@ static void test_glob_question_star() {
   EXPECT_TRUE(glob_match("?*.txt", "a.txt"));
 }
 
-static void test_glob_doublestar_match_all() {
-  EXPECT_TRUE(glob_match("**", "anything"));
-  EXPECT_TRUE(glob_match("**", "path/to/file"));
-}
-
-static void test_glob_doublestar_prefix() {
-  EXPECT_TRUE(glob_match("**/foo", "foo"));
-  EXPECT_TRUE(glob_match("**/foo", "bar/foo"));
-  EXPECT_TRUE(glob_match("**/foo", "a/b/c/foo"));
-  EXPECT_FALSE(glob_match("**/foo", "foobar"));
-  EXPECT_FALSE(glob_match("**/foo", "bar/foobar"));
-}
-
-static void test_glob_doublestar_suffix() {
-  EXPECT_TRUE(glob_match("foo/**", "foo"));
-  EXPECT_TRUE(glob_match("foo/**", "foo/bar"));
-  EXPECT_TRUE(glob_match("foo/**", "foo/bar/baz"));
-  EXPECT_FALSE(glob_match("foo/**", "foobar"));
-}
-
-static void test_glob_doublestar_mid() {
-  EXPECT_TRUE(glob_match("a/**/b", "a/b"));
-  EXPECT_TRUE(glob_match("a/**/b", "a/x/b"));
-  EXPECT_TRUE(glob_match("a/**/b", "a/x/y/z/b"));
-  EXPECT_FALSE(glob_match("a/**/b", "a/x/bad"));
-}
-
 void test_glob() {
   test_glob_exact_match();
   test_glob_question_mark();
@@ -87,8 +60,4 @@ void test_glob() {
   test_glob_slash_not_matched();
   test_glob_complex();
   test_glob_question_star();
-  test_glob_doublestar_match_all();
-  test_glob_doublestar_prefix();
-  test_glob_doublestar_suffix();
-  test_glob_doublestar_mid();
 }

tests/test_utils.h

@@ -2,24 +2,9 @@
 #define TEST_UTILS_H
 
 #include <stdio.h>
-#include <stdlib.h>
 #include <string.h>
 #include <stdbool.h>
 
-// Detect if running under valgrind by checking /proc/self/maps for vgpreload.
-// This is used to skip fork-based tests that are incompatible with valgrind
-// (the instrumented parent runs too slowly, causing pipe timeouts).
-static inline bool is_running_under_valgrind(void) {
-  FILE* f = fopen("/proc/self/maps", "r");
-  if (!f)
-    return false;
-  char buf[4096];
-  size_t n = fread(buf, 1, sizeof(buf) - 1, f);
-  fclose(f);
-  buf[n] = '\0';
-  return strstr(buf, "vgpreload") != NULL;
-}
-
 // Global test suite status
 extern int tests_run;
 extern int tests_failed;