MEDIUM/OPTIM: mux-quic: define a recv_list for demux resumption

This commit is part of the current serie which aims to refactor and
improve overall performance of QUIC MUX I/O handler.

Define a recv_list element into qcc structure. This is used to
registered every instance of qcs which are currently blocked on
demuxing, which happen on no more space in <rx.appbuf>.

The purpose of this patch is to reduce qcc_io_recv() CPU usage. Now,
only recv_list iteration is performed, instead of the previous looping
over every qcs instances. This is useful as qcc_io_recv() is called each
time qcc_io_cb() is scheduled, even if only sending condition was the
wakeup origin.

A qcs is not inserted into recv_list immediately after blocking on demux
full buffer. Instead, this is only done after unblocking via stream
rcv_buf callback, which ensure that new buffer space is available.

This should be backported up to 3.1.
This commit is contained in:
Amaury Denoyelle 2024-12-05 10:48:51 +01:00
parent 0a53a008d0
commit 4b42dd4ae0
2 changed files with 16 additions and 22 deletions

View File

@ -82,6 +82,7 @@ struct qcc {
struct eb_root streams_by_id; /* all active streams by their ID */ struct eb_root streams_by_id; /* all active streams by their ID */
struct list recv_list; /* list of qcs for which demux can be resumed */
struct list send_list; /* list of qcs ready to send (STREAM, STOP_SENDING or RESET_STREAM emission) */ struct list send_list; /* list of qcs ready to send (STREAM, STOP_SENDING or RESET_STREAM emission) */
struct list fctl_list; /* list of sending qcs blocked on conn flow control */ struct list fctl_list; /* list of sending qcs blocked on conn flow control */
struct list buf_wait_list; /* list of qcs blocked on stream desc buf */ struct list buf_wait_list; /* list of qcs blocked on stream desc buf */
@ -150,6 +151,7 @@ struct qcs {
uint64_t id; uint64_t id;
struct qc_stream_desc *stream; struct qc_stream_desc *stream;
struct list el_recv; /* element of qcc.recv_list */
struct list el_send; /* element of qcc.send_list */ struct list el_send; /* element of qcc.send_list */
struct list el_opening; /* element of qcc.opening_list */ struct list el_opening; /* element of qcc.opening_list */
struct list el_fctl; /* element of qcc.fctl_list */ struct list el_fctl; /* element of qcc.fctl_list */

View File

@ -85,6 +85,7 @@ static void qcs_free(struct qcs *qcs)
/* Safe to use even if already removed from the list. */ /* Safe to use even if already removed from the list. */
LIST_DEL_INIT(&qcs->el_opening); LIST_DEL_INIT(&qcs->el_opening);
LIST_DEL_INIT(&qcs->el_recv);
LIST_DEL_INIT(&qcs->el_send); LIST_DEL_INIT(&qcs->el_send);
LIST_DEL_INIT(&qcs->el_fctl); LIST_DEL_INIT(&qcs->el_fctl);
LIST_DEL_INIT(&qcs->el_buf); LIST_DEL_INIT(&qcs->el_buf);
@ -138,6 +139,7 @@ static struct qcs *qcs_new(struct qcc *qcc, uint64_t id, enum qcs_type type)
* These fields must be initialed before. * These fields must be initialed before.
*/ */
LIST_INIT(&qcs->el_opening); LIST_INIT(&qcs->el_opening);
LIST_INIT(&qcs->el_recv);
LIST_INIT(&qcs->el_send); LIST_INIT(&qcs->el_send);
LIST_INIT(&qcs->el_fctl); LIST_INIT(&qcs->el_fctl);
LIST_INIT(&qcs->el_buf); LIST_INIT(&qcs->el_buf);
@ -1605,6 +1607,7 @@ int qcc_recv(struct qcc *qcc, uint64_t id, uint64_t len, uint64_t offset,
if ((ncb_data(&qcs->rx.ncbuf, 0) && !(qcs->flags & QC_SF_DEM_FULL)) || fin) { if ((ncb_data(&qcs->rx.ncbuf, 0) && !(qcs->flags & QC_SF_DEM_FULL)) || fin) {
qcc_decode_qcs(qcc, qcs); qcc_decode_qcs(qcc, qcs);
LIST_DEL_INIT(&qcs->el_recv);
qcc_refresh_timeout(qcc); qcc_refresh_timeout(qcc);
} }
@ -2509,14 +2512,13 @@ static void qcc_wait_for_hs(struct qcc *qcc)
TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn); TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
} }
/* Proceed on receiving. Loop through all streams from <qcc> and use decode_qcs /* Proceed on receiving. Loop on streams subscribed in recv_list and performed
* operation. * STREAM frames decoding upon them.
* *
* Returns 0 on success else non-zero. * Returns 0 on success else non-zero.
*/ */
static int qcc_io_recv(struct qcc *qcc) static int qcc_io_recv(struct qcc *qcc)
{ {
struct eb64_node *node;
struct qcs *qcs; struct qcs *qcs;
TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn); TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
@ -2530,25 +2532,12 @@ static int qcc_io_recv(struct qcc *qcc)
if ((qcc->flags & QC_CF_WAIT_HS) && !(qcc->wait_event.events & SUB_RETRY_RECV)) if ((qcc->flags & QC_CF_WAIT_HS) && !(qcc->wait_event.events & SUB_RETRY_RECV))
qcc_wait_for_hs(qcc); qcc_wait_for_hs(qcc);
node = eb64_first(&qcc->streams_by_id); while (!LIST_ISEMPTY(&qcc->recv_list)) {
while (node) { qcs = LIST_ELEM(qcc->recv_list.n, struct qcs *, el_recv);
uint64_t id; /* No need to add an uni local stream in recv_list. */
BUG_ON(quic_stream_is_uni(qcs->id) && quic_stream_is_local(qcc, qcs->id));
qcs = eb64_entry(node, struct qcs, by_id);
id = qcs->id;
if (!ncb_data(&qcs->rx.ncbuf, 0) || (qcs->flags & QC_SF_DEM_FULL)) {
node = eb64_next(node);
continue;
}
if (quic_stream_is_uni(id) && quic_stream_is_local(qcc, id)) {
node = eb64_next(node);
continue;
}
qcc_decode_qcs(qcc, qcs); qcc_decode_qcs(qcc, qcs);
node = eb64_next(node); LIST_DEL_INIT(&qcs->el_recv);
} }
TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn); TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
@ -2998,6 +2987,7 @@ static int qmux_init(struct connection *conn, struct proxy *prx,
goto err; goto err;
} }
LIST_INIT(&qcc->recv_list);
LIST_INIT(&qcc->send_list); LIST_INIT(&qcc->send_list);
LIST_INIT(&qcc->fctl_list); LIST_INIT(&qcc->fctl_list);
LIST_INIT(&qcc->buf_wait_list); LIST_INIT(&qcc->buf_wait_list);
@ -3213,9 +3203,11 @@ static size_t qmux_strm_rcv_buf(struct stconn *sc, struct buffer *buf,
BUG_ON(!ncb_data(&qcs->rx.ncbuf, 0)); BUG_ON(!ncb_data(&qcs->rx.ncbuf, 0));
qcs->flags &= ~QC_SF_DEM_FULL; qcs->flags &= ~QC_SF_DEM_FULL;
if (!(qcc->flags & QC_CF_ERRL)) if (!(qcc->flags & QC_CF_ERRL)) {
LIST_APPEND(&qcc->recv_list, &qcs->el_recv);
qcc_wakeup(qcc); qcc_wakeup(qcc);
} }
}
TRACE_LEAVE(QMUX_EV_STRM_RECV, qcc->conn, qcs); TRACE_LEAVE(QMUX_EV_STRM_RECV, qcc->conn, qcs);