summaryrefslogtreecommitdiffstats
path: root/src/quic_stream.c
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2024-04-13 12:18:05 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2024-04-13 12:18:05 +0000
commitb46aad6df449445a9fc4aa7b32bd40005438e3f7 (patch)
tree751aa858ca01f35de800164516b298887382919d /src/quic_stream.c
parentInitial commit. (diff)
downloadhaproxy-b46aad6df449445a9fc4aa7b32bd40005438e3f7.tar.xz
haproxy-b46aad6df449445a9fc4aa7b32bd40005438e3f7.zip
Adding upstream version 2.9.5.upstream/2.9.5
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'src/quic_stream.c')
-rw-r--r--src/quic_stream.c294
1 files changed, 294 insertions, 0 deletions
diff --git a/src/quic_stream.c b/src/quic_stream.c
new file mode 100644
index 0000000..a4b984d
--- /dev/null
+++ b/src/quic_stream.c
@@ -0,0 +1,294 @@
+#include <haproxy/quic_stream.h>
+
+#include <import/eb64tree.h>
+
+#include <haproxy/api.h>
+#include <haproxy/buf.h>
+#include <haproxy/dynbuf.h>
+#include <haproxy/list.h>
+#include <haproxy/mux_quic-t.h>
+#include <haproxy/pool.h>
+#include <haproxy/quic_conn.h>
+#include <haproxy/task.h>
+
+DECLARE_STATIC_POOL(pool_head_quic_stream_desc, "qc_stream_desc",
+ sizeof(struct qc_stream_desc));
+DECLARE_STATIC_POOL(pool_head_quic_stream_buf, "qc_stream_buf",
+ sizeof(struct qc_stream_buf));
+
+
+static void qc_stream_buf_free(struct qc_stream_desc *stream,
+ struct qc_stream_buf **stream_buf)
+{
+ struct quic_conn *qc = stream->qc;
+ struct buffer *buf = &(*stream_buf)->buf;
+
+ LIST_DEL_INIT(&(*stream_buf)->list);
+
+ /* Reset current buf ptr if deleted instance is the same one. */
+ if (*stream_buf == stream->buf)
+ stream->buf = NULL;
+
+ b_free(buf);
+ offer_buffers(NULL, 1);
+ pool_free(pool_head_quic_stream_buf, *stream_buf);
+ *stream_buf = NULL;
+
+ /* notify MUX about available buffers. */
+ --qc->stream_buf_count;
+ if (qc->mux_state == QC_MUX_READY) {
+ if (qc->qcc->flags & QC_CF_CONN_FULL) {
+ qc->qcc->flags &= ~QC_CF_CONN_FULL;
+ tasklet_wakeup(qc->qcc->wait_event.tasklet);
+ }
+ }
+}
+
+/* Allocate a new stream descriptor with id <id>. The caller is responsible to
+ * store the stream in the appropriate tree. -1 special value must be used for
+ * a CRYPTO data stream, the type being ignored.
+ *
+ * Returns the newly allocated instance on success or else NULL.
+ */
+struct qc_stream_desc *qc_stream_desc_new(uint64_t id, enum qcs_type type, void *ctx,
+ struct quic_conn *qc)
+{
+ struct qc_stream_desc *stream;
+
+ stream = pool_alloc(pool_head_quic_stream_desc);
+ if (!stream)
+ return NULL;
+
+ if (id == (uint64_t)-1) {
+ stream->by_id.key = (uint64_t)-1;
+ }
+ else {
+ stream->by_id.key = id;
+ eb64_insert(&qc->streams_by_id, &stream->by_id);
+ qc->rx.strms[type].nb_streams++;
+ }
+ stream->qc = qc;
+
+ stream->buf = NULL;
+ LIST_INIT(&stream->buf_list);
+ stream->buf_offset = 0;
+
+ stream->acked_frms = EB_ROOT;
+ stream->ack_offset = 0;
+ stream->release = 0;
+ stream->ctx = ctx;
+
+ return stream;
+}
+
+/* Mark the stream descriptor <stream> as released. It will be freed as soon as
+ * all its buffered data are acknowledged. Does nothing if <stream> is already
+ * NULL.
+ *
+ * <final_size> corresponds to the last offset sent for this stream. If there
+ * is unsent data present, they will be remove first to guarantee that buffer
+ * is freed after receiving all acknowledges.
+ */
+void qc_stream_desc_release(struct qc_stream_desc *stream,
+ uint64_t final_size)
+{
+ if (!stream)
+ return;
+
+ /* A stream can be released only one time. */
+ BUG_ON(stream->release);
+
+ stream->release = 1;
+ stream->ctx = NULL;
+
+ if (stream->buf) {
+ struct qc_stream_buf *stream_buf = stream->buf;
+ struct buffer *buf = &stream_buf->buf;
+ const uint64_t tail_offset =
+ MAX(stream->buf_offset, stream->ack_offset) + b_data(buf);
+
+ /* final_size cannot be greater than all currently stored data. */
+ BUG_ON(final_size > tail_offset);
+
+ /* Remove unsent data from current buffer. */
+ if (final_size < tail_offset) {
+ b_sub(buf, tail_offset - final_size);
+ /* Remove buffer is all ACK already received. */
+ if (!b_data(buf))
+ qc_stream_buf_free(stream, &stream_buf);
+ }
+
+ /* A released stream does not use <stream.buf>. */
+ stream->buf = NULL;
+ }
+
+ if (LIST_ISEMPTY(&stream->buf_list)) {
+ /* if no buffer left we can free the stream. */
+ qc_stream_desc_free(stream, 0);
+ }
+}
+
+/* Acknowledge data at <offset> of length <len> for <stream>. It is handled
+ * only if it covers a range corresponding to stream.ack_offset. After data
+ * removal, if the stream does not contains data any more and is already
+ * released, the instance stream is freed. <stream> is set to NULL to indicate
+ * this.
+ *
+ * Returns the count of byte removed from stream. Do not forget to check if
+ * <stream> is NULL after invocation.
+ */
+int qc_stream_desc_ack(struct qc_stream_desc **stream, size_t offset, size_t len)
+{
+ struct qc_stream_desc *s = *stream;
+ struct qc_stream_buf *stream_buf;
+ struct buffer *buf;
+ size_t diff;
+
+ if (offset + len <= s->ack_offset || offset > s->ack_offset)
+ return 0;
+
+ /* There must be at least a buffer or we must not report an ACK. */
+ BUG_ON(LIST_ISEMPTY(&s->buf_list));
+
+ /* get oldest buffer from buf_list */
+ stream_buf = LIST_NEXT(&s->buf_list, struct qc_stream_buf *, list);
+ buf = &stream_buf->buf;
+
+ diff = offset + len - s->ack_offset;
+ s->ack_offset += diff;
+ b_del(buf, diff);
+
+ /* Free oldest buffer if all data acknowledged. */
+ if (!b_data(buf)) {
+ qc_stream_buf_free(s, &stream_buf);
+
+ /* Free stream instance if already released and no buffers left. */
+ if (s->release && LIST_ISEMPTY(&s->buf_list)) {
+ qc_stream_desc_free(s, 0);
+ *stream = NULL;
+ }
+ }
+
+ return diff;
+}
+
+/* Free the stream descriptor <stream> content. This function should be used
+ * when all its data have been acknowledged or on full connection closing if <closing>
+ * boolean is set to 1. It must only be called after the stream is released.
+ */
+void qc_stream_desc_free(struct qc_stream_desc *stream, int closing)
+{
+ struct qc_stream_buf *buf, *buf_back;
+ struct quic_conn *qc = stream->qc;
+ struct eb64_node *frm_node;
+ unsigned int free_count = 0;
+
+ /* This function only deals with released streams. */
+ BUG_ON(!stream->release);
+
+ /* free remaining stream buffers */
+ list_for_each_entry_safe(buf, buf_back, &stream->buf_list, list) {
+ if (!(b_data(&buf->buf)) || closing) {
+ b_free(&buf->buf);
+ LIST_DELETE(&buf->list);
+ pool_free(pool_head_quic_stream_buf, buf);
+
+ ++free_count;
+ }
+ }
+
+ if (free_count) {
+ offer_buffers(NULL, free_count);
+
+ qc->stream_buf_count -= free_count;
+ if (qc->mux_state == QC_MUX_READY) {
+ /* notify MUX about available buffers. */
+ if (qc->qcc->flags & QC_CF_CONN_FULL) {
+ qc->qcc->flags &= ~QC_CF_CONN_FULL;
+ tasklet_wakeup(qc->qcc->wait_event.tasklet);
+ }
+ }
+ }
+
+ /* qc_stream_desc might be freed before having received all its ACKs.
+ * This is the case if some frames were retransmitted.
+ */
+ frm_node = eb64_first(&stream->acked_frms);
+ while (frm_node) {
+ struct qf_stream *strm_frm;
+ struct quic_frame *frm;
+
+ strm_frm = eb64_entry(frm_node, struct qf_stream, offset);
+
+ frm_node = eb64_next(frm_node);
+ eb64_delete(&strm_frm->offset);
+
+ frm = container_of(strm_frm, struct quic_frame, stream);
+ qc_release_frm(qc, frm);
+ }
+
+ if (stream->by_id.key != (uint64_t)-1)
+ eb64_delete(&stream->by_id);
+ pool_free(pool_head_quic_stream_desc, stream);
+}
+
+/* Return the current buffer of <stream>. May be NULL if not allocated. */
+struct buffer *qc_stream_buf_get(struct qc_stream_desc *stream)
+{
+ if (!stream->buf)
+ return NULL;
+
+ return &stream->buf->buf;
+}
+
+/* Returns the count of available buffer left for <qc>. */
+static int qc_stream_buf_avail(struct quic_conn *qc)
+{
+ BUG_ON(qc->stream_buf_count > global.tune.quic_streams_buf);
+ return global.tune.quic_streams_buf - qc->stream_buf_count;
+}
+
+/* Allocate a new current buffer for <stream>. The buffer limit count for the
+ * connection is checked first. This function is not allowed if current buffer
+ * is not NULL prior to this call. The new buffer represents stream payload at
+ * offset <offset>.
+ *
+ * Returns the buffer or NULL on error. Caller may check <avail> to ensure if
+ * the connection buffer limit was reached or a fatal error was encountered.
+ */
+struct buffer *qc_stream_buf_alloc(struct qc_stream_desc *stream,
+ uint64_t offset, int *avail)
+{
+ struct quic_conn *qc = stream->qc;
+
+ /* current buffer must be released first before allocate a new one. */
+ BUG_ON(stream->buf);
+
+ *avail = qc_stream_buf_avail(qc);
+ if (!*avail)
+ return NULL;
+
+ stream->buf_offset = offset;
+ stream->buf = pool_alloc(pool_head_quic_stream_buf);
+ if (!stream->buf)
+ return NULL;
+
+ ++qc->stream_buf_count;
+
+ stream->buf->buf = BUF_NULL;
+ LIST_APPEND(&stream->buf_list, &stream->buf->list);
+
+ return &stream->buf->buf;
+}
+
+/* Release the current buffer of <stream>. It will be kept internally by
+ * the <stream>. The current buffer cannot be NULL.
+ */
+void qc_stream_buf_release(struct qc_stream_desc *stream)
+{
+ /* current buffer already released */
+ BUG_ON(!stream->buf);
+
+ stream->buf = NULL;
+ stream->buf_offset = 0;
+}