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