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)

3 files changed, 382 insertions, 1 deletions

diff --git a/include/internal/quic_sf_list.h b/include/internal/quic_sf_list.h
index 85d2fe1723..f0efcea2bf 100644
--- a/include/internal/quic_sf_list.h
+++ b/include/internal/quic_sf_list.h
@@ -47,18 +47,103 @@ typedef struct sframe_list_st {
     size_t num_frames;
     /* Offset of data not yet dropped */
     uint64_t offset;
+    /* Is head locked ? */
+    int head_locked;
 } SFRAME_LIST;
 
+/*
+ * Initializes the stream frame list fl.
+ */
 void ossl_sframe_list_init(SFRAME_LIST *fl);
+
+/*
+ * Destroys the stream frame list fl releasing any data
+ * still present inside it.
+ */
 void ossl_sframe_list_destroy(SFRAME_LIST *fl);
+
+/*
+ * Insert a stream frame data into the list.
+ * The data covers an offset range (range.start is inclusive,
+ * range.end is exclusive).
+ * fin should be set if this is the final frame of the stream.
+ * Returns an error if a frame cannot be inserted - due to
+ * STREAM_FRAME allocation error, or in case of erroneous
+ * fin flag (this is an ossl_assert() check so a caller must
+ * check it on its own too).
+ */
 int ossl_sframe_list_insert(SFRAME_LIST *fl, UINT_RANGE *range,
                             OSSL_QRX_PKT *pkt,
                             const unsigned char *data, int fin);
+
+/*
+ * Iterator to peek at the contiguous frames at the beginning
+ * of the frame list fl.
+ * The *data covers an offset range (range.start is inclusive,
+ * range.end is exclusive).
+ * *fin is set if this is the final frame of the stream.
+ * Opaque iterator *iter can be used to peek at the subsequent
+ * frame if there is any without any gap before it.
+ * Returns 1 on success.
+ * Returns 0 if there is no further contiguous frame. In that
+ * case *fin is set, if the end of the stream is reached.
+ */
 int ossl_sframe_list_peek(const SFRAME_LIST *fl, void **iter,
                           UINT_RANGE *range, const unsigned char **data,
                           int *fin);
+
+/*
+ * Drop all frames up to the offset limit.
+ * Also unlocks the head frame if locked.
+ * Returns 1 on success.
+ * Returns 0 when trying to drop frames at offsets that were not
+ * received yet. (ossl_assert() is used to check, so this is an invalid call.)
+ */
 int ossl_sframe_list_drop_frames(SFRAME_LIST *fl, uint64_t limit);
 
+/*
+ * Locks and returns the head frame of fl if it is readable - read offset is
+ * at the beginning or middle of the frame.
+ * range is set to encompass the not yet read part of the head frame,
+ * data pointer is set to appropriate offset within the frame if the read
+ * offset points in the middle of the frame,
+ * fin is set to 1 if the head frame is also the tail frame.
+ * Returns 1 on success, 0 if there is no readable data or the head
+ * frame is already locked.
+ */
+int ossl_sframe_list_lock_head(SFRAME_LIST *fl, UINT_RANGE *range,
+                               const unsigned char **data,
+                               int *fin);
+
+/*
+ * Just returns whether the head frame is locked by previous
+ * ossl_sframe_list_lock_head() call.
+ */
+int ossl_sframe_list_is_head_locked(SFRAME_LIST *fl);
+
+/*
+ * Callback function type to write stream frame data to some
+ * side storage before the packet containing the frame data
+ * is released.
+ * It should return 1 on success or 0 if there is not enough
+ * space available in the side storage.
+ */
+typedef int (sframe_list_write_at_cb)(uint64_t logical_offset,
+                                      const unsigned char *buf,
+                                      size_t buf_len,
+                                      void *cb_arg);
+
+/*
+ * Move the frame data in all the stream frames in the list fl
+ * from the packets to the side storage using the write_at_cb
+ * callback.
+ * Returns 1 if all the calls to the callback return 1.
+ * If the callback returns 0, the function stops processing further
+ * frames and returns 0.
+ */
+int ossl_sframe_list_move_data(SFRAME_LIST *fl,
+                               sframe_list_write_at_cb *write_at_cb,
+                               void *cb_arg);
 # endif
 
 #endif
diff --git a/include/internal/quic_stream.h b/include/internal/quic_stream.h
index 1fd7d85d3c..42d6ed2d7d 100644
--- a/include/internal/quic_stream.h
+++ b/include/internal/quic_stream.h
@@ -308,9 +308,11 @@ typedef struct quic_rstream_st QUIC_RSTREAM;
  * controller and statistics module. They can be NULL for unit testing.
  * If they are non-NULL, the `rxfc` is called when receive stream data
  * is read by application. `statm` is queried for current rtt.
+ * `rbuf_size` is the initial size of the ring buffer to be used
+ * when ossl_quic_rstream_move_to_rbuf() is called.
  */
 QUIC_RSTREAM *ossl_quic_rstream_new(QUIC_RXFC *rxfc,
-                                    OSSL_STATM *statm);
+                                    OSSL_STATM *statm, size_t rbuf_size);
 
 /*
  * Frees a QUIC_RSTREAM and any associated storage.
@@ -357,6 +359,55 @@ int ossl_quic_rstream_peek(QUIC_RSTREAM *qrs, unsigned char *buf, size_t size,
  */
 int ossl_quic_rstream_available(QUIC_RSTREAM *qrs, size_t *avail, int *fin);
 
