Added write_ndx() and read_ndx(), functions that allow us to transmit
[rsync/rsync.git] / sender.c
1 /*
2  * Routines only used by the sending process.
3  *
4  * Copyright (C) 1996 Andrew Tridgell
5  * Copyright (C) 1996 Paul Mackerras
6  * Copyright (C) 2003, 2004, 2005, 2006 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 2 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, write to the Free Software Foundation, Inc.,
20  * 51 Franklin Street - Fifth Floor, Boston, MA 02110-1301, USA.
21  */
22
23 #include "rsync.h"
24
25 extern int verbose;
26 extern int do_xfers;
27 extern int am_server;
28 extern int am_daemon;
29 extern int incremental;
30 extern int log_before_transfer;
31 extern int stdout_format_has_i;
32 extern int logfile_format_has_i;
33 extern int csum_length;
34 extern int append_mode;
35 extern int io_error;
36 extern int allowed_lull;
37 extern int protocol_version;
38 extern int remove_source_files;
39 extern int updating_basis_file;
40 extern int make_backups;
41 extern int do_progress;
42 extern int inplace;
43 extern int batch_fd;
44 extern int write_batch;
45 extern struct stats stats;
46 extern struct file_list *cur_flist, *first_flist;
47 extern char *stdout_format;
48
49 /**
50  * @file
51  *
52  * The sender gets checksums from the generator, calculates deltas,
53  * and transmits them to the receiver.  The sender process runs on the
54  * machine holding the source files.
55  **/
56
57 /**
58  * Receive the checksums for a buffer
59  **/
60 static struct sum_struct *receive_sums(int f)
61 {
62         struct sum_struct *s;
63         int32 i;
64         int lull_mod = allowed_lull * 5;
65         OFF_T offset = 0;
66
67         if (!(s = new(struct sum_struct)))
68                 out_of_memory("receive_sums");
69
70         read_sum_head(f, s);
71
72         s->sums = NULL;
73
74         if (verbose > 3) {
75                 rprintf(FINFO, "count=%.0f n=%ld rem=%ld\n",
76                         (double)s->count, (long)s->blength, (long)s->remainder);
77         }
78
79         if (append_mode > 0) {
80                 s->flength = (OFF_T)s->count * s->blength;
81                 if (s->remainder)
82                         s->flength -= s->blength - s->remainder;
83                 return s;
84         }
85
86         if (s->count == 0)
87                 return(s);
88
89         if (!(s->sums = new_array(struct sum_buf, s->count)))
90                 out_of_memory("receive_sums");
91
92         for (i = 0; i < s->count; i++) {
93                 s->sums[i].sum1 = read_int(f);
94                 read_buf(f, s->sums[i].sum2, s->s2length);
95
96                 s->sums[i].offset = offset;
97                 s->sums[i].flags = 0;
98
99                 if (i == s->count-1 && s->remainder != 0)
100                         s->sums[i].len = s->remainder;
101                 else
102                         s->sums[i].len = s->blength;
103                 offset += s->sums[i].len;
104
105                 if (allowed_lull && !(i % lull_mod))
106                         maybe_send_keepalive();
107
108                 if (verbose > 3) {
109                         rprintf(FINFO,
110                                 "chunk[%d] len=%d offset=%.0f sum1=%08x\n",
111                                 i, s->sums[i].len, (double)s->sums[i].offset,
112                                 s->sums[i].sum1);
113                 }
114         }
115
116         s->flength = offset;
117
118         return s;
119 }
120
121 void successful_send(int ndx)
122 {
123         char fname[MAXPATHLEN];
124         struct file_struct *file;
125         struct file_list *flist;
126
127         if (!remove_source_files)
128                 return;
129
130         if (!(flist = flist_for_ndx(ndx))) {
131                 rprintf(FERROR,
132                         "INTERNAL ERROR: unable to find flist for item %d\n",
133                         ndx);
134                 return;
135         }
136
137         file = flist->files[ndx - flist->ndx_start];
138         if (!push_flist_dir(F_ROOTDIR(file), -1))
139                 return;
140         f_name(file, fname);
141
142         if (do_unlink(fname) == 0) {
143                 if (verbose > 1)
144                         rprintf(FINFO, "sender removed %s\n", fname);
145         } else
146                 rsyserr(FERROR, errno, "sender failed to remove %s", fname);
147 }
148
149 void write_ndx_and_attrs(int f_out, int ndx, int iflags,
150                          uchar fnamecmp_type, char *buf, int len)
151 {
152         write_int(f_out, ndx);
153         if (protocol_version < 29)
154                 return;
155         write_shortint(f_out, iflags);
156         if (iflags & ITEM_BASIS_TYPE_FOLLOWS)
157                 write_byte(f_out, fnamecmp_type);
158         if (iflags & ITEM_XNAME_FOLLOWS)
159                 write_vstring(f_out, buf, len);
160 }
161
162 void send_files(int f_in, int f_out)
163 {
164         int fd = -1;
165         struct sum_struct *s;
166         struct map_struct *mbuf = NULL;
167         STRUCT_STAT st;
168         char fname[MAXPATHLEN], xname[MAXPATHLEN];
169         const char *path, *slash;
170         uchar fnamecmp_type;
171         int iflags, xlen;
172         struct file_struct *file;
173         int phase = 0, max_phase = protocol_version >= 29 ? 2 : 1;
174         struct stats initial_stats;
175         int itemizing = am_server ? logfile_format_has_i : stdout_format_has_i;
176         enum logcode log_code = log_before_transfer ? FLOG : FINFO;
177         int f_xfer = write_batch < 0 ? batch_fd : f_out;
178         int ndx, j;
179
180         if (verbose > 2)
181                 rprintf(FINFO, "send_files starting\n");
182
183         while (1) {
184                 if (incremental)
185                         send_extra_file_list(f_out, 1000);
186
187                 /* This call also sets cur_flist. */
188                 ndx = read_ndx_and_attrs(f_in, f_out, &iflags,
189                                          &fnamecmp_type, xname, &xlen);
190                 if (ndx == NDX_DONE) {
191                         if (incremental && first_flist) {
192                                 flist_free(first_flist);
193                                 if (first_flist) {
194                                         write_int(f_out, NDX_DONE);
195                                         continue;
196                                 }
197                         }
198                         if (++phase > max_phase)
199                                 break;
200                         if (verbose > 2)
201                                 rprintf(FINFO, "send_files phase=%d\n", phase);
202                         write_int(f_out, NDX_DONE);
203                         continue;
204                 }
205
206                 file = cur_flist->files[ndx - cur_flist->ndx_start];
207                 if (F_ROOTDIR(file)) {
208                         path = F_ROOTDIR(file);
209                         slash = "/";
210                 } else {
211                         path = slash = "";
212                 }
213                 if (!push_flist_dir(F_ROOTDIR(file), -1))
214                         continue;
215                 f_name(file, fname);
216
217                 if (verbose > 2)
218                         rprintf(FINFO, "send_files(%d, %s%s%s)\n", ndx, path,slash,fname);
219
220                 if (!(iflags & ITEM_TRANSFER)) {
221                         maybe_log_item(file, iflags, itemizing, xname);
222                         continue;
223                 }
224                 if (phase == 2) {
225                         rprintf(FERROR,
226                                 "got transfer request in phase 2 [%s]\n",
227                                 who_am_i());
228                         exit_cleanup(RERR_PROTOCOL);
229                 }
230
231                 if (file->flags & FLAG_FILE_SENT) {
232                         if (csum_length == SHORT_SUM_LENGTH) {
233                                 /* For inplace: redo phase turns off the backup
234                                  * flag so that we do a regular inplace send. */
235                                 make_backups = -make_backups;
236                                 append_mode = -append_mode;
237                                 csum_length = SUM_LENGTH;
238                         }
239                 } else {
240                         if (csum_length != SHORT_SUM_LENGTH) {
241                                 make_backups = -make_backups;
242                                 append_mode = -append_mode;
243                                 csum_length = SHORT_SUM_LENGTH;
244                         }
245                 }
246
247                 updating_basis_file = inplace && (protocol_version >= 29
248                         ? fnamecmp_type == FNAMECMP_FNAME : make_backups <= 0);
249
250                 stats.current_file_index = ndx;
251                 stats.num_transferred_files++;
252                 stats.total_transferred_size += F_LENGTH(file);
253
254                 if (!do_xfers) { /* log the transfer */
255                         log_item(FCLIENT, file, &stats, iflags, NULL);
256                         write_ndx_and_attrs(f_out, ndx, iflags, fnamecmp_type,
257                                             xname, xlen);
258                         continue;
259                 }
260
261                 initial_stats = stats;
262
263                 if (!(s = receive_sums(f_in))) {
264                         io_error |= IOERR_GENERAL;
265                         rprintf(FERROR, "receive_sums failed\n");
266                         exit_cleanup(RERR_PROTOCOL);
267                 }
268
269                 fd = do_open(fname, O_RDONLY, 0);
270                 if (fd == -1) {
271                         if (errno == ENOENT) {
272                                 enum logcode c = am_daemon
273                                     && protocol_version < 28 ? FERROR
274                                                              : FINFO;
275                                 io_error |= IOERR_VANISHED;
276                                 rprintf(c, "file has vanished: %s\n",
277                                         full_fname(fname));
278                         } else {
279                                 io_error |= IOERR_GENERAL;
280                                 rsyserr(FERROR, errno,
281                                         "send_files failed to open %s",
282                                         full_fname(fname));
283                         }
284                         free_sums(s);
285                         if (protocol_version >= 30)
286                                 send_msg_int(MSG_NO_SEND, ndx);
287                         continue;
288                 }
289
290                 /* map the local file */
291                 if (do_fstat(fd, &st) != 0) {
292                         io_error |= IOERR_GENERAL;
293                         rsyserr(FERROR, errno, "fstat failed");
294                         free_sums(s);
295                         close(fd);
296                         exit_cleanup(RERR_PROTOCOL);
297                 }
298
299                 if (st.st_size) {
300                         int32 read_size = MAX(s->blength * 3, MAX_MAP_SIZE);
301                         mbuf = map_file(fd, st.st_size, read_size, s->blength);
302                 } else
303                         mbuf = NULL;
304
305                 if (verbose > 2) {
306                         rprintf(FINFO, "send_files mapped %s%s%s of size %.0f\n",
307                                 path,slash,fname, (double)st.st_size);
308                 }
309
310                 write_ndx_and_attrs(f_out, ndx, iflags, fnamecmp_type,
311                                     xname, xlen);
312                 write_sum_head(f_xfer, s);
313
314                 if (verbose > 2)
315                         rprintf(FINFO, "calling match_sums %s%s%s\n", path,slash,fname);
316
317                 if (log_before_transfer)
318                         log_item(FCLIENT, file, &initial_stats, iflags, NULL);
319                 else if (!am_server && verbose && do_progress)
320                         rprintf(FCLIENT, "%s\n", fname);
321
322                 set_compression(fname);
323
324                 match_sums(f_xfer, s, mbuf, st.st_size);
325                 if (do_progress)
326                         end_progress(st.st_size);
327
328                 log_item(log_code, file, &initial_stats, iflags, NULL);
329
330                 if (mbuf) {
331                         j = unmap_file(mbuf);
332                         if (j) {
333                                 io_error |= IOERR_GENERAL;
334                                 rsyserr(FERROR, j,
335                                         "read errors mapping %s",
336                                         full_fname(fname));
337                         }
338                 }
339                 close(fd);
340
341                 free_sums(s);
342
343                 if (verbose > 2)
344                         rprintf(FINFO, "sender finished %s%s%s\n", path,slash,fname);
345
346                 /* Flag that we actually sent this entry. */
347                 file->flags |= FLAG_FILE_SENT;
348         }
349         if (make_backups < 0)
350                 make_backups = -make_backups;
351
352         if (verbose > 2)
353                 rprintf(FINFO, "send files finished\n");
354
355         match_report();
356
357         write_int(f_out, NDX_DONE);
358 }