RepoMirrors/haproxy
1
0
Fork 0
mirror of http://git.haproxy.org/git/haproxy.git/ synced 2025-02-22 13:46:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

MINOR: quic-stream: refactor ack management

Browse Source 
Acknowledge of STREAM has been complexified with the introduction of
stream multi buffers. Two functions are executing roughly the same set
of instructions in xprt_quic.c.

To simplify this, move the code complexity in a new function
qc_stream_desc_ack(). It will handle offset calculation, removal of
data, freeing oldest buffer and freeing stream instance if required.
The qc_stream_desc API is cleaner as qc_stream_desc_free_buf() ambiguous
function has been removed.
This commit is contained in:
Amaury Denoyelle 2022-04-21 09:32:53 +02:00
parent a456920491
commit 1b81dda3e0
3 changed files with 85 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/haproxy/quic_stream.h
View File

@ -10,13 +10,13 @@ struct quic_conn;
struct qc_stream_desc *qc_stream_desc_new(uint64_t id, void *ctx,
struct quic_conn *qc);
void qc_stream_desc_release(struct qc_stream_desc *stream);
int qc_stream_desc_ack(struct qc_stream_desc **stream, size_t offset, size_t len);
void qc_stream_desc_free(struct qc_stream_desc *stream);
struct buffer *qc_stream_buf_get(struct qc_stream_desc *stream);
struct buffer *qc_stream_buf_alloc(struct qc_stream_desc *stream,
uint64_t offset);
void qc_stream_buf_release(struct qc_stream_desc *stream);
int qc_stream_desc_free_buf(struct qc_stream_desc *stream);
#endif /* USE_QUIC */
#endif /* _HAPROXY_QUIC_STREAM_H_ */

98
src/quic_stream.c
View File

@ -66,6 +66,64 @@ void qc_stream_desc_release(struct qc_stream_desc *stream)
}
}
/* Acknowledge data at <offset> of length <len> for <stream>. It is handled
* only if it covers a range corresponding to stream.ack_offset. 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.
*
* Returns the count of byte removed from stream. Do not forget to check if
* <stream> is NULL after invocation.
*/
int qc_stream_desc_ack(struct qc_stream_desc **stream, size_t offset,
size_t len)
{
struct qc_stream_desc *s = *stream;
struct qc_stream_buf *stream_buf;
struct buffer *buf;
size_t diff;
if (offset + len <= s->ack_offset || offset > s->ack_offset)
return 0;
/* There must be at least a buffer or we must not report an ACK. */
BUG_ON(LIST_ISEMPTY(&s->buf_list));
/* get oldest buffer from buf_list */
stream_buf = LIST_NEXT(&s->buf_list, struct qc_stream_buf *, list);
buf = &stream_buf->buf;
diff = offset + len - s->ack_offset;
s->ack_offset += diff;
b_del(buf, diff);
/* nothing more to do if buf still not empty. */
if (b_data(buf))
return diff;
/* buf is empty and can now be freed. Do not forget to reset current
* buf ptr if we were working on it.
*/
LIST_DELETE(&stream_buf->list);
if (stream_buf == s->buf) {
/* current buf must always be last entry in buflist */
BUG_ON(!LIST_ISEMPTY(&s->buf_list));
s->buf = NULL;
}
b_free(buf);
pool_free(pool_head_quic_conn_stream_buf, stream_buf);
offer_buffers(NULL, 1);
/* Free stream instance if already released and no buffers left. */
if (s->release && LIST_ISEMPTY(&s->buf_list)) {
qc_stream_desc_free(s);
*stream = NULL;
}
return diff;
}
/* Free the stream descriptor <stream> content. This function should be used
* when all its data have been acknowledged or on full connection closing. It
* must only be called after the stream is released.
@ -159,43 +217,3 @@ void qc_stream_buf_release(struct qc_stream_desc *stream)
stream->buf = NULL;
stream->buf_offset = 0;
}
/* Free the oldest buffer of <stream>. If the stream was already released and
* does not references any buffers, it is freed. This function must only be
* called if at least one buffer is present. Use qc_stream_desc_free() to free
* a released stream.
*
* Returns 0 if the stream is not yet freed else 1.
*/
int qc_stream_desc_free_buf(struct qc_stream_desc *stream)
{
struct qc_stream_buf *stream_buf;
BUG_ON(LIST_ISEMPTY(&stream->buf_list) && !stream->buf);
if (!LIST_ISEMPTY(&stream->buf_list)) {
/* get first buffer from buf_list and remove it */
stream_buf = LIST_NEXT(&stream->buf_list, struct qc_stream_buf *,
list);
LIST_DELETE(&stream_buf->list);
}
else {
stream_buf = stream->buf;
stream->buf = NULL;
}
b_free(&stream_buf->buf);
pool_free(pool_head_quic_conn_stream_buf, stream_buf);
offer_buffers(NULL, 1);
/* If qc_stream_desc is released and does not contains any buffers, we
* can free it now.
*/
if (stream->release && LIST_ISEMPTY(&stream->buf_list)) {
qc_stream_desc_free(stream);
return 1;
}
return 0;
}

