MEDIUM: mux-quic: allow to only release one buffer

Previous commit relaxed conditions for QUIC MUX Tx buffer allocation.
This was necessary to prevent slow throughput when a standalone stream
could only allocate a single buffer. Now, when a buffer is released, it
is removed from connection buffer window, thus allowing a new
allocation. However, this virtually allows a QCS to allocate many
buffers without any limit.

To fix this, add a limit on the number of released buffer that a QCS may
have. Once this limit is reached, qcc_release_stream_txbuf() operation
will return an error, flagging QCS with QC_SF_BLK_MROOM. Each time an
already released buffer is freed after ACK reception, new
qcs_notify_buf() is called by qc_stream_desc layer to wakeup the QCS
stream if currently blocked.

The limit is configured via a define QMUX_MAX_QCS_TXBUF_RELEASED. It is
set to 1 by default. This value is low enough to guarantee minimal
memory consumption for a QUIC connection while preserving transfer
throughput. Thus, the new buffer allocation limit is now :
  qc_window_size + (1 * nb_streams)

Author: a-denoyelle

include/haproxy/mux_quic-t.h

@@ -19,6 +19,9 @@
 #include <haproxy/stconn-t.h>
 #include <haproxy/time-t.h>
 
+/* Maximim number of released buffer a QCS may have. */
+#define QMUX_MAX_QCS_TXBUF_RELEASED    1
+
 /* Stream types */
 enum qcs_type {
 	QCS_CLT_BIDI,
@@ -140,6 +143,7 @@ struct qcs {
 	} rx;
 	struct {
 		struct quic_fctl fc; /* stream flow control applied on sending */
+		int buf_released; /* count of released buffer waiting for ACK */
 	} tx;
 
 	struct eb64_node by_id;

include/haproxy/mux_quic.h

@@ -23,6 +23,7 @@ int qcs_is_close_remote(struct qcs *qcs);
 int qcs_subscribe(struct qcs *qcs, int event_type, struct wait_event *es);
 void qcs_notify_recv(struct qcs *qcs);
 void qcs_notify_send(struct qcs *qcs);
+void qcs_notify_buf(struct qcs *qcs);
 void qcc_notify_buf(struct qcc *qcc, uint64_t free_size);
 
 struct buffer *qcc_get_stream_rxbuf(struct qcs *qcs);

src/mux_quic.c

@@ -134,6 +134,8 @@ static struct qcs *qcs_new(struct qcc *qcc, uint64_t id, enum qcs_type type)
 		qfctl_init(&qcs->tx.fc, 0);
 	}
 
+	qcs->tx.buf_released = 0;
+
 	qcs->rx.ncbuf = NCBUF_NULL;
 	qcs->rx.app_buf = BUF_NULL;
 	qcs->rx.offset = qcs->rx.offset_max = 0;
@@ -524,6 +526,22 @@ void qcs_notify_send(struct qcs *qcs)
 	}
 }
 
+/* Report that a released buffer from <qcs> have been freed. A released buffer
+ * is already removed from QCC buffer window. However, <qcs> stream emission
+ * may also be blocked as it could not release anymore buffer.
+ *
+ * Noop if <qcs> is NULL.
+ */
+void qcs_notify_buf(struct qcs *qcs)
+{
+	if (qcs) {
+		BUG_ON(!qcs->tx.buf_released);
+		--qcs->tx.buf_released;
+		qcs_notify_send(qcs);
+		qcs->flags &= ~QC_SF_BLK_MROOM;
+	}
+}
+
 /* Returns true if <qcc> buffer window does not have room for a new buffer. */
 static inline int qcc_bufwnd_full(const struct qcc *qcc)
 {
@@ -1159,17 +1177,20 @@ int qcc_realign_stream_txbuf(const struct qcs *qcs, struct buffer *out)
 int qcc_release_stream_txbuf(struct qcs *qcs)
 {
 	struct buffer *buf = qc_stream_buf_get(qcs->stream);
-	const uint64_t bytes = qcs_prep_bytes(qcs);
 
-	/* Cannot release buffer if prepared data is not fully sent. */
-	if (bytes) {
+	/* Cannot release buffer if prepared data is not fully sent or already
+	 * reached the maximum number of released buffer for this stream.
+	 */
+	if (qcs_prep_bytes(qcs) ||
+	    qcs->tx.buf_released >= QMUX_MAX_QCS_TXBUF_RELEASED) {
 		qcs->flags |= QC_SF_BLK_MROOM;
 		return 1;
 	}
 
 	/* Free released buf size from buf window. */
 	qc_stream_buf_release(qcs->stream);
 	qcc_notify_buf(qcs->qcc, b_size(buf));
+	++qcs->tx.buf_released;
 
 	return 0;
 }
@@ -1977,11 +1998,11 @@ void qcc_streams_sent_done(struct qcs *qcs, uint64_t data, uint64_t offset)
 			            QMUX_EV_QCS_SEND, qcc->conn, qcs);
 		}
 		/* Release buffer if everything sent and buf is full or stream is waiting for room. */
