blob: a78f8d4a483eb58447361862090ef89304e3fd61 [file] [log] [blame]
Amaury Denoyelledeed7772021-12-03 11:36:46 +01001#include <haproxy/mux_quic.h>
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002
Amaury Denoyelleeb01f592021-10-07 16:44:05 +02003#include <import/eb64tree.h>
4
Frédéric Lécailledfbae762021-02-18 09:59:01 +01005#include <haproxy/api.h>
Frédéric Lécailledfbae762021-02-18 09:59:01 +01006#include <haproxy/connection.h>
Amaury Denoyelledeed7772021-12-03 11:36:46 +01007#include <haproxy/dynbuf.h>
Frédéric Lécaille9969adb2023-01-18 11:52:21 +01008#include <haproxy/h3.h>
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02009#include <haproxy/list.h>
Amaury Denoyelle1290f1e2022-05-13 14:49:05 +020010#include <haproxy/ncbuf.h>
Amaury Denoyelledeed7772021-12-03 11:36:46 +010011#include <haproxy/pool.h>
Frédéric Lécaille9969adb2023-01-18 11:52:21 +010012#include <haproxy/proxy.h>
Amaury Denoyelled80fbca2022-09-19 17:02:28 +020013#include <haproxy/qmux_http.h>
Amaury Denoyelle36d50bf2022-09-19 16:12:38 +020014#include <haproxy/qmux_trace.h>
Amaury Denoyelle92fa63f2022-09-30 18:11:13 +020015#include <haproxy/quic_conn.h>
Amaury Denoyelle40c24f12023-01-27 17:47:49 +010016#include <haproxy/quic_frame.h>
Amaury Denoyelleb3aa07c2023-01-24 18:20:28 +010017#include <haproxy/quic_sock.h>
Amaury Denoyelle0cc02a32022-04-19 17:21:11 +020018#include <haproxy/quic_stream.h>
Frédéric Lécaille748ece62022-05-21 23:58:40 +020019#include <haproxy/quic_tp-t.h>
Amaury Denoyelleeb01f592021-10-07 16:44:05 +020020#include <haproxy/ssl_sock-t.h>
Willy Tarreaucb086c62022-05-27 09:47:12 +020021#include <haproxy/stconn.h>
Amaury Denoyelledd4fbfb2022-03-24 16:09:16 +010022#include <haproxy/trace.h>
Frédéric Lécailledfbae762021-02-18 09:59:01 +010023
Amaury Denoyelledeed7772021-12-03 11:36:46 +010024DECLARE_POOL(pool_head_qcc, "qcc", sizeof(struct qcc));
Frédéric Lécailledfbae762021-02-18 09:59:01 +010025DECLARE_POOL(pool_head_qcs, "qcs", sizeof(struct qcs));
26
Amaury Denoyelle9fab9fd2022-05-20 15:04:38 +020027/* Emit a CONNECTION_CLOSE with error <err>. This will interrupt all future
Amaury Denoyelle5c4373a2022-05-24 14:47:48 +020028 * send/receive operations.
Amaury Denoyelle9fab9fd2022-05-20 15:04:38 +020029 */
30static void qcc_emit_cc(struct qcc *qcc, int err)
31{
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +020032 TRACE_ENTER(QMUX_EV_QCC_END, qcc->conn);
33
34 TRACE_STATE("set CONNECTION_CLOSE on quic-conn", QMUX_EV_QCC_WAKE, qcc->conn);
Amaury Denoyelle57e6db72022-07-13 15:07:56 +020035 quic_set_connection_close(qcc->conn->handle.qc, quic_err_transport(err));
Amaury Denoyelle9fab9fd2022-05-20 15:04:38 +020036 qcc->flags |= QC_CF_CC_EMIT;
37 tasklet_wakeup(qcc->wait_event.tasklet);
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +020038
39 TRACE_LEAVE(QMUX_EV_QCC_END, qcc->conn);
Amaury Denoyelle9fab9fd2022-05-20 15:04:38 +020040}
41
Amaury Denoyelle4b167002022-12-12 09:59:50 +010042static void qc_free_ncbuf(struct qcs *qcs, struct ncbuf *ncbuf)
43{
44 struct buffer buf;
45
46 if (ncb_is_null(ncbuf))
47 return;
48
49 buf = b_make(ncbuf->area, ncbuf->size, 0, 0);
50 b_free(&buf);
51 offer_buffers(NULL, 1);
52
53 *ncbuf = NCBUF_NULL;
54}
55
56/* Free <qcs> instance. This function is reserved for internal usage : it must
57 * only be called on qcs alloc error or on connection shutdown. Else
58 * qcs_destroy must be prefered to handle QUIC flow-control increase.
59 */
60static void qcs_free(struct qcs *qcs)
61{
62 struct qcc *qcc = qcs->qcc;
63
64 TRACE_ENTER(QMUX_EV_QCS_END, qcc->conn, qcs);
65
Amaury Denoyelle15337fd2022-12-20 14:47:16 +010066 /* Safe to use even if already removed from the list. */
67 LIST_DEL_INIT(&qcs->el_opening);
Amaury Denoyelle20f2a422023-01-03 14:39:24 +010068 LIST_DEL_INIT(&qcs->el_send);
Amaury Denoyelle4b167002022-12-12 09:59:50 +010069
70 /* Release stream endpoint descriptor. */
71 BUG_ON(qcs->sd && !se_fl_test(qcs->sd, SE_FL_ORPHAN));
72 sedesc_free(qcs->sd);
73
74 /* Release app-layer context. */
75 if (qcs->ctx && qcc->app_ops->detach)
76 qcc->app_ops->detach(qcs);
77
78 /* Release qc_stream_desc buffer from quic-conn layer. */
79 qc_stream_desc_release(qcs->stream);
80
81 /* Free Rx/Tx buffers. */
82 qc_free_ncbuf(qcs, &qcs->rx.ncbuf);
83 b_free(&qcs->tx.buf);
84
85 BUG_ON(!qcc->strms[qcs_id_type(qcs->id)].nb_streams);
86 --qcc->strms[qcs_id_type(qcs->id)].nb_streams;
87
88 /* Remove qcs from qcc tree. */
89 eb64_delete(&qcs->by_id);
90
91 pool_free(pool_head_qcs, qcs);
92
93 TRACE_LEAVE(QMUX_EV_QCS_END, qcc->conn);
94}
95
Amaury Denoyelledeed7772021-12-03 11:36:46 +010096/* Allocate a new QUIC streams with id <id> and type <type>. */
Amaury Denoyellea509ffb2022-07-04 15:50:33 +020097static struct qcs *qcs_new(struct qcc *qcc, uint64_t id, enum qcs_type type)
Frédéric Lécailledfbae762021-02-18 09:59:01 +010098{
Amaury Denoyelledeed7772021-12-03 11:36:46 +010099 struct qcs *qcs;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100100
Amaury Denoyelle4f137572022-03-24 17:10:00 +0100101 TRACE_ENTER(QMUX_EV_QCS_NEW, qcc->conn);
102
Amaury Denoyelledeed7772021-12-03 11:36:46 +0100103 qcs = pool_alloc(pool_head_qcs);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200104 if (!qcs) {
Amaury Denoyellec7fb0d22022-08-10 16:39:54 +0200105 TRACE_ERROR("alloc failure", QMUX_EV_QCS_NEW, qcc->conn);
Amaury Denoyelle17014a62022-04-27 15:09:27 +0200106 return NULL;
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200107 }
Amaury Denoyelle17014a62022-04-27 15:09:27 +0200108
109 qcs->stream = NULL;
110 qcs->qcc = qcc;
Willy Tarreaud7b7e0d2022-05-27 16:09:35 +0200111 qcs->sd = NULL;
Amaury Denoyelle17014a62022-04-27 15:09:27 +0200112 qcs->flags = QC_SF_NONE;
Amaury Denoyelle38e60062022-07-01 16:48:42 +0200113 qcs->st = QC_SS_IDLE;
Amaury Denoyelle47447af2022-04-27 15:17:11 +0200114 qcs->ctx = NULL;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100115
Amaury Denoyelle30e260e2022-08-03 11:17:57 +0200116 /* App callback attach may register the stream for http-request wait.
117 * These fields must be initialed before.
118 */
119 LIST_INIT(&qcs->el_opening);
Amaury Denoyelle20f2a422023-01-03 14:39:24 +0100120 LIST_INIT(&qcs->el_send);
Amaury Denoyelle30e260e2022-08-03 11:17:57 +0200121 qcs->start = TICK_ETERNITY;
122
Amaury Denoyelle4b167002022-12-12 09:59:50 +0100123 /* store transport layer stream descriptor in qcc tree */
124 qcs->id = qcs->by_id.key = id;
125 eb64_insert(&qcc->streams_by_id, &qcs->by_id);
126
127 qcc->strms[type].nb_streams++;
128
Amaury Denoyelle93fba322022-05-24 16:53:14 +0200129 /* Allocate transport layer stream descriptor. Only needed for TX. */
130 if (!quic_stream_is_uni(id) || !quic_stream_is_remote(qcc, id)) {
131 struct quic_conn *qc = qcc->conn->handle.qc;
132 qcs->stream = qc_stream_desc_new(id, type, qcs, qc);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200133 if (!qcs->stream) {
Amaury Denoyellec7fb0d22022-08-10 16:39:54 +0200134 TRACE_ERROR("qc_stream_desc alloc failure", QMUX_EV_QCS_NEW, qcc->conn, qcs);
Amaury Denoyelle93fba322022-05-24 16:53:14 +0200135 goto err;
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200136 }
Amaury Denoyelle93fba322022-05-24 16:53:14 +0200137 }
Amaury Denoyelle7272cd72022-03-29 15:15:54 +0200138
Amaury Denoyelle6ea78192022-03-07 15:47:02 +0100139 /* If stream is local, use peer remote-limit, or else the opposite. */
Amaury Denoyelle176174f2022-10-21 17:02:18 +0200140 if (quic_stream_is_bidi(id)) {
141 qcs->tx.msd = quic_stream_is_local(qcc, id) ? qcc->rfctl.msd_bidi_r :
142 qcc->rfctl.msd_bidi_l;
143 }
144 else if (quic_stream_is_local(qcc, id)) {
145 qcs->tx.msd = qcc->rfctl.msd_uni_l;
146 }
Amaury Denoyelle6ea78192022-03-07 15:47:02 +0100147
Amaury Denoyelle1290f1e2022-05-13 14:49:05 +0200148 qcs->rx.ncbuf = NCBUF_NULL;
Amaury Denoyelle9a327a72022-02-14 17:11:09 +0100149 qcs->rx.app_buf = BUF_NULL;
Amaury Denoyelled46b0f52022-05-20 15:05:07 +0200150 qcs->rx.offset = qcs->rx.offset_max = 0;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100151
Amaury Denoyelle176174f2022-10-21 17:02:18 +0200152 if (quic_stream_is_bidi(id)) {
153 qcs->rx.msd = quic_stream_is_local(qcc, id) ? qcc->lfctl.msd_bidi_l :
154 qcc->lfctl.msd_bidi_r;
155 }
156 else if (quic_stream_is_remote(qcc, id)) {
157 qcs->rx.msd = qcc->lfctl.msd_uni_r;
158 }
Amaury Denoyellea9773552022-05-16 14:38:25 +0200159 qcs->rx.msd_init = qcs->rx.msd;
Amaury Denoyelle44d09122022-04-26 11:21:10 +0200160
Amaury Denoyelledeed7772021-12-03 11:36:46 +0100161 qcs->tx.buf = BUF_NULL;
Amaury Denoyelledeed7772021-12-03 11:36:46 +0100162 qcs->tx.offset = 0;
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +0100163 qcs->tx.sent_offset = 0;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100164
Amaury Denoyelledeed7772021-12-03 11:36:46 +0100165 qcs->wait_event.tasklet = NULL;
166 qcs->wait_event.events = 0;
167 qcs->subs = NULL;
168
Amaury Denoyelle843a1192022-07-04 11:44:38 +0200169 qcs->err = 0;
170
Amaury Denoyelle3d550842023-01-24 17:42:21 +0100171 if (qcc->app_ops->attach && qcc->app_ops->attach(qcs, qcc->ctx)) {
172 TRACE_ERROR("app proto failure", QMUX_EV_QCS_NEW, qcc->conn, qcs);
173 goto err;
174 }
175
Amaury Denoyelledeed7772021-12-03 11:36:46 +0100176 out:
Amaury Denoyelle4f137572022-03-24 17:10:00 +0100177 TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn, qcs);
Amaury Denoyelledeed7772021-12-03 11:36:46 +0100178 return qcs;
Amaury Denoyelle17014a62022-04-27 15:09:27 +0200179
180 err:
Amaury Denoyelle4b167002022-12-12 09:59:50 +0100181 qcs_free(qcs);
182 TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn);
Amaury Denoyelle17014a62022-04-27 15:09:27 +0200183 return NULL;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100184}
185
Amaury Denoyelle3abeb572022-07-04 11:42:27 +0200186static forceinline struct stconn *qcs_sc(const struct qcs *qcs)
187{
188 return qcs->sd ? qcs->sd->sc : NULL;
189}
190
Amaury Denoyellebd6ec1b2022-07-25 11:53:18 +0200191/* Reset the <qcc> inactivity timeout for http-keep-alive timeout. */
192static forceinline void qcc_reset_idle_start(struct qcc *qcc)
193{
194 qcc->idle_start = now_ms;
195}
196
Amaury Denoyellec603de42022-07-25 11:21:46 +0200197/* Decrement <qcc> sc. */
198static forceinline void qcc_rm_sc(struct qcc *qcc)
199{
200 BUG_ON_HOT(!qcc->nb_sc);
201 --qcc->nb_sc;
Amaury Denoyellebd6ec1b2022-07-25 11:53:18 +0200202
203 /* Reset qcc idle start for http-keep-alive timeout. Timeout will be
204 * refreshed after this on stream detach.
205 */
206 if (!qcc->nb_sc && !qcc->nb_hreq)
207 qcc_reset_idle_start(qcc);
Amaury Denoyellec603de42022-07-25 11:21:46 +0200208}
209
210/* Decrement <qcc> hreq. */
211static forceinline void qcc_rm_hreq(struct qcc *qcc)
212{
213 BUG_ON_HOT(!qcc->nb_hreq);
214 --qcc->nb_hreq;
Amaury Denoyellebd6ec1b2022-07-25 11:53:18 +0200215
216 /* Reset qcc idle start for http-keep-alive timeout. Timeout will be
217 * refreshed after this on I/O handler.
218 */
219 if (!qcc->nb_sc && !qcc->nb_hreq)
220 qcc_reset_idle_start(qcc);
Amaury Denoyellec603de42022-07-25 11:21:46 +0200221}
222
Amaury Denoyelle418ba212022-08-02 15:57:16 +0200223static inline int qcc_is_dead(const struct qcc *qcc)
224{
225 /* Mux connection is considered dead if :
226 * - all stream-desc are detached AND
227 * = connection is on error OR
228 * = mux timeout has already fired or is unset
229 */
230 if (!qcc->nb_sc && ((qcc->conn->flags & CO_FL_ERROR) || !qcc->task))
231 return 1;
232
233 return 0;
234}
235
236/* Return true if the mux timeout should be armed. */
237static inline int qcc_may_expire(struct qcc *qcc)
238{
239 return !qcc->nb_sc;
240}
241
242/* Refresh the timeout on <qcc> if needed depending on its state. */
243static void qcc_refresh_timeout(struct qcc *qcc)
244{
245 const struct proxy *px = qcc->proxy;
246
247 TRACE_ENTER(QMUX_EV_QCC_WAKE, qcc->conn);
248
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200249 if (!qcc->task) {
250 TRACE_DEVEL("already expired", QMUX_EV_QCC_WAKE, qcc->conn);
Amaury Denoyelle418ba212022-08-02 15:57:16 +0200251 goto leave;
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200252 }
Amaury Denoyelle418ba212022-08-02 15:57:16 +0200253
Amaury Denoyelle6ec98372022-08-01 17:59:38 +0200254 /* Check if upper layer is responsible of timeout management. */
255 if (!qcc_may_expire(qcc)) {
256 TRACE_DEVEL("not eligible for timeout", QMUX_EV_QCC_WAKE, qcc->conn);
257 qcc->task->expire = TICK_ETERNITY;
258 task_queue(qcc->task);
259 goto leave;
260 }
261
Amaury Denoyelle6ec98372022-08-01 17:59:38 +0200262 /* Frontend timeout management
Amaury Denoyelleeb7d3202023-02-08 15:55:24 +0100263 * - shutdown done -> timeout client-fin
Amaury Denoyelle6ec98372022-08-01 17:59:38 +0200264 * - detached streams with data left to send -> default timeout
Amaury Denoyelle30e260e2022-08-03 11:17:57 +0200265 * - stream waiting on incomplete request or no stream yet activated -> timeout http-request
Amaury Denoyelle6ec98372022-08-01 17:59:38 +0200266 * - idle after stream processing -> timeout http-keep-alive
Amaury Denoyelleb3aa07c2023-01-24 18:20:28 +0100267 *
268 * If proxy stop-stop in progress, immediate or spread close will be
269 * processed if shutdown already one or connection is idle.
Amaury Denoyelle6ec98372022-08-01 17:59:38 +0200270 */
271 if (!conn_is_back(qcc->conn)) {
Amaury Denoyelleeb7d3202023-02-08 15:55:24 +0100272 if (qcc->nb_hreq && !(qcc->flags & QC_CF_APP_SHUT)) {
Amaury Denoyelle418ba212022-08-02 15:57:16 +0200273 TRACE_DEVEL("one or more requests still in progress", QMUX_EV_QCC_WAKE, qcc->conn);
274 qcc->task->expire = tick_add_ifset(now_ms, qcc->timeout);
Amaury Denoyelle6ec98372022-08-01 17:59:38 +0200275 task_queue(qcc->task);
276 goto leave;
Amaury Denoyelle418ba212022-08-02 15:57:16 +0200277 }
Amaury Denoyelle418ba212022-08-02 15:57:16 +0200278
Amaury Denoyelleeb7d3202023-02-08 15:55:24 +0100279 if ((!LIST_ISEMPTY(&qcc->opening_list) || unlikely(!qcc->largest_bidi_r)) &&
280 !(qcc->flags & QC_CF_APP_SHUT)) {
Amaury Denoyelle30e260e2022-08-03 11:17:57 +0200281 int timeout = px->timeout.httpreq;
282 struct qcs *qcs = NULL;
283 int base_time;
Amaury Denoyelle418ba212022-08-02 15:57:16 +0200284
Amaury Denoyelle30e260e2022-08-03 11:17:57 +0200285 /* Use start time of first stream waiting on HTTP or
286 * qcc idle if no stream not yet used.
287 */
288 if (likely(!LIST_ISEMPTY(&qcc->opening_list)))
289 qcs = LIST_ELEM(qcc->opening_list.n, struct qcs *, el_opening);
290 base_time = qcs ? qcs->start : qcc->idle_start;
291
292 TRACE_DEVEL("waiting on http request", QMUX_EV_QCC_WAKE, qcc->conn, qcs);
293 qcc->task->expire = tick_add_ifset(base_time, timeout);
294 }
295 else {
Amaury Denoyelleeb7d3202023-02-08 15:55:24 +0100296 if (qcc->flags & QC_CF_APP_SHUT) {
297 TRACE_DEVEL("connection in closing", QMUX_EV_QCC_WAKE, qcc->conn);
298 qcc->task->expire = tick_add_ifset(now_ms,
299 qcc->shut_timeout);
300 }
301 else {
302 /* Use http-request timeout if keep-alive timeout not set */
303 int timeout = tick_isset(px->timeout.httpka) ?
304 px->timeout.httpka : px->timeout.httpreq;
305 TRACE_DEVEL("at least one request achieved but none currently in progress", QMUX_EV_QCC_WAKE, qcc->conn);
306 qcc->task->expire = tick_add_ifset(qcc->idle_start, timeout);
307 }
Amaury Denoyelleb3aa07c2023-01-24 18:20:28 +0100308
309 /* If proxy soft-stop in progress and connection is
310 * inactive, close the connection immediately. If a
311 * close-spread-time is configured, randomly spread the
312 * timer over a closing window.
313 */
314 if ((qcc->proxy->flags & (PR_FL_DISABLED|PR_FL_STOPPED)) &&
315 !(global.tune.options & GTUNE_DISABLE_ACTIVE_CLOSE)) {
316
317 /* Wake timeout task immediately if window already expired. */
318 int remaining_window = tick_isset(global.close_spread_end) ?
319 tick_remain(now_ms, global.close_spread_end) : 0;
320
321 TRACE_DEVEL("proxy disabled, prepare connection soft-stop", QMUX_EV_QCC_WAKE, qcc->conn);
322 if (remaining_window) {
323 /* We don't need to reset the expire if it would
324 * already happen before the close window end.
325 */
326 if (!tick_isset(qcc->task->expire) ||
327 tick_is_le(global.close_spread_end, qcc->task->expire)) {
328 /* Set an expire value shorter than the current value
329 * because the close spread window end comes earlier.
330 */
331 qcc->task->expire = tick_add(now_ms,
332 statistical_prng_range(remaining_window));
333 }
334 }
335 else {
336 /* We are past the soft close window end, wake the timeout
337 * task up immediately.
338 */
339 qcc->task->expire = now_ms;
340 task_wakeup(qcc->task, TASK_WOKEN_TIMER);
341 }
342 }
Amaury Denoyelle30e260e2022-08-03 11:17:57 +0200343 }
Amaury Denoyelle418ba212022-08-02 15:57:16 +0200344 }
Amaury Denoyelle6ec98372022-08-01 17:59:38 +0200345
346 /* fallback to default timeout if frontend specific undefined or for
347 * backend connections.
348 */
349 if (!tick_isset(qcc->task->expire)) {
350 TRACE_DEVEL("fallback to default timeout", QMUX_EV_QCC_WAKE, qcc->conn);
351 qcc->task->expire = tick_add_ifset(now_ms, qcc->timeout);
Amaury Denoyelle418ba212022-08-02 15:57:16 +0200352 }
353
Amaury Denoyelle418ba212022-08-02 15:57:16 +0200354 task_queue(qcc->task);
355
356 leave:
357 TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn);
358}
359
Amaury Denoyelle38e60062022-07-01 16:48:42 +0200360/* Mark a stream as open if it was idle. This can be used on every
361 * successful emission/reception operation to update the stream state.
362 */
363static void qcs_idle_open(struct qcs *qcs)
364{
365 /* This operation must not be used if the stream is already closed. */
366 BUG_ON_HOT(qcs->st == QC_SS_CLO);
367
368 if (qcs->st == QC_SS_IDLE) {
Amaury Denoyelle047d86a2022-08-10 16:42:35 +0200369 TRACE_STATE("opening stream", QMUX_EV_QCS_NEW, qcs->qcc->conn, qcs);
Amaury Denoyelle38e60062022-07-01 16:48:42 +0200370 qcs->st = QC_SS_OPEN;
Amaury Denoyelle38e60062022-07-01 16:48:42 +0200371 }
372}
373
374/* Close the local channel of <qcs> instance. */
375static void qcs_close_local(struct qcs *qcs)
376{
Amaury Denoyelle047d86a2022-08-10 16:42:35 +0200377 TRACE_STATE("closing stream locally", QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
378
Amaury Denoyelle38e60062022-07-01 16:48:42 +0200379 /* The stream must have already been opened. */
380 BUG_ON_HOT(qcs->st == QC_SS_IDLE);
381
382 /* This operation cannot be used multiple times. */
383 BUG_ON_HOT(qcs->st == QC_SS_HLOC || qcs->st == QC_SS_CLO);
384
385 if (quic_stream_is_bidi(qcs->id)) {
386 qcs->st = (qcs->st == QC_SS_HREM) ? QC_SS_CLO : QC_SS_HLOC;
Amaury Denoyelleafb7b9d2022-09-19 11:58:24 +0200387
388 if (qcs->flags & QC_SF_HREQ_RECV)
389 qcc_rm_hreq(qcs->qcc);
Amaury Denoyelle38e60062022-07-01 16:48:42 +0200390 }
391 else {
392 /* Only local uni streams are valid for this operation. */
393 BUG_ON_HOT(quic_stream_is_remote(qcs->qcc, qcs->id));
394 qcs->st = QC_SS_CLO;
395 }
Amaury Denoyelle38e60062022-07-01 16:48:42 +0200396}
397
398/* Close the remote channel of <qcs> instance. */
399static void qcs_close_remote(struct qcs *qcs)
400{
Amaury Denoyelle047d86a2022-08-10 16:42:35 +0200401 TRACE_STATE("closing stream remotely", QMUX_EV_QCS_RECV, qcs->qcc->conn, qcs);
402
Amaury Denoyelle38e60062022-07-01 16:48:42 +0200403 /* The stream must have already been opened. */
404 BUG_ON_HOT(qcs->st == QC_SS_IDLE);
405
406 /* This operation cannot be used multiple times. */
407 BUG_ON_HOT(qcs->st == QC_SS_HREM || qcs->st == QC_SS_CLO);
408
409 if (quic_stream_is_bidi(qcs->id)) {
410 qcs->st = (qcs->st == QC_SS_HLOC) ? QC_SS_CLO : QC_SS_HREM;
411 }
412 else {
413 /* Only remote uni streams are valid for this operation. */
414 BUG_ON_HOT(quic_stream_is_local(qcs->qcc, qcs->id));
415 qcs->st = QC_SS_CLO;
416 }
Amaury Denoyelle38e60062022-07-01 16:48:42 +0200417}
418
419static int qcs_is_close_local(struct qcs *qcs)
420{
421 return qcs->st == QC_SS_HLOC || qcs->st == QC_SS_CLO;
422}
423
Amaury Denoyelle6eb3c4b2022-12-09 16:26:03 +0100424static int qcs_is_close_remote(struct qcs *qcs)
Amaury Denoyelle38e60062022-07-01 16:48:42 +0200425{
426 return qcs->st == QC_SS_HREM || qcs->st == QC_SS_CLO;
427}
428
Amaury Denoyelledeed7772021-12-03 11:36:46 +0100429struct buffer *qc_get_buf(struct qcs *qcs, struct buffer *bptr)
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100430{
Amaury Denoyelledeed7772021-12-03 11:36:46 +0100431 struct buffer *buf = b_alloc(bptr);
432 BUG_ON(!buf);
433 return buf;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100434}
435
Amaury Denoyellea441ec92022-07-04 15:48:57 +0200436static struct ncbuf *qc_get_ncbuf(struct qcs *qcs, struct ncbuf *ncbuf)
Amaury Denoyelle1290f1e2022-05-13 14:49:05 +0200437{
438 struct buffer buf = BUF_NULL;
439
440 if (ncb_is_null(ncbuf)) {
441 b_alloc(&buf);
442 BUG_ON(b_is_null(&buf));
443
444 *ncbuf = ncb_make(buf.area, buf.size, 0);
445 ncb_init(ncbuf, 0);
446 }
447
448 return ncbuf;
449}
450
Ilya Shipitsin3b64a282022-07-29 22:26:53 +0500451/* Notify an eventual subscriber on <qcs> or else wakeup up the stconn layer if
Amaury Denoyelle4561f842022-07-06 14:54:34 +0200452 * initialized.
453 */
454static void qcs_alert(struct qcs *qcs)
455{
456 if (qcs->subs) {
457 qcs_notify_recv(qcs);
458 qcs_notify_send(qcs);
459 }
460 else if (qcs_sc(qcs) && qcs->sd->sc->app_ops->wake) {
461 qcs->sd->sc->app_ops->wake(qcs->sd->sc);
462 }
463}
464
Amaury Denoyellea3f222d2021-12-06 11:24:00 +0100465int qcs_subscribe(struct qcs *qcs, int event_type, struct wait_event *es)
466{
Amaury Denoyelle4f137572022-03-24 17:10:00 +0100467 struct qcc *qcc = qcs->qcc;
468
469 TRACE_ENTER(QMUX_EV_STRM_SEND|QMUX_EV_STRM_RECV, qcc->conn, qcs);
Amaury Denoyellea3f222d2021-12-06 11:24:00 +0100470
471 BUG_ON(event_type & ~(SUB_RETRY_SEND|SUB_RETRY_RECV));
472 BUG_ON(qcs->subs && qcs->subs != es);
473
474 es->events |= event_type;
475 qcs->subs = es;
476
Amaury Denoyelle4f137572022-03-24 17:10:00 +0100477 if (event_type & SUB_RETRY_RECV)
478 TRACE_DEVEL("subscribe(recv)", QMUX_EV_STRM_RECV, qcc->conn, qcs);
479
480 if (event_type & SUB_RETRY_SEND)
481 TRACE_DEVEL("subscribe(send)", QMUX_EV_STRM_SEND, qcc->conn, qcs);
482
483 TRACE_LEAVE(QMUX_EV_STRM_SEND|QMUX_EV_STRM_RECV, qcc->conn, qcs);
484
Amaury Denoyellea3f222d2021-12-06 11:24:00 +0100485 return 0;
486}
487
488void qcs_notify_recv(struct qcs *qcs)
489{
490 if (qcs->subs && qcs->subs->events & SUB_RETRY_RECV) {
491 tasklet_wakeup(qcs->subs->tasklet);
492 qcs->subs->events &= ~SUB_RETRY_RECV;
493 if (!qcs->subs->events)
494 qcs->subs = NULL;
495 }
496}
497
498void qcs_notify_send(struct qcs *qcs)
499{
500 if (qcs->subs && qcs->subs->events & SUB_RETRY_SEND) {
501 tasklet_wakeup(qcs->subs->tasklet);
502 qcs->subs->events &= ~SUB_RETRY_SEND;
503 if (!qcs->subs->events)
504 qcs->subs = NULL;
505 }
506}
507
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200508/* Open a locally initiated stream for the connection <qcc>. Set <bidi> for a
509 * bidirectional stream, else an unidirectional stream is opened. The next
510 * available ID on the connection will be used according to the stream type.
511 *
512 * Returns the allocated stream instance or NULL on error.
Amaury Denoyelle8a5b27a2021-12-21 11:53:10 +0100513 */
Amaury Denoyelleb1437232022-07-08 11:53:22 +0200514struct qcs *qcc_init_stream_local(struct qcc *qcc, int bidi)
Amaury Denoyelle8a5b27a2021-12-21 11:53:10 +0100515{
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200516 struct qcs *qcs;
517 enum qcs_type type;
518 uint64_t *next;
Amaury Denoyelle8a5b27a2021-12-21 11:53:10 +0100519
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200520 TRACE_ENTER(QMUX_EV_QCS_NEW, qcc->conn);
521
522 if (bidi) {
523 next = &qcc->next_bidi_l;
524 type = conn_is_back(qcc->conn) ? QCS_CLT_BIDI : QCS_SRV_BIDI;
Amaury Denoyelle8a5b27a2021-12-21 11:53:10 +0100525 }
526 else {
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200527 next = &qcc->next_uni_l;
528 type = conn_is_back(qcc->conn) ? QCS_CLT_UNI : QCS_SRV_UNI;
529 }
530
531 /* TODO ensure that we won't overflow remote peer flow control limit on
532 * streams. Else, we should emit a STREAMS_BLOCKED frame.
533 */
Amaury Denoyelle8a5b27a2021-12-21 11:53:10 +0100534
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200535 qcs = qcs_new(qcc, *next, type);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200536 if (!qcs) {
537 TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn);
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200538 return NULL;
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200539 }
Amaury Denoyellec055e302022-02-07 16:09:06 +0100540
Amaury Denoyelle047d86a2022-08-10 16:42:35 +0200541 TRACE_PROTO("opening local stream", QMUX_EV_QCS_NEW, qcc->conn, qcs);
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200542 *next += 4;
Amaury Denoyelle8a5b27a2021-12-21 11:53:10 +0100543
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200544 TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn, qcs);
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200545 return qcs;
546}
547
548/* Open a remote initiated stream for the connection <qcc> with ID <id>. The
549 * caller is responsible to ensure that a stream with the same ID was not
550 * already opened. This function will also create all intermediaries streams
551 * with ID smaller than <id> not already opened before.
552 *
553 * Returns the allocated stream instance or NULL on error.
554 */
Amaury Denoyelleb1437232022-07-08 11:53:22 +0200555static struct qcs *qcc_init_stream_remote(struct qcc *qcc, uint64_t id)
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200556{
557 struct qcs *qcs = NULL;
558 enum qcs_type type;
Amaury Denoyellebf3c2082022-08-16 11:29:08 +0200559 uint64_t *largest, max_id;
Amaury Denoyelle8a5b27a2021-12-21 11:53:10 +0100560
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200561 TRACE_ENTER(QMUX_EV_QCS_NEW, qcc->conn);
Amaury Denoyelle7272cd72022-03-29 15:15:54 +0200562
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200563 BUG_ON_HOT(quic_stream_is_local(qcc, id));
Amaury Denoyelle8a5b27a2021-12-21 11:53:10 +0100564
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200565 if (quic_stream_is_bidi(id)) {
566 largest = &qcc->largest_bidi_r;
567 type = conn_is_back(qcc->conn) ? QCS_SRV_BIDI : QCS_CLT_BIDI;
568 }
569 else {
570 largest = &qcc->largest_uni_r;
571 type = conn_is_back(qcc->conn) ? QCS_SRV_UNI : QCS_CLT_UNI;
572 }
573
Amaury Denoyellebf3c2082022-08-16 11:29:08 +0200574 /* RFC 9000 4.6. Controlling Concurrency
575 *
576 * An endpoint that receives a frame with a stream ID exceeding the
577 * limit it has sent MUST treat this as a connection error of type
578 * STREAM_LIMIT_ERROR
579 */
580 max_id = quic_stream_is_bidi(id) ? qcc->lfctl.ms_bidi * 4 :
581 qcc->lfctl.ms_uni * 4;
582 if (id >= max_id) {
583 TRACE_ERROR("flow control error", QMUX_EV_QCS_NEW|QMUX_EV_PROTO_ERR, qcc->conn);
584 qcc_emit_cc(qcc, QC_ERR_STREAM_LIMIT_ERROR);
585 goto err;
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200586 }
587
588 /* Only stream ID not already opened can be used. */
589 BUG_ON(id < *largest);
590
591 while (id >= *largest) {
Amaury Denoyellefd79ddb2022-08-16 11:13:45 +0200592 const char *str = *largest < id ? "initializing intermediary remote stream" :
593 "initializing remote stream";
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200594
595 qcs = qcs_new(qcc, *largest, type);
596 if (!qcs) {
Amaury Denoyelle57161b72022-07-07 15:02:32 +0200597 /* TODO emit RESET_STREAM */
Amaury Denoyellec7fb0d22022-08-10 16:39:54 +0200598 TRACE_ERROR("stream fallocation failure", QMUX_EV_QCS_NEW, qcc->conn);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200599 goto err;
Amaury Denoyelle8a5b27a2021-12-21 11:53:10 +0100600 }
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200601
Amaury Denoyelle047d86a2022-08-10 16:42:35 +0200602 TRACE_PROTO(str, QMUX_EV_QCS_NEW, qcc->conn, qcs);
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200603 *largest += 4;
Amaury Denoyelle8a5b27a2021-12-21 11:53:10 +0100604 }
605
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200606 out:
607 TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn, qcs);
Amaury Denoyelle50742292022-03-29 14:57:19 +0200608 return qcs;
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200609
610 err:
611 TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn);
612 return NULL;
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200613}
614
615/* Use this function for a stream <id> which is not in <qcc> stream tree. It
616 * returns true if the associated stream is closed.
617 */
618static int qcc_stream_id_is_closed(struct qcc *qcc, uint64_t id)
619{
620 uint64_t *largest;
621
622 /* This function must only be used for stream not present in the stream tree. */
623 BUG_ON_HOT(eb64_lookup(&qcc->streams_by_id, id));
624
625 if (quic_stream_is_local(qcc, id)) {
626 largest = quic_stream_is_uni(id) ? &qcc->next_uni_l :
627 &qcc->next_bidi_l;
628 }
629 else {
630 largest = quic_stream_is_uni(id) ? &qcc->largest_uni_r :
631 &qcc->largest_bidi_r;
632 }
633
634 return id < *largest;
635}
636
637/* Retrieve the stream instance from <id> ID. This can be used when receiving
638 * STREAM, STREAM_DATA_BLOCKED, RESET_STREAM, MAX_STREAM_DATA or STOP_SENDING
Amaury Denoyelle5fbb8692022-07-06 15:43:21 +0200639 * frames. Set to false <receive_only> or <send_only> if these particular types
Amaury Denoyelle57161b72022-07-07 15:02:32 +0200640 * of streams are not allowed. If the stream instance is found, it is stored in
641 * <out>.
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200642 *
Amaury Denoyelle57161b72022-07-07 15:02:32 +0200643 * Returns 0 on success else non-zero. On error, a RESET_STREAM or a
644 * CONNECTION_CLOSE is automatically emitted. Beware that <out> may be NULL
645 * on success if the stream has already been closed.
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200646 */
Amaury Denoyelle57161b72022-07-07 15:02:32 +0200647int qcc_get_qcs(struct qcc *qcc, uint64_t id, int receive_only, int send_only,
648 struct qcs **out)
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200649{
650 struct eb64_node *node;
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200651
652 TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
Amaury Denoyelle57161b72022-07-07 15:02:32 +0200653 *out = NULL;
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200654
Amaury Denoyelle5fbb8692022-07-06 15:43:21 +0200655 if (!receive_only && quic_stream_is_uni(id) && quic_stream_is_remote(qcc, id)) {
Amaury Denoyellec7fb0d22022-08-10 16:39:54 +0200656 TRACE_ERROR("receive-only stream not allowed", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS|QMUX_EV_PROTO_ERR, qcc->conn, NULL, &id);
Amaury Denoyelle5fbb8692022-07-06 15:43:21 +0200657 qcc_emit_cc(qcc, QC_ERR_STREAM_STATE_ERROR);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200658 goto err;
Amaury Denoyelle5fbb8692022-07-06 15:43:21 +0200659 }
660
661 if (!send_only && quic_stream_is_uni(id) && quic_stream_is_local(qcc, id)) {
Amaury Denoyellec7fb0d22022-08-10 16:39:54 +0200662 TRACE_ERROR("send-only stream not allowed", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS|QMUX_EV_PROTO_ERR, qcc->conn, NULL, &id);
Amaury Denoyelle5fbb8692022-07-06 15:43:21 +0200663 qcc_emit_cc(qcc, QC_ERR_STREAM_STATE_ERROR);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200664 goto err;
Amaury Denoyelle5fbb8692022-07-06 15:43:21 +0200665 }
666
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200667 /* Search the stream in the connection tree. */
668 node = eb64_lookup(&qcc->streams_by_id, id);
669 if (node) {
Amaury Denoyelle57161b72022-07-07 15:02:32 +0200670 *out = eb64_entry(node, struct qcs, by_id);
671 TRACE_DEVEL("using stream from connection tree", QMUX_EV_QCC_RECV, qcc->conn, *out);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200672 goto out;
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200673 }
674
675 /* Check if stream is already closed. */
676 if (qcc_stream_id_is_closed(qcc, id)) {
Amaury Denoyelle047d86a2022-08-10 16:42:35 +0200677 TRACE_DATA("already closed stream", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS, qcc->conn, NULL, &id);
Amaury Denoyelle57161b72022-07-07 15:02:32 +0200678 /* Consider this as a success even if <out> is left NULL. */
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200679 goto out;
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200680 }
681
682 /* Create the stream. This is valid only for remote initiated one. A
Ilya Shipitsin3b64a282022-07-29 22:26:53 +0500683 * local stream must have already been explicitly created by the
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200684 * application protocol layer.
685 */
686 if (quic_stream_is_local(qcc, id)) {
687 /* RFC 9000 19.8. STREAM Frames
688 *
689 * An endpoint MUST terminate the connection with error
690 * STREAM_STATE_ERROR if it receives a STREAM frame for a locally
691 * initiated stream that has not yet been created, or for a send-only
692 * stream.
693 */
Amaury Denoyellec7fb0d22022-08-10 16:39:54 +0200694 TRACE_ERROR("locally initiated stream not yet created", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS|QMUX_EV_PROTO_ERR, qcc->conn, NULL, &id);
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200695 qcc_emit_cc(qcc, QC_ERR_STREAM_STATE_ERROR);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200696 goto err;
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200697 }
698 else {
699 /* Remote stream not found - try to open it. */
Amaury Denoyelle57161b72022-07-07 15:02:32 +0200700 *out = qcc_init_stream_remote(qcc, id);
701 if (!*out) {
Amaury Denoyellec7fb0d22022-08-10 16:39:54 +0200702 TRACE_ERROR("stream creation error", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS, qcc->conn, NULL, &id);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200703 goto err;
Amaury Denoyelle57161b72022-07-07 15:02:32 +0200704 }
Amaury Denoyellea509ffb2022-07-04 15:50:33 +0200705 }
Amaury Denoyelle8a5b27a2021-12-21 11:53:10 +0100706
707 out:
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200708 TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn, *out);
Amaury Denoyelle57161b72022-07-07 15:02:32 +0200709 return 0;
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200710
711 err:
712 TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
713 return 1;
Amaury Denoyelle8a5b27a2021-12-21 11:53:10 +0100714}
715
Amaury Denoyelle62eef852022-06-03 16:40:34 +0200716/* Simple function to duplicate a buffer */
717static inline struct buffer qcs_b_dup(const struct ncbuf *b)
718{
719 return b_make(ncb_orig(b), b->size, b->head, ncb_data(b, 0));
720}
721
Amaury Denoyelle36d4b5e2022-07-01 11:25:40 +0200722/* Remove <bytes> from <qcs> Rx buffer. Flow-control for received offsets may
723 * be allocated for the peer if needed.
Amaury Denoyelle62eef852022-06-03 16:40:34 +0200724 */
725static void qcs_consume(struct qcs *qcs, uint64_t bytes)
726{
727 struct qcc *qcc = qcs->qcc;
728 struct quic_frame *frm;
729 struct ncbuf *buf = &qcs->rx.ncbuf;
730 enum ncb_ret ret;
731
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +0200732 TRACE_ENTER(QMUX_EV_QCS_RECV, qcc->conn, qcs);
733
Amaury Denoyelle62eef852022-06-03 16:40:34 +0200734 ret = ncb_advance(buf, bytes);
735 if (ret) {
736 ABORT_NOW(); /* should not happens because removal only in data */
737 }
738
739 if (ncb_is_empty(buf))
740 qc_free_ncbuf(qcs, buf);
741
742 qcs->rx.offset += bytes;
Amaury Denoyellebb6296c2022-12-09 15:00:17 +0100743 /* Not necessary to emit a MAX_STREAM_DATA if all data received. */
744 if (qcs->flags & QC_SF_SIZE_KNOWN)
745 goto conn_fctl;
746
Amaury Denoyelle62eef852022-06-03 16:40:34 +0200747 if (qcs->rx.msd - qcs->rx.offset < qcs->rx.msd_init / 2) {
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +0200748 TRACE_DATA("increase stream credit via MAX_STREAM_DATA", QMUX_EV_QCS_RECV, qcc->conn, qcs);
Amaury Denoyelle40c24f12023-01-27 17:47:49 +0100749 frm = qc_frm_alloc(QUIC_FT_MAX_STREAM_DATA);
Amaury Denoyelle62eef852022-06-03 16:40:34 +0200750 BUG_ON(!frm); /* TODO handle this properly */
751
752 qcs->rx.msd = qcs->rx.offset + qcs->rx.msd_init;
753
Amaury Denoyelle62eef852022-06-03 16:40:34 +0200754 frm->max_stream_data.id = qcs->id;
755 frm->max_stream_data.max_stream_data = qcs->rx.msd;
756
757 LIST_APPEND(&qcc->lfctl.frms, &frm->list);
758 tasklet_wakeup(qcc->wait_event.tasklet);
759 }
760
Amaury Denoyellebb6296c2022-12-09 15:00:17 +0100761 conn_fctl:
Amaury Denoyelle62eef852022-06-03 16:40:34 +0200762 qcc->lfctl.offsets_consume += bytes;
763 if (qcc->lfctl.md - qcc->lfctl.offsets_consume < qcc->lfctl.md_init / 2) {
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +0200764 TRACE_DATA("increase conn credit via MAX_DATA", QMUX_EV_QCS_RECV, qcc->conn, qcs);
Amaury Denoyelle40c24f12023-01-27 17:47:49 +0100765 frm = qc_frm_alloc(QUIC_FT_MAX_DATA);
Amaury Denoyelle62eef852022-06-03 16:40:34 +0200766 BUG_ON(!frm); /* TODO handle this properly */
767
768 qcc->lfctl.md = qcc->lfctl.offsets_consume + qcc->lfctl.md_init;
769
Amaury Denoyelle62eef852022-06-03 16:40:34 +0200770 frm->max_data.max_data = qcc->lfctl.md;
771
772 LIST_APPEND(&qcs->qcc->lfctl.frms, &frm->list);
773 tasklet_wakeup(qcs->qcc->wait_event.tasklet);
774 }
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +0200775
776 TRACE_LEAVE(QMUX_EV_QCS_RECV, qcc->conn, qcs);
Amaury Denoyelle62eef852022-06-03 16:40:34 +0200777}
778
Amaury Denoyelle3a086402022-05-18 11:38:22 +0200779/* Decode the content of STREAM frames already received on the stream instance
780 * <qcs>.
781 *
782 * Returns 0 on success else non-zero.
783 */
784static int qcc_decode_qcs(struct qcc *qcc, struct qcs *qcs)
785{
Amaury Denoyelle62eef852022-06-03 16:40:34 +0200786 struct buffer b;
Amaury Denoyelle1f21ebd2022-06-07 17:30:55 +0200787 ssize_t ret;
Amaury Denoyelle6befccd2022-07-01 11:26:04 +0200788 int fin = 0;
Amaury Denoyelle62eef852022-06-03 16:40:34 +0200789
Amaury Denoyelle3a086402022-05-18 11:38:22 +0200790 TRACE_ENTER(QMUX_EV_QCS_RECV, qcc->conn, qcs);
791
Amaury Denoyelle62eef852022-06-03 16:40:34 +0200792 b = qcs_b_dup(&qcs->rx.ncbuf);
Amaury Denoyelle6befccd2022-07-01 11:26:04 +0200793
Amaury Denoyelled1310f82022-09-16 13:30:59 +0200794 /* Signal FIN to application if STREAM FIN received with all data. */
795 if (qcs_is_close_remote(qcs))
Amaury Denoyelle6befccd2022-07-01 11:26:04 +0200796 fin = 1;
797
Amaury Denoyelle663e8722022-12-09 14:58:28 +0100798 if (!(qcs->flags & QC_SF_READ_ABORTED)) {
799 ret = qcc->app_ops->decode_qcs(qcs, &b, fin);
800 if (ret < 0) {
801 TRACE_ERROR("decoding error", QMUX_EV_QCS_RECV, qcc->conn, qcs);
802 goto err;
803 }
804 }
805 else {
806 TRACE_DATA("ignore read on stream", QMUX_EV_QCS_RECV, qcc->conn, qcs);
807 ret = b_data(&b);
Amaury Denoyelle3a086402022-05-18 11:38:22 +0200808 }
809
Amaury Denoyelle381d8132023-02-17 09:51:20 +0100810 if (ret)
Amaury Denoyelle1f21ebd2022-06-07 17:30:55 +0200811 qcs_consume(qcs, ret);
Amaury Denoyelle381d8132023-02-17 09:51:20 +0100812 if (ret || (!b_data(&b) && fin))
Amaury Denoyelle62eef852022-06-03 16:40:34 +0200813 qcs_notify_recv(qcs);
Amaury Denoyelle3a086402022-05-18 11:38:22 +0200814
815 TRACE_LEAVE(QMUX_EV_QCS_RECV, qcc->conn, qcs);
Amaury Denoyelle3a086402022-05-18 11:38:22 +0200816 return 0;
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200817
818 err:
819 TRACE_LEAVE(QMUX_EV_QCS_RECV, qcc->conn, qcs);
820 return 1;
Amaury Denoyelle3a086402022-05-18 11:38:22 +0200821}
822
Amaury Denoyellef9e190e2022-05-23 16:12:15 +0200823/* Emit a CONNECTION_CLOSE_APP with error <err>. Reserved for application error
Amaury Denoyelled666d742022-07-13 15:15:58 +0200824 * code. To close the connection right away, set <immediate> : this is useful
825 * when dealing with a connection fatal error. Else a graceful shutdown will be
826 * conducted : the error-code is only registered. The lower layer is
827 * responsible to close the connection when deemed suitable. Note that in this
828 * case the error code might be overwritten if an immediate close is requested
829 * in the interval.
Amaury Denoyellef9e190e2022-05-23 16:12:15 +0200830 */
Amaury Denoyelled666d742022-07-13 15:15:58 +0200831void qcc_emit_cc_app(struct qcc *qcc, int err, int immediate)
Amaury Denoyellef9e190e2022-05-23 16:12:15 +0200832{
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +0200833 TRACE_ENTER(QMUX_EV_QCC_END, qcc->conn);
834
Amaury Denoyelled666d742022-07-13 15:15:58 +0200835 if (immediate) {
836 quic_set_connection_close(qcc->conn->handle.qc, quic_err_app(err));
837 qcc->flags |= QC_CF_CC_EMIT;
838 tasklet_wakeup(qcc->wait_event.tasklet);
839 }
840 else {
841 /* Only register the error code for graceful shutdown. */
842 qcc->conn->handle.qc->err = quic_err_app(err);
843 }
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +0200844
845 TRACE_LEAVE(QMUX_EV_QCC_END, qcc->conn);
Amaury Denoyelle843a1192022-07-04 11:44:38 +0200846}
847
848/* Prepare for the emission of RESET_STREAM on <qcs> with error code <err>. */
849void qcc_reset_stream(struct qcs *qcs, int err)
850{
851 struct qcc *qcc = qcs->qcc;
852
853 if ((qcs->flags & QC_SF_TO_RESET) || qcs_is_close_local(qcs))
854 return;
855
Amaury Denoyelle047d86a2022-08-10 16:42:35 +0200856 TRACE_STATE("reset stream", QMUX_EV_QCS_END, qcc->conn, qcs);
Amaury Denoyelle843a1192022-07-04 11:44:38 +0200857 qcs->flags |= QC_SF_TO_RESET;
858 qcs->err = err;
Amaury Denoyelle0a1154a2023-01-06 17:43:11 +0100859
860 qcc_send_stream(qcs, 1);
Amaury Denoyelle843a1192022-07-04 11:44:38 +0200861 tasklet_wakeup(qcc->wait_event.tasklet);
Amaury Denoyelle663e8722022-12-09 14:58:28 +0100862}
863
Amaury Denoyellef9b03262023-01-09 10:34:25 +0100864/* Register <qcs> stream for emission of STREAM, STOP_SENDING or RESET_STREAM.
865 * Set <urg> to 1 if stream content should be treated in priority compared to
866 * other streams.
867 */
868void qcc_send_stream(struct qcs *qcs, int urg)
Amaury Denoyelle20f2a422023-01-03 14:39:24 +0100869{
870 struct qcc *qcc = qcs->qcc;
871
872 TRACE_ENTER(QMUX_EV_QCS_SEND, qcc->conn, qcs);
873
874 /* Cannot send if already closed. */
875 BUG_ON(qcs_is_close_local(qcs));
876
Amaury Denoyellef9b03262023-01-09 10:34:25 +0100877 if (urg) {
878 LIST_DEL_INIT(&qcs->el_send);
879 LIST_INSERT(&qcc->send_list, &qcs->el_send);
880 }
881 else {
882 if (!LIST_INLIST(&qcs->el_send))
883 LIST_APPEND(&qcs->qcc->send_list, &qcs->el_send);
884 }
Amaury Denoyelle20f2a422023-01-03 14:39:24 +0100885
886 TRACE_LEAVE(QMUX_EV_QCS_SEND, qcc->conn, qcs);
887}
888
Amaury Denoyelle663e8722022-12-09 14:58:28 +0100889/* Prepare for the emission of STOP_SENDING on <qcs>. */
890void qcc_abort_stream_read(struct qcs *qcs)
891{
892 struct qcc *qcc = qcs->qcc;
893
894 TRACE_ENTER(QMUX_EV_QCC_NEW, qcc->conn, qcs);
895
896 if ((qcs->flags & QC_SF_TO_STOP_SENDING) || qcs_is_close_remote(qcs))
897 goto end;
898
899 TRACE_STATE("abort stream read", QMUX_EV_QCS_END, qcc->conn, qcs);
900 qcs->flags |= (QC_SF_TO_STOP_SENDING|QC_SF_READ_ABORTED);
Amaury Denoyelle0a1154a2023-01-06 17:43:11 +0100901
902 qcc_send_stream(qcs, 1);
Amaury Denoyelle663e8722022-12-09 14:58:28 +0100903 tasklet_wakeup(qcc->wait_event.tasklet);
904
905 end:
906 TRACE_LEAVE(QMUX_EV_QCC_NEW, qcc->conn, qcs);
Amaury Denoyellef9e190e2022-05-23 16:12:15 +0200907}
908
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +0200909/* Install the <app_ops> applicative layer of a QUIC connection on mux <qcc>.
910 * Returns 0 on success else non-zero.
911 */
912int qcc_install_app_ops(struct qcc *qcc, const struct qcc_app_ops *app_ops)
913{
914 TRACE_ENTER(QMUX_EV_QCC_NEW, qcc->conn);
915
Amaury Denoyelleb4d119f2023-01-25 17:44:36 +0100916 if (app_ops->init && !app_ops->init(qcc)) {
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +0200917 TRACE_ERROR("app ops init error", QMUX_EV_QCC_NEW, qcc->conn);
918 goto err;
919 }
920
921 TRACE_PROTO("application layer initialized", QMUX_EV_QCC_NEW, qcc->conn);
Amaury Denoyelleb4d119f2023-01-25 17:44:36 +0100922 qcc->app_ops = app_ops;
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +0200923
Amaury Denoyelle71fd0362023-01-24 17:35:37 +0100924 /* RFC 9114 7.2.4.2. Initialization
925 *
926 * Endpoints MUST NOT require any data to be
927 * received from the peer prior to sending the SETTINGS frame;
928 * settings MUST be sent as soon as the transport is ready to
929 * send data.
930 */
931 if (qcc->app_ops->finalize) {
932 if (qcc->app_ops->finalize(qcc->ctx)) {
933 TRACE_ERROR("app ops finalize error", QMUX_EV_QCC_NEW, qcc->conn);
934 goto err;
935 }
936 tasklet_wakeup(qcc->wait_event.tasklet);
937 }
938
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +0200939 TRACE_LEAVE(QMUX_EV_QCC_NEW, qcc->conn);
940 return 0;
941
942 err:
943 TRACE_LEAVE(QMUX_EV_QCC_NEW, qcc->conn);
944 return 1;
945}
946
Amaury Denoyelle3a086402022-05-18 11:38:22 +0200947/* Handle a new STREAM frame for stream with id <id>. Payload is pointed by
948 * <data> with length <len> and represents the offset <offset>. <fin> is set if
949 * the QUIC frame FIN bit is set.
Amaury Denoyelle0e3010b2022-02-28 11:37:48 +0100950 *
Amaury Denoyelle57161b72022-07-07 15:02:32 +0200951 * Returns 0 on success else non-zero. On error, the received frame should not
952 * be acknowledged.
Amaury Denoyelle0e3010b2022-02-28 11:37:48 +0100953 */
954int qcc_recv(struct qcc *qcc, uint64_t id, uint64_t len, uint64_t offset,
Amaury Denoyelle3a086402022-05-18 11:38:22 +0200955 char fin, char *data)
Amaury Denoyelle0e3010b2022-02-28 11:37:48 +0100956{
957 struct qcs *qcs;
Amaury Denoyelle1290f1e2022-05-13 14:49:05 +0200958 enum ncb_ret ret;
Amaury Denoyelle0e3010b2022-02-28 11:37:48 +0100959
Amaury Denoyelle4f137572022-03-24 17:10:00 +0100960 TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
961
Amaury Denoyelle5c4373a2022-05-24 14:47:48 +0200962 if (qcc->flags & QC_CF_CC_EMIT) {
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200963 TRACE_DATA("connection closed", QMUX_EV_QCC_RECV, qcc->conn);
964 goto err;
Amaury Denoyelle5c4373a2022-05-24 14:47:48 +0200965 }
966
Amaury Denoyelle6754d7e2022-05-23 16:12:49 +0200967 /* RFC 9000 19.8. STREAM Frames
968 *
969 * An endpoint MUST terminate the connection with error
970 * STREAM_STATE_ERROR if it receives a STREAM frame for a locally
971 * initiated stream that has not yet been created, or for a send-only
972 * stream.
973 */
Amaury Denoyelle57161b72022-07-07 15:02:32 +0200974 if (qcc_get_qcs(qcc, id, 1, 0, &qcs)) {
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200975 TRACE_DATA("qcs retrieval error", QMUX_EV_QCC_RECV, qcc->conn);
976 goto err;
Amaury Denoyelle57161b72022-07-07 15:02:32 +0200977 }
978
979 if (!qcs) {
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200980 TRACE_DATA("already closed stream", QMUX_EV_QCC_RECV, qcc->conn);
981 goto out;
Amaury Denoyelle57161b72022-07-07 15:02:32 +0200982 }
Amaury Denoyelle0e3010b2022-02-28 11:37:48 +0100983
Amaury Denoyellebf91e392022-07-04 10:02:04 +0200984 /* RFC 9000 4.5. Stream Final Size
985 *
986 * Once a final size for a stream is known, it cannot change. If a
987 * RESET_STREAM or STREAM frame is received indicating a change in the
988 * final size for the stream, an endpoint SHOULD respond with an error
989 * of type FINAL_SIZE_ERROR; see Section 11 for details on error
990 * handling.
991 */
992 if (qcs->flags & QC_SF_SIZE_KNOWN &&
993 (offset + len > qcs->rx.offset_max || (fin && offset + len < qcs->rx.offset_max))) {
Amaury Denoyellec7fb0d22022-08-10 16:39:54 +0200994 TRACE_ERROR("final size error", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV|QMUX_EV_PROTO_ERR, qcc->conn, qcs);
Amaury Denoyellebf91e392022-07-04 10:02:04 +0200995 qcc_emit_cc(qcc, QC_ERR_FINAL_SIZE_ERROR);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +0200996 goto err;
Amaury Denoyellebf91e392022-07-04 10:02:04 +0200997 }
998
Amaury Denoyelle5854fc02022-12-09 16:25:48 +0100999 if (qcs_is_close_remote(qcs)) {
1000 TRACE_DATA("skipping STREAM for remotely closed", QMUX_EV_QCC_RECV, qcc->conn);
1001 goto out;
1002 }
1003
Amaury Denoyellefa241932023-02-14 15:36:36 +01001004 if (offset + len < qcs->rx.offset ||
1005 (offset + len == qcs->rx.offset && (!fin || (qcs->flags & QC_SF_SIZE_KNOWN)))) {
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001006 TRACE_DATA("already received offset", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
1007 goto out;
Amaury Denoyelle0e3010b2022-02-28 11:37:48 +01001008 }
1009
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02001010 TRACE_PROTO("receiving STREAM", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
Amaury Denoyelle38e60062022-07-01 16:48:42 +02001011 qcs_idle_open(qcs);
1012
Amaury Denoyelled46b0f52022-05-20 15:05:07 +02001013 if (offset + len > qcs->rx.offset_max) {
1014 uint64_t diff = offset + len - qcs->rx.offset_max;
1015 qcs->rx.offset_max = offset + len;
1016 qcc->lfctl.offsets_recv += diff;
1017
1018 if (offset + len > qcs->rx.msd ||
1019 qcc->lfctl.offsets_recv > qcc->lfctl.md) {
1020 /* RFC 9000 4.1. Data Flow Control
1021 *
1022 * A receiver MUST close the connection with an error
1023 * of type FLOW_CONTROL_ERROR if the sender violates
1024 * the advertised connection or stream data limits
1025 */
Amaury Denoyellec7fb0d22022-08-10 16:39:54 +02001026 TRACE_ERROR("flow control error", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV|QMUX_EV_PROTO_ERR,
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001027 qcc->conn, qcs);
Amaury Denoyelled46b0f52022-05-20 15:05:07 +02001028 qcc_emit_cc(qcc, QC_ERR_FLOW_CONTROL_ERROR);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001029 goto err;
Amaury Denoyelled46b0f52022-05-20 15:05:07 +02001030 }
1031 }
Amaury Denoyelle0e3010b2022-02-28 11:37:48 +01001032
Amaury Denoyelle1290f1e2022-05-13 14:49:05 +02001033 if (!qc_get_ncbuf(qcs, &qcs->rx.ncbuf) || ncb_is_null(&qcs->rx.ncbuf)) {
Amaury Denoyelle0e3010b2022-02-28 11:37:48 +01001034 /* TODO should mark qcs as full */
Amaury Denoyelle1290f1e2022-05-13 14:49:05 +02001035 ABORT_NOW();
1036 return 1;
Amaury Denoyelle0e3010b2022-02-28 11:37:48 +01001037 }
1038
Amaury Denoyelle047d86a2022-08-10 16:42:35 +02001039 TRACE_DATA("newly received offset", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
Amaury Denoyelle1290f1e2022-05-13 14:49:05 +02001040 if (offset < qcs->rx.offset) {
Frédéric Lécaillea18c3332022-07-04 09:54:58 +02001041 size_t diff = qcs->rx.offset - offset;
1042
1043 len -= diff;
1044 data += diff;
Amaury Denoyelle1290f1e2022-05-13 14:49:05 +02001045 offset = qcs->rx.offset;
1046 }
Amaury Denoyelle0e3010b2022-02-28 11:37:48 +01001047
Amaury Denoyellefa241932023-02-14 15:36:36 +01001048 if (len) {
1049 ret = ncb_add(&qcs->rx.ncbuf, offset - qcs->rx.offset, data, len, NCB_ADD_COMPARE);
1050 switch (ret) {
1051 case NCB_RET_OK:
1052 break;
1053
1054 case NCB_RET_DATA_REJ:
Amaury Denoyellecc3d7162022-05-20 15:14:57 +02001055 /* RFC 9000 2.2. Sending and Receiving Data
1056 *
1057 * An endpoint could receive data for a stream at the
1058 * same stream offset multiple times. Data that has
1059 * already been received can be discarded. The data at
1060 * a given offset MUST NOT change if it is sent
1061 * multiple times; an endpoint MAY treat receipt of
1062 * different data at the same offset within a stream as
1063 * a connection error of type PROTOCOL_VIOLATION.
1064 */
Amaury Denoyellec7fb0d22022-08-10 16:39:54 +02001065 TRACE_ERROR("overlapping data rejected", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV|QMUX_EV_PROTO_ERR,
Amaury Denoyelle1290f1e2022-05-13 14:49:05 +02001066 qcc->conn, qcs);
Amaury Denoyellecc3d7162022-05-20 15:14:57 +02001067 qcc_emit_cc(qcc, QC_ERR_PROTOCOL_VIOLATION);
Amaury Denoyellefa241932023-02-14 15:36:36 +01001068 return 1;
1069
1070 case NCB_RET_GAP_SIZE:
Amaury Denoyelle047d86a2022-08-10 16:42:35 +02001071 TRACE_DATA("cannot bufferize frame due to gap size limit", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV,
1072 qcc->conn, qcs);
Amaury Denoyellefa241932023-02-14 15:36:36 +01001073 return 1;
Amaury Denoyelle1290f1e2022-05-13 14:49:05 +02001074 }
Amaury Denoyelle1290f1e2022-05-13 14:49:05 +02001075 }
Amaury Denoyelle0e3010b2022-02-28 11:37:48 +01001076
1077 if (fin)
Amaury Denoyelle3f39b402022-07-01 16:11:03 +02001078 qcs->flags |= QC_SF_SIZE_KNOWN;
Amaury Denoyelle0e3010b2022-02-28 11:37:48 +01001079
Amaury Denoyelled1310f82022-09-16 13:30:59 +02001080 if (qcs->flags & QC_SF_SIZE_KNOWN &&
1081 qcs->rx.offset_max == qcs->rx.offset + ncb_data(&qcs->rx.ncbuf, 0)) {
Amaury Denoyelle38e60062022-07-01 16:48:42 +02001082 qcs_close_remote(qcs);
Amaury Denoyelled1310f82022-09-16 13:30:59 +02001083 }
Amaury Denoyelle38e60062022-07-01 16:48:42 +02001084
Amaury Denoyellefa241932023-02-14 15:36:36 +01001085 if ((ncb_data(&qcs->rx.ncbuf, 0) && !(qcs->flags & QC_SF_DEM_FULL)) || fin) {
Amaury Denoyelle3a086402022-05-18 11:38:22 +02001086 qcc_decode_qcs(qcc, qcs);
Amaury Denoyelle418ba212022-08-02 15:57:16 +02001087 qcc_refresh_timeout(qcc);
1088 }
Amaury Denoyelle3a086402022-05-18 11:38:22 +02001089
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001090 out:
Amaury Denoyelle4f137572022-03-24 17:10:00 +01001091 TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
Amaury Denoyelle0e3010b2022-02-28 11:37:48 +01001092 return 0;
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001093
1094 err:
1095 TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
1096 return 1;
Amaury Denoyelle0e3010b2022-02-28 11:37:48 +01001097}
1098
Amaury Denoyelle1e5e5132022-03-08 16:23:03 +01001099/* Handle a new MAX_DATA frame. <max> must contains the maximum data field of
1100 * the frame.
1101 *
1102 * Returns 0 on success else non-zero.
1103 */
1104int qcc_recv_max_data(struct qcc *qcc, uint64_t max)
1105{
Amaury Denoyelle392e94e2022-07-06 15:44:16 +02001106 TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
1107
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02001108 TRACE_PROTO("receiving MAX_DATA", QMUX_EV_QCC_RECV, qcc->conn);
Amaury Denoyelle1e5e5132022-03-08 16:23:03 +01001109 if (qcc->rfctl.md < max) {
1110 qcc->rfctl.md = max;
Amaury Denoyelle392e94e2022-07-06 15:44:16 +02001111 TRACE_DEVEL("increase remote max-data", QMUX_EV_QCC_RECV, qcc->conn);
Amaury Denoyelle1e5e5132022-03-08 16:23:03 +01001112
1113 if (qcc->flags & QC_CF_BLK_MFCTL) {
1114 qcc->flags &= ~QC_CF_BLK_MFCTL;
1115 tasklet_wakeup(qcc->wait_event.tasklet);
1116 }
1117 }
Amaury Denoyelle392e94e2022-07-06 15:44:16 +02001118
1119 TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
Amaury Denoyelle1e5e5132022-03-08 16:23:03 +01001120 return 0;
1121}
1122
Amaury Denoyelle8727ff42022-03-08 10:39:55 +01001123/* Handle a new MAX_STREAM_DATA frame. <max> must contains the maximum data
1124 * field of the frame and <id> is the identifier of the QUIC stream.
1125 *
Amaury Denoyelleb68559a2022-07-06 15:45:20 +02001126 * Returns 0 on success else non-zero. On error, the received frame should not
1127 * be acknowledged.
Amaury Denoyelle8727ff42022-03-08 10:39:55 +01001128 */
1129int qcc_recv_max_stream_data(struct qcc *qcc, uint64_t id, uint64_t max)
1130{
1131 struct qcs *qcs;
Amaury Denoyelle8727ff42022-03-08 10:39:55 +01001132
Amaury Denoyelle392e94e2022-07-06 15:44:16 +02001133 TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
1134
Amaury Denoyelleb68559a2022-07-06 15:45:20 +02001135 /* RFC 9000 19.10. MAX_STREAM_DATA Frames
1136 *
1137 * Receiving a MAX_STREAM_DATA frame for a locally
1138 * initiated stream that has not yet been created MUST be treated as a
1139 * connection error of type STREAM_STATE_ERROR. An endpoint that
1140 * receives a MAX_STREAM_DATA frame for a receive-only stream MUST
1141 * terminate the connection with error STREAM_STATE_ERROR.
1142 */
1143 if (qcc_get_qcs(qcc, id, 0, 1, &qcs)) {
1144 TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
1145 return 1;
1146 }
1147
1148 if (qcs) {
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02001149 TRACE_PROTO("receiving MAX_STREAM_DATA", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
Amaury Denoyelle8727ff42022-03-08 10:39:55 +01001150 if (max > qcs->tx.msd) {
1151 qcs->tx.msd = max;
Amaury Denoyelle392e94e2022-07-06 15:44:16 +02001152 TRACE_DEVEL("increase remote max-stream-data", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
Amaury Denoyelle8727ff42022-03-08 10:39:55 +01001153
1154 if (qcs->flags & QC_SF_BLK_SFCTL) {
1155 qcs->flags &= ~QC_SF_BLK_SFCTL;
Amaury Denoyelle20f2a422023-01-03 14:39:24 +01001156 /* TODO optim: only wakeup IO-CB if stream has data to sent. */
Amaury Denoyelle8727ff42022-03-08 10:39:55 +01001157 tasklet_wakeup(qcc->wait_event.tasklet);
1158 }
1159 }
1160 }
1161
Amaury Denoyelle30e260e2022-08-03 11:17:57 +02001162 if (qcc_may_expire(qcc) && !qcc->nb_hreq)
1163 qcc_refresh_timeout(qcc);
1164
Amaury Denoyelle5854fc02022-12-09 16:25:48 +01001165 TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
1166 return 0;
1167}
1168
1169/* Handle a new RESET_STREAM frame from stream ID <id> with error code <err>
1170 * and final stream size <final_size>.
1171 *
1172 * Returns 0 on success else non-zero. On error, the received frame should not
1173 * be acknowledged.
1174 */
1175int qcc_recv_reset_stream(struct qcc *qcc, uint64_t id, uint64_t err, uint64_t final_size)
1176{
1177 struct qcs *qcs;
1178
1179 TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
1180
1181 /* RFC 9000 19.4. RESET_STREAM Frames
1182 *
1183 * An endpoint that receives a RESET_STREAM frame for a send-only stream
1184 * MUST terminate the connection with error STREAM_STATE_ERROR.
1185 */
1186 if (qcc_get_qcs(qcc, id, 1, 0, &qcs)) {
1187 TRACE_ERROR("RESET_STREAM for send-only stream received", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
1188 qcc_emit_cc(qcc, QC_ERR_STREAM_STATE_ERROR);
1189 goto err;
1190 }
1191
1192 if (!qcs || qcs_is_close_remote(qcs))
1193 goto out;
1194
1195 TRACE_PROTO("receiving RESET_STREAM", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
1196 qcs_idle_open(qcs);
1197
Amaury Denoyellee269aeb2023-01-30 12:13:22 +01001198 if (qcc->app_ops->close) {
1199 if (qcc->app_ops->close(qcs, QCC_APP_OPS_CLOSE_SIDE_RD)) {
1200 TRACE_ERROR("closure rejected by app layer", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
1201 goto out;
1202 }
1203 }
1204
Amaury Denoyelle5854fc02022-12-09 16:25:48 +01001205 if (qcs->rx.offset_max > final_size ||
1206 ((qcs->flags & QC_SF_SIZE_KNOWN) && qcs->rx.offset_max != final_size)) {
1207 TRACE_ERROR("final size error on RESET_STREAM", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
1208 qcc_emit_cc(qcc, QC_ERR_FINAL_SIZE_ERROR);
1209 goto err;
1210 }
1211
1212 qcs->flags |= QC_SF_SIZE_KNOWN;
1213 qcs_close_remote(qcs);
1214 qc_free_ncbuf(qcs, &qcs->rx.ncbuf);
1215
1216 if (qcs_sc(qcs)) {
1217 se_fl_set_error(qcs->sd);
1218 qcs_alert(qcs);
1219 }
1220
1221 out:
Amaury Denoyelle392e94e2022-07-06 15:44:16 +02001222 TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
Amaury Denoyelle8727ff42022-03-08 10:39:55 +01001223 return 0;
Amaury Denoyelle5854fc02022-12-09 16:25:48 +01001224
1225 err:
1226 TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
1227 return 1;
Amaury Denoyelle8727ff42022-03-08 10:39:55 +01001228}
1229
Amaury Denoyellea5b50752022-07-04 11:44:53 +02001230/* Handle a new STOP_SENDING frame for stream ID <id>. The error code should be
1231 * specified in <err>.
1232 *
1233 * Returns 0 on success else non-zero. On error, the received frame should not
1234 * be acknowledged.
1235 */
1236int qcc_recv_stop_sending(struct qcc *qcc, uint64_t id, uint64_t err)
1237{
1238 struct qcs *qcs;
1239
1240 TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
1241
1242 /* RFC 9000 19.5. STOP_SENDING Frames
1243 *
1244 * Receiving a STOP_SENDING frame for a
1245 * locally initiated stream that has not yet been created MUST be
1246 * treated as a connection error of type STREAM_STATE_ERROR. An
1247 * endpoint that receives a STOP_SENDING frame for a receive-only stream
1248 * MUST terminate the connection with error STREAM_STATE_ERROR.
1249 */
1250 if (qcc_get_qcs(qcc, id, 0, 1, &qcs)) {
1251 TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
1252 return 1;
1253 }
1254
1255 if (!qcs)
1256 goto out;
1257
Amaury Denoyelle047d86a2022-08-10 16:42:35 +02001258 TRACE_PROTO("receiving STOP_SENDING", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
Amaury Denoyelled7755372022-10-03 17:20:31 +02001259
1260 /* RFC 9000 3.5. Solicited State Transitions
1261 *
1262 * An endpoint is expected to send another STOP_SENDING frame if a
1263 * packet containing a previous STOP_SENDING is lost. However, once
1264 * either all stream data or a RESET_STREAM frame has been received for
1265 * the stream -- that is, the stream is in any state other than "Recv"
1266 * or "Size Known" -- sending a STOP_SENDING frame is unnecessary.
1267 */
1268
1269 /* TODO thanks to previous RFC clause, STOP_SENDING is ignored if current stream
1270 * has already been closed locally. This is useful to not emit multiple
1271 * RESET_STREAM for a single stream. This is functional if stream is
1272 * locally closed due to all data transmitted, but in this case the RFC
1273 * advices to use an explicit RESET_STREAM.
1274 */
1275 if (qcs_is_close_local(qcs)) {
1276 TRACE_STATE("ignoring STOP_SENDING", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
1277 goto out;
1278 }
1279
Amaury Denoyelle96ca1b72022-08-09 17:36:38 +02001280 qcs_idle_open(qcs);
1281
Amaury Denoyelle87f87662023-01-30 12:12:43 +01001282 if (qcc->app_ops->close) {
1283 if (qcc->app_ops->close(qcs, QCC_APP_OPS_CLOSE_SIDE_WR)) {
1284 TRACE_ERROR("closure rejected by app layer", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
1285 goto out;
1286 }
1287 }
1288
Amaury Denoyellea5b50752022-07-04 11:44:53 +02001289 /* RFC 9000 3.5. Solicited State Transitions
1290 *
1291 * An endpoint that receives a STOP_SENDING frame
1292 * MUST send a RESET_STREAM frame if the stream is in the "Ready" or
1293 * "Send" state. If the stream is in the "Data Sent" state, the
1294 * endpoint MAY defer sending the RESET_STREAM frame until the packets
1295 * containing outstanding data are acknowledged or declared lost. If
1296 * any outstanding data is declared lost, the endpoint SHOULD send a
1297 * RESET_STREAM frame instead of retransmitting the data.
1298 *
1299 * An endpoint SHOULD copy the error code from the STOP_SENDING frame to
1300 * the RESET_STREAM frame it sends, but it can use any application error
1301 * code.
1302 */
Amaury Denoyellea5b50752022-07-04 11:44:53 +02001303 qcc_reset_stream(qcs, err);
1304
Amaury Denoyelle30e260e2022-08-03 11:17:57 +02001305 if (qcc_may_expire(qcc) && !qcc->nb_hreq)
1306 qcc_refresh_timeout(qcc);
1307
Amaury Denoyellea5b50752022-07-04 11:44:53 +02001308 out:
1309 TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
1310 return 0;
1311}
1312
Amaury Denoyellec985cb12022-05-16 14:29:59 +02001313/* Signal the closing of remote stream with id <id>. Flow-control for new
1314 * streams may be allocated for the peer if needed.
1315 */
1316static int qcc_release_remote_stream(struct qcc *qcc, uint64_t id)
Amaury Denoyellec055e302022-02-07 16:09:06 +01001317{
Amaury Denoyellec985cb12022-05-16 14:29:59 +02001318 struct quic_frame *frm;
1319
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02001320 TRACE_ENTER(QMUX_EV_QCS_END, qcc->conn);
1321
Amaury Denoyellec985cb12022-05-16 14:29:59 +02001322 if (quic_stream_is_bidi(id)) {
1323 ++qcc->lfctl.cl_bidi_r;
1324 if (qcc->lfctl.cl_bidi_r > qcc->lfctl.ms_bidi_init / 2) {
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02001325 TRACE_DATA("increase max stream limit with MAX_STREAMS_BIDI", QMUX_EV_QCC_SEND, qcc->conn);
Amaury Denoyelle40c24f12023-01-27 17:47:49 +01001326 frm = qc_frm_alloc(QUIC_FT_MAX_STREAMS_BIDI);
Amaury Denoyellec985cb12022-05-16 14:29:59 +02001327 BUG_ON(!frm); /* TODO handle this properly */
1328
Amaury Denoyellec985cb12022-05-16 14:29:59 +02001329 frm->max_streams_bidi.max_streams = qcc->lfctl.ms_bidi +
1330 qcc->lfctl.cl_bidi_r;
1331 LIST_APPEND(&qcc->lfctl.frms, &frm->list);
1332 tasklet_wakeup(qcc->wait_event.tasklet);
1333
1334 qcc->lfctl.ms_bidi += qcc->lfctl.cl_bidi_r;
1335 qcc->lfctl.cl_bidi_r = 0;
1336 }
1337 }
1338 else {
Amaury Denoyelle91077312022-12-22 18:56:09 +01001339 /* TODO unidirectional stream flow control with MAX_STREAMS_UNI
1340 * emission not implemented. It should be unnecessary for
1341 * HTTP/3 but may be required if other application protocols
1342 * are supported.
Amaury Denoyellebf3c2082022-08-16 11:29:08 +02001343 */
Amaury Denoyellec985cb12022-05-16 14:29:59 +02001344 }
1345
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02001346 TRACE_LEAVE(QMUX_EV_QCS_END, qcc->conn);
1347
Amaury Denoyellec985cb12022-05-16 14:29:59 +02001348 return 0;
Amaury Denoyellec055e302022-02-07 16:09:06 +01001349}
1350
Ilya Shipitsin5e87bcf2021-12-25 11:45:52 +05001351/* detaches the QUIC stream from its QCC and releases it to the QCS pool. */
Amaury Denoyelle2873a312021-12-08 14:42:55 +01001352static void qcs_destroy(struct qcs *qcs)
1353{
Amaury Denoyelle4f137572022-03-24 17:10:00 +01001354 struct connection *conn = qcs->qcc->conn;
Amaury Denoyelled8e680c2022-03-29 15:18:44 +02001355 const uint64_t id = qcs->id;
Amaury Denoyellec055e302022-02-07 16:09:06 +01001356
Amaury Denoyelle4f137572022-03-24 17:10:00 +01001357 TRACE_ENTER(QMUX_EV_QCS_END, conn, qcs);
Amaury Denoyelle2873a312021-12-08 14:42:55 +01001358
Amaury Denoyellec985cb12022-05-16 14:29:59 +02001359 if (quic_stream_is_remote(qcs->qcc, id))
1360 qcc_release_remote_stream(qcs->qcc, id);
Amaury Denoyellec055e302022-02-07 16:09:06 +01001361
Amaury Denoyelledccbd732022-03-29 18:36:59 +02001362 qcs_free(qcs);
Amaury Denoyelle4f137572022-03-24 17:10:00 +01001363
1364 TRACE_LEAVE(QMUX_EV_QCS_END, conn);
Amaury Denoyelle2873a312021-12-08 14:42:55 +01001365}
1366
Amaury Denoyelleb9e06402022-06-10 15:16:21 +02001367/* Transfer as much as possible data on <qcs> from <in> to <out>. This is done
1368 * in respect with available flow-control at stream and connection level.
Amaury Denoyelle75d14ad2022-03-22 15:10:29 +01001369 *
Amaury Denoyellefe8f5552022-04-27 16:44:49 +02001370 * Returns the total bytes of transferred data.
Amaury Denoyelle75d14ad2022-03-22 15:10:29 +01001371 */
Amaury Denoyelleb9e06402022-06-10 15:16:21 +02001372static int qcs_xfer_data(struct qcs *qcs, struct buffer *out, struct buffer *in)
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001373{
Amaury Denoyelle05ce55e2022-03-08 10:35:42 +01001374 struct qcc *qcc = qcs->qcc;
Amaury Denoyelleda6ad202022-04-12 11:41:04 +02001375 int left, to_xfer;
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001376 int total = 0;
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001377
Amaury Denoyelle4f137572022-03-24 17:10:00 +01001378 TRACE_ENTER(QMUX_EV_QCS_SEND, qcc->conn, qcs);
Amaury Denoyelledeed7772021-12-03 11:36:46 +01001379
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001380 qc_get_buf(qcs, out);
1381
1382 /*
1383 * QCS out buffer diagram
1384 * head left to_xfer
1385 * -------------> ----------> ----->
Amaury Denoyellee0320b82022-03-11 19:12:23 +01001386 * --------------------------------------------------
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001387 * |...............|xxxxxxxxxxx|<<<<<
Amaury Denoyellee0320b82022-03-11 19:12:23 +01001388 * --------------------------------------------------
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001389 * ^ ack-off ^ sent-off ^ off
1390 *
1391 * STREAM frame
1392 * ^ ^
1393 * |xxxxxxxxxxxxxxxxx|
1394 */
1395
Amaury Denoyelle7272cd72022-03-29 15:15:54 +02001396 BUG_ON_HOT(qcs->tx.sent_offset < qcs->stream->ack_offset);
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001397 BUG_ON_HOT(qcs->tx.offset < qcs->tx.sent_offset);
Amaury Denoyelle78fa5592022-06-10 15:18:12 +02001398 BUG_ON_HOT(qcc->tx.offsets < qcc->tx.sent_offsets);
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001399
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001400 left = qcs->tx.offset - qcs->tx.sent_offset;
Amaury Denoyellefe8f5552022-04-27 16:44:49 +02001401 to_xfer = QUIC_MIN(b_data(in), b_room(out));
Amaury Denoyelle6ea78192022-03-07 15:47:02 +01001402
1403 BUG_ON_HOT(qcs->tx.offset > qcs->tx.msd);
1404 /* do not exceed flow control limit */
1405 if (qcs->tx.offset + to_xfer > qcs->tx.msd)
1406 to_xfer = qcs->tx.msd - qcs->tx.offset;
1407
Amaury Denoyelleb9e06402022-06-10 15:16:21 +02001408 BUG_ON_HOT(qcc->tx.offsets > qcc->rfctl.md);
Amaury Denoyelle05ce55e2022-03-08 10:35:42 +01001409 /* do not overcome flow control limit on connection */
Amaury Denoyelleb9e06402022-06-10 15:16:21 +02001410 if (qcc->tx.offsets + to_xfer > qcc->rfctl.md)
1411 to_xfer = qcc->rfctl.md - qcc->tx.offsets;
Amaury Denoyelle05ce55e2022-03-08 10:35:42 +01001412
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001413 if (!left && !to_xfer)
Frédéric Lécailled2ba0962021-09-20 17:50:03 +02001414 goto out;
1415
Amaury Denoyellefe8f5552022-04-27 16:44:49 +02001416 total = b_force_xfer(out, in, to_xfer);
Amaury Denoyelleda6ad202022-04-12 11:41:04 +02001417
1418 out:
1419 {
1420 struct qcs_xfer_data_trace_arg arg = {
1421 .prep = b_data(out), .xfer = total,
1422 };
1423 TRACE_LEAVE(QMUX_EV_QCS_SEND|QMUX_EV_QCS_XFER_DATA,
1424 qcc->conn, qcs, &arg);
1425 }
1426
1427 return total;
Amaury Denoyelleda6ad202022-04-12 11:41:04 +02001428}
1429
Amaury Denoyellefe8f5552022-04-27 16:44:49 +02001430/* Prepare a STREAM frame for <qcs> instance using <out> as payload. The frame
1431 * is appended in <frm_list>. Set <fin> if this is supposed to be the last
1432 * stream frame.
1433 *
1434 * Returns the length of the STREAM frame or a negative error code.
1435 */
Amaury Denoyelleda6ad202022-04-12 11:41:04 +02001436static int qcs_build_stream_frm(struct qcs *qcs, struct buffer *out, char fin,
1437 struct list *frm_list)
1438{
1439 struct qcc *qcc = qcs->qcc;
1440 struct quic_frame *frm;
1441 int head, total;
Amaury Denoyellea4569202022-04-15 17:29:25 +02001442 uint64_t base_off;
Amaury Denoyelleda6ad202022-04-12 11:41:04 +02001443
1444 TRACE_ENTER(QMUX_EV_QCS_SEND, qcc->conn, qcs);
1445
Amaury Denoyellea4569202022-04-15 17:29:25 +02001446 /* if ack_offset < buf_offset, it points to an older buffer. */
1447 base_off = MAX(qcs->stream->buf_offset, qcs->stream->ack_offset);
1448 BUG_ON(qcs->tx.sent_offset < base_off);
1449
1450 head = qcs->tx.sent_offset - base_off;
Amaury Denoyelleda6ad202022-04-12 11:41:04 +02001451 total = b_data(out) - head;
Amaury Denoyellea4569202022-04-15 17:29:25 +02001452 BUG_ON(total < 0);
1453
Amaury Denoyellee53b4892022-07-08 17:19:40 +02001454 if (!total && !fin) {
1455 /* No need to send anything if total is NULL and no FIN to signal. */
Amaury Denoyelleda6ad202022-04-12 11:41:04 +02001456 TRACE_LEAVE(QMUX_EV_QCS_SEND, qcc->conn, qcs);
1457 return 0;
1458 }
Amaury Denoyellee53b4892022-07-08 17:19:40 +02001459 BUG_ON((!total && qcs->tx.sent_offset > qcs->tx.offset) ||
1460 (total && qcs->tx.sent_offset >= qcs->tx.offset));
Amaury Denoyellea4569202022-04-15 17:29:25 +02001461 BUG_ON(qcs->tx.sent_offset + total > qcs->tx.offset);
Amaury Denoyelle78fa5592022-06-10 15:18:12 +02001462 BUG_ON(qcc->tx.sent_offsets + total > qcc->rfctl.md);
Amaury Denoyelleda6ad202022-04-12 11:41:04 +02001463
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02001464 TRACE_PROTO("sending STREAM frame", QMUX_EV_QCS_SEND, qcc->conn, qcs);
Amaury Denoyelle40c24f12023-01-27 17:47:49 +01001465 frm = qc_frm_alloc(QUIC_FT_STREAM_8);
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02001466 if (!frm) {
1467 TRACE_ERROR("frame alloc failure", QMUX_EV_QCS_SEND, qcc->conn, qcs);
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001468 goto err;
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02001469 }
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001470
Amaury Denoyelle7272cd72022-03-29 15:15:54 +02001471 frm->stream.stream = qcs->stream;
Amaury Denoyelled8e680c2022-03-29 15:18:44 +02001472 frm->stream.id = qcs->id;
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001473 frm->stream.buf = out;
1474 frm->stream.data = (unsigned char *)b_peek(out, head);
Amaury Denoyelle1dac0182023-02-02 16:45:07 +01001475 frm->stream.offset.key = 0;
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001476
Amaury Denoyellefecfa0d2021-12-07 16:50:14 +01001477 /* FIN is positioned only when the buffer has been totally emptied. */
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001478 if (fin)
1479 frm->type |= QUIC_STREAM_FRAME_TYPE_FIN_BIT;
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001480
1481 if (qcs->tx.sent_offset) {
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001482 frm->type |= QUIC_STREAM_FRAME_TYPE_OFF_BIT;
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001483 frm->stream.offset.key = qcs->tx.sent_offset;
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001484 }
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001485
Amaury Denoyelleda6ad202022-04-12 11:41:04 +02001486 frm->type |= QUIC_STREAM_FRAME_TYPE_LEN_BIT;
1487 frm->stream.len = total;
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001488
Amaury Denoyelle2c71fe52022-02-09 18:16:49 +01001489 LIST_APPEND(frm_list, &frm->list);
Amaury Denoyelle4f137572022-03-24 17:10:00 +01001490
Frédéric Lécailled2ba0962021-09-20 17:50:03 +02001491 out:
Amaury Denoyellefdcec362022-03-25 09:28:10 +01001492 {
Amaury Denoyelleda6ad202022-04-12 11:41:04 +02001493 struct qcs_build_stream_trace_arg arg = {
1494 .len = frm->stream.len, .fin = fin,
1495 .offset = frm->stream.offset.key,
Amaury Denoyellefdcec362022-03-25 09:28:10 +01001496 };
Amaury Denoyelleda6ad202022-04-12 11:41:04 +02001497 TRACE_LEAVE(QMUX_EV_QCS_SEND|QMUX_EV_QCS_BUILD_STRM,
Amaury Denoyellefdcec362022-03-25 09:28:10 +01001498 qcc->conn, qcs, &arg);
1499 }
Amaury Denoyelle4f137572022-03-24 17:10:00 +01001500
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001501 return total;
1502
1503 err:
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001504 TRACE_LEAVE(QMUX_EV_QCS_SEND, qcc->conn, qcs);
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001505 return -1;
1506}
1507
Ilya Shipitsin3b64a282022-07-29 22:26:53 +05001508/* Check after transferring data from qcs.tx.buf if FIN must be set on the next
Amaury Denoyellee53b4892022-07-08 17:19:40 +02001509 * STREAM frame for <qcs>.
1510 *
1511 * Returns true if FIN must be set else false.
1512 */
1513static int qcs_stream_fin(struct qcs *qcs)
1514{
1515 return qcs->flags & QC_SF_FIN_STREAM && !b_data(&qcs->tx.buf);
1516}
1517
Amaury Denoyelle0a1154a2023-01-06 17:43:11 +01001518/* Return true if <qcs> has data to send in new STREAM frames. */
1519static forceinline int qcs_need_sending(struct qcs *qcs)
1520{
1521 return b_data(&qcs->tx.buf) || qcs->tx.sent_offset < qcs->tx.offset ||
1522 qcs_stream_fin(qcs);
1523}
1524
Amaury Denoyelle54445d02022-03-10 16:44:14 +01001525/* This function must be called by the upper layer to inform about the sending
1526 * of a STREAM frame for <qcs> instance. The frame is of <data> length and on
1527 * <offset>.
1528 */
1529void qcc_streams_sent_done(struct qcs *qcs, uint64_t data, uint64_t offset)
1530{
Amaury Denoyelle05ce55e2022-03-08 10:35:42 +01001531 struct qcc *qcc = qcs->qcc;
1532 uint64_t diff;
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001533
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02001534 TRACE_ENTER(QMUX_EV_QCS_SEND, qcc->conn, qcs);
1535
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001536 BUG_ON(offset > qcs->tx.sent_offset);
Amaury Denoyelle78fa5592022-06-10 15:18:12 +02001537 BUG_ON(offset + data > qcs->tx.offset);
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001538
Amaury Denoyelle54445d02022-03-10 16:44:14 +01001539 /* check if the STREAM frame has already been notified. It can happen
1540 * for retransmission.
1541 */
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02001542 if (offset + data < qcs->tx.sent_offset) {
1543 TRACE_DEVEL("offset already notified", QMUX_EV_QCS_SEND, qcc->conn, qcs);
1544 goto out;
1545 }
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001546
Amaury Denoyelle38e60062022-07-01 16:48:42 +02001547 qcs_idle_open(qcs);
1548
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001549 diff = offset + data - qcs->tx.sent_offset;
Amaury Denoyellee53b4892022-07-08 17:19:40 +02001550 if (diff) {
1551 /* increase offset sum on connection */
1552 qcc->tx.sent_offsets += diff;
1553 BUG_ON_HOT(qcc->tx.sent_offsets > qcc->rfctl.md);
Amaury Denoyelle31d20572023-01-06 15:29:59 +01001554 if (qcc->tx.sent_offsets == qcc->rfctl.md) {
Amaury Denoyellee53b4892022-07-08 17:19:40 +02001555 qcc->flags |= QC_CF_BLK_MFCTL;
Amaury Denoyelle31d20572023-01-06 15:29:59 +01001556 TRACE_STATE("connection flow-control reached", QMUX_EV_QCS_SEND, qcc->conn);
1557 }
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001558
Amaury Denoyellee53b4892022-07-08 17:19:40 +02001559 /* increase offset on stream */
1560 qcs->tx.sent_offset += diff;
1561 BUG_ON_HOT(qcs->tx.sent_offset > qcs->tx.msd);
1562 BUG_ON_HOT(qcs->tx.sent_offset > qcs->tx.offset);
Amaury Denoyelle31d20572023-01-06 15:29:59 +01001563 if (qcs->tx.sent_offset == qcs->tx.msd) {
Amaury Denoyellee53b4892022-07-08 17:19:40 +02001564 qcs->flags |= QC_SF_BLK_SFCTL;
Amaury Denoyelle31d20572023-01-06 15:29:59 +01001565 TRACE_STATE("stream flow-control reached", QMUX_EV_QCS_SEND, qcc->conn, qcs);
1566 }
Amaury Denoyelle05ce55e2022-03-08 10:35:42 +01001567
Amaury Denoyellea9de7ea2023-01-06 17:16:47 +01001568 /* If qcs.stream.buf is full, release it to the lower layer. */
Amaury Denoyellee53b4892022-07-08 17:19:40 +02001569 if (qcs->tx.offset == qcs->tx.sent_offset &&
1570 b_full(&qcs->stream->buf->buf)) {
1571 qc_stream_buf_release(qcs->stream);
Amaury Denoyellee53b4892022-07-08 17:19:40 +02001572 }
1573 }
Amaury Denoyellea4569202022-04-15 17:29:25 +02001574
Amaury Denoyelle20f2a422023-01-03 14:39:24 +01001575 if (qcs->tx.offset == qcs->tx.sent_offset && !b_data(&qcs->tx.buf)) {
1576 /* Remove stream from send_list if all was sent. */
1577 LIST_DEL_INIT(&qcs->el_send);
1578 TRACE_STATE("stream sent done", QMUX_EV_QCS_SEND, qcc->conn, qcs);
1579
1580 if (qcs->flags & (QC_SF_FIN_STREAM|QC_SF_DETACH)) {
1581 /* Close stream locally. */
1582 qcs_close_local(qcs);
1583 /* Reset flag to not emit multiple FIN STREAM frames. */
1584 qcs->flags &= ~QC_SF_FIN_STREAM;
1585 }
Amaury Denoyellea4569202022-04-15 17:29:25 +02001586 }
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02001587
1588 out:
1589 TRACE_LEAVE(QMUX_EV_QCS_SEND, qcc->conn, qcs);
Amaury Denoyelle54445d02022-03-10 16:44:14 +01001590}
1591
Amaury Denoyelle2c71fe52022-02-09 18:16:49 +01001592/* Wrapper for send on transport layer. Send a list of frames <frms> for the
1593 * connection <qcc>.
1594 *
1595 * Returns 0 if all data sent with success else non-zero.
1596 */
1597static int qc_send_frames(struct qcc *qcc, struct list *frms)
1598{
Amaury Denoyelle4f137572022-03-24 17:10:00 +01001599 TRACE_ENTER(QMUX_EV_QCC_SEND, qcc->conn);
1600
1601 if (LIST_ISEMPTY(frms)) {
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001602 TRACE_DEVEL("no frames to send", QMUX_EV_QCC_SEND, qcc->conn);
1603 goto err;
Amaury Denoyelle4f137572022-03-24 17:10:00 +01001604 }
Frédéric Lécaille4e22f282022-03-18 18:38:19 +01001605
Amaury Denoyelle036cc5d2022-09-26 15:02:31 +02001606 if (!qc_send_mux(qcc->conn->handle.qc, frms))
1607 goto err;
Amaury Denoyellee9c4cc12022-03-04 15:29:53 +01001608
Amaury Denoyelledb5d1a12022-03-10 16:42:23 +01001609 /* If there is frames left at this stage, transport layer is blocked.
1610 * Subscribe on it to retry later.
1611 */
Amaury Denoyelle2c71fe52022-02-09 18:16:49 +01001612 if (!LIST_ISEMPTY(frms)) {
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001613 TRACE_DEVEL("remaining frames to send, subscribing", QMUX_EV_QCC_SEND, qcc->conn);
Amaury Denoyelle2c71fe52022-02-09 18:16:49 +01001614 qcc->conn->xprt->subscribe(qcc->conn, qcc->conn->xprt_ctx,
1615 SUB_RETRY_SEND, &qcc->wait_event);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001616 goto err;
Amaury Denoyelle2c71fe52022-02-09 18:16:49 +01001617 }
1618
Amaury Denoyelle3baab742022-08-11 18:35:55 +02001619 TRACE_LEAVE(QMUX_EV_QCC_SEND, qcc->conn);
Amaury Denoyelle843a1192022-07-04 11:44:38 +02001620 return 0;
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001621
1622 err:
Amaury Denoyelle3baab742022-08-11 18:35:55 +02001623 TRACE_LEAVE(QMUX_EV_QCC_SEND, qcc->conn);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001624 return 1;
Amaury Denoyelle843a1192022-07-04 11:44:38 +02001625}
1626
1627/* Emit a RESET_STREAM on <qcs>.
1628 *
1629 * Returns 0 if the frame has been successfully sent else non-zero.
1630 */
1631static int qcs_send_reset(struct qcs *qcs)
1632{
1633 struct list frms = LIST_HEAD_INIT(frms);
1634 struct quic_frame *frm;
1635
1636 TRACE_ENTER(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
1637
Amaury Denoyelle40c24f12023-01-27 17:47:49 +01001638 frm = qc_frm_alloc(QUIC_FT_RESET_STREAM);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001639 if (!frm) {
1640 TRACE_LEAVE(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
Amaury Denoyelle843a1192022-07-04 11:44:38 +02001641 return 1;
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001642 }
Amaury Denoyelle843a1192022-07-04 11:44:38 +02001643
Amaury Denoyelle843a1192022-07-04 11:44:38 +02001644 frm->reset_stream.id = qcs->id;
1645 frm->reset_stream.app_error_code = qcs->err;
1646 frm->reset_stream.final_size = qcs->tx.sent_offset;
1647
1648 LIST_APPEND(&frms, &frm->list);
1649 if (qc_send_frames(qcs->qcc, &frms)) {
Amaury Denoyelle57b3eaa2023-02-02 16:12:09 +01001650 qc_frm_free(&frm);
Amaury Denoyelle843a1192022-07-04 11:44:38 +02001651 TRACE_DEVEL("cannot send RESET_STREAM", QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
1652 return 1;
1653 }
1654
1655 if (qcs_sc(qcs)) {
1656 se_fl_set_error(qcs->sd);
1657 qcs_alert(qcs);
1658 }
1659
1660 qcs_close_local(qcs);
1661 qcs->flags &= ~QC_SF_TO_RESET;
1662
1663 TRACE_LEAVE(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
Amaury Denoyelle2c71fe52022-02-09 18:16:49 +01001664 return 0;
1665}
1666
Amaury Denoyelle663e8722022-12-09 14:58:28 +01001667/* Emit a STOP_SENDING on <qcs>.
1668 *
1669 * Returns 0 if the frame has been successfully sent else non-zero.
1670 */
1671static int qcs_send_stop_sending(struct qcs *qcs)
1672{
1673 struct list frms = LIST_HEAD_INIT(frms);
1674 struct quic_frame *frm;
1675 struct qcc *qcc = qcs->qcc;
1676
1677 TRACE_ENTER(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
1678
1679 /* RFC 9000 3.3. Permitted Frame Types
1680 *
1681 * A
1682 * receiver MAY send a STOP_SENDING frame in any state where it has not
1683 * received a RESET_STREAM frame -- that is, states other than "Reset
1684 * Recvd" or "Reset Read". However, there is little value in sending a
1685 * STOP_SENDING frame in the "Data Recvd" state, as all stream data has
1686 * been received. A sender could receive either of these two types of
1687 * frames in any state as a result of delayed delivery of packets.¶
1688 */
1689 if (qcs_is_close_remote(qcs)) {
1690 TRACE_STATE("skip STOP_SENDING on remote already closed", QMUX_EV_QCS_SEND, qcc->conn, qcs);
1691 goto done;
1692 }
1693
Amaury Denoyelle40c24f12023-01-27 17:47:49 +01001694 frm = qc_frm_alloc(QUIC_FT_STOP_SENDING);
Amaury Denoyelle663e8722022-12-09 14:58:28 +01001695 if (!frm) {
1696 TRACE_LEAVE(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
1697 return 1;
1698 }
1699
Amaury Denoyelle663e8722022-12-09 14:58:28 +01001700 frm->stop_sending.id = qcs->id;
1701 frm->stop_sending.app_error_code = qcs->err;
1702
1703 LIST_APPEND(&frms, &frm->list);
1704 if (qc_send_frames(qcs->qcc, &frms)) {
Amaury Denoyelle57b3eaa2023-02-02 16:12:09 +01001705 qc_frm_free(&frm);
Amaury Denoyelle663e8722022-12-09 14:58:28 +01001706 TRACE_DEVEL("cannot send STOP_SENDING", QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
1707 return 1;
1708 }
1709
1710 done:
1711 qcs->flags &= ~QC_SF_TO_STOP_SENDING;
1712
1713 TRACE_LEAVE(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
1714 return 0;
1715}
1716
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02001717/* Used internally by qc_send function. Proceed to send for <qcs>. This will
1718 * transfer data from qcs buffer to its quic_stream counterpart. A STREAM frame
Amaury Denoyelleb9e06402022-06-10 15:16:21 +02001719 * is then generated and inserted in <frms> list.
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02001720 *
1721 * Returns the total bytes transferred between qcs and quic_stream buffers. Can
1722 * be null if out buffer cannot be allocated.
1723 */
Amaury Denoyelleb9e06402022-06-10 15:16:21 +02001724static int _qc_send_qcs(struct qcs *qcs, struct list *frms)
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02001725{
1726 struct qcc *qcc = qcs->qcc;
1727 struct buffer *buf = &qcs->tx.buf;
1728 struct buffer *out = qc_stream_buf_get(qcs->stream);
1729 int xfer = 0;
Amaury Denoyellee53b4892022-07-08 17:19:40 +02001730 char fin = 0;
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02001731
Amaury Denoyelle20f2a422023-01-03 14:39:24 +01001732 /* Cannot send STREAM on remote unidirectional streams. */
1733 BUG_ON(quic_stream_is_uni(qcs->id) && quic_stream_is_remote(qcc, qcs->id));
1734
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02001735 /* Allocate <out> buffer if necessary. */
1736 if (!out) {
1737 if (qcc->flags & QC_CF_CONN_FULL)
1738 return 0;
1739
1740 out = qc_stream_buf_alloc(qcs->stream, qcs->tx.offset);
1741 if (!out) {
1742 qcc->flags |= QC_CF_CONN_FULL;
1743 return 0;
1744 }
1745 }
1746
1747 /* Transfer data from <buf> to <out>. */
1748 if (b_data(buf)) {
Amaury Denoyelleb9e06402022-06-10 15:16:21 +02001749 xfer = qcs_xfer_data(qcs, out, buf);
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02001750 if (xfer > 0) {
1751 qcs_notify_send(qcs);
1752 qcs->flags &= ~QC_SF_BLK_MROOM;
1753 }
1754
1755 qcs->tx.offset += xfer;
Amaury Denoyelle78fa5592022-06-10 15:18:12 +02001756 BUG_ON_HOT(qcs->tx.offset > qcs->tx.msd);
Amaury Denoyelleb9e06402022-06-10 15:16:21 +02001757 qcc->tx.offsets += xfer;
Amaury Denoyelle78fa5592022-06-10 15:18:12 +02001758 BUG_ON_HOT(qcc->tx.offsets > qcc->rfctl.md);
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02001759 }
1760
1761 /* out buffer cannot be emptied if qcs offsets differ. */
1762 BUG_ON(!b_data(out) && qcs->tx.sent_offset != qcs->tx.offset);
1763
Ilya Shipitsin3b64a282022-07-29 22:26:53 +05001764 /* FIN is set if all incoming data were transferred. */
Amaury Denoyellee53b4892022-07-08 17:19:40 +02001765 fin = qcs_stream_fin(qcs);
1766
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02001767 /* Build a new STREAM frame with <out> buffer. */
Amaury Denoyellee53b4892022-07-08 17:19:40 +02001768 if (qcs->tx.sent_offset != qcs->tx.offset || fin) {
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02001769 int ret;
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02001770 ret = qcs_build_stream_frm(qcs, out, fin, frms);
Amaury Denoyelleb50f3112022-04-28 14:41:50 +02001771 if (ret < 0) { ABORT_NOW(); /* TODO handle this properly */ }
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02001772 }
1773
1774 return xfer;
1775}
1776
Amaury Denoyelle75d14ad2022-03-22 15:10:29 +01001777/* Proceed to sending. Loop through all available streams for the <qcc>
1778 * instance and try to send as much as possible.
1779 *
1780 * Returns the total of bytes sent to the transport layer.
1781 */
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001782static int qc_send(struct qcc *qcc)
1783{
Amaury Denoyelle6ccfa3c2022-03-10 16:45:53 +01001784 struct list frms = LIST_HEAD_INIT(frms);
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02001785 struct qcs *qcs, *qcs_tmp;
Amaury Denoyellea9de7ea2023-01-06 17:16:47 +01001786 int total = 0;
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001787
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001788 TRACE_ENTER(QMUX_EV_QCC_SEND, qcc->conn);
Frédéric Lécaille8526f142021-09-20 17:58:22 +02001789
Amaury Denoyelle9fab9fd2022-05-20 15:04:38 +02001790 if (qcc->conn->flags & CO_FL_SOCK_WR_SH || qcc->flags & QC_CF_CC_EMIT) {
Amaury Denoyelled97fc802022-04-06 16:13:09 +02001791 qcc->conn->flags |= CO_FL_ERROR;
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001792 TRACE_DEVEL("connection on error", QMUX_EV_QCC_SEND, qcc->conn);
1793 goto err;
Amaury Denoyelled97fc802022-04-06 16:13:09 +02001794 }
1795
Amaury Denoyellec985cb12022-05-16 14:29:59 +02001796 if (!LIST_ISEMPTY(&qcc->lfctl.frms)) {
1797 if (qc_send_frames(qcc, &qcc->lfctl.frms)) {
1798 TRACE_DEVEL("flow-control frames rejected by transport, aborting send", QMUX_EV_QCC_SEND, qcc->conn);
1799 goto out;
1800 }
1801 }
Amaury Denoyellec9337802022-04-04 16:36:34 +02001802
Amaury Denoyelle05ce55e2022-03-08 10:35:42 +01001803 if (qcc->flags & QC_CF_BLK_MFCTL)
Amaury Denoyelle0a1154a2023-01-06 17:43:11 +01001804 goto err;
Amaury Denoyelle05ce55e2022-03-08 10:35:42 +01001805
Amaury Denoyelle0a1154a2023-01-06 17:43:11 +01001806 /* Send STREAM/STOP_SENDING/RESET_STREAM data for registered streams. */
1807 list_for_each_entry_safe(qcs, qcs_tmp, &qcc->send_list, el_send) {
1808 /* Stream must not be present in send_list if it has nothing to send. */
1809 BUG_ON(!(qcs->flags & (QC_SF_TO_STOP_SENDING|QC_SF_TO_RESET)) &&
1810 !qcs_need_sending(qcs));
Amaury Denoyellec6195d72022-05-23 11:39:14 +02001811
Amaury Denoyelle0a1154a2023-01-06 17:43:11 +01001812 /* Each STOP_SENDING/RESET_STREAM frame is sent individually to
1813 * guarantee its emission.
1814 *
1815 * TODO multiplex several frames in same datagram to optimize sending
1816 */
1817 if (qcs->flags & QC_SF_TO_STOP_SENDING) {
1818 if (qcs_send_stop_sending(qcs))
1819 goto out;
Amaury Denoyelle2c71fe52022-02-09 18:16:49 +01001820
Amaury Denoyelle0a1154a2023-01-06 17:43:11 +01001821 /* Remove stream from send_list if it had only STOP_SENDING
1822 * to send.
1823 */
1824 if (!(qcs->flags & QC_SF_TO_RESET) && !qcs_need_sending(qcs)) {
1825 LIST_DEL_INIT(&qcs->el_send);
1826 continue;
1827 }
Amaury Denoyellee2ec9422022-03-10 16:46:18 +01001828 }
1829
Amaury Denoyelle843a1192022-07-04 11:44:38 +02001830 if (qcs->flags & QC_SF_TO_RESET) {
Amaury Denoyelle0a1154a2023-01-06 17:43:11 +01001831 if (qcs_send_reset(qcs))
1832 goto out;
Amaury Denoyelle6ea78192022-03-07 15:47:02 +01001833
Amaury Denoyelle0a1154a2023-01-06 17:43:11 +01001834 /* RFC 9000 3.3. Permitted Frame Types
1835 *
1836 * A sender MUST NOT send
1837 * a STREAM or STREAM_DATA_BLOCKED frame for a stream in the
1838 * "Reset Sent" state or any terminal state -- that is, after
1839 * sending a RESET_STREAM frame.
1840 */
Amaury Denoyelle20f2a422023-01-03 14:39:24 +01001841 LIST_DEL_INIT(&qcs->el_send);
Amaury Denoyelled2f80a22022-04-15 17:30:49 +02001842 continue;
1843 }
Amaury Denoyellea4569202022-04-15 17:29:25 +02001844
Amaury Denoyelle20f2a422023-01-03 14:39:24 +01001845 if (!(qcs->flags & QC_SF_BLK_SFCTL))
1846 total += _qc_send_qcs(qcs, &frms);
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02001847 }
Amaury Denoyelleda6ad202022-04-12 11:41:04 +02001848
Amaury Denoyellea9de7ea2023-01-06 17:16:47 +01001849 /* Retry sending until no frame to send, data rejected or connection
1850 * flow-control limit reached.
1851 */
1852 while (qc_send_frames(qcc, &frms) == 0 && !(qcc->flags & QC_CF_BLK_MFCTL)) {
1853 /* Reloop over <qcc.send_list>. Useful for streams which have
1854 * fulfilled their qc_stream_desc buf and have now release it.
1855 */
1856 list_for_each_entry(qcs, &qcc->send_list, el_send) {
1857 /* Only streams blocked on flow-control or waiting on a
1858 * new qc_stream_desc should be present in send_list as
1859 * long as transport layer can handle all data.
1860 */
1861 BUG_ON(qcs->stream->buf && !(qcs->flags & QC_SF_BLK_SFCTL));
Amaury Denoyelleda6ad202022-04-12 11:41:04 +02001862
Amaury Denoyellea9de7ea2023-01-06 17:16:47 +01001863 if (!(qcs->flags & QC_SF_BLK_SFCTL))
1864 total += _qc_send_qcs(qcs, &frms);
1865 }
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001866 }
Frédéric Lécaille8526f142021-09-20 17:58:22 +02001867
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02001868 out:
Amaury Denoyelle43c090c2022-06-10 15:16:40 +02001869 /* Deallocate frames that the transport layer has rejected. */
1870 if (!LIST_ISEMPTY(&frms)) {
1871 struct quic_frame *frm, *frm2;
Amaury Denoyelle20f2a422023-01-03 14:39:24 +01001872
Amaury Denoyelle57b3eaa2023-02-02 16:12:09 +01001873 list_for_each_entry_safe(frm, frm2, &frms, list)
1874 qc_frm_free(&frm);
Amaury Denoyelle43c090c2022-06-10 15:16:40 +02001875 }
1876
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001877 TRACE_LEAVE(QMUX_EV_QCC_SEND, qcc->conn);
Amaury Denoyelle75d14ad2022-03-22 15:10:29 +01001878 return total;
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001879
1880 err:
1881 TRACE_LEAVE(QMUX_EV_QCC_SEND, qcc->conn);
1882 return 0;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001883}
1884
Amaury Denoyelle37c2e4a2022-05-16 13:54:59 +02001885/* Proceed on receiving. Loop through all streams from <qcc> and use decode_qcs
1886 * operation.
1887 *
1888 * Returns 0 on success else non-zero.
1889 */
1890static int qc_recv(struct qcc *qcc)
1891{
1892 struct eb64_node *node;
1893 struct qcs *qcs;
1894
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001895 TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
Amaury Denoyellee1cad8b2022-05-23 18:52:11 +02001896
Amaury Denoyelle5c4373a2022-05-24 14:47:48 +02001897 if (qcc->flags & QC_CF_CC_EMIT) {
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001898 TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
Amaury Denoyelle5c4373a2022-05-24 14:47:48 +02001899 return 0;
1900 }
1901
Amaury Denoyelle37c2e4a2022-05-16 13:54:59 +02001902 node = eb64_first(&qcc->streams_by_id);
1903 while (node) {
Amaury Denoyellef8db5aa2022-05-24 15:26:07 +02001904 uint64_t id;
1905
Amaury Denoyelle37c2e4a2022-05-16 13:54:59 +02001906 qcs = eb64_entry(node, struct qcs, by_id);
Amaury Denoyellef8db5aa2022-05-24 15:26:07 +02001907 id = qcs->id;
Amaury Denoyelle37c2e4a2022-05-16 13:54:59 +02001908
Amaury Denoyellef8db5aa2022-05-24 15:26:07 +02001909 if (!ncb_data(&qcs->rx.ncbuf, 0) || (qcs->flags & QC_SF_DEM_FULL)) {
Amaury Denoyelle37c2e4a2022-05-16 13:54:59 +02001910 node = eb64_next(node);
1911 continue;
1912 }
1913
Amaury Denoyellef8db5aa2022-05-24 15:26:07 +02001914 if (quic_stream_is_uni(id) && quic_stream_is_local(qcc, id)) {
Amaury Denoyelle37c2e4a2022-05-16 13:54:59 +02001915 node = eb64_next(node);
1916 continue;
1917 }
1918
1919 qcc_decode_qcs(qcc, qcs);
1920 node = eb64_next(node);
1921 }
1922
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02001923 TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
Amaury Denoyelle37c2e4a2022-05-16 13:54:59 +02001924 return 0;
1925}
1926
Amaury Denoyellec1a6dfd2022-07-08 14:04:21 +02001927
1928/* Release all streams which have their transfer operation achieved.
Amaury Denoyelle6a4aebf2022-02-01 10:16:05 +01001929 *
Amaury Denoyellec1a6dfd2022-07-08 14:04:21 +02001930 * Returns true if at least one stream is released.
Amaury Denoyelle6a4aebf2022-02-01 10:16:05 +01001931 */
Amaury Denoyellec1a6dfd2022-07-08 14:04:21 +02001932static int qc_purge_streams(struct qcc *qcc)
Amaury Denoyelle2873a312021-12-08 14:42:55 +01001933{
1934 struct eb64_node *node;
1935 int release = 0;
1936
Amaury Denoyelle3baab742022-08-11 18:35:55 +02001937 TRACE_ENTER(QMUX_EV_QCC_WAKE, qcc->conn);
Amaury Denoyellec1a6dfd2022-07-08 14:04:21 +02001938
Amaury Denoyelle2873a312021-12-08 14:42:55 +01001939 node = eb64_first(&qcc->streams_by_id);
1940 while (node) {
Amaury Denoyellee4301da2022-04-19 17:59:50 +02001941 struct qcs *qcs = eb64_entry(node, struct qcs, by_id);
Amaury Denoyelle2873a312021-12-08 14:42:55 +01001942 node = eb64_next(node);
1943
Amaury Denoyelle38e60062022-07-01 16:48:42 +02001944 /* Release not attached closed streams. */
1945 if (qcs->st == QC_SS_CLO && !qcs_sc(qcs)) {
Amaury Denoyelle047d86a2022-08-10 16:42:35 +02001946 TRACE_STATE("purging closed stream", QMUX_EV_QCC_WAKE, qcs->qcc->conn, qcs);
Amaury Denoyelle38e60062022-07-01 16:48:42 +02001947 qcs_destroy(qcs);
1948 release = 1;
1949 continue;
1950 }
1951
Amaury Denoyellec1a6dfd2022-07-08 14:04:21 +02001952 /* Release detached streams with empty buffer. */
Amaury Denoyelle2873a312021-12-08 14:42:55 +01001953 if (qcs->flags & QC_SF_DETACH) {
Amaury Denoyelle20d1f842022-07-11 11:23:17 +02001954 if (qcs_is_close_local(qcs)) {
Amaury Denoyelle047d86a2022-08-10 16:42:35 +02001955 TRACE_STATE("purging detached stream", QMUX_EV_QCC_WAKE, qcs->qcc->conn, qcs);
Amaury Denoyelle2873a312021-12-08 14:42:55 +01001956 qcs_destroy(qcs);
1957 release = 1;
Amaury Denoyellec1a6dfd2022-07-08 14:04:21 +02001958 continue;
Amaury Denoyelle2873a312021-12-08 14:42:55 +01001959 }
Amaury Denoyellec1a6dfd2022-07-08 14:04:21 +02001960
1961 qcc->conn->xprt->subscribe(qcc->conn, qcc->conn->xprt_ctx,
1962 SUB_RETRY_SEND, &qcc->wait_event);
Amaury Denoyelle2873a312021-12-08 14:42:55 +01001963 }
1964 }
1965
Amaury Denoyelle3baab742022-08-11 18:35:55 +02001966 TRACE_LEAVE(QMUX_EV_QCC_WAKE, qcc->conn);
Amaury Denoyelle2873a312021-12-08 14:42:55 +01001967 return release;
1968}
1969
Amaury Denoyelleb30247b2023-01-24 18:18:23 +01001970/* Execute application layer shutdown. If this operation is not defined, a
1971 * CONNECTION_CLOSE will be prepared as a fallback. This function is protected
1972 * against multiple invocation with the flag QC_CF_APP_SHUT.
Amaury Denoyellec49d5d12022-07-15 10:32:53 +02001973 */
Amaury Denoyelleb30247b2023-01-24 18:18:23 +01001974static void qc_shutdown(struct qcc *qcc)
Amaury Denoyellec49d5d12022-07-15 10:32:53 +02001975{
Amaury Denoyelleb30247b2023-01-24 18:18:23 +01001976 TRACE_ENTER(QMUX_EV_QCC_END, qcc->conn);
Amaury Denoyellec49d5d12022-07-15 10:32:53 +02001977
Amaury Denoyelleb30247b2023-01-24 18:18:23 +01001978 if (qcc->flags & QC_CF_APP_SHUT)
1979 goto out;
Amaury Denoyellec49d5d12022-07-15 10:32:53 +02001980
Amaury Denoyellef8aaf8b2022-09-14 16:23:47 +02001981 if (qcc->app_ops && qcc->app_ops->shutdown) {
Amaury Denoyellef8aaf8b2022-09-14 16:23:47 +02001982 qcc->app_ops->shutdown(qcc->ctx);
Amaury Denoyellea154dc02022-06-13 17:09:01 +02001983 qc_send(qcc);
Amaury Denoyellec49d5d12022-07-15 10:32:53 +02001984 }
1985 else {
1986 qcc_emit_cc_app(qcc, QC_ERR_NO_ERROR, 0);
1987 }
1988
Amaury Denoyelleb30247b2023-01-24 18:18:23 +01001989 out:
1990 qcc->flags |= QC_CF_APP_SHUT;
1991 TRACE_LEAVE(QMUX_EV_QCC_END, qcc->conn);
1992}
1993
Amaury Denoyelle14dbb842023-01-24 18:19:47 +01001994/* Conduct operations which should be made for <qcc> connection after
1995 * input/output. Most notably, closed streams are purged which may leave the
1996 * connection has ready to be released.
1997 *
1998 * Returns 1 if <qcc> must be released else 0.
1999 */
2000
2001static int qc_process(struct qcc *qcc)
2002{
2003 qc_purge_streams(qcc);
2004
Amaury Denoyelleb3aa07c2023-01-24 18:20:28 +01002005 /* Check if a soft-stop is in progress.
2006 *
2007 * TODO this is relevant for frontend connections only.
2008 */
2009 if (unlikely(qcc->proxy->flags & (PR_FL_DISABLED|PR_FL_STOPPED))) {
2010 int close = 1;
2011
2012 /* If using listener socket, soft-stop is not supported. The
2013 * connection must be closed immediately.
2014 */
2015 if (!qc_test_fd(qcc->conn->handle.qc)) {
2016 TRACE_DEVEL("proxy disabled with listener socket, closing connection", QMUX_EV_QCC_WAKE, qcc->conn);
2017 qcc->conn->flags |= (CO_FL_SOCK_RD_SH|CO_FL_SOCK_WR_SH);
2018 qc_send(qcc);
2019 goto out;
2020 }
2021
2022 TRACE_DEVEL("proxy disabled, prepare connection soft-stop", QMUX_EV_QCC_WAKE, qcc->conn);
2023
2024 /* If a close-spread-time option is set, we want to avoid
2025 * closing all the active HTTP3 connections at once so we add a
2026 * random factor that will spread the closing.
2027 */
2028 if (tick_isset(global.close_spread_end)) {
2029 int remaining_window = tick_remain(now_ms, global.close_spread_end);
2030 if (remaining_window) {
2031 /* This should increase the closing rate the
2032 * further along the window we are. */
2033 close = (remaining_window <= statistical_prng_range(global.close_spread_time));
2034 }
2035 }
2036 else if (global.tune.options & GTUNE_DISABLE_ACTIVE_CLOSE) {
2037 close = 0; /* let the client close his connection himself */
2038 }
2039
2040 if (close)
2041 qc_shutdown(qcc);
2042 }
2043
2044 out:
Amaury Denoyelle14dbb842023-01-24 18:19:47 +01002045 if (qcc_is_dead(qcc))
2046 return 1;
2047
2048 return 0;
2049}
2050
Amaury Denoyelleb30247b2023-01-24 18:18:23 +01002051/* release function. This one should be called to free all resources allocated
2052 * to the mux.
2053 */
2054static void qc_release(struct qcc *qcc)
2055{
2056 struct connection *conn = qcc->conn;
2057 struct eb64_node *node;
2058
2059 TRACE_ENTER(QMUX_EV_QCC_END, conn);
2060
2061 qc_shutdown(qcc);
2062
Amaury Denoyellec49d5d12022-07-15 10:32:53 +02002063 if (qcc->task) {
2064 task_destroy(qcc->task);
2065 qcc->task = NULL;
2066 }
2067
2068 if (qcc->wait_event.tasklet)
2069 tasklet_free(qcc->wait_event.tasklet);
2070 if (conn && qcc->wait_event.events) {
2071 conn->xprt->unsubscribe(conn, conn->xprt_ctx,
2072 qcc->wait_event.events,
2073 &qcc->wait_event);
2074 }
2075
2076 /* liberate remaining qcs instances */
2077 node = eb64_first(&qcc->streams_by_id);
2078 while (node) {
2079 struct qcs *qcs = eb64_entry(node, struct qcs, by_id);
2080 node = eb64_next(node);
2081 qcs_free(qcs);
2082 }
2083
2084 while (!LIST_ISEMPTY(&qcc->lfctl.frms)) {
2085 struct quic_frame *frm = LIST_ELEM(qcc->lfctl.frms.n, struct quic_frame *, list);
Amaury Denoyelle57b3eaa2023-02-02 16:12:09 +01002086 qc_frm_free(&frm);
Amaury Denoyellec49d5d12022-07-15 10:32:53 +02002087 }
2088
Amaury Denoyellef8aaf8b2022-09-14 16:23:47 +02002089 if (qcc->app_ops && qcc->app_ops->release)
2090 qcc->app_ops->release(qcc->ctx);
2091 TRACE_PROTO("application layer released", QMUX_EV_QCC_END, conn);
2092
Amaury Denoyellec49d5d12022-07-15 10:32:53 +02002093 pool_free(pool_head_qcc, qcc);
2094
2095 if (conn) {
2096 LIST_DEL_INIT(&conn->stopping_list);
2097
2098 conn->handle.qc->conn = NULL;
2099 conn->mux = NULL;
2100 conn->ctx = NULL;
2101
2102 TRACE_DEVEL("freeing conn", QMUX_EV_QCC_END, conn);
2103
2104 conn_stop_tracking(conn);
2105 conn_full_close(conn);
2106 if (conn->destroy_cb)
2107 conn->destroy_cb(conn);
2108 conn_free(conn);
2109 }
2110
2111 TRACE_LEAVE(QMUX_EV_QCC_END);
2112}
2113
Willy Tarreau41e701e2022-09-08 15:12:59 +02002114struct task *qc_io_cb(struct task *t, void *ctx, unsigned int status)
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002115{
Amaury Denoyelle769e9ff2021-10-05 11:43:50 +02002116 struct qcc *qcc = ctx;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002117
Amaury Denoyelle3baab742022-08-11 18:35:55 +02002118 TRACE_ENTER(QMUX_EV_QCC_WAKE, qcc->conn);
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002119
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002120 qc_send(qcc);
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002121
Amaury Denoyelle37c2e4a2022-05-16 13:54:59 +02002122 qc_recv(qcc);
2123
Amaury Denoyelle14dbb842023-01-24 18:19:47 +01002124 if (qc_process(qcc)) {
2125 TRACE_STATE("releasing dead connection", QMUX_EV_QCC_WAKE, qcc->conn);
2126 goto release;
2127 }
Amaury Denoyelle5fc05d12022-07-25 14:58:48 +02002128
2129 qcc_refresh_timeout(qcc);
2130
Amaury Denoyelled3973852022-07-25 14:56:54 +02002131 end:
Amaury Denoyelle3baab742022-08-11 18:35:55 +02002132 TRACE_LEAVE(QMUX_EV_QCC_WAKE, qcc->conn);
2133 return NULL;
Amaury Denoyelle4f137572022-03-24 17:10:00 +01002134
Amaury Denoyelle3baab742022-08-11 18:35:55 +02002135 release:
2136 qc_release(qcc);
2137 TRACE_LEAVE(QMUX_EV_QCC_WAKE);
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002138 return NULL;
2139}
2140
Amaury Denoyelleaebe26f2022-01-13 16:28:06 +01002141static struct task *qc_timeout_task(struct task *t, void *ctx, unsigned int state)
2142{
2143 struct qcc *qcc = ctx;
2144 int expired = tick_is_expired(t->expire, now_ms);
2145
Amaury Denoyelle4f137572022-03-24 17:10:00 +01002146 TRACE_ENTER(QMUX_EV_QCC_WAKE, qcc ? qcc->conn : NULL);
Amaury Denoyelleaebe26f2022-01-13 16:28:06 +01002147
2148 if (qcc) {
2149 if (!expired) {
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02002150 TRACE_DEVEL("not expired", QMUX_EV_QCC_WAKE, qcc->conn);
2151 goto requeue;
Amaury Denoyelleaebe26f2022-01-13 16:28:06 +01002152 }
2153
2154 if (!qcc_may_expire(qcc)) {
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02002155 TRACE_DEVEL("cannot expired", QMUX_EV_QCC_WAKE, qcc->conn);
Amaury Denoyelleaebe26f2022-01-13 16:28:06 +01002156 t->expire = TICK_ETERNITY;
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02002157 goto requeue;
Amaury Denoyelleaebe26f2022-01-13 16:28:06 +01002158 }
2159 }
2160
Amaury Denoyelleaebe26f2022-01-13 16:28:06 +01002161 task_destroy(t);
Amaury Denoyelleea3e0352022-02-21 10:05:16 +01002162
Amaury Denoyelle4f137572022-03-24 17:10:00 +01002163 if (!qcc) {
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02002164 TRACE_DEVEL("no more qcc", QMUX_EV_QCC_WAKE);
2165 goto out;
Amaury Denoyelle4f137572022-03-24 17:10:00 +01002166 }
Amaury Denoyelleea3e0352022-02-21 10:05:16 +01002167
Amaury Denoyelleaebe26f2022-01-13 16:28:06 +01002168 qcc->task = NULL;
2169
Amaury Denoyelle5fc05d12022-07-25 14:58:48 +02002170 /* TODO depending on the timeout condition, different shutdown mode
2171 * should be used. For http keep-alive or disabled proxy, a graceful
2172 * shutdown should occurs. For all other cases, an immediate close
2173 * seems legitimate.
2174 */
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02002175 if (qcc_is_dead(qcc)) {
2176 TRACE_STATE("releasing dead connection", QMUX_EV_QCC_WAKE, qcc->conn);
Amaury Denoyelleaebe26f2022-01-13 16:28:06 +01002177 qc_release(qcc);
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02002178 }
Amaury Denoyelleaebe26f2022-01-13 16:28:06 +01002179
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02002180 out:
Amaury Denoyelle4f137572022-03-24 17:10:00 +01002181 TRACE_LEAVE(QMUX_EV_QCC_WAKE);
Amaury Denoyelleaebe26f2022-01-13 16:28:06 +01002182 return NULL;
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02002183
2184 requeue:
2185 TRACE_LEAVE(QMUX_EV_QCC_WAKE);
2186 return t;
Amaury Denoyelleaebe26f2022-01-13 16:28:06 +01002187}
2188
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002189static int qc_init(struct connection *conn, struct proxy *prx,
2190 struct session *sess, struct buffer *input)
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002191{
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002192 struct qcc *qcc;
Amaury Denoyelle6ea78192022-03-07 15:47:02 +01002193 struct quic_transport_params *lparams, *rparams;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002194
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02002195 TRACE_ENTER(QMUX_EV_QCC_NEW);
2196
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002197 qcc = pool_alloc(pool_head_qcc);
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02002198 if (!qcc) {
2199 TRACE_ERROR("alloc failure", QMUX_EV_QCC_NEW);
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002200 goto fail_no_qcc;
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02002201 }
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002202
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002203 qcc->conn = conn;
2204 conn->ctx = qcc;
Amaury Denoyellec603de42022-07-25 11:21:46 +02002205 qcc->nb_hreq = qcc->nb_sc = 0;
Amaury Denoyellece1f30d2022-02-01 15:14:24 +01002206 qcc->flags = 0;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002207
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002208 qcc->app_ops = NULL;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002209
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002210 qcc->streams_by_id = EB_ROOT_UNIQUE;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002211
Amaury Denoyellef3b0ba72021-12-08 15:12:01 +01002212 /* Server parameters, params used for RX flow control. */
Willy Tarreau784b8682022-04-11 14:18:10 +02002213 lparams = &conn->handle.qc->rx.params;
Amaury Denoyellef3b0ba72021-12-08 15:12:01 +01002214
Amaury Denoyelle749cb642022-02-09 10:25:29 +01002215 qcc->rx.max_data = lparams->initial_max_data;
Amaury Denoyelleb9e06402022-06-10 15:16:21 +02002216 qcc->tx.sent_offsets = qcc->tx.offsets = 0;
Amaury Denoyellef3b0ba72021-12-08 15:12:01 +01002217
2218 /* Client initiated streams must respect the server flow control. */
Amaury Denoyelle749cb642022-02-09 10:25:29 +01002219 qcc->strms[QCS_CLT_BIDI].max_streams = lparams->initial_max_streams_bidi;
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002220 qcc->strms[QCS_CLT_BIDI].nb_streams = 0;
Amaury Denoyellef3b0ba72021-12-08 15:12:01 +01002221 qcc->strms[QCS_CLT_BIDI].rx.max_data = 0;
Amaury Denoyelle749cb642022-02-09 10:25:29 +01002222 qcc->strms[QCS_CLT_BIDI].tx.max_data = lparams->initial_max_stream_data_bidi_remote;
Amaury Denoyellef3b0ba72021-12-08 15:12:01 +01002223
Amaury Denoyelle749cb642022-02-09 10:25:29 +01002224 qcc->strms[QCS_CLT_UNI].max_streams = lparams->initial_max_streams_uni;
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002225 qcc->strms[QCS_CLT_UNI].nb_streams = 0;
Amaury Denoyellef3b0ba72021-12-08 15:12:01 +01002226 qcc->strms[QCS_CLT_UNI].rx.max_data = 0;
Amaury Denoyelle749cb642022-02-09 10:25:29 +01002227 qcc->strms[QCS_CLT_UNI].tx.max_data = lparams->initial_max_stream_data_uni;
Amaury Denoyellef3b0ba72021-12-08 15:12:01 +01002228
2229 /* Server initiated streams must respect the server flow control. */
2230 qcc->strms[QCS_SRV_BIDI].max_streams = 0;
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002231 qcc->strms[QCS_SRV_BIDI].nb_streams = 0;
Amaury Denoyelle749cb642022-02-09 10:25:29 +01002232 qcc->strms[QCS_SRV_BIDI].rx.max_data = lparams->initial_max_stream_data_bidi_local;
Amaury Denoyellef3b0ba72021-12-08 15:12:01 +01002233 qcc->strms[QCS_SRV_BIDI].tx.max_data = 0;
2234
2235 qcc->strms[QCS_SRV_UNI].max_streams = 0;
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002236 qcc->strms[QCS_SRV_UNI].nb_streams = 0;
Amaury Denoyelle749cb642022-02-09 10:25:29 +01002237 qcc->strms[QCS_SRV_UNI].rx.max_data = lparams->initial_max_stream_data_uni;
Amaury Denoyellef3b0ba72021-12-08 15:12:01 +01002238 qcc->strms[QCS_SRV_UNI].tx.max_data = 0;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002239
Amaury Denoyellec985cb12022-05-16 14:29:59 +02002240 LIST_INIT(&qcc->lfctl.frms);
Amaury Denoyelle78396e52022-03-21 17:13:32 +01002241 qcc->lfctl.ms_bidi = qcc->lfctl.ms_bidi_init = lparams->initial_max_streams_bidi;
Amaury Denoyellebf3c2082022-08-16 11:29:08 +02002242 qcc->lfctl.ms_uni = lparams->initial_max_streams_uni;
Amaury Denoyelle44d09122022-04-26 11:21:10 +02002243 qcc->lfctl.msd_bidi_l = lparams->initial_max_stream_data_bidi_local;
2244 qcc->lfctl.msd_bidi_r = lparams->initial_max_stream_data_bidi_remote;
Amaury Denoyelle176174f2022-10-21 17:02:18 +02002245 qcc->lfctl.msd_uni_r = lparams->initial_max_stream_data_uni;
Amaury Denoyelle78396e52022-03-21 17:13:32 +01002246 qcc->lfctl.cl_bidi_r = 0;
Amaury Denoyellec055e302022-02-07 16:09:06 +01002247
Amaury Denoyellec830e1e2022-05-16 16:19:59 +02002248 qcc->lfctl.md = qcc->lfctl.md_init = lparams->initial_max_data;
Amaury Denoyelled46b0f52022-05-20 15:05:07 +02002249 qcc->lfctl.offsets_recv = qcc->lfctl.offsets_consume = 0;
Amaury Denoyellec830e1e2022-05-16 16:19:59 +02002250
Willy Tarreau784b8682022-04-11 14:18:10 +02002251 rparams = &conn->handle.qc->tx.params;
Amaury Denoyelle05ce55e2022-03-08 10:35:42 +01002252 qcc->rfctl.md = rparams->initial_max_data;
Amaury Denoyelle6ea78192022-03-07 15:47:02 +01002253 qcc->rfctl.msd_bidi_l = rparams->initial_max_stream_data_bidi_local;
2254 qcc->rfctl.msd_bidi_r = rparams->initial_max_stream_data_bidi_remote;
Amaury Denoyelle176174f2022-10-21 17:02:18 +02002255 qcc->rfctl.msd_uni_l = rparams->initial_max_stream_data_uni;
Amaury Denoyelle6ea78192022-03-07 15:47:02 +01002256
Amaury Denoyellea509ffb2022-07-04 15:50:33 +02002257 if (conn_is_back(conn)) {
2258 qcc->next_bidi_l = 0x00;
2259 qcc->largest_bidi_r = 0x01;
2260 qcc->next_uni_l = 0x02;
2261 qcc->largest_uni_r = 0x03;
2262 }
2263 else {
2264 qcc->largest_bidi_r = 0x00;
2265 qcc->next_bidi_l = 0x01;
2266 qcc->largest_uni_r = 0x02;
2267 qcc->next_uni_l = 0x03;
2268 }
2269
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002270 qcc->wait_event.tasklet = tasklet_new();
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02002271 if (!qcc->wait_event.tasklet) {
2272 TRACE_ERROR("taslket alloc failure", QMUX_EV_QCC_NEW);
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002273 goto fail_no_tasklet;
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02002274 }
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002275
Amaury Denoyelle20f2a422023-01-03 14:39:24 +01002276 LIST_INIT(&qcc->send_list);
Amaury Denoyelle1b2dba52022-04-15 17:32:04 +02002277
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002278 qcc->wait_event.tasklet->process = qc_io_cb;
2279 qcc->wait_event.tasklet->context = qcc;
Frédéric Lécaillef27b66f2022-03-18 22:49:22 +01002280 qcc->wait_event.events = 0;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002281
Amaury Denoyelle07bf8f42022-07-22 16:16:03 +02002282 qcc->proxy = prx;
Amaury Denoyelleaebe26f2022-01-13 16:28:06 +01002283 /* haproxy timeouts */
Amaury Denoyellea3daaec2022-04-21 16:29:27 +02002284 qcc->task = NULL;
Amaury Denoyelleeb7d3202023-02-08 15:55:24 +01002285 if (conn_is_back(qcc->conn)) {
2286 qcc->timeout = prx->timeout.server;
2287 qcc->shut_timeout = tick_isset(prx->timeout.serverfin) ?
2288 prx->timeout.serverfin : prx->timeout.server;
2289 }
2290 else {
2291 qcc->timeout = prx->timeout.client;
2292 qcc->shut_timeout = tick_isset(prx->timeout.clientfin) ?
2293 prx->timeout.clientfin : prx->timeout.client;
2294 }
2295
Amaury Denoyellea3daaec2022-04-21 16:29:27 +02002296 if (tick_isset(qcc->timeout)) {
2297 qcc->task = task_new_here();
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02002298 if (!qcc->task) {
2299 TRACE_ERROR("timeout task alloc failure", QMUX_EV_QCC_NEW);
Amaury Denoyellea3daaec2022-04-21 16:29:27 +02002300 goto fail_no_timeout_task;
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02002301 }
Amaury Denoyellea3daaec2022-04-21 16:29:27 +02002302 qcc->task->process = qc_timeout_task;
2303 qcc->task->context = qcc;
2304 qcc->task->expire = tick_add(now_ms, qcc->timeout);
2305 }
Amaury Denoyellebd6ec1b2022-07-25 11:53:18 +02002306 qcc_reset_idle_start(qcc);
Amaury Denoyelle30e260e2022-08-03 11:17:57 +02002307 LIST_INIT(&qcc->opening_list);
Amaury Denoyelleaebe26f2022-01-13 16:28:06 +01002308
Amaury Denoyelle0e0969d2022-01-31 15:41:14 +01002309 if (!conn_is_back(conn)) {
2310 if (!LIST_INLIST(&conn->stopping_list)) {
2311 LIST_APPEND(&mux_stopping_data[tid].list,
2312 &conn->stopping_list);
2313 }
2314 }
2315
Willy Tarreau784b8682022-04-11 14:18:10 +02002316 HA_ATOMIC_STORE(&conn->handle.qc->qcc, qcc);
Amaury Denoyelleb4d119f2023-01-25 17:44:36 +01002317
2318 if (qcc_install_app_ops(qcc, conn->handle.qc->app_ops)) {
2319 TRACE_PROTO("Cannot install app layer", QMUX_EV_QCC_NEW, qcc->conn);
2320 /* prepare a CONNECTION_CLOSE frame */
2321 quic_set_connection_close(conn->handle.qc, quic_err_transport(QC_ERR_APPLICATION_ERROR));
2322 goto fail_install_app_ops;
2323 }
2324
Frédéric Lécaille9969adb2023-01-18 11:52:21 +01002325 if (qcc->app_ops == &h3_ops)
2326 proxy_inc_fe_cum_sess_ver_ctr(sess->listener, prx, 3);
2327
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002328 /* init read cycle */
2329 tasklet_wakeup(qcc->wait_event.tasklet);
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002330
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02002331 TRACE_LEAVE(QMUX_EV_QCC_NEW, qcc->conn);
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002332 return 0;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002333
Amaury Denoyelleb4d119f2023-01-25 17:44:36 +01002334 fail_install_app_ops:
2335 if (qcc->app_ops && qcc->app_ops->release)
2336 qcc->app_ops->release(qcc->ctx);
Amaury Denoyelleaebe26f2022-01-13 16:28:06 +01002337 fail_no_timeout_task:
2338 tasklet_free(qcc->wait_event.tasklet);
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002339 fail_no_tasklet:
2340 pool_free(pool_head_qcc, qcc);
2341 fail_no_qcc:
Amaury Denoyelle4c9a1642022-08-10 16:58:01 +02002342 TRACE_LEAVE(QMUX_EV_QCC_NEW);
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002343 return -1;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002344}
2345
Amaury Denoyelle2461bd52022-04-13 16:54:52 +02002346static void qc_destroy(void *ctx)
2347{
2348 struct qcc *qcc = ctx;
2349
2350 TRACE_ENTER(QMUX_EV_QCC_END, qcc->conn);
2351 qc_release(qcc);
2352 TRACE_LEAVE(QMUX_EV_QCC_END);
2353}
2354
Willy Tarreaud7b7e0d2022-05-27 16:09:35 +02002355static void qc_detach(struct sedesc *sd)
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002356{
Willy Tarreaud7b7e0d2022-05-27 16:09:35 +02002357 struct qcs *qcs = sd->se;
Amaury Denoyelle916f0ac2021-12-06 16:03:47 +01002358 struct qcc *qcc = qcs->qcc;
2359
Amaury Denoyelle4f137572022-03-24 17:10:00 +01002360 TRACE_ENTER(QMUX_EV_STRM_END, qcc->conn, qcs);
Amaury Denoyelle916f0ac2021-12-06 16:03:47 +01002361
Amaury Denoyelle38e60062022-07-01 16:48:42 +02002362 /* TODO this BUG_ON_HOT() is not correct as the stconn layer may detach
2363 * from the stream even if it is not closed remotely at the QUIC layer.
2364 * This happens for example when a stream must be closed due to a
2365 * rejected request. To better handle these cases, it will be required
2366 * to implement shutr/shutw MUX operations. Once this is done, this
2367 * BUG_ON_HOT() statement can be adjusted.
2368 */
2369 //BUG_ON_HOT(!qcs_is_close_remote(qcs));
Amaury Denoyellec603de42022-07-25 11:21:46 +02002370
2371 qcc_rm_sc(qcc);
Amaury Denoyelle06890aa2022-04-04 16:15:06 +02002372
Amaury Denoyelle20d1f842022-07-11 11:23:17 +02002373 if (!qcs_is_close_local(qcs) && !(qcc->conn->flags & CO_FL_ERROR)) {
Amaury Denoyelle047d86a2022-08-10 16:42:35 +02002374 TRACE_STATE("remaining data, detaching qcs", QMUX_EV_STRM_END, qcc->conn, qcs);
Amaury Denoyelle2873a312021-12-08 14:42:55 +01002375 qcs->flags |= QC_SF_DETACH;
Amaury Denoyelle5fc05d12022-07-25 14:58:48 +02002376 qcc_refresh_timeout(qcc);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02002377
2378 TRACE_LEAVE(QMUX_EV_STRM_END, qcc->conn, qcs);
Amaury Denoyelle2873a312021-12-08 14:42:55 +01002379 return;
2380 }
2381
Amaury Denoyelle916f0ac2021-12-06 16:03:47 +01002382 qcs_destroy(qcs);
Amaury Denoyelle1136e922022-02-01 10:33:09 +01002383
Amaury Denoyelle06890aa2022-04-04 16:15:06 +02002384 if (qcc_is_dead(qcc)) {
Amaury Denoyelle047d86a2022-08-10 16:42:35 +02002385 TRACE_STATE("killing dead connection", QMUX_EV_STRM_END, qcc->conn);
Amaury Denoyelle35a66c02022-08-12 15:56:21 +02002386 goto release;
Amaury Denoyelle06890aa2022-04-04 16:15:06 +02002387 }
Amaury Denoyellea3daaec2022-04-21 16:29:27 +02002388 else if (qcc->task) {
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02002389 TRACE_DEVEL("refreshing connection's timeout", QMUX_EV_STRM_END, qcc->conn);
Amaury Denoyelle5fc05d12022-07-25 14:58:48 +02002390 qcc_refresh_timeout(qcc);
Amaury Denoyelle916f0ac2021-12-06 16:03:47 +01002391 }
Amaury Denoyelle5fc05d12022-07-25 14:58:48 +02002392 else {
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02002393 TRACE_DEVEL("completed", QMUX_EV_STRM_END, qcc->conn);
Amaury Denoyelle5fc05d12022-07-25 14:58:48 +02002394 }
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02002395
2396 TRACE_LEAVE(QMUX_EV_STRM_END, qcc->conn);
Amaury Denoyelle35a66c02022-08-12 15:56:21 +02002397 return;
2398
2399 release:
2400 qc_release(qcc);
2401 TRACE_LEAVE(QMUX_EV_STRM_END);
2402 return;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002403}
2404
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002405/* Called from the upper layer, to receive data */
Amaury Denoyelleb7ce79c2022-11-24 10:51:19 +01002406static size_t qc_recv_buf(struct stconn *sc, struct buffer *buf,
2407 size_t count, int flags)
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002408{
Willy Tarreau3215e732022-05-27 10:09:11 +02002409 struct qcs *qcs = __sc_mux_strm(sc);
Amaury Denoyelle9a327a72022-02-14 17:11:09 +01002410 size_t ret = 0;
Amaury Denoyelleeb53e5b2022-02-14 17:11:32 +01002411 char fin = 0;
Amaury Denoyelle9a327a72022-02-14 17:11:09 +01002412
Amaury Denoyelle4f137572022-03-24 17:10:00 +01002413 TRACE_ENTER(QMUX_EV_STRM_RECV, qcs->qcc->conn, qcs);
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002414
Amaury Denoyelled80fbca2022-09-19 17:02:28 +02002415 ret = qcs_http_rcv_buf(qcs, buf, count, &fin);
Amaury Denoyelle9a327a72022-02-14 17:11:09 +01002416
Amaury Denoyelle9a327a72022-02-14 17:11:09 +01002417 if (b_data(&qcs->rx.app_buf)) {
Willy Tarreaud7b7e0d2022-05-27 16:09:35 +02002418 se_fl_set(qcs->sd, SE_FL_RCV_MORE | SE_FL_WANT_ROOM);
Amaury Denoyelle9a327a72022-02-14 17:11:09 +01002419 }
2420 else {
Willy Tarreaud7b7e0d2022-05-27 16:09:35 +02002421 se_fl_clr(qcs->sd, SE_FL_RCV_MORE | SE_FL_WANT_ROOM);
2422 if (se_fl_test(qcs->sd, SE_FL_ERR_PENDING))
2423 se_fl_set(qcs->sd, SE_FL_ERROR);
Amaury Denoyelle9a327a72022-02-14 17:11:09 +01002424
Amaury Denoyelle72a78e82022-07-29 15:34:12 +02002425 /* Set end-of-input if FIN received and all data extracted. */
Amaury Denoyelleeb53e5b2022-02-14 17:11:32 +01002426 if (fin)
Willy Tarreaud7b7e0d2022-05-27 16:09:35 +02002427 se_fl_set(qcs->sd, SE_FL_EOI);
Amaury Denoyelle9a327a72022-02-14 17:11:09 +01002428
2429 if (b_size(&qcs->rx.app_buf)) {
2430 b_free(&qcs->rx.app_buf);
2431 offer_buffers(NULL, 1);
2432 }
2433 }
2434
Amaury Denoyellef1fc0b32022-05-02 11:07:06 +02002435 if (ret) {
2436 qcs->flags &= ~QC_SF_DEM_FULL;
Amaury Denoyelle9a327a72022-02-14 17:11:09 +01002437 tasklet_wakeup(qcs->qcc->wait_event.tasklet);
Amaury Denoyellef1fc0b32022-05-02 11:07:06 +02002438 }
Amaury Denoyelle9a327a72022-02-14 17:11:09 +01002439
Amaury Denoyelle4f137572022-03-24 17:10:00 +01002440 TRACE_LEAVE(QMUX_EV_STRM_RECV, qcs->qcc->conn, qcs);
2441
Amaury Denoyelle9a327a72022-02-14 17:11:09 +01002442 return ret;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002443}
2444
Amaury Denoyelleb7ce79c2022-11-24 10:51:19 +01002445static size_t qc_send_buf(struct stconn *sc, struct buffer *buf,
2446 size_t count, int flags)
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002447{
Willy Tarreau3215e732022-05-27 10:09:11 +02002448 struct qcs *qcs = __sc_mux_strm(sc);
Amaury Denoyelle4f137572022-03-24 17:10:00 +01002449 size_t ret;
Amaury Denoyelle9534e592022-09-19 17:14:27 +02002450 char fin;
Amaury Denoyelle4f137572022-03-24 17:10:00 +01002451
2452 TRACE_ENTER(QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002453
Amaury Denoyelle3dc4e5a2022-09-13 16:49:21 +02002454 /* stream layer has been detached so no transfer must occur after. */
2455 BUG_ON_HOT(qcs->flags & QC_SF_DETACH);
2456
Amaury Denoyelle843a1192022-07-04 11:44:38 +02002457 if (qcs_is_close_local(qcs) || (qcs->flags & QC_SF_TO_RESET)) {
Amaury Denoyelle0ed617a2022-09-20 14:46:40 +02002458 ret = qcs_http_reset_buf(qcs, buf, count);
Amaury Denoyelle38e60062022-07-01 16:48:42 +02002459 goto end;
2460 }
2461
Amaury Denoyelle9534e592022-09-19 17:14:27 +02002462 ret = qcs_http_snd_buf(qcs, buf, count, &fin);
2463 if (fin)
2464 qcs->flags |= QC_SF_FIN_STREAM;
2465
Amaury Denoyelleab6cdec2023-01-10 10:41:41 +01002466 if (ret || fin) {
Amaury Denoyellef9b03262023-01-09 10:34:25 +01002467 qcc_send_stream(qcs, 0);
Amaury Denoyelle9534e592022-09-19 17:14:27 +02002468 if (!(qcs->qcc->wait_event.events & SUB_RETRY_SEND))
2469 tasklet_wakeup(qcs->qcc->wait_event.tasklet);
2470 }
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002471
Amaury Denoyelle38e60062022-07-01 16:48:42 +02002472 end:
Amaury Denoyelle4f137572022-03-24 17:10:00 +01002473 TRACE_LEAVE(QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
2474
2475 return ret;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002476}
2477
2478/* Called from the upper layer, to subscribe <es> to events <event_type>. The
2479 * event subscriber <es> is not allowed to change from a previous call as long
2480 * as at least one event is still subscribed. The <event_type> must only be a
2481 * combination of SUB_RETRY_RECV and SUB_RETRY_SEND. It always returns 0.
2482 */
Willy Tarreau3215e732022-05-27 10:09:11 +02002483static int qc_subscribe(struct stconn *sc, int event_type,
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002484 struct wait_event *es)
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002485{
Willy Tarreau3215e732022-05-27 10:09:11 +02002486 return qcs_subscribe(__sc_mux_strm(sc), event_type, es);
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002487}
2488
2489/* Called from the upper layer, to unsubscribe <es> from events <event_type>.
2490 * The <es> pointer is not allowed to differ from the one passed to the
2491 * subscribe() call. It always returns zero.
2492 */
Willy Tarreau3215e732022-05-27 10:09:11 +02002493static int qc_unsubscribe(struct stconn *sc, int event_type, struct wait_event *es)
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002494{
Willy Tarreau3215e732022-05-27 10:09:11 +02002495 struct qcs *qcs = __sc_mux_strm(sc);
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002496
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002497 BUG_ON(event_type & ~(SUB_RETRY_SEND|SUB_RETRY_RECV));
2498 BUG_ON(qcs->subs && qcs->subs != es);
2499
2500 es->events &= ~event_type;
2501 if (!es->events)
2502 qcs->subs = NULL;
2503
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002504 return 0;
2505}
2506
Amaury Denoyellefe035ec2022-04-06 15:46:30 +02002507/* Loop through all qcs from <qcc>. If CO_FL_ERROR is set on the connection,
Willy Tarreau4596fe22022-05-17 19:07:51 +02002508 * report SE_FL_ERR_PENDING|SE_FL_ERROR on the attached stream connectors and
2509 * wake them.
Amaury Denoyellefe035ec2022-04-06 15:46:30 +02002510 */
2511static int qc_wake_some_streams(struct qcc *qcc)
2512{
Amaury Denoyellefe035ec2022-04-06 15:46:30 +02002513 struct qcs *qcs;
2514 struct eb64_node *node;
2515
2516 for (node = eb64_first(&qcc->streams_by_id); node;
2517 node = eb64_next(node)) {
Amaury Denoyellee4301da2022-04-19 17:59:50 +02002518 qcs = eb64_entry(node, struct qcs, by_id);
Amaury Denoyellefe035ec2022-04-06 15:46:30 +02002519
Amaury Denoyelle3abeb572022-07-04 11:42:27 +02002520 if (!qcs_sc(qcs))
Amaury Denoyellefe035ec2022-04-06 15:46:30 +02002521 continue;
2522
2523 if (qcc->conn->flags & CO_FL_ERROR) {
Willy Tarreau35c4dd02023-01-17 16:25:29 +01002524 se_fl_set_error(qcs->sd);
Amaury Denoyelle4561f842022-07-06 14:54:34 +02002525 qcs_alert(qcs);
Amaury Denoyellefe035ec2022-04-06 15:46:30 +02002526 }
2527 }
2528
2529 return 0;
2530}
2531
Amaury Denoyelle0e0969d2022-01-31 15:41:14 +01002532static int qc_wake(struct connection *conn)
2533{
2534 struct qcc *qcc = conn->ctx;
Amaury Denoyelled97fc802022-04-06 16:13:09 +02002535
2536 TRACE_ENTER(QMUX_EV_QCC_WAKE, conn);
Amaury Denoyelle0e0969d2022-01-31 15:41:14 +01002537
Willy Tarreau784b8682022-04-11 14:18:10 +02002538 if (conn->handle.qc->flags & QUIC_FL_CONN_NOTIFY_CLOSE)
Amaury Denoyelleb515b0a2022-04-06 10:28:43 +02002539 qcc->conn->flags |= (CO_FL_SOCK_RD_SH|CO_FL_SOCK_WR_SH);
2540
Amaury Denoyelled97fc802022-04-06 16:13:09 +02002541 qc_send(qcc);
2542
Amaury Denoyelle14dbb842023-01-24 18:19:47 +01002543 if (qc_process(qcc)) {
2544 TRACE_STATE("releasing dead connection", QMUX_EV_QCC_WAKE, qcc->conn);
Amaury Denoyelled97fc802022-04-06 16:13:09 +02002545 goto release;
Amaury Denoyelle14dbb842023-01-24 18:19:47 +01002546 }
2547
2548 qc_wake_some_streams(qcc);
Amaury Denoyelled97fc802022-04-06 16:13:09 +02002549
Amaury Denoyelle5fc05d12022-07-25 14:58:48 +02002550 qcc_refresh_timeout(qcc);
2551
Amaury Denoyelled97fc802022-04-06 16:13:09 +02002552 TRACE_LEAVE(QMUX_EV_QCC_WAKE, conn);
Amaury Denoyelled97fc802022-04-06 16:13:09 +02002553 return 0;
Amaury Denoyelle0e0969d2022-01-31 15:41:14 +01002554
Amaury Denoyelled97fc802022-04-06 16:13:09 +02002555 release:
2556 qc_release(qcc);
Amaury Denoyellef0b67f92022-08-10 16:14:32 +02002557 TRACE_LEAVE(QMUX_EV_QCC_WAKE);
Amaury Denoyelle0e0969d2022-01-31 15:41:14 +01002558 return 1;
2559}
2560
Amaury Denoyellea473f192022-12-21 10:21:58 +01002561static void qc_shutw(struct stconn *sc, enum co_shw_mode mode)
2562{
2563 struct qcs *qcs = __sc_mux_strm(sc);
2564
2565 TRACE_ENTER(QMUX_EV_STRM_SHUT, qcs->qcc->conn, qcs);
2566
2567 /* If QC_SF_FIN_STREAM is not set and stream is not closed locally, it
2568 * means that upper layer reported an early closure. A RESET_STREAM is
2569 * necessary if not already scheduled.
2570 */
2571
2572 if (!qcs_is_close_local(qcs) &&
2573 !(qcs->flags & (QC_SF_FIN_STREAM|QC_SF_TO_RESET))) {
2574 qcc_reset_stream(qcs, 0);
2575 se_fl_set_error(qcs->sd);
2576 }
2577
2578 TRACE_LEAVE(QMUX_EV_STRM_SHUT, qcs->qcc->conn, qcs);
2579}
Amaury Denoyelle38e60062022-07-01 16:48:42 +02002580
Willy Tarreaub4a4fee2022-09-02 16:00:40 +02002581/* for debugging with CLI's "show sess" command. May emit multiple lines, each
2582 * new one being prefixed with <pfx>, if <pfx> is not NULL, otherwise a single
2583 * line is used. Each field starts with a space so it's safe to print it after
2584 * existing fields.
2585 */
2586static int qc_show_sd(struct buffer *msg, struct sedesc *sd, const char *pfx)
2587{
2588 struct qcs *qcs = sd->se;
2589 struct qcc *qcc;
2590 int ret = 0;
2591
2592 if (!qcs)
2593 return ret;
2594
2595 chunk_appendf(msg, " qcs=%p .flg=%#x .id=%llu .st=%s .ctx=%p, .err=%#llx",
2596 qcs, qcs->flags, (ull)qcs->id, qcs_st_to_str(qcs->st), qcs->ctx, (ull)qcs->err);
2597
2598 if (pfx)
2599 chunk_appendf(msg, "\n%s", pfx);
2600
2601 qcc = qcs->qcc;
2602 chunk_appendf(msg, " qcc=%p .flg=%#x .nbsc=%llu .nbhreq=%llu, .task=%p",
2603 qcc, qcc->flags, (ull)qcc->nb_sc, (ull)qcc->nb_hreq, qcc->task);
2604 return ret;
2605}
2606
2607
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002608static const struct mux_ops qc_ops = {
2609 .init = qc_init,
Amaury Denoyelle2461bd52022-04-13 16:54:52 +02002610 .destroy = qc_destroy,
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002611 .detach = qc_detach,
Amaury Denoyelleb7ce79c2022-11-24 10:51:19 +01002612 .rcv_buf = qc_recv_buf,
2613 .snd_buf = qc_send_buf,
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002614 .subscribe = qc_subscribe,
2615 .unsubscribe = qc_unsubscribe,
Amaury Denoyelle0e0969d2022-01-31 15:41:14 +01002616 .wake = qc_wake,
Amaury Denoyellea473f192022-12-21 10:21:58 +01002617 .shutw = qc_shutw,
Willy Tarreaub4a4fee2022-09-02 16:00:40 +02002618 .show_sd = qc_show_sd,
Willy Tarreaub5821e12022-04-26 11:54:08 +02002619 .flags = MX_FL_HTX|MX_FL_NO_UPG|MX_FL_FRAMED,
Willy Tarreau671bd5a2022-04-11 09:29:21 +02002620 .name = "QUIC",
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002621};
2622
2623static struct mux_proto_list mux_proto_quic =
Amaury Denoyelledeed7772021-12-03 11:36:46 +01002624 { .token = IST("quic"), .mode = PROTO_MODE_HTTP, .side = PROTO_SIDE_FE, .mux = &qc_ops };
Frédéric Lécailledfbae762021-02-18 09:59:01 +01002625
2626INITCALL1(STG_REGISTER, register_mux_proto, &mux_proto_quic);