+/*
+ * Sets *record to the beginning of the first readable stream data chunk and
+ * *reclen to the size of the chunk. *fin is set to 1 if the end of the
+ * chunk is the last of the stream data chunks.
+ * If there is no record available *record is set to NULL and *rec_len to 0;
+ * ossl_quic_rstream_release_record() should not be called in that case.
+ * Returns 1 on success (including calls if no record is available, or
+ * after end of the stream - in that case *fin will be set to 1 and
+ * *rec_len to 0), 0 on error.
+ * It is an error to call ossl_quic_rstream_get_record() multiple times
+ * without calling ossl_quic_rstream_release_record() in between.
+ */
+int ossl_quic_rstream_get_record(QUIC_RSTREAM *qrs,
+                                 const unsigned char **record, size_t *rec_len,
+                                 int *fin);
+
+/*
+ * Releases (possibly partially) the record returned by
+ * previous ossl_quic_rstream_get_record() call.
+ * read_len between previously returned *rec_len and SIZE_MAX indicates
+ * release of the whole record. Otherwise only part of the record is
+ * released. The remaining part of the record is unlocked, another
+ * call to ossl_quic_rstream_get_record() is needed to obtain further
+ * stream data.
+ * Returns 1 on success, 0 on error.
+ * It is an error to call ossl_quic_rstream_release_record() multiple
+ * times without calling ossl_quic_rstream_get_record() in between.
+ */
+int ossl_quic_rstream_release_record(QUIC_RSTREAM *qrs, size_t read_len);
+
+/*
+ * Moves received frame data from decrypted packets to ring buffer.
+ * This should be called when there are too many decrypted packets allocated.
+ * Returns 1 on success, 0 when it was not possible to release all
+ * referenced packets due to an insufficient size of the ring buffer.
+ * Exception is the packet from the record returned previously by
+ * ossl_quic_rstream_get_record() - that one will be always skipped.
+ */
+int ossl_quic_rstream_move_to_rbuf(QUIC_RSTREAM *qrs);
+
+/*
+ * Resizes the internal ring buffer to a new `rbuf_size` size.
+ * Returns 1 on success, 0 on error.
+ * Possible error conditions are an allocation failure, trying to resize
+ * the ring buffer when ossl_quic_rstream_get_record() was called and
+ * not yet released, or trying to resize the ring buffer to a smaller size
+ * than currently occupied.
+ */
+int ossl_quic_rstream_resize_rbuf(QUIC_RSTREAM *qrs, size_t rbuf_size);
 # endif
 
 #endif
diff --git a/include/internal/ring_buf.h b/include/internal/ring_buf.h
new file mode 100644
index 0000000000..175f78047f
--- /dev/null
+++ b/include/internal/ring_buf.h
@@ -0,0 +1,245 @@
+/*
+ * Copyright 2022 The OpenSSL Project Authors. All Rights Reserved.
+ *
+ * Licensed under the Apache License 2.0 (the "License").  You may not use
+ * this file except in compliance with the License.  You can obtain a copy
+ * in the file LICENSE in the source distribution or at
+ * https://www.openssl.org/source/license.html
+ */
+
+#ifndef OSSL_INTERNAL_RING_BUF_H
+# define OSSL_INTERNAL_RING_BUF_H
+# pragma once
+
+# include <openssl/e_os2.h>              /* For 'ossl_inline' */
+
+/*
+ * ==================================================================
+ * 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 ossl_inline int ring_buf_init(struct ring_buf *r)
+{
+    r->start = NULL;
+    r->alloc = 0;
+    r->head_offset = r->ctail_offset = 0;
+    return 1;
+}
+
+static ossl_inline void ring_buf_destroy(struct ring_buf *r)
+{
+    OPENSSL_free(r->start);
+    r->start = NULL;
+    r->alloc = 0;
+}
+
+static ossl_inline size_t ring_buf_used(struct ring_buf *r)
+{
+    return (size_t)(r->head_offset - r->ctail_offset);
+}
+
+static ossl_inline size_t ring_buf_avail(struct ring_buf *r)
+{
+    return r->alloc - ring_buf_used(r);
+}
+
+static ossl_inline int ring_buf_write_at(struct ring_buf *r,
+                                         uint64_t logical_offset,
+                                         const unsigned char *buf,
+                                         size_t buf_len)
+{
+    size_t avail, idx, l;
+    unsigned char *start = r->start;
+    int i;
+
+    avail = ring_buf_avail(r);
+    if (logical_offset < r->ctail_offset
+        || logical_offset + buf_len > r->head_offset + avail)
+        return 0;
+
+    for (i = 0; buf_len > 0 && i < 2; ++i) {
+        idx = logical_offset % r->alloc;
+        l = r->alloc - idx;
+        if (buf_len < l)
+            l = buf_len;
+
+        memcpy(start + idx, buf, l);
+        if (r->head_offset < logical_offset + l)
+            r->head_offset = logical_offset + l;
+
+        logical_offset += l;
+        buf += l;
+        buf_len -= l;
+    }
+
+    assert(buf_len == 0);
+
+    return 1;
+}
+
+static ossl_inline 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;
+}
+
+static ossl_inline const unsigned char *ring_buf_get_ptr(const struct ring_buf *r,
+                                                         uint64_t logical_offset,
+                                                         size_t *max_len)
+{
+    unsigned char *start = r->start;
+    size_t idx;
+
+    if (logical_offset >= r->head_offset || logical_offset < r->ctail_offset)
+        return NULL;
+    idx = logical_offset % r->alloc;
+    *max_len = r->alloc - idx;
+    return start + idx;
+}
+
+/*
+ * Retrieves data out of the read side 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 ossl_inline 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 ossl_inline 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;
+    /* Allow culling unpushed data */
+    if (r->head_offset < r->ctail_offset)
+        r->head_offset = r->ctail_offset;
+}
+
+static ossl_inline 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;
+}
+
+#endif                          /* OSSL_INTERNAL_RING_BUF_H */