-		if (!qcs_prep_bytes(qcs) &&
-		    (b_full(&qcs->stream->buf->buf) || qcs->flags & QC_SF_BLK_MROOM)) {
-			qcc_release_stream_txbuf(qcs);
-			qcs->flags &= ~QC_SF_BLK_MROOM;
-			qcs_notify_send(qcs);
+		if ((b_full(&qcs->stream->buf->buf) || qcs->flags & QC_SF_BLK_MROOM)) {
+			if (!qcc_release_stream_txbuf(qcs)) {
+				qcs_notify_send(qcs);
+				qcs->flags &= ~QC_SF_BLK_MROOM;
+			}
 		}
 
 		/* Add measurement for send rate. This is done at the MUX layer
@@ -3419,7 +3440,7 @@ void qcc_show_quic(struct qcc *qcc)
 		if (!quic_stream_is_uni(qcs->id) || !quic_stream_is_local(qcc, qcs->id))
 			chunk_appendf(&trash, " rxoff=%llu", (ullong)qcs->rx.offset);
 		if (!quic_stream_is_uni(qcs->id) || !quic_stream_is_remote(qcc, qcs->id))
-			chunk_appendf(&trash, " txoff=%llu", (ullong)qcs->tx.fc.off_real);
+			chunk_appendf(&trash, " txoff=%llu br=%d", (ullong)qcs->tx.fc.off_real, qcs->tx.buf_released);
 		chunk_appendf(&trash, "\n");
 		node = eb64_next(node);
 	}

src/qmux_trace.c

@@ -150,9 +150,9 @@ void qmux_dump_qcs_info(struct buffer *msg, const struct qcs *qcs)
 	              qcs_st_to_str(qcs->st), qcs->flags);
 
 	chunk_appendf(msg, " .rx=%llu/%llu", (ullong)qcs->rx.offset_max, (ullong)qcs->rx.msd);
-	chunk_appendf(msg, " .tx=%llu %llu/%llu", (ullong)qcs->tx.fc.off_soft,
-	                                          (ullong)qcs->tx.fc.off_real,
-	                                          (ullong)qcs->tx.fc.limit);
+	chunk_appendf(msg, " .tx=%llu %llu/%llu br=%d",
+	              (ullong)qcs->tx.fc.off_soft, (ullong)qcs->tx.fc.off_real,
+	              (ullong)qcs->tx.fc.limit, qcs->tx.buf_released);
 
 	chunk_appendf(msg, " .ti=%u/%u/%u",
 	              tot_time_read(&qcs->timer.base),

src/quic_stream.c

@@ -45,6 +45,10 @@ static void qc_stream_buf_free(struct qc_stream_desc *stream,
 			}
 		}
 	}
+	else {
+		/* Notify about a freed released buffer. */
+		qcs_notify_buf(stream->ctx);
+	}
 
 	if ((*stream_buf)->sbuf) {
 		pool_free(pool_head_sbuf, buf->area);