jsonrpc: Maximize receive buffer size.

Previously, we reused the jsonrpc byteq with the state from a previous
iteration with potentially less headroom. However, in case we have an
empty byteq, we can forward the pointers to maximize the available
headroom and thus, the number of bytes that we can process in a single
iteration.

Co-authored-by: Felix Huettner <felix.huettner@stackit.cloud>
Signed-off-by: Felix Huettner <felix.huettner@stackit.cloud>
Signed-off-by: Martin Morgenstern <martin.morgenstern@cloudandheat.com>
Signed-off-by: Ilya Maximets <i.maximets@ovn.org>

lib/byteq.c

@@ -198,3 +198,21 @@ byteq_advance_head(struct byteq *q, unsigned int n)
     ovs_assert(byteq_headroom(q) >= n);
     q->head += n;
 }
+
+/* Move the head and tail pointers forward to have the most headroom available.
+ * Can only be used on an empty byteq.  This is equivalent to advancing both
+ * head and tail by the current headroom size.
+ * Previous pointers returned by byteq_tail() or byteq_head() are potentially
+ * invalid afterwards. */
+void
+byteq_fast_forward(struct byteq *q)
+{
+    ovs_assert(byteq_is_empty(q));
+    unsigned int pos = q->head & (q->size - 1);
+
+    if (pos) {
+        /* Only advance head if we are not already at a multiple of size. */
+        q->head += q->size - pos;
+        q->tail = q->head;
+    }
+}

lib/byteq.h

@@ -46,5 +46,6 @@ void byteq_advance_head(struct byteq *, unsigned int n);
 int byteq_tailroom(const struct byteq *);
 const uint8_t *byteq_tail(const struct byteq *);
 void byteq_advance_tail(struct byteq *, unsigned int n);
+void byteq_fast_forward(struct byteq *);
 
 #endif /* byteq.h */

lib/jsonrpc.c

@@ -343,6 +343,7 @@ jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
             size_t chunk;
             int retval;
 
+            byteq_fast_forward(&rpc->input);
             chunk = byteq_headroom(&rpc->input);
             retval = stream_recv(rpc->stream, byteq_head(&rpc->input), chunk);
             if (retval < 0) {

tests/library.at

@@ -90,7 +90,7 @@ AT_CLEANUP
 
 AT_SETUP([byteq - basic])
 AT_KEYWORDS([byteq])
-AT_CHECK([ovstest test-byteq basic], [0], [...
+AT_CHECK([ovstest test-byteq basic], [0], [....
 ])
 AT_CLEANUP
 

tests/test-byteq.c

@@ -29,6 +29,7 @@ static void test_byteq_main(int argc, char *argv[]);
 static void test_byteq_put_get(void);
 static void test_byteq_putn_get(void);
 static void test_byteq_put_string(void);
+static void test_byteq_fast_forward(void);
 static void test_byteq_write_read(void);
 
 #define SIZE 256
@@ -80,6 +81,36 @@ test_byteq_put_string(void)
     }
 }
 
+static void
+test_byteq_fast_forward(void)
+{
+    struct byteq bq;
+    uint8_t buffer[SIZE];
+    unsigned int head;
+    unsigned int tail;
+    const char *input = "Open vSwitch";
+    const int input_len = strlen(input);
+
+    byteq_init(&bq, buffer, SIZE);
+    byteq_putn(&bq, input, input_len);
+    for (int i = 0; i < input_len; i++) {
+        ovs_assert(byteq_get(&bq) == input[i]);
+    }
+
+    ovs_assert(byteq_is_empty(&bq));
+    ovs_assert(byteq_headroom(&bq) < SIZE);
+
+    head = bq.head;
+    tail = bq.tail;
+
+    byteq_fast_forward(&bq);
+
+    ovs_assert(byteq_headroom(&bq) == SIZE);
+    ovs_assert(bq.head > head);
+    ovs_assert(bq.tail > tail);
+    ovs_assert(bq.head == bq.tail);
+}
+
 static void
 test_byteq_write_read(void)
 {
@@ -147,6 +178,7 @@ test_byteq_main(int argc, char *argv[])
         run_test(test_byteq_put_get);
         run_test(test_byteq_putn_get);
         run_test(test_byteq_put_string);
+        run_test(test_byteq_fast_forward);
         printf("\n");
     } else {
         ovs_fatal(0, "invalid argument\n"