Merge remote-tracking branch 'origin/feat/compress-threads' into dev

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) {
-  log_debug_message(LOG_DEBUG_PACK, "Starting to compress chunk");
+  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

@@ -6,6 +6,7 @@
 #include <stdint.h>
 #include <string.h>
 #include <strings.h>
+#include <unistd.h>
 #include <zstd.h>
 
 #define INITIAL_DECOMPRESS_BUF_SIZE (1024 * 1024)
@@ -38,7 +39,15 @@ bool compression_should_skip_with_suffixes(const char* path, char* const* suffix
 }
 
 Data* data_compress(Data* data_to_compress, int compression_level) {
-  log_debug_message(LOG_DEBUG_UTIL, "Starting to compress data");
+  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 || compression_threads > COMPRESSION_MAX_THREADS)
+    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);
   if (compressed_data == NULL)
@@ -59,6 +68,30 @@ Data* data_compress(Data* data_to_compress, int compression_level) {
     return NULL;
   }
 
+  if (compression_threads > 0) {
+    long online_cpus = sysconf(_SC_NPROCESSORS_ONLN);
+    int available_threads = online_cpus > 0 && online_cpus < compression_threads
+                                ? (int)online_cpus
+                                : compression_threads;
+    zret = ZSTD_CCtx_setParameter(cctx, ZSTD_c_nbWorkers, available_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

@@ -4,7 +4,11 @@
 #include "data.h"
 #include <stdbool.h>
 
+#define COMPRESSION_MAX_THREADS 64
+
 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

@@ -24,6 +24,7 @@ static void config_set_defaults(Config* config) {
   config->remove_source_files = 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

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

src/shared/file_send.c

@@ -20,19 +20,21 @@
 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_skip(file, file_descriptor, use_metadata, compression_level,
-                                          send_path, NULL, -1);
+                                          send_path, NULL, -1, 0);
 }
 
 bool file_send_single_calls_with_skip(File* file, int file_descriptor, bool use_metadata,
                                       int compression_level, bool send_path,
-                                      char* const* skip_suffixes, int skip_count) {
+                                      char* const* skip_suffixes, int skip_count,
+                                      int compression_threads) {
   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_with_suffixes(file->path, skip_suffixes, skip_count)) {
-    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;
@@ -58,17 +60,18 @@ bool file_send_single_calls_with_skip(File* file, int file_descriptor, bool use_
 bool file_send_sendfile(File* file, int file_descriptor, bool use_metadata, int compression_level,
                         bool send_path) {
   return file_send_sendfile_with_skip(file, file_descriptor, use_metadata, compression_level,
-                                      send_path, NULL, -1);
+                                      send_path, NULL, -1, 0);
 }
 
 bool file_send_sendfile_with_skip(File* file, int file_descriptor, bool use_metadata,
                                   int compression_level, bool send_path, char* const* skip_suffixes,
-                                  int skip_count) {
+                                  int skip_count, int compression_threads) {
   if (!file || !file->path || !file->data)
     return false;
   if (compression_level > 0)
     return file_send_single_calls_with_skip(file, file_descriptor, use_metadata, compression_level,
-                                            send_path, skip_suffixes, skip_count);
+                                            send_path, skip_suffixes, skip_count,
+                                            compression_threads);
 
   if (send_path && !send_str(file_descriptor, file->path))
     return false;

src/shared/file_send.h

@@ -10,11 +10,12 @@ 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_skip(File* file, int file_descriptor, bool use_metadata,
                                       int compression_level, bool send_path,
-                                      char* const* skip_suffixes, int skip_count);
+                                      char* const* skip_suffixes, int skip_count,
+                                      int compression_threads);
 bool file_send_sendfile(File* file, int file_descriptor, bool use_metadata, int compression_level,
                         bool send_path);
 bool file_send_sendfile_with_skip(File* file, int file_descriptor, bool use_metadata,
                                   int compression_level, bool send_path, char* const* skip_suffixes,
-                                  int skip_count);
+                                  int skip_count, int compression_threads);
 
 #endif