io.c

Socket and pipe I/O utilities used in rsync. [詳細]

ソースコードを見る。

データ構造
struct	flist_ndx_item
struct	flist_ndx_list
struct	msg_list_item
struct	msg_list
関数
static void	read_loop (int fd, char *buf, size_t len)
	Continue trying to read len bytes - don't return until len has been read.
static void	flist_ndx_push (struct flist_ndx_list *lp, int ndx)
static int	flist_ndx_pop (struct flist_ndx_list *lp)
static void	check_timeout (void)
void	io_set_sock_fds (int f_in, int f_out)
void	set_io_timeout (int secs)
void	set_msg_fd_in (int fd)
void	set_msg_fd_out (int fd)
static void	msg_list_add (struct msg_list *lst, int code, char *buf, int len)
static void	read_msg_fd (void)
void	increment_active_files (int ndx, int itemizing, enum logcode code)
void	decrement_active_files (int ndx)
static int	msg2genr_flush (int flush_it_all)
void	send_msg (enum msgcode code, char *buf, int len)
int	get_redo_num (int itemizing, enum logcode code)
int	get_hlink_num (void)
void	io_set_filesfrom_fds (int f_in, int f_out)
	When we're the receiver and we have a local --files-from list of names that needs to be sent over the socket to the sender, we have to do two things at the same time: send the sender a list of what files we're processing and read the incoming file+info list from the sender.
static void	whine_about_eof (int fd)
static int	read_timeout (int fd, char *buf, size_t len)
	Read from a socket with I/O timeout.
int	read_filesfrom_line (int fd, char *fname)
	Read a line into the "fname" buffer (which must be at least MAXPATHLEN characters long).
void	io_start_buffering_out (void)
void	io_start_buffering_in (void)
void	io_end_buffering (void)
void	maybe_flush_socket (void)
void	maybe_send_keepalive (void)
static int	readfd_unbuffered (int fd, char *buf, size_t len)
	Read from the file descriptor handling multiplexing - return number of bytes read.
static void	readfd (int fd, char *buffer, size_t N)
	Do a buffered read from fd.
int	read_shortint (int f)
int32	read_int (int f)
int64	read_longint (int f)
void	read_buf (int f, char *buf, size_t len)
void	read_sbuf (int f, char *buf, size_t len)
uchar	read_byte (int f)
int	read_vstring (int f, char *buf, int bufsize)
void	read_sum_head (int f, struct sum_struct *sum)
void	write_sum_head (int f, struct sum_struct *sum)
static void	sleep_for_bwlimit (int bytes_written)
	Sleep after writing to limit I/O bandwidth usage.
static void	writefd_unbuffered (int fd, char *buf, size_t len)
static void	msg2sndr_flush (void)
static void	mplex_write (enum msgcode code, char *buf, size_t len)
	Write an message to a multiplexed stream.
void	io_flush (int flush_it_all)
static void	writefd (int fd, char *buf, size_t len)
void	write_shortint (int f, int x)
void	write_int (int f, int32 x)
void	write_longint (int f, int64 x)
void	write_buf (int f, char *buf, size_t len)
void	write_sbuf (int f, char *buf)
	Write a string to the connection
void	write_byte (int f, uchar c)
void	write_vstring (int f, char *str, int len)
int	read_line (int f, char *buf, size_t maxlen)
	Read a line of up to maxlen characters into buf (not counting the trailing null).
void	io_printf (int fd, const char *format,...)
void	io_start_multiplex_out (void)
	Setup for multiplexing a MSG_* stream with the data stream.
void	io_start_multiplex_in (void)
	Setup for multiplexing a MSG_* stream with the data stream.
int	io_multiplex_write (enum msgcode code, char *buf, size_t len)
	Write an message to the multiplexed data stream.
void	close_multiplexing_in (void)
void	close_multiplexing_out (void)
	Stop output multiplexing.
void	start_write_batch (int fd)
void	stop_write_batch (void)
変数
int	bwlimit
size_t	bwlimit_writemax
int	io_timeout
int	allowed_lull
int	am_server
int	am_daemon
int	am_sender
int	am_generator
int	eol_nulls
int	read_batch
int	csum_length
int	checksum_seed
int	protocol_version
int	remove_sent_files
int	preserve_hard_links
char *	filesfrom_host
stats	stats
file_list *	the_file_list
const char	phase_unknown [] = "unknown"
int	ignore_timeout = 0
int	batch_fd = -1
int	batch_gen_fd = -1
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	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 char	io_filesfrom_buf [2048]
static char *	io_filesfrom_bp
static char	io_filesfrom_lastchar
static int	io_filesfrom_buflen
static int	defer_forwarding_messages = 0
static int	select_timeout = SELECT_TIMEOUT
static int	active_filecnt = 0
static OFF_T	active_bytecnt = 0
static struct flist_ndx_list redo_list	hlink_list
static struct msg_list msg2genr	msg2sndr
static char *	iobuf_out
static int	iobuf_out_cnt
static char *	iobuf_in
static size_t	iobuf_in_siz

---

説明

Socket and pipe I/O utilities used in rsync.

rsync provides its own multiplexing system, which is used to send stderr and stdout over a single socket. We need this because stdout normally carries the binary data stream, and stderr all our error messages.

For historical reasons this is off during the start of the connection, but it's switched on quite early using io_start_multiplex_out() and io_start_multiplex_in().

io.c で定義されています。

---

関数

static void read_loop	(	int	fd,
		char *	buf,
		size_t	len
	)			[static]

Continue trying to read len bytes - don't return until len has been read.

io.c の 732 行で定義されています。

参照先 read_timeout().

参照元 read_msg_fd()・readfd_unbuffered().

{
        while (len) {
                int n = read_timeout(fd, buf, len);

                buf += n;
                len -= n;
        }
}

static void flist_ndx_push	(	struct flist_ndx_list *	lp,
		int	ndx
	)			[static]

io.c の 119 行で定義されています。

参照先 flist_ndx_list::head・flist_ndx_item::ndx・flist_ndx_item::next・out_of_memory()・flist_ndx_list::tail.

参照元 read_msg_fd().

{
        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

)

[static]

io.c の 134 行で定義されています。

参照先 flist_ndx_list::head・flist_ndx_item::ndx・flist_ndx_item::next・flist_ndx_list::tail.

参照元 get_hlink_num()・get_redo_num().

{
        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 check_timeout

(

void

)

[static]

io.c の 152 行で定義されています。

参照先 am_daemon・am_server・FERROR・ignore_timeout・io_timeout・last_io_in・rprintf().

参照元 msg2genr_flush()・read_filesfrom_line()・read_timeout()・writefd_unbuffered().

{
        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);
        }
}

void io_set_sock_fds	(	int	f_in,
		int	f_out
	)

io.c の 177 行で定義されています。

参照先 sock_f_in・sock_f_out.

参照元 client_run()・start_daemon()・start_server().

{
        sock_f_in = f_in;
        sock_f_out = f_out;
}

void set_io_timeout

(

int

secs

)

io.c の 183 行で定義されています。

参照先 allowed_lull・io_timeout・read_batch・select_timeout.

参照元 rsync_module()・writefd_unbuffered().

{
        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;
}

void set_msg_fd_in

(

int

fd

)

io.c の 199 行で定義されています。

参照先 msg_fd_in.

参照元 client_run()・do_recv().

{
        msg_fd_in = fd;
}

void set_msg_fd_out

(

int

fd

)

io.c の 206 行で定義されています。

参照先 msg_fd_out・set_nonblocking().

