Add new data packet type

2 files changed, 98 insertions, 1 deletions

diff --git a/qarsh_packet.c b/qarsh_packet.c
index 6617362..eb4eac0 100644
--- a/qarsh_packet.c
+++ b/qarsh_packet.c
@@ -48,6 +48,7 @@ void parse_qp_kill(char *buf, int *buflen, struct qa_packet *qp);
 void parse_qp_recvfile(char *buf, int *buflen, struct qa_packet *qp);
 void parse_qp_sendfile(char *buf, int *buflen, struct qa_packet *qp);
 void parse_qp_rstat(char *buf, int *buflen, struct qa_packet *qp);
+void parse_qp_data(char *buf, int *buflen, struct qa_packet *qp);
 
 char *store_qp_hello(char *buf, struct qa_packet *qp);
 char *store_qp_returncode(char *buf, struct qa_packet *qp);
@@ -59,6 +60,7 @@ char *store_qp_kill(char *buf, struct qa_packet *qp);
 char *store_qp_recvfile(char *buf, struct qa_packet *qp);
 char *store_qp_sendfile(char *buf, struct qa_packet *qp);
 char *store_qp_rstat(char *buf, struct qa_packet *qp);
+char *store_qp_data(char *buf, struct qa_packet *qp);
 
 void free_qp_hello(struct qa_packet *qp);
 void free_qp_returncode(struct qa_packet *qp);
@@ -67,6 +69,7 @@ void free_qp_setuser(struct qa_packet *qp);
 void free_qp_recvfile(struct qa_packet *qp);
 void free_qp_sendfile(struct qa_packet *qp);
 void free_qp_rstat(struct qa_packet *qp);
+void free_qp_data(struct qa_packet *qp);
 
 void dump_qp_ack(struct qa_packet *qp);
 void dump_qp_runcmd(struct qa_packet *qp);
@@ -77,6 +80,7 @@ void dump_qp_kill(struct qa_packet *qp);
 void dump_qp_recvfile(struct qa_packet *qp);
 void dump_qp_sendfile(struct qa_packet *qp);
 void dump_qp_rstat(struct qa_packet *qp);
+void dump_qp_data(struct qa_packet *qp);
 
 
 struct packet_internals {
@@ -152,6 +156,12 @@ struct packet_internals {
 		.pi_store = store_qp_rstat,
 		.pi_free = free_qp_rstat,
 		.pi_dump = dump_qp_rstat
+	}, {
+		.pi_name = "data",
+		.pi_parse = parse_qp_data,
+		.pi_store = store_qp_data,
+		.pi_free = free_qp_data,
+		.pi_dump = dump_qp_data
 	}
 };
 	
@@ -232,6 +242,14 @@ fetch_off_t(char *buf, int *buflen, off_t *out)
 	return buf + sizeof i;
 }
 
+char *
+fetch_void(char *buf, int *buflen, int count, void *out)
+{
+	memcpy(out, buf, count);
+	*buflen -= count;
+	return buf+count;
+}
+
 void
 parse_qp_hello(char *buf, int *buflen, struct qa_packet *qp)
 {
@@ -310,6 +328,22 @@ parse_qp_rstat(char *buf, int *buflen, struct qa_packet *qp)
 	buf = fetch_off_t(buf, buflen, &(qp->qp_rstat.qp_st_size));
 }
 
+void
+parse_qp_data(char *buf, int *buflen, struct qa_packet *qp)
+{
+	buf = fetch_int(buf, buflen, (int *)&(qp->qp_data.qp_remfd));
+	buf = fetch_off_t(buf, buflen, &(qp->qp_data.qp_offset));
+	buf = fetch_int(buf, buflen, (int *)&(qp->qp_data.qp_count));
+	if (qp->qp_data.qp_count > *buflen) {
+		fprintf(stderr, "Blob is larger than rest of packet, %d > %d\n", 
+				qp->qp_data.qp_count, *buflen);
+		free(qp);
+	} else {
+		qp->qp_data.qp_blob = malloc(qp->qp_data.qp_count);
+		qp->qp_data.free_blob = 1;
+		buf = fetch_void(buf, buflen, qp->qp_data.qp_count, qp->qp_data.qp_blob);
+	}
+}
 
 struct qa_packet *
 parse_packet(char *buf, int buflen)
@@ -363,6 +397,15 @@ store_off_t(char *buf, off_t o)
 	return buf + sizeof o;
 }
 
+static char *
+store_void(char *buf, int count, void *d)
+{
+	char *cp;
+	cp = store_int(buf, count);
+	memcpy(cp, d, count);
+	return cp+count;
+}
+
 char *
 store_qp_hello(char *buf, struct qa_packet *qp)
 {
@@ -449,6 +492,15 @@ store_qp_rstat(char *buf, struct qa_packet *qp)
 	return buf;
 }
 
