Improve the QUIC_RSTREAM implementation

Add API calls to avoid copying data when reading
These are ossl_quic_rstream_get_record() and
ossl_quic_rstream_release_record().

Add side storage for the stream frame data.
When there are too many packets referenced by the
receiving stream the function ossl_quic_rstream_move_to_rbuf()
can be called to move the data to a ring buffer.

Reviewed-by: Matt Caswell <matt@openssl.org>
Reviewed-by: Hugo Landau <hlandau@openssl.org>
(Merged from https://github.com/openssl/openssl/pull/19794)

4 files changed, 251 insertions, 189 deletions

diff --git a/ssl/quic/quic_channel.c b/ssl/quic/quic_channel.c
index 1a95c0e10d..7b79e9e500 100644
--- a/ssl/quic/quic_channel.c
+++ b/ssl/quic/quic_channel.c
@@ -208,7 +208,7 @@ static int ch_init(QUIC_CHANNEL *ch)
         goto err;
 
     for (pn_space = QUIC_PN_SPACE_INITIAL; pn_space < QUIC_PN_SPACE_NUM; ++pn_space) {
-        ch->crypto_recv[pn_space] = ossl_quic_rstream_new(NULL, NULL);
+        ch->crypto_recv[pn_space] = ossl_quic_rstream_new(NULL, NULL, 0);
         if (ch->crypto_recv[pn_space] == NULL)
             goto err;
     }
@@ -221,7 +221,7 @@ static int ch_init(QUIC_CHANNEL *ch)
     if ((ch->stream0->sstream = ossl_quic_sstream_new(INIT_APP_BUF_LEN)) == NULL)
         goto err;
 
-    if ((ch->stream0->rstream = ossl_quic_rstream_new(NULL, NULL)) == NULL)
+    if ((ch->stream0->rstream = ossl_quic_rstream_new(NULL, NULL, 0)) == NULL)
         goto err;
 
     if (!ossl_quic_txfc_init(&ch->stream0->txfc, &ch->conn_txfc))
diff --git a/ssl/quic/quic_rstream.c b/ssl/quic/quic_rstream.c
index 669f8d20da..b35bd983af 100644
--- a/ssl/quic/quic_rstream.c
+++ b/ssl/quic/quic_rstream.c
@@ -6,25 +6,35 @@
 * in the file LICENSE in the source distribution or at
 * https://www.openssl.org/source/license.html
 */
+#include <openssl/err.h>
 #include "internal/common.h"
 #include "internal/time.h"
 #include "internal/quic_stream.h"
 #include "internal/quic_sf_list.h"
+#include "internal/ring_buf.h"
 
 struct quic_rstream_st {
     SFRAME_LIST fl;
     QUIC_RXFC *rxfc;
     OSSL_STATM *statm;
+    UINT_RANGE head_range;
+    struct ring_buf rbuf;
 };
 
 QUIC_RSTREAM *ossl_quic_rstream_new(QUIC_RXFC *rxfc,
-                                    OSSL_STATM *statm)
+                                    OSSL_STATM *statm, size_t rbuf_size)
 {
-    QUIC_RSTREAM *ret = OPENSSL_malloc(sizeof(*ret));
+    QUIC_RSTREAM *ret = OPENSSL_zalloc(sizeof(*ret));
 
     if (ret == NULL)
         return NULL;
 
+    ring_buf_init(&ret->rbuf);
+    if (!ring_buf_resize(&ret->rbuf, rbuf_size)) {
+        OPENSSL_free(ret);
+        return NULL;
+    }
+
     ossl_sframe_list_init(&ret->fl);
     ret->rxfc = rxfc;
     ret->statm = statm;
@@ -37,6 +47,7 @@ void ossl_quic_rstream_free(QUIC_RSTREAM *qrs)
         return;
 
     ossl_sframe_list_destroy(&qrs->fl);
+    ring_buf_destroy(&qrs->rbuf);
     OPENSSL_free(qrs);
 }
 