参照元 do_recv().

{
        msg_fd_out = fd;
        set_nonblocking(msg_fd_out);
}

static void msg_list_add	(	struct msg_list *	lst,
		int	code,
		char *	buf,
		int	len
	)			[static]

io.c の 213 行で定義されています。

参照先 msg_list::head・m・msg_list_item::next・out_of_memory()・msg_list::tail.

参照元 read_msg_fd()・send_msg().

{
        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->len = len + 4;
        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 read_msg_fd

(

void

)

[static]

io.c の 235 行で定義されています。

参照先 am_generator・am_server・buf・close_multiplexing_out()・decrement_active_files()・defer_forwarding_messages・FERROR・flist_ndx_push()・hlink_list・io_multiplex_write()・msg_list_item::len・msg2sndr・MSG_DELETED・MSG_DONE・MSG_ERROR・msg_fd_in・MSG_INFO・msg_list_add()・MSG_LOG・MSG_REDO・MSG_SOCKERR・MSG_SUCCESS・preserve_hard_links・read_loop()・remove_sent_files・rprintf()・rwrite()・who_am_i().

参照元 get_redo_num()・increment_active_files()・writefd_unbuffered().

{
        char buf[2048];
        size_t n;
        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(). */
        msg_fd_in = -1;

        read_loop(fd, buf, 4);
        tag = IVAL(buf, 0);

        len = tag & 0xFFFFFF;
        tag = (tag >> 24) - MPLEX_BASE;

        switch (tag) {
        case MSG_DONE:
                if (len != 0 || !am_generator) {
                        rprintf(FERROR, "invalid message %d:%d\n", tag, len);
                        exit_cleanup(RERR_STREAMIO);
                }
                flist_ndx_push(&redo_list, -1);
                break;
        case MSG_REDO:
                if (len != 4 || !am_generator) {
                        rprintf(FERROR, "invalid message %d:%d\n", tag, len);
                        exit_cleanup(RERR_STREAMIO);
                }
                read_loop(fd, buf, 4);
                if (remove_sent_files)
                        decrement_active_files(IVAL(buf,0));
                flist_ndx_push(&redo_list, IVAL(buf,0));
                break;
        case MSG_DELETED:
                if (len >= (int)sizeof buf || !am_generator) {
                        rprintf(FERROR, "invalid message %d:%d\n", tag, 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);
                break;
        case MSG_SUCCESS:
                if (len != 4 || !am_generator) {
                        rprintf(FERROR, "invalid message %d:%d\n", tag, len);
                        exit_cleanup(RERR_STREAMIO);
                }
                read_loop(fd, buf, len);
                if (remove_sent_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);
                }
                if (preserve_hard_links)
                        flist_ndx_push(&hlink_list, IVAL(buf,0));
                break;
        case MSG_SOCKERR:
                if (!am_generator) {
                        rprintf(FERROR, "invalid message %d:%d\n", tag, len);
                        exit_cleanup(RERR_STREAMIO);
                }
                close_multiplexing_out();
                /* FALL THROUGH */
        case MSG_INFO:
        case MSG_ERROR:
        case MSG_LOG:
                while (len) {
                        n = len;
                        if (n >= sizeof buf)
                                n = sizeof buf - 1;
                        read_loop(fd, buf, n);
                        if (am_generator && am_server && defer_forwarding_messages)
                                msg_list_add(&msg2sndr, tag, buf, n);
                        else
                                rwrite((enum logcode)tag, buf, n);
                        len -= n;
                }
                break;
        default:
                rprintf(FERROR, "unknown message %d:%d [%s]\n",
                        tag, len, who_am_i());
                exit_cleanup(RERR_STREAMIO);
        }

        msg_fd_in = fd;
}

void increment_active_files	(	int	ndx,
		int	itemizing,
		enum logcode	code
	)

io.c の 331 行で定義されています。

参照先 active_bytecnt・active_filecnt・check_for_finished_hlinks()・file_list::files・hlink_list・file_struct::length・read_msg_fd()・the_file_list.

参照元 recv_generator().

{
        /* TODO: tune these limits? */
        while (active_filecnt >= (active_bytecnt >= 128*1024 ? 10 : 50)) {
                if (hlink_list.head)
                        check_for_finished_hlinks(itemizing, code);
                read_msg_fd();
        }

        active_filecnt++;
        active_bytecnt += the_file_list->files[ndx]->length;
}

void decrement_active_files

(

int

ndx

)

io.c の 344 行で定義されています。

参照先 active_bytecnt・active_filecnt・file_list::files・file_struct::length・the_file_list.

参照元 read_msg_fd().

{
        active_filecnt--;
        active_bytecnt -= the_file_list->files[ndx]->length;
}

static int msg2genr_flush

(

int

flush_it_all

)

[static]

io.c の 353 行で定義されています。

参照先 check_timeout()・errno・msg_list::head・m・msg_fd_out・select_timeout・msg_list::tail.

参照元 io_flush()・read_timeout()・send_msg().

{
        static int written = 0;
        struct timeval tv;
        fd_set fds;

        if (msg_fd_out < 0)
                return -1;

        while (msg2genr.head) {
                struct msg_list_item *m = msg2genr.head;
                int n = write(msg_fd_out, m->buf + written, m->len - written);
                if (n < 0) {
                        if (errno == EINTR)
                                continue;
                        if (errno != EWOULDBLOCK && errno != EAGAIN)
                                return -1;
                        if (!flush_it_all)
                                return 0;
                        FD_ZERO(&fds);
                        FD_SET(msg_fd_out, &fds);
                        tv.tv_sec = select_timeout;
                        tv.tv_usec = 0;
                        if (!select(msg_fd_out+1, NULL, &fds, NULL, &tv))
                                check_timeout();
                } else if ((written += n) == m->len) {
                        msg2genr.head = m->next;
                        if (!msg2genr.head)
                                msg2genr.tail = NULL;
                        free(m);
                        written = 0;
                }
        }
        return 1;
}

void send_msg	(	enum msgcode	code,
		char *	buf,
		int	len
	)

io.c の 389 行で定義されています。

参照先 io_multiplex_write()・msg2genr_flush()・msg_fd_out・msg_list_add().

参照元 do_recv()・handle_delayed_updates()・log_delete()・recv_files()・recv_generator()・rwrite().

{
        if (msg_fd_out < 0) {
                io_multiplex_write(code, buf, len);
                return;
        }
        msg_list_add(&msg2genr, code, buf, len);
        msg2genr_flush(NORMAL_FLUSH);
}

int get_redo_num	(	int	itemizing,
		enum logcode	code
	)

io.c の 399 行で定義されています。

参照先 check_for_finished_hlinks()・flist_ndx_pop()・flist_ndx_list::head・hlink_list・read_msg_fd().

参照元 generate_files().

{
        while (1) {
                if (hlink_list.head)
                        check_for_finished_hlinks(itemizing, code);
                if (redo_list.head)
                        break;
                read_msg_fd();
        }

        return flist_ndx_pop(&redo_list);
}

int get_hlink_num

(

void

)

io.c の 412 行で定義されています。

参照先 flist_ndx_pop()・hlink_list.

参照元 check_for_finished_hlinks().

{
        return flist_ndx_pop(&hlink_list);
}

void io_set_filesfrom_fds	(	int	f_in,
		int	f_out
	)

When we're the receiver and we have a local --files-from list of names that needs to be sent over the socket to the sender, we have to do two things at the same time: send the sender a list of what files we're processing and read the incoming file+info list from the sender.

We do this by augmenting the read_timeout() function to copy this data. It uses the io_filesfrom_buf to read a block of data from f_in (when it is ready, since it might be a pipe) and then blast it out f_out (when it is ready to receive more data).

io.c の 427 行で定義されています。

参照先 io_filesfrom_bp・io_filesfrom_buf・io_filesfrom_buflen・io_filesfrom_f_in・io_filesfrom_f_out・io_filesfrom_lastchar.

参照元 client_run()・do_server_recv().

{
        io_filesfrom_f_in = f_in;
        io_filesfrom_f_out = f_out;
        io_filesfrom_bp = io_filesfrom_buf;
        io_filesfrom_lastchar = '\0';
        io_filesfrom_buflen = 0;
}

static void whine_about_eof

(

int

fd

)

[static]

io.c の 448 行で定義されています。

参照先 FERROR・kluge_around_eof・msleep()・rprintf()・sock_f_in・stats・stats::total_read・who_am_i().

参照元 read_timeout().

{
        if (kluge_around_eof && fd == sock_f_in) {
                int i;
                if (kluge_around_eof > 0)
                        exit_cleanup(0);
                /* If we're still here after 10 seconds, exit with an error. */
                for (i = 10*1000/20; i--; )
                        msleep(20);
        }

        rprintf(FERROR, RSYNC_NAME ": connection unexpectedly closed "
                "(%.0f bytes received so far) [%s]\n",
                (double)stats.total_read, who_am_i());

        exit_cleanup(RERR_STREAMIO);
}

static int read_timeout	(	int	fd,
		char *	buf,
		size_t	len
	)			[static]

Read from a socket with I/O timeout.

return the number of bytes read. If no bytes can be read then exit, never return a number <= 0.

TODO: If the remote shell connection fails, then current versions actually report an "unexpected EOF" error here. Since it's a fairly common mistake to try to use rsh when ssh is required, we should trap that: if we fail to read any data at all, we should give a better explanation. We can tell whether the connection has started by looking e.g. at whether the remote version is known yet.

io.c の 477 行で定義されています。

参照先 check_timeout()・close_multiplexing_out()・eol_nulls・errno・FERROR・FSOCKERR・msg_list::head・io_filesfrom_bp・io_filesfrom_buf・io_filesfrom_buflen・io_filesfrom_f_in・io_filesfrom_f_out・io_filesfrom_lastchar・io_flush()・io_timeout・last_io_in・msg2genr_flush()・msg_fd_out・rsyserr()・select_timeout・sock_f_in・whine_about_eof().

参照元 read_loop()・readfd_unbuffered().

{
        int n, cnt = 0;

        io_flush(NORMAL_FLUSH);

        while (cnt == 0) {
                /* until we manage to read *something* */
                fd_set r_fds, w_fds;
                struct timeval tv;
                int maxfd = fd;
                int count;

                FD_ZERO(&r_fds);
                FD_ZERO(&w_fds);
                FD_SET(fd, &r_fds);
                if (msg2genr.head) {
                        FD_SET(msg_fd_out, &w_fds);
                        if (msg_fd_out > maxfd)
                                maxfd = msg_fd_out;
                }
                if (io_filesfrom_f_out >= 0) {
                        int new_fd;
                        if (io_filesfrom_buflen == 0) {
                                if (io_filesfrom_f_in >= 0) {
                                        FD_SET(io_filesfrom_f_in, &r_fds);
                                        new_fd = io_filesfrom_f_in;
                                } else {
                                        io_filesfrom_f_out = -1;
                                        new_fd = -1;
                                }
                        } else {
                                FD_SET(io_filesfrom_f_out, &w_fds);
                                new_fd = io_filesfrom_f_out;
                        }
                        if (new_fd > maxfd)
                                maxfd = new_fd;
                }

                tv.tv_sec = select_timeout;
                tv.tv_usec = 0;

                errno = 0;

                count = select(maxfd + 1, &r_fds, &w_fds, NULL, &tv);

                if (count <= 0) {
                        if (errno == EBADF)
                                exit_cleanup(RERR_SOCKETIO);
                        check_timeout();
                        continue;
                }

                if (msg2genr.head && FD_ISSET(msg_fd_out, &w_fds))
                        msg2genr_flush(NORMAL_FLUSH);

                if (io_filesfrom_f_out >= 0) {
                        if (io_filesfrom_buflen) {
                                if (FD_ISSET(io_filesfrom_f_out, &w_fds)) {
                                        int l = write(io_filesfrom_f_out,
                                                      io_filesfrom_bp,
                                                      io_filesfrom_buflen);
                                        if (l > 0) {
                                                if (!(io_filesfrom_buflen -= l))
                                                        io_filesfrom_bp = io_filesfrom_buf;
                                                else
                                                        io_filesfrom_bp += l;
                                        } else {
                                                /* XXX should we complain? */
                                                io_filesfrom_f_out = -1;
                                        }
                                }
                        } else if (io_filesfrom_f_in >= 0) {
                                if (FD_ISSET(io_filesfrom_f_in, &r_fds)) {
                                        int l = read(io_filesfrom_f_in,
                                                     io_filesfrom_buf,
                                                     sizeof io_filesfrom_buf);
                                        if (l <= 0) {
                                                /* Send end-of-file marker */
                                                io_filesfrom_buf[0] = '\0';
                                                io_filesfrom_buf[1] = '\0';
                                                io_filesfrom_buflen = io_filesfrom_lastchar? 2 : 1;
                                                io_filesfrom_f_in = -1;
                                        } else {
                                                if (!eol_nulls) {
                                                        char *s = io_filesfrom_buf + l;
                                                        /* Transform CR and/or LF into '\0' */
                                                        while (s-- > io_filesfrom_buf) {
                                                                if (*s == '\n' || *s == '\r')
                                                                        *s = '\0';
                                                        }
                                                }
                                                if (!io_filesfrom_lastchar) {
                                                        /* Last buf ended with a '\0', so don't
                                                         * let this buf start with one. */
                                                        while (l && !*io_filesfrom_bp)
                                                                io_filesfrom_bp++, l--;
                                                }
                                                if (!l)
                                                        io_filesfrom_bp = io_filesfrom_buf;
                                                else {
                                                        char *f = io_filesfrom_bp;
                                                        char *t = f;
                                                        char *eob = f + l;
                                                        /* Eliminate any multi-'\0' runs. */
                                                        while (f != eob) {
                                                                if (!(*t++ = *f++)) {
                                                                        while (f != eob && !*f)
                                                                                f++, l--;
                                                                }
                                                        }
                                                        io_filesfrom_lastchar = f[-1];
                                                }
                                                io_filesfrom_buflen = l;
                                        }
                                }
                        }
                }

                if (!FD_ISSET(fd, &r_fds))
                        continue;

                n = read(fd, buf, len);

                if (n <= 0) {
                        if (n == 0)
                                whine_about_eof(fd); /* Doesn't return. */
                        if (errno == EINTR || errno == EWOULDBLOCK
                            || errno == EAGAIN)
                                continue;

                        /* Don't write errors on a dead socket. */
                        if (fd == sock_f_in) {
                                close_multiplexing_out();
                                rsyserr(FSOCKERR, errno, "read error");
                        } else
                                rsyserr(FERROR, errno, "read error");
                        exit_cleanup(RERR_STREAMIO);
                }

                buf += n;
                len -= n;
                cnt += n;

                if (fd == sock_f_in && io_timeout)
                        last_io_in = time(NULL);
        }

        return cnt;
}

int read_filesfrom_line	(	int	fd,
		char *	fname
	)

Read a line into the "fname" buffer (which must be at least MAXPATHLEN characters long).

io.c の 632 行で定義されています。

参照先 check_timeout()・eol_nulls・errno・filesfrom_host・select_timeout.

参照元 send_file_list().

{
        char ch, *s, *eob = fname + MAXPATHLEN - 1;
        int cnt;
        int reading_remotely = filesfrom_host != NULL;
        int nulls = eol_nulls || reading_remotely;

  start:
        s = fname;
        while (1) {
                cnt = read(fd, &ch, 1);
                if (cnt < 0 && (errno == EWOULDBLOCK
                  || errno == EINTR || errno == EAGAIN)) {
                        struct timeval tv;
                        fd_set fds;
                        FD_ZERO(&fds);
                        FD_SET(fd, &fds);
                        tv.tv_sec = select_timeout;
                        tv.tv_usec = 0;
                        if (!select(fd+1, &fds, NULL, NULL, &tv))
                                check_timeout();
                        continue;
                }
                if (cnt != 1)
                        break;
                if (nulls? !ch : (ch == '\r' || ch == '\n')) {
                        /* Skip empty lines if reading locally. */
                        if (!reading_remotely && s == fname)
                                continue;
                        break;
                }
                if (s < eob)
                        *s++ = ch;
        }
        *s = '\0';

        /* Dump comments. */
        if (*fname == '#' || *fname == ';')
                goto start;

        return s - fname;
}

void io_start_buffering_out

(

void

)

io.c の 678 行で定義されています。

参照先 iobuf_out・iobuf_out_cnt・out_of_memory().

参照元 client_run()・do_recv()・do_server_sender()・io_start_multiplex_out()・send_file_list().

{
        if (iobuf_out)
                return;
        if (!(iobuf_out = new_array(char, IO_BUFFER_SIZE)))
                out_of_memory("io_start_buffering_out");
        iobuf_out_cnt = 0;
}

void io_start_buffering_in

(

void

)

io.c の 690 行で定義されています。

参照先 iobuf_in・iobuf_in_siz・out_of_memory().

参照元 do_server_recv()・do_server_sender()・io_start_multiplex_in().

{
        if (iobuf_in)
                return;
        iobuf_in_siz = 2 * IO_BUFFER_SIZE;
        if (!(iobuf_in = new_array(char, iobuf_in_siz)))
                out_of_memory("io_start_buffering_in");
}

void io_end_buffering

(

void

)

io.c の 699 行で定義されています。

参照先 io_flush()・io_multiplexing_out・iobuf_out.

参照元 send_file_list().

{
        io_flush(NORMAL_FLUSH);
        if (!io_multiplexing_out) {
                free(iobuf_out);
                iobuf_out = NULL;
        }
}

void maybe_flush_socket

(

void

)

io.c の 708 行で定義されています。

参照先 io_flush()・iobuf_out・iobuf_out_cnt・last_io_out.

参照元 generate_files().

{
        if (iobuf_out && iobuf_out_cnt && time(NULL) - last_io_out >= 5)
                io_flush(NORMAL_FLUSH);
}

void maybe_send_keepalive

(

void

)

io.c の 714 行で定義されています。

参照先 allowed_lull・file_list::count・io_flush()・iobuf_out・iobuf_out_cnt・last_io_out・protocol_version・sock_f_out・the_file_list・write_int()・write_shortint().

参照元 delete_in_dir()・generate_files()・receive_sums().

{
        if (time(NULL) - last_io_out >= allowed_lull) {
                if (!iobuf_out || !iobuf_out_cnt) {
                        if (protocol_version < 29)
                                return; /* there's nothing we can do */
                        write_int(sock_f_out, the_file_list->count);
                        write_shortint(sock_f_out, ITEM_IS_NEW);
                }
                if (iobuf_out)
                        io_flush(NORMAL_FLUSH);
        }
}

static int readfd_unbuffered	(	int	fd,
		char *	buf,
		size_t	len
	)			[static]

Read from the file descriptor handling multiplexing - return number of bytes read.

Never returns <= 0.

io.c の 748 行で定義されています。

参照先 FERROR・io_flush()・io_multiplexing_in・iobuf_in・iobuf_in_siz・log_delete()・MSG_DATA・MSG_DELETED・MSG_ERROR・MSG_INFO・MSG_SUCCESS・out_of_memory()・read_loop()・read_timeout()・rprintf()・rwrite()・sock_f_in・successful_send()・who_am_i().

参照元 readfd()・writefd_unbuffered().

{
        static size_t remaining;
        static size_t iobuf_in_ndx;
        size_t msg_bytes;
        int tag, cnt = 0;
        char line[BIGPATHBUFLEN];

        if (!iobuf_in || fd != sock_f_in)
                return read_timeout(fd, buf, len);

        if (!io_multiplexing_in && remaining == 0) {
                remaining = read_timeout(fd, iobuf_in, iobuf_in_siz);
                iobuf_in_ndx = 0;
        }

        while (cnt == 0) {
                if (remaining) {
                        len = MIN(len, remaining);
                        memcpy(buf, iobuf_in + iobuf_in_ndx, len);
                        iobuf_in_ndx += len;
                        remaining -= len;
                        cnt = len;
                        break;
                }

                read_loop(fd, line, 4);
                tag = IVAL(line, 0);

                msg_bytes = tag & 0xFFFFFF;
                tag = (tag >> 24) - MPLEX_BASE;

                switch (tag) {
                case MSG_DATA:
                        if (msg_bytes > iobuf_in_siz) {
                                if (!(iobuf_in = realloc_array(iobuf_in, char,
                                                               msg_bytes)))
                                        out_of_memory("readfd_unbuffered");
                                iobuf_in_siz = msg_bytes;
                        }
                        read_loop(fd, iobuf_in, msg_bytes);
                        remaining = msg_bytes;
                        iobuf_in_ndx = 0;
                        break;
                case MSG_DELETED:
                        if (msg_bytes >= sizeof line)
                                goto overflow;
                        read_loop(fd, line, msg_bytes);
                        /* A directory name was sent with the trailing null */
                        if (msg_bytes > 0 && !line[msg_bytes-1])
                                log_delete(line, S_IFDIR);
                        else {
                                line[msg_bytes] = '\0';
                                log_delete(line, S_IFREG);
                        }
                        break;
                case MSG_SUCCESS:
                        if (msg_bytes != 4) {
                                rprintf(FERROR, "invalid multi-message %d:%ld [%s]\n",
                                        tag, (long)msg_bytes, who_am_i());
                                exit_cleanup(RERR_STREAMIO);
                        }
                        read_loop(fd, line, msg_bytes);
                        successful_send(IVAL(line, 0));
                        break;
                case MSG_INFO:
                case MSG_ERROR:
                        if (msg_bytes >= sizeof line) {
                            overflow:
                                rprintf(FERROR,
                                        "multiplexing overflow %d:%ld [%s]\n",
                                        tag, (long)msg_bytes, who_am_i());
                                exit_cleanup(RERR_STREAMIO);
                        }
                        read_loop(fd, line, msg_bytes);
                        rwrite((enum logcode)tag, line, msg_bytes);
                        break;
                default:
                        rprintf(FERROR, "unexpected tag %d [%s]\n",
                                tag, who_am_i());
                        exit_cleanup(RERR_STREAMIO);
                }
        }

        if (remaining == 0)
                io_flush(NORMAL_FLUSH);

        return cnt;
}

static void readfd	(	int	fd,
		char *	buffer,
		size_t	N
	)			[static]

Do a buffered read from fd.

Don't return until all n bytes have been read. If all n can't be read then exit with an error.

io.c の 843 行で定義されています。

参照先 batch_fd・readfd_unbuffered()・sock_f_in・stats・stats::total_read・write_batch_monitor_in.

参照元 read_buf()・read_byte()・read_int()・read_longint()・read_sbuf()・read_shortint()・read_vstring().

{
        int  cnt;
        size_t total = 0;

        while (total < N) {
                cnt = readfd_unbuffered(fd, buffer + total, N-total);
                total += cnt;
        }

        if (fd == write_batch_monitor_in) {
                if ((size_t)write(batch_fd, buffer, total) != total)
                        exit_cleanup(RERR_FILEIO);
        }

        if (fd == sock_f_in)
                stats.total_read += total;
}

int read_shortint

(

int

f

)

io.c の 862 行で定義されています。

参照先 readfd().

参照元 do_recv()・read_final_goodbye()・read_item_attrs().

{
        uchar b[2];
        readfd(f, (char *)b, 2);
        return (b[1] << 8) + b[0];
}

int32 read_int

(

int

f

)

io.c の 869 行で定義されています。

参照先 readfd().

参照元 do_recv()・get_next_gen_i()・read_final_goodbye()・read_longint()・read_stream_flags()・read_sum_head()・receive_acl()・receive_file_entry()・receive_rsync_acl()・receive_sums()・receive_xattr()・recv_deflated_token()・recv_file_list()・recv_files()・recv_filter_list()・recv_uid_list()・send_files()・setup_protocol()・simple_recv_token().

{
        char b[4];
        int32 num;

        readfd(f,b,4);
        num = IVAL(b,0);
        if (num == (int32)0xffffffff)
                return -1;
        return num;
}

int64 read_longint

(

int

f

)

io.c の 881 行で定義されています。

参照先 FERROR・read_int()・readfd()・rprintf().

参照元 handle_stats()・receive_file_entry().

{
        int64 num;
        char b[8];
        num = read_int(f);

        if ((int32)num != (int32)0xffffffff)
                return num;

#if SIZEOF_INT64 < 8
        rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
        exit_cleanup(RERR_UNSUPPORTED);
#else
        readfd(f,b,8);
        num = IVAL(b,0) | (((int64)IVAL(b,4))<<32);
#endif

        return num;
}

void read_buf	(	int	f,
		char *	buf,
		size_t	len
	)

io.c の 901 行で定義されています。

参照先 readfd().

参照元 read_line()・receive_data()・receive_file_entry()・receive_sums()・receive_xattr()・recv_deflated_token()・simple_recv_token().

{
        readfd(f,buf,len);
}

void read_sbuf	(	int	f,
		char *	buf,
		size_t	len
	)

io.c の 906 行で定義されています。

参照先 readfd().

参照元 receive_file_entry()・recv_filter_list()・recv_uid_list().

{
        readfd(f, buf, len);
        buf[len] = '\0';
}

uchar read_byte

(

int

f

)

io.c の 912 行で定義されています。

参照先 readfd().

参照元 read_item_attrs()・read_vstring()・receive_acl()・receive_file_entry()・receive_rsync_acl()・receive_xattr()・recv_deflated_token()・recv_file_list()・recv_uid_list().

{
        uchar c;
        readfd(f, (char *)&c, 1);
        return c;
}

int read_vstring	(	int	f,
		char *	buf,
		int	bufsize
	)

io.c の 919 行で定義されています。

参照先 FERROR・read_byte()・readfd()・rprintf().

参照元 read_item_attrs().

{
        int len = read_byte(f);

        if (len & 0x80)
                len = (len & ~0x80) * 0x100 + read_byte(f);

        if (len >= bufsize) {
                rprintf(FERROR, "over-long vstring received (%d > %d)\n",
                        len, bufsize - 1);
                return -1;
        }

        if (len)
                readfd(f, buf, len);
        buf[len] = '\0';
        return len;
}

void read_sum_head	(	int	f,
		struct sum_struct *	sum
	)

io.c の 940 行で定義されています。

参照先 sum_struct::blength・sum_struct::count・csum_length・FERROR・protocol_version・read_int()・sum_struct::remainder・rprintf()・sum_struct::s2length・who_am_i().

参照元 receive_data()・receive_sums().

{
        sum->count = read_int(f);
        if (sum->count < 0) {
                rprintf(FERROR, "Invalid checksum count %ld [%s]\n",
                        (long)sum->count, who_am_i());
                exit_cleanup(RERR_PROTOCOL);
        }
        sum->blength = read_int(f);
        if (sum->blength < 0 || sum->blength > MAX_BLOCK_SIZE) {
                rprintf(FERROR, "Invalid block length %ld [%s]\n",
                        (long)sum->blength, who_am_i());
                exit_cleanup(RERR_PROTOCOL);
        }
        sum->s2length = protocol_version < 27 ? csum_length : (int)read_int(f);
        if (sum->s2length < 0 || sum->s2length > MD4_SUM_LENGTH) {
                rprintf(FERROR, "Invalid checksum length %d [%s]\n",
                        sum->s2length, who_am_i());
                exit_cleanup(RERR_PROTOCOL);
        }
        sum->remainder = read_int(f);
        if (sum->remainder < 0 || sum->remainder > sum->blength) {
                rprintf(FERROR, "Invalid remainder length %ld [%s]\n",
                        (long)sum->remainder, who_am_i());
                exit_cleanup(RERR_PROTOCOL);
        }
}

void write_sum_head	(	int	f,
		struct sum_struct *	sum
	)

io.c の 971 行で定義されています。

参照先 sum_struct::blength・sum_struct::count・protocol_version・sum_struct::remainder・sum_struct::s2length・write_int().

参照元 generate_and_send_sums()・recv_generator()・send_files().

{
        static struct sum_struct null_sum;

        if (sum == NULL)
                sum = &null_sum;

        write_int(f, sum->count);
        write_int(f, sum->blength);
        if (protocol_version >= 27)
                write_int(f, sum->s2length);
        write_int(f, sum->remainder);
}

static void sleep_for_bwlimit

(

int

bytes_written

)

[static]

Sleep after writing to limit I/O bandwidth usage.

TODO:

Rather than sleeping after each write, it might be better to use some kind of averaging. The current algorithm seems to always use a bit less bandwidth than specified, because it doesn't make up for slow periods. But arguably this is a feature. In addition, we ought to take the time used to write the data into account.

During some phases of big transfers (file FOO is uptodate) this is called with a small bytes_written every time. As the kernel has to round small waits up to guarantee that we actually wait at least the requested number of microseconds, this can become grossly inaccurate. We therefore keep track of the bytes we've written over time and only sleep when the accumulated delay is at least 1 tenth of a second.

io.c の 1001 行で定義されています。

参照先 bwlimit・bwlimit_writemax・total_written.

参照元 writefd_unbuffered().

{
        static struct timeval prior_tv;
        static long total_written = 0; 
        struct timeval tv, start_tv;
        long elapsed_usec, sleep_usec;

#define ONE_SEC 1000000L /* # of microseconds in a second */

        if (!bwlimit_writemax)
                return;

        total_written += bytes_written; 

        gettimeofday(&start_tv, NULL);
        if (prior_tv.tv_sec) {
                elapsed_usec = (start_tv.tv_sec - prior_tv.tv_sec) * ONE_SEC
                             + (start_tv.tv_usec - prior_tv.tv_usec);
                total_written -= elapsed_usec * bwlimit / (ONE_SEC/1024);
                if (total_written < 0)
                        total_written = 0;
        }

        sleep_usec = total_written * (ONE_SEC/1024) / bwlimit;
        if (sleep_usec < ONE_SEC / 10) {
                prior_tv = start_tv;
                return;
        }

        tv.tv_sec  = sleep_usec / ONE_SEC;
        tv.tv_usec = sleep_usec % ONE_SEC;
        select(0, NULL, NULL, NULL, &tv);

        gettimeofday(&prior_tv, NULL);
        elapsed_usec = (prior_tv.tv_sec - start_tv.tv_sec) * ONE_SEC
                     + (prior_tv.tv_usec - start_tv.tv_usec);
        total_written = (sleep_usec - elapsed_usec) * bwlimit / (ONE_SEC/1024);
}

static void writefd_unbuffered	(	int	fd,
		char *	buf,
		size_t	len
	)			[static]

io.c の 1046 行で定義されています。

参照先 am_generator・bwlimit_writemax・check_timeout()・close_multiplexing_out()・defer_forwarding_messages・errno・FERROR・ignore_timeout・io_filesfrom_buf・io_multiplexing_in・io_timeout・last_io_out・msg_fd_in・msleep()・no_flush・read_msg_fd()・readfd_unbuffered()・rsyserr()・select_timeout・set_io_timeout()・sleep_for_bwlimit()・sock_f_in・sock_f_out・who_am_i().

参照元 io_flush()・mplex_write()・msg2sndr_flush()・writefd().

{
        size_t n, total = 0;
        fd_set w_fds, r_fds;
        int maxfd, count, cnt, using_r_fds;
        int defer_save = defer_forwarding_messages;
        struct timeval tv;

        no_flush++;

        while (total < len) {
                FD_ZERO(&w_fds);
                FD_SET(fd,&w_fds);
                maxfd = fd;

                if (msg_fd_in >= 0) {
                        FD_ZERO(&r_fds);
                        FD_SET(msg_fd_in,&r_fds);
                        if (msg_fd_in > maxfd)
                                maxfd = msg_fd_in;
                        using_r_fds = 1;
                } else
                        using_r_fds = 0;

                tv.tv_sec = select_timeout;
                tv.tv_usec = 0;

                errno = 0;
                count = select(maxfd + 1, using_r_fds ? &r_fds : NULL,
                               &w_fds, NULL, &tv);

                if (count <= 0) {
                        if (count < 0 && errno == EBADF)
                                exit_cleanup(RERR_SOCKETIO);
                        check_timeout();
                        continue;
                }

                if (using_r_fds && FD_ISSET(msg_fd_in, &r_fds))
                        read_msg_fd();

                if (!FD_ISSET(fd, &w_fds))
                        continue;

                n = len - total;
                if (bwlimit_writemax && n > bwlimit_writemax)
                        n = bwlimit_writemax;
                cnt = write(fd, buf + total, n);

                if (cnt <= 0) {
                        if (cnt < 0) {
                                if (errno == EINTR)
                                        continue;
                                if (errno == EWOULDBLOCK || errno == EAGAIN) {
                                        msleep(1);
                                        continue;
                                }
                        }

                        /* Don't try to write errors back across the stream. */
                        if (fd == sock_f_out)
                                close_multiplexing_out();
                        rsyserr(FERROR, errno,
                                "writefd_unbuffered failed to write %ld bytes [%s]",
                                (long)len, who_am_i());
                        /* If the other side is sending us error messages, try
                         * to grab any messages they sent before they died. */
                        while (fd == sock_f_out && io_multiplexing_in) {
                                set_io_timeout(30);
                                ignore_timeout = 0;
                                readfd_unbuffered(sock_f_in, io_filesfrom_buf,
                                                  sizeof io_filesfrom_buf);
                        }
                        exit_cleanup(RERR_STREAMIO);
                }

                total += cnt;
                defer_forwarding_messages = 1;

                if (fd == sock_f_out) {
                        if (io_timeout || am_generator)
                                last_io_out = time(NULL);
                        sleep_for_bwlimit(cnt);
                }
        }

        defer_forwarding_messages = defer_save;
        no_flush--;
}

static void msg2sndr_flush

(

void

)

[static]

io.c の 1136 行で定義されています。

参照先 defer_forwarding_messages・msg_list::head・io_multiplexing_out・m・msg2sndr・sock_f_out・stats・msg_list::tail・stats::total_written・writefd_unbuffered().

参照元 io_flush()・mplex_write().

{
        if (defer_forwarding_messages)
                return;

        while (msg2sndr.head && io_multiplexing_out) {
                struct msg_list_item *m = msg2sndr.head;
                if (!(msg2sndr.head = m->next))
                        msg2sndr.tail = NULL;
                stats.total_written += m->len;
                defer_forwarding_messages = 1;
                writefd_unbuffered(sock_f_out, m->buf, m->len);
                defer_forwarding_messages = 0;
                free(m);
        }
}

static void mplex_write	(	enum msgcode	code,
		char *	buf,
		size_t	len
	)			[static]

Write an message to a multiplexed stream.

If this fails then rsync exits.

io.c の 1157 行で定義されています。

参照先 defer_forwarding_messages・msg2sndr_flush()・sock_f_out・writefd_unbuffered().

参照元 io_flush()・io_multiplex_write().

{
        char buffer[1024];
        size_t n = len;

        SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);

        if (n > sizeof buffer - 4)
                n = 0;
        else
                memcpy(buffer + 4, buf, n);

        writefd_unbuffered(sock_f_out, buffer, n+4);

        len -= n;
        buf += n;

        if (len) {
                defer_forwarding_messages = 1;
                writefd_unbuffered(sock_f_out, buf, len);
                defer_forwarding_messages = 0;
                msg2sndr_flush();
        }
}

