feat: add threads to zstd compression

compat.c

@@ -52,6 +52,7 @@ extern int need_messages_from_generator;
 extern int delete_mode, delete_before, delete_during, delete_after;
 extern int do_compression;
 extern int do_compression_level;
+extern int do_compression_threads;
 extern int saw_stderr_opt;
 extern int msgs2stderr;
 extern char *shell_cmd;

options.c

@@ -86,6 +86,7 @@ int sparse_files = 0;
 int preallocate_files = 0;
 int do_compression = 0;
 int do_compression_level = CLVL_NOT_SPECIFIED;
+int do_compression_threads = 0; /*n = 0 use rsync thread, n >= 1 spawn n threads for compression */
 int am_root = 0; /* 0 = normal, 1 = root, 2 = --super, -1 = --fake-super */
 int am_server = 0;
 int am_sender = 0;
@@ -756,6 +757,8 @@ static struct poptOption long_options[] = {
   {"skip-compress",    0,  POPT_ARG_STRING, &skip_compress, 0, 0, 0 },
   {"compress-level",   0,  POPT_ARG_INT,    &do_compression_level, 0, 0, 0 },
   {"zl",               0,  POPT_ARG_INT,    &do_compression_level, 0, 0, 0 },
+  {"compress-threads", 0,  POPT_ARG_INT,    &do_compression_threads, 0, 0, 0 },
+  {"zt",               0,  POPT_ARG_INT,    &do_compression_threads, 0, 0, 0 },
   {0,                 'P', POPT_ARG_NONE,   0, 'P', 0, 0 },
   {"progress",         0,  POPT_ARG_VAL,    &do_progress, 1, 0, 0 },
   {"no-progress",      0,  POPT_ARG_VAL,    &do_progress, 0, 0, 0 },
@@ -2006,6 +2009,8 @@ int parse_arguments(int *argc_p, const char ***argv_p)
 			create_refuse_error(refused_compress);
 			goto cleanup;
 		}
+		if (do_compression_threads < 0)
+			do_compression_threads = 0;
 	}
 
 #ifdef HAVE_SETVBUF

rsync.1.md

@@ -513,6 +513,7 @@ has its own detailed description later in this manpage.
 --compress, -z           compress file data during the transfer
 --compress-choice=STR    choose the compression algorithm (aka --zc)
 --compress-level=NUM     explicitly set compression level (aka --zl)
+--compress-threads=NUM   explicitly set compression threads (aka --zt)
 --skip-compress=LIST     skip compressing files with suffix in LIST
 --cvs-exclude, -C        auto-ignore files in the same way CVS does
 --filter=RULE, -f        add a file-filtering RULE
@@ -2817,6 +2818,22 @@ expand it.
     report something like "`Client compress: zstd (level 3)`" (along with the
     checksum choice in effect).
 
+0.  `--compress-threads=NUM`, `--zt=NUM`
+
+    Set the number of threads to spawn when compressing data. Setting this 
+    option to 1 or more will instruct the compression library to spawn 1 or 
+    more threads for compression. Ideally, increasing the number of threads 
+    will increase transfer speed if the transfer is CPU bound on the sender.
+
+    This option does not affect decompression. 
+
+    Compression algorithms that allow threading:
+
+    - `zstd` (only when libzstd is compiled with threading support)
+
+    This option is ignored if one of the above alogithms is not selected as the
+     `--compression-choice` or if compression not enabled.
+
 0.  `--skip-compress=LIST`
 
     **NOTE:** no compression method currently supports per-file compression

support/rrsync

@@ -46,6 +46,7 @@ long_opts = {
   'compare-dest': 2,
   'compress-choice': 1,
   'compress-level': 1,
+  'compress-threads': 1,
   'copy-dest': 2,
   'copy-devices': -1,
   'copy-unsafe-links': 0,

token.c

@@ -33,6 +33,7 @@ extern int do_compression;
 extern int protocol_version;
 extern int module_id;
 extern int do_compression_level;
+extern int do_compression_threads;
 extern char *skip_compress;
 
 #ifndef Z_INSERT_ONLY
@@ -692,6 +693,8 @@ static void send_zstd_token(int f, int32 token, struct map_struct *buf, OFF_T of
 		obuf = new_array(char, OBUF_SIZE);
 
 		ZSTD_CCtx_setParameter(zstd_cctx, ZSTD_c_compressionLevel, do_compression_level);
+		ZSTD_CCtx_setParameter(zstd_cctx, ZSTD_c_nbWorkers, do_compression_threads);
+
 		zstd_out_buff.dst = obuf + 2;
 
 		comp_init_done = 1;
@@ -729,12 +732,11 @@ static void send_zstd_token(int f, int32 token, struct map_struct *buf, OFF_T of
 		zstd_in_buff.src = map_ptr(buf, offset, nb);
 		zstd_in_buff.size = nb;
 		zstd_in_buff.pos = 0;
-
+
+		int finished;
 		do {
-			if (zstd_out_buff.size == 0) {
-				zstd_out_buff.size = MAX_DATA_COUNT;
-				zstd_out_buff.pos = 0;
-			}
+			zstd_out_buff.size = MAX_DATA_COUNT;
+			zstd_out_buff.pos = 0;
 
 			/* File ended, flush */
 			if (token != -2)
@@ -752,20 +754,21 @@ static void send_zstd_token(int f, int32 token, struct map_struct *buf, OFF_T of
 			 * state and send a smaller buffer so that the remote side can
 			 * finish the file.
 			 */
-			if (zstd_out_buff.pos == zstd_out_buff.size || flush == ZSTD_e_flush) {
+			finished = (flush == ZSTD_e_flush) ? (r == 0) : (zstd_in_buff.pos == zstd_in_buff.size);
+
+			if (zstd_out_buff.pos != 0) {
 				n = zstd_out_buff.pos;
 
 				obuf[0] = DEFLATED_DATA + (n >> 8);
 				obuf[1] = n;
 				write_buf(f, obuf, n+2);
-
-				zstd_out_buff.size = 0;
 			}
 			/*
 			 * Loop while the input buffer isn't full consumed or the
 			 * internal state isn't fully flushed.
 			 */
-		} while (zstd_in_buff.pos < zstd_in_buff.size || r > 0);
+		} while (!finished);
+
 		flush_pending = token == -2;
 	}