blob: 3a7aed156ee16e23682345aa0e7446296761d689 [file] [log] [blame]
#include <haproxy/quic_stream.h>
#include <import/eb64tree.h>
#include <haproxy/api.h>
#include <haproxy/buf.h>
#include <haproxy/list.h>
#include <haproxy/dynbuf.h>
#include <haproxy/pool.h>
#include <haproxy/xprt_quic.h>
DECLARE_STATIC_POOL(pool_head_quic_conn_stream, "qc_stream_desc",
sizeof(struct qc_stream_desc));
DECLARE_STATIC_POOL(pool_head_quic_conn_stream_buf, "qc_stream_buf",
sizeof(struct qc_stream_buf));
/* Allocate a new stream descriptor with id <id>. The caller is responsible to
* store the stream in the appropriate tree.
*
* Returns the newly allocated instance on success or else NULL.
*/
struct qc_stream_desc *qc_stream_desc_new(uint64_t id, enum qcs_type type, void *ctx,
struct quic_conn *qc)
{
struct qc_stream_desc *stream;
stream = pool_alloc(pool_head_quic_conn_stream);
if (!stream)
return NULL;
stream->by_id.key = id;
eb64_insert(&qc->streams_by_id, &stream->by_id);
qc->rx.strms[type].nb_streams++;
stream->qc = qc;
stream->buf = NULL;
LIST_INIT(&stream->buf_list);
stream->buf_offset = 0;
stream->acked_frms = EB_ROOT;
stream->ack_offset = 0;
stream->release = 0;
stream->ctx = ctx;
return stream;
}
/* Mark the stream descriptor <stream> as released. It will be freed as soon as
* all its buffered data are acknowledged.
*/
void qc_stream_desc_release(struct qc_stream_desc *stream)
{
/* A stream can be released only one time. */
BUG_ON(stream->release);
stream->release = 1;
stream->ctx = NULL;
if (LIST_ISEMPTY(&stream->buf_list)) {
/* if no buffer left we can free the stream. */
qc_stream_desc_free(stream);
}
else {
/* A released stream does not use <stream.buf>. */
stream->buf = NULL;
}
}
/* 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 quic_conn *qc = s->qc;
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);
/* notify MUX about available buffers. */
--qc->stream_buf_count;
if (qc->mux_state == QC_MUX_READY) {
if (qc->qcc->flags & QC_CF_CONN_FULL) {
qc->qcc->flags &= ~QC_CF_CONN_FULL;
tasklet_wakeup(qc->qcc->wait_event.tasklet);
}
}
/* 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.
*/
void qc_stream_desc_free(struct qc_stream_desc *stream)
{
struct qc_stream_buf *buf, *buf_back;
struct quic_conn *qc = stream->qc;
struct eb64_node *frm_node;
unsigned int free_count = 0;
/* This function only deals with released streams. */
BUG_ON(!stream->release);
/* free remaining stream buffers */
list_for_each_entry_safe(buf, buf_back, &stream->buf_list, list) {
if (!(b_data(&buf->buf))) {
b_free(&buf->buf);
LIST_DELETE(&buf->list);
pool_free(pool_head_quic_conn_stream_buf, buf);
++free_count;
}
}
if (free_count) {
offer_buffers(NULL, free_count);
qc->stream_buf_count -= free_count;
if (qc->mux_state == QC_MUX_READY) {
/* notify MUX about available buffers. */
if (qc->qcc->flags & QC_CF_CONN_FULL) {
qc->qcc->flags &= ~QC_CF_CONN_FULL;
tasklet_wakeup(qc->qcc->wait_event.tasklet);
}
}
}
/* qc_stream_desc might be freed before having received all its ACKs.
* This is the case if some frames were retransmitted.
*/
frm_node = eb64_first(&stream->acked_frms);
while (frm_node) {
struct quic_stream *strm;
struct quic_frame *frm;
strm = eb64_entry(&frm_node->node, struct quic_stream, offset);
frm_node = eb64_next(frm_node);
eb64_delete(&strm->offset);
frm = container_of(strm, struct quic_frame, stream);
qc_release_frm(qc, frm);
}
eb64_delete(&stream->by_id);
pool_free(pool_head_quic_conn_stream, stream);
}
/* Return the current buffer of <stream>. May be NULL if not allocated. */
struct buffer *qc_stream_buf_get(struct qc_stream_desc *stream)
{
if (!stream->buf)
return NULL;
return &stream->buf->buf;
}
/* Check if a new stream buffer can be allocated for the connection <qc>.
* Returns a boolean.
*/
static int qc_stream_buf_avail(struct quic_conn *qc)
{
return qc->stream_buf_count < global.tune.quic_streams_buf;
}
/* Allocate a new current buffer for <stream>. The buffer limit count for the
* connection is checked first. This function is not allowed if current buffer
* is not NULL prior to this call. The new buffer represents stream payload at
* offset <offset>.
*
* Returns the buffer or NULL.
*/
struct buffer *qc_stream_buf_alloc(struct qc_stream_desc *stream,
uint64_t offset)
{
struct quic_conn *qc = stream->qc;
/* current buffer must be released first before allocate a new one. */
BUG_ON(stream->buf);
if (!qc_stream_buf_avail(qc))
return NULL;
++qc->stream_buf_count;
stream->buf_offset = offset;
stream->buf = pool_alloc(pool_head_quic_conn_stream_buf);
if (!stream->buf)
return NULL;
stream->buf->buf = BUF_NULL;
LIST_APPEND(&stream->buf_list, &stream->buf->list);
return &stream->buf->buf;
}
/* Release the current buffer of <stream>. It will be kept internally by
* the <stream>. The current buffer cannot be NULL.
*/
void qc_stream_buf_release(struct qc_stream_desc *stream)
{
/* current buffer already released */
BUG_ON(!stream->buf);
stream->buf = NULL;
stream->buf_offset = 0;
}