void io_flush

(

int

flush_it_all

)

io.c の 1182 行で定義されています。

参照先 io_multiplexing_out・iobuf_out・iobuf_out_cnt・mplex_write()・msg2genr_flush()・msg2sndr_flush()・MSG_DATA・no_flush・sock_f_out・writefd_unbuffered().

参照元 _exit_cleanup()・client_run()・do_recv()・do_server_sender()・io_end_buffering()・io_multiplex_write()・io_start_multiplex_in()・io_start_multiplex_out()・maybe_flush_socket()・maybe_send_keepalive()・read_timeout()・readfd_unbuffered()・wait_process_with_flush()・writefd().

{
        msg2genr_flush(flush_it_all);
        msg2sndr_flush();

        if (!iobuf_out_cnt || no_flush)
                return;

        if (io_multiplexing_out)
                mplex_write(MSG_DATA, iobuf_out, iobuf_out_cnt);
        else
                writefd_unbuffered(sock_f_out, iobuf_out, iobuf_out_cnt);
        iobuf_out_cnt = 0;
}

static void writefd	(	int	fd,
		char *	buf,
		size_t	len
	)			[static]

io.c の 1197 行で定義されています。

参照先 batch_fd・FERROR・io_flush()・iobuf_out・iobuf_out_cnt・msg_fd_out・rprintf()・sock_f_out・stats・stats::total_written・write_batch_monitor_out・writefd_unbuffered().

