From: Amaury Denoyelle Date: Fri, 3 Dec 2021 10:36:46 +0000 (+0100) Subject: MAJOR: mux-quic: implement a simplified mux version X-Git-Tag: v2.6-dev1~319 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=deed777766cd1248130a41e9138de7bea809df83;p=thirdparty%2Fhaproxy.git MAJOR: mux-quic: implement a simplified mux version Re-implement the QUIC mux. It will reuse the mechanics from the previous mux without all untested/unsupported features. This should ease the maintenance. Note that a lot of features are broken for the moment. They will be re-implemented on the following commits to have a clean commit history. --- diff --git a/include/haproxy/connection-t.h b/include/haproxy/connection-t.h index 09351d61a5..6a14cc090d 100644 --- a/include/haproxy/connection-t.h +++ b/include/haproxy/connection-t.h @@ -427,10 +427,6 @@ struct mux_ops { int (*show_fd)(struct buffer *, struct connection *); /* append some data about connection into chunk for "show fd"; returns non-zero if suspicious */ int (*subscribe)(struct conn_stream *cs, int event_type, struct wait_event *es); /* Subscribe to events, such as "being able to send" */ int (*unsubscribe)(struct conn_stream *cs, int event_type, struct wait_event *es); /* Unsubscribe from events */ - int (*ruqs_subscribe)(struct qcs *qcs, int event_type, struct wait_event *es); /* Subscribe to events, "being able to receive" only */ - int (*ruqs_unsubscribe)(struct qcs *qcs, int event_type, struct wait_event *es); /* Unsubscribe from events */ - int (*luqs_subscribe)(struct qcs *qcs, int event_type, struct wait_event *es); /* Subscribe to events, "being able to send" only */ - int (*luqs_unsubscribe)(struct qcs *qcs, int event_type, struct wait_event *es); /* Unsubscribe from events */ int (*avail_streams)(struct connection *conn); /* Returns the number of streams still available for a connection */ int (*avail_streams_bidi)(struct connection *conn); /* Returns the number of bidirectional streams still available for a connection */ int (*avail_streams_uni)(struct connection *conn); /* Returns the number of unidirectional streams still available for a connection */ diff --git a/include/haproxy/mux_quic-t.h b/include/haproxy/mux_quic-t.h index c05c69c300..2e869c62c5 100644 --- a/include/haproxy/mux_quic-t.h +++ b/include/haproxy/mux_quic-t.h @@ -1,85 +1,15 @@ -/* - * include/haproxy/mux_quic-t.h - * This file contains types for QUIC mux-demux. - * - * Copyright 2021 HAProxy Technologies, Frédéric Lécaille - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation, version 2.1 - * exclusively. - * - * This library is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA - */ - #ifndef _HAPROXY_MUX_QUIC_T_H #define _HAPROXY_MUX_QUIC_T_H + #ifdef USE_QUIC #ifndef USE_OPENSSL #error "Must define USE_OPENSSL" #endif -#include +#include #include #include -#include - -#include - -/* Bit shift to get the stream sub ID for internal use which is obtained - * shifting the stream IDs by this value, knowing that the - * QCS_ID_TYPE_SHIFT less significant bits identify the stream ID - * types (client initiated bidirectional, server initiated bidirectional, - * client initiated unidirectional, server initiated bidirectional). - * Note that there is no reference to such stream sub IDs in the RFC. - */ -#define QCS_ID_TYPE_MASK 0x3 -#define QCS_ID_TYPE_SHIFT 2 -/* The less significant bit of a stream ID is set for a server initiated stream */ -#define QCS_ID_SRV_INTIATOR_BIT 0x1 -/* This bit is set for unidirectional streams */ -#define QCS_ID_DIR_BIT 0x2 -#define QCS_ID_DIR_BIT_SHIFT 1 - -#define OUQS_SF_TXBUF_MALLOC 0x00000001 -#define OUQS_SF_TXBUF_FULL 0x00000002 - -/* Connection flags (32 bit), in qcc->flags */ -#define QC_CF_NONE 0x00000000 - -/* Flags indicating why writing to the mux is blocked. */ -#define QC_CF_MUX_MALLOC 0x00000001 // mux blocked on lack of connection's mux buffer -#define QC_CF_MUX_MFULL 0x00000002 // mux blocked on connection's mux buffer full -#define QC_CF_MUX_BLOCK_ANY 0x00000003 // aggregate of the mux flags above - -/* Flags indicating why writing to the demux is blocked. - * The first two ones directly affect the ability for the mux to receive data - * from the connection. The other ones affect the mux's ability to demux - * received data. - */ -#define QC_CF_DEM_DFULL 0x00000004 // demux blocked on connection's demux buffer full - -#define QC_CF_DEM_MBUSY 0x00000008 // demux blocked on connection's mux side busy -#define QC_CF_DEM_MROOM 0x00000010 // demux blocked on lack of room in mux buffer -#define QC_CF_DEM_SALLOC 0x00000020 // demux blocked on lack of stream's request buffer -#define QC_CF_DEM_SFULL 0x00000040 // demux blocked on stream request buffer full -#define QC_CF_DEM_TOOMANY 0x00000100 // demux blocked waiting for some conn_streams to leave -#define QC_CF_DEM_BLOCK_ANY 0x00000170 // aggregate of the demux flags above except DFULL - -/* other flags */ -#define QC_CF_IS_BACK 0x00008000 // this is an outgoing connection - -#define QC_CF_CC_RECV 0x00010000 // CONNECTION_CLOSE received - -extern struct pool_head *pool_head_qcs; /* Stream types */ enum qcs_type { @@ -87,32 +17,15 @@ enum qcs_type { QCS_SRV_BIDI, QCS_CLT_UNI, QCS_SRV_UNI, - /* Must be the last one */ - QCS_MAX_TYPES, -}; -/* Stream direction types */ -enum qcs_dir { - QCS_BIDI = 0, - QCS_UNI = 1, /* Must be the last one */ - QCS_MAX_DIR = 2, -}; - -/* QUIC connection state, in qcc->st0 */ -enum qc_cs { - /* Initial state */ - QC_CS_NOERR, - QC_CS_ERROR, + QCS_MAX_TYPES }; -/* QUIC connection descriptor */ struct qcc { - struct connection *conn; /* mux state */ - enum qc_cs st0; /* connection flags: QC_CF_* */ - unsigned int errcode; + struct connection *conn; uint32_t flags; - /* Stream information, one by direction and by initiator */ + struct { uint64_t max_streams; /* maximum number of concurrent streams */ uint64_t largest_id; /* Largest ID of the open streams */ @@ -126,112 +39,44 @@ struct qcc { uint64_t bytes; /* Number of bytes sent */ } tx; } strms[QCS_MAX_TYPES]; - struct { - uint64_t max_data; /* Maximum number of bytes which may be received */ - uint64_t bytes; /* Number of bytes received */ - uint64_t inmux; /* Number of bytes received but not already demultiplexed. */ - } rx; + struct { uint64_t max_data; /* Maximum number of bytes which may be sent */ - uint64_t bytes; /* Number of bytes sent */ } tx; struct eb_root streams_by_id; /* all active streams by their ID */ - int timeout; /* idle timeout duration in ticks */ - int shut_timeout; /* idle timeout duration in ticks after GOAWAY was sent */ - unsigned int nb_cs; /* number of attached conn_streams */ - unsigned int stream_cnt; /* total number of streams seen */ - struct proxy *proxy; /* the proxy this connection was created for */ - struct task *task; /* timeout management task */ - struct qc_counters *px_counters; /* quic counters attached to proxy */ - struct list send_list; /* list of blocked streams requesting to send */ - struct list fctl_list; /* list of streams blocked by connection's fctl */ - struct list blocked_list; /* list of streams blocked for other reasons (e.g. sfctl, dep) */ - struct buffer_wait buf_wait; /* wait list for buffer allocations */ struct wait_event wait_event; /* To be used if we're waiting for I/Os */ - struct wait_event *subs; /* recv wait_event the mux associated is waiting on (via quic_conn_subscribe) */ - struct mt_list qcs_rxbuf_wlist; /* list of streams waiting for their rxbuf */ - void *ctx; /* Application layer context */ - const struct qcc_app_ops *app_ops; -}; + struct wait_event *subs; -/* QUIC RX states */ -enum qcs_rx_st { - QC_RX_SS_IDLE = 0, // idle - QC_RX_SS_RECV, // receive - QC_RX_SS_SIZE_KNOWN, // stream size known - /* Terminal states */ - QC_RX_SS_DATA_RECVD, // all data received - QC_RX_SS_DATA_READ, // app. read all data - QC_RX_SS_RST_RECVD, // reset received - QC_RX_SS_RST_READ, // app. read reset -}; - -/* QUIC TX states */ -enum qcs_tx_st { - QC_TX_SS_IDLE = 0, - QC_TX_SS_READY, // ready - QC_TX_SS_SEND, // send - QC_TX_SS_DATA_SENT, // all data sent - /* Terminal states */ - QC_TX_SS_DATA_RECVD, // all data received - QC_TX_SS_RST_SENT, // reset sent - QC_TX_SS_RST_RECVD, // reset received + const struct qcc_app_ops *app_ops; + void *ctx; /* Application layer context */ }; -/* QUIC stream flags (32 bit), in qcs->flags */ #define QC_SF_NONE 0x00000000 -#define QC_SF_TXBUF_MALLOC 0x00000001 // blocked on lack of TX buffer -/* stream flags indicating the reason the stream is blocked */ -#define QC_SF_BLK_MBUSY 0x00000010 // blocked waiting for mux access (transient) -#define QC_SF_BLK_MROOM 0x00000020 // blocked waiting for room in the mux (must be in send list) -#define QC_SF_BLK_MFCTL 0x00000040 // blocked due to mux fctl (must be in fctl list) -#define QC_SF_BLK_SFCTL 0x00000080 // blocked due to stream fctl (must be in blocked list) -#define QC_SF_BLK_ANY 0x000000F0 // any of the reasons above - -#define QC_SF_NOTIFIED 0x00000800 // a paused stream was notified to try to send again - -#define QC_SF_WANT_SHUTR 0x00008000 // a stream couldn't shutr() (mux full/busy) -#define QC_SF_WANT_SHUTW 0x00010000 // a stream couldn't shutw() (mux full/busy) -#define QC_SF_KILL_CONN 0x00020000 // kill the whole connection with this stream - -#define QC_SF_FIN_STREAM 0x00040000 // FIN bit must be set for last frame of the stream -#define QC_SF_DETACH 0x00080000 - -/* QUIC stream descriptor, describing the stream as it appears in the QUIC_CONN, and as - * it is being processed in the internal HTTP representation (HTX). - */ struct qcs { - struct conn_stream *cs; - struct session *sess; struct qcc *qcc; - struct eb64_node by_id; /* place in qcc's streams_by_id */ - uint64_t id; /* stream ID */ + struct conn_stream *cs; uint32_t flags; /* QC_SF_* */ + struct { - enum qcs_rx_st st; /* RX state */ - uint64_t max_data; /* maximum number of bytes which may be received */ - uint64_t offset; /* the current offset of received data */ - uint64_t bytes; /* number of bytes received */ - struct buffer buf; /* receive buffer, always valid (buf_empty or real buffer) */ struct eb_root frms; /* received frames ordered by their offsets */ + uint64_t offset; /* the current offset of received data */ + struct buffer buf; /* receive buffer, always valid (buf_empty or real buffer) */ } rx; struct { - enum qcs_tx_st st; /* TX state */ - uint64_t max_data; /* maximum number of bytes which may be sent */ - uint64_t offset; /* the current offset of data to send */ - uint64_t bytes; /* number of bytes sent */ - uint64_t ack_offset; /* last acked ordered byte offset */ + uint64_t offset; /* the current offset of received data */ struct eb_root acked_frms; /* acked frames ordered by their offsets */ - struct buffer buf; /* transmit buffer before sending via xprt */ + uint64_t ack_offset; /* last acked ordered byte offset */ + struct buffer buf; /* transmit buffer before sending via xprt */ struct buffer xprt_buf; /* buffer for xprt sending, cleared on ACK. */ } tx; - struct wait_event *subs; /* recv wait_event the conn_stream associated is waiting on (via qc_subscribe) */ - struct list list; /* To be used when adding in qcc->send_list or qcc->fctl_lsit */ - struct tasklet *shut_tl; /* deferred shutdown tasklet, to retry to send an RST after we failed to, - * in case there's no other subscription to do it */ + + struct eb64_node by_id; /* place in qcc's streams_by_id */ + + struct wait_event wait_event; + struct wait_event *subs; }; /* QUIC application layer operations */ @@ -244,4 +89,5 @@ struct qcc_app_ops { }; #endif /* USE_QUIC */ + #endif /* _HAPROXY_MUX_QUIC_T_H */ diff --git a/include/haproxy/mux_quic.h b/include/haproxy/mux_quic.h index 7f3bd3c632..aa8572c147 100644 --- a/include/haproxy/mux_quic.h +++ b/include/haproxy/mux_quic.h @@ -1,125 +1,53 @@ -/* - * include/haproxy/mux_quic-t.h - * This file contains prototypes for QUIC mux-demux. - * - * Copyright 2021 HAProxy Technologies, Frédéric Lécaille - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation, version 2.1 - * exclusively. - * - * This library is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA - */ - #ifndef _HAPROXY_MUX_QUIC_H #define _HAPROXY_MUX_QUIC_H + #ifdef USE_QUIC #ifndef USE_OPENSSL #error "Must define USE_OPENSSL" #endif -#include +#include #include -#include void quic_mux_transport_params_update(struct qcc *qcc); -void qc_error(struct qcc *qcc, int err); -struct buffer *qc_get_buf(struct qcs *qcs, struct buffer *bptr); -struct qcs *qcc_get_stream(struct qcc *qcc, uint64_t id); -struct qcs *bidi_qcs_new(struct qcc *qcc, uint64_t id); -struct qcs *luqs_new(struct qcc *qcc); -struct qcs *ruqs_new(struct qcc *qcc, uint64_t id); -size_t luqs_snd_buf(struct qcs *qcs, struct buffer *buf, size_t count, int flags); -void qcs_release(struct qcs *qcs); +struct qcs *qcs_new(struct qcc *qcc, uint64_t id, enum qcs_type type); +void uni_qcs_free(struct qcs *qcs); -void ruqs_notify_recv(struct qcs *qcs); +struct buffer *qc_get_buf(struct qcs *qcs, struct buffer *bptr); -/* Return 1 if the stream with as ID attached to connection - * has been locally initiated, 0 if not. +/* Bit shift to get the stream sub ID for internal use which is obtained + * shifting the stream IDs by this value, knowing that the + * QCS_ID_TYPE_SHIFT less significant bits identify the stream ID + * types (client initiated bidirectional, server initiated bidirectional, + * client initiated unidirectional, server initiated bidirectional). + * Note that there is no reference to such stream sub IDs in the RFC. */ -static inline int qc_local_stream_id(struct qcc *qcc, uint64_t id) -{ - if ((objt_listener(qcc->conn->target) && (id & QCS_ID_SRV_INTIATOR_BIT)) || - (objt_server(qcc->conn->target) && !(id & QCS_ID_SRV_INTIATOR_BIT))) - return 1; - - return 0; -} - -/* Return 1 if stream has been locally initiated, 0 if not. */ -static inline int qcs_local(struct qcs *qcs) -{ - if ((objt_listener(qcs->qcc->conn->target) && (qcs->id & QCS_ID_SRV_INTIATOR_BIT)) || - (objt_server(qcs->qcc->conn->target) && !(qcs->id & QCS_ID_SRV_INTIATOR_BIT))) - return 1; - - return 0; -} - -/* Return the direction of a stream with as ID. */ -static inline enum qcs_dir qcs_id_dir(uint64_t id) -{ - return (id & QCS_ID_DIR_BIT) >> QCS_ID_DIR_BIT_SHIFT; -} - -/* Return the direction of QUIC stream. */ -static inline enum qcs_dir qcs_dir(struct qcs *qcs) -{ - return (qcs->id & QCS_ID_DIR_BIT) >> QCS_ID_DIR_BIT_SHIFT; -} +#define QCS_ID_TYPE_MASK 0x3 +#define QCS_ID_TYPE_SHIFT 2 +/* The less significant bit of a stream ID is set for a server initiated stream */ +#define QCS_ID_SRV_INTIATOR_BIT 0x1 +/* This bit is set for unidirectional streams */ +#define QCS_ID_DIR_BIT 0x2 static inline enum qcs_type qcs_id_type(uint64_t id) { return id & QCS_ID_TYPE_MASK; } -static inline enum qcs_type qcs_type_from_dir(struct qcc *qcc, enum qcs_dir dir) -{ - return (dir << QCS_ID_DIR_BIT_SHIFT) | - (!!objt_listener(qcc->conn->target) ? QCS_ID_SRV_INTIATOR_BIT : 0); -} - -static inline int64_t qcc_wnd(struct qcc *qcc) -{ - return qcc->tx.max_data - qcc->tx.bytes; -} - -/* Return 1 if is unidirectional, 0 if not. */ -static inline int qcs_uni(struct qcs *qcs) -{ - return qcs->id & QCS_ID_DIR_BIT; -} - -/* Return 1 if is bidirectional, 0 if not. */ -static inline int qcs_bidi(struct qcs *qcs) -{ - return !qcs_uni(qcs); -} - -/* Return the next stream ID with as type if succeeded, (uint64_t)-1 if not. */ -static inline uint64_t qcs_next_id(struct qcc *qcc, enum qcs_type qcs_type) +/* Return 1 if the stream with as ID attached to connection + * has been locally initiated, 0 if not. + */ +static inline int qc_local_stream_id(struct qcc *qcc, uint64_t id) { - if (qcc->strms[qcs_type].nb_streams + 1 > qcc->strms[qcs_type].max_streams) - return (uint64_t)-1; - - return (qcc->strms[qcs_type].nb_streams++ << QCS_ID_TYPE_SHIFT) | qcs_type; + return id & QCS_ID_SRV_INTIATOR_BIT; } -static inline void *qcs_new(struct qcc *qcc, uint64_t id) +static inline int qcs_get_next_id(struct qcc *qcc, enum qcs_type type) { - if (id & QCS_ID_DIR_BIT) - return ruqs_new(qcc, id); - else - return bidi_qcs_new(qcc, id); + BUG_ON(qcc->strms[type].nb_streams + 1 > qcc->strms[type].max_streams); + return (qcc->strms[type].nb_streams++ << QCS_ID_TYPE_SHIFT) | type; } #endif /* USE_QUIC */ + #endif /* _HAPROXY_MUX_QUIC_H */ diff --git a/include/haproxy/qpack-dec.h b/include/haproxy/qpack-dec.h index f45246f766..d020ced662 100644 --- a/include/haproxy/qpack-dec.h +++ b/include/haproxy/qpack-dec.h @@ -21,8 +21,6 @@ #ifndef _HAPROXY_QPACK_DEC_H #define _HAPROXY_QPACK_DEC_H -#include - struct h3_uqs; struct http_hdr; diff --git a/include/haproxy/quic_frame-t.h b/include/haproxy/quic_frame-t.h index f174073195..fcc04fc0b2 100644 --- a/include/haproxy/quic_frame-t.h +++ b/include/haproxy/quic_frame-t.h @@ -33,6 +33,8 @@ #include +#include + /* QUIC frame types. */ enum quic_frame_type { QUIC_FT_PADDING = 0x00, diff --git a/include/haproxy/xprt_quic-t.h b/include/haproxy/xprt_quic-t.h index ec18c1ec67..543d670ed8 100644 --- a/include/haproxy/xprt_quic-t.h +++ b/include/haproxy/xprt_quic-t.h @@ -32,6 +32,7 @@ #include #include +#include #include #include #include diff --git a/src/h3.c b/src/h3.c index 6414c598e0..af206565eb 100644 --- a/src/h3.c +++ b/src/h3.c @@ -23,7 +23,7 @@ #include #include #include -#include +#include #include #include #include @@ -322,8 +322,9 @@ static int h3_control_recv(struct h3_uqs *h3_uqs, void *ctx) b_del(rxbuf, flen); } - if (b_data(rxbuf)) - h3->qcc->conn->mux->ruqs_subscribe(h3_uqs->qcs, SUB_RETRY_RECV, &h3->rctrl.wait_event); + /* TODO handle the case when the buffer is not empty. This can happens + * if there is an incomplete frame. + */ return 1; } @@ -533,10 +534,8 @@ static int h3_resp_data_send(struct qcs *qcs, struct buffer *buf, size_t count) /* not enough room for headers and at least one data byte, block the * stream */ - if (b_size(&outbuf) <= hsize) { - qcs->flags |= QC_SF_BLK_MROOM; - goto end; - } + if (b_size(&outbuf) <= hsize) + ABORT_NOW(); if (b_size(&outbuf) < hsize + fsize) fsize = b_size(&outbuf) - hsize; @@ -573,9 +572,11 @@ size_t h3_snd_buf(struct conn_stream *cs, struct buffer *buf, size_t count, int int32_t idx; int ret; + fprintf(stderr, "%s\n", __func__); + htx = htx_from_buf(buf); - while (count && !htx_is_empty(htx) && !(qcs->flags & QC_SF_BLK_MROOM)) { + while (count && !htx_is_empty(htx)) { idx = htx_get_head(htx); blk = htx_get_blk(htx, idx); btype = htx_get_blk_type(blk); @@ -619,9 +620,6 @@ size_t h3_snd_buf(struct conn_stream *cs, struct buffer *buf, size_t count, int } } - if ((htx->flags & HTX_FL_EOM) && htx_is_empty(htx)) - qcs->flags |= QC_SF_FIN_STREAM; - out: if (total) { if (!(qcs->qcc->wait_event.events & SUB_RETRY_SEND)) @@ -657,7 +655,7 @@ static int h3_attach_ruqs(struct qcs *qcs, void *ctx) h3->rctrl.qcs = qcs; h3->rctrl.cb = h3_control_recv; - h3->qcc->conn->mux->ruqs_subscribe(qcs, SUB_RETRY_RECV, &h3->rctrl.wait_event); + // TODO wake-up rctrl tasklet on reception break; case H3_UNI_STRM_TP_PUSH_STREAM: /* NOT SUPPORTED */ @@ -670,7 +668,7 @@ static int h3_attach_ruqs(struct qcs *qcs, void *ctx) h3->rqpack_enc.qcs = qcs; h3->rqpack_enc.cb = qpack_decode_enc; - h3->qcc->conn->mux->ruqs_subscribe(qcs, SUB_RETRY_RECV, &h3->rqpack_enc.wait_event); + // TODO wake-up rqpack_enc tasklet on reception break; case H3_UNI_STRM_TP_QPACK_DECODER: if (h3->rqpack_dec.qcs) { @@ -680,7 +678,7 @@ static int h3_attach_ruqs(struct qcs *qcs, void *ctx) h3->rqpack_dec.qcs = qcs; h3->rqpack_dec.cb = qpack_decode_dec; - h3->qcc->conn->mux->ruqs_subscribe(qcs, SUB_RETRY_RECV, &h3->rqpack_dec.wait_event); + // TODO wake-up rqpack_dec tasklet on reception break; default: /* Error */ @@ -694,8 +692,10 @@ static int h3_attach_ruqs(struct qcs *qcs, void *ctx) static int h3_finalize(void *ctx) { struct h3 *h3 = ctx; + int lctrl_id; - h3->lctrl.qcs = luqs_new(h3->qcc); + lctrl_id = qcs_get_next_id(h3->qcc, QCS_SRV_UNI); + h3->lctrl.qcs = qcs_new(h3->qcc, lctrl_id, QCS_SRV_UNI); if (!h3->lctrl.qcs) return 0; @@ -766,7 +766,7 @@ static int h3_uqs_init(struct h3_uqs *h3_uqs, struct h3 *h3, static inline void h3_uqs_release(struct h3_uqs *h3_uqs) { if (h3_uqs->qcs) - qcs_release(h3_uqs->qcs); + uni_qcs_free(h3_uqs->qcs); } static inline void h3_uqs_release_all(struct h3 *h3) diff --git a/src/hq_interop.c b/src/hq_interop.c index 075e785b04..af59410b7b 100644 --- a/src/hq_interop.c +++ b/src/hq_interop.c @@ -81,7 +81,7 @@ static size_t hq_interop_snd_buf(struct conn_stream *cs, struct buffer *buf, res = mux_get_buf(qcs); outbuf = b_make(b_tail(res), b_contig_space(res), 0, 0); - while (count && !htx_is_empty(htx) && !(qcs->flags & QC_SF_BLK_MROOM)) { + while (count && !htx_is_empty(htx)) { /* Not implemented : QUIC on backend side */ idx = htx_get_head(htx); blk = htx_get_blk(htx, idx); @@ -116,9 +116,6 @@ static size_t hq_interop_snd_buf(struct conn_stream *cs, struct buffer *buf, } } - if ((htx->flags & HTX_FL_EOM) && htx_is_empty(htx)) - qcs->flags |= QC_SF_FIN_STREAM; - b_add(res, b_data(&outbuf)); if (total) { diff --git a/src/mux_quic.c b/src/mux_quic.c index 64990e4614..774612b8af 100644 --- a/src/mux_quic.c +++ b/src/mux_quic.c @@ -1,1236 +1,87 @@ -/* - * QUIC mux-demux for connections - * - * Copyright 2021 HAProxy Technologies, Frédéric Lécaille - * - * 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 #include #include -#include #include -#include -#include -#include -#include -#include -#include -#include +#include +#include #include -#include -#include -#include -#include - -/* 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_qcc, "qcc", sizeof(struct qcc)); 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 - * 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; - } -} - -/* 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). 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 = NULL; - - /* 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 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); -} + struct quic_transport_params *clt_params; -/* 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 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 for connection 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 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 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 on the 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 on the 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 - * 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); -} + /* Client parameters, params used to TX. */ + clt_params = &qcc->conn->qc->tx.params; -/* 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); + 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; } -/* resume each qcs eligible for sending in list head */ -__maybe_unused -static void qc_resume_each_sending_qcs(struct qcc *qcc, struct list *head) +/* Allocate a new QUIC streams with id and type . */ +struct qcs *qcs_new(struct qcc *qcc, uint64_t id, enum qcs_type type) { - struct qcs *qcs, *qcs_back; + struct qcs *qcs; - TRACE_ENTER(QC_EV_QCC_SEND|QC_EV_QCS_WAKE, qcc->conn); + qcs = pool_alloc(pool_head_qcs); + if (!qcs) + goto out; - 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; + fprintf(stderr, "%s: stream ID %llu\n", __func__, qcs->by_id.key); - qcs->flags &= ~QC_SF_BLK_ANY; + qcs->qcc = qcc; + qcs->cs = NULL; + qcs->flags = QC_SF_NONE; - if (qcs->flags & QC_SF_NOTIFIED) - continue; + qcs->by_id.key = id; + eb64_insert(&qcc->streams_by_id, &qcs->by_id); + qcc->strms[type].nb_streams++; - /* 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; - } + qcs->rx.buf = BUF_NULL; + qcs->rx.offset = 0; + qcs->rx.frms = EB_ROOT_UNIQUE; - 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); - } - } + qcs->tx.buf = BUF_NULL; + qcs->tx.xprt_buf = BUF_NULL; + qcs->tx.offset = 0; + qcs->tx.ack_offset = 0; + qcs->tx.acked_frms = EB_ROOT_UNIQUE; - TRACE_LEAVE(QC_EV_QCC_SEND|QC_EV_QCS_WAKE, qcc->conn); + qcs->wait_event.tasklet = NULL; + qcs->wait_event.events = 0; + qcs->subs = NULL; + + out: + return qcs; } -/* process Tx frames from streams to be multiplexed. Returns > 0 if it reached - * the end. +/* Free a qcs. This function must only be used for unidirectional streams. + * Bidirectional streams are released by the upper layer through qc_detach(). */ -__maybe_unused -static int qc_process_mux(struct qcc *qcc) +void uni_qcs_free(struct qcs *qcs) { - 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; + eb64_delete(&qcs->by_id); + pool_free(pool_head_qcs, qcs); } - -/* 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) +struct buffer *qc_get_buf(struct qcs *qcs, struct buffer *bptr) { - TRACE_ENTER(QC_EV_QCC_RECV, qcc->conn); - /* XXX TO DO XXX */ - TRACE_LEAVE(QC_EV_QCC_RECV, qcc->conn); - return 0; + struct buffer *buf = b_alloc(bptr); + BUG_ON(!buf); + return buf; } static int qcs_push_frame(struct qcs *qcs, struct buffer *payload, int fin, uint64_t offset) @@ -1240,6 +91,8 @@ static int qcs_push_frame(struct qcs *qcs, struct buffer *payload, int fin, uint struct quic_enc_level *qel = &qcs->qcc->conn->qc->els[QUIC_TLS_ENC_LEVEL_APP]; int total = 0, to_xfer; + fprintf(stderr, "%s\n", __func__); + qc_get_buf(qcs, buf); to_xfer = QUIC_MIN(b_data(payload), b_room(buf)); if (!to_xfer) @@ -1258,7 +111,7 @@ static int qcs_push_frame(struct qcs *qcs, struct buffer *payload, int fin, uint frm->type |= QUIC_STREAM_FRAME_TYPE_OFF_BIT; frm->stream.offset.key = offset; } - frm->stream.qcs = qcs; + frm->stream.qcs = (struct qcs *)qcs; frm->stream.buf = buf; frm->stream.id = qcs->by_id.key; if (total) { @@ -1275,19 +128,12 @@ static int qcs_push_frame(struct qcs *qcs, struct buffer *payload, int fin, uint 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; + int ret; - 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); + fprintf(stderr, "%s\n", __func__); /* TODO simple loop through all streams and check if there is frames to * send @@ -1297,307 +143,101 @@ static int qc_send(struct qcc *qcc) 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); + /* TODO handle the FIN parameter */ + ret = qcs_push_frame(qcs, buf, 0, 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); - } - } + /* TODO wake-up xprt if data were transfered */ + fprintf(stderr, "%s ret=%d\n", __func__, ret); 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; + return ret; } -/* 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; + fprintf(stderr, "%s\n", __func__); - 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); + qc_send(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) +static int qc_init(struct connection *conn, struct proxy *prx, + struct session *sess, struct buffer *input) { - TRACE_ENTER(QC_EV_QCC_WAKE); - /* XXX TO DO XXX */ - TRACE_LEAVE(QC_EV_QCC_WAKE); - return NULL; -} + struct qcc *qcc; + qcc = pool_alloc(pool_head_qcc); + if (!qcc) + goto fail_no_qcc; -/*******************************************/ -/* functions below are used by the streams */ -/*******************************************/ + qcc->conn = conn; + conn->ctx = qcc; + conn->qc->qcc = qcc; -/* - * 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; + qcc->app_ops = NULL; - 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; -} + qcc->streams_by_id = EB_ROOT_UNIQUE; -/* 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; + qcc->strms[QCS_CLT_BIDI].nb_streams = 0; + qcc->strms[QCS_CLT_BIDI].largest_id = -1; + qcc->strms[QCS_CLT_UNI].nb_streams = 0; + qcc->strms[QCS_CLT_UNI].largest_id = -1; + qcc->strms[QCS_SRV_BIDI].nb_streams = 0; + qcc->strms[QCS_SRV_BIDI].largest_id = -1; + qcc->strms[QCS_SRV_UNI].nb_streams = 0; + qcc->strms[QCS_SRV_UNI].largest_id = -1; - 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; -} + qcc->wait_event.tasklet = tasklet_new(); + if (!qcc->wait_event.tasklet) + goto fail_no_tasklet; -static int qc_ctl(struct connection *conn, enum mux_ctl_type mux_ctl, void *output) -{ - int ret = 0; - struct qcc *qcc = conn->ctx; + qcc->subs = NULL; + qcc->wait_event.tasklet->process = qc_io_cb; + qcc->wait_event.tasklet->context = qcc; - 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; - } -} + /* init read cycle */ + tasklet_wakeup(qcc->wait_event.tasklet); -/* - * Destroy the mux and the associated connection, if it is no longer used - */ -static void qc_destroy(void *ctx) -{ - struct qcc *qcc = ctx; + return 0; - 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); + fail_no_tasklet: + pool_free(pool_head_qcc, qcc); + fail_no_qcc: + return -1; } -/* - * 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) +/* 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) { + /* XXX TODO XXX */ + fprintf(stderr, "%s\n", __func__); - 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); + return 0; } -/* shutw() called by the conn_stream (mux_ops.shutw) */ -static void qc_shutw(struct conn_stream *cs, enum cs_shw_mode mode) +static size_t qc_snd_buf(struct conn_stream *cs, struct buffer *buf, + size_t count, int flags) { 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; + fprintf(stderr, "%s\n", __func__); - qc_do_shutw(qcs); - TRACE_LEAVE(QC_EV_STRM_SHUT, qcs->qcc->conn, qcs); + return qcs->qcc->app_ops->snd_buf(cs, buf, count, flags); } /* Called from the upper layer, to subscribe to events . The @@ -1605,33 +245,10 @@ static void qc_shutw(struct conn_stream *cs, enum cs_shw_mode mode) * as at least one event is still subscribed. The 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) +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); + /* XXX TODO XXX */ return 0; } @@ -1643,8 +260,6 @@ static int qc_unsubscribe(struct conn_stream *cs, int event_type, struct wait_ev { 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); @@ -1652,238 +267,19 @@ static int qc_unsubscribe(struct conn_stream *cs, int event_type, struct wait_ev 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 to events . The - * event subscriber is not allowed to change from a previous call as long - * as at least one event is still subscribed. The 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 from events . - * The 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 to events . The - * event subscriber is not allowed to change from a previous call as long - * as at least one event is still subscribed. The 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 from events . - * The 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; } -static size_t qc_snd_buf(struct conn_stream *cs, struct buffer *buf, size_t count, int flags) -{ - struct qcs *qcs = cs->ctx; - return qcs->qcc->app_ops->snd_buf(cs, buf, count, flags); -} - -/****************************************/ -/* 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, + .detach = qc_detach, .rcv_buf = qc_rcv_buf, + .snd_buf = qc_snd_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 }; + { .token = IST("quic"), .mode = PROTO_MODE_HTTP, .side = PROTO_SIDE_FE, .mux = &qc_ops }; INITCALL1(STG_REGISTER, register_mux_proto, &mux_proto_quic); - diff --git a/src/xprt_quic.c b/src/xprt_quic.c index 0ed7336c35..96d409fa7d 100644 --- a/src/xprt_quic.c +++ b/src/xprt_quic.c @@ -1426,16 +1426,12 @@ static inline void qc_treat_acked_tx_frm(struct quic_frame *frm, qcs->tx.ack_offset += strm->len; LIST_DELETE(&frm->list); pool_free(pool_head_quic_frame, frm); - qc->qcc->flags &= ~QC_CF_MUX_MFULL; stream_acked = 1; } else { eb64_insert(&qcs->tx.acked_frms, &strm->offset); } stream_acked |= qcs_try_to_consume(qcs); - - if (qcs->flags & QC_SF_DETACH) - tasklet_wakeup(qcs->shut_tl); } break; default: @@ -1989,7 +1985,9 @@ static struct eb64_node *qcc_get_qcs(struct qcc *qcc, uint64_t id) qcs = NULL; for (i = largest_id + 1; i <= sub_id; i++) { - qcs = qcs_new(qcc, (i << QCS_ID_TYPE_SHIFT) | strm_type); + uint64_t id = (i << QCS_ID_TYPE_SHIFT) | strm_type; + enum qcs_type type = id & QCS_ID_DIR_BIT ? QCS_CLT_UNI : QCS_CLT_BIDI; + qcs = qcs_new(qcc, id, type); if (!qcs) { TRACE_PROTO("Could not allocate a new stream", QUIC_EV_CONN_PSTRM, qcc->conn); @@ -2151,9 +2149,6 @@ static int qc_handle_uni_strm_frm(struct quic_rx_packet *pkt, return 0; } - if (ret) - ruqs_notify_recv(strm); - strm_frm->offset.key += ret; } /* Take this frame into an account for the stream flow control */ @@ -2297,7 +2292,7 @@ static int qc_parse_pkt_frms(struct quic_rx_packet *pkt, struct ssl_sock_ctx *ct break; case QUIC_FT_CONNECTION_CLOSE: case QUIC_FT_CONNECTION_CLOSE_APP: - conn->qcc->flags |= QC_CF_CC_RECV; + /* TODO warn the mux to close the connection */ break; case QUIC_FT_HANDSHAKE_DONE: if (objt_listener(ctx->conn->target))