Rework isccc_ccmsg to support multiple messages per tcp read

Previously, only a single controlconf message would be processed from a
single TCP read even if the TCP read buffer contained multiple messages.
Refactor the isccc_ccmsg unit to store the extra buffer in the internal
buffer and use the already read data first before reading from the
network again.

Co-authored-by: Ondřej Surý <ondrej@isc.org>
Co-authored-by: Dominik Thalhammer <dominik@thalhammer.it>

diff --git a/bin/rndc/rndc.c b/bin/rndc/rndc.c
index 2e5ae8eead6a5f139c1fd5ef47d4029228c5dd0b..1d35ded62b5d4c4287ae98c6512fca492fe4e15f 100644 (file)
--- a/bin/rndc/rndc.c
+++ b/bin/rndc/rndc.c
@@ -305,8 +305,7 @@ rndc_recvdone(isc_nmhandle_t *handle, isc_result_t result, void *arg) {
                fatal("recv failed: %s", isc_result_totext(result));
        }
 
-       source.rstart = isc_buffer_base(ccmsg->buffer);
-       source.rend = isc_buffer_used(ccmsg->buffer);
+       isccc_ccmsg_toregion(ccmsg, &source);
 
        DO("parse message",
           isccc_cc_fromwire(&source, &response, algorithm, &secret));
@@ -381,8 +380,7 @@ rndc_recvnonce(isc_nmhandle_t *handle ISC_ATTR_UNUSED, isc_result_t result,
                fatal("recv failed: %s", isc_result_totext(result));
        }
 
-       source.rstart = isc_buffer_base(ccmsg->buffer);
-       source.rend = isc_buffer_used(ccmsg->buffer);
+       isccc_ccmsg_toregion(ccmsg, &source);
 
        DO("parse message",
           isccc_cc_fromwire(&source, &response, algorithm, &secret));

diff --git a/lib/isccc/ccmsg.c b/lib/isccc/ccmsg.c
index 4c5ff61e5f49e87ec2314f0f50fca0107ce5737e..d964a8f123f28138488b387b081d0a9c1041bb2d 100644 (file)
--- a/lib/isccc/ccmsg.c
+++ b/lib/isccc/ccmsg.c
@@ -42,6 +42,35 @@
 #define CCMSG_MAGIC     ISC_MAGIC('C', 'C', 'm', 's')
 #define VALID_CCMSG(foo) ISC_MAGIC_VALID(foo, CCMSG_MAGIC)
 