参照元 write_buf()・write_byte()・write_int()・write_longint()・write_sbuf()・write_shortint()・write_vstring().

{
        if (fd == msg_fd_out) {
                rprintf(FERROR, "Internal error: wrong write used in receiver.\n");
                exit_cleanup(RERR_PROTOCOL);
        }

        if (fd == sock_f_out)
                stats.total_written += len;

        if (fd == write_batch_monitor_out) {
                if ((size_t)write(batch_fd, buf, len) != len)
                        exit_cleanup(RERR_FILEIO);
        }

        if (!iobuf_out || fd != sock_f_out) {
                writefd_unbuffered(fd, buf, len);
                return;
        }

        while (len) {
                int n = MIN((int)len, IO_BUFFER_SIZE - iobuf_out_cnt);
                if (n > 0) {
                        memcpy(iobuf_out+iobuf_out_cnt, buf, n);
                        buf += n;
                        len -= n;
                        iobuf_out_cnt += n;
                }

                if (iobuf_out_cnt == IO_BUFFER_SIZE)
                        io_flush(NORMAL_FLUSH);
        }
}

void write_shortint	(	int	f,
		int	x
	)

io.c の 1231 行で定義されています。

参照先 writefd().

参照元 itemize()・maybe_send_keepalive()・read_final_goodbye()・write_ndx_and_attrs().