62
src/xprt_quic.c
View File

@ -1444,28 +1444,24 @@ static int quic_stream_try_to_consume(struct quic_conn *qc,
while (frm_node) {
struct quic_stream *strm;
struct quic_frame *frm;
size_t offset, len;
strm = eb64_entry(&frm_node->node, struct quic_stream, offset);
if (strm->offset.key > stream->ack_offset)
offset = strm->offset.key;
len = strm->len;
if (offset > stream->ack_offset)
break;
TRACE_PROTO("stream consumed", QUIC_EV_CONN_ACKSTRM,
qc, strm, stream);
if (strm->offset.key + strm->len > stream->ack_offset) {
const size_t diff = strm->offset.key + strm->len -
stream->ack_offset;
stream->ack_offset += diff;
b_del(strm->buf, diff);
if (!b_data(strm->buf)) {
if (qc_stream_desc_free_buf(stream)) {
/* stream is freed here */
return 1;
}
}
if (qc_stream_desc_ack(&stream, offset, len)) {
TRACE_PROTO("stream consumed", QUIC_EV_CONN_ACKSTRM,
qc, strm, stream);
ret = 1;
}
if (!stream)
return 1;
frm_node = eb64_next(frm_node);
eb64_delete(&strm->offset);
@ -1492,6 +1488,8 @@ static inline void qc_treat_acked_tx_frm(struct quic_conn *qc,
struct quic_stream *strm_frm = &frm->stream;
struct eb64_node *node = NULL;
struct qc_stream_desc *stream = NULL;
const size_t offset = strm_frm->offset.key;
const size_t len = strm_frm->len;
/* 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
@ -1513,30 +1511,22 @@ static inline void qc_treat_acked_tx_frm(struct quic_conn *qc,
stream = eb64_entry(node, struct qc_stream_desc, by_id);
TRACE_PROTO("acked stream", QUIC_EV_CONN_ACKSTRM, qc, strm_frm, stream);
if (strm_frm->offset.key <= stream->ack_offset) {
if (strm_frm->offset.key + strm_frm->len > stream->ack_offset) {
const size_t diff = strm_frm->offset.key + strm_frm->len -
stream->ack_offset;
stream->ack_offset += diff;
b_del(strm_frm->buf, diff);
if (offset <= stream->ack_offset) {
if (qc_stream_desc_ack(&stream, offset, len)) {
stream_acked = 1;
if (!b_data(strm_frm->buf)) {
if (qc_stream_desc_free_buf(stream)) {
/* stream is freed at this stage,
* no need to continue.
*/
TRACE_PROTO("stream released and freed", QUIC_EV_CONN_ACKSTRM, qc);
LIST_DELETE(&frm->list);
quic_tx_packet_refdec(frm->pkt);
pool_free(pool_head_quic_frame, frm);
break;
}
}
TRACE_PROTO("stream consumed", QUIC_EV_CONN_ACKSTRM,
qc, strm_frm, stream);
}
if (!stream) {
/* no need to continue if stream freed. */
TRACE_PROTO("stream released and freed", QUIC_EV_CONN_ACKSTRM, qc);
LIST_DELETE(&frm->list);
quic_tx_packet_refdec(frm->pkt);
pool_free(pool_head_quic_frame, frm);
break;
}
TRACE_PROTO("stream consumed", QUIC_EV_CONN_ACKSTRM,
qc, strm_frm, stream);
LIST_DELETE(&frm->list);
quic_tx_packet_refdec(frm->pkt);
pool_free(pool_head_quic_frame, frm);