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

MINOR: quic: Do not forget STREAM frames received in disorder

Browse Source 
Add a function to process all STREAM frames received and ordered
by their offset (qc_treat_rx_strm_frms()) and modify
qc_handle_bidi_strm_frm() consequently.
This commit is contained in:
Frédéric Lécaille 2021-12-20 12:02:13 +01:00
parent 4137b2d316
commit f1d38cbe15
1 changed files with 72 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
src/xprt_quic.c
View File

@ -2053,6 +2053,69 @@ static size_t qc_strm_cpy(struct buffer *buf, struct quic_stream *strm_frm)
return ret; return ret;
} }
/* Copy as most as possible STREAM data from <strm_frm> into <buf> buffer.
* Also update <strm_frm> frame to reflect the data which have been consumed.
*/
static size_t qc_rx_strm_frm_cpy(struct buffer *buf,
struct quic_rx_strm_frm *strm_frm)
{
size_t ret;
ret = 0;
while (strm_frm->len) {
size_t try;
try = b_contig_space(buf);
if (!try)
break;
if (try > strm_frm->len)
try = strm_frm->len;
memcpy(b_tail(buf), strm_frm->data, try);
strm_frm->len -= try;
strm_frm->offset_node.key += try;
b_add(buf, try);
ret += try;
}
return ret;
}
/* Process as much as possible RX STREAM frames received for <qcs> */
static size_t qc_treat_rx_strm_frms(struct qcs *qcs)
{
int total;
struct eb64_node *frm_node;
total = 0;
frm_node = eb64_first(&qcs->rx.frms);
while (frm_node) {
int ret;
struct quic_rx_strm_frm *frm;
frm = eb64_entry(&frm_node->node, struct quic_rx_strm_frm, offset_node);
if (frm->offset_node.key != qcs->rx.offset)
break;
ret = qc_rx_strm_frm_cpy(&qcs->rx.buf, frm);
qcs->rx.offset += ret;
total += ret;
if (frm->len) {
/* If there is remaining data in this frame
* this is because the destination buffer is full.
*/
break;
}
frm_node = eb64_next(frm_node);
quic_rx_packet_refdec(frm->pkt);
eb64_delete(&frm->offset_node);
pool_free(pool_head_quic_rx_strm_frm, frm);
}
return total;
}
/* Handle <strm_frm> bidirectional STREAM frame. Depending on its ID, several /* Handle <strm_frm> bidirectional STREAM frame. Depending on its ID, several
* streams may be open. The data are copied to the stream RX buffer if possible. * streams may be open. The data are copied to the stream RX buffer if possible.
* If not, the STREAM frame is stored to be treated again later. * If not, the STREAM frame is stored to be treated again later.
@ -2063,6 +2126,7 @@ static int qc_handle_bidi_strm_frm(struct quic_rx_packet *pkt,
struct quic_stream *strm_frm, struct quic_stream *strm_frm,
struct quic_conn *qc) struct quic_conn *qc)
{ {
int total;
struct qcs *strm; struct qcs *strm;
struct eb64_node *strm_node, *frm_node; struct eb64_node *strm_node, *frm_node;
struct quic_rx_strm_frm *frm; struct quic_rx_strm_frm *frm;
@ -2082,6 +2146,7 @@ static int qc_handle_bidi_strm_frm(struct quic_rx_packet *pkt,
goto out; goto out;
} }
total = 0;
if (strm_frm->offset.key == strm->rx.offset) { if (strm_frm->offset.key == strm->rx.offset) {
int ret; int ret;
@ -2090,12 +2155,14 @@ static int qc_handle_bidi_strm_frm(struct quic_rx_packet *pkt,
} }
ret = qc_strm_cpy(&strm->rx.buf, strm_frm); ret = qc_strm_cpy(&strm->rx.buf, strm_frm);
if (ret && qc->qcc->app_ops->decode_qcs(strm, strm_frm->fin, qc->qcc->ctx) < 0) { total += ret;
TRACE_PROTO("Decoding error", QUIC_EV_CONN_PSTRM); strm->rx.offset += ret;
return 0;
} }
strm->rx.offset += ret; total += qc_treat_rx_strm_frms(strm);
if (total && qc->qcc->app_ops->decode_qcs(strm, strm_frm->fin, qc->qcc->ctx) < 0) {
TRACE_PROTO("Decoding error", QUIC_EV_CONN_PSTRM);
return 0;
} }
if (!strm_frm->len) if (!strm_frm->len)