{
        uchar b[2];
        b[0] = x;
        b[1] = x >> 8;
        writefd(f, (char *)b, 2);
}

void write_int	(	int	f,
		int32	x
	)

io.c の 1239 行で定義されています。

参照先 writefd().

参照元 do_recv()・generate_and_send_sums()・generate_files()・itemize()・maybe_send_keepalive()・read_final_goodbye()・recv_generator()・send_acl()・send_deflated_token()・send_file_entry()・send_file_list()・send_files()・send_filter_list()・send_ida_list()・send_rsync_acl()・send_rules()・send_uid_list()・send_xattr()・setup_protocol()・simple_send_token()・start_write_batch()・write_longint()・write_ndx_and_attrs()・write_stream_flags()・write_sum_head().

{
        char b[4];
        SIVAL(b,0,x);
        writefd(f,b,4);
}

void write_longint	(	int	f,
		int64	x
	)

io.c の 1250 行で定義されています。

参照先 FERROR・rprintf()・write_int()・writefd().

参照元 handle_stats()・send_file_entry().

{
        char b[8];

        if (x <= 0x7FFFFFFF) {
                write_int(f, (int)x);
                return;
        }

#if SIZEOF_INT64 < 8
        rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
        exit_cleanup(RERR_UNSUPPORTED);
#else
        write_int(f, (int32)0xFFFFFFFF);
        SIVAL(b,0,(x&0xFFFFFFFF));
        SIVAL(b,4,((x>>32)&0xFFFFFFFF));

        writefd(f,b,8);
#endif
}