@@ -47,6 +58,12 @@ int ossl_quic_rstream_queue_data(QUIC_RSTREAM *qrs, OSSL_QRX_PKT *pkt,
 {
     UINT_RANGE range;
 
+    if ((data == NULL && data_len != 0) || (data_len == 0 && fin == 0)) {
+        /* empty frame allowed only at the end of the stream */
+        ERR_raise(ERR_LIB_SSL, ERR_R_INTERNAL_ERROR);
+        return 0;
+    }
+
     range.start = offset;
     range.end = offset + data_len;
 
@@ -66,10 +83,34 @@ static int read_internal(QUIC_RSTREAM *qrs, unsigned char *buf, size_t size,
     while (ossl_sframe_list_peek(&qrs->fl, &iter, &range, &data, &fin_)) {
         size_t l = (size_t)(range.end - range.start);
 
-        if (l > size)
+        if (l > size) {
             l = size;
-        memcpy(buf, data, l);
+            fin_ = 0;
+        }
         offset = range.start + l;
+        if (l == 0)
+            break;
+
+        if (data == NULL) {
+            size_t max_len;
+
+            data = ring_buf_get_ptr(&qrs->rbuf, range.start, &max_len);
+            if (!ossl_assert(data != NULL))
+                return 0;
+            if (max_len < l) {
+                memcpy(buf, data, max_len);
+                size -= max_len;
+                buf += max_len;
+                readbytes_ += max_len;
+                l -= max_len;
+                data = ring_buf_get_ptr(&qrs->rbuf, range.start + max_len,
+                                        &max_len);
+                if (!ossl_assert(data != NULL) || !ossl_assert(max_len > l))
+                    return 0;
+            }
+        }
+
+        memcpy(buf, data, l);
         size -= l;
         buf += l;
         readbytes_ += l;
@@ -77,8 +118,10 @@ static int read_internal(QUIC_RSTREAM *qrs, unsigned char *buf, size_t size,
             break;
     }
 
-    if (drop && offset != 0)
+    if (drop && offset != 0) {
         ret = ossl_sframe_list_drop_frames(&qrs->fl, offset);
+        ring_buf_cpop_range(&qrs->rbuf, 0, offset - 1);
+    }
 
     if (ret) {
         *readbytes = readbytes_;
@@ -88,8 +131,7 @@ static int read_internal(QUIC_RSTREAM *qrs, unsigned char *buf, size_t size,
     return ret;
 }
 
-int ossl_quic_rstream_read(QUIC_RSTREAM *qrs, unsigned char *buf, size_t size,
-                           size_t *readbytes, int *fin)
+static OSSL_TIME get_rtt(QUIC_RSTREAM *qrs)
 {
     OSSL_TIME rtt;
 
@@ -101,6 +143,13 @@ int ossl_quic_rstream_read(QUIC_RSTREAM *qrs, unsigned char *buf, size_t size,
     } else {
         rtt = ossl_time_zero();
     }
+    return rtt;
+}
+
+int ossl_quic_rstream_read(QUIC_RSTREAM *qrs, unsigned char *buf, size_t size,
+                           size_t *readbytes, int *fin)
+{
+    OSSL_TIME rtt = get_rtt(qrs);
 
     if (!read_internal(qrs, buf, size, readbytes, fin, 1))
         return 0;
@@ -135,3 +184,105 @@ int ossl_quic_rstream_available(QUIC_RSTREAM *qrs, size_t *avail, int *fin)
 #endif
     return 1;
 }
+
+int ossl_quic_rstream_get_record(QUIC_RSTREAM *qrs,
+                                 const unsigned char **record, size_t *rec_len,
+                                 int *fin)
+{
+    const unsigned char *record_ = NULL;
+    size_t rec_len_, max_len;
+
+    if (!ossl_sframe_list_lock_head(&qrs->fl, &qrs->head_range, &record_, fin)) {
+        /* No head frame to lock and return */
+        *record = NULL;
+        *rec_len = 0;
+        return 1;
+    }
+
+    /* if final empty frame, we drop it immediately */
+    if (qrs->head_range.end == qrs->head_range.start) {
+        if (!ossl_assert(*fin))
+            return 0;
+        if (!ossl_sframe_list_drop_frames(&qrs->fl, qrs->head_range.end))
+            return 0;
+    }
+
+    rec_len_ = (size_t)(qrs->head_range.end - qrs->head_range.start);
+
+    if (record_ == NULL && rec_len_ != 0) {
+        record_ = ring_buf_get_ptr(&qrs->rbuf, qrs->head_range.start,
+                                   &max_len);
+        if (!ossl_assert(record_ != NULL))
+            return 0;
+        if (max_len < rec_len_) {
+            rec_len_ = max_len;
+            qrs->head_range.end = qrs->head_range.start + max_len;
+        }
+    }
+
+    *rec_len = rec_len_;
+    *record = record_;
+    return 1;
+}
+
+
+int ossl_quic_rstream_release_record(QUIC_RSTREAM *qrs, size_t read_len)
+{
+    uint64_t offset;
+
+    if (!ossl_sframe_list_is_head_locked(&qrs->fl))
+        return 0;
+
+    if (read_len > qrs->head_range.end - qrs->head_range.start) {
+        if (read_len != SIZE_MAX)
+            return 0;
+        offset = qrs->head_range.end;
+    } else {
+        offset = qrs->head_range.start + read_len;
+    }
+
+    if (!ossl_sframe_list_drop_frames(&qrs->fl, offset))
+        return 0;
+
+    if (offset > 0)
+        ring_buf_cpop_range(&qrs->rbuf, 0, offset - 1);
+
+    if (qrs->rxfc != NULL) {
+        OSSL_TIME rtt = get_rtt(qrs);
+
+        if (!ossl_quic_rxfc_on_retire(qrs->rxfc, offset, rtt))
+            return 0;
+    }
+
+    return 1;
+}
+
+static int write_at_ring_buf_cb(uint64_t logical_offset,
+                                const unsigned char *buf,
+                                size_t buf_len,
+                                void *cb_arg)
+{
+    struct ring_buf *rbuf = cb_arg;
+
+    return ring_buf_write_at(rbuf, logical_offset, buf, buf_len);
+}
+
+int ossl_quic_rstream_move_to_rbuf(QUIC_RSTREAM *qrs)
+{
+    if (ring_buf_avail(&qrs->rbuf) == 0)
+        return 0;
+    return ossl_sframe_list_move_data(&qrs->fl,
+                                      write_at_ring_buf_cb, &qrs->rbuf);
+}
+
+int ossl_quic_rstream_resize_rbuf(QUIC_RSTREAM *qrs, size_t rbuf_size)
+{
+    /* TODO(QUIC): Do we need to distinguish different error conditions ? */
+    if (ossl_sframe_list_is_head_locked(&qrs->fl))
+        return 0;
+
+    if (!ring_buf_resize(&qrs->rbuf, rbuf_size))
+        return 0;
+
+    return 1;
+}
diff --git a/ssl/quic/quic_sf_list.c b/ssl/quic/quic_sf_list.c
index 2924b78e1f..56803b2a97 100644
--- a/ssl/quic/quic_sf_list.c
+++ b/ssl/quic/quic_sf_list.c
@@ -97,14 +97,14 @@ int ossl_sframe_list_insert(SFRAME_LIST *fl, UINT_RANGE *range,
         goto end;
     }
 
-    /* TODO(QUIC): Check for fl->num_frames and start copying if too many */
-
     /* optimize insertion at the end */
     if (fl->tail->range.start < range->start) {
         if (fl->tail->range.end >= range->end)
             goto end;
 
-        return append_frame(fl, range, pkt, data);
+        if (!append_frame(fl, range, pkt, data))
+            return 0;
+        goto end;
     }
 
     prev_frame = NULL;
@@ -200,7 +200,10 @@ int ossl_sframe_list_peek(const SFRAME_LIST *fl, void **iter,
     }
 
     range->end = sf->range.end;
-    *data = sf->data + (start - sf->range.start);
+    if (sf->data != NULL)
+        *data = sf->data + (start - sf->range.start);
+    else
+        *data = NULL;
     *fin = sf->next == NULL ? fl->fin : 0;
     *iter = sf;
     return 1;
@@ -234,5 +237,88 @@ int ossl_sframe_list_drop_frames(SFRAME_LIST *fl, uint64_t limit)
     else
         fl->tail = NULL;
 
+    fl->head_locked = 0;
+
+    return 1;
+}
+
+int ossl_sframe_list_lock_head(SFRAME_LIST *fl, UINT_RANGE *range,
+                               const unsigned char **data,
+                               int *fin)
+{
+    int ret;
+    void *iter = NULL;
+
+    if (fl->head_locked)
+        return 0;
+
+    ret = ossl_sframe_list_peek(fl, &iter, range, data, fin);
+    if (ret)
+        fl->head_locked = 1;
+    return ret;
+}
+
+int ossl_sframe_list_is_head_locked(SFRAME_LIST *fl)
+{
+    return fl->head_locked;
+}
+
+int ossl_sframe_list_move_data(SFRAME_LIST *fl,
+                               sframe_list_write_at_cb *write_at_cb,
+                               void *cb_arg)
+{
+    STREAM_FRAME *sf = fl->head, *prev_frame = NULL;
+    uint64_t limit = fl->offset;
+
+    if (sf == NULL)
+        return 1;
+
+    if (fl->head_locked)
+        sf = sf->next;
+
+    for (; sf != NULL; sf = sf->next) {
+        size_t len;
+        const unsigned char *data = sf->data;
+
+        if (limit < sf->range.start)
+            limit = sf->range.start;
+
+        if (data != NULL) {
+            if (limit > sf->range.start)
+                data += (size_t)(limit - sf->range.start);
+            len = (size_t)(sf->range.end - limit);
+
+            if (!write_at_cb(limit, data, len, cb_arg))
+                /* data did not fit */
+                return 0;
+
+            /* release the packet */
+            sf->data = NULL;
+            ossl_qrx_pkt_release(sf->pkt);
+            sf->pkt = NULL;
+        }
+
+        limit = sf->range.end;
+
+        /* merge contiguous frames */
+        if (prev_frame != NULL
+            && prev_frame->range.end >= sf->range.start) {
+            prev_frame->range.end = sf->range.end;
+            prev_frame->next = sf->next;
+
+            if (sf->next != NULL)
+                sf->next->prev = prev_frame;
+            else
+                fl->tail = prev_frame;
+
+            --fl->num_frames;
+            stream_frame_free(fl, sf);
+            sf = prev_frame;
+            continue;
+        }
+
+        prev_frame = sf;
+    }
+
     return 1;
 }
diff --git a/ssl/quic/quic_sstream.c b/ssl/quic/quic_sstream.c
index 8dc7002c41..a0ef4e9eae 100644
--- a/ssl/quic/quic_sstream.c
+++ b/ssl/quic/quic_sstream.c
@@ -10,182 +10,7 @@
 #include "internal/quic_stream.h"
 #include "internal/uint_set.h"
 #include "internal/common.h"
-
-/*
- * ==================================================================
- * Byte-wise ring buffer which supports pushing and popping blocks of multiple
- * bytes at a time. The logical offset of each byte for the purposes of a QUIC
- * stream is tracked. Bytes can be popped from the ring buffer in two stages;
- * first they are popped, and then they are culled. Bytes which have been popped
- * but not yet culled will not be overwritten, and can be restored.
- */
-struct ring_buf {
-    void       *start;
-    size_t      alloc;        /* size of buffer allocation in bytes */
-
-    /*
-     * Logical offset of the head (where we append to). This is the current size
-     * of the QUIC stream. This increases monotonically.
-     */
-    uint64_t    head_offset;
-
-    /*
-     * Logical offset of the cull tail. Data is no longer needed and is
-     * deallocated as the cull tail advances, which occurs as data is
-     * acknowledged. This increases monotonically.
-     */
-    uint64_t    ctail_offset;
-};
-
-static int ring_buf_init(struct ring_buf *r)
-{
-    r->start = NULL;
-    r->alloc = 0;
-    r->head_offset = r->ctail_offset = 0;
-    return 1;
-}
-
-static void ring_buf_destroy(struct ring_buf *r)
-{
-    OPENSSL_free(r->start);
-    r->start = NULL;
-    r->alloc = 0;
-}
-
-static size_t ring_buf_used(struct ring_buf *r)
-{
-    return (size_t)(r->head_offset - r->ctail_offset);
-}
-
-static size_t ring_buf_avail(struct ring_buf *r)
-{
-    return r->alloc - ring_buf_used(r);
-}
-
-static size_t ring_buf_push(struct ring_buf *r,
-                            const unsigned char *buf, size_t buf_len)
-{
-    size_t pushed = 0, avail, idx, l, i;
-    unsigned char *start = r->start;
-
-    for (i = 0;; ++i) {
-        avail = ring_buf_avail(r);
-        if (buf_len > avail)
-            buf_len = avail;
-
-        if (buf_len == 0)
-            break;
-
-        assert(i < 2);
-
-        idx = r->head_offset % r->alloc;
-        l = r->alloc - idx;
-        if (buf_len < l)
-            l = buf_len;
-
-        memcpy(start + idx, buf, l);
-        r->head_offset  += l;
-        buf             += l;
-        buf_len         -= l;
-        pushed          += l;
-    }
-
-    return pushed;
-}
-
-/*
- * Retrieves data out of the read size of the ring buffer starting at the given
- * logical offset. *buf is set to point to a contiguous span of bytes and
- * *buf_len is set to the number of contiguous bytes. After this function
- * returns, there may or may not be more bytes available at the logical offset
- * of (logical_offset + *buf_len) by calling this function again. If the logical
- * offset is out of the range retained by the ring buffer, returns 0, else
- * returns 1. A logical offset at the end of the range retained by the ring
- * buffer is not considered an error and is returned with a *buf_len of 0.
- *
- * The ring buffer state is not changed.
- */
-static int ring_buf_get_buf_at(const struct ring_buf *r,
-                               uint64_t logical_offset,
-                               const unsigned char **buf, size_t *buf_len)
-{
-    const unsigned char *start = r->start;
-    size_t idx, l;
-
-    if (logical_offset > r->head_offset || logical_offset < r->ctail_offset)
-        return 0;
-
-    if (r->alloc == 0) {
-        *buf        = NULL;
-        *buf_len    = 0;
-        return 1;
-    }
-
-    idx = logical_offset % r->alloc;
-    l   = (size_t)(r->head_offset - logical_offset);
-    if (l > r->alloc - idx)
-        l = r->alloc - idx;
-
-    *buf        = start + idx;
-    *buf_len    = l;
-    return 1;
-}
-
-static void ring_buf_cpop_range(struct ring_buf *r,
-                                uint64_t start, uint64_t end)
-{
-    assert(end >= start);
-
-    if (start > r->ctail_offset)
-        return;
-
-    r->ctail_offset = end + 1;
-}
-
-static int ring_buf_resize(struct ring_buf *r, size_t num_bytes)
-{
-    struct ring_buf rnew = {0};
-    const unsigned char *src = NULL;
-    size_t src_len = 0, copied = 0;
-
-    if (num_bytes == r->alloc)
-        return 1;
-
-    if (num_bytes < ring_buf_used(r))
-        return 0;
-
-    rnew.start = OPENSSL_malloc(num_bytes);
-    if (rnew.start == NULL)
-        return 0;
-
-    rnew.alloc          = num_bytes;
-    rnew.head_offset    = r->head_offset - ring_buf_used(r);
-    rnew.ctail_offset   = rnew.head_offset;
-
-    for (;;) {
-        if (!ring_buf_get_buf_at(r, r->ctail_offset + copied, &src, &src_len)) {
-            OPENSSL_free(rnew.start);
-            return 0;
-        }
-
-        if (src_len == 0)
-            break;
-
-        if (ring_buf_push(&rnew, src, src_len) != src_len) {
-            OPENSSL_free(rnew.start);
-            return 0;
-        }
-
-        copied += src_len;
-    }
-
-    assert(rnew.head_offset == r->head_offset);
-    rnew.ctail_offset   = r->ctail_offset;
-
-    OPENSSL_free(r->start);
-    memcpy(r, &rnew, sizeof(*r));
-    return 1;
-}
+#include "internal/ring_buf.h"
 
 /*
  * ==================================================================