Matt McCutchen's Web Site
/
rsync
/
rsync.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Added parsing for --delete-delay.
[rsync/rsync.git]
/
io.c
diff --git
a/io.c
b/io.c
index
e235d3e
..
fc2ec29
100644
(file)
--- a/
io.c
+++ b/
io.c
@@
-46,7
+46,7
@@
extern int read_batch;
extern int csum_length;
extern int checksum_seed;
extern int protocol_version;
extern int csum_length;
extern int checksum_seed;
extern int protocol_version;
-extern int remove_s
ent
_files;
+extern int remove_s
ource
_files;
extern int preserve_hard_links;
extern char *filesfrom_host;
extern struct stats stats;
extern int preserve_hard_links;
extern char *filesfrom_host;
extern struct stats stats;
@@
-204,7
+204,7
@@
void set_msg_fd_out(int fd)
}
/* Add a message to the pending MSG_* list. */
}
/* Add a message to the pending MSG_* list. */
-static void msg_list_add(struct msg_list *lst, int code, char *buf, int len)
+static void msg_list_add(struct msg_list *lst, int code, c
onst c
har *buf, int len)
{
struct msg_list_item *m;
int sz = len + 4 + sizeof m[0] - 1;
{
struct msg_list_item *m;
int sz = len + 4 + sizeof m[0] - 1;
@@
-257,7
+257,7
@@
static void read_msg_fd(void)
exit_cleanup(RERR_STREAMIO);
}
read_loop(fd, buf, 4);
exit_cleanup(RERR_STREAMIO);
}
read_loop(fd, buf, 4);
- if (remove_s
ent
_files)
+ if (remove_s
ource
_files)
decrement_active_files(IVAL(buf,0));
flist_ndx_push(&redo_list, IVAL(buf,0));
break;
decrement_active_files(IVAL(buf,0));
flist_ndx_push(&redo_list, IVAL(buf,0));
break;
@@
-267,10
+267,7
@@
static void read_msg_fd(void)
exit_cleanup(RERR_STREAMIO);
}
read_loop(fd, buf, len);
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) {
break;
case MSG_SUCCESS:
if (len != 4 || !am_generator) {
@@
-278,12
+275,9
@@
static void read_msg_fd(void)
exit_cleanup(RERR_STREAMIO);
}
read_loop(fd, buf, len);
exit_cleanup(RERR_STREAMIO);
}
read_loop(fd, buf, len);
- if (remove_s
ent
_files) {
+ if (remove_s
ource
_files) {
decrement_active_files(IVAL(buf,0));
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));
}
if (preserve_hard_links)
flist_ndx_push(&hlink_list, IVAL(buf,0));
@@
-294,7
+288,6
@@
static void read_msg_fd(void)
exit_cleanup(RERR_STREAMIO);
}
close_multiplexing_out();
exit_cleanup(RERR_STREAMIO);
}
close_multiplexing_out();
- defer_forwarding_messages = 0;
/* FALL THROUGH */
case MSG_INFO:
case MSG_ERROR:
/* FALL THROUGH */
case MSG_INFO:
case MSG_ERROR:
@@
-304,11
+297,7
@@
static void read_msg_fd(void)
if (n >= sizeof buf)
n = sizeof buf - 1;
read_loop(fd, buf, n);
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((enum logcode)tag, buf, n);
len -= n;
}
break;
len -= n;
}
break;
@@
-322,7
+311,7
@@
static void read_msg_fd(void)
}
/* This is used by the generator to limit how many file transfers can
}
/* This is used by the generator to limit how many file transfers can
- * be active at once when --remove-s
ent
-files is specified. Without
+ * be active at once when --remove-s
ource
-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)
{
* this, sender-side deletions were mostly happening at the end. */
void increment_active_files(int ndx, int itemizing, enum logcode code)
{
@@
-382,14
+371,19
@@
static int msg2genr_flush(int flush_it_all)
return 1;
}
return 1;
}
-
void send_msg(enum msgcode code,
char *buf, int len)
+
int send_msg(enum msgcode code, const
char *buf, int len)
{
if (msg_fd_out < 0) {
{
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);
}
msg_list_add(&msg2genr, code, buf, len);
msg2genr_flush(NORMAL_FLUSH);
+ return 1;
}
int get_redo_num(int itemizing, enum logcode code)
}
int get_redo_num(int itemizing, enum logcode code)
@@
-639,13
+633,19
@@
int read_filesfrom_line(int fd, char *fname)
if (cnt < 0 && (errno == EWOULDBLOCK
|| errno == EINTR || errno == EAGAIN)) {
struct timeval tv;
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;
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();
check_timeout();
+ if (FD_ISSET(fd, &e_fds)) {
+ rsyserr(FINFO, errno,
+ "select exception on fd %d", fd);
+ }
continue;
}
if (cnt != 1)
continue;
}
if (cnt != 1)
@@
-894,12
+894,12
@@
int64 read_longint(int f)
return num;
}
return num;
}
-void read_buf(int f,
char *buf,
size_t len)
+void read_buf(int f,
char *buf,
size_t len)
{
readfd(f,buf,len);
}
{
readfd(f,buf,len);
}
-void read_sbuf(int f,
char *buf,
size_t len)
+void read_sbuf(int f,
char *buf,
size_t len)
{
readfd(f, buf, len);
buf[len] = '\0';
{
readfd(f, buf, len);
buf[len] = '\0';
@@
-1039,10
+1039,10
@@
static void sleep_for_bwlimit(int bytes_written)
*
* This function underlies the multiplexing system. The body of the
* application never calls this function directly. */
*
* This function underlies the multiplexing system. The body of the
* application never calls this function directly. */
-static void writefd_unbuffered(int fd,
char *buf,
size_t len)
+static void writefd_unbuffered(int fd,
const char *buf,
size_t len)
{
size_t n, total = 0;
{
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;
int maxfd, count, cnt, using_r_fds;
int defer_save = defer_forwarding_messages;
struct timeval tv;
@@
-1051,12
+1051,14
@@
static void writefd_unbuffered(int fd,char *buf,size_t len)
while (total < len) {
FD_ZERO(&w_fds);
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);
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;
if (msg_fd_in > maxfd)
maxfd = msg_fd_in;
using_r_fds = 1;
@@
-1068,7
+1070,7
@@
static void writefd_unbuffered(int fd,char *buf,size_t len)
errno = 0;
count = select(maxfd + 1, using_r_fds ? &r_fds : NULL,
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)
if (count <= 0) {
if (count < 0 && errno == EBADF)
@@
-1077,6
+1079,11
@@
static void writefd_unbuffered(int fd,char *buf,size_t len)
continue;
}
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();
if (using_r_fds && FD_ISSET(msg_fd_in, &r_fds))
read_msg_fd();
@@
-1136,20
+1143,11
@@
static void msg2sndr_flush(void)
while (msg2sndr.head && io_multiplexing_out) {
struct msg_list_item *m = msg2sndr.head;
while (msg2sndr.head && io_multiplexing_out) {
struct msg_list_item *m = msg2sndr.head;
- int tag = (IVAL(m->buf, 0) >> 24) - MPLEX_BASE;
if (!(msg2sndr.head = m->next))
msg2sndr.tail = NULL;
if (!(msg2sndr.head = m->next))
msg2sndr.tail = NULL;
+ stats.total_written += m->len;
defer_forwarding_messages = 1;
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);
}
defer_forwarding_messages = 0;
free(m);
}
@@
-1159,7
+1157,7
@@
static void msg2sndr_flush(void)
* Write an message to a multiplexed stream. If this fails then rsync
* exits.
**/
* Write an message to a multiplexed stream. If this fails then rsync
* exits.
**/
-static void mplex_write(enum msgcode code, char *buf, size_t len)
+static void mplex_write(enum msgcode code, c
onst c
har *buf, size_t len)
{
char buffer[1024];
size_t n = len;
{
char buffer[1024];
size_t n = len;
@@
-1199,7
+1197,7
@@
void io_flush(int flush_it_all)
iobuf_out_cnt = 0;
}
iobuf_out_cnt = 0;
}
-static void writefd(int fd,
char *buf,
size_t len)
+static void writefd(int fd,
const char *buf,
size_t len)
{
if (fd == msg_fd_out) {
rprintf(FERROR, "Internal error: wrong write used in receiver.\n");
{
if (fd == msg_fd_out) {
rprintf(FERROR, "Internal error: wrong write used in receiver.\n");
@@
-1273,13
+1271,13
@@
void write_longint(int f, int64 x)
#endif
}
#endif
}
-void write_buf(int f,
char *buf,
size_t len)
+void write_buf(int f,
const char *buf,
size_t len)
{
writefd(f,buf,len);
}
/** Write a string to the connection */
{
writefd(f,buf,len);
}
/** Write a string to the connection */
-void write_sbuf(int f, char *buf)
+void write_sbuf(int f, c
onst c
har *buf)
{
writefd(f, buf, strlen(buf));
}
{
writefd(f, buf, strlen(buf));
}
@@
-1289,7
+1287,7
@@
void write_byte(int f, uchar c)
writefd(f, (char *)&c, 1);
}
writefd(f, (char *)&c, 1);
}
-void write_vstring(int f, char *str, int len)
+void write_vstring(int f, c
onst c
har *str, int len)
{
uchar lenbuf[3], *lb = lenbuf;
{
uchar lenbuf[3], *lb = lenbuf;
@@
-1372,7
+1370,7
@@
void io_start_multiplex_in(void)
}
/** Write an message to the multiplexed data stream. */
}
/** Write an message to the multiplexed data stream. */
-int io_multiplex_write(enum msgcode code, char *buf, size_t len)
+int io_multiplex_write(enum msgcode code, c
onst c
har *buf, size_t len)
{
if (!io_multiplexing_out)
return 0;
{
if (!io_multiplexing_out)
return 0;