void write_buf	(	int	f,
		char *	buf,
		size_t	len
	)

io.c の 1271 行で定義されています。

参照先 writefd().

参照元 finish_pre_exec()・generate_and_send_sums()・send_deflated_token()・send_file_entry()・send_rules()・send_uid_list()・send_xattr()・simple_send_token()・write_filter_rules().

{
        writefd(f,buf,len);
}

void write_sbuf	(	int	f,
		char *	buf
	)

Write a string to the connection

io.c の 1277 行で定義されています。

参照先 writefd().

参照元 io_printf()・write_batch_shell_file()・write_filter_rules().

{
        writefd(f, buf, strlen(buf));
}

void write_byte	(	int	f,
		uchar	c
	)

io.c の 1282 行で定義されています。

参照先 writefd().

参照元 finish_pre_exec()・itemize()・rsync_module()・send_acl()・send_deflated_token()・send_file_entry()・send_ida_list()・send_rsync_acl()・send_rules()・send_uid_list()・send_xattr()・write_batch_shell_file()・write_filter_rules()・write_ndx_and_attrs().

{
        writefd(f, (char *)&c, 1);
}

void write_vstring	(	int	f,
		char *	str,
		int	len
	)

io.c の 1287 行で定義されています。

参照先 FERROR・rprintf()・writefd().

