extern int csum_length;
extern int checksum_seed;
extern int protocol_version;
-extern int remove_sent_files;
+extern int remove_source_files;
extern int preserve_hard_links;
extern char *filesfrom_host;
extern struct stats stats;
exit_cleanup(RERR_STREAMIO);
}
read_loop(fd, buf, 4);
- if (remove_sent_files)
+ if (remove_source_files)
decrement_active_files(IVAL(buf,0));
flist_ndx_push(&redo_list, IVAL(buf,0));
break;
exit_cleanup(RERR_STREAMIO);
}
read_loop(fd, buf, len);
- if (defer_forwarding_messages)
- msg_list_add(&msg2sndr, MSG_DELETED, buf, len);
- else
- io_multiplex_write(MSG_DELETED, buf, len);
+ send_msg(MSG_DELETED, buf, len);
break;
case MSG_SUCCESS:
if (len != 4 || !am_generator) {
exit_cleanup(RERR_STREAMIO);
}
read_loop(fd, buf, len);
- if (remove_sent_files) {
+ if (remove_source_files) {
decrement_active_files(IVAL(buf,0));
- if (defer_forwarding_messages)
- msg_list_add(&msg2sndr, MSG_SUCCESS, buf, len);
- else
- io_multiplex_write(MSG_SUCCESS, buf, len);
+ send_msg(MSG_SUCCESS, buf, len);
}
if (preserve_hard_links)
flist_ndx_push(&hlink_list, IVAL(buf,0));
exit_cleanup(RERR_STREAMIO);
}
close_multiplexing_out();
- defer_forwarding_messages = 0;
/* FALL THROUGH */
case MSG_INFO:
case MSG_ERROR:
if (n >= sizeof buf)
n = sizeof buf - 1;
read_loop(fd, buf, n);
- if (am_generator && am_server
- && defer_forwarding_messages && tag != MSG_LOG)
- msg_list_add(&msg2sndr, tag, buf, n);
- else
- rwrite((enum logcode)tag, buf, n);
+ rwrite(tag, buf, n);
len -= n;
}
break;
}
/* This is used by the generator to limit how many file transfers can
- * be active at once when --remove-sent-files is specified. Without
+ * 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)
{
return 1;
}
-void send_msg(enum msgcode code, char *buf, int len)
+int send_msg(enum msgcode code, char *buf, int len)
{
if (msg_fd_out < 0) {
- io_multiplex_write(code, buf, len);
- return;
+ if (!defer_forwarding_messages)
+ return io_multiplex_write(code, buf, len);
+ if (!io_multiplexing_out)
+ return 0;
+ msg_list_add(&msg2sndr, code, buf, len);
+ return 1;
}
msg_list_add(&msg2genr, code, buf, len);
msg2genr_flush(NORMAL_FLUSH);
+ return 1;
}
int get_redo_num(int itemizing, enum logcode code)
if (cnt < 0 && (errno == EWOULDBLOCK
|| errno == EINTR || errno == EAGAIN)) {
struct timeval tv;
- fd_set fds;
- FD_ZERO(&fds);
- FD_SET(fd, &fds);
+ fd_set r_fds, e_fds;
+ FD_ZERO(&r_fds);
+ FD_SET(fd, &r_fds);
+ FD_ZERO(&e_fds);
+ FD_SET(fd, &e_fds);
tv.tv_sec = select_timeout;
tv.tv_usec = 0;
- if (!select(fd+1, &fds, NULL, NULL, &tv))
+ if (!select(fd+1, &r_fds, NULL, &e_fds, &tv))
check_timeout();
+ if (FD_ISSET(fd, &e_fds)) {
+ rsyserr(FINFO, errno,
+ "select exception on fd %d", fd);
+ }
continue;
}
if (cnt != 1)
static void writefd_unbuffered(int fd,char *buf,size_t len)
{
size_t n, total = 0;
- fd_set w_fds, r_fds;
+ fd_set w_fds, r_fds, e_fds;
int maxfd, count, cnt, using_r_fds;
int defer_save = defer_forwarding_messages;
struct timeval tv;
while (total < len) {
FD_ZERO(&w_fds);
- FD_SET(fd,&w_fds);
+ FD_SET(fd, &w_fds);
+ FD_ZERO(&e_fds);
+ FD_SET(fd, &e_fds);
maxfd = fd;
if (msg_fd_in >= 0) {
FD_ZERO(&r_fds);
- FD_SET(msg_fd_in,&r_fds);
+ FD_SET(msg_fd_in, &r_fds);
if (msg_fd_in > maxfd)
maxfd = msg_fd_in;
using_r_fds = 1;
errno = 0;
count = select(maxfd + 1, using_r_fds ? &r_fds : NULL,
- &w_fds, NULL, &tv);
+ &w_fds, &e_fds, &tv);
if (count <= 0) {
if (count < 0 && errno == EBADF)
continue;
}
+ if (FD_ISSET(fd, &e_fds)) {
+ rsyserr(FINFO, errno,
+ "select exception on fd %d", fd);
+ }
+
if (using_r_fds && FD_ISSET(msg_fd_in, &r_fds))
read_msg_fd();
while (msg2sndr.head && io_multiplexing_out) {
struct msg_list_item *m = msg2sndr.head;
- int tag = *((uchar*)m->buf+3) - MPLEX_BASE;
if (!(msg2sndr.head = m->next))
msg2sndr.tail = NULL;
+ stats.total_written += m->len;
defer_forwarding_messages = 1;
- switch (tag) {
- case MSG_INFO:
- case MSG_ERROR:
- rwrite((enum logcode)tag, m->buf + 4, m->len - 4);
- break;
- default:
- stats.total_written += m->len;
- writefd_unbuffered(sock_f_out, m->buf, m->len);
- break;
- }
+ writefd_unbuffered(sock_f_out, m->buf, m->len);
defer_forwarding_messages = 0;
free(m);
}