+char *
+store_qp_data(char *buf, struct qa_packet *qp)
+{
+	buf = store_int(buf, qp->qp_data.qp_remfd);
+	buf = store_off_t(buf, qp->qp_data.qp_offset);
+	buf = store_void(buf, qp->qp_data.qp_count, qp->qp_data.qp_blob);
+	return buf;
+}
+
 int
 qptostr(struct qa_packet *qp, char *qpstr, int maxsize)
 {
@@ -628,6 +680,27 @@ make_qp_rstat(const char *path, const struct stat *sb)
 	return qp;
 }
 
+struct qa_packet *
+make_qp_data(const int remfd, const off_t offset, const int count, void *blob)
+{
+	struct qa_packet *qp;
+	qp = malloc(sizeof *qp);
+	assert(qp);
+	memset(qp, 0, sizeof *qp);
+
+	qp->qp_type = QP_DATA;
+	qp->qp_data.qp_remfd = remfd;
+	qp->qp_data.qp_offset = offset;
+	qp->qp_data.qp_count = count;
+	/* data packets are different, the blob is just a pointer which
+	 * must stick around until after the packet is sent.  This is to
+	 * avoid multiple data copies */
+	qp->qp_data.free_blob = 0;
+	qp->qp_data.qp_blob = blob;
+
+	return qp;
+}
+
 /*
  * Packet deallocation functions
  */
@@ -674,6 +747,11 @@ free_qp_rstat(struct qa_packet *qp)
 	condfree(qp->qp_rstat.qp_path);
 }
 
+void
+free_qp_data(struct qa_packet *qp)
+{
+	if (qp->qp_data.free_blob) free(qp->qp_data.qp_blob);
+}
 
 void
 qpfree(struct qa_packet *qp)
@@ -762,6 +840,14 @@ dump_qp_rstat(struct qa_packet *qp)
 }
 
 void
+dump_qp_data(struct qa_packet *qp)
+{
+	fprintf(stderr, "\tremfd: %d\n", qp->qp_data.qp_remfd);
+	fprintf(stderr, "\toffset: %lld\n", (long long int)qp->qp_data.qp_offset);
+	fprintf(stderr, "\tcount: %d\n", qp->qp_data.qp_count);
+}
+
+void
 dump_qp(struct qa_packet *qp)
 {
 	fprintf(stderr, "%s #%d\n", QP_NAME(qp->qp_type), qp->qp_seq);
diff --git a/qarsh_packet.h b/qarsh_packet.h
index ec391c3..42f4b61 100644
--- a/qarsh_packet.h
+++ b/qarsh_packet.h
@@ -36,7 +36,8 @@ enum qa_packet_type {
 	QP_KILL = 7,
 	QP_RECVFILE = 8,
 	QP_SENDFILE = 9,
-	QP_RSTAT = 10
+	QP_RSTAT = 10,
+	QP_DATA = 11
 };
 
 struct qp_hello_pkt {
@@ -96,6 +97,13 @@ struct qp_rstat_pkt {
 	off_t qp_st_size;
 };
 
+struct qp_data_pkt {
+	int qp_remfd;		/* remote fd to write to */
+	off_t qp_offset;	/* offset in file for this data */
+	int qp_count;		/* length of data buffer */
+	int free_blob;		/* Whether qpfree should free the blob */
+	void *qp_blob;
+};
 
 #define QP_VERSION 1
 
@@ -113,6 +121,7 @@ struct qa_packet {
 		struct qp_recvfile_pkt recvfile;
 		struct qp_sendfile_pkt sendfile;
 		struct qp_rstat_pkt rstat;
+		struct qp_data_pkt data;
 	} qp_u;
 };
 
@@ -126,6 +135,7 @@ struct qa_packet {
 #define qp_recvfile qp_u.recvfile
 #define qp_sendfile qp_u.sendfile
 #define qp_rstat qp_u.rstat
+#define qp_data qp_u.data
 
 /* Prototypes */
 char *qp_packet_type(enum qa_packet_type t);
@@ -140,6 +150,7 @@ struct qa_packet *make_qp_kill(int sig);
 struct qa_packet *make_qp_recvfile(const char *path, int if_port, off_t count, mode_t mode);
 struct qa_packet *make_qp_sendfile(const char *path, int of_port);
 struct qa_packet *make_qp_rstat(const char *path, const struct stat *sb);
+struct qa_packet *make_qp_data(const int remfd, const off_t offset, const int count, void *blob);
 int qptostr(struct qa_packet *qp, char *qpstr, int maxsize);
 void qpfree(struct qa_packet *qp);
 void dump_qp(struct qa_packet *qp);