参照元 itemize()・write_ndx_and_attrs().

{
        uchar lenbuf[3], *lb = lenbuf;

        if (len > 0x7F) {
                if (len > 0x7FFF) {
                        rprintf(FERROR,
                                "attempting to send over-long vstring (%d > %d)\n",
                                len, 0x7FFF);
                        exit_cleanup(RERR_PROTOCOL);
                }
                *lb++ = len / 0x100 + 0x80;
        }
        *lb = len;

        writefd(f, (char*)lenbuf, lb - lenbuf + 1);
        if (len)
                writefd(f, str, len);
}

int read_line	(	int	f,
		char *	buf,
		size_t	maxlen
	)

Read a line of up to maxlen characters into buf (not counting the trailing null).

Strips the (required) trailing newline and all carriage returns.

戻り値:

1 for success; 0 for I/O error or truncation.

io.c の 1314 行で定義されています。

参照先 read_buf().

参照元 auth_server()・rsync_module()・start_daemon()・start_inband_exchange().

{
        while (maxlen) {
                buf[0] = 0;
                read_buf(f, buf, 1);
                if (buf[0] == 0)
                        return 0;
                if (buf[0] == '\n')
                        break;
                if (buf[0] != '\r') {
                        buf++;
                        maxlen--;
                }
        }
        *buf = '\0';
        return maxlen > 0;
}

void io_printf	(	int	fd,
		const char *	format,
			...
	)

io.c の 1332 行で定義されています。

参照先 buf・FERROR・msg_list_item::len・rprintf()・vsnprintf()・write_sbuf().

参照元 auth_client()・auth_server()・rsync_module()・send_listing()・start_daemon()・start_inband_exchange().

{
        va_list ap;
        char buf[BIGPATHBUFLEN];
        int len;

        va_start(ap, format);
        len = vsnprintf(buf, sizeof buf, format, ap);
        va_end(ap);

        if (len < 0)
                exit_cleanup(RERR_STREAMIO);

        if (len > (int)sizeof buf) {
                rprintf(FERROR, "io_printf() was too long for the buffer.\n");
                exit_cleanup(RERR_STREAMIO);
        }

        write_sbuf(fd, buf);
}

void io_start_multiplex_out

(

void

)

Setup for multiplexing a MSG_* stream with the data stream.

io.c の 1354 行で定義されています。

参照先 io_flush()・io_multiplexing_out・io_start_buffering_out().

参照元 client_run()・rsync_module()・start_server().

{
        io_flush(NORMAL_FLUSH);
        io_start_buffering_out();
        io_multiplexing_out = 1;
}

void io_start_multiplex_in

(

void

)

Setup for multiplexing a MSG_* stream with the data stream.

io.c の 1362 行で定義されています。

参照先 io_flush()・io_multiplexing_in・io_start_buffering_in().

参照元 client_run()・start_inband_exchange()・start_server().

{
        io_flush(NORMAL_FLUSH);
        io_start_buffering_in();
        io_multiplexing_in = 1;
}

int io_multiplex_write	(	enum msgcode	code,
		char *	buf,
		size_t	len
	)

Write an message to the multiplexed data stream.

io.c の 1370 行で定義されています。

参照先 io_flush()・io_multiplexing_out・mplex_write()・stats・stats::total_written.

参照元 read_msg_fd()・rwrite()・send_msg().

{
        if (!io_multiplexing_out)
                return 0;

        io_flush(NORMAL_FLUSH);
        stats.total_written += (len+4);
        mplex_write(code, buf, len);
        return 1;
}

void close_multiplexing_in

(

void

)

io.c の 1381 行で定義されています。

参照先 io_multiplexing_in.

参照元 do_recv().

{
        io_multiplexing_in = 0;
}

void close_multiplexing_out

(

void

)

