mirrors/haproxy
1
0
Fork 0
mirror of https://git.haproxy.org/git/haproxy.git/ synced 2025-08-07 07:37:02 +02:00
Code Issues Projects Releases Wiki Activity

MINOR: quic: move buffered ACK to streambuf

Browse Source 
QUIC streamdesc layer is used to manage QUIC MUX stream txbuf data
storage until acknowledgment. Currently, it only supports in-order
acknowledgment at the stream level. This requires to be able to buffer
out-of-order ACKs until they can be handled.

Previously, these ACKs were stored in a tree to the streamdesc instance.
Move this indexed storage at the streambuf instance.

This commit is purely an architecture change. However, it will allow to
extend ACK management in future patches, such as the ability to merge
overlapping out-of-order ACKs.
This commit is contained in:
Amaury Denoyelle 2024-10-01 11:13:41 +02:00
parent 943e48dadd
commit 62558a9285
3 changed files with 85 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/haproxy/quic_stream-t.h
View File

@ -15,6 +15,7 @@
* can be freed in strict order.
*/
struct qc_stream_buf {
struct eb_root acked_frms; /* storage for out-of-order ACKs */
struct eb64_node offset_node; /* node for qc_stream_desc buf tree */
struct buffer buf; /* STREAM payload */
int sbuf;
@ -41,7 +42,6 @@ struct qc_stream_desc {
uint64_t ack_offset; /* last acknowledged offset */
struct eb_root buf_tree; /* list of active and released buffers */
struct eb_root acked_frms; /* ACK frames tree for non-contiguous ACK ranges */
int flags; /* QC_SD_FL_* values */

76
src/quic_rx.c
View File

@ -210,37 +210,59 @@ static int quic_stream_try_to_consume(struct quic_conn *qc,
{
int ret;
struct eb64_node *frm_node;
struct qc_stream_buf *stream_buf;
struct eb64_node *buf_node;
TRACE_ENTER(QUIC_EV_CONN_ACKSTRM, qc);
ret = 0;
frm_node = eb64_first(&stream->acked_frms);
while (frm_node) {
struct qf_stream *strm_frm;
struct quic_frame *frm;
size_t offset, len;
int fin, ack;
buf_node = eb64_first(&stream->buf_tree);
if (buf_node) {
stream_buf = eb64_entry(buf_node, struct qc_stream_buf,
offset_node);
strm_frm = eb64_entry(frm_node, struct qf_stream, offset);
frm = container_of(strm_frm, struct quic_frame, stream);
offset = strm_frm->offset.key;
len = strm_frm->len;
fin = frm->type & QUIC_STREAM_FRAME_TYPE_FIN_BIT;
frm_node = eb64_first(&stream_buf->acked_frms);
while (frm_node) {
struct qf_stream *strm_frm;
struct quic_frame *frm;
size_t offset, len;
int fin;
if (offset > stream->ack_offset)
break;
strm_frm = eb64_entry(frm_node, struct qf_stream, offset);
frm = container_of(strm_frm, struct quic_frame, stream);
offset = strm_frm->offset.key;
len = strm_frm->len;
fin = frm->type & QUIC_STREAM_FRAME_TYPE_FIN_BIT;
ack = qc_stream_desc_ack(stream, offset, len, fin);
if (ack) {
TRACE_DEVEL("stream consumed", QUIC_EV_CONN_ACKSTRM,
qc, strm_frm, stream);
ret = 1;
if (offset > stream->ack_offset)
break;
/* First delete frm from tree. This prevents BUG_ON() if
* stream_buf instance is removed via qc_stream_desc_ack().
*/
eb64_delete(frm_node);
qc_release_frm(qc, frm);
if (qc_stream_desc_ack(stream, offset, len, fin)) {
TRACE_DEVEL("stream consumed", QUIC_EV_CONN_ACKSTRM,
qc, strm_frm, stream);
ret = 1;
}
/* Always retrieve first buffer as the previously used
* instance could have been removed during qc_stream_desc_ack().
*/
buf_node = eb64_first(&stream->buf_tree);
if (buf_node) {
stream_buf = eb64_entry(buf_node, struct qc_stream_buf,
offset_node);
frm_node = eb64_first(&stream_buf->acked_frms);
BUG_ON(!frm_node && !b_data(&stream_buf->buf));
}
else {
frm_node = NULL;
}
}
frm_node = eb64_next(frm_node);
eb64_delete(&strm_frm->offset);
qc_release_frm(qc, frm);
}
TRACE_LEAVE(QUIC_EV_CONN_ACKSTRM, qc);
@ -302,7 +324,13 @@ static void qc_handle_newly_acked_frm(struct quic_conn *qc, struct quic_frame *f
qc_release_frm(qc, frm);
}
else {
eb64_insert(&stream->acked_frms, &strm_frm->offset);
struct qc_stream_buf *stream_buf;
struct eb64_node *buf_node;
buf_node = eb64_lookup_le(&stream->buf_tree, offset);
BUG_ON(!buf_node);
stream_buf = eb64_entry(buf_node, struct qc_stream_buf, offset_node);
eb64_insert(&stream_buf->acked_frms, &strm_frm->offset);
}
}
break;

50
src/quic_stream.c
View File

@ -9,6 +9,7 @@
#include <haproxy/mux_quic.h>
#include <haproxy/pool.h>
#include <haproxy/quic_conn.h>
#include <haproxy/quic_frame.h>
#include <haproxy/task.h>
DECLARE_STATIC_POOL(pool_head_quic_stream_desc, "qc_stream_desc",
@ -24,6 +25,9 @@ static void qc_stream_buf_free(struct qc_stream_desc *stream,
struct buffer *buf = &(*stream_buf)->buf;
uint64_t free_size;
/* Caller is responsible to remove buffered ACK frames before destroying a buffer instance. */
BUG_ON(!eb_is_empty(&(*stream_buf)->acked_frms));
eb64_delete(&(*stream_buf)->offset_node);
/* Reset current buf ptr if deleted instance is the same one. */
@ -75,7 +79,6 @@ struct qc_stream_desc *qc_stream_desc_new(uint64_t id, enum qcs_type type, void
stream->buf_tree = EB_ROOT_UNIQUE;
stream->buf_offset = 0;
stream->acked_frms = EB_ROOT;
stream->ack_offset = 0;
stream->flags = 0;
stream->ctx = ctx;
@ -166,6 +169,20 @@ int qc_stream_desc_ack(struct qc_stream_desc *stream, size_t offset, size_t len,
/* Free oldest buffer if all data acknowledged. */
if (!b_data(buf)) {
/* Remove buffered ACK before deleting buffer instance. */
while (!eb_is_empty(&stream_buf->acked_frms)) {
struct quic_conn *qc = stream->qc;
struct eb64_node *frm_node;
struct qf_stream *strm_frm;
struct quic_frame *frm;
frm_node = eb64_first(&stream_buf->acked_frms);
eb64_delete(frm_node);
strm_frm = eb64_entry(frm_node, struct qf_stream, offset);
frm = container_of(strm_frm, struct quic_frame, stream);
qc_release_frm(qc, frm);
}
qc_stream_buf_free(stream, &stream_buf);
buf = NULL;
}
@ -204,6 +221,19 @@ void qc_stream_desc_free(struct qc_stream_desc *stream, int closing)
*/
BUG_ON(b_data(&buf->buf) && !closing);
/* qc_stream_desc might be freed before having received all its ACKs. */
while (!eb_is_empty(&buf->acked_frms)) {
struct qf_stream *strm_frm;
struct quic_frame *frm;
frm_node = eb64_first(&buf->acked_frms);
eb64_delete(frm_node);
strm_frm = eb64_entry(frm_node, struct qf_stream, offset);
frm = container_of(strm_frm, struct quic_frame, stream);
qc_release_frm(qc, frm);
}
if (buf->sbuf)
pool_free(pool_head_sbuf, buf->buf.area);
else
@ -217,23 +247,6 @@ void qc_stream_desc_free(struct qc_stream_desc *stream, int closing)
if (free_count)
offer_buffers(NULL, free_count);
/* 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);
@ -265,6 +278,7 @@ struct buffer *qc_stream_buf_alloc(struct qc_stream_desc *stream,
if (!stream->buf)
return NULL;
stream->buf->acked_frms = EB_ROOT;
stream->buf->buf = BUF_NULL;
stream->buf->offset_node.key = offset;