ctdb-common: Fix a bug in packet reading code for generic socket I/O

queue->offset currently points to the end of available data.  However,
after processing one packet the beginning of the next packet is not
marked explicitly and caused the same packet to be processed again.

BUG: https://bugzilla.samba.org/show_bug.cgi?id=12500

Signed-off-by: Amitay Isaacs <amitay@gmail.com>
Reviewed-by: Martin Schwenke <martin@meltin.net>

diff --git a/ctdb/common/sock_io.c b/ctdb/common/sock_io.c
index b3581fc4721a4be4498b71fceea1b25022e45b7f..7431eec3250bb4678ca40ca1331b8289b05a8c66 100644 (file)
--- a/ctdb/common/sock_io.c
+++ b/ctdb/common/sock_io.c
@@ -76,7 +76,7 @@ struct sock_queue {
        struct tevent_queue *queue;
        struct tevent_fd *fde;
        uint8_t *buf;
-       size_t buflen, offset;
+       size_t buflen, begin, end;
 };
 
 static bool sock_queue_set_fd(struct sock_queue *queue, int fd);
@@ -181,20 +181,20 @@ static void sock_queue_handler(struct tevent_context *ev,
                goto fail;
        }
 
-       if (num_ready > queue->buflen - queue->offset) {
+       if (num_ready > queue->buflen - queue->end) {
                queue->buf = talloc_realloc_size(queue, queue->buf,
-                                                queue->offset + num_ready);
+                                                queue->end + num_ready);
                if (queue->buf == NULL) {
                        goto fail;
                }
-               queue->buflen = queue->offset + num_ready;
+               queue->buflen = queue->end + num_ready;
        }
 
-       nread = sys_read(queue->fd, queue->buf + queue->offset, num_ready);
+       nread = sys_read(queue->fd, queue->buf + queue->end, num_ready);
        if (nread < 0) {
                goto fail;
        }
-       queue->offset += nread;
+       queue->end += nread;
 
        sock_queue_process(queue);
        return;
@@ -207,33 +207,35 @@ static void sock_queue_process(struct sock_queue *queue)
 {
        uint32_t pkt_size;
 
-       if (queue->offset < sizeof(uint32_t)) {
+       if ((queue->end - queue->begin) < sizeof(uint32_t)) {
                /* not enough data */
                return;
        }
 
-       pkt_size = *(uint32_t *)queue->buf;
+       pkt_size = *(uint32_t *)(queue->buf + queue->begin);
        if (pkt_size == 0) {
                D_ERR("Invalid packet of length 0\n");
                queue->callback(NULL, 0, queue->private_data);
        }
 
-       if (queue->offset < pkt_size) {
+       if ((queue->end - queue->begin) < pkt_size) {
                /* not enough data */
                return;
        }
 
-       queue->callback(queue->buf, pkt_size, queue->private_data);
-       queue->offset += pkt_size;
+       queue->callback(queue->buf + queue->begin, pkt_size,
+                       queue->private_data);
+       queue->begin += pkt_size;
 
-       if (queue->offset < queue->buflen) {
+       if (queue->begin < queue->end) {
                /* more data to be processed */
                tevent_schedule_immediate(queue->im, queue->ev,
                                          sock_queue_process_event, queue);
        } else {
                TALLOC_FREE(queue->buf);
                queue->buflen = 0;
-               queue->offset = 0;
+               queue->begin = 0;
+               queue->end = 0;
        }
 }