+/*
+ * Try parsing a message from the internal read_buffer and set state
+ * accordingly. Returns true if a message was successfully parsed, false if not.
+ * If no message could be parsed the ccmsg struct remains untouched.
+ */
+static isc_result_t
+try_parse_message(isccc_ccmsg_t *ccmsg) {
+       REQUIRE(ccmsg != NULL);
+
+       uint32_t len = 0;
+       if (isc_buffer_peekuint32(ccmsg->buffer, &len) != ISC_R_SUCCESS) {
+               return ISC_R_NOMORE;
+       }
+       if (len == 0) {
+               return ISC_R_UNEXPECTEDEND;
+       }
+       if (len > ccmsg->maxsize) {
+               return ISC_R_RANGE;
+       }
+       if (isc_buffer_remaininglength(ccmsg->buffer) < sizeof(uint32_t) + len)
+       {
+               return ISC_R_NOMORE;
+       }
+       /* Skip the size we just peeked */
+       isc_buffer_forward(ccmsg->buffer, sizeof(uint32_t));
+       ccmsg->size = len;
+       return ISC_R_SUCCESS;
+}
+
 static void
 recv_data(isc_nmhandle_t *handle, isc_result_t eresult, isc_region_t *region,
          void *arg) {
@@ -56,50 +85,20 @@ recv_data(isc_nmhandle_t *handle, isc_result_t eresult, isc_region_t *region,
 
        REQUIRE(region != NULL);
 
-       if (!ccmsg->length_received) {
-               if (region->length < sizeof(uint32_t)) {
-                       eresult = ISC_R_UNEXPECTEDEND;
-                       goto done;
-               }
-
-               ccmsg->size = ntohl(*(uint32_t *)region->base);
-
-               if (ccmsg->size == 0) {
-                       eresult = ISC_R_UNEXPECTEDEND;
-                       goto done;
-               }
-               if (ccmsg->size > ccmsg->maxsize) {
-                       eresult = ISC_R_RANGE;
-                       goto done;
-               }
-
-               isc_region_consume(region, sizeof(uint32_t));
-               isc_buffer_allocate(ccmsg->mctx, &ccmsg->buffer, ccmsg->size);
-
-               ccmsg->length_received = true;
+       /* Copy the received data to our reassembly buffer */
+       eresult = isc_buffer_copyregion(ccmsg->buffer, region);
+       if (eresult != ISC_R_SUCCESS) {
+               goto done;
        }
+       isc_region_consume(region, region->length);
 
-       /*
-        * If there's no more data, wait for more
-        */
-       if (region->length == 0) {
+       /* Try to parse a single message of the buffer */
+       eresult = try_parse_message(ccmsg);
+       /* No results from parsing, we need more data */
+       if (eresult == ISC_R_NOMORE) {
                return;
        }
 
-       /* We have some data in the buffer, read it */
-
-       size_t size = ISC_MIN(isc_buffer_availablelength(ccmsg->buffer),
-                             region->length);
-       isc_buffer_putmem(ccmsg->buffer, region->base, size);
-       isc_region_consume(region, size);
-
-       if (isc_buffer_usedlength(ccmsg->buffer) == ccmsg->size) {
-               goto done;
-       }
-
-       /* Wait for more data to come */
-       return;
-
 done:
        isc_nm_read_stop(handle);
        ccmsg->recv_cb(handle, eresult, ccmsg->recv_cbarg);
@@ -120,6 +119,10 @@ isccc_ccmsg_init(isc_mem_t *mctx, isc_nmhandle_t *handle,
                .mctx = mctx,
        };
 
+       /* Preallocate the buffer to maximum single TCP read */
+       isc_buffer_allocate(ccmsg->mctx, &ccmsg->buffer,
+                           UINT16_MAX + sizeof(uint16_t));
+
        isc_nmhandle_attach(handle, &ccmsg->handle);
 }
 
@@ -134,15 +137,25 @@ void
 isccc_ccmsg_readmessage(isccc_ccmsg_t *ccmsg, isc_nm_cb_t cb, void *cbarg) {
        REQUIRE(VALID_CCMSG(ccmsg));
 
-       if (ccmsg->buffer != NULL) {
-               isc_buffer_free(&ccmsg->buffer);
+       if (ccmsg->size != 0) {
+               /* Remove the previously read message from the buffer */
+               isc_buffer_forward(ccmsg->buffer, ccmsg->size);
+               ccmsg->size = 0;
+               isc_buffer_trycompact(ccmsg->buffer);
        }
 
        ccmsg->recv_cb = cb;
        ccmsg->recv_cbarg = cbarg;
-       ccmsg->length_received = false;
 
-       isc_nm_read(ccmsg->handle, recv_data, ccmsg);
+       /* If we have previous data still in the buffer, try to parse it */
+       isc_result_t result = try_parse_message(ccmsg);
+       if (result == ISC_R_NOMORE) {
+               /* We need to read more data */
+               isc_nm_read(ccmsg->handle, recv_data, ccmsg);
+               return;
+       }
+
+       ccmsg->recv_cb(ccmsg->handle, result, ccmsg->recv_cbarg);
 }
 
 static void
@@ -187,18 +200,19 @@ isccc_ccmsg_disconnect(isccc_ccmsg_t *ccmsg) {
 void
 isccc_ccmsg_invalidate(isccc_ccmsg_t *ccmsg) {
        REQUIRE(VALID_CCMSG(ccmsg));
+       REQUIRE(ccmsg->handle == NULL);
 
        ccmsg->magic = 0;
 
-       if (ccmsg->buffer != NULL) {
-               isc_buffer_free(&ccmsg->buffer);
-       }
+       isc_buffer_free(&ccmsg->buffer);
 }
 
 void
 isccc_ccmsg_toregion(isccc_ccmsg_t *ccmsg, isccc_region_t *ccregion) {
        REQUIRE(VALID_CCMSG(ccmsg));
+       REQUIRE(ccmsg->buffer);
+       REQUIRE(isc_buffer_remaininglength(ccmsg->buffer) >= ccmsg->size);
 
-       ccregion->rstart = isc_buffer_base(ccmsg->buffer);
-       ccregion->rend = isc_buffer_used(ccmsg->buffer);
+       ccregion->rstart = isc_buffer_current(ccmsg->buffer);
+       ccregion->rend = ccregion->rstart + ccmsg->size;
 }

diff --git a/lib/isccc/include/isccc/ccmsg.h b/lib/isccc/include/isccc/ccmsg.h
index 5120732901cfe3b6cf8da7b54b3f6941615f2ce2..07ca331f1e46e41ab3bed5c34d99a85c1291a84b 100644 (file)
--- a/lib/isccc/include/isccc/ccmsg.h
+++ b/lib/isccc/include/isccc/ccmsg.h
@@ -45,7 +45,6 @@ typedef struct isccc_ccmsg {
        /* private (don't touch!) */
        unsigned int    magic;
        uint32_t        size;
-       bool            length_received;
        isc_buffer_t   *buffer;
        unsigned int    maxsize;
        isc_mem_t      *mctx;