[SCM] The rsync repository. - branch master updated
Rsync CVS commit messages
rsync-cvs at lists.samba.org
Mon May 25 20:46:56 UTC 2020
The branch, master has been updated
via c394e861 Include lz4 compression support.
via 4aaadc2f Include zstd compression support.
from abef92c0 Fix handling of a compressor that has no off_level.
https://git.samba.org/?p=rsync.git;a=shortlog;h=master
- Log -----------------------------------------------------------------
commit c394e861845682d23b6dd6b9114d763fc8c493a8
Author: Wayne Davison <wayne at opencoder.net>
Date: Mon May 25 13:43:25 2020 -0700
Include lz4 compression support.
Based on a patch that was emailed to me without a valid return address.
commit 4aaadc2f2970a22676df60be593c63fa37b49700
Author: Wayne Davison <wayne at opencoder.net>
Date: Mon May 25 13:31:30 2020 -0700
Include zstd compression support.
Based on a patch by Sebastian A. Siewior. Fixes bug #14338.
-----------------------------------------------------------------------
Summary of changes:
compat.c | 8 +
configure.ac | 26 +++-
token.c | 469 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
3 files changed, 502 insertions(+), 1 deletion(-)
Changeset truncated at 500 lines:
diff --git a/compat.c b/compat.c
index e89fac26..4bef8820 100644
--- a/compat.c
+++ b/compat.c
@@ -90,6 +90,14 @@ struct name_num_obj valid_compressions = {
"compress", NULL, NULL, 0, 0, {
{ CPRES_ZLIBX, "zlibx", NULL },
{ CPRES_ZLIB, "zlib", NULL },
+#ifdef SUPPORT_ZSTD
+ /* TODO decide where in the default preference order this should go. */
+ { CPRES_ZSTD, "zstd", NULL },
+#endif
+#ifdef SUPPORT_LZ4
+ /* TODO decide where in the default preference order this should go. */
+ { CPRES_LZ4, "lz4", NULL },
+#endif
{ CPRES_NONE, "none", NULL },
{ 0, NULL, NULL }
}
diff --git a/configure.ac b/configure.ac
index cbf93056..2b65ef9d 100644
--- a/configure.ac
+++ b/configure.ac
@@ -381,7 +381,7 @@ AC_CHECK_HEADERS(sys/fcntl.h sys/select.h fcntl.h sys/time.h sys/unistd.h \
netdb.h malloc.h float.h limits.h iconv.h libcharset.h langinfo.h \
sys/acl.h acl/libacl.h attr/xattr.h sys/xattr.h sys/extattr.h \
popt.h popt/popt.h linux/falloc.h netinet/in_systm.h netinet/ip.h \
- zlib.h xxhash.h openssl/md4.h openssl/md5.h)
+ zlib.h xxhash.h openssl/md4.h openssl/md5.h zstd.h lz4.h)
AC_HEADER_MAJOR_FIXED
AC_MSG_CHECKING([whether to enable use of openssl crypto library])
@@ -408,6 +408,30 @@ else
AC_MSG_RESULT(no)
fi
+AC_MSG_CHECKING([whether to enable zstd compression])
+AC_ARG_ENABLE([zstd],
+ AC_HELP_STRING([--disable-zstd], [disable zstd compression]))
+AH_TEMPLATE([SUPPORT_ZSTD],
+[Undefine if you do not want zstd compression. By default this is defined.])
+if test x"$enable_zstd" != x"no" && test x"$ac_cv_header_zstd_h" = x"yes"; then
+ AC_MSG_RESULT(yes)
+ AC_SEARCH_LIBS(ZSTD_minCLevel, zstd, [AC_DEFINE(SUPPORT_ZSTD)])
+else
+ AC_MSG_RESULT(no)
+fi
+
+AC_MSG_CHECKING([whether to enable LZ4 compression])
+AC_ARG_ENABLE([lz4],
+ AC_HELP_STRING([--disable-lz4], [disable LZ4 compression]))
+AH_TEMPLATE([SUPPORT_LZ4],
+[Undefine if you do not want LZ4 compression. By default this is defined.])
+if test x"$enable_lz4" != x"no" && test x"$ac_cv_header_lz4_h" = x"yes"; then
+ AC_MSG_RESULT(yes)
+ AC_SEARCH_LIBS(LZ4_compress, lz4, [AC_DEFINE(SUPPORT_LZ4)])
+else
+ AC_MSG_RESULT(no)
+fi
+
AC_CACHE_CHECK([if makedev takes 3 args],rsync_cv_MAKEDEV_TAKES_3_ARGS,[
AC_RUN_IFELSE([AC_LANG_SOURCE([[
#include <sys/types.h>
diff --git a/token.c b/token.c
index aa7ed42d..f92b9b5b 100644
--- a/token.c
+++ b/token.c
@@ -22,6 +22,12 @@
#include "rsync.h"
#include "itypes.h"
#include <zlib.h>
+#ifdef SUPPORT_ZSTD
+#include <zstd.h>
+#endif
+#ifdef SUPPORT_LZ4
+#include <lz4.h>
+#endif
extern int do_compression;
extern int protocol_version;
@@ -58,6 +64,22 @@ void init_compression_level(void)
if (do_compression_level == Z_DEFAULT_COMPRESSION)
do_compression_level = def_level;
break;
+#ifdef SUPPORT_ZSTD
+ case CPRES_ZSTD:
+ min_level = ZSTD_minCLevel();
+ max_level = ZSTD_maxCLevel();
+ def_level = 3;
+ off_level = CLVL_NOT_SPECIFIED;
+ break;
+#endif
+#ifdef SUPPORT_LZ4
+ case CPRES_LZ4:
+ min_level = 0;
+ max_level = 0;
+ def_level = 0;
+ off_level = CLVL_NOT_SPECIFIED;
+ break;
+#endif
default: /* paranoia to prevent missing case values */
exit_cleanup(RERR_UNSUPPORTED);
}
@@ -648,6 +670,431 @@ static void see_deflate_token(char *buf, int32 len)
} while (len || rx_strm.avail_out == 0);
}
+#ifdef SUPPORT_ZSTD
+
+static ZSTD_inBuffer zstd_in_buff;
+static ZSTD_outBuffer zstd_out_buff;
+static ZSTD_CCtx *zstd_cctx;
+
+static void send_zstd_token(int f, int32 token, struct map_struct *buf,
+ OFF_T offset, int32 nb)
+{
+ static int comp_init_done, flush_pending;
+ ZSTD_EndDirective flush = ZSTD_e_continue;
+ int32 n, r;
+
+ /* initialization */
+ if (!comp_init_done) {
+
+ zstd_cctx = ZSTD_createCCtx();
+ if (!zstd_cctx) {
+ rprintf(FERROR, "compression init failed\n");
+ exit_cleanup(RERR_PROTOCOL);
+ }
+
+ obuf = new_array(char, MAX_DATA_COUNT + 2);
+ if (!obuf)
+ out_of_memory("send_deflated_token");
+
+ ZSTD_CCtx_setParameter(zstd_cctx, ZSTD_c_compressionLevel,
+ do_compression_level);
+ zstd_out_buff.dst = obuf + 2;
+
+ comp_init_done = 1;
+ }
+
+ if (last_token == -1) {
+ last_run_end = 0;
+ run_start = token;
+ flush_pending = 0;
+ } else if (last_token == -2) {
+ run_start = token;
+
+ } else if (nb != 0 || token != last_token + 1
+ || token >= run_start + 65536) {
+
+ /* output previous run */
+ r = run_start - last_run_end;
+ n = last_token - run_start;
+
+ if (r >= 0 && r <= 63) {
+ write_byte(f, (n==0? TOKEN_REL: TOKENRUN_REL) + r);
+ } else {
+ write_byte(f, (n==0? TOKEN_LONG: TOKENRUN_LONG));
+ write_int(f, run_start);
+ }
+ if (n != 0) {
+ write_byte(f, n);
+ write_byte(f, n >> 8);
+ }
+ last_run_end = last_token;
+ run_start = token;
+ }
+
+ last_token = token;
+
+ if (nb || flush_pending) {
+
+ zstd_in_buff.src = map_ptr(buf, offset, nb);
+ zstd_in_buff.size = nb;
+ zstd_in_buff.pos = 0;
+
+ do {
+ if (zstd_out_buff.size == 0) {
+ zstd_out_buff.size = MAX_DATA_COUNT;
+ zstd_out_buff.pos = 0;
+ }
+
+ /* File ended, flush */
+ if (token != -2)
+ flush = ZSTD_e_flush;
+
+ r = ZSTD_compressStream2(zstd_cctx, &zstd_out_buff, &zstd_in_buff, flush);
+ if (ZSTD_isError(r)) {
+ rprintf(FERROR, "ZSTD_compressStream returned %d\n", r);
+ exit_cleanup(RERR_STREAMIO);
+ }
+
+ /*
+ * Nothing is sent if the buffer isn't full so avoid smaller
+ * transfers. If a file is finished then we flush the internal
+ * 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) {
+ 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);
+ flush_pending = token == -2;
+ }
+
+ if (token == -1) {
+ /* end of file - clean up */
+ write_byte(f, END_FLAG);
+ }
+}
+
+static ZSTD_DCtx *zstd_dctx;
+
+static int32 recv_zstd_token(int f, char **data)
+{
+ static int decomp_init_done;
+ static int out_buffer_size;
+ int32 n, flag;
+ int r;
+
+ if (!decomp_init_done) {
+
+ zstd_dctx = ZSTD_createDCtx();
+ if (!zstd_dctx) {
+ rprintf(FERROR, "ZSTD_createDStream failed\n");
+ exit_cleanup(RERR_PROTOCOL);
+ }
+
+ /* Output buffer fits two decompressed blocks */
+ out_buffer_size = ZSTD_DStreamOutSize() * 2;
+ cbuf = new_array(char, MAX_DATA_COUNT);
+ dbuf = new_array(char, out_buffer_size);
+ if (!cbuf || !dbuf)
+ out_of_memory("recv_zstd_token");
+
+ zstd_in_buff.src = cbuf;
+ zstd_out_buff.dst = dbuf;
+
+ decomp_init_done = 1;
+ }
+
+ do {
+ switch (recv_state) {
+ case r_init:
+ recv_state = r_idle;
+ rx_token = 0;
+ break;
+
+ case r_idle:
+ flag = read_byte(f);
+ if ((flag & 0xC0) == DEFLATED_DATA) {
+ n = ((flag & 0x3f) << 8) + read_byte(f);
+ read_buf(f, cbuf, n);
+
+ zstd_in_buff.size = n;
+ zstd_in_buff.pos = 0;
+
+ recv_state = r_inflating;
+
+ } else if (flag == END_FLAG) {
+ /* that's all folks */
+ recv_state = r_init;
+ return 0;
+
+ } else {
+ /* here we have a token of some kind */
+ if (flag & TOKEN_REL) {
+ rx_token += flag & 0x3f;
+ flag >>= 6;
+ } else
+ rx_token = read_int(f);
+ if (flag & 1) {
+ rx_run = read_byte(f);
+ rx_run += read_byte(f) << 8;
+ recv_state = r_running;
+ }
+ return -1 - rx_token;
+ }
+ break;
+
+ case r_inflating:
+ zstd_out_buff.size = out_buffer_size;
+ zstd_out_buff.pos = 0;
+
+ r = ZSTD_decompressStream(zstd_dctx, &zstd_out_buff, &zstd_in_buff);
+ n = zstd_out_buff.pos;
+ if (ZSTD_isError(r)) {
+ rprintf(FERROR, "ZSTD decomp returned %d (%d bytes)\n", r, n);
+ exit_cleanup(RERR_STREAMIO);
+ }
+
+ /*
+ * If the input buffer is fully consumed and the output
+ * buffer is not full then next step is to read more
+ * data.
+ */
+ if (zstd_in_buff.size == zstd_in_buff.pos && n < out_buffer_size)
+ recv_state = r_idle;
+
+ if (n != 0) {
+ *data = dbuf;
+ return n;
+ }
+ break;
+
+ case r_running:
+ ++rx_token;
+ if (--rx_run == 0)
+ recv_state = r_idle;
+ return -1 - rx_token;
+ break;
+
+ case r_inflated:
+ break;
+ }
+ } while (1);
+}
+#endif /* SUPPORT_ZSTD */
+
+#ifdef SUPPORT_LZ4
+static void
+send_compressed_token(int f, int32 token, struct map_struct *buf, OFF_T offset, int32 nb)
+{
+ static int init_done, flush_pending;
+ int size = MAX(LZ4_compressBound(CHUNK_SIZE), MAX_DATA_COUNT+2);
+ int32 n, r;
+
+ if (last_token == -1) {
+ if (!init_done) {
+ if ((obuf = new_array(char, size)) == NULL)
+ out_of_memory("send_compressed_token");
+ init_done = 1;
+ }
+ last_run_end = 0;
+ run_start = token;
+ flush_pending = 0;
+ } else if (last_token == -2) {
+ run_start = token;
+ } else if (nb != 0 || token != last_token + 1
+ || token >= run_start + 65536) {
+ /* output previous run */
+ r = run_start - last_run_end;
+ n = last_token - run_start;
+ if (r >= 0 && r <= 63) {
+ write_byte(f, (n==0? TOKEN_REL: TOKENRUN_REL) + r);
+ } else {
+ write_byte(f, (n==0? TOKEN_LONG: TOKENRUN_LONG));
+ write_int(f, run_start);
+ }
+ if (n != 0) {
+ write_byte(f, n);
+ write_byte(f, n >> 8);
+ }
+ last_run_end = last_token;
+ run_start = token;
+ }
+
+ last_token = token;
+
+ if (nb != 0 || flush_pending) {
+ int available_in, available_out = 0;
+ const char *next_in;
+
+ do {
+ char *ptr = obuf;
+ char *next_out = obuf + 2;
+
+ if (available_out == 0) {
+ available_in = MIN(nb, MAX_DATA_COUNT);
+ next_in = map_ptr(buf, offset, available_in);
+ } else
+ available_in /= 2;
+
+ available_out = LZ4_compress(next_in, next_out, available_in);
+ if (!available_out) {
+ rprintf(FERROR, "compress returned %d\n", available_out);
+ exit_cleanup(RERR_STREAMIO);
+ }
+ if (available_out <= MAX_DATA_COUNT) {
+ ptr[0] = DEFLATED_DATA + (available_out >> 8);
+ ptr[1] = available_out;
+
+ write_buf(f, ptr, available_out + 2);
+
+ available_out = 0;
+ nb -= available_in;
+ offset += available_in;
+ }
+ } while (nb != 0);
+ flush_pending = token == -2;
+ }
+ if (token == -1)
+ /* end of file - clean up */
+ write_byte(f, END_FLAG);
+}
+
+static int32 recv_compressed_token(int f, char **data)
+{
+ static int32 saved_flag;
+ static int init_done;
+ int32 n, flag;
+ int size = MAX(LZ4_compressBound(CHUNK_SIZE), MAX_DATA_COUNT+2);
+ static const char *next_in;
+ static int avail_in;
+ int avail_out;
+
+ for (;;) {
+ switch (recv_state) {
+ case r_init:
+ if (!init_done) {
+ if (!(cbuf = new_array(char, MAX_DATA_COUNT))
+ || !(dbuf = new_array(char, size)))
+ out_of_memory("recv_compressed_token");
+ init_done = 1;
+ }
+ recv_state = r_idle;
+ rx_token = 0;
+ break;
+ case r_idle:
+ case r_inflated:
+ if (saved_flag) {
+ flag = saved_flag & 0xff;
+ saved_flag = 0;
+ } else
+ flag = read_byte(f);
+ if ((flag & 0xC0) == DEFLATED_DATA) {
+ n = ((flag & 0x3f) << 8) + read_byte(f);
+ read_buf(f, cbuf, n);
+ next_in = (char *)cbuf;
+ avail_in = n;
+ recv_state = r_inflating;
+ break;
+ }
+
+ if (recv_state == r_inflated)
+ recv_state = r_idle;
+
+ if (flag == END_FLAG) {
+ /* that's all folks */
+ recv_state = r_init;
+ return 0;
+ }
+
+ /* here we have a token of some kind */
+ if (flag & TOKEN_REL) {
+ rx_token += flag & 0x3f;
+ flag >>= 6;
+ } else
+ rx_token = read_int(f);
+ if (flag & 1) {
+ rx_run = read_byte(f);
+ rx_run += read_byte(f) << 8;
+ recv_state = r_running;
+ }
+ return -1 - rx_token;
+
+ case r_inflating:
+ avail_out = LZ4_decompress_safe(next_in, dbuf, avail_in, size);
+ if (avail_out < 0) {
+ rprintf(FERROR, "uncompress failed: %d\n", avail_out);
+ exit_cleanup(RERR_STREAMIO);
+ }
+ recv_state = r_inflated;
+ *data = dbuf;
+ return avail_out;
+
+ case r_running:
+ ++rx_token;
+ if (--rx_run == 0)
+ recv_state = r_idle;
+ return -1 - rx_token;
+ }
+ }
+
+}
+
+# if 0
+static void see_uncompressed_token(char *buf, int32 len)
+{
+ static const char *next_in;
+ static int avail_in;
+ int avail_out;
+
+ int32 blklen;
+ char hdr[5];
+
+ avail_in = 0;
+ blklen = 0;
+ hdr[0] = 0;
+ do {
--
The rsync repository.
More information about the rsync-cvs
mailing list