blob: df6d147960be0c975e27eca5dceb0ec1360c90ee [file] [log] [blame]
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001/*
2 * QUIC mux-demux for connections
3 *
4 * Copyright 2021 HAProxy Technologies, Frédéric Lécaille <flecaille@haproxy.com>
5 *
6 * This program is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU General Public License
8 * as published by the Free Software Foundation; either version
9 * 2 of the License, or (at your option) any later version.
10 *
11 */
12
Amaury Denoyelleeb01f592021-10-07 16:44:05 +020013#include <import/eb64tree.h>
14
Frédéric Lécailledfbae762021-02-18 09:59:01 +010015#include <haproxy/api.h>
16#include <haproxy/cfgparse.h>
17#include <haproxy/connection.h>
18#include <haproxy/h3.h>
19#include <haproxy/istbuf.h>
20#include <haproxy/log.h>
21#include <haproxy/mux_quic.h>
22#include <haproxy/net_helper.h>
23#include <haproxy/quic_frame.h>
24#include <haproxy/session-t.h>
Amaury Denoyelleeb01f592021-10-07 16:44:05 +020025#include <haproxy/ssl_sock-t.h>
Frédéric Lécailledfbae762021-02-18 09:59:01 +010026#include <haproxy/stats.h>
27#include <haproxy/stream.h>
28#include <haproxy/stream_interface.h>
29#include <haproxy/trace.h>
30
31/* dummy streams returned for closed, error, refused, idle and states */
32static const struct qcs *qc_closed_stream;
33
Frédéric Lécailledfbae762021-02-18 09:59:01 +010034#define QC_SS_MASK(state) (1UL << (state))
35#define QC_SS_IDLE_BIT (1UL << QC_SS_IDLE)
36#define QC_SS_RLOC_BIT (1UL << QC_SS_RLOC)
37#define QC_SS_RREM_BIT (1UL << QC_SS_RREM)
38#define QC_SS_OPEN_BIT (1UL << QC_SS_OPEN)
39#define QC_SS_HREM_BIT (1UL << QC_SS_HREM)
40#define QC_SS_HLOC_BIT (1UL << QC_SS_HLOC)
41#define QC_SS_ERROR_BIT (1UL << QC_SS_ERROR)
42#define QC_SS_CLOSED_BIT (1UL << QC_SS_CLOSED)
43
44
45/* trace source and events */
46static void qc_trace(enum trace_level level, uint64_t mask, \
47 const struct trace_source *src,
48 const struct ist where, const struct ist func,
49 const void *a1, const void *a2, const void *a3, const void *a4);
50
51/* The event representation is split like this :
52 * strm - application layer
53 * qcs - internal QUIC stream
54 * qcc - internal QUIC connection
55 * conn - external connection
56 *
57 */
58static const struct trace_event qc_trace_events[] = {
59#define QC_EV_QCC_NEW (1ULL << 0)
60 { .mask = QC_EV_QCC_NEW, .name = "qcc_new", .desc = "new QUIC connection" },
61#define QC_EV_QCC_RECV (1ULL << 1)
62 { .mask = QC_EV_QCC_RECV, .name = "qcc_recv", .desc = "Rx on QUIC connection" },
63#define QC_EV_QCC_SEND (1ULL << 2)
64 { .mask = QC_EV_QCC_SEND, .name = "qcc_send", .desc = "Tx on QUIC connection" },
65#define QC_EV_QCC_FCTL (1ULL << 3)
66 { .mask = QC_EV_QCC_FCTL, .name = "qcc_fctl", .desc = "QUIC connection flow-controlled" },
67#define QC_EV_QCC_BLK (1ULL << 4)
68 { .mask = QC_EV_QCC_BLK, .name = "qcc_blk", .desc = "QUIC connection blocked" },
69#define QC_EV_QCC_WAKE (1ULL << 5)
70 { .mask = QC_EV_QCC_WAKE, .name = "qcc_wake", .desc = "QUIC connection woken up" },
71#define QC_EV_QCC_END (1ULL << 6)
72 { .mask = QC_EV_QCC_END, .name = "qcc_end", .desc = "QUIC connection terminated" },
73#define QC_EV_QCC_ERR (1ULL << 7)
74 { .mask = QC_EV_QCC_ERR, .name = "qcc_err", .desc = "error on QUIC connection" },
75#define QC_EV_TX_FRAME (1ULL << 8)
76 { .mask = QC_EV_TX_FRAME, .name = "tx_frame", .desc = "transmission of any QUIC frame" },
77#define QC_EV_QCS_NEW (1ULL << 9)
78 { .mask = QC_EV_QCS_NEW, .name = "qcs_new", .desc = "new QUIC stream" },
79#define QC_EV_QCS_GET (1ULL << 10)
80 { .mask = QC_EV_QCS_GET, .name = "qcs_get", .desc = "get QUIC stream by ID" },
81#define QC_EV_QCS_SEND (1ULL << 11)
82 { .mask = QC_EV_QCS_SEND, .name = "qcs_send", .desc = "Tx for QUIC stream" },
83#define QC_EV_QCS_FCTL (1ULL << 12)
84 { .mask = QC_EV_QCS_FCTL, .name = "qcs_fctl", .desc = "QUIC stream flow-controlled" },
85#define QC_EV_QCS_BLK (1ULL << 13)
86 { .mask = QC_EV_QCS_BLK, .name = "qcs_blk", .desc = "QUIC stream blocked" },
87#define QC_EV_QCS_WAKE (1ULL << 14)
88 { .mask = QC_EV_QCS_WAKE, .name = "qcs_wake", .desc = "QUIC stream woken up" },
89#define QC_EV_QCS_END (1ULL << 15)
90 { .mask = QC_EV_QCS_END, .name = "qcs_end", .desc = "QUIC stream terminated" },
91#define QC_EV_QCS_ERR (1ULL << 16)
92 { .mask = QC_EV_QCS_ERR, .name = "qcs_err", .desc = "error on QUIC stream" },
93#define QC_EV_STRM_NEW (1ULL << 17)
94 { .mask = QC_EV_STRM_NEW, .name = "strm_new", .desc = "app-layer stream creation" },
95#define QC_EV_STRM_RECV (1ULL << 18)
96 { .mask = QC_EV_STRM_RECV, .name = "strm_recv", .desc = "receiving data for stream" },
97#define QC_EV_STRM_SEND (1ULL << 19)
98 { .mask = QC_EV_STRM_SEND, .name = "strm_send", .desc = "sending data for stream" },
99#define QC_EV_STRM_FULL (1ULL << 20)
100 { .mask = QC_EV_STRM_FULL, .name = "strm_full", .desc = "stream buffer full" },
101#define QC_EV_STRM_WAKE (1ULL << 21)
102 { .mask = QC_EV_STRM_WAKE, .name = "strm_wake", .desc = "stream woken up" },
103#define QC_EV_STRM_SHUT (1ULL << 22)
104 { .mask = QC_EV_STRM_SHUT, .name = "strm_shut", .desc = "stream shutdown" },
105#define QC_EV_STRM_END (1ULL << 23)
106 { .mask = QC_EV_STRM_END, .name = "strm_end", .desc = "detaching app-layer stream" },
107#define QC_EV_STRM_ERR (1ULL << 24)
108 { .mask = QC_EV_STRM_ERR, .name = "strm_err", .desc = "stream error" },
109 { }
110};
111
112static const struct name_desc qc_trace_lockon_args[4] = {
113 /* arg1 */ { /* already used by the connection */ },
114 /* arg2 */ { .name = "qcs", .desc = "QUIC stream" },
115 /* arg3 */ { },
116 /* arg4 */ { }
117};
118
119static const struct name_desc qc_trace_decoding[] = {
120#define QC_VERB_CLEAN 1
121 { .name="clean", .desc="only user-friendly stuff, generally suitable for level \"user\"" },
122#define QC_VERB_MINIMAL 2
123 { .name="minimal", .desc="report only qcc/qcs state and flags, no real decoding" },
124#define QC_VERB_SIMPLE 3
125 { .name="simple", .desc="add request/response status line or frame info when available" },
126#define QC_VERB_ADVANCED 4
127 { .name="advanced", .desc="add header fields or frame decoding when available" },
128#define QC_VERB_COMPLETE 5
129 { .name="complete", .desc="add full data dump when available" },
130 { /* end */ }
131};
132
133static struct trace_source trace_mux_quic = {
134 .name = IST("mux_quic"),
135 .desc = "QUIC multiplexer",
136 .arg_def = TRC_ARG1_CONN, // TRACE()'s first argument is always a connection
137 .default_cb = qc_trace,
138 .known_events = qc_trace_events,
139 .lockon_args = qc_trace_lockon_args,
140 .decoding = qc_trace_decoding,
141 .report_events = ~0, // report everything by default
142};
143
144#define TRACE_SOURCE &trace_mux_quic
145INITCALL1(STG_REGISTER, trace_register_source, TRACE_SOURCE);
146
147/* quic stats module */
148enum {
149 QC_ST_RESET_STREAM_RCVD,
150
151 QC_ST_CONN_PROTO_ERR,
152 QC_ST_STRM_PROTO_ERR,
153 QC_ST_RESET_STREAM_SENT,
154
155 QC_ST_OPEN_CONN,
156 QC_ST_OPEN_STREAM,
157 QC_ST_TOTAL_CONN,
158 QC_ST_TOTAL_STREAM,
159
160 QC_STATS_COUNT /* must be the last member of the enum */
161};
162
163static struct name_desc qc_stats[] = {
164 [QC_ST_RESET_STREAM_RCVD] = { .name = "qc_rst_stream_rcvd",
165 .desc = "Total number of received RESET_STREAM frames" },
166
167 [QC_ST_CONN_PROTO_ERR] = { .name = "qc_detected_conn_protocol_errors",
168 .desc = "Total number of connection protocol errors" },
169 [QC_ST_STRM_PROTO_ERR] = { .name = "qc_detected_strm_protocol_errors",
170 .desc = "Total number of stream protocol errors" },
171 [QC_ST_RESET_STREAM_SENT] = { .name = "qc_rst_stream_resp",
172 .desc = "Total number of RESET_STREAM sent on detected error" },
173
174 [QC_ST_OPEN_CONN] = { .name = "qc_open_connections",
175 .desc = "Count of currently open connections" },
176 [QC_ST_OPEN_STREAM] = { .name = "qc_backend_open_streams",
177 .desc = "Count of currently open streams" },
178 [QC_ST_TOTAL_CONN] = { .name = "qc_open_connections",
179 .desc = "Total number of connections" },
180 [QC_ST_TOTAL_STREAM] = { .name = "qc_backend_open_streams",
181 .desc = "Total number of streams" },
182};
183
184static struct qc_counters {
185 long long rst_stream_rcvd; /* total number of RESET_STREAM frame received */
186
187 long long conn_proto_err; /* total number of protocol errors detected */
188 long long strm_proto_err; /* total number of protocol errors detected */
189 long long rst_stream_resp; /* total number of RESET_STREAM frame sent on error */
190
191 long long open_conns; /* count of currently open connections */
192 long long open_streams; /* count of currently open streams */
193 long long total_conns; /* total number of connections */
194 long long total_streams; /* total number of streams */
195} qc_counters;
196
197static void qc_fill_stats(void *data, struct field *stats)
198{
199 struct qc_counters *counters = data;
200
201 stats[QC_ST_RESET_STREAM_RCVD] = mkf_u64(FN_COUNTER, counters->rst_stream_rcvd);
202
203 stats[QC_ST_CONN_PROTO_ERR] = mkf_u64(FN_COUNTER, counters->conn_proto_err);
204 stats[QC_ST_STRM_PROTO_ERR] = mkf_u64(FN_COUNTER, counters->strm_proto_err);
205 stats[QC_ST_RESET_STREAM_SENT] = mkf_u64(FN_COUNTER, counters->rst_stream_resp);
206
207 stats[QC_ST_OPEN_CONN] = mkf_u64(FN_GAUGE, counters->open_conns);
208 stats[QC_ST_OPEN_STREAM] = mkf_u64(FN_GAUGE, counters->open_streams);
209 stats[QC_ST_TOTAL_CONN] = mkf_u64(FN_COUNTER, counters->total_conns);
210 stats[QC_ST_TOTAL_STREAM] = mkf_u64(FN_COUNTER, counters->total_streams);
211}
212
213static struct stats_module qc_stats_module = {
214 .name = "quic",
215 .fill_stats = qc_fill_stats,
216 .stats = qc_stats,
217 .stats_count = QC_STATS_COUNT,
218 .counters = &qc_counters,
219 .counters_size = sizeof(qc_counters),
220 .domain_flags = MK_STATS_PROXY_DOMAIN(STATS_PX_CAP_FE|STATS_PX_CAP_BE),
221 .clearable = 1,
222};
223
224INITCALL1(STG_REGISTER, stats_register_module, &qc_stats_module);
225
226/* the qcc connection pool */
227DECLARE_STATIC_POOL(pool_head_qcc, "qcc", sizeof(struct qcc));
228/* the qcs stream pool */
229DECLARE_POOL(pool_head_qcs, "qcs", sizeof(struct qcs));
230
231static struct task *qc_timeout_task(struct task *t, void *context, unsigned int state);
232static int qc_send(struct qcc *qcc);
233static int qc_recv(struct qcc *qcc);
234static int qc_process(struct qcc *qcc);
235static struct task *qc_io_cb(struct task *t, void *ctx, unsigned int state);
236static inline struct qcs *qcc_st_by_id(struct qcc *qcc, int id);
237static struct task *qc_deferred_shut(struct task *t, void *ctx, unsigned int state);
238static struct qcs *qcc_bck_stream_new(struct qcc *qcc, int dir,
239 struct conn_stream *cs, struct session *sess);
240static void qcs_alert(struct qcs *qcs);
241
242/* returns a qcc state as an abbreviated 3-letter string, or "???" if unknown */
243static inline const char *qcc_st_to_str(enum qc_cs st)
244{
245 switch (st) {
246 case QC_CS_NOERR: return "NER";
247 default: return "???";
248 }
249}
250
251/* marks an error on the connection */
252void qc_error(struct qcc *qcc, int err)
253{
254 TRACE_POINT(QC_EV_QCC_ERR, qcc->conn, 0, 0, (void *)(long)(err));
255 qcc->errcode = err;
256 qcc->st0 = QC_CS_ERROR;
257}
258
259static inline const char *qcs_rx_st_to_str(enum qcs_rx_st st)
260{
261 switch (st) {
262 case QC_RX_SS_IDLE: return "IDL";
263 case QC_RX_SS_RECV: return "RCV";
264 case QC_RX_SS_SIZE_KNOWN: return "SKNWN";
265 case QC_RX_SS_DATA_RECVD: return "DATARCVD";
266 case QC_RX_SS_DATA_READ : return "DATAREAD";
267 case QC_RX_SS_RST_RECVD: return "RSTRCVD";
268 case QC_RX_SS_RST_READ: return "RSTREAD";
269 default: return "???";
270 }
271}
272
273static inline const char *qcs_tx_st_to_str(enum qcs_tx_st st)
274{
275 switch (st) {
276 case QC_TX_SS_IDLE: return "IDL";
277 case QC_TX_SS_READY: return "READY";
278 case QC_TX_SS_SEND: return "SEND";
279 case QC_TX_SS_DATA_SENT: return "DATASENT";
280 case QC_TX_SS_DATA_RECVD: return "DATARCVD";
281 case QC_TX_SS_RST_SENT: return "RSTSENT";
282 case QC_TX_SS_RST_RECVD: return "RSTRCVD";
283 default: return "???";
284 }
285}
286
287/* the QUIC traces always expect that arg1, if non-null, is of type connection
288 * (from which we can derive qcc), that arg2, if non-null, is of type qcs.
289 */
290static void qc_trace(enum trace_level level, uint64_t mask, const struct trace_source *src,
291 const struct ist where, const struct ist func,
292 const void *a1, const void *a2, const void *a3, const void *a4)
293{
294 const struct connection *conn = a1;
295 const struct qcc *qcc = conn ? conn->ctx : NULL;
296 const struct qcs *qcs = a2;
297
298 if (!qcc)
299 return;
300
301 if (src->verbosity > QC_VERB_CLEAN) {
302 chunk_appendf(&trace_buf, " : qcc=%p(%c,%s)",
303 qcc, conn_is_back(conn) ? 'B' : 'F', qcc_st_to_str(qcc->st0));
304 if (qcs) {
305 chunk_appendf(&trace_buf, " qcs=%p(rx.%s,tx.%s)",
306 qcs, qcs_rx_st_to_str(qcs->rx.st), qcs_tx_st_to_str(qcs->tx.st));
307 }
308 }
309}
310
311
312/* Detect a pending read0 for a QUIC connection. It happens if a read0 is pending
313 * on the connection AND if there is no more data in the demux buffer. The
314 * function returns 1 to report a read0 or 0 otherwise.
315 */
316__maybe_unused
317static int qcc_read0_pending(struct qcc *qcc)
318{
319 if (conn_xprt_read0_pending(qcc->conn) && !qcc->rx.inmux)
320 return 1;
321 return 0;
322}
323
324/* returns true if the connection is allowed to expire, false otherwise. A
325 * connection may expire when:
326 * - it has no stream
327 * - it has data in the mux buffer
328 * - it has streams in the blocked list
329 * - it has streams in the fctl list
330 * - it has streams in the send list
331 * Otherwise it means some streams are waiting in the data layer and it should
332 * not expire.
333 */
334__maybe_unused
335static inline int qcc_may_expire(const struct qcc *qcc)
336{
337 return eb_is_empty(&qcc->streams_by_id) ||
338 br_data(qcc->mbuf) ||
339 !LIST_ISEMPTY(&qcc->blocked_list) ||
340 !LIST_ISEMPTY(&qcc->fctl_list) ||
341 !LIST_ISEMPTY(&qcc->send_list);
342}
343
344static __inline int
345qcc_is_dead(const struct qcc *qcc)
346{
Amaury Denoyelleac8ee252021-10-08 17:57:03 +0200347#if 0
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100348 if (eb_is_empty(&qcc->streams_by_id) && /* don't close if streams exist */
349 ((qcc->conn->flags & CO_FL_ERROR) || /* errors close immediately */
350 (qcc->st0 >= QC_CS_ERROR && !qcc->task) || /* a timeout stroke earlier */
351 (!(qcc->conn->owner)) || /* Nobody's left to take care of the connection, drop it now */
352 (!br_data(qcc->mbuf) && /* mux buffer empty, also process clean events below */
353 conn_xprt_read0_pending(qcc->conn))))
354 return 1;
Amaury Denoyelleac8ee252021-10-08 17:57:03 +0200355#endif
356 if (!qcc->strms[QCS_CLT_BIDI].nb_streams)
357 return 1;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100358
359 return 0;
360}
361
362/*****************************************************/
363/* functions below are for dynamic buffer management */
364/*****************************************************/
365
366/* indicates whether or not the we may call the qc_recv() function to attempt
367 * to receive data into the buffer and/or demux pending data. The condition is
368 * a bit complex due to some API limits for now. The rules are the following :
369 * - if an error or a shutdown was detected on the connection and the buffer
370 * is empty, we must not attempt to receive
371 * - if the demux buf failed to be allocated, we must not try to receive and
372 * we know there is nothing pending
373 * - if no flag indicates a blocking condition, we may attempt to receive,
374 * regardless of whether the demux buffer is full or not, so that only
375 * de demux part decides whether or not to block. This is needed because
376 * the connection API indeed prevents us from re-enabling receipt that is
377 * already enabled in a polled state, so we must always immediately stop
378 * as soon as the demux can't proceed so as never to hit an end of read
379 * with data pending in the buffers.
380 * - otherwise must may not attempt
381 */
382static inline int qc_recv_allowed(const struct qcc *qcc)
383{
384 if (qcc->rx.inmux == 0 &&
385 (qcc->st0 >= QC_CS_ERROR ||
386 qcc->conn->flags & CO_FL_ERROR ||
387 conn_xprt_read0_pending(qcc->conn)))
388 return 0;
389
390 if (!(qcc->flags & QC_CF_DEM_BLOCK_ANY))
391 return 1;
392
393 return 0;
394}
395
396/* restarts reading on the connection if it was not enabled */
397static inline void qcc_restart_reading(const struct qcc *qcc, int consider_buffer)
398{
399 if (!qc_recv_allowed(qcc))
400 return;
401
402 if ((!consider_buffer || !qcc->rx.inmux)
403 && (qcc->wait_event.events & SUB_RETRY_RECV))
404 return;
405
406 tasklet_wakeup(qcc->wait_event.tasklet);
407}
408
409/* Tries to grab a buffer and to re-enable processing on mux <target>. The qcc
410 * flags are used to figure what buffer was requested. It returns 1 if the
411 * allocation succeeds, in which case the connection is woken up, or 0 if it's
412 * impossible to wake up and we prefer to be woken up later.
413 */
414static int qc_buf_available(void *target)
415{
416 struct qcc *qcc = target;
417
418 if ((qcc->flags & QC_CF_MUX_MALLOC) && b_alloc(br_tail(qcc->mbuf))) {
419 qcc->flags &= ~QC_CF_MUX_MALLOC;
420
421 if (qcc->flags & QC_CF_DEM_MROOM) {
422 qcc->flags &= ~QC_CF_DEM_MROOM;
423 qcc_restart_reading(qcc, 1);
424 }
425 return 1;
426 }
427
428#if 0
429 if ((qcc->flags & QC_CF_DEM_SALLOC) &&
430 (qcs = qcc_st_by_id(qcc, qcc->dsi)) && qcs->cs &&
431 b_alloc_margin(&qcs->rxbuf, 0)) {
432 qcc->flags &= ~QC_CF_DEM_SALLOC;
433 qcc_restart_reading(qcc, 1);
434 return 1;
435 }
436#endif
437
438 return 0;
439}
440
441struct buffer *qc_get_buf(struct qcc *qcc, struct buffer *bptr)
442{
443 struct buffer *buf = NULL;
444
445 if (likely(!LIST_INLIST(&qcc->buf_wait.list)) &&
446 unlikely((buf = b_alloc(bptr)) == NULL)) {
447 qcc->buf_wait.target = qcc;
448 qcc->buf_wait.wakeup_cb = qc_buf_available;
Willy Tarreaub4e34762021-09-30 19:02:18 +0200449 LIST_APPEND(&th_ctx->buffer_wq, &qcc->buf_wait.list);
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100450 }
451
452 return buf;
453}
454
455__maybe_unused
456static inline void qc_release_buf(struct qcc *qcc, struct buffer *bptr)
457{
458 if (bptr->size) {
459 b_free(bptr);
460 offer_buffers(NULL, 1);
461 }
462}
463
464static inline void qc_release_mbuf(struct qcc *qcc)
465{
466 struct buffer *buf;
467 unsigned int count = 0;
468
469 while (b_size(buf = br_head_pick(qcc->mbuf))) {
470 b_free(buf);
471 count++;
472 }
473 if (count)
474 offer_buffers(NULL, count);
475}
476
477/* returns the number of streams in use on a connection to figure if it's
478 * idle or not. We check nb_cs and not nb_streams as the caller will want
479 * to know if it was the last one after a detach().
480 */
481static int qc_used_streams(struct connection *conn)
482{
483 struct qcc *qcc = conn->ctx;
484
485 return qcc->nb_cs;
486}
487
488/* returns the number of concurrent streams available on the connection with <dir>
489 * as direction
490 */
491static int qc_avail_streams(struct connection *conn, enum qcs_dir dir)
492{
493 struct qcc *qcc = conn->ctx;
494 enum qcs_type qcs_type;
495
496 if (qcc->st0 >= QC_CS_ERROR)
497 return 0;
498
499 qcs_type = qcs_type_from_dir(qcc, dir);
500
501 return qcc->strms[qcs_type].max_streams - qcc->strms[qcs_type].nb_streams;
502}
503
504
505/* returns the number of concurrent bidirectional streams available on the
506 * connection.
507 */
508static int qc_avail_streams_bidi(struct connection *conn)
509{
510 return qc_avail_streams(conn, QCS_BIDI);
511}
512
513/* returns the number of concurrent unidirectional streams available on the
514 * connection.
515 */
516static int qc_avail_streams_uni(struct connection *conn)
517{
518 return qc_avail_streams(conn, QCS_UNI);
519}
520
521/*****************************************************************/
522/* functions below are dedicated to the mux setup and management */
523/*****************************************************************/
524
525/* Update the mux transport parameter after having received remote transpot parameters */
526void quic_mux_transport_params_update(struct qcc *qcc)
527{
528 if (objt_listener(qcc->conn->target)) {
529 struct quic_transport_params *clt_params;
530
531 /* Client parameters, params used to TX. */
532 clt_params = &qcc->conn->qc->tx.params;
533
534 qcc->tx.max_data = clt_params->initial_max_data;
535 /* Client initiated streams must respect the server flow control. */
536 qcc->strms[QCS_CLT_BIDI].rx.max_data = clt_params->initial_max_stream_data_bidi_local;
537 qcc->strms[QCS_CLT_UNI].rx.max_data = clt_params->initial_max_stream_data_uni;
538
539 /* Server initiated streams must respect the server flow control. */
540 qcc->strms[QCS_SRV_BIDI].max_streams = clt_params->initial_max_streams_bidi;
541 qcc->strms[QCS_SRV_BIDI].tx.max_data = clt_params->initial_max_stream_data_bidi_remote;
542
543 qcc->strms[QCS_SRV_UNI].max_streams = clt_params->initial_max_streams_uni;
544 qcc->strms[QCS_SRV_UNI].tx.max_data = clt_params->initial_max_stream_data_uni;
545 }
546 else {
547 struct quic_transport_params *srv_params;
548
549 /* server parameters, TX params. */
550 srv_params = &qcc->conn->qc->tx.params;
551
552 qcc->tx.max_data = srv_params->initial_max_data;
553 /* Client initiated streams must respect the server flow control. */
554 qcc->strms[QCS_CLT_BIDI].max_streams = srv_params->initial_max_streams_bidi;
555 qcc->strms[QCS_CLT_BIDI].tx.max_data = srv_params->initial_max_stream_data_bidi_remote;
556
557 qcc->strms[QCS_CLT_UNI].max_streams = srv_params->initial_max_streams_uni;
558 qcc->strms[QCS_CLT_UNI].tx.max_data = srv_params->initial_max_stream_data_uni;
559
560 /* Server initiated streams must respect the server flow control. */
561 qcc->strms[QCS_SRV_BIDI].rx.max_data = srv_params->initial_max_stream_data_bidi_local;
562 qcc->strms[QCS_SRV_UNI].rx.max_data = srv_params->initial_max_stream_data_uni;
563 }
564
565 /* Now that we have all the flow control information, we can finalize the application
566 * context.
567 */
568 qcc->app_ops->finalize(qcc->ctx);
569}
570
571/* Initialize the mux once it's attached. For outgoing connections, the context
572 * is already initialized before installing the mux, so we detect incoming
573 * connections from the fact that the context is still NULL (even during mux
574 * upgrades). <input> is always used as Input buffer and may contain data. It is
575 * the caller responsibility to not reuse it anymore. Returns < 0 on error.
576 */
577static int qc_init(struct connection *conn, struct proxy *prx,
578 struct session *sess, struct buffer *input)
579{
580 struct qcc *qcc;
581 struct task *t = NULL;
582 void *conn_ctx = conn->ctx;
583
584 TRACE_ENTER(QC_EV_QCC_NEW);
585
586 qcc = pool_alloc(pool_head_qcc);
587 if (!qcc)
588 goto fail_no_qcc;
589
590 if (conn_is_back(conn)) {
591 qcc->flags = QC_CF_IS_BACK;
592 qcc->shut_timeout = qcc->timeout = prx->timeout.server;
593 if (tick_isset(prx->timeout.serverfin))
594 qcc->shut_timeout = prx->timeout.serverfin;
595
596 qcc->px_counters = EXTRA_COUNTERS_GET(prx->extra_counters_be,
597 &qc_stats_module);
598 } else {
599 qcc->flags = QC_CF_NONE;
600 qcc->shut_timeout = qcc->timeout = prx->timeout.client;
601 if (tick_isset(prx->timeout.clientfin))
602 qcc->shut_timeout = prx->timeout.clientfin;
603
604 qcc->px_counters = EXTRA_COUNTERS_GET(prx->extra_counters_fe,
605 &qc_stats_module);
606 }
607
608 qcc->proxy = prx;
609 qcc->task = NULL;
610 if (tick_isset(qcc->timeout)) {
Willy Tarreaubeeabf52021-10-01 18:23:30 +0200611 t = task_new_here();
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100612 if (!t)
613 goto fail;
614
615 qcc->task = t;
616 t->process = qc_timeout_task;
617 t->context = qcc;
618 t->expire = tick_add(now_ms, qcc->timeout);
619 }
620
Amaury Denoyellecde91122021-09-22 15:28:27 +0200621 qcc->subs = NULL;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100622 qcc->wait_event.tasklet = tasklet_new();
623 if (!qcc->wait_event.tasklet)
624 goto fail;
625
626 qcc->wait_event.tasklet->process = qc_io_cb;
627 qcc->wait_event.tasklet->context = qcc;
628 qcc->wait_event.events = 0;
629
630 /* Initialize the context. */
631 qcc->st0 = QC_CS_NOERR;
632 qcc->conn = conn;
633 qcc->conn->qc->qcc = qcc;
634
635 /* Application layer initialization. */
636 qcc->app_ops = &h3_ops;
637 if (!qcc->app_ops->init(qcc))
638 goto fail;
639
640 /* The transports parameters which control the data sent have been stored
641 * in ->tx.params. The ones which control the received data are stored in
642 * in ->rx.params.
643 */
644 if (objt_listener(qcc->conn->target)) {
645 struct quic_transport_params *srv_params;
646
647 /* Server parameters, params used for RX flow control. */
648 srv_params = &conn->qc->rx.params;
649
650 qcc->rx.max_data = srv_params->initial_max_data;
651 qcc->tx.max_data = 0;
652 /* Client initiated streams must respect the server flow control. */
653 qcc->strms[QCS_CLT_BIDI].max_streams = srv_params->initial_max_streams_bidi;
654 qcc->strms[QCS_CLT_BIDI].nb_streams = 0;
655 qcc->strms[QCS_CLT_BIDI].largest_id = -1;
656 qcc->strms[QCS_CLT_BIDI].rx.max_data = 0;
657 qcc->strms[QCS_CLT_BIDI].tx.max_data = srv_params->initial_max_stream_data_bidi_remote;
658
659 qcc->strms[QCS_CLT_UNI].max_streams = srv_params->initial_max_streams_uni;
660 qcc->strms[QCS_CLT_UNI].nb_streams = 0;
661 qcc->strms[QCS_CLT_UNI].largest_id = -1;
662 qcc->strms[QCS_CLT_UNI].rx.max_data = 0;
663 qcc->strms[QCS_CLT_UNI].tx.max_data = srv_params->initial_max_stream_data_uni;
664
665 /* Server initiated streams must respect the server flow control. */
666 qcc->strms[QCS_SRV_BIDI].max_streams = 0;
667 qcc->strms[QCS_SRV_BIDI].nb_streams = 0;
668 qcc->strms[QCS_SRV_BIDI].largest_id = -1;
669 qcc->strms[QCS_SRV_BIDI].rx.max_data = srv_params->initial_max_stream_data_bidi_local;
670 qcc->strms[QCS_SRV_BIDI].tx.max_data = 0;
671
672 qcc->strms[QCS_SRV_UNI].max_streams = 0;
673 qcc->strms[QCS_SRV_UNI].nb_streams = 0;
674 qcc->strms[QCS_SRV_UNI].largest_id = -1;
675 qcc->strms[QCS_SRV_UNI].rx.max_data = srv_params->initial_max_stream_data_uni;
676 qcc->strms[QCS_SRV_UNI].tx.max_data = 0;
677 }
678 else {
679 struct quic_transport_params *clt_params;
680
681 /* client parameters, RX params. */
682 clt_params = &conn->qc->rx.params;
683
684 qcc->rx.max_data = clt_params->initial_max_data;
685 qcc->tx.max_data = 0;
686 /* Client initiated streams must respect the server flow control. */
687 qcc->strms[QCS_CLT_BIDI].max_streams = 0;
688 qcc->strms[QCS_CLT_BIDI].nb_streams = 0;
689 qcc->strms[QCS_CLT_BIDI].largest_id = -1;
690 qcc->strms[QCS_CLT_BIDI].rx.max_data = clt_params->initial_max_stream_data_bidi_local;
691 qcc->strms[QCS_CLT_BIDI].tx.max_data = 0;
692
693 qcc->strms[QCS_CLT_UNI].max_streams = 0;
694 qcc->strms[QCS_CLT_UNI].nb_streams = 0;
695 qcc->strms[QCS_CLT_UNI].largest_id = -1;
696 qcc->strms[QCS_CLT_UNI].rx.max_data = clt_params->initial_max_stream_data_uni;
697 qcc->strms[QCS_CLT_UNI].tx.max_data = 0;
698
699 /* Server initiated streams must respect the server flow control. */
700 qcc->strms[QCS_SRV_BIDI].max_streams = clt_params->initial_max_streams_bidi;
701 qcc->strms[QCS_SRV_BIDI].nb_streams = 0;
702 qcc->strms[QCS_SRV_BIDI].largest_id = -1;
703 qcc->strms[QCS_SRV_BIDI].rx.max_data = 0;
704 qcc->strms[QCS_SRV_BIDI].tx.max_data = clt_params->initial_max_stream_data_bidi_remote;
705
706 qcc->strms[QCS_SRV_UNI].max_streams = clt_params->initial_max_streams_uni;
707 qcc->strms[QCS_SRV_UNI].nb_streams = 0;
708 qcc->strms[QCS_SRV_UNI].largest_id = -1;
709 qcc->strms[QCS_SRV_UNI].rx.max_data = 0;
710 qcc->strms[QCS_SRV_UNI].tx.max_data = clt_params->initial_max_stream_data_uni;
711
712 }
713
714 /* Initialize the streams counters. */
715 qcc->nb_cs = 0;
716 qcc->stream_cnt = 0;
717
718 br_init(qcc->mbuf, sizeof(qcc->mbuf) / sizeof(qcc->mbuf[0]));
719 qcc->streams_by_id = EB_ROOT_UNIQUE;
720 LIST_INIT(&qcc->send_list);
721 LIST_INIT(&qcc->fctl_list);
722 LIST_INIT(&qcc->blocked_list);
723 LIST_INIT(&qcc->buf_wait.list);
724 MT_LIST_INIT(&qcc->qcs_rxbuf_wlist);
725
Frédéric Lécaille01abc462021-07-21 09:34:27 +0200726 HA_ATOMIC_STORE(&conn->ctx, qcc);
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100727
728 if (t)
729 task_queue(t);
730
731 if (qcc->flags & QC_CF_IS_BACK) {
732 /* FIXME: For outgoing connections we need to immediately allocate streams.
733 * This highly depends on the QUIC application needs.
734 */
735 }
736
737 HA_ATOMIC_ADD(&qcc->px_counters->open_conns, 1);
738 HA_ATOMIC_ADD(&qcc->px_counters->total_conns, 1);
739
740 /* prepare to read something */
741 qcc_restart_reading(qcc, 1);
742 TRACE_LEAVE(QC_EV_QCC_NEW, conn);
743 return 0;
744
745 fail:
746 task_destroy(t);
747 if (qcc->wait_event.tasklet)
748 tasklet_free(qcc->wait_event.tasklet);
749 pool_free(pool_head_qcc, qcc);
750 fail_no_qcc:
751 conn->ctx = conn_ctx; /* restore saved ctx */
752 TRACE_DEVEL("leaving in error", QC_EV_QCC_NEW|QC_EV_QCC_END|QC_EV_QCC_ERR);
753 return -1;
754}
755
756/* returns the stream associated with id <id> or NULL if not found */
757__maybe_unused
758static inline struct qcs *qcc_st_by_id(struct qcc *qcc, int id)
759{
760 struct eb64_node *node;
761
762 node = eb64_lookup(&qcc->streams_by_id, id);
763 if (!node)
764 return (struct qcs *)qc_closed_stream;
765
766 return container_of(node, struct qcs, by_id);
767}
768
769/* release function. This one should be called to free all resources allocated
770 * to the mux.
771 */
772static void qc_release(struct qcc *qcc)
773{
774 struct connection *conn = NULL;
775
776 TRACE_ENTER(QC_EV_QCC_END);
777
778 if (qcc) {
779 /* The connection must be aattached to this mux to be released */
780 if (qcc->conn && qcc->conn->ctx == qcc)
781 conn = qcc->conn;
782
783 TRACE_DEVEL("freeing qcc", QC_EV_QCC_END, conn);
784
785 if (LIST_INLIST(&qcc->buf_wait.list))
786 LIST_DELETE(&qcc->buf_wait.list);
787
788 qc_release_mbuf(qcc);
789
790 if (qcc->task) {
791 qcc->task->context = NULL;
792 task_wakeup(qcc->task, TASK_WOKEN_OTHER);
793 qcc->task = NULL;
794 }
795 if (qcc->wait_event.tasklet)
796 tasklet_free(qcc->wait_event.tasklet);
797 if (conn && qcc->wait_event.events != 0)
798 conn->xprt->unsubscribe(conn, conn->xprt_ctx, qcc->wait_event.events,
799 &qcc->wait_event);
800
801 HA_ATOMIC_SUB(&qcc->px_counters->open_conns, 1);
802
803 pool_free(pool_head_qcc, qcc);
804 }
805
806 if (conn) {
807 conn->mux = NULL;
808 conn->ctx = NULL;
809 TRACE_DEVEL("freeing conn", QC_EV_QCC_END, conn);
810
811 conn_stop_tracking(conn);
812 conn_full_close(conn);
813 if (conn->destroy_cb)
814 conn->destroy_cb(conn);
815 conn_free(conn);
816 }
817
818 TRACE_LEAVE(QC_EV_QCC_END);
819}
820
821
822/******************************************************/
823/* functions below are for the QUIC protocol processing */
824/******************************************************/
825
826/* attempt to notify the data layer of recv availability */
827__maybe_unused
828static void qcs_notify_recv(struct qcs *qcs)
829{
830 if (qcs->subs && qcs->subs->events & SUB_RETRY_RECV) {
831 TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn, qcs);
832 tasklet_wakeup(qcs->subs->tasklet);
833 qcs->subs->events &= ~SUB_RETRY_RECV;
834 if (!qcs->subs->events)
835 qcs->subs = NULL;
836 }
837}
838
839/* attempt to notify the data layer of send availability */
840__maybe_unused
841static void qcs_notify_send(struct qcs *qcs)
842{
843 if (qcs->subs && qcs->subs->events & SUB_RETRY_SEND) {
844 TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn, qcs);
845 qcs->flags |= QC_SF_NOTIFIED;
846 tasklet_wakeup(qcs->subs->tasklet);
847 qcs->subs->events &= ~SUB_RETRY_SEND;
848 if (!qcs->subs->events)
849 qcs->subs = NULL;
850 }
851 else if (qcs->flags & (QC_SF_WANT_SHUTR | QC_SF_WANT_SHUTW)) {
852 TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn, qcs);
853 tasklet_wakeup(qcs->shut_tl);
854 }
855}
856
857/* alerts the data layer, trying to wake it up by all means, following
858 * this sequence :
859 * - if the qcs' data layer is subscribed to recv, then it's woken up for recv
860 * - if its subscribed to send, then it's woken up for send
861 * - if it was subscribed to neither, its ->wake() callback is called
862 * It is safe to call this function with a closed stream which doesn't have a
863 * conn_stream anymore.
864 */
865__maybe_unused
866static void qcs_alert(struct qcs *qcs)
867{
868 TRACE_ENTER(QC_EV_QCS_WAKE, qcs->qcc->conn, qcs);
869
870 if (qcs->subs ||
871 (qcs->flags & (QC_SF_WANT_SHUTR | QC_SF_WANT_SHUTW))) {
872 qcs_notify_recv(qcs);
873 qcs_notify_send(qcs);
874 }
875 else if (qcs->cs && qcs->cs->data_cb->wake != NULL) {
876 TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn, qcs);
877 qcs->cs->data_cb->wake(qcs->cs);
878 }
879
880 TRACE_LEAVE(QC_EV_QCS_WAKE, qcs->qcc->conn, qcs);
881}
882
883/* marks stream <qcs> as CLOSED and decrement the number of active streams for
884 * its connection if the stream was not yet closed. Please use this exclusively
885 * before closing a stream to ensure stream count is well maintained.
886 */
887static inline void qcs_close(struct qcs *qcs)
888{
889 TRACE_ENTER(QC_EV_QCS_END, qcs->qcc->conn, qcs);
890 /* XXX TO DO XXX */
891 TRACE_LEAVE(QC_EV_QCS_END, qcs->qcc->conn, qcs);
892}
893
894/* detaches an QUIC stream from its QCC and releases it to the QCS pool. */
895/* qcs_destroy should only ever be called by the thread that owns the stream,
896 * that means that a tasklet should be used if we want to destroy the qcs
897 * from another thread
898 */
899static void qcs_destroy(struct qcs *qcs)
900{
901 struct connection *conn = qcs->qcc->conn;
902
903 TRACE_ENTER(QC_EV_QCS_END, conn, qcs);
904
905 qcs_close(qcs);
906 eb64_delete(&qcs->by_id);
907 if (b_size(&qcs->rx.buf)) {
908 b_free(&qcs->rx.buf);
909 offer_buffers(NULL, 1);
910 }
911
Amaury Denoyelled3d97c62021-10-05 11:45:58 +0200912 b_free(&qcs->tx.buf);
913 b_free(&qcs->tx.xprt_buf);
914
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100915 if (qcs->subs)
916 qcs->subs->events = 0;
917
918 /* There's no need to explicitly call unsubscribe here, the only
919 * reference left would be in the qcc send_list/fctl_list, and if
920 * we're in it, we're getting out anyway
921 */
922 LIST_DEL_INIT(&qcs->list);
Amaury Denoyelled595f102021-09-24 10:05:30 +0200923 --qcs->qcc->strms[qcs_id_type(qcs->id)].nb_streams;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100924
925 /* ditto, calling tasklet_free() here should be ok */
926 tasklet_free(qcs->shut_tl);
927 pool_free(pool_head_qcs, qcs);
928
929 TRACE_LEAVE(QC_EV_QCS_END, conn);
930}
931
932/* allocates a new bidirection stream <id> for connection <qcc> and adds it into qcc's
933 * stream tree. In case of error, nothing is added and NULL is returned. The
934 * causes of errors can be any failed memory allocation. The caller is
935 * responsible for checking if the connection may support an extra stream
936 * prior to calling this function.
937 */
938struct qcs *bidi_qcs_new(struct qcc *qcc, uint64_t id)
939{
940 struct qcs *qcs;
941 enum qcs_type qcs_type;
942
943 TRACE_ENTER(QC_EV_QCS_NEW, qcc->conn);
944
945 qcs = pool_alloc(pool_head_qcs);
946 if (!qcs)
947 goto out;
948
949 qcs->shut_tl = tasklet_new();
950 if (!qcs->shut_tl) {
951 pool_free(pool_head_qcs, qcs);
952 goto out;
953 }
954
955 qcs_type = qcs_id_type(id);
956 qcs->qcc = qcc;
957 qcs->cs = NULL;
958 qcs->id = qcs->by_id.key = id;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100959 qcs->flags = QC_SF_NONE;
960
961 qcs->rx.buf = BUF_NULL;
962 qcs->rx.st = QC_RX_SS_IDLE;
963 qcs->rx.bytes = qcs->rx.offset = 0;
964 qcs->rx.max_data = qcc->strms[qcs_type].rx.max_data;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100965 qcs->rx.buf = BUF_NULL;
Frédéric Lécaille785d3bd2021-09-10 09:13:39 +0200966 qcs->rx.frms = EB_ROOT_UNIQUE;
967
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100968 qcs->tx.st = QC_TX_SS_IDLE;
Frédéric Lécaille785d3bd2021-09-10 09:13:39 +0200969 qcs->tx.bytes = qcs->tx.offset = qcs->tx.ack_offset = 0;
970 qcs->tx.acked_frms = EB_ROOT_UNIQUE;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100971 qcs->tx.max_data = qcc->strms[qcs_type].tx.max_data;
Frédéric Lécaille785d3bd2021-09-10 09:13:39 +0200972 qcs->tx.buf = BUF_NULL;
Amaury Denoyelled3d97c62021-10-05 11:45:58 +0200973 qcs->tx.xprt_buf = BUF_NULL;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100974
975 eb64_insert(&qcc->streams_by_id, &qcs->by_id);
976 qcc->strms[qcs_type].nb_streams++;
977 qcc->stream_cnt++;
978 qcs->subs = NULL;
979 LIST_INIT(&qcs->list);
980 qcs->shut_tl->process = qc_deferred_shut;
981 qcs->shut_tl->context = qcs;
982
983 HA_ATOMIC_ADD(&qcc->px_counters->open_streams, 1);
984 HA_ATOMIC_ADD(&qcc->px_counters->total_streams, 1);
985
986 TRACE_LEAVE(QC_EV_QCS_NEW, qcc->conn, qcs);
987 return qcs;
988
989 out:
990 TRACE_DEVEL("leaving in error", QC_EV_QCS_ERR|QC_EV_QCS_END, qcc->conn);
991 return NULL;
992}
993
994/* Release <qcs> outgoing uni-stream */
995void qcs_release(struct qcs *qcs)
996{
997 eb64_delete(&qcs->by_id);
998 pool_free(pool_head_qcs, qcs);
999}
1000
1001/* Allocates a locally initiated unidirectional stream. */
1002struct qcs *luqs_new(struct qcc *qcc)
1003{
1004 struct qcs *qcs;
1005 uint64_t next_id;
1006 enum qcs_type qcs_type;
1007
1008 TRACE_ENTER(QC_EV_QCS_NEW, qcc->conn);
1009
1010 qcs = NULL;
1011 /* QCS_ID_DIR_BIT bit is set for unidirectional stream. */
1012 if (objt_listener(qcc->conn->target))
1013 qcs_type = QCS_ID_SRV_INTIATOR_BIT | QCS_ID_DIR_BIT;
1014 else
1015 qcs_type = QCS_ID_DIR_BIT;
1016
1017 next_id = qcs_next_id(qcc, qcs_type);
1018 if (next_id == (uint64_t)-1) {
1019 TRACE_PROTO("No more stream available", QC_EV_QCS_NEW, qcc->conn);
1020 goto out;
1021 }
1022
1023 qcs = pool_alloc(pool_head_qcs);
1024 if (!qcs)
1025 goto out;
1026
1027 qcs->qcc = qcc;
1028 qcs->cs = NULL;
1029 qcs->id = qcs->by_id.key = next_id;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001030 qcs->flags = QC_SF_NONE;
1031
Frédéric Lécaille785d3bd2021-09-10 09:13:39 +02001032 qcs->tx.st = QC_TX_SS_IDLE;
1033 qcs->tx.max_data = qcc->strms[qcs_type].tx.max_data;
1034 qcs->tx.offset = qcs->tx.bytes = qcs->tx.ack_offset = 0;
1035 qcs->tx.acked_frms = EB_ROOT_UNIQUE;
1036 qcs->tx.buf = BUF_NULL;
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001037 qcs->tx.xprt_buf = BUF_NULL;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001038
1039 qcs->subs = NULL;
1040 LIST_INIT(&qcs->list);
Amaury Denoyelle139814a2021-09-24 10:03:16 +02001041 eb64_insert(&qcc->streams_by_id, &qcs->by_id);
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001042
1043 TRACE_LEAVE(QC_EV_QCS_NEW, qcc->conn);
1044 return qcs;
1045
1046 out:
1047 if (qcs)
1048 pool_free(pool_head_qcs, qcs);
1049 TRACE_DEVEL("leaving in error", QC_EV_QCS_ERR|QC_EV_QCS_END, qcc->conn);
1050 return NULL;
1051}
1052
1053/* Allocates a remotely initiated unidirectional stream. */
1054struct qcs *ruqs_new(struct qcc *qcc, uint64_t id)
1055{
1056 struct qcs *qcs;
1057 enum qcs_type qcs_type;
1058
1059 TRACE_ENTER(QC_EV_QCS_NEW, qcc->conn);
1060 qcs = pool_alloc(pool_head_qcs);
1061 if (!qcs)
1062 goto out;
1063
1064 qcs_type = qcs_id_type(id);
1065
1066 qcs->qcc = qcc;
1067 qcs->cs = NULL;
1068
1069 qcs->qcc = qcc;
1070 qcs->id = qcs->by_id.key = id;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001071 qcs->flags = QC_SF_NONE;
1072
1073 qcs->rx.st = QC_RX_SS_IDLE;
1074 qcs->rx.max_data = qcc->strms[qcs_type].rx.max_data;
1075 qcs->rx.offset = qcs->rx.bytes = 0;
1076 qcs->rx.buf = BUF_NULL;
Frédéric Lécaille785d3bd2021-09-10 09:13:39 +02001077 qcs->rx.frms = EB_ROOT_UNIQUE;
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001078 qcs->tx.buf = BUF_NULL;
1079 qcs->tx.xprt_buf = BUF_NULL;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001080
1081 qcs->subs = NULL;
1082 LIST_INIT(&qcs->list);
Amaury Denoyelle139814a2021-09-24 10:03:16 +02001083 eb64_insert(&qcc->streams_by_id, &qcs->by_id);
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001084
1085 TRACE_LEAVE(QC_EV_QCS_NEW, qcc->conn);
1086 return qcs;
1087
1088 out:
1089 TRACE_DEVEL("leaving in error", QC_EV_QCS_ERR|QC_EV_QCS_END, qcc->conn);
1090 return NULL;
1091}
1092
1093/* attempt to notify the data layer of recv availability */
1094void ruqs_notify_recv(struct qcs *qcs)
1095{
1096 if (qcs->subs && qcs->subs->events & SUB_RETRY_RECV) {
1097 TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn);
1098 tasklet_wakeup(qcs->subs->tasklet);
1099 qcs->subs->events &= ~SUB_RETRY_RECV;
1100 if (!qcs->subs->events)
1101 qcs->subs = NULL;
1102 }
1103}
1104
1105/* Allocates a new stream associated to conn_stream <cs> on the qcc connection
1106 * with dir as direction and returns it, or NULL in case of memory allocation
1107 * error or if the highest possible stream ID was reached.
1108 */
1109static struct qcs *qcc_bck_stream_new(struct qcc *qcc, int dir,
1110 struct conn_stream *cs, struct session *sess)
1111{
1112 struct qcs *qcs = NULL;
1113 enum qcs_type qcs_type;
1114
1115 TRACE_ENTER(QC_EV_QCS_NEW, qcc->conn);
1116
1117 qcs_type = qcs_type_from_dir(qcc, dir);
1118 if (qcc->strms[qcs_type].largest_id + 1 >= qcc->strms[qcs_type].max_streams)
1119 goto out;
1120
1121 /* Defer choosing the ID until we send the first message to create the stream */
1122 qcs = bidi_qcs_new(qcc, qcc->strms[qcs_type].largest_id + 1);
1123 if (!qcs)
1124 goto out;
1125
1126 qcs->cs = cs;
1127 qcs->sess = sess;
1128 cs->ctx = qcs;
1129 qcc->nb_cs++;
1130
1131 out:
1132 if (likely(qcs))
1133 TRACE_LEAVE(QC_EV_QCS_NEW, qcc->conn, qcs);
1134 else
1135 TRACE_LEAVE(QC_EV_QCS_NEW|QC_EV_QCS_ERR|QC_EV_QCS_END, qcc->conn, qcs);
1136 return qcs;
1137}
1138
1139/* Allocates a new bidirectional stream associated to conn_stream <cs> on the <qcc> connection
1140 * and returns it, or NULL in case of memory allocation error or if the highest
1141 * possible stream ID was reached.
1142 */
1143__maybe_unused
1144static struct qcs *qcc_bck_stream_new_bidi(struct qcc *qcc,
1145 struct conn_stream *cs, struct session *sess)
1146{
1147 return qcc_bck_stream_new(qcc, QCS_BIDI, cs, sess);
1148}
1149
1150/* Allocates a new unidirectional stream associated to conn_stream <cs> on the <qcc> connection
1151 * and returns it, or NULL in case of memory allocation error or if the highest
1152 * possible stream ID was reached.
1153 */
1154__maybe_unused
1155static struct qcs *qcc_bck_stream_new_uni(struct qcc *qcc,
1156 struct conn_stream *cs, struct session *sess)
1157{
1158 return qcc_bck_stream_new(qcc, QCS_UNI, cs, sess);
1159}
1160
1161
1162/* wake a specific stream and assign its conn_stream some CS_FL_* flags among
1163 * CS_FL_ERR_PENDING and CS_FL_ERROR if needed. The stream's state
1164 * is automatically updated accordingly. If the stream is orphaned, it is
1165 * destroyed.
1166 */
1167static void qcs_wake_one_stream(struct qcs *qcs)
1168{
1169 struct qcc *qcc = qcs->qcc;
1170
1171 TRACE_ENTER(QC_EV_QCS_WAKE, qcc->conn, qcs);
1172 if (!qcs->cs) {
1173 /* this stream was already orphaned */
1174 qcs_destroy(qcs);
1175 TRACE_DEVEL("leaving with no qcs", QC_EV_QCS_WAKE, qcc->conn);
1176 return;
1177 }
1178 /* XXX TO DO XXX */
1179 TRACE_LEAVE(QC_EV_QCS_WAKE, qcc->conn);
1180}
1181
1182/* wake the streams attached to the connection, whose id is greater than <last>
1183 * or unassigned.
1184 */
1185static void qc_wake_some_streams(struct qcc *qcc, int last)
1186{
1187 struct eb64_node *node;
1188 struct qcs *qcs;
1189
1190 TRACE_ENTER(QC_EV_QCS_WAKE, qcc->conn);
1191
1192 /* Wake all streams with ID > last */
1193 node = eb64_lookup_ge(&qcc->streams_by_id, last + 1);
1194 while (node) {
1195 qcs = container_of(node, struct qcs, by_id);
1196 node = eb64_next(node);
1197 qcs_wake_one_stream(qcs);
1198 }
1199
1200 /* Wake all streams with unassigned ID (ID == 0) */
1201 node = eb64_lookup(&qcc->streams_by_id, 0);
1202 while (node) {
1203 qcs = container_of(node, struct qcs, by_id);
1204 if (qcs->id > 0)
1205 break;
1206 node = eb64_next(node);
1207 qcs_wake_one_stream(qcs);
1208 }
1209
1210 TRACE_LEAVE(QC_EV_QCS_WAKE, qcc->conn);
1211}
1212
1213/* Wake up all blocked streams whose window size has become positive after the
1214 * mux's initial window was adjusted. This should be done after having processed
1215 * SETTINGS frames which have updated the mux's initial window size.
1216 */
1217__maybe_unused
1218static void qcc_unblock_sfctl(struct qcc *qcc)
1219{
1220 TRACE_ENTER(QC_EV_QCC_WAKE, qcc->conn);
1221 /* XXX TO DO XXX */
1222 TRACE_LEAVE(QC_EV_QCC_WAKE, qcc->conn);
1223}
1224
1225/* process Rx frames to be demultiplexed */
1226__maybe_unused
1227static void qc_process_demux(struct qcc *qcc)
1228{
1229 TRACE_ENTER(QC_EV_QCC_WAKE, qcc->conn);
1230 /* XXX TO DO XXX */
1231 TRACE_LEAVE(QC_EV_QCC_WAKE, qcc->conn);
1232}
1233
1234/* resume each qcs eligible for sending in list head <head> */
1235__maybe_unused
1236static void qc_resume_each_sending_qcs(struct qcc *qcc, struct list *head)
1237{
1238 struct qcs *qcs, *qcs_back;
1239
1240 TRACE_ENTER(QC_EV_QCC_SEND|QC_EV_QCS_WAKE, qcc->conn);
1241
1242 list_for_each_entry_safe(qcs, qcs_back, head, list) {
1243 if (qcc_wnd(qcc) <= 0 ||
1244 qcc->flags & QC_CF_MUX_BLOCK_ANY ||
1245 qcc->st0 >= QC_CS_ERROR)
1246 break;
1247
1248 qcs->flags &= ~QC_SF_BLK_ANY;
1249
1250 if (qcs->flags & QC_SF_NOTIFIED)
1251 continue;
1252
1253 /* If the sender changed his mind and unsubscribed, let's just
1254 * remove the stream from the send_list.
1255 */
1256 if (!(qcs->flags & (QC_SF_WANT_SHUTR|QC_SF_WANT_SHUTW)) &&
1257 (!qcs->subs || !(qcs->subs->events & SUB_RETRY_SEND))) {
1258 LIST_DEL_INIT(&qcs->list);
1259 continue;
1260 }
1261
1262 if (qcs->subs && qcs->subs->events & SUB_RETRY_SEND) {
1263 qcs->flags |= QC_SF_NOTIFIED;
1264 tasklet_wakeup(qcs->subs->tasklet);
1265 qcs->subs->events &= ~SUB_RETRY_SEND;
1266 if (!qcs->subs->events)
1267 qcs->subs = NULL;
1268 }
1269 else if (qcs->flags & (QC_SF_WANT_SHUTR|QC_SF_WANT_SHUTW)) {
1270 tasklet_wakeup(qcs->shut_tl);
1271 }
1272 }
1273
1274 TRACE_LEAVE(QC_EV_QCC_SEND|QC_EV_QCS_WAKE, qcc->conn);
1275}
1276
1277/* process Tx frames from streams to be multiplexed. Returns > 0 if it reached
1278 * the end.
1279 */
1280__maybe_unused
1281static int qc_process_mux(struct qcc *qcc)
1282{
1283 TRACE_ENTER(QC_EV_QCC_WAKE, qcc->conn);
Frédéric Lécaille1d402402021-09-20 17:53:17 +02001284
1285 /* First we always process the flow control list because the streams
1286 * waiting there were already elected for immediate emission but were
1287 * blocked just on this.
1288 */
1289 qc_resume_each_sending_qcs(qcc, &qcc->fctl_list);
1290 qc_resume_each_sending_qcs(qcc, &qcc->send_list);
1291
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001292 TRACE_LEAVE(QC_EV_QCC_WAKE, qcc->conn);
Frédéric Lécaille1d402402021-09-20 17:53:17 +02001293 return 1;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001294}
1295
1296
1297/* Attempt to read data, and subscribe if none available.
1298 * The function returns 1 if data has been received, otherwise zero.
1299 */
1300__maybe_unused
1301static int qc_recv(struct qcc *qcc)
1302{
1303 TRACE_ENTER(QC_EV_QCC_RECV, qcc->conn);
1304 /* XXX TO DO XXX */
1305 TRACE_LEAVE(QC_EV_QCC_RECV, qcc->conn);
1306 return 0;
1307}
1308
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001309static int qcs_push_frame(struct qcs *qcs, struct buffer *payload, int fin, uint64_t offset)
1310{
1311 struct quic_frame *frm;
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001312 struct buffer *buf = &qcs->tx.xprt_buf;
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001313 struct quic_enc_level *qel = &qcs->qcc->conn->qc->els[QUIC_TLS_ENC_LEVEL_APP];
Frédéric Lécailled2ba0962021-09-20 17:50:03 +02001314 int total = 0, to_xfer;
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001315
1316 qc_get_buf(qcs->qcc, buf);
Frédéric Lécailled2ba0962021-09-20 17:50:03 +02001317 to_xfer = QUIC_MIN(b_data(payload), b_room(buf));
1318 if (!to_xfer)
1319 goto out;
1320
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001321 frm = pool_zalloc(pool_head_quic_frame);
1322 if (!frm)
1323 goto err;
1324
Frédéric Lécailled2ba0962021-09-20 17:50:03 +02001325 total = b_force_xfer(buf, payload, to_xfer);
1326 fin = fin && !b_data(payload);
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001327 frm->type = QUIC_FT_STREAM_8;
1328 if (fin)
1329 frm->type |= QUIC_STREAM_FRAME_TYPE_FIN_BIT;
1330 if (offset) {
1331 frm->type |= QUIC_STREAM_FRAME_TYPE_OFF_BIT;
1332 frm->stream.offset.key = offset;
1333 }
1334 frm->stream.qcs = qcs;
1335 frm->stream.buf = buf;
1336 frm->stream.id = qcs->by_id.key;
1337 if (total) {
1338 frm->type |= QUIC_STREAM_FRAME_TYPE_LEN_BIT;
1339 frm->stream.len = total;
1340 }
1341
1342 MT_LIST_APPEND(&qel->pktns->tx.frms, &frm->mt_list);
Frédéric Lécailled2ba0962021-09-20 17:50:03 +02001343 out:
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001344 fprintf(stderr, "%s: total=%d fin=%d offset=%lu\n", __func__, total, fin, offset);
1345 return total;
1346
1347 err:
1348 return -1;
1349}
1350
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001351/* Try to send data if possible.
1352 * The function returns 1 if data have been sent, otherwise zero.
1353 */
1354static int qc_send(struct qcc *qcc)
1355{
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001356 struct eb64_node *node;
Amaury Denoyellea543eb12021-10-06 14:53:13 +02001357 int ret, done, xprt_wake = 0;
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001358
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001359 TRACE_ENTER(QC_EV_QCC_SEND, qcc->conn);
Frédéric Lécaille8526f142021-09-20 17:58:22 +02001360 ret = done = 0;
1361 /* fill as much as we can into the current buffer */
1362 while (((qcc->flags & (QC_CF_MUX_MFULL|QC_CF_MUX_MALLOC)) == 0) && !done)
1363 done = qc_process_mux(qcc);
1364
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001365 /* TODO simple loop through all streams and check if there is frames to
1366 * send
1367 */
1368 node = eb64_first(&qcc->streams_by_id);
1369 while (node) {
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001370 struct qcs *qcs = container_of(node, struct qcs, by_id);
1371 struct buffer *buf = &qcs->tx.buf;
1372 if (b_data(buf)) {
1373 char fin = 0;
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001374
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001375 /* if FIN is activated, ensure the buffer to
1376 * send is the last
1377 */
1378 if (qcs->flags & QC_SF_FIN_STREAM) {
1379 BUG_ON(b_data(&qcs->tx.buf) < b_data(buf));
1380 fin = (b_data(&qcs->tx.buf) - b_data(buf) == 0);
1381 }
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001382
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001383 ret = qcs_push_frame(qcs, buf, fin, qcs->tx.offset);
1384 if (ret < 0)
1385 ABORT_NOW();
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001386
Amaury Denoyellea543eb12021-10-06 14:53:13 +02001387 if (ret > 0) {
1388 xprt_wake = 1;
1389 if (qcs->flags & QC_SF_BLK_MROOM) {
1390 qcs->flags &= ~QC_SF_BLK_MROOM;
1391 qcs_notify_send(qcs);
1392 }
1393 }
1394
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001395 qcs->tx.offset += ret;
Amaury Denoyellea543eb12021-10-06 14:53:13 +02001396
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001397 if (b_data(buf)) {
1398 qcc->conn->xprt->subscribe(qcc->conn, qcc->conn->xprt_ctx,
1399 SUB_RETRY_SEND, &qcc->wait_event);
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001400 }
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001401 }
1402 node = eb64_next(node);
1403 }
Frédéric Lécaille8526f142021-09-20 17:58:22 +02001404
Amaury Denoyellea543eb12021-10-06 14:53:13 +02001405 if (xprt_wake)
1406 tasklet_wakeup(((struct ssl_sock_ctx *)(qcc->conn->xprt_ctx))->wait_event.tasklet);
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001407
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001408 TRACE_LEAVE(QC_EV_QCC_SEND, qcc->conn);
1409 return 0;
1410}
1411
1412/* this is the tasklet referenced in qcc->wait_event.tasklet */
1413static struct task *qc_io_cb(struct task *t, void *ctx, unsigned int status)
1414{
Amaury Denoyelle769e9ff2021-10-05 11:43:50 +02001415 struct qcc *qcc = ctx;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001416 int ret = 0;
1417
1418
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001419 if (!(qcc->wait_event.events & SUB_RETRY_SEND))
1420 ret = qc_send(qcc);
1421#if 0
1422 if (!(qcc->wait_event.events & SUB_RETRY_RECV))
1423 ret |= qc_recv(qcc);
1424#endif
1425 // TODO redefine the proper condition here
1426 //if (ret || qcc->rx.inmux)
1427 ret = qc_process(qcc);
1428
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001429leave:
1430 TRACE_LEAVE(QC_EV_QCC_WAKE);
1431 return NULL;
1432}
1433
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001434/* callback called on any event by the connection handler.
1435 * It applies changes and returns zero, or < 0 if it wants immediate
1436 * destruction of the connection (which normally doesn not happen in quic).
1437 */
1438static int qc_process(struct qcc *qcc)
1439{
1440 struct connection *conn = qcc->conn;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001441
1442 TRACE_ENTER(QC_EV_QCC_WAKE, conn);
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001443 TRACE_LEAVE(QC_EV_QCC_WAKE, conn);
1444 return 0;
1445}
1446
1447/* wake-up function called by the connection layer (mux_ops.wake) */
1448static int qc_wake(struct connection *conn)
1449{
1450 struct qcc *qcc = conn->ctx;
1451 int ret;
1452
1453 TRACE_ENTER(QC_EV_QCC_WAKE, conn);
1454 ret = qc_process(qcc);
1455 if (ret >= 0)
1456 qc_wake_some_streams(qcc, 0);
1457 TRACE_LEAVE(QC_EV_QCC_WAKE);
1458 return ret;
1459}
1460
1461/* Connection timeout management. The principle is that if there's no receipt
1462 * nor sending for a certain amount of time, the connection is closed. If the
1463 * MUX buffer still has lying data or is not allocatable, the connection is
1464 * immediately killed. If it's allocatable and empty, we attempt to send a
1465 * GOAWAY frame.
1466 */
1467static struct task *qc_timeout_task(struct task *t, void *context, unsigned int state)
1468{
1469 TRACE_ENTER(QC_EV_QCC_WAKE);
1470 /* XXX TO DO XXX */
1471 TRACE_LEAVE(QC_EV_QCC_WAKE);
1472 return NULL;
1473}
1474
1475
1476/*******************************************/
1477/* functions below are used by the streams */
1478/*******************************************/
1479
1480/*
1481 * Attach a new stream to a connection
1482 * (Used for outgoing connections)
1483 */
1484static struct conn_stream *qc_attach(struct connection *conn, struct session *sess)
1485{
1486 struct conn_stream *cs;
1487 struct qcs *qcs;
1488 struct qcc *qcc = conn->ctx;
1489
1490 TRACE_ENTER(QC_EV_QCS_NEW, conn);
1491 cs = cs_new(conn, conn->target);
1492 if (!cs) {
1493 TRACE_DEVEL("leaving on CS allocation failure", QC_EV_QCS_NEW|QC_EV_QCS_ERR, conn);
1494 return NULL;
1495 }
1496 qcs = qcc_bck_stream_new(qcc, QCS_BIDI, cs, sess);
1497 if (!qcs) {
1498 TRACE_DEVEL("leaving on stream creation failure", QC_EV_QCS_NEW|QC_EV_QCS_ERR, conn);
1499 cs_free(cs);
1500 return NULL;
1501 }
1502 TRACE_LEAVE(QC_EV_QCS_NEW, conn, qcs);
1503 return cs;
1504}
1505
1506/* Retrieves the first valid conn_stream from this connection, or returns NULL.
1507 * We have to scan because we may have some orphan streams. It might be
1508 * beneficial to scan backwards from the end to reduce the likeliness to find
1509 * orphans.
1510 */
1511static const struct conn_stream *qc_get_first_cs(const struct connection *conn)
1512{
1513 struct qcc *qcc = conn->ctx;
1514 struct qcs *qcs;
1515 struct eb64_node *node;
1516
1517 node = eb64_first(&qcc->streams_by_id);
1518 while (node) {
1519 qcs = container_of(node, struct qcs, by_id);
1520 if (qcs->cs)
1521 return qcs->cs;
1522 node = eb64_next(node);
1523 }
1524 return NULL;
1525}
1526
1527static int qc_ctl(struct connection *conn, enum mux_ctl_type mux_ctl, void *output)
1528{
1529 int ret = 0;
1530 struct qcc *qcc = conn->ctx;
1531
1532 switch (mux_ctl) {
1533 case MUX_STATUS:
1534 /* Only consider the mux to be ready if we had no error. */
1535 if (qcc->st0 < QC_CS_ERROR)
1536 ret |= MUX_STATUS_READY;
1537 return ret;
1538 case MUX_EXIT_STATUS:
1539 return MUX_ES_UNKNOWN;
1540 default:
1541 return -1;
1542 }
1543}
1544
1545/*
1546 * Destroy the mux and the associated connection, if it is no longer used
1547 */
1548static void qc_destroy(void *ctx)
1549{
1550 struct qcc *qcc = ctx;
1551
1552 TRACE_ENTER(QC_EV_QCC_END, qcc->conn);
1553 if (eb_is_empty(&qcc->streams_by_id) || !qcc->conn || qcc->conn->ctx != qcc)
1554 qc_release(qcc);
1555 TRACE_LEAVE(QC_EV_QCC_END);
1556}
1557
1558/*
1559 * Detach the stream from the connection and possibly release the connection.
1560 */
1561static void qc_detach(struct conn_stream *cs)
1562{
1563 struct qcs *qcs = cs->ctx;
Amaury Denoyellecd28b272021-09-22 14:48:32 +02001564 struct qcc *qcc = qcs->qcc;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001565
1566 TRACE_ENTER(QC_EV_STRM_END, qcs ? qcs->qcc->conn : NULL, qcs);
Amaury Denoyellecae07912021-10-08 17:57:41 +02001567 if (b_data(&qcs->tx.buf) || b_data(&qcs->tx.xprt_buf)) {
1568 qcs->flags |= QC_SF_DETACH;
1569 goto out;
1570 }
1571
Amaury Denoyellecd28b272021-09-22 14:48:32 +02001572 qcs_destroy(qcs);
Amaury Denoyelleac8ee252021-10-08 17:57:03 +02001573 if (qcc_is_dead(qcc))
Amaury Denoyellecd28b272021-09-22 14:48:32 +02001574 qc_release(qcc);
Amaury Denoyellecae07912021-10-08 17:57:41 +02001575
1576 out:
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001577 TRACE_LEAVE(QC_EV_STRM_END, qcs ? qcs->qcc->conn : NULL);
1578}
1579
1580/* Performs a synchronous or asynchronous shutr(). */
1581static void qc_do_shutr(struct qcs *qcs)
1582{
1583 struct qcc *qcc = qcs->qcc;
1584
1585 TRACE_ENTER(QC_EV_STRM_SHUT, qcc->conn, qcs);
1586 /* XXX TO DO XXX */
1587 TRACE_LEAVE(QC_EV_STRM_SHUT, qcc->conn, qcs);
1588 return;
1589}
1590
1591/* Performs a synchronous or asynchronous shutw(). */
1592static void qc_do_shutw(struct qcs *qcs)
1593{
1594 struct qcc *qcc = qcs->qcc;
1595
1596 TRACE_ENTER(QC_EV_STRM_SHUT, qcc->conn, qcs);
1597 /* XXX TO DO XXX */
1598 TRACE_LEAVE(QC_EV_STRM_SHUT, qcc->conn, qcs);
1599 return;
1600}
1601
1602/* This is the tasklet referenced in qcs->shut_tl, it is used for
1603 * deferred shutdowns when the qc_detach() was done but the mux buffer was full
1604 * and prevented the last frame from being emitted.
1605 */
1606static struct task *qc_deferred_shut(struct task *t, void *ctx, unsigned int state)
1607{
1608 struct qcs *qcs = ctx;
1609 struct qcc *qcc = qcs->qcc;
1610
1611 TRACE_ENTER(QC_EV_STRM_SHUT, qcc->conn, qcs);
1612
Amaury Denoyellecae07912021-10-08 17:57:41 +02001613 if (qcs->flags & QC_SF_NOTIFIED ||
1614 (b_data(&qcs->tx.buf) || b_data(&qcs->tx.xprt_buf))) {
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001615 /* some data processing remains to be done first */
1616 goto end;
1617 }
1618
1619 if (qcs->flags & QC_SF_WANT_SHUTW)
1620 qc_do_shutw(qcs);
1621
1622 if (qcs->flags & QC_SF_WANT_SHUTR)
1623 qc_do_shutr(qcs);
1624
1625 if (!(qcs->flags & (QC_SF_WANT_SHUTR|QC_SF_WANT_SHUTW))) {
1626 /* We're done trying to send, remove ourself from the send_list */
1627 LIST_DEL_INIT(&qcs->list);
1628
1629 if (!qcs->cs) {
1630 qcs_destroy(qcs);
1631 if (qcc_is_dead(qcc))
1632 qc_release(qcc);
1633 }
1634 }
1635
1636 end:
1637 TRACE_LEAVE(QC_EV_STRM_SHUT);
1638 return NULL;
1639}
1640
1641/* shutr() called by the conn_stream (mux_ops.shutr) */
1642static void qc_shutr(struct conn_stream *cs, enum cs_shr_mode mode)
1643{
1644
1645 struct qcs *qcs = cs->ctx;
1646
1647 TRACE_ENTER(QC_EV_STRM_SHUT, qcs->qcc->conn, qcs);
1648 if (cs->flags & CS_FL_KILL_CONN)
1649 qcs->flags |= QC_SF_KILL_CONN;
1650
1651 if (mode)
1652 qc_do_shutr(qcs);
1653
1654 TRACE_LEAVE(QC_EV_STRM_SHUT, qcs->qcc->conn, qcs);
1655}
1656
1657/* shutw() called by the conn_stream (mux_ops.shutw) */
1658static void qc_shutw(struct conn_stream *cs, enum cs_shw_mode mode)
1659{
1660 struct qcs *qcs = cs->ctx;
1661
1662 TRACE_ENTER(QC_EV_STRM_SHUT, qcs->qcc->conn, qcs);
1663 if (cs->flags & CS_FL_KILL_CONN)
1664 qcs->flags |= QC_SF_KILL_CONN;
1665
1666 qc_do_shutw(qcs);
1667 TRACE_LEAVE(QC_EV_STRM_SHUT, qcs->qcc->conn, qcs);
1668}
1669
1670/* Called from the upper layer, to subscribe <es> to events <event_type>. The
1671 * event subscriber <es> is not allowed to change from a previous call as long
1672 * as at least one event is still subscribed. The <event_type> must only be a
1673 * combination of SUB_RETRY_RECV and SUB_RETRY_SEND. It always returns 0.
1674 */
1675static int qc_subscribe(struct conn_stream *cs, int event_type, struct wait_event *es)
1676{
1677 struct qcs *qcs = cs->ctx;
1678 struct qcc *qcc = qcs->qcc;
1679
1680 TRACE_ENTER(QC_EV_STRM_SEND|QC_EV_STRM_RECV, qcc->conn, qcs);
1681
1682 BUG_ON(event_type & ~(SUB_RETRY_SEND|SUB_RETRY_RECV));
1683 BUG_ON(qcs->subs && qcs->subs != es);
1684
1685 es->events |= event_type;
1686 qcs->subs = es;
1687
1688 if (event_type & SUB_RETRY_RECV)
1689 TRACE_DEVEL("subscribe(recv)", QC_EV_STRM_RECV, qcc->conn, qcs);
1690
1691 if (event_type & SUB_RETRY_SEND) {
1692 TRACE_DEVEL("subscribe(send)", QC_EV_STRM_SEND, qcc->conn, qcs);
1693 if (!(qcs->flags & QC_SF_BLK_SFCTL) &&
1694 !LIST_INLIST(&qcs->list)) {
1695 if (qcs->flags & QC_SF_BLK_MFCTL)
1696 LIST_APPEND(&qcc->fctl_list, &qcs->list);
1697 else
1698 LIST_APPEND(&qcc->send_list, &qcs->list);
1699 }
1700 }
1701 TRACE_LEAVE(QC_EV_STRM_SEND|QC_EV_STRM_RECV, qcc->conn, qcs);
1702 return 0;
1703}
1704
1705/* Called from the upper layer, to unsubscribe <es> from events <event_type>.
1706 * The <es> pointer is not allowed to differ from the one passed to the
1707 * subscribe() call. It always returns zero.
1708 */
1709static int qc_unsubscribe(struct conn_stream *cs, int event_type, struct wait_event *es)
1710{
1711 struct qcs *qcs = cs->ctx;
1712
1713 TRACE_ENTER(QC_EV_STRM_SEND|QC_EV_STRM_RECV, qcs->qcc->conn, qcs);
1714
1715 BUG_ON(event_type & ~(SUB_RETRY_SEND|SUB_RETRY_RECV));
1716 BUG_ON(qcs->subs && qcs->subs != es);
1717
1718 es->events &= ~event_type;
1719 if (!es->events)
1720 qcs->subs = NULL;
1721
1722 if (event_type & SUB_RETRY_RECV)
1723 TRACE_DEVEL("unsubscribe(recv)", QC_EV_STRM_RECV, qcs->qcc->conn, qcs);
1724
1725 if (event_type & SUB_RETRY_SEND) {
1726 TRACE_DEVEL("subscribe(send)", QC_EV_STRM_SEND, qcs->qcc->conn, qcs);
1727 qcs->flags &= ~QC_SF_NOTIFIED;
1728 if (!(qcs->flags & (QC_SF_WANT_SHUTR | QC_SF_WANT_SHUTW)))
1729 LIST_DEL_INIT(&qcs->list);
1730 }
1731
1732 TRACE_LEAVE(QC_EV_STRM_SEND|QC_EV_STRM_RECV, qcs->qcc->conn, qcs);
1733 return 0;
1734}
1735
1736
1737/* Called from the upper layer, to subscribe <es> to events <event_type>. The
1738 * event subscriber <es> is not allowed to change from a previous call as long
1739 * as at least one event is still subscribed. The <event_type> must only be a
1740 * SUB_RETRY_RECV. It always returns 0.
1741 */
1742static int ruqs_subscribe(struct qcs *qcs, int event_type, struct wait_event *es)
1743{
1744 struct qcc *qcc = qcs->qcc;
1745
1746 TRACE_ENTER(QC_EV_STRM_RECV, qcc->conn, qcs);
1747
1748 BUG_ON(event_type & ~SUB_RETRY_RECV);
1749 BUG_ON(qcs->subs && qcs->subs != es);
1750
1751 es->events |= event_type;
1752 qcs->subs = es;
1753
1754 if (event_type & SUB_RETRY_RECV)
1755 TRACE_DEVEL("subscribe(recv)", QC_EV_STRM_RECV, qcc->conn, qcs);
1756
1757 TRACE_LEAVE(QC_EV_STRM_RECV, qcc->conn, qcs);
1758 return 0;
1759}
1760
1761/* Called from the upper layer, to unsubscribe <es> from events <event_type>.
1762 * The <es> pointer is not allowed to differ from the one passed to the
1763 * subscribe() call. It always returns zero.
1764 */
1765static int ruqs_unsubscribe(struct qcs *qcs, int event_type, struct wait_event *es)
1766{
1767 TRACE_ENTER(QC_EV_STRM_RECV, qcs->qcc->conn, qcs);
1768
1769 BUG_ON(event_type & ~SUB_RETRY_RECV);
1770 BUG_ON(qcs->subs && qcs->subs != es);
1771
1772 es->events &= ~event_type;
1773 if (!es->events)
1774 qcs->subs = NULL;
1775
1776 if (event_type & SUB_RETRY_RECV)
1777 TRACE_DEVEL("unsubscribe(recv)", QC_EV_STRM_RECV, qcs->qcc->conn, qcs);
1778
1779 TRACE_LEAVE(QC_EV_STRM_RECV, qcs->qcc->conn, qcs);
1780 return 0;
1781}
1782
1783/* Called from the upper layer, to subscribe <es> to events <event_type>. The
1784 * event subscriber <es> is not allowed to change from a previous call as long
1785 * as at least one event is still subscribed. The <event_type> must only be
1786 * SUB_RETRY_SEND. It always returns 0.
1787 */
1788static int luqs_subscribe(struct qcs *qcs, int event_type, struct wait_event *es)
1789{
1790 struct qcc *qcc = qcs->qcc;
1791
1792 TRACE_ENTER(QC_EV_STRM_SEND, qcc->conn, qcs);
1793
1794 BUG_ON(event_type & ~SUB_RETRY_SEND);
1795 BUG_ON(qcs->subs && qcs->subs != es);
1796
1797 es->events |= event_type;
1798 qcs->subs = es;
1799
1800 if (event_type & SUB_RETRY_SEND) {
1801 TRACE_DEVEL("subscribe(send)", QC_EV_STRM_SEND, qcc->conn, qcs);
1802 if (!(qcs->flags & QC_SF_BLK_SFCTL) &&
1803 !LIST_INLIST(&qcs->list)) {
1804 if (qcs->flags & QC_SF_BLK_MFCTL)
1805 LIST_APPEND(&qcc->fctl_list, &qcs->list);
1806 else
1807 LIST_APPEND(&qcc->send_list, &qcs->list);
1808 }
1809 }
1810
1811 TRACE_LEAVE(QC_EV_STRM_SEND, qcc->conn, qcs);
1812 return 0;
1813}
1814
1815/* Called from the upper layer, to unsubscribe <es> from events <event_type>.
1816 * The <es> pointer is not allowed to differ from the one passed to the
1817 * subscribe() call. It always returns zero.
1818 */
1819static int luqs_unsubscribe(struct qcs *qcs, int event_type, struct wait_event *es)
1820{
1821 struct qcc *qcc = qcs->qcc;
1822
1823 TRACE_ENTER(QC_EV_STRM_SEND, qcc->conn, qcs);
1824
1825 BUG_ON(event_type & ~SUB_RETRY_SEND);
1826 BUG_ON(qcs->subs && qcs->subs != es);
1827
1828 es->events &= ~event_type;
1829 if (!es->events)
1830 qcs->subs = NULL;
1831
1832 if (event_type & SUB_RETRY_SEND) {
1833 TRACE_DEVEL("subscribe(send)", QC_EV_STRM_SEND, qcc->conn, qcs);
1834 qcs->flags &= ~QC_SF_NOTIFIED;
1835 if (!(qcs->flags & (QC_SF_WANT_SHUTR | QC_SF_WANT_SHUTW)))
1836 LIST_DEL_INIT(&qcs->list);
1837 }
1838
1839 TRACE_LEAVE(QC_EV_STRM_SEND, qcc->conn, qcs);
1840 return 0;
1841}
1842
1843/* Called from the upper layer, to receive data */
1844static size_t qc_rcv_buf(struct conn_stream *cs, struct buffer *buf, size_t count, int flags)
1845{
1846 struct qcs *qcs = cs->ctx;
1847 struct qcc *qcc = qcs->qcc;
1848 int ret;
1849
1850 ret = 0;
1851 TRACE_ENTER(QC_EV_STRM_RECV, qcc->conn, qcs);
1852 /* XXX TO DO XXX */
1853 TRACE_LEAVE(QC_EV_STRM_RECV, qcc->conn, qcs);
1854 return ret;
1855}
1856
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001857/* for debugging with CLI's "show fd" command */
1858static int qc_show_fd(struct buffer *msg, struct connection *conn)
1859{
1860 struct qcc *qcc = conn->ctx;
1861 struct qcs *qcs = NULL;
1862 struct eb64_node *node;
1863 int fctl_cnt = 0;
1864 int send_cnt = 0;
1865 int tree_cnt = 0;
1866 int orph_cnt = 0;
1867 struct buffer *hmbuf, *tmbuf;
1868
1869 if (!qcc)
1870 return 0;
1871
1872 list_for_each_entry(qcs, &qcc->fctl_list, list)
1873 fctl_cnt++;
1874
1875 list_for_each_entry(qcs, &qcc->send_list, list)
1876 send_cnt++;
1877
1878 qcs = NULL;
1879 node = eb64_first(&qcc->streams_by_id);
1880 while (node) {
1881 qcs = container_of(node, struct qcs, by_id);
1882 tree_cnt++;
1883 if (!qcs->cs)
1884 orph_cnt++;
1885 node = eb64_next(node);
1886 }
1887
1888 hmbuf = br_head(qcc->mbuf);
1889 tmbuf = br_tail(qcc->mbuf);
1890 chunk_appendf(msg, " qcc.st0=%s .flg=0x%04x"
1891 " clt.nb_streams_bidi=%llu srv.nb_streams_bidi=%llu"
1892 " clt.nb_streams_uni=%llu srv.nb_streams_uni=%llu"
1893 " .nbcs=%u .fctl_cnt=%d .send_cnt=%d .tree_cnt=%d"
1894 " .orph_cnt=%d .sub=%d"
1895 " .mbuf=[%u..%u|%u],h=[%u@%p+%u/%u],t=[%u@%p+%u/%u]",
1896 qcc_st_to_str(qcc->st0), qcc->flags,
1897 (unsigned long long)qcc->strms[QCS_CLT_BIDI].nb_streams,
1898 (unsigned long long)qcc->strms[QCS_SRV_BIDI].nb_streams,
1899 (unsigned long long)qcc->strms[QCS_CLT_UNI].nb_streams,
1900 (unsigned long long)qcc->strms[QCS_SRV_UNI].nb_streams,
1901 qcc->nb_cs, fctl_cnt, send_cnt, tree_cnt, orph_cnt,
1902 qcc->wait_event.events,
1903 br_head_idx(qcc->mbuf), br_tail_idx(qcc->mbuf), br_size(qcc->mbuf),
1904 (unsigned int)b_data(hmbuf), b_orig(hmbuf),
1905 (unsigned int)b_head_ofs(hmbuf), (unsigned int)b_size(hmbuf),
1906 (unsigned int)b_data(tmbuf), b_orig(tmbuf),
1907 (unsigned int)b_head_ofs(tmbuf), (unsigned int)b_size(tmbuf));
1908
1909 if (qcs) {
1910 chunk_appendf(msg, " last_qcs=%p .id=%llu rx.st=%s tx.st=%s .flg=0x%04x .rxbuf=%u@%p+%u/%u .cs=%p",
1911 qcs, (unsigned long long)qcs->id,
1912 qcs_rx_st_to_str(qcs->rx.st), qcs_tx_st_to_str(qcs->tx.st), qcs->flags,
1913 (unsigned int)b_data(&qcs->rx.buf), b_orig(&qcs->rx.buf),
1914 (unsigned int)b_head_ofs(&qcs->rx.buf), (unsigned int)b_size(&qcs->rx.buf),
1915 qcs->cs);
1916 if (qcs->cs)
1917 chunk_appendf(msg, " .cs.flg=0x%08x .cs.data=%p",
1918 qcs->cs->flags, qcs->cs->data);
1919 }
1920
1921 return 0;
1922}
1923
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001924/****************************************/
1925/* MUX initialization and instantiation */
1926/***************************************/
1927
1928/* The mux operations */
1929static const struct mux_ops qc_ops = {
1930 .init = qc_init,
1931 .wake = qc_wake,
Amaury Denoyelle26dfd902021-08-24 16:33:53 +02001932 //.snd_buf = qc_snd_buf,
1933 .snd_buf = h3_snd_buf,
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001934 .rcv_buf = qc_rcv_buf,
1935 .subscribe = qc_subscribe,
1936 .unsubscribe = qc_unsubscribe,
1937 .ruqs_subscribe = ruqs_subscribe,
1938 .ruqs_unsubscribe = ruqs_unsubscribe,
1939 .luqs_subscribe = luqs_subscribe,
1940 .luqs_unsubscribe = luqs_unsubscribe,
1941 .attach = qc_attach,
1942 .get_first_cs = qc_get_first_cs,
1943 .detach = qc_detach,
1944 .destroy = qc_destroy,
1945 .avail_streams_bidi = qc_avail_streams_bidi,
1946 .avail_streams_uni = qc_avail_streams_uni,
1947 .used_streams = qc_used_streams,
1948 .shutr = qc_shutr,
1949 .shutw = qc_shutw,
1950 .ctl = qc_ctl,
1951 .show_fd = qc_show_fd,
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001952 .flags = MX_FL_CLEAN_ABRT|MX_FL_HTX|MX_FL_HOL_RISK,
1953 .name = "QUIC",
1954};
1955
1956static struct mux_proto_list mux_proto_quic =
1957 { .token = IST("quic"), .mode = PROTO_MODE_HTTP, .side = PROTO_SIDE_BOTH, .mux = &qc_ops };
1958
1959INITCALL1(STG_REGISTER, register_mux_proto, &mux_proto_quic);
1960