+#include "ifuncs.h"
+
+/** If no timeout is specified then use a 60 second select timeout */
+#define SELECT_TIMEOUT 60
+
+extern int bwlimit;
+extern size_t bwlimit_writemax;
+extern int io_timeout;
+extern int am_server;
+extern int am_daemon;
+extern int am_sender;
+extern int am_generator;
+extern int inc_recurse;
+extern int io_error;
+extern int eol_nulls;
+extern int flist_eof;
+extern int list_only;
+extern int read_batch;
+extern int protect_args;
+extern int checksum_seed;
+extern int protocol_version;
+extern int remove_source_files;
+extern int preserve_hard_links;
+extern struct stats stats;
+extern struct file_list *cur_flist;
+#ifdef ICONV_OPTION
+extern int filesfrom_convert;
+extern iconv_t ic_send, ic_recv;
+#endif
+
+int csum_length = SHORT_SUM_LENGTH; /* initial value */
+int allowed_lull = 0;
+int ignore_timeout = 0;
+int batch_fd = -1;
+int msgdone_cnt = 0;
+
+/* Ignore an EOF error if non-zero. See whine_about_eof(). */
+int kluge_around_eof = 0;
+
+int msg_fd_in = -1;
+int msg_fd_out = -1;
+int sock_f_in = -1;
+int sock_f_out = -1;
+
+static int iobuf_f_in = -1;
+static char *iobuf_in;
+static size_t iobuf_in_siz;
+static size_t iobuf_in_ndx;
+static size_t iobuf_in_remaining;
+
+static int iobuf_f_out = -1;
+static char *iobuf_out;
+static int iobuf_out_cnt;
+
+int flist_forward_from = -1;
+
+static int io_multiplexing_out;
+static int io_multiplexing_in;
+static time_t last_io_in;
+static time_t last_io_out;
+static int no_flush;
+
+static int write_batch_monitor_in = -1;
+static int write_batch_monitor_out = -1;
+
+static int io_filesfrom_f_in = -1;
+static int io_filesfrom_f_out = -1;
+static xbuf ff_buf = EMPTY_XBUF;
+static char ff_lastchar;
+#ifdef ICONV_OPTION
+static xbuf iconv_buf = EMPTY_XBUF;
+#endif
+static int defer_forwarding_messages = 0, defer_forwarding_keep = 0;
+static int select_timeout = SELECT_TIMEOUT;
+static int active_filecnt = 0;
+static OFF_T active_bytecnt = 0;
+static int first_message = 1;
+
+static char int_byte_extra[64] = {
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, /* (00 - 3F)/4 */
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, /* (40 - 7F)/4 */
+ 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, /* (80 - BF)/4 */
+ 2, 2, 2, 2, 2, 2, 2, 2, 3, 3, 3, 3, 4, 4, 5, 6, /* (C0 - FF)/4 */
+};
+
+#define REMOTE_OPTION_ERROR "rsync: on remote machine: -"
+#define REMOTE_OPTION_ERROR2 ": unknown option"
+
+enum festatus { FES_SUCCESS, FES_REDO, FES_NO_SEND };
+
+static void readfd(int fd, char *buffer, size_t N);
+static void writefd(int fd, const char *buf, size_t len);
+static void writefd_unbuffered(int fd, const char *buf, size_t len);
+static void mplex_write(int fd, enum msgcode code, const char *buf, size_t len, int convert);
+
+struct flist_ndx_item {
+ struct flist_ndx_item *next;
+ int ndx;
+};
+
+struct flist_ndx_list {
+ struct flist_ndx_item *head, *tail;
+};
+
+static struct flist_ndx_list redo_list, hlink_list;
+
+struct msg_list_item {
+ struct msg_list_item *next;
+ char convert;
+ char buf[1];
+};
+
+struct msg_list {
+ struct msg_list_item *head, *tail;
+};
+
+static struct msg_list msg_queue;
+
+static void flist_ndx_push(struct flist_ndx_list *lp, int ndx)
+{
+ struct flist_ndx_item *item;
+
+ if (!(item = new(struct flist_ndx_item)))
+ out_of_memory("flist_ndx_push");
+ item->next = NULL;
+ item->ndx = ndx;
+ if (lp->tail)
+ lp->tail->next = item;
+ else
+ lp->head = item;
+ lp->tail = item;
+}
+
+static int flist_ndx_pop(struct flist_ndx_list *lp)
+{
+ struct flist_ndx_item *next;
+ int ndx;
+
+ if (!lp->head)
+ return -1;
+
+ ndx = lp->head->ndx;
+ next = lp->head->next;
+ free(lp->head);
+ lp->head = next;
+ if (!next)
+ lp->tail = NULL;
+
+ return ndx;
+}
+
+static void got_flist_entry_status(enum festatus status, const char *buf)
+{
+ int ndx = IVAL(buf, 0);
+ struct file_list *flist = flist_for_ndx(ndx);
+
+ assert(flist != NULL);
+
+ if (remove_source_files) {
+ active_filecnt--;
+ active_bytecnt -= F_LENGTH(flist->files[ndx - flist->ndx_start]);
+ }
+
+ if (inc_recurse)
+ flist->in_progress--;
+
+ switch (status) {
+ case FES_SUCCESS:
+ if (remove_source_files)
+ send_msg(MSG_SUCCESS, buf, 4, 0);
+ if (preserve_hard_links) {
+ struct file_struct *file = flist->files[ndx - flist->ndx_start];
+ if (F_IS_HLINKED(file)) {
+ flist_ndx_push(&hlink_list, ndx);
+ flist->in_progress++;
+ }
+ }
+ break;
+ case FES_REDO:
+ if (inc_recurse)
+ flist->to_redo++;
+ flist_ndx_push(&redo_list, ndx);
+ break;
+ case FES_NO_SEND:
+ break;
+ }
+}
+
+static void check_timeout(void)
+{
+ time_t t;
+
+ if (!io_timeout || ignore_timeout)
+ return;
+
+ if (!last_io_in) {
+ last_io_in = time(NULL);
+ return;
+ }
+
+ t = time(NULL);
+
+ if (t - last_io_in >= io_timeout) {
+ if (!am_server && !am_daemon) {
+ rprintf(FERROR, "io timeout after %d seconds -- exiting\n",
+ (int)(t-last_io_in));
+ }
+ exit_cleanup(RERR_TIMEOUT);
+ }
+}
+
+/* Note the fds used for the main socket (which might really be a pipe
+ * for a local transfer, but we can ignore that). */
+void io_set_sock_fds(int f_in, int f_out)
+{
+ sock_f_in = f_in;
+ sock_f_out = f_out;
+}
+
+void set_io_timeout(int secs)
+{
+ io_timeout = secs;
+
+ if (!io_timeout || io_timeout > SELECT_TIMEOUT)
+ select_timeout = SELECT_TIMEOUT;
+ else
+ select_timeout = io_timeout;
+
+ allowed_lull = read_batch ? 0 : (io_timeout + 1) / 2;
+}
+
+/* Setup the fd used to receive MSG_* messages. Only needed during the
+ * early stages of being a local sender (up through the sending of the
+ * file list) or when we're the generator (to fetch the messages from
+ * the receiver). */
+void set_msg_fd_in(int fd)
+{
+ msg_fd_in = fd;
+}
+
+/* Setup the fd used to send our MSG_* messages. Only needed when
+ * we're the receiver (to send our messages to the generator). */
+void set_msg_fd_out(int fd)
+{
+ msg_fd_out = fd;
+ set_nonblocking(msg_fd_out);
+}
+
+/* Add a message to the pending MSG_* list. */
+static void msg_list_add(struct msg_list *lst, int code, const char *buf, int len, int convert)
+{
+ struct msg_list_item *m;
+ int sz = len + 4 + sizeof m[0] - 1;
+
+ if (!(m = (struct msg_list_item *)new_array(char, sz)))
+ out_of_memory("msg_list_add");
+ m->next = NULL;
+ m->convert = convert;
+ SIVAL(m->buf, 0, ((code+MPLEX_BASE)<<24) | len);
+ memcpy(m->buf + 4, buf, len);
+ if (lst->tail)
+ lst->tail->next = m;
+ else
+ lst->head = m;
+ lst->tail = m;
+}
+
+static void msg_flush(void)
+{
+ if (am_generator) {
+ while (msg_queue.head && io_multiplexing_out) {
+ struct msg_list_item *m = msg_queue.head;
+ int len = IVAL(m->buf, 0) & 0xFFFFFF;
+ int tag = *((uchar*)m->buf+3) - MPLEX_BASE;
+ if (!(msg_queue.head = m->next))
+ msg_queue.tail = NULL;
+ stats.total_written += len + 4;
+ defer_forwarding_messages++;
+ mplex_write(sock_f_out, tag, m->buf + 4, len, m->convert);
+ defer_forwarding_messages--;
+ free(m);
+ }
+ } else {
+ while (msg_queue.head) {
+ struct msg_list_item *m = msg_queue.head;
+ int len = IVAL(m->buf, 0) & 0xFFFFFF;
+ int tag = *((uchar*)m->buf+3) - MPLEX_BASE;
+ if (!(msg_queue.head = m->next))
+ msg_queue.tail = NULL;
+ defer_forwarding_messages++;
+ mplex_write(msg_fd_out, tag, m->buf + 4, len, m->convert);
+ defer_forwarding_messages--;
+ free(m);
+ }
+ }
+}
+
+static void check_for_d_option_error(const char *msg)
+{
+ static char rsync263_opts[] = "BCDHIKLPRSTWabceghlnopqrtuvxz";
+ char *colon;
+ int saw_d = 0;
+
+ if (*msg != 'r'
+ || strncmp(msg, REMOTE_OPTION_ERROR, sizeof REMOTE_OPTION_ERROR - 1) != 0)
+ return;
+
+ msg += sizeof REMOTE_OPTION_ERROR - 1;
+ if (*msg == '-' || (colon = strchr(msg, ':')) == NULL
+ || strncmp(colon, REMOTE_OPTION_ERROR2, sizeof REMOTE_OPTION_ERROR2 - 1) != 0)
+ return;
+
+ for ( ; *msg != ':'; msg++) {
+ if (*msg == 'd')
+ saw_d = 1;
+ else if (*msg == 'e')
+ break;
+ else if (strchr(rsync263_opts, *msg) == NULL)
+ return;
+ }
+
+ if (saw_d) {
+ rprintf(FWARNING,
+ "*** Try using \"--old-d\" if remote rsync is <= 2.6.3 ***\n");
+ }
+}
+
+/* Read a message from the MSG_* fd and handle it. This is called either
+ * during the early stages of being a local sender (up through the sending
+ * of the file list) or when we're the generator (to fetch the messages
+ * from the receiver). */
+static void read_msg_fd(void)
+{
+ char buf[2048];
+ size_t n;
+ struct file_list *flist;
+ int fd = msg_fd_in;
+ int tag, len;
+
+ /* Temporarily disable msg_fd_in. This is needed to avoid looping back
+ * to this routine from writefd_unbuffered(). */
+ no_flush++;
+ msg_fd_in = -1;
+ defer_forwarding_messages++;
+
+ readfd(fd, buf, 4);
+ tag = IVAL(buf, 0);
+
+ len = tag & 0xFFFFFF;
+ tag = (tag >> 24) - MPLEX_BASE;
+
+ switch (tag) {
+ case MSG_DONE:
+ if (len < 0 || len > 1 || !am_generator) {
+ invalid_msg:
+ rprintf(FERROR, "invalid message %d:%d [%s%s]\n",
+ tag, len, who_am_i(),
+ inc_recurse ? "/inc" : "");
+ exit_cleanup(RERR_STREAMIO);
+ }
+ if (len) {
+ readfd(fd, buf, len);
+ stats.total_read = read_varlong(fd, 3);
+ }
+ msgdone_cnt++;
+ break;
+ case MSG_REDO:
+ if (len != 4 || !am_generator)
+ goto invalid_msg;
+ readfd(fd, buf, 4);
+ got_flist_entry_status(FES_REDO, buf);
+ break;
+ case MSG_FLIST:
+ if (len != 4 || !am_generator || !inc_recurse)
+ goto invalid_msg;
+ readfd(fd, buf, 4);
+ /* Read extra file list from receiver. */
+ assert(iobuf_in != NULL);
+ assert(iobuf_f_in == fd);
+ if (DEBUG_GTE(FLIST, 2)) {
+ rprintf(FINFO, "[%s] receiving flist for dir %d\n",
+ who_am_i(), IVAL(buf,0));
+ }
+ flist = recv_file_list(fd);
+ flist->parent_ndx = IVAL(buf,0);
+#ifdef SUPPORT_HARD_LINKS
+ if (preserve_hard_links)
+ match_hard_links(flist);
+#endif
+ break;
+ case MSG_FLIST_EOF:
+ if (len != 0 || !am_generator || !inc_recurse)
+ goto invalid_msg;
+ flist_eof = 1;
+ break;
+ case MSG_IO_ERROR:
+ if (len != 4)
+ goto invalid_msg;
+ readfd(fd, buf, len);
+ io_error |= IVAL(buf, 0);
+ break;
+ case MSG_DELETED:
+ if (len >= (int)sizeof buf || !am_generator)
+ goto invalid_msg;
+ readfd(fd, buf, len);
+ send_msg(MSG_DELETED, buf, len, 1);
+ break;
+ case MSG_SUCCESS:
+ if (len != 4 || !am_generator)
+ goto invalid_msg;
+ readfd(fd, buf, 4);
+ got_flist_entry_status(FES_SUCCESS, buf);
+ break;
+ case MSG_NO_SEND:
+ if (len != 4 || !am_generator)
+ goto invalid_msg;
+ readfd(fd, buf, 4);
+ got_flist_entry_status(FES_NO_SEND, buf);
+ break;
+ case MSG_ERROR_SOCKET:
+ case MSG_CLIENT:
+ if (!am_generator)
+ goto invalid_msg;
+ if (tag == MSG_ERROR_SOCKET)
+ io_end_multiplex_out();
+ /* FALL THROUGH */
+ case MSG_INFO:
+ case MSG_ERROR:
+ case MSG_ERROR_XFER:
+ case MSG_WARNING:
+ case MSG_LOG:
+ while (len) {
+ n = len;
+ if (n >= sizeof buf)
+ n = sizeof buf - 1;
+ readfd(fd, buf, n);
+ rwrite((enum logcode)tag, buf, n, !am_generator);
+ len -= n;
+ }
+ break;
+ default:
+ rprintf(FERROR, "unknown message %d:%d [%s]\n",
+ tag, len, who_am_i());
+ exit_cleanup(RERR_STREAMIO);
+ }
+
+ no_flush--;
+ msg_fd_in = fd;
+ if (!--defer_forwarding_messages && !no_flush)
+ msg_flush();
+}
+
+/* This is used by the generator to limit how many file transfers can
+ * be active at once when --remove-source-files is specified. Without
+ * this, sender-side deletions were mostly happening at the end. */
+void increment_active_files(int ndx, int itemizing, enum logcode code)
+{
+ /* TODO: tune these limits? */
+ while (active_filecnt >= (active_bytecnt >= 128*1024 ? 10 : 50)) {
+ check_for_finished_files(itemizing, code, 0);
+ if (iobuf_out_cnt)
+ io_flush(NORMAL_FLUSH);
+ else
+ read_msg_fd();
+ }