summaryrefslogtreecommitdiffstats
path: root/src/mux_quic.c
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2024-04-28 09:35:11 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2024-04-28 09:35:11 +0000
commitda76459dc21b5af2449af2d36eb95226cb186ce2 (patch)
tree542ebb3c1e796fac2742495b8437331727bbbfa0 /src/mux_quic.c
parentInitial commit. (diff)
downloadhaproxy-da76459dc21b5af2449af2d36eb95226cb186ce2.tar.xz
haproxy-da76459dc21b5af2449af2d36eb95226cb186ce2.zip
Adding upstream version 2.6.12.upstream/2.6.12upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'src/mux_quic.c')
-rw-r--r--src/mux_quic.c2321
1 files changed, 2321 insertions, 0 deletions
diff --git a/src/mux_quic.c b/src/mux_quic.c
new file mode 100644
index 0000000..a8c7e19
--- /dev/null
+++ b/src/mux_quic.c
@@ -0,0 +1,2321 @@
+#include <haproxy/mux_quic.h>
+
+#include <import/eb64tree.h>
+
+#include <haproxy/api.h>
+#include <haproxy/connection.h>
+#include <haproxy/dynbuf.h>
+#include <haproxy/list.h>
+#include <haproxy/ncbuf.h>
+#include <haproxy/pool.h>
+#include <haproxy/qmux_http.h>
+#include <haproxy/qmux_trace.h>
+#include <haproxy/quic_conn.h>
+#include <haproxy/quic_stream.h>
+#include <haproxy/quic_tp-t.h>
+#include <haproxy/ssl_sock-t.h>
+#include <haproxy/stconn.h>
+#include <haproxy/trace.h>
+
+DECLARE_POOL(pool_head_qcc, "qcc", sizeof(struct qcc));
+DECLARE_POOL(pool_head_qcs, "qcs", sizeof(struct qcs));
+
+/* Emit a CONNECTION_CLOSE with error <err>. This will interrupt all future
+ * send/receive operations.
+ */
+static void qcc_emit_cc(struct qcc *qcc, int err)
+{
+ TRACE_ENTER(QMUX_EV_QCC_END, qcc->conn);
+
+ TRACE_STATE("set CONNECTION_CLOSE on quic-conn", QMUX_EV_QCC_WAKE, qcc->conn);
+ quic_set_connection_close(qcc->conn->handle.qc, quic_err_transport(err));
+ qcc->flags |= QC_CF_CC_EMIT;
+ tasklet_wakeup(qcc->wait_event.tasklet);
+
+ TRACE_LEAVE(QMUX_EV_QCC_END, qcc->conn);
+}
+
+static void qc_free_ncbuf(struct qcs *qcs, struct ncbuf *ncbuf)
+{
+ struct buffer buf;
+
+ if (ncb_is_null(ncbuf))
+ return;
+
+ buf = b_make(ncbuf->area, ncbuf->size, 0, 0);
+ b_free(&buf);
+ offer_buffers(NULL, 1);
+
+ *ncbuf = NCBUF_NULL;
+}
+
+/* Free <qcs> instance. This function is reserved for internal usage : it must
+ * only be called on qcs alloc error or on connection shutdown. Else
+ * qcs_destroy must be prefered to handle QUIC flow-control increase.
+ */
+static void qcs_free(struct qcs *qcs)
+{
+ struct qcc *qcc = qcs->qcc;
+
+ TRACE_ENTER(QMUX_EV_QCS_END, qcc->conn, qcs);
+
+ /* Safe to use even if already removed from the list. */
+ LIST_DEL_INIT(&qcs->el_opening);
+
+ /* Release stream endpoint descriptor. */
+ BUG_ON(qcs->sd && !se_fl_test(qcs->sd, SE_FL_ORPHAN));
+ sedesc_free(qcs->sd);
+
+ /* Release app-layer context. */
+ if (qcs->ctx && qcc->app_ops->detach)
+ qcc->app_ops->detach(qcs);
+
+ /* Release qc_stream_desc buffer from quic-conn layer. */
+ qc_stream_desc_release(qcs->stream);
+
+ /* Free Rx/Tx buffers. */
+ qc_free_ncbuf(qcs, &qcs->rx.ncbuf);
+ b_free(&qcs->tx.buf);
+
+ BUG_ON(!qcc->strms[qcs_id_type(qcs->id)].nb_streams);
+ --qcc->strms[qcs_id_type(qcs->id)].nb_streams;
+
+ /* Remove qcs from qcc tree. */
+ eb64_delete(&qcs->by_id);
+
+ pool_free(pool_head_qcs, qcs);
+
+ TRACE_LEAVE(QMUX_EV_QCS_END, qcc->conn);
+}
+
+/* Allocate a new QUIC streams with id <id> and type <type>. */
+static struct qcs *qcs_new(struct qcc *qcc, uint64_t id, enum qcs_type type)
+{
+ struct qcs *qcs;
+
+ TRACE_ENTER(QMUX_EV_QCS_NEW, qcc->conn);
+
+ qcs = pool_alloc(pool_head_qcs);
+ if (!qcs) {
+ TRACE_ERROR("alloc failure", QMUX_EV_QCS_NEW, qcc->conn);
+ return NULL;
+ }
+
+ qcs->stream = NULL;
+ qcs->qcc = qcc;
+ qcs->sd = NULL;
+ qcs->flags = QC_SF_NONE;
+ qcs->st = QC_SS_IDLE;
+ qcs->ctx = NULL;
+
+ /* App callback attach may register the stream for http-request wait.
+ * These fields must be initialed before.
+ */
+ LIST_INIT(&qcs->el_opening);
+ qcs->start = TICK_ETERNITY;
+
+ /* store transport layer stream descriptor in qcc tree */
+ qcs->id = qcs->by_id.key = id;
+ eb64_insert(&qcc->streams_by_id, &qcs->by_id);
+
+ qcc->strms[type].nb_streams++;
+
+ /* Allocate transport layer stream descriptor. Only needed for TX. */
+ if (!quic_stream_is_uni(id) || !quic_stream_is_remote(qcc, id)) {
+ struct quic_conn *qc = qcc->conn->handle.qc;
+ qcs->stream = qc_stream_desc_new(id, type, qcs, qc);
+ if (!qcs->stream) {
+ TRACE_ERROR("qc_stream_desc alloc failure", QMUX_EV_QCS_NEW, qcc->conn, qcs);
+ goto err;
+ }
+ }
+
+ if (qcc->app_ops->attach) {
+ if (qcc->app_ops->attach(qcs, qcc->ctx)) {
+ TRACE_ERROR("app proto failure", QMUX_EV_QCS_NEW, qcc->conn, qcs);
+ goto err;
+ }
+ }
+
+ /* If stream is local, use peer remote-limit, or else the opposite. */
+ if (quic_stream_is_bidi(id)) {
+ qcs->tx.msd = quic_stream_is_local(qcc, id) ? qcc->rfctl.msd_bidi_r :
+ qcc->rfctl.msd_bidi_l;
+ }
+ else if (quic_stream_is_local(qcc, id)) {
+ qcs->tx.msd = qcc->rfctl.msd_uni_l;
+ }
+
+ qcs->rx.ncbuf = NCBUF_NULL;
+ qcs->rx.app_buf = BUF_NULL;
+ qcs->rx.offset = qcs->rx.offset_max = 0;
+
+ if (quic_stream_is_bidi(id)) {
+ qcs->rx.msd = quic_stream_is_local(qcc, id) ? qcc->lfctl.msd_bidi_l :
+ qcc->lfctl.msd_bidi_r;
+ }
+ else if (quic_stream_is_remote(qcc, id)) {
+ qcs->rx.msd = qcc->lfctl.msd_uni_r;
+ }
+ qcs->rx.msd_init = qcs->rx.msd;
+
+ qcs->tx.buf = BUF_NULL;
+ qcs->tx.offset = 0;
+ qcs->tx.sent_offset = 0;
+
+ qcs->wait_event.tasklet = NULL;
+ qcs->wait_event.events = 0;
+ qcs->subs = NULL;
+
+ qcs->err = 0;
+
+ out:
+ TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn, qcs);
+ return qcs;
+
+ err:
+ qcs_free(qcs);
+ TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn);
+ return NULL;
+}
+
+static forceinline struct stconn *qcs_sc(const struct qcs *qcs)
+{
+ return qcs->sd ? qcs->sd->sc : NULL;
+}
+
+/* Reset the <qcc> inactivity timeout for http-keep-alive timeout. */
+static forceinline void qcc_reset_idle_start(struct qcc *qcc)
+{
+ qcc->idle_start = now_ms;
+}
+
+/* Decrement <qcc> sc. */
+static forceinline void qcc_rm_sc(struct qcc *qcc)
+{
+ BUG_ON_HOT(!qcc->nb_sc);
+ --qcc->nb_sc;
+
+ /* Reset qcc idle start for http-keep-alive timeout. Timeout will be
+ * refreshed after this on stream detach.
+ */
+ if (!qcc->nb_sc && !qcc->nb_hreq)
+ qcc_reset_idle_start(qcc);
+}
+
+/* Decrement <qcc> hreq. */
+static forceinline void qcc_rm_hreq(struct qcc *qcc)
+{
+ BUG_ON_HOT(!qcc->nb_hreq);
+ --qcc->nb_hreq;
+
+ /* Reset qcc idle start for http-keep-alive timeout. Timeout will be
+ * refreshed after this on I/O handler.
+ */
+ if (!qcc->nb_sc && !qcc->nb_hreq)
+ qcc_reset_idle_start(qcc);
+}
+
+static inline int qcc_is_dead(const struct qcc *qcc)
+{
+ /* Mux connection is considered dead if :
+ * - all stream-desc are detached AND
+ * = connection is on error OR
+ * = mux timeout has already fired or is unset
+ */
+ if (!qcc->nb_sc && ((qcc->conn->flags & CO_FL_ERROR) || !qcc->task))
+ return 1;
+
+ return 0;
+}
+
+/* Return true if the mux timeout should be armed. */
+static inline int qcc_may_expire(struct qcc *qcc)
+{
+ return !qcc->nb_sc;
+}
+
+/* Refresh the timeout on <qcc> if needed depending on its state. */
+static void qcc_refresh_timeout(struct qcc *qcc)
+{
+ const struct proxy *px = qcc->proxy;
+
+ TRACE_ENTER(QMUX_EV_QCC_WAKE, qcc->conn);
+
+ if (!qcc->task) {
+ TRACE_DEVEL("already expired", QMUX_EV_QCC_WAKE, qcc->conn);
+ goto leave;
+ }
+
+ /* Check if upper layer is responsible of timeout management. */
+ if (!qcc_may_expire(qcc)) {
+ TRACE_DEVEL("not eligible for timeout", QMUX_EV_QCC_WAKE, qcc->conn);
+ qcc->task->expire = TICK_ETERNITY;
+ task_queue(qcc->task);
+ goto leave;
+ }
+
+ /* TODO if connection is idle on frontend and proxy is disabled, remove
+ * it with global close_spread delay applied.
+ */
+
+ /* TODO implement client/server-fin timeout for graceful shutdown */
+
+ /* Frontend timeout management
+ * - detached streams with data left to send -> default timeout
+ * - stream waiting on incomplete request or no stream yet activated -> timeout http-request
+ * - idle after stream processing -> timeout http-keep-alive
+ */
+ if (!conn_is_back(qcc->conn)) {
+ if (qcc->nb_hreq) {
+ TRACE_DEVEL("one or more requests still in progress", QMUX_EV_QCC_WAKE, qcc->conn);
+ qcc->task->expire = tick_add_ifset(now_ms, qcc->timeout);
+ task_queue(qcc->task);
+ goto leave;
+ }
+
+ if (!LIST_ISEMPTY(&qcc->opening_list) || unlikely(!qcc->largest_bidi_r)) {
+ int timeout = px->timeout.httpreq;
+ struct qcs *qcs = NULL;
+ int base_time;
+
+ /* Use start time of first stream waiting on HTTP or
+ * qcc idle if no stream not yet used.
+ */
+ if (likely(!LIST_ISEMPTY(&qcc->opening_list)))
+ qcs = LIST_ELEM(qcc->opening_list.n, struct qcs *, el_opening);
+ base_time = qcs ? qcs->start : qcc->idle_start;
+
+ TRACE_DEVEL("waiting on http request", QMUX_EV_QCC_WAKE, qcc->conn, qcs);
+ qcc->task->expire = tick_add_ifset(base_time, timeout);
+ }
+ else {
+ /* Use http-request timeout if keep-alive timeout not set */
+ int timeout = tick_isset(px->timeout.httpka) ?
+ px->timeout.httpka : px->timeout.httpreq;
+
+ TRACE_DEVEL("at least one request achieved but none currently in progress", QMUX_EV_QCC_WAKE, qcc->conn);
+ qcc->task->expire = tick_add_ifset(qcc->idle_start, timeout);
+ }
+ }
+
+ /* fallback to default timeout if frontend specific undefined or for
+ * backend connections.
+ */
+ if (!tick_isset(qcc->task->expire)) {
+ TRACE_DEVEL("fallback to default timeout", QMUX_EV_QCC_WAKE, qcc->conn);
+ qcc->task->expire = tick_add_ifset(now_ms, qcc->timeout);
+ }
+
+ task_queue(qcc->task);
+
+ leave:
+ TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn);
+}
+
+/* Mark a stream as open if it was idle. This can be used on every
+ * successful emission/reception operation to update the stream state.
+ */
+static void qcs_idle_open(struct qcs *qcs)
+{
+ /* This operation must not be used if the stream is already closed. */
+ BUG_ON_HOT(qcs->st == QC_SS_CLO);
+
+ if (qcs->st == QC_SS_IDLE) {
+ TRACE_STATE("opening stream", QMUX_EV_QCS_NEW, qcs->qcc->conn, qcs);
+ qcs->st = QC_SS_OPEN;
+ }
+}
+
+/* Close the local channel of <qcs> instance. */
+static void qcs_close_local(struct qcs *qcs)
+{
+ TRACE_STATE("closing stream locally", QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
+
+ /* The stream must have already been opened. */
+ BUG_ON_HOT(qcs->st == QC_SS_IDLE);
+
+ /* This operation cannot be used multiple times. */
+ BUG_ON_HOT(qcs->st == QC_SS_HLOC || qcs->st == QC_SS_CLO);
+
+ if (quic_stream_is_bidi(qcs->id)) {
+ qcs->st = (qcs->st == QC_SS_HREM) ? QC_SS_CLO : QC_SS_HLOC;
+
+ if (qcs->flags & QC_SF_HREQ_RECV)
+ qcc_rm_hreq(qcs->qcc);
+ }
+ else {
+ /* Only local uni streams are valid for this operation. */
+ BUG_ON_HOT(quic_stream_is_remote(qcs->qcc, qcs->id));
+ qcs->st = QC_SS_CLO;
+ }
+}
+
+/* Close the remote channel of <qcs> instance. */
+static void qcs_close_remote(struct qcs *qcs)
+{
+ TRACE_STATE("closing stream remotely", QMUX_EV_QCS_RECV, qcs->qcc->conn, qcs);
+
+ /* The stream must have already been opened. */
+ BUG_ON_HOT(qcs->st == QC_SS_IDLE);
+
+ /* This operation cannot be used multiple times. */
+ BUG_ON_HOT(qcs->st == QC_SS_HREM || qcs->st == QC_SS_CLO);
+
+ if (quic_stream_is_bidi(qcs->id)) {
+ qcs->st = (qcs->st == QC_SS_HLOC) ? QC_SS_CLO : QC_SS_HREM;
+ }
+ else {
+ /* Only remote uni streams are valid for this operation. */
+ BUG_ON_HOT(quic_stream_is_local(qcs->qcc, qcs->id));
+ qcs->st = QC_SS_CLO;
+ }
+}
+
+static int qcs_is_close_local(struct qcs *qcs)
+{
+ return qcs->st == QC_SS_HLOC || qcs->st == QC_SS_CLO;
+}
+
+static __maybe_unused int qcs_is_close_remote(struct qcs *qcs)
+{
+ return qcs->st == QC_SS_HREM || qcs->st == QC_SS_CLO;
+}
+
+struct buffer *qc_get_buf(struct qcs *qcs, struct buffer *bptr)
+{
+ struct buffer *buf = b_alloc(bptr);
+ BUG_ON(!buf);
+ return buf;
+}
+
+static struct ncbuf *qc_get_ncbuf(struct qcs *qcs, struct ncbuf *ncbuf)
+{
+ struct buffer buf = BUF_NULL;
+
+ if (ncb_is_null(ncbuf)) {
+ b_alloc(&buf);
+ BUG_ON(b_is_null(&buf));
+
+ *ncbuf = ncb_make(buf.area, buf.size, 0);
+ ncb_init(ncbuf, 0);
+ }
+
+ return ncbuf;
+}
+
+/* Notify an eventual subscriber on <qcs> or else wakeup up the stconn layer if
+ * initialized.
+ */
+static void qcs_alert(struct qcs *qcs)
+{
+ if (qcs->subs) {
+ qcs_notify_recv(qcs);
+ qcs_notify_send(qcs);
+ }
+ else if (qcs_sc(qcs) && qcs->sd->sc->app_ops->wake) {
+ qcs->sd->sc->app_ops->wake(qcs->sd->sc);
+ }
+}
+
+int qcs_subscribe(struct qcs *qcs, int event_type, struct wait_event *es)
+{
+ struct qcc *qcc = qcs->qcc;
+
+ TRACE_ENTER(QMUX_EV_STRM_SEND|QMUX_EV_STRM_RECV, qcc->conn, qcs);
+
+ BUG_ON(event_type & ~(SUB_RETRY_SEND|SUB_RETRY_RECV));
+ BUG_ON(qcs->subs && qcs->subs != es);
+
+ es->events |= event_type;
+ qcs->subs = es;
+
+ if (event_type & SUB_RETRY_RECV)
+ TRACE_DEVEL("subscribe(recv)", QMUX_EV_STRM_RECV, qcc->conn, qcs);
+
+ if (event_type & SUB_RETRY_SEND)
+ TRACE_DEVEL("subscribe(send)", QMUX_EV_STRM_SEND, qcc->conn, qcs);
+
+ TRACE_LEAVE(QMUX_EV_STRM_SEND|QMUX_EV_STRM_RECV, qcc->conn, qcs);
+
+ return 0;
+}
+
+void qcs_notify_recv(struct qcs *qcs)
+{
+ if (qcs->subs && qcs->subs->events & SUB_RETRY_RECV) {
+ tasklet_wakeup(qcs->subs->tasklet);
+ qcs->subs->events &= ~SUB_RETRY_RECV;
+ if (!qcs->subs->events)
+ qcs->subs = NULL;
+ }
+}
+
+void qcs_notify_send(struct qcs *qcs)
+{
+ if (qcs->subs && qcs->subs->events & SUB_RETRY_SEND) {
+ tasklet_wakeup(qcs->subs->tasklet);
+ qcs->subs->events &= ~SUB_RETRY_SEND;
+ if (!qcs->subs->events)
+ qcs->subs = NULL;
+ }
+}
+
+/* Open a locally initiated stream for the connection <qcc>. Set <bidi> for a
+ * bidirectional stream, else an unidirectional stream is opened. The next
+ * available ID on the connection will be used according to the stream type.
+ *
+ * Returns the allocated stream instance or NULL on error.
+ */
+struct qcs *qcc_init_stream_local(struct qcc *qcc, int bidi)
+{
+ struct qcs *qcs;
+ enum qcs_type type;
+ uint64_t *next;
+
+ TRACE_ENTER(QMUX_EV_QCS_NEW, qcc->conn);
+
+ if (bidi) {
+ next = &qcc->next_bidi_l;
+ type = conn_is_back(qcc->conn) ? QCS_CLT_BIDI : QCS_SRV_BIDI;
+ }
+ else {
+ next = &qcc->next_uni_l;
+ type = conn_is_back(qcc->conn) ? QCS_CLT_UNI : QCS_SRV_UNI;
+ }
+
+ /* TODO ensure that we won't overflow remote peer flow control limit on
+ * streams. Else, we should emit a STREAMS_BLOCKED frame.
+ */
+
+ qcs = qcs_new(qcc, *next, type);
+ if (!qcs) {
+ TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn);
+ return NULL;
+ }
+
+ TRACE_PROTO("opening local stream", QMUX_EV_QCS_NEW, qcc->conn, qcs);
+ *next += 4;
+
+ TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn, qcs);
+ return qcs;
+}
+
+/* Open a remote initiated stream for the connection <qcc> with ID <id>. The
+ * caller is responsible to ensure that a stream with the same ID was not
+ * already opened. This function will also create all intermediaries streams
+ * with ID smaller than <id> not already opened before.
+ *
+ * Returns the allocated stream instance or NULL on error.
+ */
+static struct qcs *qcc_init_stream_remote(struct qcc *qcc, uint64_t id)
+{
+ struct qcs *qcs = NULL;
+ enum qcs_type type;
+ uint64_t *largest, max_id;
+
+ TRACE_ENTER(QMUX_EV_QCS_NEW, qcc->conn);
+
+ BUG_ON_HOT(quic_stream_is_local(qcc, id));
+
+ if (quic_stream_is_bidi(id)) {
+ largest = &qcc->largest_bidi_r;
+ type = conn_is_back(qcc->conn) ? QCS_SRV_BIDI : QCS_CLT_BIDI;
+ }
+ else {
+ largest = &qcc->largest_uni_r;
+ type = conn_is_back(qcc->conn) ? QCS_SRV_UNI : QCS_CLT_UNI;
+ }
+
+ /* RFC 9000 4.6. Controlling Concurrency
+ *
+ * An endpoint that receives a frame with a stream ID exceeding the
+ * limit it has sent MUST treat this as a connection error of type
+ * STREAM_LIMIT_ERROR
+ */
+ max_id = quic_stream_is_bidi(id) ? qcc->lfctl.ms_bidi * 4 :
+ qcc->lfctl.ms_uni * 4;
+ if (id >= max_id) {
+ TRACE_ERROR("flow control error", QMUX_EV_QCS_NEW|QMUX_EV_PROTO_ERR, qcc->conn);
+ qcc_emit_cc(qcc, QC_ERR_STREAM_LIMIT_ERROR);
+ goto err;
+ }
+
+ /* Only stream ID not already opened can be used. */
+ BUG_ON(id < *largest);
+
+ while (id >= *largest) {
+ const char *str = *largest < id ? "initializing intermediary remote stream" :
+ "initializing remote stream";
+
+ qcs = qcs_new(qcc, *largest, type);
+ if (!qcs) {
+ /* TODO emit RESET_STREAM */
+ TRACE_ERROR("stream fallocation failure", QMUX_EV_QCS_NEW, qcc->conn);
+ goto err;
+ }
+
+ TRACE_PROTO(str, QMUX_EV_QCS_NEW, qcc->conn, qcs);
+ *largest += 4;
+ }
+
+ out:
+ TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn, qcs);
+ return qcs;
+
+ err:
+ TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn);
+ return NULL;
+}
+
+/* Use this function for a stream <id> which is not in <qcc> stream tree. It
+ * returns true if the associated stream is closed.
+ */
+static int qcc_stream_id_is_closed(struct qcc *qcc, uint64_t id)
+{
+ uint64_t *largest;
+
+ /* This function must only be used for stream not present in the stream tree. */
+ BUG_ON_HOT(eb64_lookup(&qcc->streams_by_id, id));
+
+ if (quic_stream_is_local(qcc, id)) {
+ largest = quic_stream_is_uni(id) ? &qcc->next_uni_l :
+ &qcc->next_bidi_l;
+ }
+ else {
+ largest = quic_stream_is_uni(id) ? &qcc->largest_uni_r :
+ &qcc->largest_bidi_r;
+ }
+
+ return id < *largest;
+}
+
+/* Retrieve the stream instance from <id> ID. This can be used when receiving
+ * STREAM, STREAM_DATA_BLOCKED, RESET_STREAM, MAX_STREAM_DATA or STOP_SENDING
+ * frames. Set to false <receive_only> or <send_only> if these particular types
+ * of streams are not allowed. If the stream instance is found, it is stored in
+ * <out>.
+ *
+ * Returns 0 on success else non-zero. On error, a RESET_STREAM or a
+ * CONNECTION_CLOSE is automatically emitted. Beware that <out> may be NULL
+ * on success if the stream has already been closed.
+ */
+int qcc_get_qcs(struct qcc *qcc, uint64_t id, int receive_only, int send_only,
+ struct qcs **out)
+{
+ struct eb64_node *node;
+
+ TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
+ *out = NULL;
+
+ if (!receive_only && quic_stream_is_uni(id) && quic_stream_is_remote(qcc, id)) {
+ TRACE_ERROR("receive-only stream not allowed", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS|QMUX_EV_PROTO_ERR, qcc->conn, NULL, &id);
+ qcc_emit_cc(qcc, QC_ERR_STREAM_STATE_ERROR);
+ goto err;
+ }
+
+ if (!send_only && quic_stream_is_uni(id) && quic_stream_is_local(qcc, id)) {
+ TRACE_ERROR("send-only stream not allowed", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS|QMUX_EV_PROTO_ERR, qcc->conn, NULL, &id);
+ qcc_emit_cc(qcc, QC_ERR_STREAM_STATE_ERROR);
+ goto err;
+ }
+
+ /* Search the stream in the connection tree. */
+ node = eb64_lookup(&qcc->streams_by_id, id);
+ if (node) {
+ *out = eb64_entry(node, struct qcs, by_id);
+ TRACE_DEVEL("using stream from connection tree", QMUX_EV_QCC_RECV, qcc->conn, *out);
+ goto out;
+ }
+
+ /* Check if stream is already closed. */
+ if (qcc_stream_id_is_closed(qcc, id)) {
+ TRACE_DATA("already closed stream", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS, qcc->conn, NULL, &id);
+ /* Consider this as a success even if <out> is left NULL. */
+ goto out;
+ }
+
+ /* Create the stream. This is valid only for remote initiated one. A
+ * local stream must have already been explicitely created by the
+ * application protocol layer.
+ */
+ if (quic_stream_is_local(qcc, id)) {
+ /* RFC 9000 19.8. STREAM Frames
+ *
+ * An endpoint MUST terminate the connection with error
+ * STREAM_STATE_ERROR if it receives a STREAM frame for a locally
+ * initiated stream that has not yet been created, or for a send-only
+ * stream.
+ */
+ TRACE_ERROR("locally initiated stream not yet created", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS|QMUX_EV_PROTO_ERR, qcc->conn, NULL, &id);
+ qcc_emit_cc(qcc, QC_ERR_STREAM_STATE_ERROR);
+ goto err;
+ }
+ else {
+ /* Remote stream not found - try to open it. */
+ *out = qcc_init_stream_remote(qcc, id);
+ if (!*out) {
+ TRACE_ERROR("stream creation error", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS, qcc->conn, NULL, &id);
+ goto err;
+ }
+ }
+
+ out:
+ TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn, *out);
+ return 0;
+
+ err:
+ TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
+ return 1;
+}
+
+/* Simple function to duplicate a buffer */
+static inline struct buffer qcs_b_dup(const struct ncbuf *b)
+{
+ return b_make(ncb_orig(b), b->size, b->head, ncb_data(b, 0));
+}
+
+/* Remove <bytes> from <qcs> Rx buffer. Flow-control for received offsets may
+ * be allocated for the peer if needed.
+ */
+static void qcs_consume(struct qcs *qcs, uint64_t bytes)
+{
+ struct qcc *qcc = qcs->qcc;
+ struct quic_frame *frm;
+ struct ncbuf *buf = &qcs->rx.ncbuf;
+ enum ncb_ret ret;
+
+ TRACE_ENTER(QMUX_EV_QCS_RECV, qcc->conn, qcs);
+
+ ret = ncb_advance(buf, bytes);
+ if (ret) {
+ ABORT_NOW(); /* should not happens because removal only in data */
+ }
+
+ if (ncb_is_empty(buf))
+ qc_free_ncbuf(qcs, buf);
+
+ qcs->rx.offset += bytes;
+ if (qcs->rx.msd - qcs->rx.offset < qcs->rx.msd_init / 2) {
+ TRACE_DATA("increase stream credit via MAX_STREAM_DATA", QMUX_EV_QCS_RECV, qcc->conn, qcs);
+ frm = pool_zalloc(pool_head_quic_frame);
+ BUG_ON(!frm); /* TODO handle this properly */
+
+ qcs->rx.msd = qcs->rx.offset + qcs->rx.msd_init;
+
+ LIST_INIT(&frm->reflist);
+ frm->type = QUIC_FT_MAX_STREAM_DATA;
+ frm->max_stream_data.id = qcs->id;
+ frm->max_stream_data.max_stream_data = qcs->rx.msd;
+
+ LIST_APPEND(&qcc->lfctl.frms, &frm->list);
+ tasklet_wakeup(qcc->wait_event.tasklet);
+ }
+
+ qcc->lfctl.offsets_consume += bytes;
+ if (qcc->lfctl.md - qcc->lfctl.offsets_consume < qcc->lfctl.md_init / 2) {
+ TRACE_DATA("increase conn credit via MAX_DATA", QMUX_EV_QCS_RECV, qcc->conn, qcs);
+ frm = pool_zalloc(pool_head_quic_frame);
+ BUG_ON(!frm); /* TODO handle this properly */
+
+ qcc->lfctl.md = qcc->lfctl.offsets_consume + qcc->lfctl.md_init;
+
+ LIST_INIT(&frm->reflist);
+ frm->type = QUIC_FT_MAX_DATA;
+ frm->max_data.max_data = qcc->lfctl.md;
+
+ LIST_APPEND(&qcs->qcc->lfctl.frms, &frm->list);
+ tasklet_wakeup(qcs->qcc->wait_event.tasklet);
+ }
+
+ TRACE_LEAVE(QMUX_EV_QCS_RECV, qcc->conn, qcs);
+}
+
+/* Decode the content of STREAM frames already received on the stream instance
+ * <qcs>.
+ *
+ * Returns 0 on success else non-zero.
+ */
+static int qcc_decode_qcs(struct qcc *qcc, struct qcs *qcs)
+{
+ struct buffer b;
+ ssize_t ret;
+ int fin = 0;
+
+ TRACE_ENTER(QMUX_EV_QCS_RECV, qcc->conn, qcs);
+
+ b = qcs_b_dup(&qcs->rx.ncbuf);
+
+ /* Signal FIN to application if STREAM FIN received with all data. */
+ if (qcs_is_close_remote(qcs))
+ fin = 1;
+
+ ret = qcc->app_ops->decode_qcs(qcs, &b, fin);
+ if (ret < 0) {
+ TRACE_ERROR("decoding error", QMUX_EV_QCS_RECV, qcc->conn, qcs);
+ goto err;
+ }
+
+ if (ret)
+ qcs_consume(qcs, ret);
+ if (ret || (!b_data(&b) && fin))
+ qcs_notify_recv(qcs);
+
+ TRACE_LEAVE(QMUX_EV_QCS_RECV, qcc->conn, qcs);
+ return 0;
+
+ err:
+ TRACE_LEAVE(QMUX_EV_QCS_RECV, qcc->conn, qcs);
+ return 1;
+}
+
+/* Emit a CONNECTION_CLOSE_APP with error <err>. Reserved for application error
+ * code. To close the connection right away, set <immediate> : this is useful
+ * when dealing with a connection fatal error. Else a graceful shutdown will be
+ * conducted : the error-code is only registered. The lower layer is
+ * responsible to close the connection when deemed suitable. Note that in this
+ * case the error code might be overwritten if an immediate close is requested
+ * in the interval.
+ */
+void qcc_emit_cc_app(struct qcc *qcc, int err, int immediate)
+{
+ TRACE_ENTER(QMUX_EV_QCC_END, qcc->conn);
+
+ if (immediate) {
+ quic_set_connection_close(qcc->conn->handle.qc, quic_err_app(err));
+ qcc->flags |= QC_CF_CC_EMIT;
+ tasklet_wakeup(qcc->wait_event.tasklet);
+ }
+ else {
+ /* Only register the error code for graceful shutdown. */
+ qcc->conn->handle.qc->err = quic_err_app(err);
+ }
+
+ TRACE_LEAVE(QMUX_EV_QCC_END, qcc->conn);
+}
+
+/* Prepare for the emission of RESET_STREAM on <qcs> with error code <err>. */
+void qcc_reset_stream(struct qcs *qcs, int err)
+{
+ struct qcc *qcc = qcs->qcc;
+
+ if ((qcs->flags & QC_SF_TO_RESET) || qcs_is_close_local(qcs))
+ return;
+
+ TRACE_STATE("reset stream", QMUX_EV_QCS_END, qcc->conn, qcs);
+ qcs->flags |= QC_SF_TO_RESET;
+ qcs->err = err;
+
+ /* Remove prepared stream data from connection flow-control calcul. */
+ if (qcs->tx.offset > qcs->tx.sent_offset) {
+ const uint64_t diff = qcs->tx.offset - qcs->tx.sent_offset;
+ BUG_ON(qcc->tx.offsets - diff < qcc->tx.sent_offsets);
+ qcc->tx.offsets -= diff;
+ /* Reset qcs offset to prevent BUG_ON() on qcs_destroy(). */
+ qcs->tx.offset = qcs->tx.sent_offset;
+ }
+
+ tasklet_wakeup(qcc->wait_event.tasklet);
+}
+
+/* Install the <app_ops> applicative layer of a QUIC connection on mux <qcc>.
+ * Returns 0 on success else non-zero.
+ */
+int qcc_install_app_ops(struct qcc *qcc, const struct qcc_app_ops *app_ops)
+{
+ TRACE_ENTER(QMUX_EV_QCC_NEW, qcc->conn);
+
+ qcc->app_ops = app_ops;
+ if (qcc->app_ops->init && !qcc->app_ops->init(qcc)) {
+ TRACE_ERROR("app ops init error", QMUX_EV_QCC_NEW, qcc->conn);
+ goto err;
+ }
+
+ TRACE_PROTO("application layer initialized", QMUX_EV_QCC_NEW, qcc->conn);
+
+ TRACE_LEAVE(QMUX_EV_QCC_NEW, qcc->conn);
+ return 0;
+
+ err:
+ TRACE_LEAVE(QMUX_EV_QCC_NEW, qcc->conn);
+ return 1;
+}
+
+/* Handle a new STREAM frame for stream with id <id>. Payload is pointed by
+ * <data> with length <len> and represents the offset <offset>. <fin> is set if
+ * the QUIC frame FIN bit is set.
+ *
+ * Returns 0 on success else non-zero. On error, the received frame should not
+ * be acknowledged.
+ */
+int qcc_recv(struct qcc *qcc, uint64_t id, uint64_t len, uint64_t offset,
+ char fin, char *data)
+{
+ struct qcs *qcs;
+ enum ncb_ret ret;
+
+ TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
+
+ if (qcc->flags & QC_CF_CC_EMIT) {
+ TRACE_DATA("connection closed", QMUX_EV_QCC_RECV, qcc->conn);
+ goto err;
+ }
+
+ /* RFC 9000 19.8. STREAM Frames
+ *
+ * An endpoint MUST terminate the connection with error
+ * STREAM_STATE_ERROR if it receives a STREAM frame for a locally
+ * initiated stream that has not yet been created, or for a send-only
+ * stream.
+ */
+ if (qcc_get_qcs(qcc, id, 1, 0, &qcs)) {
+ TRACE_DATA("qcs retrieval error", QMUX_EV_QCC_RECV, qcc->conn);
+ goto err;
+ }
+
+ if (!qcs) {
+ TRACE_DATA("already closed stream", QMUX_EV_QCC_RECV, qcc->conn);
+ goto out;
+ }
+
+ /* RFC 9000 4.5. Stream Final Size
+ *
+ * Once a final size for a stream is known, it cannot change. If a
+ * RESET_STREAM or STREAM frame is received indicating a change in the
+ * final size for the stream, an endpoint SHOULD respond with an error
+ * of type FINAL_SIZE_ERROR; see Section 11 for details on error
+ * handling.
+ */
+ if (qcs->flags & QC_SF_SIZE_KNOWN &&
+ (offset + len > qcs->rx.offset_max || (fin && offset + len < qcs->rx.offset_max))) {
+ TRACE_ERROR("final size error", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV|QMUX_EV_PROTO_ERR, qcc->conn, qcs);
+ qcc_emit_cc(qcc, QC_ERR_FINAL_SIZE_ERROR);
+ goto err;
+ }
+
+ if (offset + len < qcs->rx.offset ||
+ (offset + len == qcs->rx.offset && (!fin || (qcs->flags & QC_SF_SIZE_KNOWN)))) {
+ TRACE_DATA("already received offset", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
+ goto out;
+ }
+
+ TRACE_PROTO("receiving STREAM", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
+ qcs_idle_open(qcs);
+
+ if (offset + len > qcs->rx.offset_max) {
+ uint64_t diff = offset + len - qcs->rx.offset_max;
+ qcs->rx.offset_max = offset + len;
+ qcc->lfctl.offsets_recv += diff;
+
+ if (offset + len > qcs->rx.msd ||
+ qcc->lfctl.offsets_recv > qcc->lfctl.md) {
+ /* RFC 9000 4.1. Data Flow Control
+ *
+ * A receiver MUST close the connection with an error
+ * of type FLOW_CONTROL_ERROR if the sender violates
+ * the advertised connection or stream data limits
+ */
+ TRACE_ERROR("flow control error", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV|QMUX_EV_PROTO_ERR,
+ qcc->conn, qcs);
+ qcc_emit_cc(qcc, QC_ERR_FLOW_CONTROL_ERROR);
+ goto err;
+ }
+ }
+
+ if (!qc_get_ncbuf(qcs, &qcs->rx.ncbuf) || ncb_is_null(&qcs->rx.ncbuf)) {
+ /* TODO should mark qcs as full */
+ ABORT_NOW();
+ return 1;
+ }
+
+ TRACE_DATA("newly received offset", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
+ if (offset < qcs->rx.offset) {
+ size_t diff = qcs->rx.offset - offset;
+
+ len -= diff;
+ data += diff;
+ offset = qcs->rx.offset;
+ }
+
+ if (len) {
+ ret = ncb_add(&qcs->rx.ncbuf, offset - qcs->rx.offset, data, len, NCB_ADD_COMPARE);
+ switch (ret) {
+ case NCB_RET_OK:
+ break;
+
+ case NCB_RET_DATA_REJ:
+ /* RFC 9000 2.2. Sending and Receiving Data
+ *
+ * An endpoint could receive data for a stream at the
+ * same stream offset multiple times. Data that has
+ * already been received can be discarded. The data at
+ * a given offset MUST NOT change if it is sent
+ * multiple times; an endpoint MAY treat receipt of
+ * different data at the same offset within a stream as
+ * a connection error of type PROTOCOL_VIOLATION.
+ */
+ TRACE_ERROR("overlapping data rejected", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV|QMUX_EV_PROTO_ERR,
+ qcc->conn, qcs);
+ qcc_emit_cc(qcc, QC_ERR_PROTOCOL_VIOLATION);
+ return 1;
+
+ case NCB_RET_GAP_SIZE:
+ TRACE_DATA("cannot bufferize frame due to gap size limit", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV,
+ qcc->conn, qcs);
+ return 1;
+ }
+ }
+
+ if (fin)
+ qcs->flags |= QC_SF_SIZE_KNOWN;
+
+ if (qcs->flags & QC_SF_SIZE_KNOWN &&
+ qcs->rx.offset_max == qcs->rx.offset + ncb_data(&qcs->rx.ncbuf, 0)) {
+ qcs_close_remote(qcs);
+ }
+
+ if ((ncb_data(&qcs->rx.ncbuf, 0) && !(qcs->flags & QC_SF_DEM_FULL)) || fin) {
+ qcc_decode_qcs(qcc, qcs);
+ qcc_refresh_timeout(qcc);
+ }
+
+ if (qcs->flags & QC_SF_READ_ABORTED) {
+ /* TODO should send a STOP_SENDING */
+ qcs_free(qcs);
+ }
+
+ out:
+ TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
+ return 0;
+
+ err:
+ TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
+ return 1;
+}
+
+/* Handle a new MAX_DATA frame. <max> must contains the maximum data field of
+ * the frame.
+ *
+ * Returns 0 on success else non-zero.
+ */
+int qcc_recv_max_data(struct qcc *qcc, uint64_t max)
+{
+ TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
+
+ TRACE_PROTO("receiving MAX_DATA", QMUX_EV_QCC_RECV, qcc->conn);
+ if (qcc->rfctl.md < max) {
+ qcc->rfctl.md = max;
+ TRACE_DEVEL("increase remote max-data", QMUX_EV_QCC_RECV, qcc->conn);
+
+ if (qcc->flags & QC_CF_BLK_MFCTL) {
+ qcc->flags &= ~QC_CF_BLK_MFCTL;
+ tasklet_wakeup(qcc->wait_event.tasklet);
+ }
+ }
+
+ TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
+ return 0;
+}
+
+/* Handle a new MAX_STREAM_DATA frame. <max> must contains the maximum data
+ * field of the frame and <id> is the identifier of the QUIC stream.
+ *
+ * Returns 0 on success else non-zero. On error, the received frame should not
+ * be acknowledged.
+ */
+int qcc_recv_max_stream_data(struct qcc *qcc, uint64_t id, uint64_t max)
+{
+ struct qcs *qcs;
+
+ TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
+
+ /* RFC 9000 19.10. MAX_STREAM_DATA Frames
+ *
+ * Receiving a MAX_STREAM_DATA frame for a locally
+ * initiated stream that has not yet been created MUST be treated as a
+ * connection error of type STREAM_STATE_ERROR. An endpoint that
+ * receives a MAX_STREAM_DATA frame for a receive-only stream MUST
+ * terminate the connection with error STREAM_STATE_ERROR.
+ */
+ if (qcc_get_qcs(qcc, id, 0, 1, &qcs)) {
+ TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
+ return 1;
+ }
+
+ if (qcs) {
+ TRACE_PROTO("receiving MAX_STREAM_DATA", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
+ if (max > qcs->tx.msd) {
+ qcs->tx.msd = max;
+ TRACE_DEVEL("increase remote max-stream-data", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
+
+ if (qcs->flags & QC_SF_BLK_SFCTL) {
+ qcs->flags &= ~QC_SF_BLK_SFCTL;
+ tasklet_wakeup(qcc->wait_event.tasklet);
+ }
+ }
+ }
+
+ if (qcc_may_expire(qcc) && !qcc->nb_hreq)
+ qcc_refresh_timeout(qcc);
+
+ TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
+ return 0;
+}
+
+/* Handle a new STOP_SENDING frame for stream ID <id>. The error code should be
+ * specified in <err>.
+ *
+ * Returns 0 on success else non-zero. On error, the received frame should not
+ * be acknowledged.
+ */
+int qcc_recv_stop_sending(struct qcc *qcc, uint64_t id, uint64_t err)
+{
+ struct qcs *qcs;
+
+ TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
+
+ /* RFC 9000 19.5. STOP_SENDING Frames
+ *
+ * Receiving a STOP_SENDING frame for a
+ * locally initiated stream that has not yet been created MUST be
+ * treated as a connection error of type STREAM_STATE_ERROR. An
+ * endpoint that receives a STOP_SENDING frame for a receive-only stream
+ * MUST terminate the connection with error STREAM_STATE_ERROR.
+ */
+ if (qcc_get_qcs(qcc, id, 0, 1, &qcs)) {
+ TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
+ return 1;
+ }
+
+ if (!qcs)
+ goto out;
+
+ TRACE_PROTO("receiving STOP_SENDING", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
+
+ /* RFC 9000 3.5. Solicited State Transitions
+ *
+ * An endpoint is expected to send another STOP_SENDING frame if a
+ * packet containing a previous STOP_SENDING is lost. However, once
+ * either all stream data or a RESET_STREAM frame has been received for
+ * the stream -- that is, the stream is in any state other than "Recv"
+ * or "Size Known" -- sending a STOP_SENDING frame is unnecessary.
+ */
+
+ /* TODO thanks to previous RFC clause, STOP_SENDING is ignored if current stream
+ * has already been closed locally. This is useful to not emit multiple
+ * RESET_STREAM for a single stream. This is functional if stream is
+ * locally closed due to all data transmitted, but in this case the RFC
+ * advices to use an explicit RESET_STREAM.
+ */
+ if (qcs_is_close_local(qcs)) {
+ TRACE_STATE("ignoring STOP_SENDING", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
+ goto out;
+ }
+
+ qcs_idle_open(qcs);
+
+ if (qcc->app_ops->close) {
+ if (qcc->app_ops->close(qcs, QCC_APP_OPS_CLOSE_SIDE_WR)) {
+ TRACE_ERROR("closure rejected by app layer", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
+ goto out;
+ }
+ }
+
+ /* RFC 9000 3.5. Solicited State Transitions
+ *
+ * An endpoint that receives a STOP_SENDING frame
+ * MUST send a RESET_STREAM frame if the stream is in the "Ready" or
+ * "Send" state. If the stream is in the "Data Sent" state, the
+ * endpoint MAY defer sending the RESET_STREAM frame until the packets
+ * containing outstanding data are acknowledged or declared lost. If
+ * any outstanding data is declared lost, the endpoint SHOULD send a
+ * RESET_STREAM frame instead of retransmitting the data.
+ *
+ * An endpoint SHOULD copy the error code from the STOP_SENDING frame to
+ * the RESET_STREAM frame it sends, but it can use any application error
+ * code.
+ */
+ qcc_reset_stream(qcs, err);
+
+ if (qcc_may_expire(qcc) && !qcc->nb_hreq)
+ qcc_refresh_timeout(qcc);
+
+ out:
+ TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
+ return 0;
+}
+
+/* Signal the closing of remote stream with id <id>. Flow-control for new
+ * streams may be allocated for the peer if needed.
+ */
+static int qcc_release_remote_stream(struct qcc *qcc, uint64_t id)
+{
+ struct quic_frame *frm;
+
+ TRACE_ENTER(QMUX_EV_QCS_END, qcc->conn);
+
+ if (quic_stream_is_bidi(id)) {
+ ++qcc->lfctl.cl_bidi_r;
+ if (qcc->lfctl.cl_bidi_r > qcc->lfctl.ms_bidi_init / 2) {
+ TRACE_DATA("increase max stream limit with MAX_STREAMS_BIDI", QMUX_EV_QCC_SEND, qcc->conn);
+ frm = pool_zalloc(pool_head_quic_frame);
+ BUG_ON(!frm); /* TODO handle this properly */
+
+ LIST_INIT(&frm->reflist);
+ frm->type = QUIC_FT_MAX_STREAMS_BIDI;
+ frm->max_streams_bidi.max_streams = qcc->lfctl.ms_bidi +
+ qcc->lfctl.cl_bidi_r;
+ LIST_APPEND(&qcc->lfctl.frms, &frm->list);
+ tasklet_wakeup(qcc->wait_event.tasklet);
+
+ qcc->lfctl.ms_bidi += qcc->lfctl.cl_bidi_r;
+ qcc->lfctl.cl_bidi_r = 0;
+ }
+ }
+ else {
+ /* TODO unidirectional stream flow control with MAX_STREAMS_UNI
+ * emission not implemented. It should be unnecessary for
+ * HTTP/3 but may be required if other application protocols
+ * are supported.
+ */
+ }
+
+ TRACE_LEAVE(QMUX_EV_QCS_END, qcc->conn);
+
+ return 0;
+}
+
+/* detaches the QUIC stream from its QCC and releases it to the QCS pool. */
+static void qcs_destroy(struct qcs *qcs)
+{
+ struct connection *conn = qcs->qcc->conn;
+ const uint64_t id = qcs->id;
+
+ TRACE_ENTER(QMUX_EV_QCS_END, conn, qcs);
+
+ /* MUST not removed a stream with sending prepared data left. This is
+ * to ensure consistency on connection flow-control calculation.
+ */
+ BUG_ON(qcs->tx.offset < qcs->tx.sent_offset);
+
+ if (quic_stream_is_remote(qcs->qcc, id))
+ qcc_release_remote_stream(qcs->qcc, id);
+
+ qcs_free(qcs);
+
+ TRACE_LEAVE(QMUX_EV_QCS_END, conn);
+}
+
+/* Transfer as much as possible data on <qcs> from <in> to <out>. This is done
+ * in respect with available flow-control at stream and connection level.
+ *
+ * Returns the total bytes of transferred data.
+ */
+static int qcs_xfer_data(struct qcs *qcs, struct buffer *out, struct buffer *in)
+{
+ struct qcc *qcc = qcs->qcc;
+ int left, to_xfer;
+ int total = 0;
+
+ TRACE_ENTER(QMUX_EV_QCS_SEND, qcc->conn, qcs);
+
+ qc_get_buf(qcs, out);
+
+ /*
+ * QCS out buffer diagram
+ * head left to_xfer
+ * -------------> ----------> ----->
+ * --------------------------------------------------
+ * |...............|xxxxxxxxxxx|<<<<<
+ * --------------------------------------------------
+ * ^ ack-off ^ sent-off ^ off
+ *
+ * STREAM frame
+ * ^ ^
+ * |xxxxxxxxxxxxxxxxx|
+ */
+
+ BUG_ON_HOT(qcs->tx.sent_offset < qcs->stream->ack_offset);
+ BUG_ON_HOT(qcs->tx.offset < qcs->tx.sent_offset);
+ BUG_ON_HOT(qcc->tx.offsets < qcc->tx.sent_offsets);
+
+ left = qcs->tx.offset - qcs->tx.sent_offset;
+ to_xfer = QUIC_MIN(b_data(in), b_room(out));
+
+ BUG_ON_HOT(qcs->tx.offset > qcs->tx.msd);
+ /* do not exceed flow control limit */
+ if (qcs->tx.offset + to_xfer > qcs->tx.msd)
+ to_xfer = qcs->tx.msd - qcs->tx.offset;
+
+ BUG_ON_HOT(qcc->tx.offsets > qcc->rfctl.md);
+ /* do not overcome flow control limit on connection */
+ if (qcc->tx.offsets + to_xfer > qcc->rfctl.md)
+ to_xfer = qcc->rfctl.md - qcc->tx.offsets;
+
+ if (!left && !to_xfer)
+ goto out;
+
+ total = b_force_xfer(out, in, to_xfer);
+
+ out:
+ {
+ struct qcs_xfer_data_trace_arg arg = {
+ .prep = b_data(out), .xfer = total,
+ };
+ TRACE_LEAVE(QMUX_EV_QCS_SEND|QMUX_EV_QCS_XFER_DATA,
+ qcc->conn, qcs, &arg);
+ }
+
+ return total;
+}
+
+/* Prepare a STREAM frame for <qcs> instance using <out> as payload. The frame
+ * is appended in <frm_list>. Set <fin> if this is supposed to be the last
+ * stream frame.
+ *
+ * Returns the length of the STREAM frame or a negative error code.
+ */
+static int qcs_build_stream_frm(struct qcs *qcs, struct buffer *out, char fin,
+ struct list *frm_list)
+{
+ struct qcc *qcc = qcs->qcc;
+ struct quic_frame *frm;
+ int head, total;
+ uint64_t base_off;
+
+ TRACE_ENTER(QMUX_EV_QCS_SEND, qcc->conn, qcs);
+
+ /* if ack_offset < buf_offset, it points to an older buffer. */
+ base_off = MAX(qcs->stream->buf_offset, qcs->stream->ack_offset);
+ BUG_ON(qcs->tx.sent_offset < base_off);
+
+ head = qcs->tx.sent_offset - base_off;
+ total = b_data(out) - head;
+ BUG_ON(total < 0);
+
+ if (!total && !fin) {
+ /* No need to send anything if total is NULL and no FIN to signal. */
+ TRACE_LEAVE(QMUX_EV_QCS_SEND, qcc->conn, qcs);
+ return 0;
+ }
+ BUG_ON((!total && qcs->tx.sent_offset > qcs->tx.offset) ||
+ (total && qcs->tx.sent_offset >= qcs->tx.offset));
+ BUG_ON(qcs->tx.sent_offset + total > qcs->tx.offset);
+ BUG_ON(qcc->tx.sent_offsets + total > qcc->rfctl.md);
+
+ TRACE_PROTO("sending STREAM frame", QMUX_EV_QCS_SEND, qcc->conn, qcs);
+ frm = pool_zalloc(pool_head_quic_frame);
+ if (!frm) {
+ TRACE_ERROR("frame alloc failure", QMUX_EV_QCS_SEND, qcc->conn, qcs);
+ goto err;
+ }
+
+ LIST_INIT(&frm->reflist);
+ frm->type = QUIC_FT_STREAM_8;
+ frm->stream.stream = qcs->stream;
+ frm->stream.id = qcs->id;
+ frm->stream.buf = out;
+ frm->stream.data = (unsigned char *)b_peek(out, head);
+ frm->stream.dup = 0;
+
+ /* FIN is positioned only when the buffer has been totally emptied. */
+ if (fin)
+ frm->type |= QUIC_STREAM_FRAME_TYPE_FIN_BIT;
+
+ if (qcs->tx.sent_offset) {
+ frm->type |= QUIC_STREAM_FRAME_TYPE_OFF_BIT;
+ frm->stream.offset.key = qcs->tx.sent_offset;
+ }
+
+ frm->type |= QUIC_STREAM_FRAME_TYPE_LEN_BIT;
+ frm->stream.len = total;
+
+ LIST_APPEND(frm_list, &frm->list);
+
+ out:
+ {
+ struct qcs_build_stream_trace_arg arg = {
+ .len = frm->stream.len, .fin = fin,
+ .offset = frm->stream.offset.key,
+ };
+ TRACE_LEAVE(QMUX_EV_QCS_SEND|QMUX_EV_QCS_BUILD_STRM,
+ qcc->conn, qcs, &arg);
+ }
+
+ return total;
+
+ err:
+ TRACE_LEAVE(QMUX_EV_QCS_SEND, qcc->conn, qcs);
+ return -1;
+}
+
+/* Check after transferring data from qcs.tx.buf if FIN must be set on the next
+ * STREAM frame for <qcs>.
+ *
+ * Returns true if FIN must be set else false.
+ */
+static int qcs_stream_fin(struct qcs *qcs)
+{
+ return qcs->flags & QC_SF_FIN_STREAM && !b_data(&qcs->tx.buf);
+}
+
+/* This function must be called by the upper layer to inform about the sending
+ * of a STREAM frame for <qcs> instance. The frame is of <data> length and on
+ * <offset>.
+ */
+void qcc_streams_sent_done(struct qcs *qcs, uint64_t data, uint64_t offset)
+{
+ struct qcc *qcc = qcs->qcc;
+ uint64_t diff;
+
+ TRACE_ENTER(QMUX_EV_QCS_SEND, qcc->conn, qcs);
+
+ BUG_ON(offset > qcs->tx.sent_offset);
+ BUG_ON(offset + data > qcs->tx.offset);
+
+ /* check if the STREAM frame has already been notified. It can happen
+ * for retransmission.
+ */
+ if (offset + data < qcs->tx.sent_offset) {
+ TRACE_DEVEL("offset already notified", QMUX_EV_QCS_SEND, qcc->conn, qcs);
+ goto out;
+ }
+
+ qcs_idle_open(qcs);
+
+ diff = offset + data - qcs->tx.sent_offset;
+ if (diff) {
+ /* increase offset sum on connection */
+ qcc->tx.sent_offsets += diff;
+ BUG_ON_HOT(qcc->tx.sent_offsets > qcc->rfctl.md);
+ if (qcc->tx.sent_offsets == qcc->rfctl.md) {
+ qcc->flags |= QC_CF_BLK_MFCTL;
+ TRACE_STATE("connection flow-control reached", QMUX_EV_QCS_SEND, qcc->conn);
+ }
+
+ /* increase offset on stream */
+ qcs->tx.sent_offset += diff;
+ BUG_ON_HOT(qcs->tx.sent_offset > qcs->tx.msd);
+ BUG_ON_HOT(qcs->tx.sent_offset > qcs->tx.offset);
+ if (qcs->tx.sent_offset == qcs->tx.msd) {
+ qcs->flags |= QC_SF_BLK_SFCTL;
+ TRACE_STATE("stream flow-control reached", QMUX_EV_QCS_SEND, qcc->conn, qcs);
+ }
+
+ if (qcs->tx.offset == qcs->tx.sent_offset &&
+ b_full(&qcs->stream->buf->buf)) {
+ qc_stream_buf_release(qcs->stream);
+ /* prepare qcs for immediate send retry if data to send */
+ if (b_data(&qcs->tx.buf))
+ LIST_APPEND(&qcc->send_retry_list, &qcs->el);
+ }
+ }
+
+ if (qcs->tx.offset == qcs->tx.sent_offset && !b_data(&qcs->tx.buf) &&
+ qcs->flags & (QC_SF_FIN_STREAM|QC_SF_DETACH)) {
+ /* Close stream locally. */
+ qcs_close_local(qcs);
+ /* Reset flag to not emit multiple FIN STREAM frames. */
+ qcs->flags &= ~QC_SF_FIN_STREAM;
+ }
+
+ out:
+ TRACE_LEAVE(QMUX_EV_QCS_SEND, qcc->conn, qcs);
+}
+
+/* Wrapper for send on transport layer. Send a list of frames <frms> for the
+ * connection <qcc>.
+ *
+ * Returns 0 if all data sent with success else non-zero.
+ */
+static int qc_send_frames(struct qcc *qcc, struct list *frms)
+{
+ TRACE_ENTER(QMUX_EV_QCC_SEND, qcc->conn);
+
+ if (LIST_ISEMPTY(frms)) {
+ TRACE_DEVEL("no frames to send", QMUX_EV_QCC_SEND, qcc->conn);
+ goto err;
+ }
+
+ LIST_INIT(&qcc->send_retry_list);
+
+ if (!qc_send_mux(qcc->conn->handle.qc, frms))
+ goto err;
+
+ /* If there is frames left at this stage, transport layer is blocked.
+ * Subscribe on it to retry later.
+ */
+ if (!LIST_ISEMPTY(frms)) {
+ TRACE_DEVEL("remaining frames to send, subscribing", QMUX_EV_QCC_SEND, qcc->conn);
+ qcc->conn->xprt->subscribe(qcc->conn, qcc->conn->xprt_ctx,
+ SUB_RETRY_SEND, &qcc->wait_event);
+ goto err;
+ }
+
+ TRACE_LEAVE(QMUX_EV_QCC_SEND, qcc->conn);
+ return 0;
+
+ err:
+ TRACE_LEAVE(QMUX_EV_QCC_SEND, qcc->conn);
+ return 1;
+}
+
+/* Emit a RESET_STREAM on <qcs>.
+ *
+ * Returns 0 if the frame has been successfully sent else non-zero.
+ */
+static int qcs_send_reset(struct qcs *qcs)
+{
+ struct list frms = LIST_HEAD_INIT(frms);
+ struct quic_frame *frm;
+
+ TRACE_ENTER(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
+
+ frm = pool_zalloc(pool_head_quic_frame);
+ if (!frm) {
+ TRACE_LEAVE(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
+ return 1;
+ }
+
+ LIST_INIT(&frm->reflist);
+ frm->type = QUIC_FT_RESET_STREAM;
+ frm->reset_stream.id = qcs->id;
+ frm->reset_stream.app_error_code = qcs->err;
+ frm->reset_stream.final_size = qcs->tx.sent_offset;
+
+ LIST_APPEND(&frms, &frm->list);
+ if (qc_send_frames(qcs->qcc, &frms)) {
+ pool_free(pool_head_quic_frame, frm);
+ TRACE_DEVEL("cannot send RESET_STREAM", QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
+ return 1;
+ }
+
+ if (qcs_sc(qcs)) {
+ se_fl_set_error(qcs->sd);
+ qcs_alert(qcs);
+ }
+
+ qcs_close_local(qcs);
+ qcs->flags &= ~QC_SF_TO_RESET;
+
+ TRACE_LEAVE(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
+ return 0;
+}
+
+/* Used internally by qc_send function. Proceed to send for <qcs>. This will
+ * transfer data from qcs buffer to its quic_stream counterpart. A STREAM frame
+ * is then generated and inserted in <frms> list.
+ *
+ * Returns the total bytes transferred between qcs and quic_stream buffers. Can
+ * be null if out buffer cannot be allocated.
+ */
+static int _qc_send_qcs(struct qcs *qcs, struct list *frms)
+{
+ struct qcc *qcc = qcs->qcc;
+ struct buffer *buf = &qcs->tx.buf;
+ struct buffer *out = qc_stream_buf_get(qcs->stream);
+ int xfer = 0;
+ char fin = 0;
+
+ /* Allocate <out> buffer if necessary. */
+ if (!out) {
+ if (qcc->flags & QC_CF_CONN_FULL)
+ return 0;
+
+ out = qc_stream_buf_alloc(qcs->stream, qcs->tx.offset);
+ if (!out) {
+ qcc->flags |= QC_CF_CONN_FULL;
+ return 0;
+ }
+ }
+
+ /* Transfer data from <buf> to <out>. */
+ if (b_data(buf)) {
+ xfer = qcs_xfer_data(qcs, out, buf);
+ if (xfer > 0) {
+ qcs_notify_send(qcs);
+ qcs->flags &= ~QC_SF_BLK_MROOM;
+ }
+
+ qcs->tx.offset += xfer;
+ BUG_ON_HOT(qcs->tx.offset > qcs->tx.msd);
+ qcc->tx.offsets += xfer;
+ BUG_ON_HOT(qcc->tx.offsets > qcc->rfctl.md);
+ }
+
+ /* out buffer cannot be emptied if qcs offsets differ. */
+ BUG_ON(!b_data(out) && qcs->tx.sent_offset != qcs->tx.offset);
+
+ /* FIN is set if all incoming data were transferred. */
+ fin = qcs_stream_fin(qcs);
+
+ /* Build a new STREAM frame with <out> buffer. */
+ if (qcs->tx.sent_offset != qcs->tx.offset || fin) {
+ int ret;
+ ret = qcs_build_stream_frm(qcs, out, fin, frms);
+ if (ret < 0) { ABORT_NOW(); /* TODO handle this properly */ }
+ }
+
+ return xfer;
+}
+
+/* Proceed to sending. Loop through all available streams for the <qcc>
+ * instance and try to send as much as possible.
+ *
+ * Returns the total of bytes sent to the transport layer.
+ */
+static int qc_send(struct qcc *qcc)
+{
+ struct list frms = LIST_HEAD_INIT(frms);
+ struct eb64_node *node;
+ struct qcs *qcs, *qcs_tmp;
+ int total = 0, tmp_total = 0;
+
+ TRACE_ENTER(QMUX_EV_QCC_SEND, qcc->conn);
+
+ if (qcc->conn->flags & CO_FL_SOCK_WR_SH || qcc->flags & QC_CF_CC_EMIT) {
+ qcc->conn->flags |= CO_FL_ERROR;
+ TRACE_DEVEL("connection on error", QMUX_EV_QCC_SEND, qcc->conn);
+ goto err;
+ }
+
+ if (!LIST_ISEMPTY(&qcc->lfctl.frms)) {
+ if (qc_send_frames(qcc, &qcc->lfctl.frms)) {
+ TRACE_DEVEL("flow-control frames rejected by transport, aborting send", QMUX_EV_QCC_SEND, qcc->conn);
+ goto out;
+ }
+ }
+
+ if (qcc->flags & QC_CF_BLK_MFCTL)
+ return 0;
+
+ if (!(qcc->flags & QC_CF_APP_FINAL) && !eb_is_empty(&qcc->streams_by_id) &&
+ qcc->app_ops->finalize) {
+ /* Finalize the application layer before sending any stream.
+ * For h3 this consists in preparing the control stream data (SETTINGS h3).
+ */
+ qcc->app_ops->finalize(qcc->ctx);
+ qcc->flags |= QC_CF_APP_FINAL;
+ }
+
+ /* loop through all streams, construct STREAM frames if data available.
+ * TODO optimize the loop to favor streams which are not too heavy.
+ */
+ node = eb64_first(&qcc->streams_by_id);
+ while (node) {
+ int ret;
+ uint64_t id;
+
+ qcs = eb64_entry(node, struct qcs, by_id);
+ id = qcs->id;
+
+ if (quic_stream_is_uni(id) && quic_stream_is_remote(qcc, id)) {
+ node = eb64_next(node);
+ continue;
+ }
+
+ if (qcs->flags & QC_SF_TO_RESET) {
+ qcs_send_reset(qcs);
+ node = eb64_next(node);
+ continue;
+ }
+
+ if (qcs_is_close_local(qcs)) {
+ node = eb64_next(node);
+ continue;
+ }
+
+ if (qcs->flags & QC_SF_BLK_SFCTL) {
+ node = eb64_next(node);
+ continue;
+ }
+
+ /* Check if there is something to send. */
+ if (!b_data(&qcs->tx.buf) && !qcs_stream_fin(qcs) &&
+ !qc_stream_buf_get(qcs->stream)) {
+ node = eb64_next(node);
+ continue;
+ }
+
+ ret = _qc_send_qcs(qcs, &frms);
+ total += ret;
+ node = eb64_next(node);
+ }
+
+ if (qc_send_frames(qcc, &frms)) {
+ /* data rejected by transport layer, do not retry. */
+ goto out;
+ }
+
+ retry:
+ tmp_total = 0;
+ list_for_each_entry_safe(qcs, qcs_tmp, &qcc->send_retry_list, el) {
+ int ret;
+ BUG_ON(!b_data(&qcs->tx.buf));
+ BUG_ON(qc_stream_buf_get(qcs->stream));
+
+ ret = _qc_send_qcs(qcs, &frms);
+ tmp_total += ret;
+ LIST_DELETE(&qcs->el);
+ }
+
+ total += tmp_total;
+ if (!qc_send_frames(qcc, &frms) && !LIST_ISEMPTY(&qcc->send_retry_list))
+ goto retry;
+
+ out:
+ /* Deallocate frames that the transport layer has rejected. */
+ if (!LIST_ISEMPTY(&frms)) {
+ struct quic_frame *frm, *frm2;
+ list_for_each_entry_safe(frm, frm2, &frms, list) {
+ LIST_DELETE(&frm->list);
+ pool_free(pool_head_quic_frame, frm);
+ }
+ }
+
+ TRACE_LEAVE(QMUX_EV_QCC_SEND, qcc->conn);
+ return total;
+
+ err:
+ TRACE_LEAVE(QMUX_EV_QCC_SEND, qcc->conn);
+ return 0;
+}
+
+/* Proceed on receiving. Loop through all streams from <qcc> and use decode_qcs
+ * operation.
+ *
+ * Returns 0 on success else non-zero.
+ */
+static int qc_recv(struct qcc *qcc)
+{
+ struct eb64_node *node;
+ struct qcs *qcs;
+
+ TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
+
+ if (qcc->flags & QC_CF_CC_EMIT) {
+ TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
+ return 0;
+ }
+
+ node = eb64_first(&qcc->streams_by_id);
+ while (node) {
+ uint64_t 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);
+ node = eb64_next(node);
+
+ if (qcs->flags & QC_SF_READ_ABORTED) {
+ /* TODO should send a STOP_SENDING */
+ qcs_free(qcs);
+ }
+ }
+
+ TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
+ return 0;
+}
+
+
+/* Release all streams which have their transfer operation achieved.
+ *
+ * Returns true if at least one stream is released.
+ */
+static int qc_purge_streams(struct qcc *qcc)
+{
+ struct eb64_node *node;
+ int release = 0;
+
+ TRACE_ENTER(QMUX_EV_QCC_WAKE, qcc->conn);
+
+ node = eb64_first(&qcc->streams_by_id);
+ while (node) {
+ struct qcs *qcs = eb64_entry(node, struct qcs, by_id);
+ node = eb64_next(node);
+
+ /* Release not attached closed streams. */
+ if (qcs->st == QC_SS_CLO && !qcs_sc(qcs)) {
+ TRACE_STATE("purging closed stream", QMUX_EV_QCC_WAKE, qcs->qcc->conn, qcs);
+ qcs_destroy(qcs);
+ release = 1;
+ continue;
+ }
+
+ /* Release detached streams with empty buffer. */
+ if (qcs->flags & QC_SF_DETACH) {
+ if (qcs_is_close_local(qcs)) {
+ TRACE_STATE("purging detached stream", QMUX_EV_QCC_WAKE, qcs->qcc->conn, qcs);
+ qcs_destroy(qcs);
+ release = 1;
+ continue;
+ }
+
+ qcc->conn->xprt->subscribe(qcc->conn, qcc->conn->xprt_ctx,
+ SUB_RETRY_SEND, &qcc->wait_event);
+ }
+ }
+
+ TRACE_LEAVE(QMUX_EV_QCC_WAKE, qcc->conn);
+ return release;
+}
+
+/* release function. This one should be called to free all resources allocated
+ * to the mux.
+ */
+static void qc_release(struct qcc *qcc)
+{
+ struct connection *conn = qcc->conn;
+ struct eb64_node *node;
+
+ TRACE_ENTER(QMUX_EV_QCC_END, conn);
+
+ if (!(qcc->flags & QC_CF_CC_EMIT)) {
+ if (qcc->app_ops && qcc->app_ops->shutdown) {
+ /* Application protocol with dedicated connection closing
+ * procedure.
+ */
+ qcc->app_ops->shutdown(qcc->ctx);
+
+ /* useful if application protocol should emit some closing
+ * frames. For example HTTP/3 GOAWAY frame.
+ */
+ qc_send(qcc);
+ }
+ else {
+ qcc_emit_cc_app(qcc, QC_ERR_NO_ERROR, 0);
+ }
+ }
+
+ if (qcc->task) {
+ task_destroy(qcc->task);
+ qcc->task = NULL;
+ }
+
+ if (qcc->wait_event.tasklet)
+ tasklet_free(qcc->wait_event.tasklet);
+ if (conn && qcc->wait_event.events) {
+ conn->xprt->unsubscribe(conn, conn->xprt_ctx,
+ qcc->wait_event.events,
+ &qcc->wait_event);
+ }
+
+ /* liberate remaining qcs instances */
+ node = eb64_first(&qcc->streams_by_id);
+ while (node) {
+ struct qcs *qcs = eb64_entry(node, struct qcs, by_id);
+ node = eb64_next(node);
+ qcs_free(qcs);
+ }
+
+ while (!LIST_ISEMPTY(&qcc->lfctl.frms)) {
+ struct quic_frame *frm = LIST_ELEM(qcc->lfctl.frms.n, struct quic_frame *, list);
+ LIST_DELETE(&frm->list);
+ pool_free(pool_head_quic_frame, frm);
+ }
+
+ if (qcc->app_ops && qcc->app_ops->release)
+ qcc->app_ops->release(qcc->ctx);
+ TRACE_PROTO("application layer released", QMUX_EV_QCC_END, conn);
+
+ pool_free(pool_head_qcc, qcc);
+
+ if (conn) {
+ LIST_DEL_INIT(&conn->stopping_list);
+
+ conn->handle.qc->conn = NULL;
+ conn->mux = NULL;
+ conn->ctx = NULL;
+
+ TRACE_DEVEL("freeing conn", QMUX_EV_QCC_END, conn);
+
+ conn_stop_tracking(conn);
+ conn_full_close(conn);
+ if (conn->destroy_cb)
+ conn->destroy_cb(conn);
+ conn_free(conn);
+ }
+
+ TRACE_LEAVE(QMUX_EV_QCC_END);
+}
+
+static struct task *qc_io_cb(struct task *t, void *ctx, unsigned int status)
+{
+ struct qcc *qcc = ctx;
+
+ TRACE_ENTER(QMUX_EV_QCC_WAKE, qcc->conn);
+
+ qc_send(qcc);
+
+ if (qc_purge_streams(qcc)) {
+ if (qcc_is_dead(qcc)) {
+ TRACE_STATE("releasing dead connection", QMUX_EV_QCC_WAKE, qcc->conn);
+ goto release;
+ }
+ }
+
+ qc_recv(qcc);
+
+ /* TODO check if qcc proxy is disabled. If yes, use graceful shutdown
+ * to close the connection.
+ */
+
+ qcc_refresh_timeout(qcc);
+
+ end:
+ TRACE_LEAVE(QMUX_EV_QCC_WAKE, qcc->conn);
+ return NULL;
+
+ release:
+ qc_release(qcc);
+ TRACE_LEAVE(QMUX_EV_QCC_WAKE);
+ return NULL;
+}
+
+static struct task *qc_timeout_task(struct task *t, void *ctx, unsigned int state)
+{
+ struct qcc *qcc = ctx;
+ int expired = tick_is_expired(t->expire, now_ms);
+
+ TRACE_ENTER(QMUX_EV_QCC_WAKE, qcc ? qcc->conn : NULL);
+
+ if (qcc) {
+ if (!expired) {
+ TRACE_DEVEL("not expired", QMUX_EV_QCC_WAKE, qcc->conn);
+ goto requeue;
+ }
+
+ if (!qcc_may_expire(qcc)) {
+ TRACE_DEVEL("cannot expired", QMUX_EV_QCC_WAKE, qcc->conn);
+ t->expire = TICK_ETERNITY;
+ goto requeue;
+ }
+ }
+
+ task_destroy(t);
+
+ if (!qcc) {
+ TRACE_DEVEL("no more qcc", QMUX_EV_QCC_WAKE);
+ goto out;
+ }
+
+ qcc->task = NULL;
+
+ /* TODO depending on the timeout condition, different shutdown mode
+ * should be used. For http keep-alive or disabled proxy, a graceful
+ * shutdown should occurs. For all other cases, an immediate close
+ * seems legitimate.
+ */
+ if (qcc_is_dead(qcc)) {
+ TRACE_STATE("releasing dead connection", QMUX_EV_QCC_WAKE, qcc->conn);
+ qc_release(qcc);
+ }
+
+ out:
+ TRACE_LEAVE(QMUX_EV_QCC_WAKE);
+ return NULL;
+
+ requeue:
+ TRACE_LEAVE(QMUX_EV_QCC_WAKE);
+ return t;
+}
+
+static int qc_init(struct connection *conn, struct proxy *prx,
+ struct session *sess, struct buffer *input)
+{
+ struct qcc *qcc;
+ struct quic_transport_params *lparams, *rparams;
+
+ TRACE_ENTER(QMUX_EV_QCC_NEW);
+
+ qcc = pool_alloc(pool_head_qcc);
+ if (!qcc) {
+ TRACE_ERROR("alloc failure", QMUX_EV_QCC_NEW);
+ goto fail_no_qcc;
+ }
+
+ qcc->conn = conn;
+ conn->ctx = qcc;
+ qcc->nb_hreq = qcc->nb_sc = 0;
+ qcc->flags = 0;
+
+ qcc->app_ops = NULL;
+ if (qcc_install_app_ops(qcc, conn->handle.qc->app_ops)) {
+ TRACE_PROTO("Cannot install app layer", QMUX_EV_QCC_NEW, qcc->conn);
+ /* prepare a CONNECTION_CLOSE frame */
+ quic_set_connection_close(conn->handle.qc, quic_err_transport(QC_ERR_APPLICATION_ERROR));
+ goto fail_no_tasklet;
+ }
+
+ qcc->streams_by_id = EB_ROOT_UNIQUE;
+
+ /* Server parameters, params used for RX flow control. */
+ lparams = &conn->handle.qc->rx.params;
+
+ qcc->rx.max_data = lparams->initial_max_data;
+ qcc->tx.sent_offsets = qcc->tx.offsets = 0;
+
+ /* Client initiated streams must respect the server flow control. */
+ qcc->strms[QCS_CLT_BIDI].max_streams = lparams->initial_max_streams_bidi;
+ qcc->strms[QCS_CLT_BIDI].nb_streams = 0;
+ qcc->strms[QCS_CLT_BIDI].rx.max_data = 0;
+ qcc->strms[QCS_CLT_BIDI].tx.max_data = lparams->initial_max_stream_data_bidi_remote;
+
+ qcc->strms[QCS_CLT_UNI].max_streams = lparams->initial_max_streams_uni;
+ qcc->strms[QCS_CLT_UNI].nb_streams = 0;
+ qcc->strms[QCS_CLT_UNI].rx.max_data = 0;
+ qcc->strms[QCS_CLT_UNI].tx.max_data = lparams->initial_max_stream_data_uni;
+
+ /* Server initiated streams must respect the server flow control. */
+ qcc->strms[QCS_SRV_BIDI].max_streams = 0;
+ qcc->strms[QCS_SRV_BIDI].nb_streams = 0;
+ qcc->strms[QCS_SRV_BIDI].rx.max_data = lparams->initial_max_stream_data_bidi_local;
+ qcc->strms[QCS_SRV_BIDI].tx.max_data = 0;
+
+ qcc->strms[QCS_SRV_UNI].max_streams = 0;
+ qcc->strms[QCS_SRV_UNI].nb_streams = 0;
+ qcc->strms[QCS_SRV_UNI].rx.max_data = lparams->initial_max_stream_data_uni;
+ qcc->strms[QCS_SRV_UNI].tx.max_data = 0;
+
+ LIST_INIT(&qcc->lfctl.frms);
+ qcc->lfctl.ms_bidi = qcc->lfctl.ms_bidi_init = lparams->initial_max_streams_bidi;
+ qcc->lfctl.ms_uni = lparams->initial_max_streams_uni;
+ qcc->lfctl.msd_bidi_l = lparams->initial_max_stream_data_bidi_local;
+ qcc->lfctl.msd_bidi_r = lparams->initial_max_stream_data_bidi_remote;
+ qcc->lfctl.msd_uni_r = lparams->initial_max_stream_data_uni;
+ qcc->lfctl.cl_bidi_r = 0;
+
+ qcc->lfctl.md = qcc->lfctl.md_init = lparams->initial_max_data;
+ qcc->lfctl.offsets_recv = qcc->lfctl.offsets_consume = 0;
+
+ rparams = &conn->handle.qc->tx.params;
+ qcc->rfctl.md = rparams->initial_max_data;
+ qcc->rfctl.msd_bidi_l = rparams->initial_max_stream_data_bidi_local;
+ qcc->rfctl.msd_bidi_r = rparams->initial_max_stream_data_bidi_remote;
+ qcc->rfctl.msd_uni_l = rparams->initial_max_stream_data_uni;
+
+ if (conn_is_back(conn)) {
+ qcc->next_bidi_l = 0x00;
+ qcc->largest_bidi_r = 0x01;
+ qcc->next_uni_l = 0x02;
+ qcc->largest_uni_r = 0x03;
+ }
+ else {
+ qcc->largest_bidi_r = 0x00;
+ qcc->next_bidi_l = 0x01;
+ qcc->largest_uni_r = 0x02;
+ qcc->next_uni_l = 0x03;
+ }
+
+ qcc->wait_event.tasklet = tasklet_new();
+ if (!qcc->wait_event.tasklet) {
+ TRACE_ERROR("taslket alloc failure", QMUX_EV_QCC_NEW);
+ goto fail_no_tasklet;
+ }
+
+ LIST_INIT(&qcc->send_retry_list);
+
+ qcc->wait_event.tasklet->process = qc_io_cb;
+ qcc->wait_event.tasklet->context = qcc;
+ qcc->wait_event.events = 0;
+
+ qcc->proxy = prx;
+ /* haproxy timeouts */
+ qcc->task = NULL;
+ qcc->timeout = conn_is_back(qcc->conn) ? prx->timeout.server :
+ prx->timeout.client;
+ if (tick_isset(qcc->timeout)) {
+ qcc->task = task_new_here();
+ if (!qcc->task) {
+ TRACE_ERROR("timeout task alloc failure", QMUX_EV_QCC_NEW);
+ goto fail_no_timeout_task;
+ }
+ qcc->task->process = qc_timeout_task;
+ qcc->task->context = qcc;
+ qcc->task->expire = tick_add(now_ms, qcc->timeout);
+ }
+ qcc_reset_idle_start(qcc);
+ LIST_INIT(&qcc->opening_list);
+
+ if (!conn_is_back(conn)) {
+ if (!LIST_INLIST(&conn->stopping_list)) {
+ LIST_APPEND(&mux_stopping_data[tid].list,
+ &conn->stopping_list);
+ }
+ }
+
+ HA_ATOMIC_STORE(&conn->handle.qc->qcc, qcc);
+ /* init read cycle */
+ tasklet_wakeup(qcc->wait_event.tasklet);
+
+ TRACE_LEAVE(QMUX_EV_QCC_NEW, qcc->conn);
+ return 0;
+
+ fail_no_timeout_task:
+ tasklet_free(qcc->wait_event.tasklet);
+ fail_no_tasklet:
+ if (qcc->app_ops && qcc->app_ops->release)
+ qcc->app_ops->release(qcc->ctx);
+ pool_free(pool_head_qcc, qcc);
+ fail_no_qcc:
+ TRACE_LEAVE(QMUX_EV_QCC_NEW);
+ return -1;
+}
+
+static void qc_destroy(void *ctx)
+{
+ struct qcc *qcc = ctx;
+
+ TRACE_ENTER(QMUX_EV_QCC_END, qcc->conn);
+ qc_release(qcc);
+ TRACE_LEAVE(QMUX_EV_QCC_END);
+}
+
+static void qc_detach(struct sedesc *sd)
+{
+ struct qcs *qcs = sd->se;
+ struct qcc *qcc = qcs->qcc;
+
+ TRACE_ENTER(QMUX_EV_STRM_END, qcc->conn, qcs);
+
+ /* TODO this BUG_ON_HOT() is not correct as the stconn layer may detach
+ * from the stream even if it is not closed remotely at the QUIC layer.
+ * This happens for example when a stream must be closed due to a
+ * rejected request. To better handle these cases, it will be required
+ * to implement shutr/shutw MUX operations. Once this is done, this
+ * BUG_ON_HOT() statement can be adjusted.
+ */
+ //BUG_ON_HOT(!qcs_is_close_remote(qcs));
+
+ qcc_rm_sc(qcc);
+
+ if (!qcs_is_close_local(qcs) && !(qcc->conn->flags & CO_FL_ERROR)) {
+ TRACE_STATE("remaining data, detaching qcs", QMUX_EV_STRM_END, qcc->conn, qcs);
+ qcs->flags |= QC_SF_DETACH;
+ qcc_refresh_timeout(qcc);
+
+ TRACE_LEAVE(QMUX_EV_STRM_END, qcc->conn, qcs);
+ return;
+ }
+
+ qcs_destroy(qcs);
+
+ if (qcc_is_dead(qcc)) {
+ TRACE_STATE("killing dead connection", QMUX_EV_STRM_END, qcc->conn);
+ goto release;
+ }
+ else if (qcc->task) {
+ TRACE_DEVEL("refreshing connection's timeout", QMUX_EV_STRM_END, qcc->conn);
+ qcc_refresh_timeout(qcc);
+ }
+ else {
+ TRACE_DEVEL("completed", QMUX_EV_STRM_END, qcc->conn);
+ }
+
+ TRACE_LEAVE(QMUX_EV_STRM_END, qcc->conn);
+ return;
+
+ release:
+ qc_release(qcc);
+ TRACE_LEAVE(QMUX_EV_STRM_END);
+ return;
+}
+
+/* Called from the upper layer, to receive data */
+static size_t qc_rcv_buf(struct stconn *sc, struct buffer *buf,
+ size_t count, int flags)
+{
+ struct qcs *qcs = __sc_mux_strm(sc);
+ size_t ret = 0;
+ char fin = 0;
+
+ TRACE_ENTER(QMUX_EV_STRM_RECV, qcs->qcc->conn, qcs);
+
+ ret = qcs_http_rcv_buf(qcs, buf, count, &fin);
+
+ if (b_data(&qcs->rx.app_buf)) {
+ se_fl_set(qcs->sd, SE_FL_RCV_MORE | SE_FL_WANT_ROOM);
+ }
+ else {
+ se_fl_clr(qcs->sd, SE_FL_RCV_MORE | SE_FL_WANT_ROOM);
+ if (se_fl_test(qcs->sd, SE_FL_ERR_PENDING))
+ se_fl_set(qcs->sd, SE_FL_ERROR);
+
+ /* Set end-of-input if FIN received and all data extracted. */
+ if (fin)
+ se_fl_set(qcs->sd, SE_FL_EOI);
+
+ if (b_size(&qcs->rx.app_buf)) {
+ b_free(&qcs->rx.app_buf);
+ offer_buffers(NULL, 1);
+ }
+ }
+
+ if (ret) {
+ qcs->flags &= ~QC_SF_DEM_FULL;
+ tasklet_wakeup(qcs->qcc->wait_event.tasklet);
+ }
+
+ TRACE_LEAVE(QMUX_EV_STRM_RECV, qcs->qcc->conn, qcs);
+
+ return ret;
+}
+
+static size_t qc_snd_buf(struct stconn *sc, struct buffer *buf,
+ size_t count, int flags)
+{
+ struct qcs *qcs = __sc_mux_strm(sc);
+ size_t ret;
+ char fin;
+
+ TRACE_ENTER(QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
+
+ /* stream layer has been detached so no transfer must occur after. */
+ BUG_ON_HOT(qcs->flags & QC_SF_DETACH);
+
+ if (qcs_is_close_local(qcs) || (qcs->flags & QC_SF_TO_RESET)) {
+ ret = qcs_http_reset_buf(qcs, buf, count);
+ goto end;
+ }
+
+ ret = qcs_http_snd_buf(qcs, buf, count, &fin);
+ if (fin)
+ qcs->flags |= QC_SF_FIN_STREAM;
+
+ if (ret || fin) {
+ if (!(qcs->qcc->wait_event.events & SUB_RETRY_SEND))
+ tasklet_wakeup(qcs->qcc->wait_event.tasklet);
+ }
+
+ end:
+ TRACE_LEAVE(QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
+
+ return ret;
+}
+
+/* Called from the upper layer, to subscribe <es> to events <event_type>. The
+ * event subscriber <es> is not allowed to change from a previous call as long
+ * as at least one event is still subscribed. The <event_type> must only be a
+ * combination of SUB_RETRY_RECV and SUB_RETRY_SEND. It always returns 0.
+ */
+static int qc_subscribe(struct stconn *sc, int event_type,
+ struct wait_event *es)
+{
+ return qcs_subscribe(__sc_mux_strm(sc), event_type, es);
+}
+
+/* Called from the upper layer, to unsubscribe <es> from events <event_type>.
+ * The <es> pointer is not allowed to differ from the one passed to the
+ * subscribe() call. It always returns zero.
+ */
+static int qc_unsubscribe(struct stconn *sc, int event_type, struct wait_event *es)
+{
+ struct qcs *qcs = __sc_mux_strm(sc);
+
+ BUG_ON(event_type & ~(SUB_RETRY_SEND|SUB_RETRY_RECV));
+ BUG_ON(qcs->subs && qcs->subs != es);
+
+ es->events &= ~event_type;
+ if (!es->events)
+ qcs->subs = NULL;
+
+ return 0;
+}
+
+/* Loop through all qcs from <qcc>. If CO_FL_ERROR is set on the connection,
+ * report SE_FL_ERR_PENDING|SE_FL_ERROR on the attached stream connectors and
+ * wake them.
+ */
+static int qc_wake_some_streams(struct qcc *qcc)
+{
+ struct qcs *qcs;
+ struct eb64_node *node;
+
+ for (node = eb64_first(&qcc->streams_by_id); node;
+ node = eb64_next(node)) {
+ qcs = eb64_entry(node, struct qcs, by_id);
+
+ if (!qcs_sc(qcs))
+ continue;
+
+ if (qcc->conn->flags & CO_FL_ERROR) {
+ se_fl_set(qcs->sd, SE_FL_ERR_PENDING);
+ if (se_fl_test(qcs->sd, SE_FL_EOS))
+ se_fl_set(qcs->sd, SE_FL_ERROR);
+
+ qcs_alert(qcs);
+ }
+ }
+
+ return 0;
+}
+
+static int qc_wake(struct connection *conn)
+{
+ struct qcc *qcc = conn->ctx;
+ struct proxy *prx = conn->handle.qc->li->bind_conf->frontend;
+
+ TRACE_ENTER(QMUX_EV_QCC_WAKE, conn);
+
+ /* Check if a soft-stop is in progress.
+ *
+ * TODO this is relevant for frontend connections only.
+ *
+ * TODO Client should be notified with a H3 GOAWAY and then a
+ * CONNECTION_CLOSE. However, quic-conn uses the listener socket for
+ * sending which at this stage is already closed.
+ */
+ if (unlikely(prx->flags & (PR_FL_DISABLED|PR_FL_STOPPED)))
+ qcc->conn->flags |= (CO_FL_SOCK_RD_SH|CO_FL_SOCK_WR_SH);
+
+ if (conn->handle.qc->flags & QUIC_FL_CONN_NOTIFY_CLOSE)
+ qcc->conn->flags |= (CO_FL_SOCK_RD_SH|CO_FL_SOCK_WR_SH);
+
+ qc_send(qcc);
+
+ qc_wake_some_streams(qcc);
+
+ if (qcc_is_dead(qcc))
+ goto release;
+
+ qcc_refresh_timeout(qcc);
+
+ TRACE_LEAVE(QMUX_EV_QCC_WAKE, conn);
+ return 0;
+
+ release:
+ TRACE_STATE("releasing dead connection", QMUX_EV_QCC_WAKE, qcc->conn);
+ qc_release(qcc);
+ TRACE_LEAVE(QMUX_EV_QCC_WAKE);
+ return 1;
+}
+
+
+static const struct mux_ops qc_ops = {
+ .init = qc_init,
+ .destroy = qc_destroy,
+ .detach = qc_detach,
+ .rcv_buf = qc_rcv_buf,
+ .snd_buf = qc_snd_buf,
+ .subscribe = qc_subscribe,
+ .unsubscribe = qc_unsubscribe,
+ .wake = qc_wake,
+ .flags = MX_FL_HTX|MX_FL_NO_UPG|MX_FL_FRAMED,
+ .name = "QUIC",
+};
+
+static struct mux_proto_list mux_proto_quic =
+ { .token = IST("quic"), .mode = PROTO_MODE_HTTP, .side = PROTO_SIDE_FE, .mux = &qc_ops };
+
+INITCALL1(STG_REGISTER, register_mux_proto, &mux_proto_quic);