Rewrite send file to handle all packets

This change breaks the qacp protocol!

Before there was a chance we would exit before receiving and checking
all packets from qarshd.  Now we look at all packets and check them.
Use data allow packets and larger buffers.
Handle errors on the write end.

2 files changed, 81 insertions, 52 deletions

diff --git a/qacp.c b/qacp.c
index 3afc35e..295c2d9 100644
--- a/qacp.c
+++ b/qacp.c
@@ -148,6 +148,11 @@ qacp_sendonefile(const char *host, const char *srcfile, const char *destfile)
 	struct stat sb;
 	const int bufsize = QARSH_MAX_PACKET_SIZE/2;
 	char buf[bufsize];
+	fd_set rfds, wfds;
+	int nfd;
+	struct timeval timeout;
+	int dallow;
+	int complete;
 
 
 	if ((fd = open(srcfile, O_RDONLY)) <0) {
@@ -174,59 +179,80 @@ qacp_sendonefile(const char *host, const char *srcfile, const char *destfile)
 	send_packet(qacp_fd, qp);
 	qpfree(qp);
 
-	/* Await our return code from qarshd */
-	qp = recv_packet(qacp_fd);
-	if (qp && qp->qp_type == QP_RETURNCODE) {
-		if (qp->qp_returncode.qp_rc != 0) {
-			fprintf(stderr, "Remote side failed: %s\n",
-					qp->qp_returncode.qp_strerror);
-			qpfree(qp);
-			goto sendone_error;
-		}
-		qpfree(qp);
-	} else {
-		fprintf(stderr, "Did not receive response to recvfile\n");
-		goto sendone_failure;
-	}
-
 	offset = 0;
-	do {
-		nbytes = read(fd, buf, bufsize);
-		if (nbytes < 0) {
-			fprintf(stderr, "read() error: %s\n", strerror(errno));
-			qp = make_qp_returncode(-1, errno, strerror(errno));
-		} else {
-			qp = make_qp_data(0, offset, nbytes, buf);
+	dallow = 0;
+	complete = 0;
+	for (;;) {
+		FD_ZERO(&rfds);
+		FD_SET(qacp_fd, &rfds);
+		FD_ZERO(&wfds);
+		if (!complete && dallow > 0) FD_SET(qacp_fd, &wfds);
+		timeout.tv_sec = 10;
+		timeout.tv_usec = 0;
+
+		nfd = select(qacp_fd+1, &rfds, &wfds, 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_DALLOW) {
+					if (qp->qp_dallow.qp_count > 0) {
+						dallow += qp->qp_dallow.qp_count;
+					} else {
+						dallow = 0;
+					}
+				} 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);
+			}
+			if (nfd && FD_ISSET(qacp_fd, &wfds)) {
+				/* Create a data packet to send */
+				nbytes = read(fd, buf, dallow);
+				if (nbytes < 0) {
+					fprintf(stderr, "read() error: %s\n",
+							strerror(errno));
+					qp = make_qp_returncode(-1, errno, strerror(errno));
+				} else {
+					qp = make_qp_data(0, offset, nbytes, buf);
+					if (nbytes == 0) { /* Transfer is complete */
+						complete = 1;
+					} else {
+						offset += nbytes;
+						dallow -= nbytes;
+					}
+				}
+				send_packet(qacp_fd, qp);
+				qpfree(qp);
+			}
 		}
-		send_packet(qacp_fd, qp);
-		offset += nbytes;
-		qpfree(qp);
-	} while (nbytes > 0);
-sendone_failure:
-
-	close(fd);
-
-	/* 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);
-		qpfree(qp);
-		exit(125);
 	}
-
-	qpfree(qp);
+	close(fd);
 
 	if (!quiet) {
 		printf("%-30.30s  -> %s:%s\n", srcfile, host, destfile);
 	}
 
 	return;
-sendone_error:
-	close(qacp_fd);
-	exit(125);
 }
 
 
diff --git a/qarshd.c b/qarshd.c
index d686078..78b9dda 100644
--- a/qarshd.c
+++ b/qarshd.c
@@ -183,13 +183,13 @@ prepare_recvfile(struct qa_packet *qp)
 	recvsize = qp->qp_recvfile.qp_count;
 	received = 0;
 
-	return make_qp_returncode(0, 0, "Ready to receive");
+	return make_qp_data_allow(0, QARSHD_BUFSIZE);
 }
 
 struct qa_packet *
 receive_data(struct qa_packet *qp)
 {
-	ssize_t nwrote;
+	ssize_t ret, nwrote;
 
 	if (qp->qp_data.qp_count == 0) { /* EOF */
 		close(receivefd);
@@ -198,13 +198,16 @@ receive_data(struct qa_packet *qp)
 	}
 	if (debug) syslog(LOG_DEBUG, "Data for %d, %d@%ld\n",
 			receivefd, qp->qp_data.qp_count, qp->qp_data.qp_offset);
-	nwrote = write(receivefd, qp->qp_data.qp_blob, qp->qp_data.qp_count);
-
-	if (nwrote < 0) {
-		return make_qp_returncode(-1, errno, strerror(errno));
-	}
+	nwrote = 0;
+	do {
+		ret = write(receivefd, qp->qp_data.qp_blob + nwrote, qp->qp_data.qp_count - nwrote);
+		if (ret < 0) {
+			return make_qp_returncode(-1, errno, strerror(errno));
+		} 
+		nwrote += ret;
+	} while (nwrote < qp->qp_data.qp_count);
 	received += nwrote;
-	return NULL;
+	return make_qp_data_allow(0, nwrote);
 }
 
 ssize_t