| /* |
| * QUIC mux-demux for connections |
| * |
| * Copyright 2021 HAProxy Technologies, Frédéric Lécaille <flecaille@haproxy.com> |
| * |
| * This program is free software; you can redistribute it and/or |
| * modify it under the terms of the GNU General Public License |
| * as published by the Free Software Foundation; either version |
| * 2 of the License, or (at your option) any later version. |
| * |
| */ |
| |
| #include <import/eb64tree.h> |
| |
| #include <haproxy/api.h> |
| #include <haproxy/cfgparse.h> |
| #include <haproxy/connection.h> |
| #include <haproxy/h3.h> |
| #include <haproxy/istbuf.h> |
| #include <haproxy/log.h> |
| #include <haproxy/mux_quic.h> |
| #include <haproxy/net_helper.h> |
| #include <haproxy/quic_frame.h> |
| #include <haproxy/session-t.h> |
| #include <haproxy/ssl_sock-t.h> |
| #include <haproxy/stats.h> |
| #include <haproxy/stream.h> |
| #include <haproxy/stream_interface.h> |
| #include <haproxy/trace.h> |
| |
| /* dummy streams returned for closed, error, refused, idle and states */ |
| static const struct qcs *qc_closed_stream; |
| |
| #define QC_SS_MASK(state) (1UL << (state)) |
| #define QC_SS_IDLE_BIT (1UL << QC_SS_IDLE) |
| #define QC_SS_RLOC_BIT (1UL << QC_SS_RLOC) |
| #define QC_SS_RREM_BIT (1UL << QC_SS_RREM) |
| #define QC_SS_OPEN_BIT (1UL << QC_SS_OPEN) |
| #define QC_SS_HREM_BIT (1UL << QC_SS_HREM) |
| #define QC_SS_HLOC_BIT (1UL << QC_SS_HLOC) |
| #define QC_SS_ERROR_BIT (1UL << QC_SS_ERROR) |
| #define QC_SS_CLOSED_BIT (1UL << QC_SS_CLOSED) |
| |
| |
| /* trace source and events */ |
| static void qc_trace(enum trace_level level, uint64_t mask, \ |
| const struct trace_source *src, |
| const struct ist where, const struct ist func, |
| const void *a1, const void *a2, const void *a3, const void *a4); |
| |
| /* The event representation is split like this : |
| * strm - application layer |
| * qcs - internal QUIC stream |
| * qcc - internal QUIC connection |
| * conn - external connection |
| * |
| */ |
| static const struct trace_event qc_trace_events[] = { |
| #define QC_EV_QCC_NEW (1ULL << 0) |
| { .mask = QC_EV_QCC_NEW, .name = "qcc_new", .desc = "new QUIC connection" }, |
| #define QC_EV_QCC_RECV (1ULL << 1) |
| { .mask = QC_EV_QCC_RECV, .name = "qcc_recv", .desc = "Rx on QUIC connection" }, |
| #define QC_EV_QCC_SEND (1ULL << 2) |
| { .mask = QC_EV_QCC_SEND, .name = "qcc_send", .desc = "Tx on QUIC connection" }, |
| #define QC_EV_QCC_FCTL (1ULL << 3) |
| { .mask = QC_EV_QCC_FCTL, .name = "qcc_fctl", .desc = "QUIC connection flow-controlled" }, |
| #define QC_EV_QCC_BLK (1ULL << 4) |
| { .mask = QC_EV_QCC_BLK, .name = "qcc_blk", .desc = "QUIC connection blocked" }, |
| #define QC_EV_QCC_WAKE (1ULL << 5) |
| { .mask = QC_EV_QCC_WAKE, .name = "qcc_wake", .desc = "QUIC connection woken up" }, |
| #define QC_EV_QCC_END (1ULL << 6) |
| { .mask = QC_EV_QCC_END, .name = "qcc_end", .desc = "QUIC connection terminated" }, |
| #define QC_EV_QCC_ERR (1ULL << 7) |
| { .mask = QC_EV_QCC_ERR, .name = "qcc_err", .desc = "error on QUIC connection" }, |
| #define QC_EV_TX_FRAME (1ULL << 8) |
| { .mask = QC_EV_TX_FRAME, .name = "tx_frame", .desc = "transmission of any QUIC frame" }, |
| #define QC_EV_QCS_NEW (1ULL << 9) |
| { .mask = QC_EV_QCS_NEW, .name = "qcs_new", .desc = "new QUIC stream" }, |
| #define QC_EV_QCS_GET (1ULL << 10) |
| { .mask = QC_EV_QCS_GET, .name = "qcs_get", .desc = "get QUIC stream by ID" }, |
| #define QC_EV_QCS_SEND (1ULL << 11) |
| { .mask = QC_EV_QCS_SEND, .name = "qcs_send", .desc = "Tx for QUIC stream" }, |
| #define QC_EV_QCS_FCTL (1ULL << 12) |
| { .mask = QC_EV_QCS_FCTL, .name = "qcs_fctl", .desc = "QUIC stream flow-controlled" }, |
| #define QC_EV_QCS_BLK (1ULL << 13) |
| { .mask = QC_EV_QCS_BLK, .name = "qcs_blk", .desc = "QUIC stream blocked" }, |
| #define QC_EV_QCS_WAKE (1ULL << 14) |
| { .mask = QC_EV_QCS_WAKE, .name = "qcs_wake", .desc = "QUIC stream woken up" }, |
| #define QC_EV_QCS_END (1ULL << 15) |
| { .mask = QC_EV_QCS_END, .name = "qcs_end", .desc = "QUIC stream terminated" }, |
| #define QC_EV_QCS_ERR (1ULL << 16) |
| { .mask = QC_EV_QCS_ERR, .name = "qcs_err", .desc = "error on QUIC stream" }, |
| #define QC_EV_STRM_NEW (1ULL << 17) |
| { .mask = QC_EV_STRM_NEW, .name = "strm_new", .desc = "app-layer stream creation" }, |
| #define QC_EV_STRM_RECV (1ULL << 18) |
| { .mask = QC_EV_STRM_RECV, .name = "strm_recv", .desc = "receiving data for stream" }, |
| #define QC_EV_STRM_SEND (1ULL << 19) |
| { .mask = QC_EV_STRM_SEND, .name = "strm_send", .desc = "sending data for stream" }, |
| #define QC_EV_STRM_FULL (1ULL << 20) |
| { .mask = QC_EV_STRM_FULL, .name = "strm_full", .desc = "stream buffer full" }, |
| #define QC_EV_STRM_WAKE (1ULL << 21) |
| { .mask = QC_EV_STRM_WAKE, .name = "strm_wake", .desc = "stream woken up" }, |
| #define QC_EV_STRM_SHUT (1ULL << 22) |
| { .mask = QC_EV_STRM_SHUT, .name = "strm_shut", .desc = "stream shutdown" }, |
| #define QC_EV_STRM_END (1ULL << 23) |
| { .mask = QC_EV_STRM_END, .name = "strm_end", .desc = "detaching app-layer stream" }, |
| #define QC_EV_STRM_ERR (1ULL << 24) |
| { .mask = QC_EV_STRM_ERR, .name = "strm_err", .desc = "stream error" }, |
| { } |
| }; |
| |
| static const struct name_desc qc_trace_lockon_args[4] = { |
| /* arg1 */ { /* already used by the connection */ }, |
| /* arg2 */ { .name = "qcs", .desc = "QUIC stream" }, |
| /* arg3 */ { }, |
| /* arg4 */ { } |
| }; |
| |
| static const struct name_desc qc_trace_decoding[] = { |
| #define QC_VERB_CLEAN 1 |
| { .name="clean", .desc="only user-friendly stuff, generally suitable for level \"user\"" }, |
| #define QC_VERB_MINIMAL 2 |
| { .name="minimal", .desc="report only qcc/qcs state and flags, no real decoding" }, |
| #define QC_VERB_SIMPLE 3 |
| { .name="simple", .desc="add request/response status line or frame info when available" }, |
| #define QC_VERB_ADVANCED 4 |
| { .name="advanced", .desc="add header fields or frame decoding when available" }, |
| #define QC_VERB_COMPLETE 5 |
| { .name="complete", .desc="add full data dump when available" }, |
| { /* end */ } |
| }; |
| |
| static struct trace_source trace_mux_quic = { |
| .name = IST("mux_quic"), |
| .desc = "QUIC multiplexer", |
| .arg_def = TRC_ARG1_CONN, // TRACE()'s first argument is always a connection |
| .default_cb = qc_trace, |
| .known_events = qc_trace_events, |
| .lockon_args = qc_trace_lockon_args, |
| .decoding = qc_trace_decoding, |
| .report_events = ~0, // report everything by default |
| }; |
| |
| #define TRACE_SOURCE &trace_mux_quic |
| INITCALL1(STG_REGISTER, trace_register_source, TRACE_SOURCE); |
| |
| /* quic stats module */ |
| enum { |
| QC_ST_RESET_STREAM_RCVD, |
| |
| QC_ST_CONN_PROTO_ERR, |
| QC_ST_STRM_PROTO_ERR, |
| QC_ST_RESET_STREAM_SENT, |
| |
| QC_ST_OPEN_CONN, |
| QC_ST_OPEN_STREAM, |
| QC_ST_TOTAL_CONN, |
| QC_ST_TOTAL_STREAM, |
| |
| QC_STATS_COUNT /* must be the last member of the enum */ |
| }; |
| |
| static struct name_desc qc_stats[] = { |
| [QC_ST_RESET_STREAM_RCVD] = { .name = "qc_rst_stream_rcvd", |
| .desc = "Total number of received RESET_STREAM frames" }, |
| |
| [QC_ST_CONN_PROTO_ERR] = { .name = "qc_detected_conn_protocol_errors", |
| .desc = "Total number of connection protocol errors" }, |
| [QC_ST_STRM_PROTO_ERR] = { .name = "qc_detected_strm_protocol_errors", |
| .desc = "Total number of stream protocol errors" }, |
| [QC_ST_RESET_STREAM_SENT] = { .name = "qc_rst_stream_resp", |
| .desc = "Total number of RESET_STREAM sent on detected error" }, |
| |
| [QC_ST_OPEN_CONN] = { .name = "qc_open_connections", |
| .desc = "Count of currently open connections" }, |
| [QC_ST_OPEN_STREAM] = { .name = "qc_backend_open_streams", |
| .desc = "Count of currently open streams" }, |
| [QC_ST_TOTAL_CONN] = { .name = "qc_open_connections", |
| .desc = "Total number of connections" }, |
| [QC_ST_TOTAL_STREAM] = { .name = "qc_backend_open_streams", |
| .desc = "Total number of streams" }, |
| }; |
| |
| static struct qc_counters { |
| long long rst_stream_rcvd; /* total number of RESET_STREAM frame received */ |
| |
| long long conn_proto_err; /* total number of protocol errors detected */ |
| long long strm_proto_err; /* total number of protocol errors detected */ |
| long long rst_stream_resp; /* total number of RESET_STREAM frame sent on error */ |
| |
| long long open_conns; /* count of currently open connections */ |
| long long open_streams; /* count of currently open streams */ |
| long long total_conns; /* total number of connections */ |
| long long total_streams; /* total number of streams */ |
| } qc_counters; |
| |
| static void qc_fill_stats(void *data, struct field *stats) |
| { |
| struct qc_counters *counters = data; |
| |
| stats[QC_ST_RESET_STREAM_RCVD] = mkf_u64(FN_COUNTER, counters->rst_stream_rcvd); |
| |
| stats[QC_ST_CONN_PROTO_ERR] = mkf_u64(FN_COUNTER, counters->conn_proto_err); |
| stats[QC_ST_STRM_PROTO_ERR] = mkf_u64(FN_COUNTER, counters->strm_proto_err); |
| stats[QC_ST_RESET_STREAM_SENT] = mkf_u64(FN_COUNTER, counters->rst_stream_resp); |
| |
| stats[QC_ST_OPEN_CONN] = mkf_u64(FN_GAUGE, counters->open_conns); |
| stats[QC_ST_OPEN_STREAM] = mkf_u64(FN_GAUGE, counters->open_streams); |
| stats[QC_ST_TOTAL_CONN] = mkf_u64(FN_COUNTER, counters->total_conns); |
| stats[QC_ST_TOTAL_STREAM] = mkf_u64(FN_COUNTER, counters->total_streams); |
| } |
| |
| static struct stats_module qc_stats_module = { |
| .name = "quic", |
| .fill_stats = qc_fill_stats, |
| .stats = qc_stats, |
| .stats_count = QC_STATS_COUNT, |
| .counters = &qc_counters, |
| .counters_size = sizeof(qc_counters), |
| .domain_flags = MK_STATS_PROXY_DOMAIN(STATS_PX_CAP_FE|STATS_PX_CAP_BE), |
| .clearable = 1, |
| }; |
| |
| INITCALL1(STG_REGISTER, stats_register_module, &qc_stats_module); |
| |
| /* the qcc connection pool */ |
| DECLARE_STATIC_POOL(pool_head_qcc, "qcc", sizeof(struct qcc)); |
| /* the qcs stream pool */ |
| DECLARE_POOL(pool_head_qcs, "qcs", sizeof(struct qcs)); |
| |
| static struct task *qc_timeout_task(struct task *t, void *context, unsigned int state); |
| static int qc_send(struct qcc *qcc); |
| static int qc_recv(struct qcc *qcc); |
| static int qc_process(struct qcc *qcc); |
| static struct task *qc_io_cb(struct task *t, void *ctx, unsigned int state); |
| static inline struct qcs *qcc_st_by_id(struct qcc *qcc, int id); |
| static struct task *qc_deferred_shut(struct task *t, void *ctx, unsigned int state); |
| static struct qcs *qcc_bck_stream_new(struct qcc *qcc, int dir, |
| struct conn_stream *cs, struct session *sess); |
| static void qcs_alert(struct qcs *qcs); |
| |
| /* returns a qcc state as an abbreviated 3-letter string, or "???" if unknown */ |
| static inline const char *qcc_st_to_str(enum qc_cs st) |
| { |
| switch (st) { |
| case QC_CS_NOERR: return "NER"; |
| default: return "???"; |
| } |
| } |
| |
| /* marks an error on the connection */ |
| void qc_error(struct qcc *qcc, int err) |
| { |
| TRACE_POINT(QC_EV_QCC_ERR, qcc->conn, 0, 0, (void *)(long)(err)); |
| qcc->errcode = err; |
| qcc->st0 = QC_CS_ERROR; |
| } |
| |
| static inline const char *qcs_rx_st_to_str(enum qcs_rx_st st) |
| { |
| switch (st) { |
| case QC_RX_SS_IDLE: return "IDL"; |
| case QC_RX_SS_RECV: return "RCV"; |
| case QC_RX_SS_SIZE_KNOWN: return "SKNWN"; |
| case QC_RX_SS_DATA_RECVD: return "DATARCVD"; |
| case QC_RX_SS_DATA_READ : return "DATAREAD"; |
| case QC_RX_SS_RST_RECVD: return "RSTRCVD"; |
| case QC_RX_SS_RST_READ: return "RSTREAD"; |
| default: return "???"; |
| } |
| } |
| |
| static inline const char *qcs_tx_st_to_str(enum qcs_tx_st st) |
| { |
| switch (st) { |
| case QC_TX_SS_IDLE: return "IDL"; |
| case QC_TX_SS_READY: return "READY"; |
| case QC_TX_SS_SEND: return "SEND"; |
| case QC_TX_SS_DATA_SENT: return "DATASENT"; |
| case QC_TX_SS_DATA_RECVD: return "DATARCVD"; |
| case QC_TX_SS_RST_SENT: return "RSTSENT"; |
| case QC_TX_SS_RST_RECVD: return "RSTRCVD"; |
| default: return "???"; |
| } |
| } |
| |
| /* the QUIC traces always expect that arg1, if non-null, is of type connection |
| * (from which we can derive qcc), that arg2, if non-null, is of type qcs. |
| */ |
| static void qc_trace(enum trace_level level, uint64_t mask, const struct trace_source *src, |
| const struct ist where, const struct ist func, |
| const void *a1, const void *a2, const void *a3, const void *a4) |
| { |
| const struct connection *conn = a1; |
| const struct qcc *qcc = conn ? conn->ctx : NULL; |
| const struct qcs *qcs = a2; |
| |
| if (!qcc) |
| return; |
| |
| if (src->verbosity > QC_VERB_CLEAN) { |
| chunk_appendf(&trace_buf, " : qcc=%p(%c,%s)", |
| qcc, conn_is_back(conn) ? 'B' : 'F', qcc_st_to_str(qcc->st0)); |
| if (qcs) { |
| chunk_appendf(&trace_buf, " qcs=%p(rx.%s,tx.%s)", |
| qcs, qcs_rx_st_to_str(qcs->rx.st), qcs_tx_st_to_str(qcs->tx.st)); |
| } |
| } |
| } |
| |
| |
| /* Detect a pending read0 for a QUIC connection. It happens if a read0 is pending |
| * on the connection AND if there is no more data in the demux buffer. The |
| * function returns 1 to report a read0 or 0 otherwise. |
| */ |
| __maybe_unused |
| static int qcc_read0_pending(struct qcc *qcc) |
| { |
| if (conn_xprt_read0_pending(qcc->conn) && !qcc->rx.inmux) |
| return 1; |
| return 0; |
| } |
| |
| static __inline int |
| qcc_is_dead(const struct qcc *qcc) |
| { |
| #if 0 |
| if (eb_is_empty(&qcc->streams_by_id) && /* don't close if streams exist */ |
| ((qcc->conn->flags & CO_FL_ERROR) || /* errors close immediately */ |
| (qcc->st0 >= QC_CS_ERROR && !qcc->task) || /* a timeout stroke earlier */ |
| (!(qcc->conn->owner)) || /* Nobody's left to take care of the connection, drop it now */ |
| (!br_data(qcc->mbuf) && /* mux buffer empty, also process clean events below */ |
| conn_xprt_read0_pending(qcc->conn)))) |
| return 1; |
| #endif |
| if (!qcc->strms[QCS_CLT_BIDI].nb_streams) |
| return 1; |
| |
| return 0; |
| } |
| |
| /*****************************************************/ |
| /* functions below are for dynamic buffer management */ |
| /*****************************************************/ |
| |
| /* indicates whether or not the we may call the qc_recv() function to attempt |
| * to receive data into the buffer and/or demux pending data. The condition is |
| * a bit complex due to some API limits for now. The rules are the following : |
| * - if an error or a shutdown was detected on the connection and the buffer |
| * is empty, we must not attempt to receive |
| * - if the demux buf failed to be allocated, we must not try to receive and |
| * we know there is nothing pending |
| * - if no flag indicates a blocking condition, we may attempt to receive, |
| * regardless of whether the demux buffer is full or not, so that only |
| * de demux part decides whether or not to block. This is needed because |
| * the connection API indeed prevents us from re-enabling receipt that is |
| * already enabled in a polled state, so we must always immediately stop |
| * as soon as the demux can't proceed so as never to hit an end of read |
| * with data pending in the buffers. |
| * - otherwise must may not attempt |
| */ |
| static inline int qc_recv_allowed(const struct qcc *qcc) |
| { |
| if (qcc->rx.inmux == 0 && |
| (qcc->st0 >= QC_CS_ERROR || |
| qcc->conn->flags & CO_FL_ERROR || |
| conn_xprt_read0_pending(qcc->conn))) |
| return 0; |
| |
| if (!(qcc->flags & QC_CF_DEM_BLOCK_ANY)) |
| return 1; |
| |
| return 0; |
| } |
| |
| /* restarts reading on the connection if it was not enabled */ |
| static inline void qcc_restart_reading(const struct qcc *qcc, int consider_buffer) |
| { |
| if (!qc_recv_allowed(qcc)) |
| return; |
| |
| if ((!consider_buffer || !qcc->rx.inmux) |
| && (qcc->wait_event.events & SUB_RETRY_RECV)) |
| return; |
| |
| tasklet_wakeup(qcc->wait_event.tasklet); |
| } |
| |
| static int qc_buf_available(void *target) |
| { |
| struct qcs *qcs = target; |
| if (!b_alloc(&qcs->tx.buf)) |
| return 0; |
| |
| qcc_restart_reading(qcs->qcc, 1); |
| return 1; |
| } |
| |
| struct buffer *qc_get_buf(struct qcs *qcs, struct buffer *bptr) |
| { |
| struct buffer *buf = NULL; |
| struct qcc *qcc = qcs->qcc; |
| |
| if (likely(!LIST_INLIST(&qcc->buf_wait.list)) && |
| unlikely((buf = b_alloc(bptr)) == NULL)) { |
| qcc->buf_wait.target = qcs; |
| qcc->buf_wait.wakeup_cb = qc_buf_available; |
| LIST_APPEND(&th_ctx->buffer_wq, &qcc->buf_wait.list); |
| } |
| |
| return buf; |
| } |
| |
| /* returns the number of streams in use on a connection to figure if it's |
| * idle or not. We check nb_cs and not nb_streams as the caller will want |
| * to know if it was the last one after a detach(). |
| */ |
| static int qc_used_streams(struct connection *conn) |
| { |
| struct qcc *qcc = conn->ctx; |
| |
| return qcc->nb_cs; |
| } |
| |
| /* returns the number of concurrent streams available on the connection with <dir> |
| * as direction |
| */ |
| static int qc_avail_streams(struct connection *conn, enum qcs_dir dir) |
| { |
| struct qcc *qcc = conn->ctx; |
| enum qcs_type qcs_type; |
| |
| if (qcc->st0 >= QC_CS_ERROR) |
| return 0; |
| |
| qcs_type = qcs_type_from_dir(qcc, dir); |
| |
| return qcc->strms[qcs_type].max_streams - qcc->strms[qcs_type].nb_streams; |
| } |
| |
| |
| /* returns the number of concurrent bidirectional streams available on the |
| * connection. |
| */ |
| static int qc_avail_streams_bidi(struct connection *conn) |
| { |
| return qc_avail_streams(conn, QCS_BIDI); |
| } |
| |
| /* returns the number of concurrent unidirectional streams available on the |
| * connection. |
| */ |
| static int qc_avail_streams_uni(struct connection *conn) |
| { |
| return qc_avail_streams(conn, QCS_UNI); |
| } |
| |
| /*****************************************************************/ |
| /* functions below are dedicated to the mux setup and management */ |
| /*****************************************************************/ |
| |
| /* Update the mux transport parameter after having received remote transpot parameters */ |
| void quic_mux_transport_params_update(struct qcc *qcc) |
| { |
| if (objt_listener(qcc->conn->target)) { |
| struct quic_transport_params *clt_params; |
| |
| /* Client parameters, params used to TX. */ |
| clt_params = &qcc->conn->qc->tx.params; |
| |
| qcc->tx.max_data = clt_params->initial_max_data; |
| /* Client initiated streams must respect the server flow control. */ |
| qcc->strms[QCS_CLT_BIDI].rx.max_data = clt_params->initial_max_stream_data_bidi_local; |
| qcc->strms[QCS_CLT_UNI].rx.max_data = clt_params->initial_max_stream_data_uni; |
| |
| /* Server initiated streams must respect the server flow control. */ |
| qcc->strms[QCS_SRV_BIDI].max_streams = clt_params->initial_max_streams_bidi; |
| qcc->strms[QCS_SRV_BIDI].tx.max_data = clt_params->initial_max_stream_data_bidi_remote; |
| |
| qcc->strms[QCS_SRV_UNI].max_streams = clt_params->initial_max_streams_uni; |
| qcc->strms[QCS_SRV_UNI].tx.max_data = clt_params->initial_max_stream_data_uni; |
| } |
| else { |
| struct quic_transport_params *srv_params; |
| |
| /* server parameters, TX params. */ |
| srv_params = &qcc->conn->qc->tx.params; |
| |
| qcc->tx.max_data = srv_params->initial_max_data; |
| /* Client initiated streams must respect the server flow control. */ |
| qcc->strms[QCS_CLT_BIDI].max_streams = srv_params->initial_max_streams_bidi; |
| qcc->strms[QCS_CLT_BIDI].tx.max_data = srv_params->initial_max_stream_data_bidi_remote; |
| |
| qcc->strms[QCS_CLT_UNI].max_streams = srv_params->initial_max_streams_uni; |
| qcc->strms[QCS_CLT_UNI].tx.max_data = srv_params->initial_max_stream_data_uni; |
| |
| /* Server initiated streams must respect the server flow control. */ |
| qcc->strms[QCS_SRV_BIDI].rx.max_data = srv_params->initial_max_stream_data_bidi_local; |
| qcc->strms[QCS_SRV_UNI].rx.max_data = srv_params->initial_max_stream_data_uni; |
| } |
| |
| /* Now that we have all the flow control information, we can finalize the application |
| * context. |
| */ |
| qcc->app_ops->finalize(qcc->ctx); |
| } |
| |
| /* Initialize the mux once it's attached. For outgoing connections, the context |
| * is already initialized before installing the mux, so we detect incoming |
| * connections from the fact that the context is still NULL (even during mux |
| * upgrades). <input> is always used as Input buffer and may contain data. It is |
| * the caller responsibility to not reuse it anymore. Returns < 0 on error. |
| */ |
| static int qc_init(struct connection *conn, struct proxy *prx, |
| struct session *sess, struct buffer *input) |
| { |
| struct qcc *qcc; |
| struct task *t = NULL; |
| void *conn_ctx = conn->ctx; |
| |
| TRACE_ENTER(QC_EV_QCC_NEW); |
| |
| qcc = pool_alloc(pool_head_qcc); |
| if (!qcc) |
| goto fail_no_qcc; |
| |
| if (conn_is_back(conn)) { |
| qcc->flags = QC_CF_IS_BACK; |
| qcc->shut_timeout = qcc->timeout = prx->timeout.server; |
| if (tick_isset(prx->timeout.serverfin)) |
| qcc->shut_timeout = prx->timeout.serverfin; |
| |
| qcc->px_counters = EXTRA_COUNTERS_GET(prx->extra_counters_be, |
| &qc_stats_module); |
| } else { |
| qcc->flags = QC_CF_NONE; |
| qcc->shut_timeout = qcc->timeout = prx->timeout.client; |
| if (tick_isset(prx->timeout.clientfin)) |
| qcc->shut_timeout = prx->timeout.clientfin; |
| |
| qcc->px_counters = EXTRA_COUNTERS_GET(prx->extra_counters_fe, |
| &qc_stats_module); |
| } |
| |
| qcc->proxy = prx; |
| qcc->task = NULL; |
| if (tick_isset(qcc->timeout)) { |
| t = task_new_here(); |
| if (!t) |
| goto fail; |
| |
| qcc->task = t; |
| t->process = qc_timeout_task; |
| t->context = qcc; |
| t->expire = tick_add(now_ms, qcc->timeout); |
| } |
| |
| qcc->subs = NULL; |
| qcc->wait_event.tasklet = tasklet_new(); |
| if (!qcc->wait_event.tasklet) |
| goto fail; |
| |
| qcc->wait_event.tasklet->process = qc_io_cb; |
| qcc->wait_event.tasklet->context = qcc; |
| qcc->wait_event.events = 0; |
| |
| /* Initialize the context. */ |
| qcc->st0 = QC_CS_NOERR; |
| qcc->conn = conn; |
| qcc->conn->qc->qcc = qcc; |
| |
| /* Application layer initialization. */ |
| qcc->app_ops = &h3_ops; |
| if (!qcc->app_ops->init(qcc)) |
| goto fail; |
| |
| /* The transports parameters which control the data sent have been stored |
| * in ->tx.params. The ones which control the received data are stored in |
| * in ->rx.params. |
| */ |
| if (objt_listener(qcc->conn->target)) { |
| struct quic_transport_params *srv_params; |
| |
| /* Server parameters, params used for RX flow control. */ |
| srv_params = &conn->qc->rx.params; |
| |
| qcc->rx.max_data = srv_params->initial_max_data; |
| qcc->tx.max_data = 0; |
| /* Client initiated streams must respect the server flow control. */ |
| qcc->strms[QCS_CLT_BIDI].max_streams = srv_params->initial_max_streams_bidi; |
| qcc->strms[QCS_CLT_BIDI].nb_streams = 0; |
| qcc->strms[QCS_CLT_BIDI].largest_id = -1; |
| qcc->strms[QCS_CLT_BIDI].rx.max_data = 0; |
| qcc->strms[QCS_CLT_BIDI].tx.max_data = srv_params->initial_max_stream_data_bidi_remote; |
| |
| qcc->strms[QCS_CLT_UNI].max_streams = srv_params->initial_max_streams_uni; |
| qcc->strms[QCS_CLT_UNI].nb_streams = 0; |
| qcc->strms[QCS_CLT_UNI].largest_id = -1; |
| qcc->strms[QCS_CLT_UNI].rx.max_data = 0; |
| qcc->strms[QCS_CLT_UNI].tx.max_data = srv_params->initial_max_stream_data_uni; |
| |
| /* Server initiated streams must respect the server flow control. */ |
| qcc->strms[QCS_SRV_BIDI].max_streams = 0; |
| qcc->strms[QCS_SRV_BIDI].nb_streams = 0; |
| qcc->strms[QCS_SRV_BIDI].largest_id = -1; |
| qcc->strms[QCS_SRV_BIDI].rx.max_data = srv_params->initial_max_stream_data_bidi_local; |
| qcc->strms[QCS_SRV_BIDI].tx.max_data = 0; |
| |
| qcc->strms[QCS_SRV_UNI].max_streams = 0; |
| qcc->strms[QCS_SRV_UNI].nb_streams = 0; |
| qcc->strms[QCS_SRV_UNI].largest_id = -1; |
| qcc->strms[QCS_SRV_UNI].rx.max_data = srv_params->initial_max_stream_data_uni; |
| qcc->strms[QCS_SRV_UNI].tx.max_data = 0; |
| } |
| else { |
| struct quic_transport_params *clt_params; |
| |
| /* client parameters, RX params. */ |
| clt_params = &conn->qc->rx.params; |
| |
| qcc->rx.max_data = clt_params->initial_max_data; |
| qcc->tx.max_data = 0; |
| /* Client initiated streams must respect the server flow control. */ |
| qcc->strms[QCS_CLT_BIDI].max_streams = 0; |
| qcc->strms[QCS_CLT_BIDI].nb_streams = 0; |
| qcc->strms[QCS_CLT_BIDI].largest_id = -1; |
| qcc->strms[QCS_CLT_BIDI].rx.max_data = clt_params->initial_max_stream_data_bidi_local; |
| qcc->strms[QCS_CLT_BIDI].tx.max_data = 0; |
| |
| qcc->strms[QCS_CLT_UNI].max_streams = 0; |
| qcc->strms[QCS_CLT_UNI].nb_streams = 0; |
| qcc->strms[QCS_CLT_UNI].largest_id = -1; |
| qcc->strms[QCS_CLT_UNI].rx.max_data = clt_params->initial_max_stream_data_uni; |
| qcc->strms[QCS_CLT_UNI].tx.max_data = 0; |
| |
| /* Server initiated streams must respect the server flow control. */ |
| qcc->strms[QCS_SRV_BIDI].max_streams = clt_params->initial_max_streams_bidi; |
| qcc->strms[QCS_SRV_BIDI].nb_streams = 0; |
| qcc->strms[QCS_SRV_BIDI].largest_id = -1; |
| qcc->strms[QCS_SRV_BIDI].rx.max_data = 0; |
| qcc->strms[QCS_SRV_BIDI].tx.max_data = clt_params->initial_max_stream_data_bidi_remote; |
| |
| qcc->strms[QCS_SRV_UNI].max_streams = clt_params->initial_max_streams_uni; |
| qcc->strms[QCS_SRV_UNI].nb_streams = 0; |
| qcc->strms[QCS_SRV_UNI].largest_id = -1; |
| qcc->strms[QCS_SRV_UNI].rx.max_data = 0; |
| qcc->strms[QCS_SRV_UNI].tx.max_data = clt_params->initial_max_stream_data_uni; |
| |
| } |
| |
| /* Initialize the streams counters. */ |
| qcc->nb_cs = 0; |
| qcc->stream_cnt = 0; |
| |
| qcc->streams_by_id = EB_ROOT_UNIQUE; |
| LIST_INIT(&qcc->send_list); |
| LIST_INIT(&qcc->fctl_list); |
| LIST_INIT(&qcc->blocked_list); |
| LIST_INIT(&qcc->buf_wait.list); |
| MT_LIST_INIT(&qcc->qcs_rxbuf_wlist); |
| |
| HA_ATOMIC_STORE(&conn->ctx, qcc); |
| |
| if (t) |
| task_queue(t); |
| |
| if (qcc->flags & QC_CF_IS_BACK) { |
| /* FIXME: For outgoing connections we need to immediately allocate streams. |
| * This highly depends on the QUIC application needs. |
| */ |
| } |
| |
| HA_ATOMIC_ADD(&qcc->px_counters->open_conns, 1); |
| HA_ATOMIC_ADD(&qcc->px_counters->total_conns, 1); |
| |
| /* prepare to read something */ |
| qcc_restart_reading(qcc, 1); |
| TRACE_LEAVE(QC_EV_QCC_NEW, conn); |
| return 0; |
| |
| fail: |
| task_destroy(t); |
| if (qcc->wait_event.tasklet) |
| tasklet_free(qcc->wait_event.tasklet); |
| pool_free(pool_head_qcc, qcc); |
| fail_no_qcc: |
| conn->ctx = conn_ctx; /* restore saved ctx */ |
| TRACE_DEVEL("leaving in error", QC_EV_QCC_NEW|QC_EV_QCC_END|QC_EV_QCC_ERR); |
| return -1; |
| } |
| |
| /* returns the stream associated with id <id> or NULL if not found */ |
| __maybe_unused |
| static inline struct qcs *qcc_st_by_id(struct qcc *qcc, int id) |
| { |
| struct eb64_node *node; |
| |
| node = eb64_lookup(&qcc->streams_by_id, id); |
| if (!node) |
| return (struct qcs *)qc_closed_stream; |
| |
| return container_of(node, struct qcs, by_id); |
| } |
| |
| /* release function. This one should be called to free all resources allocated |
| * to the mux. |
| */ |
| static void qc_release(struct qcc *qcc) |
| { |
| struct connection *conn = NULL; |
| |
| TRACE_ENTER(QC_EV_QCC_END); |
| |
| if (qcc) { |
| /* The connection must be aattached to this mux to be released */ |
| if (qcc->conn && qcc->conn->ctx == qcc) |
| conn = qcc->conn; |
| |
| TRACE_DEVEL("freeing qcc", QC_EV_QCC_END, conn); |
| |
| if (LIST_INLIST(&qcc->buf_wait.list)) |
| LIST_DELETE(&qcc->buf_wait.list); |
| |
| if (qcc->task) { |
| qcc->task->context = NULL; |
| task_wakeup(qcc->task, TASK_WOKEN_OTHER); |
| qcc->task = NULL; |
| } |
| if (qcc->wait_event.tasklet) |
| tasklet_free(qcc->wait_event.tasklet); |
| if (conn && qcc->wait_event.events != 0) |
| conn->xprt->unsubscribe(conn, conn->xprt_ctx, qcc->wait_event.events, |
| &qcc->wait_event); |
| |
| HA_ATOMIC_SUB(&qcc->px_counters->open_conns, 1); |
| |
| pool_free(pool_head_qcc, qcc); |
| } |
| |
| if (conn) { |
| conn->mux = NULL; |
| conn->ctx = NULL; |
| TRACE_DEVEL("freeing conn", QC_EV_QCC_END, conn); |
| |
| conn_stop_tracking(conn); |
| conn_full_close(conn); |
| if (conn->destroy_cb) |
| conn->destroy_cb(conn); |
| conn_free(conn); |
| } |
| |
| TRACE_LEAVE(QC_EV_QCC_END); |
| } |
| |
| |
| /******************************************************/ |
| /* functions below are for the QUIC protocol processing */ |
| /******************************************************/ |
| |
| /* attempt to notify the data layer of recv availability */ |
| __maybe_unused |
| static void qcs_notify_recv(struct qcs *qcs) |
| { |
| if (qcs->subs && qcs->subs->events & SUB_RETRY_RECV) { |
| TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn, qcs); |
| tasklet_wakeup(qcs->subs->tasklet); |
| qcs->subs->events &= ~SUB_RETRY_RECV; |
| if (!qcs->subs->events) |
| qcs->subs = NULL; |
| } |
| } |
| |
| /* attempt to notify the data layer of send availability */ |
| __maybe_unused |
| static void qcs_notify_send(struct qcs *qcs) |
| { |
| if (qcs->subs && qcs->subs->events & SUB_RETRY_SEND) { |
| TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn, qcs); |
| qcs->flags |= QC_SF_NOTIFIED; |
| tasklet_wakeup(qcs->subs->tasklet); |
| qcs->subs->events &= ~SUB_RETRY_SEND; |
| if (!qcs->subs->events) |
| qcs->subs = NULL; |
| } |
| else if (qcs->flags & (QC_SF_WANT_SHUTR | QC_SF_WANT_SHUTW)) { |
| TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn, qcs); |
| tasklet_wakeup(qcs->shut_tl); |
| } |
| } |
| |
| /* alerts the data layer, trying to wake it up by all means, following |
| * this sequence : |
| * - if the qcs' data layer is subscribed to recv, then it's woken up for recv |
| * - if its subscribed to send, then it's woken up for send |
| * - if it was subscribed to neither, its ->wake() callback is called |
| * It is safe to call this function with a closed stream which doesn't have a |
| * conn_stream anymore. |
| */ |
| __maybe_unused |
| static void qcs_alert(struct qcs *qcs) |
| { |
| TRACE_ENTER(QC_EV_QCS_WAKE, qcs->qcc->conn, qcs); |
| |
| if (qcs->subs || |
| (qcs->flags & (QC_SF_WANT_SHUTR | QC_SF_WANT_SHUTW))) { |
| qcs_notify_recv(qcs); |
| qcs_notify_send(qcs); |
| } |
| else if (qcs->cs && qcs->cs->data_cb->wake != NULL) { |
| TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn, qcs); |
| qcs->cs->data_cb->wake(qcs->cs); |
| } |
| |
| TRACE_LEAVE(QC_EV_QCS_WAKE, qcs->qcc->conn, qcs); |
| } |
| |
| /* marks stream <qcs> as CLOSED and decrement the number of active streams for |
| * its connection if the stream was not yet closed. Please use this exclusively |
| * before closing a stream to ensure stream count is well maintained. |
| */ |
| static inline void qcs_close(struct qcs *qcs) |
| { |
| TRACE_ENTER(QC_EV_QCS_END, qcs->qcc->conn, qcs); |
| /* XXX TO DO XXX */ |
| TRACE_LEAVE(QC_EV_QCS_END, qcs->qcc->conn, qcs); |
| } |
| |
| /* detaches an QUIC stream from its QCC and releases it to the QCS pool. */ |
| /* qcs_destroy should only ever be called by the thread that owns the stream, |
| * that means that a tasklet should be used if we want to destroy the qcs |
| * from another thread |
| */ |
| static void qcs_destroy(struct qcs *qcs) |
| { |
| struct connection *conn = qcs->qcc->conn; |
| |
| TRACE_ENTER(QC_EV_QCS_END, conn, qcs); |
| |
| qcs_close(qcs); |
| eb64_delete(&qcs->by_id); |
| if (b_size(&qcs->rx.buf)) { |
| b_free(&qcs->rx.buf); |
| offer_buffers(NULL, 1); |
| } |
| |
| b_free(&qcs->tx.buf); |
| b_free(&qcs->tx.xprt_buf); |
| |
| if (qcs->subs) |
| qcs->subs->events = 0; |
| |
| /* There's no need to explicitly call unsubscribe here, the only |
| * reference left would be in the qcc send_list/fctl_list, and if |
| * we're in it, we're getting out anyway |
| */ |
| LIST_DEL_INIT(&qcs->list); |
| --qcs->qcc->strms[qcs_id_type(qcs->id)].nb_streams; |
| |
| /* ditto, calling tasklet_free() here should be ok */ |
| tasklet_free(qcs->shut_tl); |
| pool_free(pool_head_qcs, qcs); |
| |
| TRACE_LEAVE(QC_EV_QCS_END, conn); |
| } |
| |
| /* allocates a new bidirection stream <id> for connection <qcc> and adds it into qcc's |
| * stream tree. In case of error, nothing is added and NULL is returned. The |
| * causes of errors can be any failed memory allocation. The caller is |
| * responsible for checking if the connection may support an extra stream |
| * prior to calling this function. |
| */ |
| struct qcs *bidi_qcs_new(struct qcc *qcc, uint64_t id) |
| { |
| struct qcs *qcs; |
| enum qcs_type qcs_type; |
| |
| TRACE_ENTER(QC_EV_QCS_NEW, qcc->conn); |
| |
| qcs = pool_alloc(pool_head_qcs); |
| if (!qcs) |
| goto out; |
| |
| qcs->shut_tl = tasklet_new(); |
| if (!qcs->shut_tl) { |
| pool_free(pool_head_qcs, qcs); |
| goto out; |
| } |
| |
| qcs_type = qcs_id_type(id); |
| qcs->qcc = qcc; |
| qcs->cs = NULL; |
| qcs->id = qcs->by_id.key = id; |
| qcs->flags = QC_SF_NONE; |
| |
| qcs->rx.buf = BUF_NULL; |
| qcs->rx.st = QC_RX_SS_IDLE; |
| qcs->rx.bytes = qcs->rx.offset = 0; |
| qcs->rx.max_data = qcc->strms[qcs_type].rx.max_data; |
| qcs->rx.buf = BUF_NULL; |
| qcs->rx.frms = EB_ROOT_UNIQUE; |
| |
| qcs->tx.st = QC_TX_SS_IDLE; |
| qcs->tx.bytes = qcs->tx.offset = qcs->tx.ack_offset = 0; |
| qcs->tx.acked_frms = EB_ROOT_UNIQUE; |
| qcs->tx.max_data = qcc->strms[qcs_type].tx.max_data; |
| qcs->tx.buf = BUF_NULL; |
| qcs->tx.xprt_buf = BUF_NULL; |
| |
| eb64_insert(&qcc->streams_by_id, &qcs->by_id); |
| qcc->strms[qcs_type].nb_streams++; |
| qcc->stream_cnt++; |
| qcs->subs = NULL; |
| LIST_INIT(&qcs->list); |
| qcs->shut_tl->process = qc_deferred_shut; |
| qcs->shut_tl->context = qcs; |
| |
| HA_ATOMIC_ADD(&qcc->px_counters->open_streams, 1); |
| HA_ATOMIC_ADD(&qcc->px_counters->total_streams, 1); |
| |
| TRACE_LEAVE(QC_EV_QCS_NEW, qcc->conn, qcs); |
| return qcs; |
| |
| out: |
| TRACE_DEVEL("leaving in error", QC_EV_QCS_ERR|QC_EV_QCS_END, qcc->conn); |
| return NULL; |
| } |
| |
| /* Release <qcs> outgoing uni-stream */ |
| void qcs_release(struct qcs *qcs) |
| { |
| eb64_delete(&qcs->by_id); |
| pool_free(pool_head_qcs, qcs); |
| } |
| |
| /* Allocates a locally initiated unidirectional stream. */ |
| struct qcs *luqs_new(struct qcc *qcc) |
| { |
| struct qcs *qcs; |
| uint64_t next_id; |
| enum qcs_type qcs_type; |
| |
| TRACE_ENTER(QC_EV_QCS_NEW, qcc->conn); |
| |
| qcs = NULL; |
| /* QCS_ID_DIR_BIT bit is set for unidirectional stream. */ |
| if (objt_listener(qcc->conn->target)) |
| qcs_type = QCS_ID_SRV_INTIATOR_BIT | QCS_ID_DIR_BIT; |
| else |
| qcs_type = QCS_ID_DIR_BIT; |
| |
| next_id = qcs_next_id(qcc, qcs_type); |
| if (next_id == (uint64_t)-1) { |
| TRACE_PROTO("No more stream available", QC_EV_QCS_NEW, qcc->conn); |
| goto out; |
| } |
| |
| qcs = pool_alloc(pool_head_qcs); |
| if (!qcs) |
| goto out; |
| |
| qcs->qcc = qcc; |
| qcs->cs = NULL; |
| qcs->id = qcs->by_id.key = next_id; |
| qcs->flags = QC_SF_NONE; |
| |
| qcs->tx.st = QC_TX_SS_IDLE; |
| qcs->tx.max_data = qcc->strms[qcs_type].tx.max_data; |
| qcs->tx.offset = qcs->tx.bytes = qcs->tx.ack_offset = 0; |
| qcs->tx.acked_frms = EB_ROOT_UNIQUE; |
| qcs->tx.buf = BUF_NULL; |
| qcs->tx.xprt_buf = BUF_NULL; |
| |
| qcs->subs = NULL; |
| LIST_INIT(&qcs->list); |
| eb64_insert(&qcc->streams_by_id, &qcs->by_id); |
| |
| TRACE_LEAVE(QC_EV_QCS_NEW, qcc->conn); |
| return qcs; |
| |
| out: |
| if (qcs) |
| pool_free(pool_head_qcs, qcs); |
| TRACE_DEVEL("leaving in error", QC_EV_QCS_ERR|QC_EV_QCS_END, qcc->conn); |
| return NULL; |
| } |
| |
| /* Allocates a remotely initiated unidirectional stream. */ |
| struct qcs *ruqs_new(struct qcc *qcc, uint64_t id) |
| { |
| struct qcs *qcs; |
| enum qcs_type qcs_type; |
| |
| TRACE_ENTER(QC_EV_QCS_NEW, qcc->conn); |
| qcs = pool_alloc(pool_head_qcs); |
| if (!qcs) |
| goto out; |
| |
| qcs_type = qcs_id_type(id); |
| |
| qcs->qcc = qcc; |
| qcs->cs = NULL; |
| |
| qcs->qcc = qcc; |
| qcs->id = qcs->by_id.key = id; |
| qcs->flags = QC_SF_NONE; |
| |
| qcs->rx.st = QC_RX_SS_IDLE; |
| qcs->rx.max_data = qcc->strms[qcs_type].rx.max_data; |
| qcs->rx.offset = qcs->rx.bytes = 0; |
| qcs->rx.buf = BUF_NULL; |
| qcs->rx.frms = EB_ROOT_UNIQUE; |
| qcs->tx.buf = BUF_NULL; |
| qcs->tx.xprt_buf = BUF_NULL; |
| |
| qcs->subs = NULL; |
| LIST_INIT(&qcs->list); |
| eb64_insert(&qcc->streams_by_id, &qcs->by_id); |
| |
| TRACE_LEAVE(QC_EV_QCS_NEW, qcc->conn); |
| return qcs; |
| |
| out: |
| TRACE_DEVEL("leaving in error", QC_EV_QCS_ERR|QC_EV_QCS_END, qcc->conn); |
| return NULL; |
| } |
| |
| /* attempt to notify the data layer of recv availability */ |
| void ruqs_notify_recv(struct qcs *qcs) |
| { |
| if (qcs->subs && qcs->subs->events & SUB_RETRY_RECV) { |
| TRACE_POINT(QC_EV_STRM_WAKE, qcs->qcc->conn); |
| tasklet_wakeup(qcs->subs->tasklet); |
| qcs->subs->events &= ~SUB_RETRY_RECV; |
| if (!qcs->subs->events) |
| qcs->subs = NULL; |
| } |
| } |
| |
| /* Allocates a new stream associated to conn_stream <cs> on the qcc connection |
| * with dir as direction and returns it, or NULL in case of memory allocation |
| * error or if the highest possible stream ID was reached. |
| */ |
| static struct qcs *qcc_bck_stream_new(struct qcc *qcc, int dir, |
| struct conn_stream *cs, struct session *sess) |
| { |
| struct qcs *qcs = NULL; |
| enum qcs_type qcs_type; |
| |
| TRACE_ENTER(QC_EV_QCS_NEW, qcc->conn); |
| |
| qcs_type = qcs_type_from_dir(qcc, dir); |
| if (qcc->strms[qcs_type].largest_id + 1 >= qcc->strms[qcs_type].max_streams) |
| goto out; |
| |
| /* Defer choosing the ID until we send the first message to create the stream */ |
| qcs = bidi_qcs_new(qcc, qcc->strms[qcs_type].largest_id + 1); |
| if (!qcs) |
| goto out; |
| |
| qcs->cs = cs; |
| qcs->sess = sess; |
| cs->ctx = qcs; |
| qcc->nb_cs++; |
| |
| out: |
| if (likely(qcs)) |
| TRACE_LEAVE(QC_EV_QCS_NEW, qcc->conn, qcs); |
| else |
| TRACE_LEAVE(QC_EV_QCS_NEW|QC_EV_QCS_ERR|QC_EV_QCS_END, qcc->conn, qcs); |
| return qcs; |
| } |
| |
| /* Allocates a new bidirectional stream associated to conn_stream <cs> on the <qcc> connection |
| * and returns it, or NULL in case of memory allocation error or if the highest |
| * possible stream ID was reached. |
| */ |
| __maybe_unused |
| static struct qcs *qcc_bck_stream_new_bidi(struct qcc *qcc, |
| struct conn_stream *cs, struct session *sess) |
| { |
| return qcc_bck_stream_new(qcc, QCS_BIDI, cs, sess); |
| } |
| |
| /* Allocates a new unidirectional stream associated to conn_stream <cs> on the <qcc> connection |
| * and returns it, or NULL in case of memory allocation error or if the highest |
| * possible stream ID was reached. |
| */ |
| __maybe_unused |
| static struct qcs *qcc_bck_stream_new_uni(struct qcc *qcc, |
| struct conn_stream *cs, struct session *sess) |
| { |
| return qcc_bck_stream_new(qcc, QCS_UNI, cs, sess); |
| } |
| |
| |
| /* wake a specific stream and assign its conn_stream some CS_FL_* flags among |
| * CS_FL_ERR_PENDING and CS_FL_ERROR if needed. The stream's state |
| * is automatically updated accordingly. If the stream is orphaned, it is |
| * destroyed. |
| */ |
| static void qcs_wake_one_stream(struct qcs *qcs) |
| { |
| struct qcc *qcc = qcs->qcc; |
| |
| TRACE_ENTER(QC_EV_QCS_WAKE, qcc->conn, qcs); |
| if (!qcs->cs) { |
| /* this stream was already orphaned */ |
| qcs_destroy(qcs); |
| TRACE_DEVEL("leaving with no qcs", QC_EV_QCS_WAKE, qcc->conn); |
| return; |
| } |
| /* XXX TO DO XXX */ |
| TRACE_LEAVE(QC_EV_QCS_WAKE, qcc->conn); |
| } |
| |
| /* wake the streams attached to the connection, whose id is greater than <last> |
| * or unassigned. |
| */ |
| static void qc_wake_some_streams(struct qcc *qcc, int last) |
| { |
| struct eb64_node *node; |
| struct qcs *qcs; |
| |
| TRACE_ENTER(QC_EV_QCS_WAKE, qcc->conn); |
| |
| /* Wake all streams with ID > last */ |
| node = eb64_lookup_ge(&qcc->streams_by_id, last + 1); |
| while (node) { |
| qcs = container_of(node, struct qcs, by_id); |
| node = eb64_next(node); |
| qcs_wake_one_stream(qcs); |
| } |
| |
| /* Wake all streams with unassigned ID (ID == 0) */ |
| node = eb64_lookup(&qcc->streams_by_id, 0); |
| while (node) { |
| qcs = container_of(node, struct qcs, by_id); |
| if (qcs->id > 0) |
| break; |
| node = eb64_next(node); |
| qcs_wake_one_stream(qcs); |
| } |
| |
| TRACE_LEAVE(QC_EV_QCS_WAKE, qcc->conn); |
| } |
| |
| /* Wake up all blocked streams whose window size has become positive after the |
| * mux's initial window was adjusted. This should be done after having processed |
| * SETTINGS frames which have updated the mux's initial window size. |
| */ |
| __maybe_unused |
| static void qcc_unblock_sfctl(struct qcc *qcc) |
| { |
| TRACE_ENTER(QC_EV_QCC_WAKE, qcc->conn); |
| /* XXX TO DO XXX */ |
| TRACE_LEAVE(QC_EV_QCC_WAKE, qcc->conn); |
| } |
| |
| /* process Rx frames to be demultiplexed */ |
| __maybe_unused |
| static void qc_process_demux(struct qcc *qcc) |
| { |
| TRACE_ENTER(QC_EV_QCC_WAKE, qcc->conn); |
| /* XXX TO DO XXX */ |
| TRACE_LEAVE(QC_EV_QCC_WAKE, qcc->conn); |
| } |
| |
| /* resume each qcs eligible for sending in list head <head> */ |
| __maybe_unused |
| static void qc_resume_each_sending_qcs(struct qcc *qcc, struct list *head) |
| { |
| struct qcs *qcs, *qcs_back; |
| |
| TRACE_ENTER(QC_EV_QCC_SEND|QC_EV_QCS_WAKE, qcc->conn); |
| |
| list_for_each_entry_safe(qcs, qcs_back, head, list) { |
| if (qcc_wnd(qcc) <= 0 || |
| qcc->flags & QC_CF_MUX_BLOCK_ANY || |
| qcc->st0 >= QC_CS_ERROR) |
| break; |
| |
| qcs->flags &= ~QC_SF_BLK_ANY; |
| |
| if (qcs->flags & QC_SF_NOTIFIED) |
| continue; |
| |
| /* If the sender changed his mind and unsubscribed, let's just |
| * remove the stream from the send_list. |
| */ |
| if (!(qcs->flags & (QC_SF_WANT_SHUTR|QC_SF_WANT_SHUTW)) && |
| (!qcs->subs || !(qcs->subs->events & SUB_RETRY_SEND))) { |
| LIST_DEL_INIT(&qcs->list); |
| continue; |
| } |
| |
| if (qcs->subs && qcs->subs->events & SUB_RETRY_SEND) { |
| qcs->flags |= QC_SF_NOTIFIED; |
| tasklet_wakeup(qcs->subs->tasklet); |
| qcs->subs->events &= ~SUB_RETRY_SEND; |
| if (!qcs->subs->events) |
| qcs->subs = NULL; |
| } |
| else if (qcs->flags & (QC_SF_WANT_SHUTR|QC_SF_WANT_SHUTW)) { |
| tasklet_wakeup(qcs->shut_tl); |
| } |
| } |
| |
| TRACE_LEAVE(QC_EV_QCC_SEND|QC_EV_QCS_WAKE, qcc->conn); |
| } |
| |
| /* process Tx frames from streams to be multiplexed. Returns > 0 if it reached |
| * the end. |
| */ |
| __maybe_unused |
| static int qc_process_mux(struct qcc *qcc) |
| { |
| TRACE_ENTER(QC_EV_QCC_WAKE, qcc->conn); |
| |
| /* First we always process the flow control list because the streams |
| * waiting there were already elected for immediate emission but were |
| * blocked just on this. |
| */ |
| qc_resume_each_sending_qcs(qcc, &qcc->fctl_list); |
| qc_resume_each_sending_qcs(qcc, &qcc->send_list); |
| |
| TRACE_LEAVE(QC_EV_QCC_WAKE, qcc->conn); |
| return 1; |
| } |
| |
| |
| /* Attempt to read data, and subscribe if none available. |
| * The function returns 1 if data has been received, otherwise zero. |
| */ |
| __maybe_unused |
| static int qc_recv(struct qcc *qcc) |
| { |
| TRACE_ENTER(QC_EV_QCC_RECV, qcc->conn); |
| /* XXX TO DO XXX */ |
| TRACE_LEAVE(QC_EV_QCC_RECV, qcc->conn); |
| return 0; |
| } |
| |
| static int qcs_push_frame(struct qcs *qcs, struct buffer *payload, int fin, uint64_t offset) |
| { |
| struct quic_frame *frm; |
| struct buffer *buf = &qcs->tx.xprt_buf; |
| struct quic_enc_level *qel = &qcs->qcc->conn->qc->els[QUIC_TLS_ENC_LEVEL_APP]; |
| int total = 0, to_xfer; |
| |
| qc_get_buf(qcs, buf); |
| to_xfer = QUIC_MIN(b_data(payload), b_room(buf)); |
| if (!to_xfer) |
| goto out; |
| |
| frm = pool_zalloc(pool_head_quic_frame); |
| if (!frm) |
| goto err; |
| |
| total = b_force_xfer(buf, payload, to_xfer); |
| fin = fin && !b_data(payload); |
| frm->type = QUIC_FT_STREAM_8; |
| if (fin) |
| frm->type |= QUIC_STREAM_FRAME_TYPE_FIN_BIT; |
| if (offset) { |
| frm->type |= QUIC_STREAM_FRAME_TYPE_OFF_BIT; |
| frm->stream.offset.key = offset; |
| } |
| frm->stream.qcs = qcs; |
| frm->stream.buf = buf; |
| frm->stream.id = qcs->by_id.key; |
| if (total) { |
| frm->type |= QUIC_STREAM_FRAME_TYPE_LEN_BIT; |
| frm->stream.len = total; |
| } |
| |
| MT_LIST_APPEND(&qel->pktns->tx.frms, &frm->mt_list); |
| out: |
| fprintf(stderr, "%s: total=%d fin=%d offset=%lu\n", __func__, total, fin, offset); |
| return total; |
| |
| err: |
| return -1; |
| } |
| |
| /* Try to send data if possible. |
| * The function returns 1 if data have been sent, otherwise zero. |
| */ |
| static int qc_send(struct qcc *qcc) |
| { |
| struct eb64_node *node; |
| int ret, done, xprt_wake = 0; |
| |
| TRACE_ENTER(QC_EV_QCC_SEND, qcc->conn); |
| ret = done = 0; |
| /* fill as much as we can into the current buffer */ |
| while (((qcc->flags & (QC_CF_MUX_MFULL|QC_CF_MUX_MALLOC)) == 0) && !done) |
| done = qc_process_mux(qcc); |
| |
| /* TODO simple loop through all streams and check if there is frames to |
| * send |
| */ |
| node = eb64_first(&qcc->streams_by_id); |
| while (node) { |
| struct qcs *qcs = container_of(node, struct qcs, by_id); |
| struct buffer *buf = &qcs->tx.buf; |
| if (b_data(buf)) { |
| char fin = 0; |
| |
| /* if FIN is activated, ensure the buffer to |
| * send is the last |
| */ |
| if (qcs->flags & QC_SF_FIN_STREAM) { |
| BUG_ON(b_data(&qcs->tx.buf) < b_data(buf)); |
| fin = (b_data(&qcs->tx.buf) - b_data(buf) == 0); |
| } |
| |
| ret = qcs_push_frame(qcs, buf, fin, qcs->tx.offset); |
| if (ret < 0) |
| ABORT_NOW(); |
| |
| if (ret > 0) { |
| xprt_wake = 1; |
| if (qcs->flags & QC_SF_BLK_MROOM) { |
| qcs->flags &= ~QC_SF_BLK_MROOM; |
| qcs_notify_send(qcs); |
| } |
| } |
| |
| qcs->tx.offset += ret; |
| |
| if (b_data(buf)) { |
| qcc->conn->xprt->subscribe(qcc->conn, qcc->conn->xprt_ctx, |
| SUB_RETRY_SEND, &qcc->wait_event); |
| } |
| } |
| node = eb64_next(node); |
| } |
| |
| if (xprt_wake) |
| tasklet_wakeup(((struct ssl_sock_ctx *)(qcc->conn->xprt_ctx))->wait_event.tasklet); |
| |
| TRACE_LEAVE(QC_EV_QCC_SEND, qcc->conn); |
| return 0; |
| } |
| |
| /* this is the tasklet referenced in qcc->wait_event.tasklet */ |
| static struct task *qc_io_cb(struct task *t, void *ctx, unsigned int status) |
| { |
| struct qcc *qcc = ctx; |
| //int ret = 0; |
| |
| |
| if (!(qcc->wait_event.events & SUB_RETRY_SEND)) |
| //ret = qc_send(qcc); |
| qc_send(qcc); |
| #if 0 |
| if (!(qcc->wait_event.events & SUB_RETRY_RECV)) |
| ret |= qc_recv(qcc); |
| #endif |
| // TODO redefine the proper condition here |
| //if (ret || qcc->rx.inmux) |
| //ret = qc_process(qcc); |
| qc_process(qcc); |
| |
| leave: |
| TRACE_LEAVE(QC_EV_QCC_WAKE); |
| return NULL; |
| } |
| |
| /* callback called on any event by the connection handler. |
| * It applies changes and returns zero, or < 0 if it wants immediate |
| * destruction of the connection (which normally doesn not happen in quic). |
| */ |
| static int qc_process(struct qcc *qcc) |
| { |
| struct connection *conn = qcc->conn; |
| |
| TRACE_ENTER(QC_EV_QCC_WAKE, conn); |
| TRACE_LEAVE(QC_EV_QCC_WAKE, conn); |
| return 0; |
| } |
| |
| /* wake-up function called by the connection layer (mux_ops.wake) */ |
| static int qc_wake(struct connection *conn) |
| { |
| struct qcc *qcc = conn->ctx; |
| int ret; |
| |
| TRACE_ENTER(QC_EV_QCC_WAKE, conn); |
| ret = qc_process(qcc); |
| if (ret >= 0) |
| qc_wake_some_streams(qcc, 0); |
| TRACE_LEAVE(QC_EV_QCC_WAKE); |
| return ret; |
| } |
| |
| /* Connection timeout management. The principle is that if there's no receipt |
| * nor sending for a certain amount of time, the connection is closed. If the |
| * MUX buffer still has lying data or is not allocatable, the connection is |
| * immediately killed. If it's allocatable and empty, we attempt to send a |
| * GOAWAY frame. |
| */ |
| static struct task *qc_timeout_task(struct task *t, void *context, unsigned int state) |
| { |
| TRACE_ENTER(QC_EV_QCC_WAKE); |
| /* XXX TO DO XXX */ |
| TRACE_LEAVE(QC_EV_QCC_WAKE); |
| return NULL; |
| } |
| |
| |
| /*******************************************/ |
| /* functions below are used by the streams */ |
| /*******************************************/ |
| |
| /* |
| * Attach a new stream to a connection |
| * (Used for outgoing connections) |
| */ |
| static struct conn_stream *qc_attach(struct connection *conn, struct session *sess) |
| { |
| struct conn_stream *cs; |
| struct qcs *qcs; |
| struct qcc *qcc = conn->ctx; |
| |
| TRACE_ENTER(QC_EV_QCS_NEW, conn); |
| cs = cs_new(conn, conn->target); |
| if (!cs) { |
| TRACE_DEVEL("leaving on CS allocation failure", QC_EV_QCS_NEW|QC_EV_QCS_ERR, conn); |
| return NULL; |
| } |
| qcs = qcc_bck_stream_new(qcc, QCS_BIDI, cs, sess); |
| if (!qcs) { |
| TRACE_DEVEL("leaving on stream creation failure", QC_EV_QCS_NEW|QC_EV_QCS_ERR, conn); |
| cs_free(cs); |
| return NULL; |
| } |
| TRACE_LEAVE(QC_EV_QCS_NEW, conn, qcs); |
| return cs; |
| } |
| |
| /* Retrieves the first valid conn_stream from this connection, or returns NULL. |
| * We have to scan because we may have some orphan streams. It might be |
| * beneficial to scan backwards from the end to reduce the likeliness to find |
| * orphans. |
| */ |
| static const struct conn_stream *qc_get_first_cs(const struct connection *conn) |
| { |
| struct qcc *qcc = conn->ctx; |
| struct qcs *qcs; |
| struct eb64_node *node; |
| |
| node = eb64_first(&qcc->streams_by_id); |
| while (node) { |
| qcs = container_of(node, struct qcs, by_id); |
| if (qcs->cs) |
| return qcs->cs; |
| node = eb64_next(node); |
| } |
| return NULL; |
| } |
| |
| static int qc_ctl(struct connection *conn, enum mux_ctl_type mux_ctl, void *output) |
| { |
| int ret = 0; |
| struct qcc *qcc = conn->ctx; |
| |
| switch (mux_ctl) { |
| case MUX_STATUS: |
| /* Only consider the mux to be ready if we had no error. */ |
| if (qcc->st0 < QC_CS_ERROR) |
| ret |= MUX_STATUS_READY; |
| return ret; |
| case MUX_EXIT_STATUS: |
| return MUX_ES_UNKNOWN; |
| default: |
| return -1; |
| } |
| } |
| |
| /* |
| * Destroy the mux and the associated connection, if it is no longer used |
| */ |
| static void qc_destroy(void *ctx) |
| { |
| struct qcc *qcc = ctx; |
| |
| TRACE_ENTER(QC_EV_QCC_END, qcc->conn); |
| if (eb_is_empty(&qcc->streams_by_id) || !qcc->conn || qcc->conn->ctx != qcc) |
| qc_release(qcc); |
| TRACE_LEAVE(QC_EV_QCC_END); |
| } |
| |
| /* |
| * Detach the stream from the connection and possibly release the connection. |
| */ |
| static void qc_detach(struct conn_stream *cs) |
| { |
| struct qcs *qcs = cs->ctx; |
| struct qcc *qcc = qcs->qcc; |
| |
| TRACE_ENTER(QC_EV_STRM_END, qcs ? qcs->qcc->conn : NULL, qcs); |
| if (b_data(&qcs->tx.buf) || |
| (b_data(&qcs->tx.xprt_buf) && !(qcc->flags & QC_CF_CC_RECV))) { |
| qcs->flags |= QC_SF_DETACH; |
| goto out; |
| } |
| |
| qcs_destroy(qcs); |
| if (qcc_is_dead(qcc)) { |
| qc_release(qcc); |
| TRACE_LEAVE(QC_EV_STRM_END, NULL); |
| return; |
| } |
| |
| out: |
| TRACE_LEAVE(QC_EV_STRM_END, qcs ? qcs->qcc->conn : NULL); |
| } |
| |
| /* Performs a synchronous or asynchronous shutr(). */ |
| static void qc_do_shutr(struct qcs *qcs) |
| { |
| struct qcc *qcc = qcs->qcc; |
| |
| TRACE_ENTER(QC_EV_STRM_SHUT, qcc->conn, qcs); |
| /* XXX TO DO XXX */ |
| TRACE_LEAVE(QC_EV_STRM_SHUT, qcc->conn, qcs); |
| return; |
| } |
| |
| /* Performs a synchronous or asynchronous shutw(). */ |
| static void qc_do_shutw(struct qcs *qcs) |
| { |
| struct qcc *qcc = qcs->qcc; |
| |
| TRACE_ENTER(QC_EV_STRM_SHUT, qcc->conn, qcs); |
| /* XXX TO DO XXX */ |
| TRACE_LEAVE(QC_EV_STRM_SHUT, qcc->conn, qcs); |
| return; |
| } |
| |
| /* This is the tasklet referenced in qcs->shut_tl, it is used for |
| * deferred shutdowns when the qc_detach() was done but the mux buffer was full |
| * and prevented the last frame from being emitted. |
| */ |
| static struct task *qc_deferred_shut(struct task *t, void *ctx, unsigned int state) |
| { |
| struct qcs *qcs = ctx; |
| struct qcc *qcc = qcs->qcc; |
| |
| TRACE_ENTER(QC_EV_STRM_SHUT, qcc->conn, qcs); |
| |
| if (qcs->flags & QC_SF_NOTIFIED || |
| (b_data(&qcs->tx.buf) || b_data(&qcs->tx.xprt_buf))) { |
| /* some data processing remains to be done first */ |
| goto end; |
| } |
| |
| if (qcs->flags & QC_SF_WANT_SHUTW) |
| qc_do_shutw(qcs); |
| |
| if (qcs->flags & QC_SF_WANT_SHUTR) |
| qc_do_shutr(qcs); |
| |
| if (!(qcs->flags & (QC_SF_WANT_SHUTR|QC_SF_WANT_SHUTW))) { |
| /* We're done trying to send, remove ourself from the send_list */ |
| LIST_DEL_INIT(&qcs->list); |
| |
| if (!qcs->cs) { |
| qcs_destroy(qcs); |
| if (qcc_is_dead(qcc)) |
| qc_release(qcc); |
| } |
| } |
| |
| end: |
| TRACE_LEAVE(QC_EV_STRM_SHUT); |
| return NULL; |
| } |
| |
| /* shutr() called by the conn_stream (mux_ops.shutr) */ |
| static void qc_shutr(struct conn_stream *cs, enum cs_shr_mode mode) |
| { |
| |
| struct qcs *qcs = cs->ctx; |
| |
| TRACE_ENTER(QC_EV_STRM_SHUT, qcs->qcc->conn, qcs); |
| if (cs->flags & CS_FL_KILL_CONN) |
| qcs->flags |= QC_SF_KILL_CONN; |
| |
| if (mode) |
| qc_do_shutr(qcs); |
| |
| TRACE_LEAVE(QC_EV_STRM_SHUT, qcs->qcc->conn, qcs); |
| } |
| |
| /* shutw() called by the conn_stream (mux_ops.shutw) */ |
| static void qc_shutw(struct conn_stream *cs, enum cs_shw_mode mode) |
| { |
| struct qcs *qcs = cs->ctx; |
| |
| TRACE_ENTER(QC_EV_STRM_SHUT, qcs->qcc->conn, qcs); |
| if (cs->flags & CS_FL_KILL_CONN) |
| qcs->flags |= QC_SF_KILL_CONN; |
| |
| qc_do_shutw(qcs); |
| TRACE_LEAVE(QC_EV_STRM_SHUT, qcs->qcc->conn, qcs); |
| } |
| |
| /* Called from the upper layer, to subscribe <es> to events <event_type>. The |
| * event subscriber <es> is not allowed to change from a previous call as long |
| * as at least one event is still subscribed. The <event_type> must only be a |
| * combination of SUB_RETRY_RECV and SUB_RETRY_SEND. It always returns 0. |
| */ |
| static int qc_subscribe(struct conn_stream *cs, int event_type, struct wait_event *es) |
| { |
| struct qcs *qcs = cs->ctx; |
| struct qcc *qcc = qcs->qcc; |
| |
| TRACE_ENTER(QC_EV_STRM_SEND|QC_EV_STRM_RECV, qcc->conn, qcs); |
| |
| BUG_ON(event_type & ~(SUB_RETRY_SEND|SUB_RETRY_RECV)); |
| BUG_ON(qcs->subs && qcs->subs != es); |
| |
| es->events |= event_type; |
| qcs->subs = es; |
| |
| if (event_type & SUB_RETRY_RECV) |
| TRACE_DEVEL("subscribe(recv)", QC_EV_STRM_RECV, qcc->conn, qcs); |
| |
| if (event_type & SUB_RETRY_SEND) { |
| TRACE_DEVEL("subscribe(send)", QC_EV_STRM_SEND, qcc->conn, qcs); |
| if (!(qcs->flags & QC_SF_BLK_SFCTL) && |
| !LIST_INLIST(&qcs->list)) { |
| if (qcs->flags & QC_SF_BLK_MFCTL) |
| LIST_APPEND(&qcc->fctl_list, &qcs->list); |
| else |
| LIST_APPEND(&qcc->send_list, &qcs->list); |
| } |
| } |
| TRACE_LEAVE(QC_EV_STRM_SEND|QC_EV_STRM_RECV, qcc->conn, qcs); |
| return 0; |
| } |
| |
| /* Called from the upper layer, to unsubscribe <es> from events <event_type>. |
| * The <es> pointer is not allowed to differ from the one passed to the |
| * subscribe() call. It always returns zero. |
| */ |
| static int qc_unsubscribe(struct conn_stream *cs, int event_type, struct wait_event *es) |
| { |
| struct qcs *qcs = cs->ctx; |
| |
| TRACE_ENTER(QC_EV_STRM_SEND|QC_EV_STRM_RECV, qcs->qcc->conn, qcs); |
| |
| BUG_ON(event_type & ~(SUB_RETRY_SEND|SUB_RETRY_RECV)); |
| BUG_ON(qcs->subs && qcs->subs != es); |
| |
| es->events &= ~event_type; |
| if (!es->events) |
| qcs->subs = NULL; |
| |
| if (event_type & SUB_RETRY_RECV) |
| TRACE_DEVEL("unsubscribe(recv)", QC_EV_STRM_RECV, qcs->qcc->conn, qcs); |
| |
| if (event_type & SUB_RETRY_SEND) { |
| TRACE_DEVEL("subscribe(send)", QC_EV_STRM_SEND, qcs->qcc->conn, qcs); |
| qcs->flags &= ~QC_SF_NOTIFIED; |
| if (!(qcs->flags & (QC_SF_WANT_SHUTR | QC_SF_WANT_SHUTW))) |
| LIST_DEL_INIT(&qcs->list); |
| } |
| |
| TRACE_LEAVE(QC_EV_STRM_SEND|QC_EV_STRM_RECV, qcs->qcc->conn, qcs); |
| return 0; |
| } |
| |
| |
| /* Called from the upper layer, to subscribe <es> to events <event_type>. The |
| * event subscriber <es> is not allowed to change from a previous call as long |
| * as at least one event is still subscribed. The <event_type> must only be a |
| * SUB_RETRY_RECV. It always returns 0. |
| */ |
| static int ruqs_subscribe(struct qcs *qcs, int event_type, struct wait_event *es) |
| { |
| struct qcc *qcc = qcs->qcc; |
| |
| TRACE_ENTER(QC_EV_STRM_RECV, qcc->conn, qcs); |
| |
| BUG_ON(event_type & ~SUB_RETRY_RECV); |
| BUG_ON(qcs->subs && qcs->subs != es); |
| |
| es->events |= event_type; |
| qcs->subs = es; |
| |
| if (event_type & SUB_RETRY_RECV) |
| TRACE_DEVEL("subscribe(recv)", QC_EV_STRM_RECV, qcc->conn, qcs); |
| |
| TRACE_LEAVE(QC_EV_STRM_RECV, qcc->conn, qcs); |
| return 0; |
| } |
| |
| /* Called from the upper layer, to unsubscribe <es> from events <event_type>. |
| * The <es> pointer is not allowed to differ from the one passed to the |
| * subscribe() call. It always returns zero. |
| */ |
| static int ruqs_unsubscribe(struct qcs *qcs, int event_type, struct wait_event *es) |
| { |
| TRACE_ENTER(QC_EV_STRM_RECV, qcs->qcc->conn, qcs); |
| |
| BUG_ON(event_type & ~SUB_RETRY_RECV); |
| BUG_ON(qcs->subs && qcs->subs != es); |
| |
| es->events &= ~event_type; |
| if (!es->events) |
| qcs->subs = NULL; |
| |
| if (event_type & SUB_RETRY_RECV) |
| TRACE_DEVEL("unsubscribe(recv)", QC_EV_STRM_RECV, qcs->qcc->conn, qcs); |
| |
| TRACE_LEAVE(QC_EV_STRM_RECV, qcs->qcc->conn, qcs); |
| return 0; |
| } |
| |
| /* Called from the upper layer, to subscribe <es> to events <event_type>. The |
| * event subscriber <es> is not allowed to change from a previous call as long |
| * as at least one event is still subscribed. The <event_type> must only be |
| * SUB_RETRY_SEND. It always returns 0. |
| */ |
| static int luqs_subscribe(struct qcs *qcs, int event_type, struct wait_event *es) |
| { |
| struct qcc *qcc = qcs->qcc; |
| |
| TRACE_ENTER(QC_EV_STRM_SEND, qcc->conn, qcs); |
| |
| BUG_ON(event_type & ~SUB_RETRY_SEND); |
| BUG_ON(qcs->subs && qcs->subs != es); |
| |
| es->events |= event_type; |
| qcs->subs = es; |
| |
| if (event_type & SUB_RETRY_SEND) { |
| TRACE_DEVEL("subscribe(send)", QC_EV_STRM_SEND, qcc->conn, qcs); |
| if (!(qcs->flags & QC_SF_BLK_SFCTL) && |
| !LIST_INLIST(&qcs->list)) { |
| if (qcs->flags & QC_SF_BLK_MFCTL) |
| LIST_APPEND(&qcc->fctl_list, &qcs->list); |
| else |
| LIST_APPEND(&qcc->send_list, &qcs->list); |
| } |
| } |
| |
| TRACE_LEAVE(QC_EV_STRM_SEND, qcc->conn, qcs); |
| return 0; |
| } |
| |
| /* Called from the upper layer, to unsubscribe <es> from events <event_type>. |
| * The <es> pointer is not allowed to differ from the one passed to the |
| * subscribe() call. It always returns zero. |
| */ |
| static int luqs_unsubscribe(struct qcs *qcs, int event_type, struct wait_event *es) |
| { |
| struct qcc *qcc = qcs->qcc; |
| |
| TRACE_ENTER(QC_EV_STRM_SEND, qcc->conn, qcs); |
| |
| BUG_ON(event_type & ~SUB_RETRY_SEND); |
| BUG_ON(qcs->subs && qcs->subs != es); |
| |
| es->events &= ~event_type; |
| if (!es->events) |
| qcs->subs = NULL; |
| |
| if (event_type & SUB_RETRY_SEND) { |
| TRACE_DEVEL("subscribe(send)", QC_EV_STRM_SEND, qcc->conn, qcs); |
| qcs->flags &= ~QC_SF_NOTIFIED; |
| if (!(qcs->flags & (QC_SF_WANT_SHUTR | QC_SF_WANT_SHUTW))) |
| LIST_DEL_INIT(&qcs->list); |
| } |
| |
| TRACE_LEAVE(QC_EV_STRM_SEND, qcc->conn, qcs); |
| return 0; |
| } |
| |
| /* Called from the upper layer, to receive data */ |
| static size_t qc_rcv_buf(struct conn_stream *cs, struct buffer *buf, size_t count, int flags) |
| { |
| struct qcs *qcs = cs->ctx; |
| struct qcc *qcc = qcs->qcc; |
| int ret; |
| |
| ret = 0; |
| TRACE_ENTER(QC_EV_STRM_RECV, qcc->conn, qcs); |
| /* XXX TO DO XXX */ |
| TRACE_LEAVE(QC_EV_STRM_RECV, qcc->conn, qcs); |
| return ret; |
| } |
| |
| /* for debugging with CLI's "show fd" command */ |
| static int qc_show_fd(struct buffer *msg, struct connection *conn) |
| { |
| struct qcc *qcc = conn->ctx; |
| struct qcs *qcs = NULL; |
| struct eb64_node *node; |
| int fctl_cnt = 0; |
| int send_cnt = 0; |
| int tree_cnt = 0; |
| int orph_cnt = 0; |
| |
| if (!qcc) |
| return 0; |
| |
| list_for_each_entry(qcs, &qcc->fctl_list, list) |
| fctl_cnt++; |
| |
| list_for_each_entry(qcs, &qcc->send_list, list) |
| send_cnt++; |
| |
| qcs = NULL; |
| node = eb64_first(&qcc->streams_by_id); |
| while (node) { |
| qcs = container_of(node, struct qcs, by_id); |
| tree_cnt++; |
| if (!qcs->cs) |
| orph_cnt++; |
| node = eb64_next(node); |
| } |
| |
| chunk_appendf(msg, " qcc.st0=%s .flg=0x%04x" |
| " clt.nb_streams_bidi=%llu srv.nb_streams_bidi=%llu" |
| " clt.nb_streams_uni=%llu srv.nb_streams_uni=%llu" |
| " .nbcs=%u .fctl_cnt=%d .send_cnt=%d .tree_cnt=%d" |
| " .orph_cnt=%d .sub=%d", |
| qcc_st_to_str(qcc->st0), qcc->flags, |
| (unsigned long long)qcc->strms[QCS_CLT_BIDI].nb_streams, |
| (unsigned long long)qcc->strms[QCS_SRV_BIDI].nb_streams, |
| (unsigned long long)qcc->strms[QCS_CLT_UNI].nb_streams, |
| (unsigned long long)qcc->strms[QCS_SRV_UNI].nb_streams, |
| qcc->nb_cs, fctl_cnt, send_cnt, tree_cnt, orph_cnt, |
| qcc->wait_event.events); |
| |
| if (qcs) { |
| chunk_appendf(msg, " last_qcs=%p .id=%llu rx.st=%s tx.st=%s .flg=0x%04x .rxbuf=%u@%p+%u/%u .cs=%p", |
| qcs, (unsigned long long)qcs->id, |
| qcs_rx_st_to_str(qcs->rx.st), qcs_tx_st_to_str(qcs->tx.st), qcs->flags, |
| (unsigned int)b_data(&qcs->rx.buf), b_orig(&qcs->rx.buf), |
| (unsigned int)b_head_ofs(&qcs->rx.buf), (unsigned int)b_size(&qcs->rx.buf), |
| qcs->cs); |
| if (qcs->cs) |
| chunk_appendf(msg, " .cs.flg=0x%08x .cs.data=%p", |
| qcs->cs->flags, qcs->cs->data); |
| } |
| |
| return 0; |
| } |
| |
| /****************************************/ |
| /* MUX initialization and instantiation */ |
| /***************************************/ |
| |
| /* The mux operations */ |
| static const struct mux_ops qc_ops = { |
| .init = qc_init, |
| .wake = qc_wake, |
| //.snd_buf = qc_snd_buf, |
| .snd_buf = h3_snd_buf, |
| .rcv_buf = qc_rcv_buf, |
| .subscribe = qc_subscribe, |
| .unsubscribe = qc_unsubscribe, |
| .ruqs_subscribe = ruqs_subscribe, |
| .ruqs_unsubscribe = ruqs_unsubscribe, |
| .luqs_subscribe = luqs_subscribe, |
| .luqs_unsubscribe = luqs_unsubscribe, |
| .attach = qc_attach, |
| .get_first_cs = qc_get_first_cs, |
| .detach = qc_detach, |
| .destroy = qc_destroy, |
| .avail_streams_bidi = qc_avail_streams_bidi, |
| .avail_streams_uni = qc_avail_streams_uni, |
| .used_streams = qc_used_streams, |
| .shutr = qc_shutr, |
| .shutw = qc_shutw, |
| .ctl = qc_ctl, |
| .show_fd = qc_show_fd, |
| .flags = MX_FL_CLEAN_ABRT|MX_FL_HTX|MX_FL_HOL_RISK, |
| .name = "QUIC", |
| }; |
| |
| static struct mux_proto_list mux_proto_quic = |
| { .token = IST("quic"), .mode = PROTO_MODE_HTTP, .side = PROTO_SIDE_BOTH, .mux = &qc_ops }; |
| |
| INITCALL1(STG_REGISTER, register_mux_proto, &mux_proto_quic); |
| |