mirrors/haproxy
1
0
Fork 0
mirror of https://git.haproxy.org/git/haproxy.git/ synced 2025-11-26 21:31:01 +01:00
Code Issues Projects Releases Wiki Activity

MINOR: quic: do not remove qc_stream_desc automatically on ACK handling

Browse Source 
qc_stream_desc_ack() is used to handle ACK received for STREAM frame. It
removes acknowledged data from their underlying buffer.

If all data were removed after ACK handling, qc_stream_desc instance
would automatically be freed at the end of qc_stream_desc_ack().
However, this renders the function complicated to use. Simplify this by
removing this automatic removal. Now, caller is responsible to check
after ACK handling if qc_stream_desc instance can be removed. This is
easily done using qc_stream_desc_done() helper.
This commit is contained in:
Amaury Denoyelle 2024-09-26 16:14:40 +02:00
parent db68f8ed86
commit f4a83fbb14
3 changed files with 35 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
include/haproxy/quic_stream.h
View File

@ -12,7 +12,7 @@ struct qc_stream_desc *qc_stream_desc_new(uint64_t id, enum qcs_type, void *ctx,
struct quic_conn *qc);
void qc_stream_desc_release(struct qc_stream_desc *stream, uint64_t final_size,
void *new_ctx);
int qc_stream_desc_ack(struct qc_stream_desc **stream, size_t offset, size_t len, int fin);
int qc_stream_desc_ack(struct qc_stream_desc *stream, size_t offset, size_t len, int fin);
void qc_stream_desc_free(struct qc_stream_desc *stream, int closing);
struct buffer *qc_stream_buf_get(struct qc_stream_desc *stream);
@ -21,6 +21,13 @@ struct buffer *qc_stream_buf_alloc(struct qc_stream_desc *stream,
struct buffer *qc_stream_buf_realloc(struct qc_stream_desc *stream);
void qc_stream_buf_release(struct qc_stream_desc *stream);
/* Returns true if nothing to ack yet for stream <s> including FIN bit. */
static inline int qc_stream_desc_done(const struct qc_stream_desc *s)
{
return (s->flags & (QC_SD_FL_RELEASE|QC_SD_FL_WAIT_FOR_FIN)) == QC_SD_FL_RELEASE &&
LIST_ISEMPTY(&s->buf_list);
}
/* Reports emission of STREAM frame starting at <offset> and of length <len>,
* related to <stream> data storage.
*/

43
src/quic_rx.c
View File

@ -219,7 +219,7 @@ static int quic_stream_try_to_consume(struct quic_conn *qc,
struct qf_stream *strm_frm;
struct quic_frame *frm;
size_t offset, len;
int fin;
int fin, ack;
strm_frm = eb64_entry(frm_node, struct qf_stream, offset);
frm = container_of(strm_frm, struct quic_frame, stream);
@ -230,30 +230,19 @@ static int quic_stream_try_to_consume(struct quic_conn *qc,
if (offset > stream->ack_offset)
break;
if (qc_stream_desc_ack(&stream, offset, len, fin)) {
/* cf. next comment : frame may be freed at this stage. */
ack = qc_stream_desc_ack(stream, offset, len, fin);
if (ack) {
TRACE_DEVEL("stream consumed", QUIC_EV_CONN_ACKSTRM,
qc, stream ? strm_frm : NULL, stream);
qc, strm_frm, stream);
ret = 1;
}
/* If stream is NULL after qc_stream_desc_ack(), it means frame
* has been freed. with the stream frames tree. Nothing to do
* anymore in here.
*/
if (!stream) {
qc_check_close_on_released_mux(qc);
ret = 1;
goto leave;
}
frm_node = eb64_next(frm_node);
eb64_delete(&strm_frm->offset);
qc_release_frm(qc, frm);
}
leave:
TRACE_LEAVE(QUIC_EV_CONN_ACKSTRM, qc);
return ret;
}
@ -276,6 +265,7 @@ static void qc_handle_newly_acked_frm(struct quic_conn *qc, struct quic_frame *f
const size_t offset = strm_frm->offset.key;
const size_t len = strm_frm->len;
const int fin = frm->type & QUIC_STREAM_FRAME_TYPE_FIN_BIT;
int ack;
/* do not use strm_frm->stream as the qc_stream_desc instance
* might be freed at this stage. Use the id to do a proper
@ -295,28 +285,25 @@ static void qc_handle_newly_acked_frm(struct quic_conn *qc, struct quic_frame *f
TRACE_DEVEL("acked stream", QUIC_EV_CONN_ACKSTRM, qc, strm_frm, stream);
if (offset <= stream->ack_offset) {
if (qc_stream_desc_ack(&stream, offset, len, fin)) {
ack = qc_stream_desc_ack(stream, offset, len, fin);
if (ack || fin) {
TRACE_DEVEL("stream consumed", QUIC_EV_CONN_ACKSTRM,
qc, strm_frm, stream);
quic_stream_try_to_consume(qc, stream);
if (qc_stream_desc_done(stream)) {
/* no need to continue if stream freed. */
TRACE_DEVEL("stream released and freed", QUIC_EV_CONN_ACKSTRM, qc);
qc_check_close_on_released_mux(qc);
}
}
if (!stream) {
/* no need to continue if stream freed. */
TRACE_DEVEL("stream released and freed", QUIC_EV_CONN_ACKSTRM, qc);
qc_release_frm(qc, frm);
qc_check_close_on_released_mux(qc);
break;
}
TRACE_DEVEL("stream consumed", QUIC_EV_CONN_ACKSTRM,
qc, strm_frm, stream);
qc_release_frm(qc, frm);
}
else {
eb64_insert(&stream->acked_frms, &strm_frm->offset);
}
quic_stream_try_to_consume(qc, stream);
}
break;
default:

