blob: d60df2b6c075c360d19321dd0c95758c7e30d964 [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
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100324static __inline int
325qcc_is_dead(const struct qcc *qcc)
326{
Amaury Denoyelleac8ee252021-10-08 17:57:03 +0200327#if 0
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100328 if (eb_is_empty(&qcc->streams_by_id) && /* don't close if streams exist */
329 ((qcc->conn->flags & CO_FL_ERROR) || /* errors close immediately */
330 (qcc->st0 >= QC_CS_ERROR && !qcc->task) || /* a timeout stroke earlier */
331 (!(qcc->conn->owner)) || /* Nobody's left to take care of the connection, drop it now */
332 (!br_data(qcc->mbuf) && /* mux buffer empty, also process clean events below */
333 conn_xprt_read0_pending(qcc->conn))))
334 return 1;
Amaury Denoyelleac8ee252021-10-08 17:57:03 +0200335#endif
336 if (!qcc->strms[QCS_CLT_BIDI].nb_streams)
337 return 1;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100338
339 return 0;
340}
341
342/*****************************************************/
343/* functions below are for dynamic buffer management */
344/*****************************************************/
345
346/* indicates whether or not the we may call the qc_recv() function to attempt
347 * to receive data into the buffer and/or demux pending data. The condition is
348 * a bit complex due to some API limits for now. The rules are the following :
349 * - if an error or a shutdown was detected on the connection and the buffer
350 * is empty, we must not attempt to receive
351 * - if the demux buf failed to be allocated, we must not try to receive and
352 * we know there is nothing pending
353 * - if no flag indicates a blocking condition, we may attempt to receive,
354 * regardless of whether the demux buffer is full or not, so that only
355 * de demux part decides whether or not to block. This is needed because
356 * the connection API indeed prevents us from re-enabling receipt that is
357 * already enabled in a polled state, so we must always immediately stop
358 * as soon as the demux can't proceed so as never to hit an end of read
359 * with data pending in the buffers.
360 * - otherwise must may not attempt
361 */
362static inline int qc_recv_allowed(const struct qcc *qcc)
363{
364 if (qcc->rx.inmux == 0 &&
365 (qcc->st0 >= QC_CS_ERROR ||
366 qcc->conn->flags & CO_FL_ERROR ||
367 conn_xprt_read0_pending(qcc->conn)))
368 return 0;
369
370 if (!(qcc->flags & QC_CF_DEM_BLOCK_ANY))
371 return 1;
372
373 return 0;
374}
375
376/* restarts reading on the connection if it was not enabled */
377static inline void qcc_restart_reading(const struct qcc *qcc, int consider_buffer)
378{
379 if (!qc_recv_allowed(qcc))
380 return;
381
382 if ((!consider_buffer || !qcc->rx.inmux)
383 && (qcc->wait_event.events & SUB_RETRY_RECV))
384 return;
385
386 tasklet_wakeup(qcc->wait_event.tasklet);
387}
388
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100389static int qc_buf_available(void *target)
390{
Amaury Denoyelle1e308ff2021-10-12 18:14:12 +0200391 struct qcs *qcs = target;
392 if (!b_alloc(&qcs->tx.buf))
393 return 0;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100394
Amaury Denoyelle1e308ff2021-10-12 18:14:12 +0200395 qcc_restart_reading(qcs->qcc, 1);
396 return 1;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100397}
398
Amaury Denoyelle1e308ff2021-10-12 18:14:12 +0200399struct buffer *qc_get_buf(struct qcs *qcs, struct buffer *bptr)
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100400{
401 struct buffer *buf = NULL;
Amaury Denoyelle1e308ff2021-10-12 18:14:12 +0200402 struct qcc *qcc = qcs->qcc;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100403
404 if (likely(!LIST_INLIST(&qcc->buf_wait.list)) &&
405 unlikely((buf = b_alloc(bptr)) == NULL)) {
Amaury Denoyelle1e308ff2021-10-12 18:14:12 +0200406 qcc->buf_wait.target = qcs;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100407 qcc->buf_wait.wakeup_cb = qc_buf_available;
Willy Tarreaub4e34762021-09-30 19:02:18 +0200408 LIST_APPEND(&th_ctx->buffer_wq, &qcc->buf_wait.list);
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100409 }
410
411 return buf;
412}
413
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100414/* returns the number of streams in use on a connection to figure if it's
415 * idle or not. We check nb_cs and not nb_streams as the caller will want
416 * to know if it was the last one after a detach().
417 */
418static int qc_used_streams(struct connection *conn)
419{
420 struct qcc *qcc = conn->ctx;
421
422 return qcc->nb_cs;
423}
424
425/* returns the number of concurrent streams available on the connection with <dir>
426 * as direction
427 */
428static int qc_avail_streams(struct connection *conn, enum qcs_dir dir)
429{
430 struct qcc *qcc = conn->ctx;
431 enum qcs_type qcs_type;
432
433 if (qcc->st0 >= QC_CS_ERROR)
434 return 0;
435
436 qcs_type = qcs_type_from_dir(qcc, dir);
437
438 return qcc->strms[qcs_type].max_streams - qcc->strms[qcs_type].nb_streams;
439}
440
441
442/* returns the number of concurrent bidirectional streams available on the
443 * connection.
444 */
445static int qc_avail_streams_bidi(struct connection *conn)
446{
447 return qc_avail_streams(conn, QCS_BIDI);
448}
449
450/* returns the number of concurrent unidirectional streams available on the
451 * connection.
452 */
453static int qc_avail_streams_uni(struct connection *conn)
454{
455 return qc_avail_streams(conn, QCS_UNI);
456}
457
458/*****************************************************************/
459/* functions below are dedicated to the mux setup and management */
460/*****************************************************************/
461
462/* Update the mux transport parameter after having received remote transpot parameters */
463void quic_mux_transport_params_update(struct qcc *qcc)
464{
465 if (objt_listener(qcc->conn->target)) {
466 struct quic_transport_params *clt_params;
467
468 /* Client parameters, params used to TX. */
469 clt_params = &qcc->conn->qc->tx.params;
470
471 qcc->tx.max_data = clt_params->initial_max_data;
472 /* Client initiated streams must respect the server flow control. */
473 qcc->strms[QCS_CLT_BIDI].rx.max_data = clt_params->initial_max_stream_data_bidi_local;
474 qcc->strms[QCS_CLT_UNI].rx.max_data = clt_params->initial_max_stream_data_uni;
475
476 /* Server initiated streams must respect the server flow control. */
477 qcc->strms[QCS_SRV_BIDI].max_streams = clt_params->initial_max_streams_bidi;
478 qcc->strms[QCS_SRV_BIDI].tx.max_data = clt_params->initial_max_stream_data_bidi_remote;
479
480 qcc->strms[QCS_SRV_UNI].max_streams = clt_params->initial_max_streams_uni;
481 qcc->strms[QCS_SRV_UNI].tx.max_data = clt_params->initial_max_stream_data_uni;
482 }
483 else {
484 struct quic_transport_params *srv_params;
485
486 /* server parameters, TX params. */
487 srv_params = &qcc->conn->qc->tx.params;
488
489 qcc->tx.max_data = srv_params->initial_max_data;
490 /* Client initiated streams must respect the server flow control. */
491 qcc->strms[QCS_CLT_BIDI].max_streams = srv_params->initial_max_streams_bidi;
492 qcc->strms[QCS_CLT_BIDI].tx.max_data = srv_params->initial_max_stream_data_bidi_remote;
493
494 qcc->strms[QCS_CLT_UNI].max_streams = srv_params->initial_max_streams_uni;
495 qcc->strms[QCS_CLT_UNI].tx.max_data = srv_params->initial_max_stream_data_uni;
496
497 /* Server initiated streams must respect the server flow control. */
498 qcc->strms[QCS_SRV_BIDI].rx.max_data = srv_params->initial_max_stream_data_bidi_local;
499 qcc->strms[QCS_SRV_UNI].rx.max_data = srv_params->initial_max_stream_data_uni;
500 }
501
502 /* Now that we have all the flow control information, we can finalize the application
503 * context.
504 */
505 qcc->app_ops->finalize(qcc->ctx);
506}
507
508/* Initialize the mux once it's attached. For outgoing connections, the context
509 * is already initialized before installing the mux, so we detect incoming
510 * connections from the fact that the context is still NULL (even during mux
511 * upgrades). <input> is always used as Input buffer and may contain data. It is
512 * the caller responsibility to not reuse it anymore. Returns < 0 on error.
513 */
514static int qc_init(struct connection *conn, struct proxy *prx,
515 struct session *sess, struct buffer *input)
516{
517 struct qcc *qcc;
518 struct task *t = NULL;
519 void *conn_ctx = conn->ctx;
520
521 TRACE_ENTER(QC_EV_QCC_NEW);
522
523 qcc = pool_alloc(pool_head_qcc);
524 if (!qcc)
525 goto fail_no_qcc;
526
527 if (conn_is_back(conn)) {
528 qcc->flags = QC_CF_IS_BACK;
529 qcc->shut_timeout = qcc->timeout = prx->timeout.server;
530 if (tick_isset(prx->timeout.serverfin))
531 qcc->shut_timeout = prx->timeout.serverfin;
532
533 qcc->px_counters = EXTRA_COUNTERS_GET(prx->extra_counters_be,
534 &qc_stats_module);
535 } else {
536 qcc->flags = QC_CF_NONE;
537 qcc->shut_timeout = qcc->timeout = prx->timeout.client;
538 if (tick_isset(prx->timeout.clientfin))
539 qcc->shut_timeout = prx->timeout.clientfin;
540
541 qcc->px_counters = EXTRA_COUNTERS_GET(prx->extra_counters_fe,
542 &qc_stats_module);
543 }
544
545 qcc->proxy = prx;
546 qcc->task = NULL;
547 if (tick_isset(qcc->timeout)) {
Willy Tarreaubeeabf52021-10-01 18:23:30 +0200548 t = task_new_here();
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100549 if (!t)
550 goto fail;
551
552 qcc->task = t;
553 t->process = qc_timeout_task;
554 t->context = qcc;
555 t->expire = tick_add(now_ms, qcc->timeout);
556 }
557
Amaury Denoyellecde91122021-09-22 15:28:27 +0200558 qcc->subs = NULL;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100559 qcc->wait_event.tasklet = tasklet_new();
560 if (!qcc->wait_event.tasklet)
561 goto fail;
562
563 qcc->wait_event.tasklet->process = qc_io_cb;
564 qcc->wait_event.tasklet->context = qcc;
565 qcc->wait_event.events = 0;
566
567 /* Initialize the context. */
568 qcc->st0 = QC_CS_NOERR;
569 qcc->conn = conn;
570 qcc->conn->qc->qcc = qcc;
571
572 /* Application layer initialization. */
573 qcc->app_ops = &h3_ops;
574 if (!qcc->app_ops->init(qcc))
575 goto fail;
576
577 /* The transports parameters which control the data sent have been stored
578 * in ->tx.params. The ones which control the received data are stored in
579 * in ->rx.params.
580 */
581 if (objt_listener(qcc->conn->target)) {
582 struct quic_transport_params *srv_params;
583
584 /* Server parameters, params used for RX flow control. */
585 srv_params = &conn->qc->rx.params;
586
587 qcc->rx.max_data = srv_params->initial_max_data;
588 qcc->tx.max_data = 0;
589 /* Client initiated streams must respect the server flow control. */
590 qcc->strms[QCS_CLT_BIDI].max_streams = srv_params->initial_max_streams_bidi;
591 qcc->strms[QCS_CLT_BIDI].nb_streams = 0;
592 qcc->strms[QCS_CLT_BIDI].largest_id = -1;
593 qcc->strms[QCS_CLT_BIDI].rx.max_data = 0;
594 qcc->strms[QCS_CLT_BIDI].tx.max_data = srv_params->initial_max_stream_data_bidi_remote;
595
596 qcc->strms[QCS_CLT_UNI].max_streams = srv_params->initial_max_streams_uni;
597 qcc->strms[QCS_CLT_UNI].nb_streams = 0;
598 qcc->strms[QCS_CLT_UNI].largest_id = -1;
599 qcc->strms[QCS_CLT_UNI].rx.max_data = 0;
600 qcc->strms[QCS_CLT_UNI].tx.max_data = srv_params->initial_max_stream_data_uni;
601
602 /* Server initiated streams must respect the server flow control. */
603 qcc->strms[QCS_SRV_BIDI].max_streams = 0;
604 qcc->strms[QCS_SRV_BIDI].nb_streams = 0;
605 qcc->strms[QCS_SRV_BIDI].largest_id = -1;
606 qcc->strms[QCS_SRV_BIDI].rx.max_data = srv_params->initial_max_stream_data_bidi_local;
607 qcc->strms[QCS_SRV_BIDI].tx.max_data = 0;
608
609 qcc->strms[QCS_SRV_UNI].max_streams = 0;
610 qcc->strms[QCS_SRV_UNI].nb_streams = 0;
611 qcc->strms[QCS_SRV_UNI].largest_id = -1;
612 qcc->strms[QCS_SRV_UNI].rx.max_data = srv_params->initial_max_stream_data_uni;
613 qcc->strms[QCS_SRV_UNI].tx.max_data = 0;
614 }
615 else {
616 struct quic_transport_params *clt_params;
617
618 /* client parameters, RX params. */
619 clt_params = &conn->qc->rx.params;
620
621 qcc->rx.max_data = clt_params->initial_max_data;
622 qcc->tx.max_data = 0;
623 /* Client initiated streams must respect the server flow control. */
624 qcc->strms[QCS_CLT_BIDI].max_streams = 0;
625 qcc->strms[QCS_CLT_BIDI].nb_streams = 0;
626 qcc->strms[QCS_CLT_BIDI].largest_id = -1;
627 qcc->strms[QCS_CLT_BIDI].rx.max_data = clt_params->initial_max_stream_data_bidi_local;
628 qcc->strms[QCS_CLT_BIDI].tx.max_data = 0;
629
630 qcc->strms[QCS_CLT_UNI].max_streams = 0;
631 qcc->strms[QCS_CLT_UNI].nb_streams = 0;
632 qcc->strms[QCS_CLT_UNI].largest_id = -1;
633 qcc->strms[QCS_CLT_UNI].rx.max_data = clt_params->initial_max_stream_data_uni;
634 qcc->strms[QCS_CLT_UNI].tx.max_data = 0;
635
636 /* Server initiated streams must respect the server flow control. */
637 qcc->strms[QCS_SRV_BIDI].max_streams = clt_params->initial_max_streams_bidi;
638 qcc->strms[QCS_SRV_BIDI].nb_streams = 0;
639 qcc->strms[QCS_SRV_BIDI].largest_id = -1;
640 qcc->strms[QCS_SRV_BIDI].rx.max_data = 0;
641 qcc->strms[QCS_SRV_BIDI].tx.max_data = clt_params->initial_max_stream_data_bidi_remote;
642
643 qcc->strms[QCS_SRV_UNI].max_streams = clt_params->initial_max_streams_uni;
644 qcc->strms[QCS_SRV_UNI].nb_streams = 0;
645 qcc->strms[QCS_SRV_UNI].largest_id = -1;
646 qcc->strms[QCS_SRV_UNI].rx.max_data = 0;
647 qcc->strms[QCS_SRV_UNI].tx.max_data = clt_params->initial_max_stream_data_uni;
648
649 }
650
651 /* Initialize the streams counters. */
652 qcc->nb_cs = 0;
653 qcc->stream_cnt = 0;
654
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100655 qcc->streams_by_id = EB_ROOT_UNIQUE;
656 LIST_INIT(&qcc->send_list);
657 LIST_INIT(&qcc->fctl_list);
658 LIST_INIT(&qcc->blocked_list);
659 LIST_INIT(&qcc->buf_wait.list);
660 MT_LIST_INIT(&qcc->qcs_rxbuf_wlist);
661
Frédéric Lécaille01abc462021-07-21 09:34:27 +0200662 HA_ATOMIC_STORE(&conn->ctx, qcc);
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100663
664 if (t)
665 task_queue(t);
666
667 if (qcc->flags & QC_CF_IS_BACK) {
668 /* FIXME: For outgoing connections we need to immediately allocate streams.
669 * This highly depends on the QUIC application needs.
670 */
671 }
672
673 HA_ATOMIC_ADD(&qcc->px_counters->open_conns, 1);
674 HA_ATOMIC_ADD(&qcc->px_counters->total_conns, 1);
675
676 /* prepare to read something */
677 qcc_restart_reading(qcc, 1);
678 TRACE_LEAVE(QC_EV_QCC_NEW, conn);
679 return 0;
680
681 fail:
682 task_destroy(t);
683 if (qcc->wait_event.tasklet)
684 tasklet_free(qcc->wait_event.tasklet);
685 pool_free(pool_head_qcc, qcc);
686 fail_no_qcc:
687 conn->ctx = conn_ctx; /* restore saved ctx */
688 TRACE_DEVEL("leaving in error", QC_EV_QCC_NEW|QC_EV_QCC_END|QC_EV_QCC_ERR);
689 return -1;
690}
691
692/* returns the stream associated with id <id> or NULL if not found */
693__maybe_unused
694static inline struct qcs *qcc_st_by_id(struct qcc *qcc, int id)
695{
696 struct eb64_node *node;
697
698 node = eb64_lookup(&qcc->streams_by_id, id);
699 if (!node)
700 return (struct qcs *)qc_closed_stream;
701
702 return container_of(node, struct qcs, by_id);
703}
704
705/* release function. This one should be called to free all resources allocated
706 * to the mux.
707 */
708static void qc_release(struct qcc *qcc)
709{
710 struct connection *conn = NULL;
711
712 TRACE_ENTER(QC_EV_QCC_END);
713
714 if (qcc) {
715 /* The connection must be aattached to this mux to be released */
716 if (qcc->conn && qcc->conn->ctx == qcc)
717 conn = qcc->conn;
718
719 TRACE_DEVEL("freeing qcc", QC_EV_QCC_END, conn);
720
721 if (LIST_INLIST(&qcc->buf_wait.list))
722 LIST_DELETE(&qcc->buf_wait.list);
723
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100724 if (qcc->task) {
725 qcc->task->context = NULL;
726 task_wakeup(qcc->task, TASK_WOKEN_OTHER);
727 qcc->task = NULL;
728 }
729 if (qcc->wait_event.tasklet)
730 tasklet_free(qcc->wait_event.tasklet);
731 if (conn && qcc->wait_event.events != 0)
732 conn->xprt->unsubscribe(conn, conn->xprt_ctx, qcc->wait_event.events,
733 &qcc->wait_event);
734
735 HA_ATOMIC_SUB(&qcc->px_counters->open_conns, 1);
736
737 pool_free(pool_head_qcc, qcc);
738 }
739
740 if (conn) {
741 conn->mux = NULL;
742 conn->ctx = NULL;
743 TRACE_DEVEL("freeing conn", QC_EV_QCC_END, conn);
744
745 conn_stop_tracking(conn);
746 conn_full_close(conn);
747 if (conn->destroy_cb)
748 conn->destroy_cb(conn);
749 conn_free(conn);
750 }
751
752 TRACE_LEAVE(QC_EV_QCC_END);
753}
754
755
756/******************************************************/
757/* functions below are for the QUIC protocol processing */
758/******************************************************/
759
760/* attempt to notify the data layer of recv availability */
761__maybe_unused
762static void qcs_notify_recv(struct qcs *qcs)
763{
764 if (qcs->subs && qcs->subs->events & SUB_RETRY_RECV) {
765 TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn, qcs);
766 tasklet_wakeup(qcs->subs->tasklet);
767 qcs->subs->events &= ~SUB_RETRY_RECV;
768 if (!qcs->subs->events)
769 qcs->subs = NULL;
770 }
771}
772
773/* attempt to notify the data layer of send availability */
774__maybe_unused
775static void qcs_notify_send(struct qcs *qcs)
776{
777 if (qcs->subs && qcs->subs->events & SUB_RETRY_SEND) {
778 TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn, qcs);
779 qcs->flags |= QC_SF_NOTIFIED;
780 tasklet_wakeup(qcs->subs->tasklet);
781 qcs->subs->events &= ~SUB_RETRY_SEND;
782 if (!qcs->subs->events)
783 qcs->subs = NULL;
784 }
785 else if (qcs->flags & (QC_SF_WANT_SHUTR | QC_SF_WANT_SHUTW)) {
786 TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn, qcs);
787 tasklet_wakeup(qcs->shut_tl);
788 }
789}
790
791/* alerts the data layer, trying to wake it up by all means, following
792 * this sequence :
793 * - if the qcs' data layer is subscribed to recv, then it's woken up for recv
794 * - if its subscribed to send, then it's woken up for send
795 * - if it was subscribed to neither, its ->wake() callback is called
796 * It is safe to call this function with a closed stream which doesn't have a
797 * conn_stream anymore.
798 */
799__maybe_unused
800static void qcs_alert(struct qcs *qcs)
801{
802 TRACE_ENTER(QC_EV_QCS_WAKE, qcs->qcc->conn, qcs);
803
804 if (qcs->subs ||
805 (qcs->flags & (QC_SF_WANT_SHUTR | QC_SF_WANT_SHUTW))) {
806 qcs_notify_recv(qcs);
807 qcs_notify_send(qcs);
808 }
809 else if (qcs->cs && qcs->cs->data_cb->wake != NULL) {
810 TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn, qcs);
811 qcs->cs->data_cb->wake(qcs->cs);
812 }
813
814 TRACE_LEAVE(QC_EV_QCS_WAKE, qcs->qcc->conn, qcs);
815}
816
817/* marks stream <qcs> as CLOSED and decrement the number of active streams for
818 * its connection if the stream was not yet closed. Please use this exclusively
819 * before closing a stream to ensure stream count is well maintained.
820 */
821static inline void qcs_close(struct qcs *qcs)
822{
823 TRACE_ENTER(QC_EV_QCS_END, qcs->qcc->conn, qcs);
824 /* XXX TO DO XXX */
825 TRACE_LEAVE(QC_EV_QCS_END, qcs->qcc->conn, qcs);
826}
827
828/* detaches an QUIC stream from its QCC and releases it to the QCS pool. */
829/* qcs_destroy should only ever be called by the thread that owns the stream,
830 * that means that a tasklet should be used if we want to destroy the qcs
831 * from another thread
832 */
833static void qcs_destroy(struct qcs *qcs)
834{
835 struct connection *conn = qcs->qcc->conn;
836
837 TRACE_ENTER(QC_EV_QCS_END, conn, qcs);
838
839 qcs_close(qcs);
840 eb64_delete(&qcs->by_id);
841 if (b_size(&qcs->rx.buf)) {
842 b_free(&qcs->rx.buf);
843 offer_buffers(NULL, 1);
844 }
845
Amaury Denoyelled3d97c62021-10-05 11:45:58 +0200846 b_free(&qcs->tx.buf);
847 b_free(&qcs->tx.xprt_buf);
848
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100849 if (qcs->subs)
850 qcs->subs->events = 0;
851
852 /* There's no need to explicitly call unsubscribe here, the only
853 * reference left would be in the qcc send_list/fctl_list, and if
854 * we're in it, we're getting out anyway
855 */
856 LIST_DEL_INIT(&qcs->list);
Amaury Denoyelled595f102021-09-24 10:05:30 +0200857 --qcs->qcc->strms[qcs_id_type(qcs->id)].nb_streams;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100858
859 /* ditto, calling tasklet_free() here should be ok */
860 tasklet_free(qcs->shut_tl);
861 pool_free(pool_head_qcs, qcs);
862
863 TRACE_LEAVE(QC_EV_QCS_END, conn);
864}
865
866/* allocates a new bidirection stream <id> for connection <qcc> and adds it into qcc's
867 * stream tree. In case of error, nothing is added and NULL is returned. The
868 * causes of errors can be any failed memory allocation. The caller is
869 * responsible for checking if the connection may support an extra stream
870 * prior to calling this function.
871 */
872struct qcs *bidi_qcs_new(struct qcc *qcc, uint64_t id)
873{
874 struct qcs *qcs;
875 enum qcs_type qcs_type;
876
877 TRACE_ENTER(QC_EV_QCS_NEW, qcc->conn);
878
879 qcs = pool_alloc(pool_head_qcs);
880 if (!qcs)
881 goto out;
882
883 qcs->shut_tl = tasklet_new();
884 if (!qcs->shut_tl) {
885 pool_free(pool_head_qcs, qcs);
886 goto out;
887 }
888
889 qcs_type = qcs_id_type(id);
890 qcs->qcc = qcc;
891 qcs->cs = NULL;
892 qcs->id = qcs->by_id.key = id;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100893 qcs->flags = QC_SF_NONE;
894
895 qcs->rx.buf = BUF_NULL;
896 qcs->rx.st = QC_RX_SS_IDLE;
897 qcs->rx.bytes = qcs->rx.offset = 0;
898 qcs->rx.max_data = qcc->strms[qcs_type].rx.max_data;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100899 qcs->rx.buf = BUF_NULL;
Frédéric Lécaille785d3bd2021-09-10 09:13:39 +0200900 qcs->rx.frms = EB_ROOT_UNIQUE;
901
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100902 qcs->tx.st = QC_TX_SS_IDLE;
Frédéric Lécaille785d3bd2021-09-10 09:13:39 +0200903 qcs->tx.bytes = qcs->tx.offset = qcs->tx.ack_offset = 0;
904 qcs->tx.acked_frms = EB_ROOT_UNIQUE;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100905 qcs->tx.max_data = qcc->strms[qcs_type].tx.max_data;
Frédéric Lécaille785d3bd2021-09-10 09:13:39 +0200906 qcs->tx.buf = BUF_NULL;
Amaury Denoyelled3d97c62021-10-05 11:45:58 +0200907 qcs->tx.xprt_buf = BUF_NULL;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100908
909 eb64_insert(&qcc->streams_by_id, &qcs->by_id);
910 qcc->strms[qcs_type].nb_streams++;
911 qcc->stream_cnt++;
912 qcs->subs = NULL;
913 LIST_INIT(&qcs->list);
914 qcs->shut_tl->process = qc_deferred_shut;
915 qcs->shut_tl->context = qcs;
916
917 HA_ATOMIC_ADD(&qcc->px_counters->open_streams, 1);
918 HA_ATOMIC_ADD(&qcc->px_counters->total_streams, 1);
919
920 TRACE_LEAVE(QC_EV_QCS_NEW, qcc->conn, qcs);
921 return qcs;
922
923 out:
924 TRACE_DEVEL("leaving in error", QC_EV_QCS_ERR|QC_EV_QCS_END, qcc->conn);
925 return NULL;
926}
927
928/* Release <qcs> outgoing uni-stream */
929void qcs_release(struct qcs *qcs)
930{
931 eb64_delete(&qcs->by_id);
932 pool_free(pool_head_qcs, qcs);
933}
934
935/* Allocates a locally initiated unidirectional stream. */
936struct qcs *luqs_new(struct qcc *qcc)
937{
938 struct qcs *qcs;
939 uint64_t next_id;
940 enum qcs_type qcs_type;
941
942 TRACE_ENTER(QC_EV_QCS_NEW, qcc->conn);
943
944 qcs = NULL;
945 /* QCS_ID_DIR_BIT bit is set for unidirectional stream. */
946 if (objt_listener(qcc->conn->target))
947 qcs_type = QCS_ID_SRV_INTIATOR_BIT | QCS_ID_DIR_BIT;
948 else
949 qcs_type = QCS_ID_DIR_BIT;
950
951 next_id = qcs_next_id(qcc, qcs_type);
952 if (next_id == (uint64_t)-1) {
953 TRACE_PROTO("No more stream available", QC_EV_QCS_NEW, qcc->conn);
954 goto out;
955 }
956
957 qcs = pool_alloc(pool_head_qcs);
958 if (!qcs)
959 goto out;
960
961 qcs->qcc = qcc;
962 qcs->cs = NULL;
963 qcs->id = qcs->by_id.key = next_id;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100964 qcs->flags = QC_SF_NONE;
965
Frédéric Lécaille785d3bd2021-09-10 09:13:39 +0200966 qcs->tx.st = QC_TX_SS_IDLE;
967 qcs->tx.max_data = qcc->strms[qcs_type].tx.max_data;
968 qcs->tx.offset = qcs->tx.bytes = qcs->tx.ack_offset = 0;
969 qcs->tx.acked_frms = EB_ROOT_UNIQUE;
970 qcs->tx.buf = BUF_NULL;
Amaury Denoyelled3d97c62021-10-05 11:45:58 +0200971 qcs->tx.xprt_buf = BUF_NULL;
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100972
973 qcs->subs = NULL;
974 LIST_INIT(&qcs->list);
Amaury Denoyelle139814a2021-09-24 10:03:16 +0200975 eb64_insert(&qcc->streams_by_id, &qcs->by_id);
Frédéric Lécailledfbae762021-02-18 09:59:01 +0100976
977 TRACE_LEAVE(QC_EV_QCS_NEW, qcc->conn);
978 return qcs;
979
980 out:
981 if (qcs)
982 pool_free(pool_head_qcs, qcs);
983 TRACE_DEVEL("leaving in error", QC_EV_QCS_ERR|QC_EV_QCS_END, qcc->conn);
984 return NULL;
985}
986
987/* Allocates a remotely initiated unidirectional stream. */
988struct qcs *ruqs_new(struct qcc *qcc, uint64_t id)
989{
990 struct qcs *qcs;
991 enum qcs_type qcs_type;
992
993 TRACE_ENTER(QC_EV_QCS_NEW, qcc->conn);
994 qcs = pool_alloc(pool_head_qcs);
995 if (!qcs)
996 goto out;
997
998 qcs_type = qcs_id_type(id);
999
1000 qcs->qcc = qcc;
1001 qcs->cs = NULL;
1002
1003 qcs->qcc = qcc;
1004 qcs->id = qcs->by_id.key = id;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001005 qcs->flags = QC_SF_NONE;
1006
1007 qcs->rx.st = QC_RX_SS_IDLE;
1008 qcs->rx.max_data = qcc->strms[qcs_type].rx.max_data;
1009 qcs->rx.offset = qcs->rx.bytes = 0;
1010 qcs->rx.buf = BUF_NULL;
Frédéric Lécaille785d3bd2021-09-10 09:13:39 +02001011 qcs->rx.frms = EB_ROOT_UNIQUE;
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001012 qcs->tx.buf = BUF_NULL;
1013 qcs->tx.xprt_buf = BUF_NULL;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001014
1015 qcs->subs = NULL;
1016 LIST_INIT(&qcs->list);
Amaury Denoyelle139814a2021-09-24 10:03:16 +02001017 eb64_insert(&qcc->streams_by_id, &qcs->by_id);
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001018
1019 TRACE_LEAVE(QC_EV_QCS_NEW, qcc->conn);
1020 return qcs;
1021
1022 out:
1023 TRACE_DEVEL("leaving in error", QC_EV_QCS_ERR|QC_EV_QCS_END, qcc->conn);
1024 return NULL;
1025}
1026
1027/* attempt to notify the data layer of recv availability */
1028void ruqs_notify_recv(struct qcs *qcs)
1029{
1030 if (qcs->subs && qcs->subs->events & SUB_RETRY_RECV) {
1031 TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn);
1032 tasklet_wakeup(qcs->subs->tasklet);
1033 qcs->subs->events &= ~SUB_RETRY_RECV;
1034 if (!qcs->subs->events)
1035 qcs->subs = NULL;
1036 }
1037}
1038
1039/* Allocates a new stream associated to conn_stream <cs> on the qcc connection
1040 * with dir as direction and returns it, or NULL in case of memory allocation
1041 * error or if the highest possible stream ID was reached.
1042 */
1043static struct qcs *qcc_bck_stream_new(struct qcc *qcc, int dir,
1044 struct conn_stream *cs, struct session *sess)
1045{
1046 struct qcs *qcs = NULL;
1047 enum qcs_type qcs_type;
1048
1049 TRACE_ENTER(QC_EV_QCS_NEW, qcc->conn);
1050
1051 qcs_type = qcs_type_from_dir(qcc, dir);
1052 if (qcc->strms[qcs_type].largest_id + 1 >= qcc->strms[qcs_type].max_streams)
1053 goto out;
1054
1055 /* Defer choosing the ID until we send the first message to create the stream */
1056 qcs = bidi_qcs_new(qcc, qcc->strms[qcs_type].largest_id + 1);
1057 if (!qcs)
1058 goto out;
1059
1060 qcs->cs = cs;
1061 qcs->sess = sess;
1062 cs->ctx = qcs;
1063 qcc->nb_cs++;
1064
1065 out:
1066 if (likely(qcs))
1067 TRACE_LEAVE(QC_EV_QCS_NEW, qcc->conn, qcs);
1068 else
1069 TRACE_LEAVE(QC_EV_QCS_NEW|QC_EV_QCS_ERR|QC_EV_QCS_END, qcc->conn, qcs);
1070 return qcs;
1071}
1072
1073/* Allocates a new bidirectional stream associated to conn_stream <cs> on the <qcc> connection
1074 * and returns it, or NULL in case of memory allocation error or if the highest
1075 * possible stream ID was reached.
1076 */
1077__maybe_unused
1078static struct qcs *qcc_bck_stream_new_bidi(struct qcc *qcc,
1079 struct conn_stream *cs, struct session *sess)
1080{
1081 return qcc_bck_stream_new(qcc, QCS_BIDI, cs, sess);
1082}
1083
1084/* Allocates a new unidirectional stream associated to conn_stream <cs> on the <qcc> connection
1085 * and returns it, or NULL in case of memory allocation error or if the highest
1086 * possible stream ID was reached.
1087 */
1088__maybe_unused
1089static struct qcs *qcc_bck_stream_new_uni(struct qcc *qcc,
1090 struct conn_stream *cs, struct session *sess)
1091{
1092 return qcc_bck_stream_new(qcc, QCS_UNI, cs, sess);
1093}
1094
1095
1096/* wake a specific stream and assign its conn_stream some CS_FL_* flags among
1097 * CS_FL_ERR_PENDING and CS_FL_ERROR if needed. The stream's state
1098 * is automatically updated accordingly. If the stream is orphaned, it is
1099 * destroyed.
1100 */
1101static void qcs_wake_one_stream(struct qcs *qcs)
1102{
1103 struct qcc *qcc = qcs->qcc;
1104
1105 TRACE_ENTER(QC_EV_QCS_WAKE, qcc->conn, qcs);
1106 if (!qcs->cs) {
1107 /* this stream was already orphaned */
1108 qcs_destroy(qcs);
1109 TRACE_DEVEL("leaving with no qcs", QC_EV_QCS_WAKE, qcc->conn);
1110 return;
1111 }
1112 /* XXX TO DO XXX */
1113 TRACE_LEAVE(QC_EV_QCS_WAKE, qcc->conn);
1114}
1115
1116/* wake the streams attached to the connection, whose id is greater than <last>
1117 * or unassigned.
1118 */
1119static void qc_wake_some_streams(struct qcc *qcc, int last)
1120{
1121 struct eb64_node *node;
1122 struct qcs *qcs;
1123
1124 TRACE_ENTER(QC_EV_QCS_WAKE, qcc->conn);
1125
1126 /* Wake all streams with ID > last */
1127 node = eb64_lookup_ge(&qcc->streams_by_id, last + 1);
1128 while (node) {
1129 qcs = container_of(node, struct qcs, by_id);
1130 node = eb64_next(node);
1131 qcs_wake_one_stream(qcs);
1132 }
1133
1134 /* Wake all streams with unassigned ID (ID == 0) */
1135 node = eb64_lookup(&qcc->streams_by_id, 0);
1136 while (node) {
1137 qcs = container_of(node, struct qcs, by_id);
1138 if (qcs->id > 0)
1139 break;
1140 node = eb64_next(node);
1141 qcs_wake_one_stream(qcs);
1142 }
1143
1144 TRACE_LEAVE(QC_EV_QCS_WAKE, qcc->conn);
1145}
1146
1147/* Wake up all blocked streams whose window size has become positive after the
1148 * mux's initial window was adjusted. This should be done after having processed
1149 * SETTINGS frames which have updated the mux's initial window size.
1150 */
1151__maybe_unused
1152static void qcc_unblock_sfctl(struct qcc *qcc)
1153{
1154 TRACE_ENTER(QC_EV_QCC_WAKE, qcc->conn);
1155 /* XXX TO DO XXX */
1156 TRACE_LEAVE(QC_EV_QCC_WAKE, qcc->conn);
1157}
1158
1159/* process Rx frames to be demultiplexed */
1160__maybe_unused
1161static void qc_process_demux(struct qcc *qcc)
1162{
1163 TRACE_ENTER(QC_EV_QCC_WAKE, qcc->conn);
1164 /* XXX TO DO XXX */
1165 TRACE_LEAVE(QC_EV_QCC_WAKE, qcc->conn);
1166}
1167
1168/* resume each qcs eligible for sending in list head <head> */
1169__maybe_unused
1170static void qc_resume_each_sending_qcs(struct qcc *qcc, struct list *head)
1171{
1172 struct qcs *qcs, *qcs_back;
1173
1174 TRACE_ENTER(QC_EV_QCC_SEND|QC_EV_QCS_WAKE, qcc->conn);
1175
1176 list_for_each_entry_safe(qcs, qcs_back, head, list) {
1177 if (qcc_wnd(qcc) <= 0 ||
1178 qcc->flags & QC_CF_MUX_BLOCK_ANY ||
1179 qcc->st0 >= QC_CS_ERROR)
1180 break;
1181
1182 qcs->flags &= ~QC_SF_BLK_ANY;
1183
1184 if (qcs->flags & QC_SF_NOTIFIED)
1185 continue;
1186
1187 /* If the sender changed his mind and unsubscribed, let's just
1188 * remove the stream from the send_list.
1189 */
1190 if (!(qcs->flags & (QC_SF_WANT_SHUTR|QC_SF_WANT_SHUTW)) &&
1191 (!qcs->subs || !(qcs->subs->events & SUB_RETRY_SEND))) {
1192 LIST_DEL_INIT(&qcs->list);
1193 continue;
1194 }
1195
1196 if (qcs->subs && qcs->subs->events & SUB_RETRY_SEND) {
1197 qcs->flags |= QC_SF_NOTIFIED;
1198 tasklet_wakeup(qcs->subs->tasklet);
1199 qcs->subs->events &= ~SUB_RETRY_SEND;
1200 if (!qcs->subs->events)
1201 qcs->subs = NULL;
1202 }
1203 else if (qcs->flags & (QC_SF_WANT_SHUTR|QC_SF_WANT_SHUTW)) {
1204 tasklet_wakeup(qcs->shut_tl);
1205 }
1206 }
1207
1208 TRACE_LEAVE(QC_EV_QCC_SEND|QC_EV_QCS_WAKE, qcc->conn);
1209}
1210
1211/* process Tx frames from streams to be multiplexed. Returns > 0 if it reached
1212 * the end.
1213 */
1214__maybe_unused
1215static int qc_process_mux(struct qcc *qcc)
1216{
1217 TRACE_ENTER(QC_EV_QCC_WAKE, qcc->conn);
Frédéric Lécaille1d402402021-09-20 17:53:17 +02001218
1219 /* First we always process the flow control list because the streams
1220 * waiting there were already elected for immediate emission but were
1221 * blocked just on this.
1222 */
1223 qc_resume_each_sending_qcs(qcc, &qcc->fctl_list);
1224 qc_resume_each_sending_qcs(qcc, &qcc->send_list);
1225
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001226 TRACE_LEAVE(QC_EV_QCC_WAKE, qcc->conn);
Frédéric Lécaille1d402402021-09-20 17:53:17 +02001227 return 1;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001228}
1229
1230
1231/* Attempt to read data, and subscribe if none available.
1232 * The function returns 1 if data has been received, otherwise zero.
1233 */
1234__maybe_unused
1235static int qc_recv(struct qcc *qcc)
1236{
1237 TRACE_ENTER(QC_EV_QCC_RECV, qcc->conn);
1238 /* XXX TO DO XXX */
1239 TRACE_LEAVE(QC_EV_QCC_RECV, qcc->conn);
1240 return 0;
1241}
1242
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001243static int qcs_push_frame(struct qcs *qcs, struct buffer *payload, int fin, uint64_t offset)
1244{
1245 struct quic_frame *frm;
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001246 struct buffer *buf = &qcs->tx.xprt_buf;
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001247 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 +02001248 int total = 0, to_xfer;
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001249
Amaury Denoyelle1e308ff2021-10-12 18:14:12 +02001250 qc_get_buf(qcs, buf);
Frédéric Lécailled2ba0962021-09-20 17:50:03 +02001251 to_xfer = QUIC_MIN(b_data(payload), b_room(buf));
1252 if (!to_xfer)
1253 goto out;
1254
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001255 frm = pool_zalloc(pool_head_quic_frame);
1256 if (!frm)
1257 goto err;
1258
Frédéric Lécailled2ba0962021-09-20 17:50:03 +02001259 total = b_force_xfer(buf, payload, to_xfer);
1260 fin = fin && !b_data(payload);
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001261 frm->type = QUIC_FT_STREAM_8;
1262 if (fin)
1263 frm->type |= QUIC_STREAM_FRAME_TYPE_FIN_BIT;
1264 if (offset) {
1265 frm->type |= QUIC_STREAM_FRAME_TYPE_OFF_BIT;
1266 frm->stream.offset.key = offset;
1267 }
1268 frm->stream.qcs = qcs;
1269 frm->stream.buf = buf;
1270 frm->stream.id = qcs->by_id.key;
1271 if (total) {
1272 frm->type |= QUIC_STREAM_FRAME_TYPE_LEN_BIT;
1273 frm->stream.len = total;
1274 }
1275
1276 MT_LIST_APPEND(&qel->pktns->tx.frms, &frm->mt_list);
Frédéric Lécailled2ba0962021-09-20 17:50:03 +02001277 out:
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001278 fprintf(stderr, "%s: total=%d fin=%d offset=%lu\n", __func__, total, fin, offset);
1279 return total;
1280
1281 err:
1282 return -1;
1283}
1284
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001285/* Try to send data if possible.
1286 * The function returns 1 if data have been sent, otherwise zero.
1287 */
1288static int qc_send(struct qcc *qcc)
1289{
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001290 struct eb64_node *node;
Amaury Denoyellea543eb12021-10-06 14:53:13 +02001291 int ret, done, xprt_wake = 0;
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001292
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001293 TRACE_ENTER(QC_EV_QCC_SEND, qcc->conn);
Frédéric Lécaille8526f142021-09-20 17:58:22 +02001294 ret = done = 0;
1295 /* fill as much as we can into the current buffer */
1296 while (((qcc->flags & (QC_CF_MUX_MFULL|QC_CF_MUX_MALLOC)) == 0) && !done)
1297 done = qc_process_mux(qcc);
1298
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001299 /* TODO simple loop through all streams and check if there is frames to
1300 * send
1301 */
1302 node = eb64_first(&qcc->streams_by_id);
1303 while (node) {
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001304 struct qcs *qcs = container_of(node, struct qcs, by_id);
1305 struct buffer *buf = &qcs->tx.buf;
1306 if (b_data(buf)) {
1307 char fin = 0;
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001308
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001309 /* if FIN is activated, ensure the buffer to
1310 * send is the last
1311 */
1312 if (qcs->flags & QC_SF_FIN_STREAM) {
1313 BUG_ON(b_data(&qcs->tx.buf) < b_data(buf));
1314 fin = (b_data(&qcs->tx.buf) - b_data(buf) == 0);
1315 }
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001316
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001317 ret = qcs_push_frame(qcs, buf, fin, qcs->tx.offset);
1318 if (ret < 0)
1319 ABORT_NOW();
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001320
Amaury Denoyellea543eb12021-10-06 14:53:13 +02001321 if (ret > 0) {
1322 xprt_wake = 1;
1323 if (qcs->flags & QC_SF_BLK_MROOM) {
1324 qcs->flags &= ~QC_SF_BLK_MROOM;
1325 qcs_notify_send(qcs);
1326 }
1327 }
1328
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001329 qcs->tx.offset += ret;
Amaury Denoyellea543eb12021-10-06 14:53:13 +02001330
Amaury Denoyelled3d97c62021-10-05 11:45:58 +02001331 if (b_data(buf)) {
1332 qcc->conn->xprt->subscribe(qcc->conn, qcc->conn->xprt_ctx,
1333 SUB_RETRY_SEND, &qcc->wait_event);
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001334 }
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001335 }
1336 node = eb64_next(node);
1337 }
Frédéric Lécaille8526f142021-09-20 17:58:22 +02001338
Amaury Denoyellea543eb12021-10-06 14:53:13 +02001339 if (xprt_wake)
1340 tasklet_wakeup(((struct ssl_sock_ctx *)(qcc->conn->xprt_ctx))->wait_event.tasklet);
Frédéric Lécaille578a7892021-09-13 16:13:00 +02001341
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001342 TRACE_LEAVE(QC_EV_QCC_SEND, qcc->conn);
1343 return 0;
1344}
1345
1346/* this is the tasklet referenced in qcc->wait_event.tasklet */
1347static struct task *qc_io_cb(struct task *t, void *ctx, unsigned int status)
1348{
Amaury Denoyelle769e9ff2021-10-05 11:43:50 +02001349 struct qcc *qcc = ctx;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001350 int ret = 0;
1351
1352
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001353 if (!(qcc->wait_event.events & SUB_RETRY_SEND))
1354 ret = qc_send(qcc);
1355#if 0
1356 if (!(qcc->wait_event.events & SUB_RETRY_RECV))
1357 ret |= qc_recv(qcc);
1358#endif
1359 // TODO redefine the proper condition here
1360 //if (ret || qcc->rx.inmux)
1361 ret = qc_process(qcc);
1362
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001363leave:
1364 TRACE_LEAVE(QC_EV_QCC_WAKE);
1365 return NULL;
1366}
1367
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001368/* callback called on any event by the connection handler.
1369 * It applies changes and returns zero, or < 0 if it wants immediate
1370 * destruction of the connection (which normally doesn not happen in quic).
1371 */
1372static int qc_process(struct qcc *qcc)
1373{
1374 struct connection *conn = qcc->conn;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001375
1376 TRACE_ENTER(QC_EV_QCC_WAKE, conn);
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001377 TRACE_LEAVE(QC_EV_QCC_WAKE, conn);
1378 return 0;
1379}
1380
1381/* wake-up function called by the connection layer (mux_ops.wake) */
1382static int qc_wake(struct connection *conn)
1383{
1384 struct qcc *qcc = conn->ctx;
1385 int ret;
1386
1387 TRACE_ENTER(QC_EV_QCC_WAKE, conn);
1388 ret = qc_process(qcc);
1389 if (ret >= 0)
1390 qc_wake_some_streams(qcc, 0);
1391 TRACE_LEAVE(QC_EV_QCC_WAKE);
1392 return ret;
1393}
1394
1395/* Connection timeout management. The principle is that if there's no receipt
1396 * nor sending for a certain amount of time, the connection is closed. If the
1397 * MUX buffer still has lying data or is not allocatable, the connection is
1398 * immediately killed. If it's allocatable and empty, we attempt to send a
1399 * GOAWAY frame.
1400 */
1401static struct task *qc_timeout_task(struct task *t, void *context, unsigned int state)
1402{
1403 TRACE_ENTER(QC_EV_QCC_WAKE);
1404 /* XXX TO DO XXX */
1405 TRACE_LEAVE(QC_EV_QCC_WAKE);
1406 return NULL;
1407}
1408
1409
1410/*******************************************/
1411/* functions below are used by the streams */
1412/*******************************************/
1413
1414/*
1415 * Attach a new stream to a connection
1416 * (Used for outgoing connections)
1417 */
1418static struct conn_stream *qc_attach(struct connection *conn, struct session *sess)
1419{
1420 struct conn_stream *cs;
1421 struct qcs *qcs;
1422 struct qcc *qcc = conn->ctx;
1423
1424 TRACE_ENTER(QC_EV_QCS_NEW, conn);
1425 cs = cs_new(conn, conn->target);
1426 if (!cs) {
1427 TRACE_DEVEL("leaving on CS allocation failure", QC_EV_QCS_NEW|QC_EV_QCS_ERR, conn);
1428 return NULL;
1429 }
1430 qcs = qcc_bck_stream_new(qcc, QCS_BIDI, cs, sess);
1431 if (!qcs) {
1432 TRACE_DEVEL("leaving on stream creation failure", QC_EV_QCS_NEW|QC_EV_QCS_ERR, conn);
1433 cs_free(cs);
1434 return NULL;
1435 }
1436 TRACE_LEAVE(QC_EV_QCS_NEW, conn, qcs);
1437 return cs;
1438}
1439
1440/* Retrieves the first valid conn_stream from this connection, or returns NULL.
1441 * We have to scan because we may have some orphan streams. It might be
1442 * beneficial to scan backwards from the end to reduce the likeliness to find
1443 * orphans.
1444 */
1445static const struct conn_stream *qc_get_first_cs(const struct connection *conn)
1446{
1447 struct qcc *qcc = conn->ctx;
1448 struct qcs *qcs;
1449 struct eb64_node *node;
1450
1451 node = eb64_first(&qcc->streams_by_id);
1452 while (node) {
1453 qcs = container_of(node, struct qcs, by_id);
1454 if (qcs->cs)
1455 return qcs->cs;
1456 node = eb64_next(node);
1457 }
1458 return NULL;
1459}
1460
1461static int qc_ctl(struct connection *conn, enum mux_ctl_type mux_ctl, void *output)
1462{
1463 int ret = 0;
1464 struct qcc *qcc = conn->ctx;
1465
1466 switch (mux_ctl) {
1467 case MUX_STATUS:
1468 /* Only consider the mux to be ready if we had no error. */
1469 if (qcc->st0 < QC_CS_ERROR)
1470 ret |= MUX_STATUS_READY;
1471 return ret;
1472 case MUX_EXIT_STATUS:
1473 return MUX_ES_UNKNOWN;
1474 default:
1475 return -1;
1476 }
1477}
1478
1479/*
1480 * Destroy the mux and the associated connection, if it is no longer used
1481 */
1482static void qc_destroy(void *ctx)
1483{
1484 struct qcc *qcc = ctx;
1485
1486 TRACE_ENTER(QC_EV_QCC_END, qcc->conn);
1487 if (eb_is_empty(&qcc->streams_by_id) || !qcc->conn || qcc->conn->ctx != qcc)
1488 qc_release(qcc);
1489 TRACE_LEAVE(QC_EV_QCC_END);
1490}
1491
1492/*
1493 * Detach the stream from the connection and possibly release the connection.
1494 */
1495static void qc_detach(struct conn_stream *cs)
1496{
1497 struct qcs *qcs = cs->ctx;
Amaury Denoyellecd28b272021-09-22 14:48:32 +02001498 struct qcc *qcc = qcs->qcc;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001499
1500 TRACE_ENTER(QC_EV_STRM_END, qcs ? qcs->qcc->conn : NULL, qcs);
Amaury Denoyelle493bb1d2021-10-13 16:34:49 +02001501 if (b_data(&qcs->tx.buf) ||
1502 (b_data(&qcs->tx.xprt_buf) && !(qcc->flags & QC_CF_CC_RECV))) {
Amaury Denoyellecae07912021-10-08 17:57:41 +02001503 qcs->flags |= QC_SF_DETACH;
1504 goto out;
1505 }
1506
Amaury Denoyellecd28b272021-09-22 14:48:32 +02001507 qcs_destroy(qcs);
Amaury Denoyelle493bb1d2021-10-13 16:34:49 +02001508 if (qcc_is_dead(qcc)) {
Amaury Denoyellecd28b272021-09-22 14:48:32 +02001509 qc_release(qcc);
Amaury Denoyelle493bb1d2021-10-13 16:34:49 +02001510 TRACE_LEAVE(QC_EV_STRM_END, NULL);
1511 return;
1512 }
Amaury Denoyellecae07912021-10-08 17:57:41 +02001513
1514 out:
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001515 TRACE_LEAVE(QC_EV_STRM_END, qcs ? qcs->qcc->conn : NULL);
1516}
1517
1518/* Performs a synchronous or asynchronous shutr(). */
1519static void qc_do_shutr(struct qcs *qcs)
1520{
1521 struct qcc *qcc = qcs->qcc;
1522
1523 TRACE_ENTER(QC_EV_STRM_SHUT, qcc->conn, qcs);
1524 /* XXX TO DO XXX */
1525 TRACE_LEAVE(QC_EV_STRM_SHUT, qcc->conn, qcs);
1526 return;
1527}
1528
1529/* Performs a synchronous or asynchronous shutw(). */
1530static void qc_do_shutw(struct qcs *qcs)
1531{
1532 struct qcc *qcc = qcs->qcc;
1533
1534 TRACE_ENTER(QC_EV_STRM_SHUT, qcc->conn, qcs);
1535 /* XXX TO DO XXX */
1536 TRACE_LEAVE(QC_EV_STRM_SHUT, qcc->conn, qcs);
1537 return;
1538}
1539
1540/* This is the tasklet referenced in qcs->shut_tl, it is used for
1541 * deferred shutdowns when the qc_detach() was done but the mux buffer was full
1542 * and prevented the last frame from being emitted.
1543 */
1544static struct task *qc_deferred_shut(struct task *t, void *ctx, unsigned int state)
1545{
1546 struct qcs *qcs = ctx;
1547 struct qcc *qcc = qcs->qcc;
1548
1549 TRACE_ENTER(QC_EV_STRM_SHUT, qcc->conn, qcs);
1550
Amaury Denoyellecae07912021-10-08 17:57:41 +02001551 if (qcs->flags & QC_SF_NOTIFIED ||
1552 (b_data(&qcs->tx.buf) || b_data(&qcs->tx.xprt_buf))) {
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001553 /* some data processing remains to be done first */
1554 goto end;
1555 }
1556
1557 if (qcs->flags & QC_SF_WANT_SHUTW)
1558 qc_do_shutw(qcs);
1559
1560 if (qcs->flags & QC_SF_WANT_SHUTR)
1561 qc_do_shutr(qcs);
1562
1563 if (!(qcs->flags & (QC_SF_WANT_SHUTR|QC_SF_WANT_SHUTW))) {
1564 /* We're done trying to send, remove ourself from the send_list */
1565 LIST_DEL_INIT(&qcs->list);
1566
1567 if (!qcs->cs) {
1568 qcs_destroy(qcs);
1569 if (qcc_is_dead(qcc))
1570 qc_release(qcc);
1571 }
1572 }
1573
1574 end:
1575 TRACE_LEAVE(QC_EV_STRM_SHUT);
1576 return NULL;
1577}
1578
1579/* shutr() called by the conn_stream (mux_ops.shutr) */
1580static void qc_shutr(struct conn_stream *cs, enum cs_shr_mode mode)
1581{
1582
1583 struct qcs *qcs = cs->ctx;
1584
1585 TRACE_ENTER(QC_EV_STRM_SHUT, qcs->qcc->conn, qcs);
1586 if (cs->flags & CS_FL_KILL_CONN)
1587 qcs->flags |= QC_SF_KILL_CONN;
1588
1589 if (mode)
1590 qc_do_shutr(qcs);
1591
1592 TRACE_LEAVE(QC_EV_STRM_SHUT, qcs->qcc->conn, qcs);
1593}
1594
1595/* shutw() called by the conn_stream (mux_ops.shutw) */
1596static void qc_shutw(struct conn_stream *cs, enum cs_shw_mode mode)
1597{
1598 struct qcs *qcs = cs->ctx;
1599
1600 TRACE_ENTER(QC_EV_STRM_SHUT, qcs->qcc->conn, qcs);
1601 if (cs->flags & CS_FL_KILL_CONN)
1602 qcs->flags |= QC_SF_KILL_CONN;
1603
1604 qc_do_shutw(qcs);
1605 TRACE_LEAVE(QC_EV_STRM_SHUT, qcs->qcc->conn, qcs);
1606}
1607
1608/* Called from the upper layer, to subscribe <es> to events <event_type>. The
1609 * event subscriber <es> is not allowed to change from a previous call as long
1610 * as at least one event is still subscribed. The <event_type> must only be a
1611 * combination of SUB_RETRY_RECV and SUB_RETRY_SEND. It always returns 0.
1612 */
1613static int qc_subscribe(struct conn_stream *cs, int event_type, struct wait_event *es)
1614{
1615 struct qcs *qcs = cs->ctx;
1616 struct qcc *qcc = qcs->qcc;
1617
1618 TRACE_ENTER(QC_EV_STRM_SEND|QC_EV_STRM_RECV, qcc->conn, qcs);
1619
1620 BUG_ON(event_type & ~(SUB_RETRY_SEND|SUB_RETRY_RECV));
1621 BUG_ON(qcs->subs && qcs->subs != es);
1622
1623 es->events |= event_type;
1624 qcs->subs = es;
1625
1626 if (event_type & SUB_RETRY_RECV)
1627 TRACE_DEVEL("subscribe(recv)", QC_EV_STRM_RECV, qcc->conn, qcs);
1628
1629 if (event_type & SUB_RETRY_SEND) {
1630 TRACE_DEVEL("subscribe(send)", QC_EV_STRM_SEND, qcc->conn, qcs);
1631 if (!(qcs->flags & QC_SF_BLK_SFCTL) &&
1632 !LIST_INLIST(&qcs->list)) {
1633 if (qcs->flags & QC_SF_BLK_MFCTL)
1634 LIST_APPEND(&qcc->fctl_list, &qcs->list);
1635 else
1636 LIST_APPEND(&qcc->send_list, &qcs->list);
1637 }
1638 }
1639 TRACE_LEAVE(QC_EV_STRM_SEND|QC_EV_STRM_RECV, qcc->conn, qcs);
1640 return 0;
1641}
1642
1643/* Called from the upper layer, to unsubscribe <es> from events <event_type>.
1644 * The <es> pointer is not allowed to differ from the one passed to the
1645 * subscribe() call. It always returns zero.
1646 */
1647static int qc_unsubscribe(struct conn_stream *cs, int event_type, struct wait_event *es)
1648{
1649 struct qcs *qcs = cs->ctx;
1650
1651 TRACE_ENTER(QC_EV_STRM_SEND|QC_EV_STRM_RECV, qcs->qcc->conn, qcs);
1652
1653 BUG_ON(event_type & ~(SUB_RETRY_SEND|SUB_RETRY_RECV));
1654 BUG_ON(qcs->subs && qcs->subs != es);
1655
1656 es->events &= ~event_type;
1657 if (!es->events)
1658 qcs->subs = NULL;
1659
1660 if (event_type & SUB_RETRY_RECV)
1661 TRACE_DEVEL("unsubscribe(recv)", QC_EV_STRM_RECV, qcs->qcc->conn, qcs);
1662
1663 if (event_type & SUB_RETRY_SEND) {
1664 TRACE_DEVEL("subscribe(send)", QC_EV_STRM_SEND, qcs->qcc->conn, qcs);
1665 qcs->flags &= ~QC_SF_NOTIFIED;
1666 if (!(qcs->flags & (QC_SF_WANT_SHUTR | QC_SF_WANT_SHUTW)))
1667 LIST_DEL_INIT(&qcs->list);
1668 }
1669
1670 TRACE_LEAVE(QC_EV_STRM_SEND|QC_EV_STRM_RECV, qcs->qcc->conn, qcs);
1671 return 0;
1672}
1673
1674
1675/* Called from the upper layer, to subscribe <es> to events <event_type>. The
1676 * event subscriber <es> is not allowed to change from a previous call as long
1677 * as at least one event is still subscribed. The <event_type> must only be a
1678 * SUB_RETRY_RECV. It always returns 0.
1679 */
1680static int ruqs_subscribe(struct qcs *qcs, int event_type, struct wait_event *es)
1681{
1682 struct qcc *qcc = qcs->qcc;
1683
1684 TRACE_ENTER(QC_EV_STRM_RECV, qcc->conn, qcs);
1685
1686 BUG_ON(event_type & ~SUB_RETRY_RECV);
1687 BUG_ON(qcs->subs && qcs->subs != es);
1688
1689 es->events |= event_type;
1690 qcs->subs = es;
1691
1692 if (event_type & SUB_RETRY_RECV)
1693 TRACE_DEVEL("subscribe(recv)", QC_EV_STRM_RECV, qcc->conn, qcs);
1694
1695 TRACE_LEAVE(QC_EV_STRM_RECV, qcc->conn, qcs);
1696 return 0;
1697}
1698
1699/* Called from the upper layer, to unsubscribe <es> from events <event_type>.
1700 * The <es> pointer is not allowed to differ from the one passed to the
1701 * subscribe() call. It always returns zero.
1702 */
1703static int ruqs_unsubscribe(struct qcs *qcs, int event_type, struct wait_event *es)
1704{
1705 TRACE_ENTER(QC_EV_STRM_RECV, qcs->qcc->conn, qcs);
1706
1707 BUG_ON(event_type & ~SUB_RETRY_RECV);
1708 BUG_ON(qcs->subs && qcs->subs != es);
1709
1710 es->events &= ~event_type;
1711 if (!es->events)
1712 qcs->subs = NULL;
1713
1714 if (event_type & SUB_RETRY_RECV)
1715 TRACE_DEVEL("unsubscribe(recv)", QC_EV_STRM_RECV, qcs->qcc->conn, qcs);
1716
1717 TRACE_LEAVE(QC_EV_STRM_RECV, qcs->qcc->conn, qcs);
1718 return 0;
1719}
1720
1721/* Called from the upper layer, to subscribe <es> to events <event_type>. The
1722 * event subscriber <es> is not allowed to change from a previous call as long
1723 * as at least one event is still subscribed. The <event_type> must only be
1724 * SUB_RETRY_SEND. It always returns 0.
1725 */
1726static int luqs_subscribe(struct qcs *qcs, int event_type, struct wait_event *es)
1727{
1728 struct qcc *qcc = qcs->qcc;
1729
1730 TRACE_ENTER(QC_EV_STRM_SEND, qcc->conn, qcs);
1731
1732 BUG_ON(event_type & ~SUB_RETRY_SEND);
1733 BUG_ON(qcs->subs && qcs->subs != es);
1734
1735 es->events |= event_type;
1736 qcs->subs = es;
1737
1738 if (event_type & SUB_RETRY_SEND) {
1739 TRACE_DEVEL("subscribe(send)", QC_EV_STRM_SEND, qcc->conn, qcs);
1740 if (!(qcs->flags & QC_SF_BLK_SFCTL) &&
1741 !LIST_INLIST(&qcs->list)) {
1742 if (qcs->flags & QC_SF_BLK_MFCTL)
1743 LIST_APPEND(&qcc->fctl_list, &qcs->list);
1744 else
1745 LIST_APPEND(&qcc->send_list, &qcs->list);
1746 }
1747 }
1748
1749 TRACE_LEAVE(QC_EV_STRM_SEND, qcc->conn, qcs);
1750 return 0;
1751}
1752
1753/* Called from the upper layer, to unsubscribe <es> from events <event_type>.
1754 * The <es> pointer is not allowed to differ from the one passed to the
1755 * subscribe() call. It always returns zero.
1756 */
1757static int luqs_unsubscribe(struct qcs *qcs, int event_type, struct wait_event *es)
1758{
1759 struct qcc *qcc = qcs->qcc;
1760
1761 TRACE_ENTER(QC_EV_STRM_SEND, qcc->conn, qcs);
1762
1763 BUG_ON(event_type & ~SUB_RETRY_SEND);
1764 BUG_ON(qcs->subs && qcs->subs != es);
1765
1766 es->events &= ~event_type;
1767 if (!es->events)
1768 qcs->subs = NULL;
1769
1770 if (event_type & SUB_RETRY_SEND) {
1771 TRACE_DEVEL("subscribe(send)", QC_EV_STRM_SEND, qcc->conn, qcs);
1772 qcs->flags &= ~QC_SF_NOTIFIED;
1773 if (!(qcs->flags & (QC_SF_WANT_SHUTR | QC_SF_WANT_SHUTW)))
1774 LIST_DEL_INIT(&qcs->list);
1775 }
1776
1777 TRACE_LEAVE(QC_EV_STRM_SEND, qcc->conn, qcs);
1778 return 0;
1779}
1780
1781/* Called from the upper layer, to receive data */
1782static size_t qc_rcv_buf(struct conn_stream *cs, struct buffer *buf, size_t count, int flags)
1783{
1784 struct qcs *qcs = cs->ctx;
1785 struct qcc *qcc = qcs->qcc;
1786 int ret;
1787
1788 ret = 0;
1789 TRACE_ENTER(QC_EV_STRM_RECV, qcc->conn, qcs);
1790 /* XXX TO DO XXX */
1791 TRACE_LEAVE(QC_EV_STRM_RECV, qcc->conn, qcs);
1792 return ret;
1793}
1794
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001795/* for debugging with CLI's "show fd" command */
1796static int qc_show_fd(struct buffer *msg, struct connection *conn)
1797{
1798 struct qcc *qcc = conn->ctx;
1799 struct qcs *qcs = NULL;
1800 struct eb64_node *node;
1801 int fctl_cnt = 0;
1802 int send_cnt = 0;
1803 int tree_cnt = 0;
1804 int orph_cnt = 0;
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001805
1806 if (!qcc)
1807 return 0;
1808
1809 list_for_each_entry(qcs, &qcc->fctl_list, list)
1810 fctl_cnt++;
1811
1812 list_for_each_entry(qcs, &qcc->send_list, list)
1813 send_cnt++;
1814
1815 qcs = NULL;
1816 node = eb64_first(&qcc->streams_by_id);
1817 while (node) {
1818 qcs = container_of(node, struct qcs, by_id);
1819 tree_cnt++;
1820 if (!qcs->cs)
1821 orph_cnt++;
1822 node = eb64_next(node);
1823 }
1824
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001825 chunk_appendf(msg, " qcc.st0=%s .flg=0x%04x"
1826 " clt.nb_streams_bidi=%llu srv.nb_streams_bidi=%llu"
1827 " clt.nb_streams_uni=%llu srv.nb_streams_uni=%llu"
1828 " .nbcs=%u .fctl_cnt=%d .send_cnt=%d .tree_cnt=%d"
Amaury Denoyelle1e308ff2021-10-12 18:14:12 +02001829 " .orph_cnt=%d .sub=%d",
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001830 qcc_st_to_str(qcc->st0), qcc->flags,
1831 (unsigned long long)qcc->strms[QCS_CLT_BIDI].nb_streams,
1832 (unsigned long long)qcc->strms[QCS_SRV_BIDI].nb_streams,
1833 (unsigned long long)qcc->strms[QCS_CLT_UNI].nb_streams,
1834 (unsigned long long)qcc->strms[QCS_SRV_UNI].nb_streams,
1835 qcc->nb_cs, fctl_cnt, send_cnt, tree_cnt, orph_cnt,
Amaury Denoyelle1e308ff2021-10-12 18:14:12 +02001836 qcc->wait_event.events);
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001837
1838 if (qcs) {
1839 chunk_appendf(msg, " last_qcs=%p .id=%llu rx.st=%s tx.st=%s .flg=0x%04x .rxbuf=%u@%p+%u/%u .cs=%p",
1840 qcs, (unsigned long long)qcs->id,
1841 qcs_rx_st_to_str(qcs->rx.st), qcs_tx_st_to_str(qcs->tx.st), qcs->flags,
1842 (unsigned int)b_data(&qcs->rx.buf), b_orig(&qcs->rx.buf),
1843 (unsigned int)b_head_ofs(&qcs->rx.buf), (unsigned int)b_size(&qcs->rx.buf),
1844 qcs->cs);
1845 if (qcs->cs)
1846 chunk_appendf(msg, " .cs.flg=0x%08x .cs.data=%p",
1847 qcs->cs->flags, qcs->cs->data);
1848 }
1849
1850 return 0;
1851}
1852
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001853/****************************************/
1854/* MUX initialization and instantiation */
1855/***************************************/
1856
1857/* The mux operations */
1858static const struct mux_ops qc_ops = {
1859 .init = qc_init,
1860 .wake = qc_wake,
Amaury Denoyelle26dfd902021-08-24 16:33:53 +02001861 //.snd_buf = qc_snd_buf,
1862 .snd_buf = h3_snd_buf,
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001863 .rcv_buf = qc_rcv_buf,
1864 .subscribe = qc_subscribe,
1865 .unsubscribe = qc_unsubscribe,
1866 .ruqs_subscribe = ruqs_subscribe,
1867 .ruqs_unsubscribe = ruqs_unsubscribe,
1868 .luqs_subscribe = luqs_subscribe,
1869 .luqs_unsubscribe = luqs_unsubscribe,
1870 .attach = qc_attach,
1871 .get_first_cs = qc_get_first_cs,
1872 .detach = qc_detach,
1873 .destroy = qc_destroy,
1874 .avail_streams_bidi = qc_avail_streams_bidi,
1875 .avail_streams_uni = qc_avail_streams_uni,
1876 .used_streams = qc_used_streams,
1877 .shutr = qc_shutr,
1878 .shutw = qc_shutw,
1879 .ctl = qc_ctl,
1880 .show_fd = qc_show_fd,
Frédéric Lécailledfbae762021-02-18 09:59:01 +01001881 .flags = MX_FL_CLEAN_ABRT|MX_FL_HTX|MX_FL_HOL_RISK,
1882 .name = "QUIC",
1883};
1884
1885static struct mux_proto_list mux_proto_quic =
1886 { .token = IST("quic"), .mode = PROTO_MODE_HTTP, .side = PROTO_SIDE_BOTH, .mux = &qc_ops };
1887
1888INITCALL1(STG_REGISTER, register_mux_proto, &mux_proto_quic);
1889