[Patch 1/2] tabled: fix bugs in streaming of data

[Pete Zaitcev <zaitcev@redhat.com>]

This patch fixes 3 plain coding bugs in the streaming mechanism.

1. Whenever we loop back to "restart" label, we have to clear n_iov.
   Otherwise, the loop that fills iov[] writes beyond the end of the
   array. The result is binary garbage on console and a hang.

   Behold the evils of initialized variables. Fix is obvious.

   BTW, array indexes are signed.

2. When we consume a queue element partially, we ought to advance its
   buffer pointer when we decrement the length. Otherwise, we resend
   the data that was sent already, causing a corruption in transfer.

   The fix is obvious (add tmp->buf += sz).

3. The accounting (through unfortunately named write_cnt) was incorrect
   because wr->len may be decremented by the time the element is freed.

Since we're at it, we implement a couple of safety improvements.
First, we change the API to the callback a little. From now on, it
does not have access to the struct client_write anymore, and cannot
accidentially rely on wr->buf (which can be changed now).
Also, we renamed "len" into "togo" and named the new member "length",
to make sure that all use instances for "len" were reviewed.

Signed-Off-By: Pete Zaitcev <zaitcev@redhat.com>

---
 server/object.c |    8 +++-----
 server/server.c |   31 ++++++++++++++++---------------
 server/tabled.h |   11 ++++++-----
 3 files changed, 25 insertions(+), 25 deletions(-)

commit c3671739c217b4b7a63d82543d94318f368a109f
Author: Master <zaitcev@lembas.zaitcev.lan>
Date:   Tue Jan 5 00:04:15 2010 -0700

    Bugfixes for the object streaming to client.

diff --git a/server/object.c b/server/object.c
index 103b36e..d61a446 100644
--- a/server/object.c
+++ b/server/object.c
@@ -983,8 +983,7 @@ void cli_in_end(struct client *cli)
 	stor_close(&cli->in_ce);
 }
 
-static bool object_get_more(struct client *cli, struct client_write *wr,
-			    bool done);
+static bool object_get_more(struct client *cli, void *cb_data, bool done);
 
 /*
  * Return true iff cli_writeq was called. This is compatible with the
@@ -1036,12 +1035,11 @@ err_out:
 }
 
 /* callback from the client side: a queued write is being disposed */
-static bool object_get_more(struct client *cli, struct client_write *wr,
-			    bool done)
+static bool object_get_more(struct client *cli, void *cb_data, bool done)
 {
 
 	/* free now-written buffer */
-	free(wr->cb_data);
+	free(cb_data);
 
 	/* do not queue more, if !completion or fd was closed early */
 	if (!done)	/* FIXME We used to test for input errors here. */
diff --git a/server/server.c b/server/server.c
index 32bb8a4..f16d2ab 100644
--- a/server/server.c
+++ b/server/server.c
@@ -386,10 +386,10 @@ static bool cli_write_free(struct client *cli, struct client_write *tmp,
 {
 	bool rcb = false;
 
-	cli->write_cnt -= tmp->len;
+	cli->write_cnt -= tmp->length;
 	list_del(&tmp->node);
 	if (tmp->cb)
-		rcb = tmp->cb(cli, tmp, done);
+		rcb = tmp->cb(cli, tmp->cb_data, done);
 	free(tmp);
 
 	return rcb;
@@ -487,7 +487,7 @@ static bool cli_evt_recycle(struct client *cli, unsigned int events)
 
 static void cli_writable(struct client *cli)
 {
-	unsigned int n_iov = 0;
+	int n_iov;
 	struct client_write *tmp;
 	ssize_t rc;
 	struct iovec iov[CLI_MAX_WR_IOV];
@@ -497,13 +497,14 @@ restart:
 	more_work = false;
 
 	/* accumulate pending writes into iovec */
+	n_iov = 0;
 	list_for_each_entry(tmp, &cli->write_q, node) {
+		if (n_iov == CLI_MAX_WR_IOV)
+			break;
 		/* bleh, struct iovec should declare iov_base const */
 		iov[n_iov].iov_base = (void *) tmp->buf;
-		iov[n_iov].iov_len = tmp->len;
+		iov[n_iov].iov_len = tmp->togo;
 		n_iov++;
-		if (n_iov == CLI_MAX_WR_IOV)
-			break;
 	}
 
 	/* execute non-blocking write */
@@ -527,14 +528,15 @@ do_write:
 		tmp = list_entry(cli->write_q.next, struct client_write, node);
 
 		/* mark data consumed by decreasing tmp->len */
-		sz = (tmp->len < rc) ? tmp->len : rc;
-		tmp->len -= sz;
+		sz = (tmp->togo < rc) ? tmp->togo : rc;
+		tmp->togo -= sz;
+		tmp->buf += sz;
 		rc -= sz;
 
 		/* if tmp->len reaches zero, write is complete,
 		 * call callback and clean up
 		 */
-		if (tmp->len == 0)
+		if (tmp->togo == 0)
 			if (cli_write_free(cli, tmp, true))
 				more_work = true;
 	}
@@ -600,11 +602,12 @@ int cli_writeq(struct client *cli, const void *buf, unsigned int buflen,
 		return -ENOMEM;
 
 	wr->buf = buf;
-	wr->len = buflen;
+	wr->togo = buflen;
+	wr->length = buflen;
 	wr->cb = cb;
 	wr->cb_data = cb_data;
 	list_add_tail(&wr->node, &cli->write_q);
-	cli->write_cnt += wr->len;
+	cli->write_cnt += buflen;
 
 	return 0;
 }
@@ -645,11 +648,9 @@ do_read:
 	return 0;
 }
 
-bool cli_cb_free(struct client *cli, struct client_write *wr,
-			bool done)
+bool cli_cb_free(struct client *cli, void *cb_data, bool done)
 {
-	free(wr->cb_data);
-
+	free(cb_data);
 	return false;
 }
 
diff --git a/server/tabled.h b/server/tabled.h
index 59c474a..a0d3400 100644
--- a/server/tabled.h
+++ b/server/tabled.h
@@ -103,11 +103,13 @@ struct storage_node {
 };
 
 typedef bool (*cli_evt_func)(struct client *, unsigned int);
-typedef bool (*cli_write_func)(struct client *, struct client_write *, bool);
+typedef bool (*cli_write_func)(struct client *, void *, bool);
 
 struct client_write {
-	const void		*buf;		/* write buffer */
-	int			len;		/* write buffer length */
+	const void		*buf;		/* write buffer pointer */
+	int			togo;		/* write buffer remainder */
+
+	int			length;		/* length for accounting */
 	cli_write_func		cb;		/* callback */
 	void			*cb_data;	/* data passed to cb */
 
@@ -314,8 +316,7 @@ extern bool cli_resp_xml(struct client *cli, int http_status,
 extern int cli_writeq(struct client *cli, const void *buf, unsigned int buflen,
 		     cli_write_func cb, void *cb_data);
 extern size_t cli_wqueued(struct client *cli);
-extern bool cli_cb_free(struct client *cli, struct client_write *wr,
-			bool done);
+extern bool cli_cb_free(struct client *cli, void *cb_data, bool done);
 extern bool cli_write_start(struct client *cli);
 extern int cli_req_avail(struct client *cli);
 extern void applog(int prio, const char *fmt, ...);