40
src/quic_stream.c
View File

@ -19,12 +19,6 @@ DECLARE_STATIC_POOL(pool_head_quic_stream_buf, "qc_stream_buf",
static struct pool_head *pool_head_sbuf;
/* Returns true if nothing to ack yet for stream <s> including FIN bit. */
static inline int qc_stream_desc_done(const struct qc_stream_desc *s)
{
return !(s->flags & QC_SD_FL_WAIT_FOR_FIN) && LIST_ISEMPTY(&s->buf_list);
}
static void qc_stream_buf_free(struct qc_stream_desc *stream,
struct qc_stream_buf **stream_buf)
{
@ -139,58 +133,48 @@ void qc_stream_desc_release(struct qc_stream_desc *stream,
}
/* Acknowledge data at <offset> of length <len> for <stream> with <fin> set for
* the final data. After data removal, if the stream does not contains data
* any more and is already released, the instance stream is freed. <stream> is
* set to NULL to indicate this.
* the final data.
*
* Returns the count of byte removed from stream. Do not forget to check if
* <stream> is NULL after invocation.
* Returns the count of byte removed from stream.
*/
int qc_stream_desc_ack(struct qc_stream_desc **stream, size_t offset, size_t len,
int qc_stream_desc_ack(struct qc_stream_desc *stream, size_t offset, size_t len,
int fin)
{
struct qc_stream_desc *s = *stream;
struct qc_stream_buf *stream_buf = NULL;
struct buffer *buf = NULL;
size_t diff;
/* Cannot advertise FIN for an inferior data range. */
BUG_ON(fin && offset + len < s->ack_offset);
BUG_ON(fin && offset + len < stream->ack_offset);
/* No support now for out-of-order ACK reporting. */
BUG_ON(offset > s->ack_offset);
BUG_ON(offset > stream->ack_offset);
if (offset + len < s->ack_offset)
if (offset + len < stream->ack_offset)
return 0;
diff = offset + len - s->ack_offset;
diff = offset + len - stream->ack_offset;
if (diff) {
/* Buf list cannot be empty if there is still unacked data. */
BUG_ON(LIST_ISEMPTY(&s->buf_list));
BUG_ON(LIST_ISEMPTY(&stream->buf_list));
/* get oldest buffer from buf_list */
stream_buf = LIST_NEXT(&s->buf_list, struct qc_stream_buf *, list);
stream_buf = LIST_NEXT(&stream->buf_list, struct qc_stream_buf *, list);
buf = &stream_buf->buf;
s->ack_offset += diff;
stream->ack_offset += diff;
b_del(buf, diff);
/* Free oldest buffer if all data acknowledged. */
if (!b_data(buf)) {
qc_stream_buf_free(s, &stream_buf);
qc_stream_buf_free(stream, &stream_buf);
buf = NULL;
}
}
if (fin) {
/* Mark FIN as acknowledged. */
s->flags &= ~QC_SD_FL_WAIT_FOR_FIN;
}
/* Free stream instance if already released and everything acknowledged. */
if ((s->flags & QC_SD_FL_RELEASE) && qc_stream_desc_done(s)) {
qc_stream_desc_free(s, 0);
*stream = NULL;
stream->flags &= ~QC_SD_FL_WAIT_FOR_FIN;
}
return diff;