summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorNathan Straz <nstraz@redhat.com>2013-09-27 11:32:53 -0400
committerNathan Straz <nstraz@redhat.com>2013-10-02 14:11:33 -0400
commitdb70c6e59af1634d3734019e6e2e17c95f6c6d97 (patch)
tree70e419c16889221d58e2c3949ac0e54c5787bc16
parent1aa4bac6bceb8bd1ce0914134d91f59ed46d4a90 (diff)
downloadqarsh-db70c6e59af1634d3734019e6e2e17c95f6c6d97.zip
qarsh-db70c6e59af1634d3734019e6e2e17c95f6c6d97.tar.gz
qarsh-db70c6e59af1634d3734019e6e2e17c95f6c6d97.tar.xz
Rewrite qacp receive file to handle all packets
-rw-r--r--qacp.c103
-rw-r--r--qarshd.c55
2 files changed, 106 insertions, 52 deletions
diff --git a/qacp.c b/qacp.c
index 295c2d9..34c0399 100644
--- a/qacp.c
+++ b/qacp.c
@@ -260,11 +260,14 @@ void
qacp_recvonefile(const char *host, const char *srcfile, const char *destfile)
{
struct qa_packet *qp;
+ struct qa_packet *rp;
int outfd;
ssize_t nwrote;
- off_t nleft;
struct qp_rstat_pkt *rstatp;
int rstaterrno;
+ fd_set rfds;
+ int nfd;
+ struct timeval timeout;
rstatp = qacp_rstat(srcfile, &rstaterrno);
if (!rstatp) {
@@ -272,7 +275,7 @@ qacp_recvonefile(const char *host, const char *srcfile, const char *destfile)
return;
}
- if ((outfd = open(destfile, O_TRUNC|O_CREAT|O_WRONLY)) <0) {
+ if ((outfd = open(destfile, O_TRUNC|O_CREAT|O_WRONLY)) < 0) {
fprintf(stderr, "Could not open %s: %s\n", destfile, strerror(errno));
close(qacp_fd);
exit(errno);
@@ -288,49 +291,67 @@ qacp_recvonefile(const char *host, const char *srcfile, const char *destfile)
send_packet(qacp_fd, qp);
qpfree(qp);
- /* Read/write file */
- nleft = rstatp->qp_st_size;
- while (nleft > 0) {
- qp = recv_packet(qacp_fd);
- if (qp) {
- assert(qp->qp_type == QP_DATA);
- assert(qp->qp_data.qp_remfd == 0);
- assert(rstatp->qp_st_size - nleft == qp->qp_data.qp_offset);
- nwrote = write(outfd, qp->qp_data.qp_blob, qp->qp_data.qp_count);
- if (nwrote < 0) {
- fprintf(stderr, "write() error: %s\n", strerror(errno));
- break;
+ qp = make_qp_data_allow(0, QARSH_MAX_PACKET_SIZE/2);
+ send_packet(qacp_fd, qp);
+ qpfree(qp);
+
+ for (;;) {
+ FD_ZERO(&rfds);
+ FD_SET(qacp_fd, &rfds);
+
+ timeout.tv_sec = 10;
+ timeout.tv_usec = 0;
+
+ nfd = select(qacp_fd+1, &rfds, NULL, NULL, &timeout);
+ if (nfd < 0) {
+ if (errno == EINTR) {
+ continue;
+ } else {
+ fprintf(stderr, "select errno %d, %s\n", errno, strerror(errno));
+ }
+ } else if (nfd > 0) {
+ if (nfd && FD_ISSET(qacp_fd, &rfds)) {
+ qp = recv_packet(qacp_fd);
+ if (qp == NULL) {
+ fprintf(stderr, "No packet\n");
+ break;
+ }
+ if (qp->qp_type == QP_DATA) {
+ if (qp->qp_data.qp_count == 0) {
+ break;
+ }
+ if ((nwrote = write(outfd, qp->qp_data.qp_blob, qp->qp_data.qp_count)) == -1) {
+ fprintf(stderr, "write() error: %s\n", strerror(errno));
+ rp = make_qp_returncode(-1, errno, strerror(errno));
+ send_packet(qacp_fd, rp);
+ qpfree(rp);
+ break;
+ } else if (nwrote == qp->qp_data.qp_count) {
+ rp = make_qp_data_allow(0, nwrote);
+ send_packet(qacp_fd, rp);
+ qpfree(rp);
+ } else {
+ fprintf(stderr, "short write\n");
+ break;
+ }
+ } else if (qp->qp_type == QP_RETURNCODE) {
+ if (qp->qp_returncode.qp_rc == 0) { /* success */
+ break;
+ } else if (qp->qp_returncode.qp_rc == -1) {
+ fprintf(stderr, "Transfer failed: %s\n",
+ qp->qp_returncode.qp_strerror);
+ exit(125);
+ }
+ } else {
+ fprintf(stderr, "Received unexpected packet\n");
+ dump_qp(qp);
+ break;
+ }
+ qpfree(qp);
}
- assert(nwrote == qp->qp_data.qp_count);
- nleft -= qp->qp_data.qp_count;
- qpfree(qp);
- } else {
- fprintf(stderr, "Did not receive a packet\n");
- break;
}
-
- }
-
- if (nleft != 0) {
- unlink(destfile);
- fprintf(stderr, "Short file transfer of %s, "
- "%lld bytes lost, wanted %lld\n", srcfile,
- (long long int)nleft,
- (long long int)rstatp->qp_st_size);
}
close(outfd);
- free_rstat(rstatp);
-
- /* Await our return code from qarshd */
- qp = recv_packet(qacp_fd);
- if (qp && qp->qp_type == QP_RETURNCODE
- && qp->qp_returncode.qp_rc == -1) {
- fprintf(stderr, "Remote side failed, %s\n",
- qp->qp_returncode.qp_strerror);
- close(qacp_fd);
- exit(125);
- }
- if (qp) qpfree(qp);
if (!quiet) {
printf("%s:%s -> %30.30s\n", host, srcfile, destfile);
diff --git a/qarshd.c b/qarshd.c
index 78b9dda..e8a5ec7 100644
--- a/qarshd.c
+++ b/qarshd.c
@@ -58,6 +58,7 @@ pid_t child_pid;
sigset_t orig_sigmask;
int childfds[3] = { -1, -1, -1 }; /* pipes to child for stdin/stdout/stderr */
int receivefd = -1;
+int sendfd = -1;
off_t recvsize = 0;
off_t received = 0;
@@ -193,6 +194,7 @@ receive_data(struct qa_packet *qp)
if (qp->qp_data.qp_count == 0) { /* EOF */
close(receivefd);
+ receivefd = -1;
if (debug) syslog(LOG_DEBUG, "Transfer complete\n");
return make_qp_returncode(0, 0, "Transfer complete");
}
@@ -210,6 +212,23 @@ receive_data(struct qa_packet *qp)
return make_qp_data_allow(0, nwrote);
}
+struct qa_packet *
+prepare_sendfile(struct qa_packet *qp)
+{
+ int fd;
+
+ syslog(LOG_INFO, "Sending file %s\n", qp->qp_sendfile.qp_path);
+
+ if ((fd = open(qp->qp_sendfile.qp_path, O_RDONLY)) == -1) {
+ syslog(LOG_WARNING, "Could not open %s to send file: %s\n",
+ qp->qp_sendfile.qp_path, strerror(errno));
+ return make_qp_returncode(-1, errno, strerror(errno));
+ }
+
+ sendfd = fd;
+ return NULL;
+}
+
ssize_t
pushfile(const char *path)
{
@@ -485,7 +504,8 @@ handle_packets()
int nfd;
struct timeval timeout;
struct qa_packet *qp = NULL, *rp = NULL;
-
+ const int bufsize = QARSH_MAX_PACKET_SIZE/2;
+ char buf[bufsize];
off_t nbytes;
if (dopause) {
@@ -537,7 +557,7 @@ handle_packets()
send_packet(qoutfd, rp);
qpfree(rp);
break;
- case QP_DATA:
+ case QP_DATA: /* qacp sent some data */
rp = receive_data(qp);
if (rp) {
send_packet(qoutfd, rp);
@@ -545,16 +565,29 @@ handle_packets()
}
break;
case QP_SENDFILE:
- syslog(LOG_INFO, "Sending file %s\n",
- qp->qp_sendfile.qp_path);
- nbytes = pushfile(qp->qp_sendfile.qp_path);
- if (nbytes < 0) {
- rp = make_qp_returncode(-1, errno, strerror(errno));
- } else {
- rp = make_qp_returncode(0, 0, "Transfer Complete");
+ rp = prepare_sendfile(qp);
+ if (rp) {
+ send_packet(qoutfd, rp);
+ qpfree(rp);
}
- send_packet(qoutfd, rp);
- qpfree(rp);
+ break;
+ case QP_DALLOW: /* qacp is ready to receive some data */
+ if (sendfd != -1) {
+ nbytes = read(sendfd, buf, bufsize < qp->qp_dallow.qp_count ? bufsize : qp->qp_dallow.qp_count);
+ if (nbytes == -1) {
+ qp = make_qp_returncode(-1, errno, strerror(errno));
+ } else {
+ rp = make_qp_data(0, 0, nbytes, buf);
+ if (nbytes == 0) {
+ close(sendfd);
+ sendfd = -1;
+ }
+ }
+ send_packet(qoutfd, rp);
+ qpfree(rp);
+ }
+ break;
+ case QP_RETURNCODE: /* qacp either completed or hit an error */
break;
case QP_RSTAT:
if (debug) syslog(LOG_DEBUG, "Got a QP_RSTAT with path = %s\n",