feat: add compression worker threads

src/client/client_cli.c

@@ -211,6 +211,13 @@ static int apply_table_option(Config* config, const OptionEntry* entry, const ch
 int parse_args(Config* config, int argc, char* argv[], int* positional_args,
                int* positional_count) {
   for (int i = 1; i < argc; i++) {
+    const char* threads_prefix = "--compress-threads=";
+    if (strncmp(argv[i], threads_prefix, strlen(threads_prefix)) == 0) {
+      if (set_positive_int_option(&config->compression_threads, argv[i] + strlen(threads_prefix),
+                                  "--compress-threads") != 0)
+        return -1;
+      continue;
+    }
     const OptionEntry* entry = find_table_option(argv[i]);
     if (entry) {
       if (entry->kind != OPT_FLAG) {
@@ -361,6 +368,10 @@ int parse_args(Config* config, int argc, char* argv[], int* positional_args,
         log_message(LOG_LEVEL_ERROR, "--compress-level must be between 1 and 22");
         return -1;
       }
+    } else if (opt_is(argv[i], "--compress-threads", NULL) && i + 1 < argc) {
+      if (set_positive_int_option(&config->compression_threads, argv[++i], "--compress-threads") !=
+          0)
+        return -1;
     } else if (argv[i][0] == '-') {
       fprintf(stderr, "Unknown option: %s\n", argv[i]);
       print_usage();

src/client/client_send.c

@@ -235,7 +235,8 @@ static int send_delta(Client* client, File* file, DeltaSignature* sig, Config* c
 
   Data* to_send = delta_data;
   if (config->use_compression) {
-    to_send = data_compress(delta_data, config->compression_level);
+    to_send = data_compress_with_threads(delta_data, config->compression_level,
+                                         config->compression_threads);
     data_destroy(delta_data);
     if (!to_send)
       return send_status(client->file_descriptor, STATUS_NEXT) ? 1 : -1;
@@ -251,13 +252,15 @@ static int send_delta(Client* client, File* file, DeltaSignature* sig, Config* c
   return ok ? 0 : -1;
 }
 
-typedef bool (*file_send_fn)(File*, int, bool, int, bool);
+typedef bool (*file_send_fn)(File*, int, bool, int, int, bool);
 
 // Send a single file directly (non-incremental path).
-static bool send_file_direct(File* file, int fd, bool use_metadata, int compression_level) {
+static bool send_file_direct(File* file, int fd, bool use_metadata, int compression_level,
+                             int compression_threads) {
   if (!send_status(fd, STATUS_NEXT))
     return false;
-  return file_send_single_calls(file, fd, use_metadata, compression_level, true);
+  return file_send_single_calls_with_threads(file, fd, use_metadata, compression_level,
+                                             compression_threads, true);
 }
 
 // Send a single file directly via sendfile (non-incremental path).
@@ -278,7 +281,8 @@ static int send_single_file(Client* client, File* file, Config* config, bool use
       return send_file_direct_sendfile(file, client->file_descriptor, config->use_metadata) ? 0
                                                                                             : -1;
     }
-    return send_file_direct(file, client->file_descriptor, config->use_metadata, compression_level)
+    return send_file_direct(file, client->file_descriptor, config->use_metadata, compression_level,
+                            config->compression_threads)
                ? 0
                : -1;
   }
@@ -310,7 +314,7 @@ static int send_single_file(Client* client, File* file, Config* config, bool use
   }
 
   // Incremental path with single_calls (supports compression and delta)
-  file_send_fn send_fn = (file_send_fn)file_send_single_calls;
+  file_send_fn send_fn = (file_send_fn)file_send_single_calls_with_threads;
   DeltaSignature* sig = NULL;
   int rc = incremental_check(client, file, config, &sig);
   if (rc < 0) {
@@ -338,7 +342,8 @@ static int send_single_file(Client* client, File* file, Config* config, bool use
         return -1;
     }
   }
-  if (!send_fn(file, client->file_descriptor, config->use_metadata, compression_level, false))
+  if (!send_fn(file, client->file_descriptor, config->use_metadata, compression_level,
+               config->compression_threads, false))
     return -1;
   return 0;
 }
@@ -349,7 +354,8 @@ int send_chunk(Client* client, Chunk* chunk, Config* config) {
       return -1;
     Data* data;
     if (config->use_compression) {
-      data = chunk_compress(chunk, config->compression_level, config->use_metadata);
+      data = chunk_compress_with_threads(chunk, config->compression_level, config->use_metadata,
+                                         config->compression_threads);
     } else {
       data = chunk_serialize(chunk, config->use_metadata);
     }

src/client/client_validation.c

@@ -15,6 +15,10 @@ bool validate_config(const Config* config) {
                                  "(chunk serialization)");
     return false;
   }
+  if (config->compression_threads > 0 && !config->use_compression) {
+    log_message(LOG_LEVEL_ERROR, "--compress-threads requires compression (-c or -z)");
+    return false;
+  }
   if (config->transport == TRANSPORT_SSH && config->use_sendfile) {
     log_message(LOG_LEVEL_ERROR, "-f/--sendfile is not supported with SSH transport");
     return false;

src/client/usage.c

@@ -69,6 +69,7 @@ void print_usage(void) {
   printf("  -S, --sparse        Handle sparse files efficiently\n");
   printf("  --inplace           Update files in-place (no temp+rename)\n");
   printf("  --compress-level <n>    Compression level (default: 5)\n");
+  printf("  --compress-threads <n> Compression worker threads (requires zstd threaded support)\n");
   printf("  --help              Show this help\n");
   printf("  -V, --version       Show version\n");
 }

src/shared/chunk.c

@@ -290,11 +290,16 @@ Chunk* chunk_deserialize(Data* data, bool use_metadata) {
 }
 
 Data* chunk_compress(Chunk* chunk, int compression_level, bool use_metadata) {
+  return chunk_compress_with_threads(chunk, compression_level, use_metadata, 0);
+}
+
+Data* chunk_compress_with_threads(Chunk* chunk, int compression_level, bool use_metadata,
+                                  int compression_threads) {
   log_message(LOG_LEVEL_DEBUG, "Starting to compress chunk");
   Data* serialized = chunk_serialize(chunk, use_metadata);
   if (serialized == NULL)
     return NULL;
-  Data* compressed = data_compress(serialized, compression_level);
+  Data* compressed = data_compress_with_threads(serialized, compression_level, compression_threads);
   data_destroy(serialized);
   if (compressed == NULL)
     return NULL;

src/shared/chunk.h

@@ -19,6 +19,8 @@ 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);
+Data* chunk_compress_with_threads(Chunk* chunk, int compression_level, bool use_metadata,
+                                  int compression_threads);
 Chunk* receive_chunk_data(int fd, const Config* config);
 
 #endif

src/shared/compression.c

@@ -28,6 +28,14 @@ bool compression_should_skip(const char* path) {
 }
 
 Data* data_compress(Data* data_to_compress, int compression_level) {
+  return data_compress_with_threads(data_to_compress, compression_level, 0);
+}
+
+Data* data_compress_with_threads(Data* data_to_compress, int compression_level,
+                                 int compression_threads) {
+  if (!data_to_compress || (!data_to_compress->data && data_to_compress->size != 0) ||
+      compression_threads < 0)
+    return NULL;
   log_message(LOG_LEVEL_DEBUG, "Starting to compress data");
   size_t dst_size = ZSTD_compressBound(data_to_compress->size);
   Data* compressed_data = data_create_empty(dst_size);
@@ -49,6 +57,26 @@ Data* data_compress(Data* data_to_compress, int compression_level) {
     return NULL;
   }
 
+  if (compression_threads > 0) {
+    zret = ZSTD_CCtx_setParameter(cctx, ZSTD_c_nbWorkers, compression_threads);
+    if (ZSTD_isError(zret)) {
+      log_message(LOG_LEVEL_ERROR, "Failed to set compression threads: %s",
+                  ZSTD_getErrorName(zret));
+      ZSTD_freeCCtx(cctx);
+      data_destroy(compressed_data);
+      return NULL;
+    }
+    /* Streaming compression needs the source size before threaded mode can end a frame. */
+    zret = ZSTD_CCtx_setPledgedSrcSize(cctx, data_to_compress->size);
+    if (ZSTD_isError(zret)) {
+      log_message(LOG_LEVEL_ERROR, "Failed to set compression source size: %s",
+                  ZSTD_getErrorName(zret));
+      ZSTD_freeCCtx(cctx);
+      data_destroy(compressed_data);
+      return NULL;
+    }
+  }
+
   ZSTD_inBuffer input = {data_to_compress->data, data_to_compress->size, 0};
   ZSTD_outBuffer output = {compressed_data->data, dst_size, 0};
 

src/shared/compression.h

@@ -5,6 +5,8 @@
 #include <stdbool.h>
 
 Data* data_compress(Data* data_to_compress, int compression_level);
+Data* data_compress_with_threads(Data* data_to_compress, int compression_level,
+                                 int compression_threads);
 Data* data_decompress(Data* compressed_data);
 Data* data_decompress_limited(Data* compressed_data, size_t maximum_size);
 bool compression_should_skip(const char* path);

src/shared/config.c

@@ -21,6 +21,7 @@ static void config_set_defaults(Config* config) {
   config->dry_run = false;
   config->use_delete = false;
   config->compression_level = 5;
+  config->compression_threads = 0;
   config->use_sendfile = false;
   config->chunk_size = DEFAULT_CHUNK_SIZE;
   config->ssh_port = 22;

src/shared/config.h

@@ -22,6 +22,7 @@ typedef struct Config {
   bool dry_run;
   bool use_delete;
   int compression_level;
+  int compression_threads;
   unsigned long long chunk_size;
   int ssh_port;
   TransportType transport;

src/shared/file_send.c

@@ -19,12 +19,20 @@
 
 bool file_send_single_calls(File* file, int file_descriptor, bool use_metadata,
                             int compression_level, bool send_path) {
+  return file_send_single_calls_with_threads(file, file_descriptor, use_metadata, compression_level,
+                                             0, send_path);
+}
+
+bool file_send_single_calls_with_threads(File* file, int file_descriptor, bool use_metadata,
+                                         int compression_level, int compression_threads,
+                                         bool send_path) {
   if (!file || !file->path || !file->data || (file->data->size != 0 && !file->data->data))
     return false;
   const Data* data_to_send = file->data;
   Data* compressed_data = NULL;
   if (compression_level > 0 && !compression_should_skip(file->path)) {
-    compressed_data = data_compress(file->data, compression_level);
+    compressed_data =
+        data_compress_with_threads(file->data, compression_level, compression_threads);
     if (compressed_data == NULL) {
       log_message(LOG_LEVEL_ERROR, "Failed to compress file data");
       return false;

src/shared/file_send.h

@@ -8,6 +8,9 @@
 
 bool file_send_single_calls(File* file, int file_descriptor, bool use_metadata,
                             int compression_level, bool send_path);
+bool file_send_single_calls_with_threads(File* file, int file_descriptor, bool use_metadata,
+                                         int compression_level, int compression_threads,
+                                         bool send_path);
 bool file_send_sendfile(File* file, int file_descriptor, bool use_metadata, int compression_level,
                         bool send_path);