Stop output multiplexing.

io.c の 1387 行で定義されています。

参照先 io_multiplexing_out.

参照元 do_recv()・read_msg_fd()・read_timeout()・writefd_unbuffered().

{
        io_multiplexing_out = 0;
}

void start_write_batch

(

int

fd

)

io.c の 1392 行で定義されています。

参照先 am_sender・batch_fd・checksum_seed・protocol_version・write_batch_monitor_in・write_batch_monitor_out・write_int()・write_stream_flags().

参照元 client_run().

{
        write_stream_flags(batch_fd);

        /* Some communication has already taken place, but we don't
         * enable batch writing until here so that we can write a
         * canonical record of the communication even though the
         * actual communication so far depends on whether a daemon
         * is involved. */
        write_int(batch_fd, protocol_version);
        write_int(batch_fd, checksum_seed);

        if (am_sender)
                write_batch_monitor_out = fd;
        else
                write_batch_monitor_in = fd;
}

void stop_write_batch

(

void

)

io.c の 1410 行で定義されています。

参照先 write_batch_monitor_in・write_batch_monitor_out.

参照元 do_recv().

{
        write_batch_monitor_out = -1;
        write_batch_monitor_in = -1;
}

---

変数

int bwlimit

options.c の 103 行で定義されています。

参照元 server_options()・sleep_for_bwlimit().

size_t bwlimit_writemax

options.c の 105 行で定義されています。

参照元 sleep_for_bwlimit()・writefd_unbuffered().

int io_timeout

options.c の 84 行で定義されています。

int allowed_lull

options.c の 85 行で定義されています。

int am_server

options.c の 75 行で定義されています。

int am_daemon

options.c の 94 行で定義されています。

int am_sender

options.c の 76 行で定義されています。

int am_generator

options.c の 77 行で定義されています。

参照元 do_recv()・handle_stats()・read_msg_fd()・who_am_i()・writefd_unbuffered().

int eol_nulls

options.c の 90 行で定義されています。

int read_batch

options.c の 139 行で定義されています。

int csum_length

checksum.c の 22 行で定義されています。

int checksum_seed

options.c の 115 行で定義されています。

int protocol_version

mdfour.c の 209 行で定義されています。

int remove_sent_files

options.c の 68 行で定義されています。

int preserve_hard_links

options.c の 47 行で定義されています。

char* filesfrom_host

options.c の 89 行で定義されています。

参照元 client_run()・read_filesfrom_line()・server_options()・start_client().

struct stats stats

log.c の 59 行で定義されています。

struct file_list* the_file_list

main.c の 66 行で定義されています。

const char phase_unknown[] = "unknown"

io.c の 61 行で定義されています。

int ignore_timeout = 0

io.c の 62 行で定義されています。

参照元 check_timeout()・generate_files()・writefd_unbuffered().

int batch_fd = -1

io.c の 63 行で定義されています。

参照元 do_cmd()・handle_stats()・main()・readfd()・send_files()・start_write_batch()・writefd().

int batch_gen_fd = -1

io.c の 64 行で定義されています。

参照元 do_cmd()・recv_files().

int kluge_around_eof = 0

io.c の 67 行で定義されています。

int msg_fd_in = -1

io.c の 69 行で定義されています。

参照元 read_msg_fd()・set_msg_fd_in()・writefd_unbuffered().

int msg_fd_out = -1

io.c の 70 行で定義されています。

参照元 msg2genr_flush()・read_timeout()・rwrite()・send_msg()・set_msg_fd_out()・writefd().

int sock_f_in = -1

io.c の 71 行で定義されています。

参照元 io_set_sock_fds()・read_timeout()・readfd()・readfd_unbuffered()・whine_about_eof()・writefd_unbuffered().

int sock_f_out = -1

io.c の 72 行で定義されています。

参照元 io_flush()・io_set_sock_fds()・itemize()・maybe_send_keepalive()・mplex_write()・msg2sndr_flush()・writefd()・writefd_unbuffered().

int io_multiplexing_out [static]

io.c の 74 行で定義されています。

参照元 close_multiplexing_out()・io_end_buffering()・io_flush()・io_multiplex_write()・io_start_multiplex_out()・msg2sndr_flush().

int io_multiplexing_in [static]

io.c の 75 行で定義されています。

参照元 close_multiplexing_in()・io_start_multiplex_in()・readfd_unbuffered()・writefd_unbuffered().

time_t last_io_in [static]

io.c の 76 行で定義されています。

参照元 check_timeout()・read_timeout().

time_t last_io_out [static]

io.c の 77 行で定義されています。

参照元 maybe_flush_socket()・maybe_send_keepalive()・writefd_unbuffered().

int no_flush [static]

io.c の 78 行で定義されています。

参照元 io_flush()・writefd_unbuffered().

int write_batch_monitor_in = -1 [static]

io.c の 80 行で定義されています。

参照元 readfd()・start_write_batch()・stop_write_batch().

int write_batch_monitor_out = -1 [static]

io.c の 81 行で定義されています。

参照元 start_write_batch()・stop_write_batch()・writefd().

int io_filesfrom_f_in = -1 [static]

io.c の 83 行で定義されています。

参照元 io_set_filesfrom_fds()・read_timeout().

int io_filesfrom_f_out = -1 [static]

io.c の 84 行で定義されています。

参照元 io_set_filesfrom_fds()・read_timeout().

char io_filesfrom_buf[2048] [static]

io.c の 85 行で定義されています。

参照元 io_set_filesfrom_fds()・read_timeout()・writefd_unbuffered().

char* io_filesfrom_bp [static]

io.c の 86 行で定義されています。

参照元 io_set_filesfrom_fds()・read_timeout().

char io_filesfrom_lastchar [static]

io.c の 87 行で定義されています。

参照元 io_set_filesfrom_fds()・read_timeout().

int io_filesfrom_buflen [static]

io.c の 88 行で定義されています。

参照元 io_set_filesfrom_fds()・read_timeout().

int defer_forwarding_messages = 0 [static]

io.c の 89 行で定義されています。

参照元 mplex_write()・msg2sndr_flush()・read_msg_fd()・writefd_unbuffered().

int select_timeout = SELECT_TIMEOUT [static]

io.c の 90 行で定義されています。

参照元 msg2genr_flush()・read_filesfrom_line()・read_timeout()・set_io_timeout()・writefd_unbuffered().

int active_filecnt = 0 [static]

io.c の 91 行で定義されています。

参照元 decrement_active_files()・increment_active_files().

OFF_T active_bytecnt = 0 [static]

io.c の 92 行で定義されています。

参照元 decrement_active_files()・increment_active_files().

struct flist_ndx_list redo_list hlink_list [static]

io.c の 105 行で定義されています。

struct msg_list msg2genr msg2sndr [static]

io.c の 117 行で定義されています。

参照元 msg2sndr_flush()・read_msg_fd().

char* iobuf_out [static]

io.c の 675 行で定義されています。

参照元 io_end_buffering()・io_flush()・io_start_buffering_out()・maybe_flush_socket()・maybe_send_keepalive()・writefd().

int iobuf_out_cnt [static]

io.c の 676 行で定義されています。

参照元 io_flush()・io_start_buffering_out()・maybe_flush_socket()・maybe_send_keepalive()・writefd().

char* iobuf_in [static]

io.c の 687 行で定義されています。

参照元 io_start_buffering_in()・readfd_unbuffered().

size_t iobuf_in_siz [static]

io.c の 688 行で定義されています。

参照元 io_start_buffering_in()・readfd_unbuffered().

---