MINOR: quic: implement stream descriptor for transport layer

Currently, the mux qcs streams manage the Tx buffering, even after
sending it to the transport layer. Buffers are emptied when
acknowledgement are treated by the transport layer. This complicates the
MUX liberation and we may loose some data after the MUX free.

Change this paradigm by moving the buffering on the transport layer. For
this goal, a new type is implemented as low-level stream at the
transport layer, as a counterpart of qcs mux instances. This structure
is called qc_stream_desc. This will allow to free the qcs/qcc instances
without having to wait for acknowledge reception.

For the moment, the quic-conn is responsible to store the qc_stream_desc
in a new tree named streams_by_id. This will sligthly change in the next
commits to remove the qcs node which has a similar purpose :
qc_stream_desc instances will be shared between the qcc MUX and the
quic-conn.

This patch only introduces the new type definition and the function to
manipulate it. The following commit will bring the rearchitecture in the
qcs structure.

diff --git a/include/haproxy/xprt_quic-t.h b/include/haproxy/xprt_quic-t.h
index 4e31f1c40470499a4bdc968ff11827d6d7d22d7c..f4b160a10efb77b9c84c46e1fe82477c72fc1aa2 100644 (file)
--- a/include/haproxy/xprt_quic-t.h
+++ b/include/haproxy/xprt_quic-t.h
@@ -745,6 +745,9 @@ struct quic_conn {
 
        struct listener *li; /* only valid for frontend connections */
        struct mt_list accept_list; /* chaining element used for accept, only valid for frontend connections */
+
+       struct eb_root streams_by_id; /* stream-descriptors tree */
+
        /* MUX */
        struct qcc *qcc;
        struct task *timer_task;
@@ -756,5 +759,26 @@ struct quic_conn {
        const struct qcc_app_ops *app_ops;
 };
 
+/* QUIC STREAM descriptor.
+ *
+ * This structure is the low-level counterpart of the QUIC STREAM at the MUX
+ * layer. It provides a node for tree-storage and buffering for Tx.
+ *
+ * Once the MUX has finished to transfer data on a STREAM, it must release its
+ * QUIC STREAM descriptor. The descriptor will be kept by the quic_conn until
+ * all acknowledgement has been received.
+ */
+struct qc_stream_desc {
+       struct eb64_node by_id; /* id of the stream used for <streams_by_id> tree */
+
+       struct buffer buf; /* buffer for STREAM data on Tx, emptied on acknowledge */
+       uint64_t ack_offset; /* last acknowledged offset */
+       struct eb_root acked_frms; /* ACK frames tree for non-contiguous ACK ranges */
+
+       int release; /* set to 1 when the MUX has finished to use this stream */
+
+       void *ctx; /* MUX specific context */
+};
+
 #endif /* USE_QUIC */
 #endif /* _HAPROXY_XPRT_QUIC_T_H */

diff --git a/include/haproxy/xprt_quic.h b/include/haproxy/xprt_quic.h
index 3348d82cf60f7b7d0b583a2b6c29742e31660a50..cc6cd49c64e0fcc992d614e2f6f37c8413c869ec 100644 (file)
--- a/include/haproxy/xprt_quic.h
+++ b/include/haproxy/xprt_quic.h
@@ -1243,5 +1243,8 @@ int quic_lstnr_dgram_dispatch(unsigned char *buf, size_t len, void *owner,
                               struct quic_dgram *new_dgram, struct list *dgrams);
 int qc_send_app_pkts(struct quic_conn *qc, struct list *frms);
 
+struct qc_stream_desc *qc_stream_desc_new(struct quic_conn *qc, uint64_t id, void *ctx);
+void qc_stream_desc_release(struct qc_stream_desc *stream);
+
 #endif /* USE_QUIC */
 #endif /* _HAPROXY_XPRT_QUIC_H */

diff --git a/src/xprt_quic.c b/src/xprt_quic.c
index 56c688515fb0a05985d0b82ed76bcf64e43c158d..36b50fd5b345950cfff19c0f63a660f755e6df1d 100644 (file)
--- a/src/xprt_quic.c
+++ b/src/xprt_quic.c
@@ -169,6 +169,7 @@ DECLARE_POOL(pool_head_quic_rx_strm_frm, "quic_rx_strm_frm", sizeof(struct quic_
 DECLARE_STATIC_POOL(pool_head_quic_crypto_buf, "quic_crypto_buf_pool", sizeof(struct quic_crypto_buf));
 DECLARE_POOL(pool_head_quic_frame, "quic_frame_pool", sizeof(struct quic_frame));
 DECLARE_STATIC_POOL(pool_head_quic_arng, "quic_arng_pool", sizeof(struct quic_arng_node));
+DECLARE_STATIC_POOL(pool_head_quic_conn_stream, "qc_stream_desc", sizeof(struct qc_stream_desc));
 
 static struct quic_tx_packet *qc_build_pkt(unsigned char **pos, const unsigned char *buf_end,
                                            struct quic_enc_level *qel, struct list *frms,
@@ -1399,6 +1400,27 @@ static int qc_pkt_decrypt(struct quic_rx_packet *pkt, struct quic_enc_level *qel
        return 1;
 }
 
+/* Free the stream descriptor <stream> buffer. This function should be used
+ * when all its data have been acknowledged. If the stream was released by the
+ * upper layer, the stream descriptor will be freed.
+ *
+ * Returns 0 if the stream was not freed else non-zero.
+ */
+static int qc_stream_desc_free(struct qc_stream_desc *stream)
+{
+       b_free(&stream->buf);
+       offer_buffers(NULL, 1);
+
+       if (stream->release) {
+               eb64_delete(&stream->by_id);
+               pool_free(pool_head_quic_conn_stream, stream);
+
+               return 1;
+       }
+
+       return 0;
+}
+
 /* Remove from <qcs> stream the acknowledged frames.
  *
  * Returns 1 if at least one frame was removed else 0.
@@ -3689,6 +3711,19 @@ static void quic_conn_release(struct quic_conn *qc)
 {
        int i;
        struct ssl_sock_ctx *conn_ctx;
+       struct eb64_node *node;
+
+       /* free remaining stream descriptors */
+       node = eb64_first(&qc->streams_by_id);
+       while (node) {
+               struct qc_stream_desc *stream;
+
+               stream = eb64_entry(node, struct qc_stream_desc, by_id);
+               node = eb64_next(node);
+
+               eb64_delete(&stream->by_id);
+               pool_free(pool_head_quic_conn_stream, stream);
+       }
 
        if (qc->idle_timer_task) {
                task_destroy(qc->idle_timer_task);
@@ -3909,6 +3944,8 @@ static struct quic_conn *qc_new_conn(unsigned int version, int ipv4,
        /* required to use MTLIST_IN_LIST */
        MT_LIST_INIT(&qc->accept_list);
 
+       qc->streams_by_id = EB_ROOT_UNIQUE;
+
        TRACE_LEAVE(QUIC_EV_CONN_INIT, qc);
 
        return qc;
@@ -5681,6 +5718,43 @@ int quic_lstnr_dgram_dispatch(unsigned char *buf, size_t len, void *owner,
        return 0;
 }
 
+/* Allocate a new stream descriptor with id <id>. The stream will be stored
+ * inside the <qc> connection.
+ *
+ * Returns the newly allocated instance on success or else NULL.
+ */
+struct qc_stream_desc *qc_stream_desc_new(struct quic_conn *qc, uint64_t id, void *ctx)
+{
+       struct qc_stream_desc *stream;
+
+       stream = pool_alloc(pool_head_quic_conn_stream);
+       if (!stream)
+               return NULL;
+
+       stream->by_id.key = id;
+       eb64_insert(&qc->streams_by_id, &stream->by_id);
+
+       stream->buf = BUF_NULL;
+       stream->acked_frms = EB_ROOT;
+       stream->ack_offset = 0;
+       stream->release = 0;
+       stream->ctx = ctx;
+
+       return stream;
+}
+
+/* Mark the stream descriptor <stream> as released by the upper layer. It will
+ * be freed as soon as all its buffered data are acknowledged.
+ */
+void qc_stream_desc_release(struct qc_stream_desc *stream)
+{
+       stream->release = 1;
+       stream->ctx = NULL;
+
+       if (!b_data(&stream->buf))
+               qc_stream_desc_free(stream);
+}
+
 /* Function to automatically activate QUIC traces on stdout.
  * Activated via the compilation flag -DENABLE_QUIC_STDOUT_TRACES.
  * Main use for now is in the docker image for QUIC interop testing.