?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
Fix overzealous setting of mtime & tweak time comparisons
[rsync.git] / token.c
blobebecd8abc701fac0cbfdd70097d791af7ee00a5e
1 /*
2 * Routines used by the file-transfer code.
3 *
4 * Copyright (C) 1996 Andrew Tridgell
5 * Copyright (C) 1996 Paul Mackerras
6 * Copyright (C) 2003-2020 Wayne Davison
7 *
8 * This program is free software; you can redistribute it and/or modify
9 * it under the terms of the GNU General Public License as published by
10 * the Free Software Foundation; either version 3 of the License, or
11 * (at your option) any later version.
12 *
13 * This program is distributed in the hope that it will be useful,
14 * but WITHOUT ANY WARRANTY; without even the implied warranty of
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 * GNU General Public License for more details.
17 *
18 * You should have received a copy of the GNU General Public License along
19 * with this program; if not, visit the http://fsf.org website.
20 */
21
22 #include "rsync.h"
23 #include "itypes.h"
24 #include <zlib.h>
25 #ifdef SUPPORT_ZSTD
26 #include <zstd.h>
27 #endif
28 #ifdef SUPPORT_LZ4
29 #include <lz4.h>
30 #endif
31
32 extern int do_compression;
33 extern int protocol_version;
34 extern int module_id;
35 extern int do_compression_level;
36 extern char *skip_compress;
37
38 #ifndef Z_INSERT_ONLY
39 #define Z_INSERT_ONLY Z_SYNC_FLUSH
40 #endif
41
42 static int compression_level; /* The compression level for the current file. */
43 static int skip_compression_level; /* The least possible compressing for handling skip-compress files. */
44 static int per_file_default_level; /* The default level that each new file gets prior to checking its suffix. */
45
46 struct suffix_tree {
47 struct suffix_tree *sibling;
48 struct suffix_tree *child;
49 char letter, word_end;
50 };
51
52 static char *match_list;
53 static struct suffix_tree *suftree;
54
55 void init_compression_level(void)
56 {
57 int min_level, max_level, def_level, off_level;
58
59 switch (do_compression) {
60 case CPRES_NONE:
61 return;
62 case CPRES_ZLIB:
63 case CPRES_ZLIBX:
64 min_level = 1;
65 max_level = Z_BEST_COMPRESSION;
66 def_level = 6; /* Z_DEFAULT_COMPRESSION is -1, so set it to the real default */
67 off_level = skip_compression_level = Z_NO_COMPRESSION;
68 if (do_compression_level == Z_DEFAULT_COMPRESSION)
69 do_compression_level = def_level;
70 break;
71 #ifdef SUPPORT_ZSTD
72 case CPRES_ZSTD:
73 min_level = skip_compression_level = ZSTD_minCLevel();
74 max_level = ZSTD_maxCLevel();
75 def_level = ZSTD_CLEVEL_DEFAULT;
76 off_level = CLVL_NOT_SPECIFIED;
77 if (do_compression_level == 0)
78 do_compression_level = def_level;
79 break;
80 #endif
81 #ifdef SUPPORT_LZ4
82 case CPRES_LZ4:
83 min_level = skip_compression_level = 0;
84 max_level = 0;
85 def_level = 0;
86 off_level = CLVL_NOT_SPECIFIED;
87 break;
88 #endif
89 default: /* paranoia to prevent missing case values */
90 assert(0);
91 }
92
93 if (do_compression_level == CLVL_NOT_SPECIFIED)
94 do_compression_level = def_level;
95 else if (do_compression_level == off_level) {
96 do_compression = CPRES_NONE;
97 return;
98 }
99
100 /* We don't bother with any errors or warnings -- just make sure that the values are valid. */
101 if (do_compression_level < min_level)
102 do_compression_level = min_level;
103 else if (do_compression_level > max_level)
104 do_compression_level = max_level;
105 }
106
107 static void add_suffix(struct suffix_tree **prior, char ltr, const char *str)
108 {
109 struct suffix_tree *node, *newnode;
110
111 if (ltr == '[') {
112 const char *after = strchr(str, ']');
113 /* Treat "[foo" and "[]" as having a literal '['. */
114 if (after && after++ != str+1) {
115 while ((ltr = *str++) != ']')
116 add_suffix(prior, ltr, after);
117 return;
118 }
119 }
120
121 for (node = *prior; node; prior = &node->sibling, node = node->sibling) {
122 if (node->letter == ltr) {
123 if (*str)
124 add_suffix(&node->child, *str, str+1);
125 else
126 node->word_end = 1;
127 return;
128 }
129 if (node->letter > ltr)
130 break;
131 }
132 if (!(newnode = new(struct suffix_tree)))
133 out_of_memory("add_suffix");
134 newnode->sibling = node;
135 newnode->child = NULL;
136 newnode->letter = ltr;
137 *prior = newnode;
138 if (*str) {
139 add_suffix(&newnode->child, *str, str+1);
140 newnode->word_end = 0;
141 } else
142 newnode->word_end = 1;
143 }
144
145 static void add_nocompress_suffixes(const char *str)
146 {
147 char *buf, *t;
148 const char *f = str;
149
150 if (!(buf = new_array(char, strlen(f) + 1)))
151 out_of_memory("add_nocompress_suffixes");
152
153 while (*f) {
154 if (*f == '/') {
155 f++;
156 continue;
157 }
158
159 t = buf;
160 do {
161 if (isUpper(f))
162 *t++ = toLower(f);
163 else
164 *t++ = *f;
165 } while (*++f != '/' && *f);
166 *t++ = '\0';
167
168 add_suffix(&suftree, *buf, buf+1);
169 }
170
171 free(buf);
172 }
173
174 static void init_set_compression(void)
175 {
176 const char *f;
177 char *t, *start;
178
179 if (skip_compress)
180 add_nocompress_suffixes(skip_compress);
181
182 /* A non-daemon transfer skips the default suffix list if the
183 * user specified --skip-compress. */
184 if (skip_compress && module_id < 0)
185 f = "";
186 else
187 f = lp_dont_compress(module_id);
188
189 if (!(match_list = t = new_array(char, strlen(f) + 2)))
190 out_of_memory("set_compression");
191
192 per_file_default_level = do_compression_level;
193
194 while (*f) {
195 if (*f == ' ') {
196 f++;
197 continue;
198 }
199
200 start = t;
201 do {
202 if (isUpper(f))
203 *t++ = toLower(f);
204 else
205 *t++ = *f;
206 } while (*++f != ' ' && *f);
207 *t++ = '\0';
208
209 if (t - start == 1+1 && *start == '*') {
210 /* Optimize a match-string of "*". */
211 *match_list = '\0';
212 suftree = NULL;
213 per_file_default_level = skip_compression_level;
214 break;
215 }
216
217 /* Move *.foo items into the stuffix tree. */
218 if (*start == '*' && start[1] == '.' && start[2]
219 && !strpbrk(start+2, ".?*")) {
220 add_suffix(&suftree, start[2], start+3);
221 t = start;
222 }
223 }
224 *t++ = '\0';
225 }
226
227 /* determine the compression level based on a wildcard filename list */
228 void set_compression(const char *fname)
229 {
230 const struct suffix_tree *node;
231 const char *s;
232 char ltr;
233
234 if (!do_compression)
235 return;
236
237 if (!match_list)
238 init_set_compression();
239
240 compression_level = per_file_default_level;
241
242 if (!*match_list && !suftree)
243 return;
244
245 if ((s = strrchr(fname, '/')) != NULL)
246 fname = s + 1;
247
248 for (s = match_list; *s; s += strlen(s) + 1) {
249 if (iwildmatch(s, fname)) {
250 compression_level = skip_compression_level;
251 return;
252 }
253 }
254
255 if (!(node = suftree) || !(s = strrchr(fname, '.'))
256 || s == fname || !(ltr = *++s))
257 return;
258
259 while (1) {
260 if (isUpper(&ltr))
261 ltr = toLower(&ltr);
262 while (node->letter != ltr) {
263 if (node->letter > ltr)
264 return;
265 if (!(node = node->sibling))
266 return;
267 }
268 if ((ltr = *++s) == '\0') {
269 if (node->word_end)
270 compression_level = skip_compression_level;
271 return;
272 }
273 if (!(node = node->child))
274 return;
275 }
276 }
277
278 /* non-compressing recv token */
279 static int32 simple_recv_token(int f, char **data)
280 {
281 static int32 residue;
282 static char *buf;
283 int32 n;
284
285 if (!buf) {
286 buf = new_array(char, CHUNK_SIZE);
287 if (!buf)
288 out_of_memory("simple_recv_token");
289 }
290
291 if (residue == 0) {
292 int32 i = read_int(f);
293 if (i <= 0)
294 return i;
295 residue = i;
296 }
297
298 *data = buf;
299 n = MIN(CHUNK_SIZE,residue);
300 residue -= n;
301 read_buf(f,buf,n);
302 return n;
303 }
304
305 /* non-compressing send token */
306 static void simple_send_token(int f, int32 token, struct map_struct *buf,
307 OFF_T offset, int32 n)
308 {
309 if (n > 0) {
310 int32 len = 0;
311 while (len < n) {
312 int32 n1 = MIN(CHUNK_SIZE, n-len);
313 write_int(f, n1);
314 write_buf(f, map_ptr(buf, offset+len, n1), n1);
315 len += n1;
316 }
317 }
318 /* a -2 token means to send data only and no token */
319 if (token != -2)
320 write_int(f, -(token+1));
321 }
322
323 /* Flag bytes in compressed stream are encoded as follows: */
324 #define END_FLAG 0 /* that's all folks */
325 #define TOKEN_LONG 0x20 /* followed by 32-bit token number */
326 #define TOKENRUN_LONG 0x21 /* ditto with 16-bit run count */
327 #define DEFLATED_DATA 0x40 /* + 6-bit high len, then low len byte */
328 #define TOKEN_REL 0x80 /* + 6-bit relative token number */
329 #define TOKENRUN_REL 0xc0 /* ditto with 16-bit run count */
330
331 #define MAX_DATA_COUNT 16383 /* fit 14 bit count into 2 bytes with flags */
332
333 /* zlib.h says that if we want to be able to compress something in a single
334 * call, avail_out must be at least 0.1% larger than avail_in plus 12 bytes.
335 * We'll add in 0.1%+16, just to be safe (and we'll avoid floating point,
336 * to ensure that this is a compile-time value). */
337 #define AVAIL_OUT_SIZE(avail_in_size) ((avail_in_size)*1001/1000+16)
338
339 /* For coding runs of tokens */
340 static int32 last_token = -1;
341 static int32 run_start;
342 static int32 last_run_end;
343
344 /* Deflation state */
345 static z_stream tx_strm;
346
347 /* Output buffer */
348 static char *obuf;
349
350 /* We want obuf to be able to hold both MAX_DATA_COUNT+2 bytes as well as
351 * AVAIL_OUT_SIZE(CHUNK_SIZE) bytes, so make sure that it's large enough. */
352 #if MAX_DATA_COUNT+2 > AVAIL_OUT_SIZE(CHUNK_SIZE)
353 #define OBUF_SIZE (MAX_DATA_COUNT+2)
354 #else
355 #define OBUF_SIZE AVAIL_OUT_SIZE(CHUNK_SIZE)
356 #endif
357
358 /* Send a deflated token */
359 static void
360 send_deflated_token(int f, int32 token, struct map_struct *buf, OFF_T offset,
361 int32 nb, int32 toklen)
362 {
363 static int init_done, flush_pending;
364 int32 n, r;
365
366 if (last_token == -1) {
367 /* initialization */
368 if (!init_done) {
369 tx_strm.next_in = NULL;
370 tx_strm.zalloc = NULL;
371 tx_strm.zfree = NULL;
372 if (deflateInit2(&tx_strm, compression_level,
373 Z_DEFLATED, -15, 8,
374 Z_DEFAULT_STRATEGY) != Z_OK) {
375 rprintf(FERROR, "compression init failed\n");
376 exit_cleanup(RERR_PROTOCOL);
377 }
378 if ((obuf = new_array(char, OBUF_SIZE)) == NULL)
379 out_of_memory("send_deflated_token");
380 init_done = 1;
381 } else
382 deflateReset(&tx_strm);
383 last_run_end = 0;
384 run_start = token;
385 flush_pending = 0;
386 } else if (last_token == -2) {
387 run_start = token;
388 } else if (nb != 0 || token != last_token + 1
389 || token >= run_start + 65536) {
390 /* output previous run */
391 r = run_start - last_run_end;
392 n = last_token - run_start;
393 if (r >= 0 && r <= 63) {
394 write_byte(f, (n==0? TOKEN_REL: TOKENRUN_REL) + r);
395 } else {
396 write_byte(f, (n==0? TOKEN_LONG: TOKENRUN_LONG));
397 write_int(f, run_start);
398 }
399 if (n != 0) {
400 write_byte(f, n);
401 write_byte(f, n >> 8);
402 }
403 last_run_end = last_token;
404 run_start = token;
405 }
406
407 last_token = token;
408
409 if (nb != 0 || flush_pending) {
410 /* deflate the data starting at offset */
411 int flush = Z_NO_FLUSH;
412 tx_strm.avail_in = 0;
413 tx_strm.avail_out = 0;
414 do {
415 if (tx_strm.avail_in == 0 && nb != 0) {
416 /* give it some more input */
417 n = MIN(nb, CHUNK_SIZE);
418 tx_strm.next_in = (Bytef *)
419 map_ptr(buf, offset, n);
420 tx_strm.avail_in = n;
421 nb -= n;
422 offset += n;
423 }
424 if (tx_strm.avail_out == 0) {
425 tx_strm.next_out = (Bytef *)(obuf + 2);
426 tx_strm.avail_out = MAX_DATA_COUNT;
427 if (flush != Z_NO_FLUSH) {
428 /*
429 * We left the last 4 bytes in the
430 * buffer, in case they are the
431 * last 4. Move them to the front.
432 */
433 memcpy(tx_strm.next_out,
434 obuf+MAX_DATA_COUNT-2, 4);
435 tx_strm.next_out += 4;
436 tx_strm.avail_out -= 4;
437 }
438 }
439 if (nb == 0 && token != -2)
440 flush = Z_SYNC_FLUSH;
441 r = deflate(&tx_strm, flush);
442 if (r != Z_OK) {
443 rprintf(FERROR, "deflate returned %d\n", r);
444 exit_cleanup(RERR_STREAMIO);
445 }
446 if (nb == 0 || tx_strm.avail_out == 0) {
447 n = MAX_DATA_COUNT - tx_strm.avail_out;
448 if (flush != Z_NO_FLUSH) {
449 /*
450 * We have to trim off the last 4
451 * bytes of output when flushing
452 * (they are just 0, 0, ff, ff).
453 */
454 n -= 4;
455 }
456 if (n > 0) {
457 obuf[0] = DEFLATED_DATA + (n >> 8);
458 obuf[1] = n;
459 write_buf(f, obuf, n+2);
460 }
461 }
462 } while (nb != 0 || tx_strm.avail_out == 0);
463 flush_pending = token == -2;
464 }
465
466 if (token == -1) {
467 /* end of file - clean up */
468 write_byte(f, END_FLAG);
469 } else if (token != -2 && do_compression == CPRES_ZLIB) {
470 /* Add the data in the current block to the compressor's
471 * history and hash table. */
472 do {
473 /* Break up long sections in the same way that
474 * see_deflate_token() does. */
475 int32 n1 = toklen > 0xffff ? 0xffff : toklen;
476 toklen -= n1;
477 tx_strm.next_in = (Bytef *)map_ptr(buf, offset, n1);
478 tx_strm.avail_in = n1;
479 if (protocol_version >= 31) /* Newer protocols avoid a data-duplicating bug */
480 offset += n1;
481 tx_strm.next_out = (Bytef *) obuf;
482 tx_strm.avail_out = AVAIL_OUT_SIZE(CHUNK_SIZE);
483 r = deflate(&tx_strm, Z_INSERT_ONLY);
484 if (r != Z_OK || tx_strm.avail_in != 0) {
485 rprintf(FERROR, "deflate on token returned %d (%d bytes left)\n",
486 r, tx_strm.avail_in);
487 exit_cleanup(RERR_STREAMIO);
488 }
489 } while (toklen > 0);
490 }
491 }
492
493 /* tells us what the receiver is in the middle of doing */
494 static enum { r_init, r_idle, r_running, r_inflating, r_inflated } recv_state;
495
496 /* for inflating stuff */
497 static z_stream rx_strm;
498 static char *cbuf;
499 static char *dbuf;
500
501 /* for decoding runs of tokens */
502 static int32 rx_token;
503 static int32 rx_run;
504
505 /* Receive a deflated token and inflate it */
506 static int32 recv_deflated_token(int f, char **data)
507 {
508 static int init_done;
509 static int32 saved_flag;
510 int32 n, flag;
511 int r;
512
513 for (;;) {
514 switch (recv_state) {
515 case r_init:
516 if (!init_done) {
517 rx_strm.next_out = NULL;
518 rx_strm.zalloc = NULL;
519 rx_strm.zfree = NULL;
520 if (inflateInit2(&rx_strm, -15) != Z_OK) {
521 rprintf(FERROR, "inflate init failed\n");
522 exit_cleanup(RERR_PROTOCOL);
523 }
524 if (!(cbuf = new_array(char, MAX_DATA_COUNT))
525 || !(dbuf = new_array(char, AVAIL_OUT_SIZE(CHUNK_SIZE))))
526 out_of_memory("recv_deflated_token");
527 init_done = 1;
528 } else {
529 inflateReset(&rx_strm);
530 }
531 recv_state = r_idle;
532 rx_token = 0;
533 break;
534
535 case r_idle:
536 case r_inflated:
537 if (saved_flag) {
538 flag = saved_flag & 0xff;
539 saved_flag = 0;
540 } else
541 flag = read_byte(f);
542 if ((flag & 0xC0) == DEFLATED_DATA) {
543 n = ((flag & 0x3f) << 8) + read_byte(f);
544 read_buf(f, cbuf, n);
545 rx_strm.next_in = (Bytef *)cbuf;
546 rx_strm.avail_in = n;
547 recv_state = r_inflating;
548 break;
549 }
550 if (recv_state == r_inflated) {
551 /* check previous inflated stuff ended correctly */
552 rx_strm.avail_in = 0;
553 rx_strm.next_out = (Bytef *)dbuf;
554 rx_strm.avail_out = AVAIL_OUT_SIZE(CHUNK_SIZE);
555 r = inflate(&rx_strm, Z_SYNC_FLUSH);
556 n = AVAIL_OUT_SIZE(CHUNK_SIZE) - rx_strm.avail_out;
557 /*
558 * Z_BUF_ERROR just means no progress was
559 * made, i.e. the decompressor didn't have
560 * any pending output for us.
561 */
562 if (r != Z_OK && r != Z_BUF_ERROR) {
563 rprintf(FERROR, "inflate flush returned %d (%d bytes)\n",
564 r, n);
565 exit_cleanup(RERR_STREAMIO);
566 }
567 if (n != 0 && r != Z_BUF_ERROR) {
568 /* have to return some more data and
569 save the flag for later. */
570 saved_flag = flag + 0x10000;
571 *data = dbuf;
572 return n;
573 }
574 /*
575 * At this point the decompressor should
576 * be expecting to see the 0, 0, ff, ff bytes.
577 */
578 if (!inflateSyncPoint(&rx_strm)) {
579 rprintf(FERROR, "decompressor lost sync!\n");
580 exit_cleanup(RERR_STREAMIO);
581 }
582 rx_strm.avail_in = 4;
583 rx_strm.next_in = (Bytef *)cbuf;
584 cbuf[0] = cbuf[1] = 0;
585 cbuf[2] = cbuf[3] = 0xff;
586 inflate(&rx_strm, Z_SYNC_FLUSH);
587 recv_state = r_idle;
588 }
589 if (flag == END_FLAG) {
590 /* that's all folks */
591 recv_state = r_init;
592 return 0;
593 }
594
595 /* here we have a token of some kind */
596 if (flag & TOKEN_REL) {
597 rx_token += flag & 0x3f;
598 flag >>= 6;
599 } else
600 rx_token = read_int(f);
601 if (flag & 1) {
602 rx_run = read_byte(f);
603 rx_run += read_byte(f) << 8;
604 recv_state = r_running;
605 }
606 return -1 - rx_token;
607
608 case r_inflating:
609 rx_strm.next_out = (Bytef *)dbuf;
610 rx_strm.avail_out = AVAIL_OUT_SIZE(CHUNK_SIZE);
611 r = inflate(&rx_strm, Z_NO_FLUSH);
612 n = AVAIL_OUT_SIZE(CHUNK_SIZE) - rx_strm.avail_out;
613 if (r != Z_OK) {
614 rprintf(FERROR, "inflate returned %d (%d bytes)\n", r, n);
615 exit_cleanup(RERR_STREAMIO);
616 }
617 if (rx_strm.avail_in == 0)
618 recv_state = r_inflated;
619 if (n != 0) {
620 *data = dbuf;
621 return n;
622 }
623 break;
624
625 case r_running:
626 ++rx_token;
627 if (--rx_run == 0)
628 recv_state = r_idle;
629 return -1 - rx_token;
630 }
631 }
632 }
633
634 /*
635 * put the data corresponding to a token that we've just returned
636 * from recv_deflated_token into the decompressor's history buffer.
637 */
638 static void see_deflate_token(char *buf, int32 len)
639 {
640 int r;
641 int32 blklen;
642 unsigned char hdr[5];
643
644 rx_strm.avail_in = 0;
645 blklen = 0;
646 hdr[0] = 0;
647 do {
648 if (rx_strm.avail_in == 0 && len != 0) {
649 if (blklen == 0) {
650 /* Give it a fake stored-block header. */
651 rx_strm.next_in = (Bytef *)hdr;
652 rx_strm.avail_in = 5;
653 blklen = len;
654 if (blklen > 0xffff)
655 blklen = 0xffff;
656 hdr[1] = blklen;
657 hdr[2] = blklen >> 8;
658 hdr[3] = ~hdr[1];
659 hdr[4] = ~hdr[2];
660 } else {
661 rx_strm.next_in = (Bytef *)buf;
662 rx_strm.avail_in = blklen;
663 if (protocol_version >= 31) /* Newer protocols avoid a data-duplicating bug */
664 buf += blklen;
665 len -= blklen;
666 blklen = 0;
667 }
668 }
669 rx_strm.next_out = (Bytef *)dbuf;
670 rx_strm.avail_out = AVAIL_OUT_SIZE(CHUNK_SIZE);
671 r = inflate(&rx_strm, Z_SYNC_FLUSH);
672 if (r != Z_OK && r != Z_BUF_ERROR) {
673 rprintf(FERROR, "inflate (token) returned %d\n", r);
674 exit_cleanup(RERR_STREAMIO);
675 }
676 } while (len || rx_strm.avail_out == 0);
677 }
678
679 #ifdef SUPPORT_ZSTD
680
681 static ZSTD_inBuffer zstd_in_buff;
682 static ZSTD_outBuffer zstd_out_buff;
683 static ZSTD_CCtx *zstd_cctx;
684
685 static void send_zstd_token(int f, int32 token, struct map_struct *buf,
686 OFF_T offset, int32 nb)
687 {
688 static int comp_init_done, flush_pending;
689 ZSTD_EndDirective flush = ZSTD_e_continue;
690 int32 n, r;
691
692 /* initialization */
693 if (!comp_init_done) {
694
695 zstd_cctx = ZSTD_createCCtx();
696 if (!zstd_cctx) {
697 rprintf(FERROR, "compression init failed\n");
698 exit_cleanup(RERR_PROTOCOL);
699 }
700
701 obuf = new_array(char, OBUF_SIZE);
702 if (!obuf)
703 out_of_memory("send_deflated_token");
704
705 ZSTD_CCtx_setParameter(zstd_cctx, ZSTD_c_compressionLevel,
706 do_compression_level);
707 zstd_out_buff.dst = obuf + 2;
708
709 comp_init_done = 1;
710 }
711
712 if (last_token == -1) {
713 last_run_end = 0;
714 run_start = token;
715 flush_pending = 0;
716 } else if (last_token == -2) {
717 run_start = token;
718
719 } else if (nb != 0 || token != last_token + 1
720 || token >= run_start + 65536) {
721
722 /* output previous run */
723 r = run_start - last_run_end;
724 n = last_token - run_start;
725
726 if (r >= 0 && r <= 63) {
727 write_byte(f, (n==0? TOKEN_REL: TOKENRUN_REL) + r);
728 } else {
729 write_byte(f, (n==0? TOKEN_LONG: TOKENRUN_LONG));
730 write_int(f, run_start);
731 }
732 if (n != 0) {
733 write_byte(f, n);
734 write_byte(f, n >> 8);
735 }
736 last_run_end = last_token;
737 run_start = token;
738 }
739
740 last_token = token;
741
742 if (nb || flush_pending) {
743
744 zstd_in_buff.src = map_ptr(buf, offset, nb);
745 zstd_in_buff.size = nb;
746 zstd_in_buff.pos = 0;
747
748 do {
749 if (zstd_out_buff.size == 0) {
750 zstd_out_buff.size = MAX_DATA_COUNT;
751 zstd_out_buff.pos = 0;
752 }
753
754 /* File ended, flush */
755 if (token != -2)
756 flush = ZSTD_e_flush;
757
758 r = ZSTD_compressStream2(zstd_cctx, &zstd_out_buff, &zstd_in_buff, flush);
759 if (ZSTD_isError(r)) {
760 rprintf(FERROR, "ZSTD_compressStream returned %d\n", r);
761 exit_cleanup(RERR_STREAMIO);
762 }
763
764 /*
765 * Nothing is sent if the buffer isn't full so avoid smaller
766 * transfers. If a file is finished then we flush the internal
767 * state and send a smaller buffer so that the remote side can
768 * finish the file.
769 */
770 if (zstd_out_buff.pos == zstd_out_buff.size || flush == ZSTD_e_flush) {
771 n = zstd_out_buff.pos;
772
773 obuf[0] = DEFLATED_DATA + (n >> 8);
774 obuf[1] = n;
775 write_buf(f, obuf, n+2);
776
777 zstd_out_buff.size = 0;
778 }
779 /*
780 * Loop while the input buffer isn't full consumed or the
781 * internal state isn't fully flushed.
782 */
783 } while (zstd_in_buff.pos < zstd_in_buff.size || r > 0);
784 flush_pending = token == -2;
785 }
786
787 if (token == -1) {
788 /* end of file - clean up */
789 write_byte(f, END_FLAG);
790 }
791 }
792
793 static ZSTD_DCtx *zstd_dctx;
794
795 static int32 recv_zstd_token(int f, char **data)
796 {
797 static int decomp_init_done;
798 static int out_buffer_size;
799 int32 n, flag;
800 int r;
801
802 if (!decomp_init_done) {
803
804 zstd_dctx = ZSTD_createDCtx();
805 if (!zstd_dctx) {
806 rprintf(FERROR, "ZSTD_createDStream failed\n");
807 exit_cleanup(RERR_PROTOCOL);
808 }
809
810 /* Output buffer fits two decompressed blocks */
811 out_buffer_size = ZSTD_DStreamOutSize() * 2;
812 cbuf = new_array(char, MAX_DATA_COUNT);
813 dbuf = new_array(char, out_buffer_size);
814 if (!cbuf || !dbuf)
815 out_of_memory("recv_zstd_token");
816
817 zstd_in_buff.src = cbuf;
818 zstd_out_buff.dst = dbuf;
819
820 decomp_init_done = 1;
821 }
822
823 do {
824 switch (recv_state) {
825 case r_init:
826 recv_state = r_idle;
827 rx_token = 0;
828 break;
829
830 case r_idle:
831 flag = read_byte(f);
832 if ((flag & 0xC0) == DEFLATED_DATA) {
833 n = ((flag & 0x3f) << 8) + read_byte(f);
834 read_buf(f, cbuf, n);
835
836 zstd_in_buff.size = n;
837 zstd_in_buff.pos = 0;
838
839 recv_state = r_inflating;
840
841 } else if (flag == END_FLAG) {
842 /* that's all folks */
843 recv_state = r_init;
844 return 0;
845
846 } else {
847 /* here we have a token of some kind */
848 if (flag & TOKEN_REL) {
849 rx_token += flag & 0x3f;
850 flag >>= 6;
851 } else
852 rx_token = read_int(f);
853 if (flag & 1) {
854 rx_run = read_byte(f);
855 rx_run += read_byte(f) << 8;
856 recv_state = r_running;
857 }
858 return -1 - rx_token;
859 }
860 break;
861
862 case r_inflating:
863 zstd_out_buff.size = out_buffer_size;
864 zstd_out_buff.pos = 0;
865
866 r = ZSTD_decompressStream(zstd_dctx, &zstd_out_buff, &zstd_in_buff);
867 n = zstd_out_buff.pos;
868 if (ZSTD_isError(r)) {
869 rprintf(FERROR, "ZSTD decomp returned %d (%d bytes)\n", r, n);
870 exit_cleanup(RERR_STREAMIO);
871 }
872
873 /*
874 * If the input buffer is fully consumed and the output
875 * buffer is not full then next step is to read more
876 * data.
877 */
878 if (zstd_in_buff.size == zstd_in_buff.pos && n < out_buffer_size)
879 recv_state = r_idle;
880
881 if (n != 0) {
882 *data = dbuf;
883 return n;
884 }
885 break;
886
887 case r_running:
888 ++rx_token;
889 if (--rx_run == 0)
890 recv_state = r_idle;
891 return -1 - rx_token;
892 break;
893
894 case r_inflated:
895 break;
896 }
897 } while (1);
898 }
899 #endif /* SUPPORT_ZSTD */
900
901 #ifdef SUPPORT_LZ4
902 static void
903 send_compressed_token(int f, int32 token, struct map_struct *buf, OFF_T offset, int32 nb)
904 {
905 static int init_done, flush_pending;
906 int size = MAX(LZ4_compressBound(CHUNK_SIZE), MAX_DATA_COUNT+2);
907 int32 n, r;
908
909 if (last_token == -1) {
910 if (!init_done) {
911 if ((obuf = new_array(char, size)) == NULL)
912 out_of_memory("send_compressed_token");
913 init_done = 1;
914 }
915 last_run_end = 0;
916 run_start = token;
917 flush_pending = 0;
918 } else if (last_token == -2) {
919 run_start = token;
920 } else if (nb != 0 || token != last_token + 1
921 || token >= run_start + 65536) {
922 /* output previous run */
923 r = run_start - last_run_end;
924 n = last_token - run_start;
925 if (r >= 0 && r <= 63) {
926 write_byte(f, (n==0? TOKEN_REL: TOKENRUN_REL) + r);
927 } else {
928 write_byte(f, (n==0? TOKEN_LONG: TOKENRUN_LONG));
929 write_int(f, run_start);
930 }
931 if (n != 0) {
932 write_byte(f, n);
933 write_byte(f, n >> 8);
934 }
935 last_run_end = last_token;
936 run_start = token;
937 }
938
939 last_token = token;
940
941 if (nb != 0 || flush_pending) {
942 int available_in, available_out = 0;
943 const char *next_in;
944
945 do {
946 char *ptr = obuf;
947 char *next_out = obuf + 2;
948
949 if (available_out == 0) {
950 available_in = MIN(nb, MAX_DATA_COUNT);
951 next_in = map_ptr(buf, offset, available_in);
952 } else
953 available_in /= 2;
954
955 available_out = LZ4_compress_default(next_in, next_out, available_in, size - 2);
956 if (!available_out) {
957 rprintf(FERROR, "compress returned %d\n", available_out);
958 exit_cleanup(RERR_STREAMIO);
959 }
960 if (available_out <= MAX_DATA_COUNT) {
961 ptr[0] = DEFLATED_DATA + (available_out >> 8);
962 ptr[1] = available_out;
963
964 write_buf(f, ptr, available_out + 2);
965
966 available_out = 0;
967 nb -= available_in;
968 offset += available_in;
969 }
970 } while (nb != 0);
971 flush_pending = token == -2;
972 }
973 if (token == -1)
974 /* end of file - clean up */
975 write_byte(f, END_FLAG);
976 }
977
978 static int32 recv_compressed_token(int f, char **data)
979 {
980 static int32 saved_flag;
981 static int init_done;
982 int32 n, flag;
983 int size = MAX(LZ4_compressBound(CHUNK_SIZE), MAX_DATA_COUNT+2);
984 static const char *next_in;
985 static int avail_in;
986 int avail_out;
987
988 for (;;) {
989 switch (recv_state) {
990 case r_init:
991 if (!init_done) {
992 if (!(cbuf = new_array(char, MAX_DATA_COUNT))
993 || !(dbuf = new_array(char, size)))
994 out_of_memory("recv_compressed_token");
995 init_done = 1;
996 }
997 recv_state = r_idle;
998 rx_token = 0;
999 break;
1000 case r_idle:
1001 case r_inflated:
1002 if (saved_flag) {
1003 flag = saved_flag & 0xff;
1004 saved_flag = 0;
1005 } else
1006 flag = read_byte(f);
1007 if ((flag & 0xC0) == DEFLATED_DATA) {
1008 n = ((flag & 0x3f) << 8) + read_byte(f);
1009 read_buf(f, cbuf, n);
1010 next_in = (char *)cbuf;
1011 avail_in = n;
1012 recv_state = r_inflating;
1013 break;
1014 }
1015
1016 if (recv_state == r_inflated)
1017 recv_state = r_idle;
1018
1019 if (flag == END_FLAG) {
1020 /* that's all folks */
1021 recv_state = r_init;
1022 return 0;
1023 }
1024
1025 /* here we have a token of some kind */
1026 if (flag & TOKEN_REL) {
1027 rx_token += flag & 0x3f;
1028 flag >>= 6;
1029 } else
1030 rx_token = read_int(f);
1031 if (flag & 1) {
1032 rx_run = read_byte(f);
1033 rx_run += read_byte(f) << 8;
1034 recv_state = r_running;
1035 }
1036 return -1 - rx_token;
1037
1038 case r_inflating:
1039 avail_out = LZ4_decompress_safe(next_in, dbuf, avail_in, size);
1040 if (avail_out < 0) {
1041 rprintf(FERROR, "uncompress failed: %d\n", avail_out);
1042 exit_cleanup(RERR_STREAMIO);
1043 }
1044 recv_state = r_inflated;
1045 *data = dbuf;
1046 return avail_out;
1047
1048 case r_running:
1049 ++rx_token;
1050 if (--rx_run == 0)
1051 recv_state = r_idle;
1052 return -1 - rx_token;
1053 }
1054 }
1055
1056 }
1057
1058 # if 0
1059 static void see_uncompressed_token(char *buf, int32 len)
1060 {
1061 static const char *next_in;
1062 static int avail_in;
1063 int avail_out;
1064
1065 int32 blklen;
1066 char hdr[5];
1067
1068 avail_in = 0;
1069 blklen = 0;
1070 hdr[0] = 0;
1071 do {
1072 if (avail_in == 0 && len != 0) {
1073 if (blklen == 0) {
1074 /* Give it a fake stored-block header. */
1075 next_in = hdr;
1076 avail_in = 5;
1077 blklen = len;
1078 if (blklen > 0xffff)
1079 blklen = 0xffff;
1080 hdr[1] = blklen;
1081 hdr[2] = blklen >> 8;
1082 hdr[3] = ~hdr[1];
1083 hdr[4] = ~hdr[2];
1084 } else {
1085 next_in = (char *)buf;
1086 avail_in = blklen;
1087 if (protocol_version >= 31) /* Newer protocols avoid a data-duplicating bug */
1088 buf += blklen;
1089 len -= blklen;
1090 blklen = 0;
1091 }
1092 }
1093 avail_out = LZ4_decompress_safe(next_in, dbuf, avail_in, LZ4_compressBound(CHUNK_SIZE));
1094 if (avail_out < 0) {
1095 rprintf(FERROR, "uncompress failed: %d\n", avail_out);
1096 exit_cleanup(RERR_STREAMIO);
1097 }
1098
1099 } while (len);
1100 }
1101 # endif /* 0 */
1102 #endif /* SUPPORT_LZ4 */
1103
1104 /**
1105 * Transmit a verbatim buffer of length @p n followed by a token.
1106 * If token == -1 then we have reached EOF
1107 * If n == 0 then don't send a buffer
1108 */
1109 void send_token(int f, int32 token, struct map_struct *buf, OFF_T offset,
1110 int32 n, int32 toklen)
1111 {
1112 switch (do_compression) {
1113 case CPRES_NONE:
1114 simple_send_token(f, token, buf, offset, n);
1115 break;
1116 case CPRES_ZLIB:
1117 case CPRES_ZLIBX:
1118 send_deflated_token(f, token, buf, offset, n, toklen);
1119 break;
1120 #ifdef SUPPORT_ZSTD
1121 case CPRES_ZSTD:
1122 send_zstd_token(f, token, buf, offset, n);
1123 break;
1124 #endif
1125 #ifdef SUPPORT_LZ4
1126 case CPRES_LZ4:
1127 send_compressed_token(f, token, buf, offset, n);
1128 break;
1129 #endif
1130 default:
1131 assert(0);
1132 }
1133 }
1134
1135 /*
1136 * receive a token or buffer from the other end. If the return value is >0 then
1137 * it is a data buffer of that length, and *data will point at the data.
1138 * if the return value is -i then it represents token i-1
1139 * if the return value is 0 then the end has been reached
1140 */
1141 int32 recv_token(int f, char **data)
1142 {
1143 int tok;
1144
1145 switch (do_compression) {
1146 case CPRES_NONE:
1147 tok = simple_recv_token(f,data);
1148 break;
1149 case CPRES_ZLIB:
1150 case CPRES_ZLIBX:
1151 tok = recv_deflated_token(f, data);
1152 break;
1153 #ifdef SUPPORT_ZSTD
1154 case CPRES_ZSTD:
1155 tok = recv_zstd_token(f, data);
1156 break;
1157 #endif
1158 #ifdef SUPPORT_LZ4
1159 case CPRES_LZ4:
1160 tok = recv_compressed_token(f, data);
1161 break;
1162 #endif
1163 default:
1164 assert(0);
1165 }
1166 return tok;
1167 }
1168
1169 /*
1170 * look at the data corresponding to a token, if necessary
1171 */
1172 void see_token(char *data, int32 toklen)
1173 {
1174 switch (do_compression) {
1175 case CPRES_NONE:
1176 break;
1177 case CPRES_ZLIB:
1178 see_deflate_token(data, toklen);
1179 break;
1180 case CPRES_ZLIBX:
1181 break;
1182 #ifdef SUPPORT_LZ4
1183 case CPRES_LZ4:
1184 /*see_uncompressed_token(data, toklen);*/
1185 break;
1186 #endif
1187 #ifdef SUPPORT_LZ4
1188 case CPRES_ZSTD:
1189 break;
1190 #endif
1191 default:
1192 assert(0);
1193 }
1194 }
rsync: a fast, versatile, remote (and local) file-copying tool