From: Frantisek Tobias Date: Wed, 3 Jun 2026 13:11:29 +0000 (+0200) Subject: daemon/quic: rewrite timer system, let terminal tasks traverse the entire DoQ group... X-Git-Tag: v6.4.1^2~8 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=fef18094996f571da32ef423b8355c02b47f8982;p=thirdparty%2Fknot-resolver.git daemon/quic: rewrite timer system, let terminal tasks traverse the entire DoQ group to assure gracefull termination the timer system was incorrectly oblivious to some ngtcp2 states, this revrite amends missing functionality improving performance stability and correctness. Terminal events now reach all layers, this most importantly solves issues with dangling qr_tasks. --- diff --git a/daemon/quic_common.c b/daemon/quic_common.c index 5cff5a1af..c5d15ee45 100644 --- a/daemon/quic_common.c +++ b/daemon/quic_common.c @@ -1,7 +1,9 @@ /* Copyright (C) CZ.NIC, z.s.p.o. * SPDX-License-Identifier: GPL-3.0-or-later */ + #include +#include #include "contrib/openbsd/siphash.h" #include "lib/dnssec.h" @@ -64,6 +66,78 @@ bool kr_quic_conn_timeout(struct pl_quic_conn_sess_data *conn, uint64_t *now) return *now > ngtcp2_conn_get_expiry(conn->conn); } +void quic_hs_timeout(uv_timer_t *timer) +{ + struct session2 *s = timer->data; + session2_timer_stop(s); + + struct pl_quic_conn_sess_data *conn = + protolayer_sess_data_get_proto(s, PROTOLAYER_TYPE_QUIC_CONN); + session2_event(conn->h.session, PROTOLAYER_EVENT_CONNECT_TIMEOUT, NULL); +} + +void quic_idle_timeout(uv_timer_t *timer) +{ + struct session2 *s = timer->data; + struct pl_quic_conn_sess_data *conn = + protolayer_sess_data_get_proto(s, PROTOLAYER_TYPE_QUIC_CONN); + uint64_t now = quic_timestamp(); + int ret = ngtcp2_conn_handle_expiry(conn->conn, now); + if (ret != 0) { + kr_log_debug(DOQ, "Result of ngtcp2_conn_handle_expiry: %s (%d)\n", + ngtcp2_strerror(ret), ret); + } + if (ret == NGTCP2_ERR_IDLE_CLOSE) { + /* idle equal max_idle_timeout, don't send CONNECTION_CLOSE */ + session2_force_close(s); + } else if (ret < 0) { + session2_close(s); + } else { + quic_flush_streams(conn); + /* let the connection itself send whatever data it needs to */ + session2_wrap(s, protolayer_payload_wire_buf(&s->wire_buf, true), + &conn->comm_storage, NULL, NULL, NULL); + } +} + +int quic_set_timeout(struct session2 *s, uint64_t ms, uv_timer_cb timeout_cb) +{ + uv_timer_stop(&s->timer); + return uv_timer_start(&s->timer, timeout_cb, ms, 0); +} + +int quic_set_hs_timeout(struct session2 *s, uint64_t ms) +{ + return quic_set_timeout(s, ms, quic_hs_timeout); +} + +int quic_set_idle_timeout(struct session2 *s, uint64_t ms) +{ + return quic_set_timeout(s, ms, quic_idle_timeout); +} + +void quic_reset_expiry(struct pl_quic_conn_sess_data *conn) +{ + kr_require(conn); + struct session2 *s = conn->h.session; + ngtcp2_tstamp expiry = ngtcp2_conn_get_expiry(conn->conn); + + uv_timer_stop(&s->timer); + if (expiry == UINT64_MAX) { + uv_timer_start(&s->timer, quic_idle_timeout, + QUIC_MAX_IDLE_TIMEOUT, 0); + return; + } + uint64_t now = quic_timestamp(); + if (expiry > now) { + uint64_t ceil_delay_ns = expiry - now + NGTCP2_MILLISECONDS - 1; + uint64_t delay_ms = ceil_delay_ns / NGTCP2_MILLISECONDS; + uv_timer_start(&s->timer, quic_idle_timeout, delay_ms, 0); + } else { + uv_timer_start(&s->timer, quic_idle_timeout, 0, 0); + } +} + void init_random_cid(ngtcp2_cid *cid, size_t len) { if (len == 0) @@ -150,7 +224,9 @@ int set_application_error(struct pl_quic_conn_sess_data *conn, if (kr_fails_assert(conn && msglen < 128)) return kr_error(EINVAL); - memcpy(&conn->err_msg_buffer, msg, msglen); + if (msg && msglen > 0) { + memcpy(&conn->err_msg_buffer, msg, msglen); + } ngtcp2_ccerr_set_application_error(&conn->ccerr, error_code, conn->err_msg_buffer, msglen); diff --git a/daemon/quic_common.h b/daemon/quic_common.h index af249dcf1..71a5696d4 100644 --- a/daemon/quic_common.h +++ b/daemon/quic_common.h @@ -60,13 +60,21 @@ typedef enum { #define MAX_QUIC_FRAME_SIZE 65536 #define QUIC_MAX_SEND_PER_RECV 4 -#define QUIC_CONN_IDLE_TIMEOUT (5 * NGTCP2_SECONDS) -#define QUIC_HS_IDLE_TIMEOUT (5 * NGTCP2_SECONDS) - -/* HACK adjust pointer of conn->streams head so it points to - * struct pl_quic_stream_sess_data, this is hacky */ -#define container_of(ptr, type, member) \ - ((type *)((char *)(ptr) - offsetof(type, member))) +/* The connection can reach a state when the internal ngtcp2 state machine is + * not waiting for any event. In such cases the ngtcp2_conn_get_expiry + * returns UINT64_MAX => meaining no timer event. If the peer abruptly + * stops communicating in such a state the connection would be active + * untill a new connection prompts a conn table sweep. Instead of waiting + * for a new connection we set the timer to the QUIC_MAX_TIMEOUT to assure + * that every connection will terminate within a reasonable time period + * regardless of the incoming traffic. This is just a cleanliness measure. */ +#define QUIC_MAX_IDLE_TIMEOUT (15 * NGTCP2_SECONDS) +#define QUIC_CONN_IDLE_TIMEOUT (2 * NGTCP2_SECONDS) +#define QUIC_HS_IDLE_TIMEOUT (3 * NGTCP2_SECONDS) + +#define container_of(ptr, type, member) ({ \ + const typeof( ((type *)0)->member ) *__mptr = (ptr); \ + (type *)((char *)__mptr - offsetof(type,member));}) typedef enum { KNOT_QUIC_TABLE_CLIENT_ONLY = (1 << 0), @@ -131,6 +139,11 @@ int kr_quic_table_add(struct pl_quic_conn_sess_data *conn_sess, bool init_unique_cid(ngtcp2_cid *cid, size_t len, kr_quic_table_t *table); void init_random_cid(ngtcp2_cid *cid, size_t len); uint64_t quic_timestamp(void); + +void quic_hs_timeout(uv_timer_t *timer); +void quic_idle_timeout(uv_timer_t *timer); +void quic_reset_expiry(struct pl_quic_conn_sess_data *conn); + kr_quic_cid_t **kr_quic_table_lookup2(const ngtcp2_cid *cid, kr_quic_table_t *table); struct pl_quic_conn_sess_data *kr_quic_table_lookup(const ngtcp2_cid *cid, diff --git a/daemon/quic_conn.c b/daemon/quic_conn.c index 551b37f06..daef519c4 100644 --- a/daemon/quic_conn.c +++ b/daemon/quic_conn.c @@ -3,10 +3,12 @@ */ #include "quic_conn.h" +#include "lib/proto.h" #include "network.h" #include "quic_common.h" #include "quic_stream.h" #include "lib/dnssec.h" +#include "session2.h" #include "worker.h" #include #include @@ -94,8 +96,16 @@ static struct tls_credentials *tls_credentials_reserve( static int handshake_completed_cb(ngtcp2_conn *ngconn, void *user_data) { - (void)ngconn; - QUIC_SET_HS_COMPLETED((struct pl_quic_conn_sess_data *)user_data); + struct pl_quic_conn_sess_data *conn = user_data; + QUIC_SET_HS_COMPLETED(conn); + + /* Notify worker that waiting tasks can be sent out */ + if (conn->h.session->outgoing) { + // pass + } else { + quic_reset_expiry(conn); + } + return kr_ok(); } @@ -193,6 +203,7 @@ finished: msg, sizeof(msg) - 1); return NGTCP2_ERR_CALLBACK_FAILURE; } + session2_inc_refs(stream->h.session); queue_push(conn->pending_unwrap, stream); } @@ -247,6 +258,8 @@ static int stream_open_cb(ngtcp2_conn *ngconn, ++conn->streams_count; ngtcp2_conn_set_stream_user_data(ngconn, stream_id, stream); + session2_event(stream->h.session, PROTOLAYER_EVENT_CONNECT, NULL); + return NGTCP2_NO_ERROR; } @@ -255,9 +268,9 @@ static int stream_close_cb(ngtcp2_conn *ngconn, uint32_t flags, void *user_data, void *stream_user_data) { ngtcp2_conn_extend_max_streams_bidi(ngconn, 1); + struct pl_quic_conn_sess_data *conn = user_data; struct pl_quic_stream_sess_data *stream = stream_user_data; - rem_node(&stream->list_node); session2_close(stream->h.session); --conn->streams_count; ++conn->finished_streams; @@ -601,6 +614,33 @@ int quic_init_server_conn(struct pl_quic_conn_sess_data *conn, return ret; } +int quic_flush_streams(struct pl_quic_conn_sess_data *conn) +{ + kr_require(conn); + node_t *s_node; + WALK_LIST(s_node, conn->streams) { + struct pl_quic_stream_sess_data *s = + container_of(s_node, struct pl_quic_stream_sess_data, + list_node); + /* Skip streams that have no data to send */ + if (s->unsent_obuf == NULL) + continue; + + if (session2_wrap_after(s->h.session, + PROTOLAYER_TYPE_DNS_SINGLE_STREAM, + protolayer_payload_wire_buf( + &s->h.session->wire_buf, true), + &conn->comm_storage, + NULL, + NULL) < 0) { + /* Either closing or invalid */ + session2_close(s->h.session); + } + } + + return kr_ok(); +} + static void copy_comm_storage( struct pl_quic_conn_sess_data *conn, struct comm_info *comm) @@ -640,7 +680,7 @@ int send_special(ngtcp2_version_cid *dec_cids, kr_quic_table_t *table, .size = NGTCP2_MAX_UDP_PAYLOAD_SIZE, .start = 0, }; - struct wire_buf *save = ctx->payload.wire_buf; + struct wire_buf *save_wb = ctx->payload.wire_buf; ctx->payload.wire_buf = &err_wb; uint64_t now = quic_timestamp(); @@ -663,6 +703,7 @@ int send_special(ngtcp2_version_cid *dec_cids, kr_quic_table_t *table, dec_cids->dcid, dec_cids->dcidlen, supported_quic, sizeof(supported_quic) / sizeof(*supported_quic) ); + break; case QUIC_SEND_RETRY: kr_require(dec_cids); ret = write_retry_packet(ctx->payload.wire_buf, @@ -710,7 +751,7 @@ int send_special(ngtcp2_version_cid *dec_cids, kr_quic_table_t *table, } mm_free(&ctx->pool, ctx->payload.wire_buf); - ctx->payload.wire_buf = save; + ctx->payload.wire_buf = save_wb; return ret; } @@ -736,25 +777,22 @@ static enum protolayer_iter_cb_result pl_quic_conn_unwrap(void *sess_data, if ((ret = quic_init_server_conn(conn, ctx)) != kr_ok()) { kr_log_error(DOQ, "Failed to initiate quic connection (%d)\n", ret); - session2_force_close(conn->h.session); + QUIC_SET_DRAINING(conn); return protolayer_break(ctx, ret); } } quic_doq_error_t doq_error; - uv_timer_again(&conn->h.session->timer); ret = handle_packet(conn, ctx, &doq_error); if (ret != kr_ok()) { - ret = send_special(&conn->dec_cids, + (void)send_special(&conn->dec_cids, conn->table_ref, ctx, ret, conn, conn->h.session, &doq_error); - if (ret == kr_ok()) { - ngtcp2_conn_update_pkt_tx_time(conn->conn, - quic_timestamp()); - } return protolayer_break(ctx, kr_ok()); } + quic_reset_expiry(conn); + if (queue_len(conn->pending_unwrap) == 0) { ret = session2_wrap(conn->h.session, ctx->payload, @@ -767,11 +805,13 @@ static enum protolayer_iter_cb_result pl_quic_conn_unwrap(void *sess_data, } while (queue_len(conn->pending_unwrap) > 0) { - session2_unwrap(queue_head(conn->pending_unwrap)->h.session, + struct session2 *s = queue_head(conn->pending_unwrap)->h.session; + session2_unwrap(s, ctx->payload, NULL /* &conn->comm_storage */, ctx->finished_cb, ctx->finished_cb_baton); + session2_dec_refs(s); queue_pop(conn->pending_unwrap); } @@ -787,6 +827,8 @@ static enum protolayer_iter_cb_result pl_quic_conn_wrap(void *sess_data, return protolayer_break(ctx, kr_ok()); } + kr_quic_set_addrs(ctx, &conn->path); + if (ctx->payload.type != PROTOLAYER_PAYLOAD_IOVEC) { ngtcp2_ssize sent = 0; ngtcp2_conn_info info = { 0 }; @@ -804,24 +846,27 @@ static enum protolayer_iter_cb_result pl_quic_conn_wrap(void *sess_data, &sent, NGTCP2_WRITE_STREAM_FLAG_NONE, -1, NULL, 0, quic_timestamp()); + if (nwrite > 0) + ngtcp2_conn_update_pkt_tx_time(conn->conn, quic_timestamp()); + if (nwrite == 0) + quic_reset_expiry(conn); + if (nwrite <= 0) { if (nwrite == NGTCP2_ERR_NOMEM) { - size_t inc = MIN(ctx->payload.wire_buf->size, 1024); - char *new_buf = realloc(ctx->payload.wire_buf->buf, inc); + size_t new_size = ctx->payload.wire_buf->size + 1024; + char *new_buf = realloc(ctx->payload.wire_buf->buf, new_size); kr_require(new_buf); - ctx->payload.wire_buf->buf = new_buf; - ctx->payload.wire_buf->end += inc; - ctx->payload.wire_buf->size += inc; + ctx->payload.wire_buf->size = new_size; + ctx->payload.wire_buf->end = new_size; } - return protolayer_break(ctx, kr_error(EINVAL)); + return protolayer_break(ctx, kr_ok()); } wire_buf_consume(ctx->payload.wire_buf, nwrite); } - ngtcp2_conn_update_pkt_tx_time(conn->conn, quic_timestamp()); if (conn->state & QUIC_STATE_CLOSING) { QUIC_SET_DRAINING(conn); } @@ -905,9 +950,7 @@ static int pl_quic_conn_sess_init(struct session2 *session, void *sess_data, voi return kr_error(EINVAL); } - session2_timer_start(session, PROTOLAYER_EVENT_CONNECT_TIMEOUT, - QUIC_CONN_IDLE_TIMEOUT / NGTCP2_MILLISECONDS, - QUIC_CONN_IDLE_TIMEOUT / NGTCP2_MILLISECONDS); + quic_set_hs_timeout(session, QUIC_CONN_IDLE_TIMEOUT / NGTCP2_MILLISECONDS); return kr_ok(); } @@ -919,16 +962,6 @@ static int pl_quic_conn_sess_deinit(struct session2 *session, void *sess_data) session2_timer_stop(session); struct pl_quic_stream_sess_data *s_node; - WALK_LIST_FIRST(s_node, conn->streams) { - struct pl_quic_stream_sess_data *s = - container_of(s_node, struct pl_quic_stream_sess_data, list_node); - rem_node(&s->list_node); - session2_close(s->h.session); - /* These streams die with the connection, stream_close_cb - * will not be called so adjust counters here. */ - --conn->streams_count; - ++conn->finished_streams; - } kr_log_debug(DOQ, "Closing connection, %s useful, served %zu streams\n", conn->finished_streams ? "was" : "wasn't", @@ -967,24 +1000,103 @@ static enum protolayer_event_cb_result pl_quic_conn_event_unwrap( struct session2 *session, void *sess_data) { struct pl_quic_conn_sess_data *conn = sess_data; - if (event == PROTOLAYER_EVENT_CONNECT_TIMEOUT) { - session2_event(conn->h.session->transport.parent, event, conn); - return PROTOLAYER_EVENT_CONSUME; - } + if (event == PROTOLAYER_EVENT_DISCONNECT + || event == PROTOLAYER_EVENT_CLOSE + || event == PROTOLAYER_EVENT_FORCE_CLOSE + || event == PROTOLAYER_EVENT_GENERAL_TIMEOUT + || event == PROTOLAYER_EVENT_CONNECT_TIMEOUT) { + while (queue_len(conn->pending_unwrap) > 0) { + struct session2 *s = + queue_head(conn->pending_unwrap)->h.session; + session2_dec_refs(s); + } + + node_t *s_node; + uint64_t last_stream_id = -1; + bool forced = false; + WALK_LIST_FIRST(s_node, conn->streams) { + struct pl_quic_stream_sess_data *s = + container_of(s_node, + struct pl_quic_stream_sess_data, + list_node); + + /* stream has a pending task to finish. If force + * try force_close, otherwise defer conn termination */ + if (s->stream_id == last_stream_id) { + if (event == PROTOLAYER_EVENT_FORCE_CLOSE + && !forced) { + session2_force_close(s->h.session); + forced = true; + continue; + } + + session2_touch(session); + session2_timer_restart(session); + return PROTOLAYER_EVENT_CONSUME; + } + + last_stream_id = s->stream_id; + + (void)ngtcp2_conn_shutdown_stream(s->conn, 0, s->stream_id, 0); + + session2_close(s->h.session); + /* These streams die with the connection, stream_close_cb + * will not be called so adjust counters here. */ + --conn->streams_count; + ++conn->finished_streams; + } + + if (!EMPTY_LIST(conn->streams)) { + return PROTOLAYER_EVENT_CONSUME; + } - if (event == PROTOLAYER_EVENT_DISCONNECT || - event == PROTOLAYER_EVENT_CLOSE || - event == PROTOLAYER_EVENT_FORCE_CLOSE) { if (!conn->disconnected) { + if (event == PROTOLAYER_EVENT_CLOSE) { + QUIC_SET_CLOSING(conn); + } else { + QUIC_SET_DRAINING(conn); + } conn->disconnected = true; - session2_dec_refs(session); } + return PROTOLAYER_EVENT_PROPAGATE; + } + + return PROTOLAYER_EVENT_PROPAGATE; +} + +static enum protolayer_event_cb_result pl_quic_conn_event_wrap( + enum protolayer_event_type event, void **baton, + struct session2 *session, void *sess_data) +{ + struct pl_quic_conn_sess_data *conn = sess_data; + if (event == PROTOLAYER_EVENT_FORCE_CLOSE + || event == PROTOLAYER_EVENT_CLOSE + || event == PROTOLAYER_EVENT_GENERAL_TIMEOUT + || event == PROTOLAYER_EVENT_CONNECT_TIMEOUT) { + if (baton && *baton) { + struct pl_quic_stream_sess_data *stream = *baton; + rem_node(&stream->list_node); + session2_dec_refs(stream->h.session); + } + + if (EMPTY_LIST(conn->streams) && conn->disconnected) { + /* Connection can be terminated */ + *baton = conn; + return PROTOLAYER_EVENT_PROPAGATE; + } + + /* Close terminates the timer. We failed to close + * due to a stream waiting for a task to finish + * so we restart the timer and try again */ + session2_touch(session); + session2_timer_restart(session); return PROTOLAYER_EVENT_CONSUME; } return PROTOLAYER_EVENT_PROPAGATE; } + __attribute__((constructor)) static void quic_conn_protolayers_init(void) { @@ -995,5 +1107,6 @@ static void quic_conn_protolayers_init(void) .unwrap = pl_quic_conn_unwrap, .wrap = pl_quic_conn_wrap, .event_unwrap = pl_quic_conn_event_unwrap, + .event_wrap = pl_quic_conn_event_wrap, }; } diff --git a/daemon/quic_conn.h b/daemon/quic_conn.h index 38926bff1..83de65aa5 100644 --- a/daemon/quic_conn.h +++ b/daemon/quic_conn.h @@ -102,6 +102,7 @@ struct pl_quic_conn_sess_data { kr_quic_table_t *table_ref; }; +int quic_flush_streams(struct pl_quic_conn_sess_data *conn); int send_special(ngtcp2_version_cid *dec_cids, kr_quic_table_t *table, struct protolayer_iter_ctx *ctx, int action, diff --git a/daemon/quic_demux.c b/daemon/quic_demux.c index bded3fd2e..efba345b8 100644 --- a/daemon/quic_demux.c +++ b/daemon/quic_demux.c @@ -7,6 +7,8 @@ #include "quic_conn.h" #include "quic_demux.h" #include "lib/dnssec.h" +#include "session2.h" +#include "worker.h" #include /* Toggle sending retry for new connections. This is a way to validate the @@ -18,15 +20,14 @@ void kr_quic_table_rem(struct pl_quic_conn_sess_data *conn, kr_quic_table_t *tab static int cmp_expiry_heap_nodes(void *c1, void *c2) { - if (((struct pl_quic_conn_sess_data *)c1)->h.heap_value < - ((struct pl_quic_conn_sess_data *)c2)->h.heap_value) - return -1; - if (((struct pl_quic_conn_sess_data *)c1)->h.heap_value > ((struct pl_quic_conn_sess_data *)c2)->h.heap_value) return 1; - return 0; + /* If the heap_values equal stop the swapping subroutine as there is + * no clear hierarchy. This also protects new connections from being + * terminated by kr_quic_table_sweep right after initialisation. */ + return -1; } static void conn_heap_reschedule(struct pl_quic_conn_sess_data *conn, @@ -131,9 +132,9 @@ void kr_quic_table_sweep(struct kr_quic_table *table, * This is to prevent closing brand new connections which has * crippling effects on the number of answered queries when * conn limits are reached. */ - } else if (table->usage >= table->max_conns && - // c->streams_count <= 0 && - c->finished_streams > 0) { + } else if (table->usage >= table->max_conns + && c->finished_streams > 0 + && !c->disconnected) { quic_doq_error_t doq_error = DOQ_EXCESSIVE_LOAD; send_special(&c->dec_cids, c->table_ref, ctx, QUIC_SEND_CONN_CLOSE, c, c->h.session, @@ -493,8 +494,7 @@ static enum protolayer_event_cb_result pl_quic_demux_event_unwrap( struct pl_quic_conn_sess_data *c = *(struct pl_quic_conn_sess_data **)HHEAD( demux->conn_table->expiry_heap); - kr_quic_table_rem(c, demux->conn_table); - session2_close(c->h.session); + session2_force_close(c->h.session); } session2_dec_refs(session); @@ -524,7 +524,6 @@ static enum protolayer_event_cb_result pl_quic_demux_event_unwrap( if (event == PROTOLAYER_EVENT_DISCONNECT || event == PROTOLAYER_EVENT_CONNECT_TIMEOUT) { - kr_quic_table_rem(conn, demux->conn_table); session2_close(conn->h.session); return PROTOLAYER_EVENT_CONSUME; } @@ -532,6 +531,29 @@ static enum protolayer_event_cb_result pl_quic_demux_event_unwrap( return PROTOLAYER_EVENT_PROPAGATE; } +static enum protolayer_event_cb_result pl_quic_demux_event_wrap( + enum protolayer_event_type event, void **baton, + struct session2 *session, void *sess_data) +{ + if (event == PROTOLAYER_EVENT_FORCE_CLOSE + || event == PROTOLAYER_EVENT_CLOSE + || event == PROTOLAYER_EVENT_GENERAL_TIMEOUT + || event == PROTOLAYER_EVENT_CONNECT_TIMEOUT) { + kr_require(baton && *baton); + struct pl_quic_demux_sess_data *demux = sess_data; + struct pl_quic_conn_sess_data *conn = *baton; + kr_quic_table_rem(conn, demux->conn_table); + + kr_require(EMPTY_LIST(conn->streams)); + session2_dec_refs(conn->h.session); + + return PROTOLAYER_EVENT_CONSUME; + } + + return PROTOLAYER_EVENT_PROPAGATE; +} + + __attribute__((constructor)) static void quic_demux_protolayers_init(void) { @@ -542,5 +564,6 @@ static void quic_demux_protolayers_init(void) .sess_deinit = pl_quic_demux_sess_deinit, .unwrap = pl_quic_demux_unwrap, .event_unwrap = pl_quic_demux_event_unwrap, + .event_wrap = pl_quic_demux_event_wrap, }; } diff --git a/daemon/quic_stream.c b/daemon/quic_stream.c index 3050d744c..fa6c6f2c5 100644 --- a/daemon/quic_stream.c +++ b/daemon/quic_stream.c @@ -4,6 +4,9 @@ #include "lib/resolve.h" #include "quic_common.h" +#include "quic_conn.h" +#include "session2.h" +#include #include "quic_stream.h" /* forward declaration */ @@ -24,6 +27,22 @@ static enum protolayer_iter_cb_result pl_quic_stream_unwrap(void *sess_data, return protolayer_continue(ctx); } +void kr_quic_stream_mark_sent(struct pl_quic_stream_sess_data *stream, + size_t amount_sent) +{ + stream->unsent_offset += amount_sent; + kr_assert(stream->unsent_offset <= stream->unsent_obuf->len); + if (stream->unsent_offset == stream->unsent_obuf->len) { + stream->unsent_offset = 0; + stream->unsent_obuf = + (struct kr_quic_obuf *)stream->unsent_obuf->node.next; + // already behind the tail of list + if (stream->unsent_obuf->node.next == NULL) { + stream->unsent_obuf = NULL; + } + } +} + uint8_t *kr_quic_stream_add_data(struct pl_quic_stream_sess_data *s, uint8_t *data, size_t len) { @@ -54,39 +73,74 @@ static enum protolayer_iter_cb_result pl_quic_stream_wrap(void *sess_data, void *iter_data, struct protolayer_iter_ctx *ctx) { struct pl_quic_stream_sess_data *stream = sess_data; - ngtcp2_ssize sent = 0; - - if (unlikely(kr_quic_stream_add_data(stream, ctx->payload.iovec.iov[1].iov_base, - ctx->payload.iovec.iov[1].iov_len) == NULL)) { - return kr_error(ENOMEM); - } - ctx->payload = protolayer_payload_wire_buf(&stream->outbuf, false); + /* new data only comes as a iovec, flushing the stream with no new data + * to append uses wire_buf */ + if (likely(ctx->payload.type == PROTOLAYER_PAYLOAD_IOVEC)) { + if (unlikely(kr_quic_stream_add_data(stream, ctx->payload.iovec.iov[1].iov_base, + ctx->payload.iovec.iov[1].iov_len) == NULL)) { + return kr_error(ENOMEM); + } - size_t uf = stream->unsent_offset; - struct kr_quic_obuf *uo = stream->unsent_obuf; - if (uo == NULL) { - return protolayer_break(ctx, kr_ok()); + ctx->payload = protolayer_payload_wire_buf(&stream->outbuf, + false); } if (wire_buf_data_length(&stream->outbuf) != 0) { wire_buf_reset(&stream->outbuf); } - bool fin = (((node_t *)uo->node.next)->next == NULL)/* && ignore_last == 0 */; - int nwrite = send_stream(stream, ctx, uo->buf + uf, uo->len - uf - 0/* ignore_last*/, - fin, &sent); + ngtcp2_ssize sent; + uint16_t sent_msgs = 0; + do { + sent = 0; + size_t uf = stream->unsent_offset; + struct kr_quic_obuf *uo = stream->unsent_obuf; + if (uo == NULL) { + if (sent_msgs > 0) { + break; + } + + protolayer_break(ctx, kr_ok()); + } - if (nwrite <= 0) { - if (nwrite == NGTCP2_ERR_NOMEM) { - kr_log_error(DOQ, "Insufficient memory available\n"); - return protolayer_break(ctx, kr_error(ENOMEM)); + if (wire_buf_data_length(ctx->payload.wire_buf) != 0) { + wire_buf_reset(ctx->payload.wire_buf); } - return protolayer_break(ctx, kr_ok()); - } + bool fin = (((node_t *)uo->node.next)->next == NULL)/* && ignore_last == 0 */; + int nwrite = send_stream(stream, ctx, uo->buf + uf, uo->len - uf - 0/* ignore_last*/, + fin, &sent); + if (nwrite < 0) { + if (nwrite == NGTCP2_ERR_NOMEM) { + kr_log_error(DOQ, "Insufficient memory available\n"); + } - return protolayer_continue(ctx); + return protolayer_break(ctx, kr_ok()); + } + sent_msgs++; + + if (sent > 0) { + kr_quic_stream_mark_sent(stream, sent); + } + + protolayer_finished_cb finished_cb = NULL; + void *finished_baton = NULL; + if (sent > 0 && stream->unsent_obuf == NULL && stream->h.session->outgoing + && false) { + finished_cb = ctx->finished_cb; + finished_baton = ctx->finished_cb_baton; + } + + session2_wrap_after(stream->h.session->transport.parent, + PROTOLAYER_TYPE_QUIC_CONN, + ctx->payload, + ctx->comm, + finished_cb, + finished_baton); + } while (sent > 0 && sent_msgs < QUIC_MAX_SEND_PER_RECV); + + return protolayer_break(ctx, kr_ok()); } static int send_stream(struct pl_quic_stream_sess_data *stream, @@ -94,7 +148,7 @@ static int send_stream(struct pl_quic_stream_sess_data *stream, size_t len, bool fin, ngtcp2_ssize *sent) { if (!stream->conn_ref || !QUIC_CAN_SEND(stream->conn_ref)) { - return protolayer_break(ctx, kr_ok()); + return 0; } int64_t stream_id = stream->stream_id; @@ -131,22 +185,6 @@ static int send_stream(struct pl_quic_stream_sess_data *stream, return nwrite; } -void kr_quic_stream_mark_sent(struct pl_quic_stream_sess_data *stream, - size_t amount_sent) -{ - stream->unsent_offset += amount_sent; - kr_assert(stream->unsent_offset <= stream->unsent_obuf->len); - if (stream->unsent_offset == stream->unsent_obuf->len) { - stream->unsent_offset = 0; - stream->unsent_obuf = - (struct kr_quic_obuf *)stream->unsent_obuf->node.next; - // already behind the tail of list - if (stream->unsent_obuf->node.next == NULL) { - stream->unsent_obuf = NULL; - } - } -} - static int pl_quic_stream_sess_init(struct session2 *session, void *sess_data, void *param) { @@ -195,7 +233,13 @@ void kr_quic_stream_ack_data(struct pl_quic_stream_sess_data *stream, static int pl_quic_stream_sess_deinit(struct session2 *session, void *sess_data) { struct pl_quic_stream_sess_data *stream = sess_data; - ngtcp2_conn_shutdown_stream(stream->conn, 0, stream->stream_id, 0); + if (stream->conn) { + ngtcp2_conn_shutdown_stream(stream->conn, 0, stream->stream_id, 0); + } + + if (!session2_tasklist_is_empty(session)) { + session2_tasklist_finalize_expired(session); + } kr_quic_stream_ack_data(stream, stream->stream_id, SIZE_MAX, false); wire_buf_deinit(&stream->pers_inbuf); wire_buf_deinit(&stream->outbuf); @@ -206,8 +250,33 @@ static enum protolayer_event_cb_result pl_quic_stream_event_unwrap( enum protolayer_event_type event, void **baton, struct session2 *session, void *sess_data) { - if (event == PROTOLAYER_EVENT_CLOSE) { - session2_dec_refs(session); + if (event == PROTOLAYER_EVENT_CLOSE + || event == PROTOLAYER_EVENT_FORCE_CLOSE) { + struct pl_quic_stream_sess_data *stream = + protolayer_sess_data_get_proto(session, + PROTOLAYER_TYPE_QUIC_STREAM); + kr_quic_stream_ack_data(stream, + stream->stream_id, + SIZE_MAX, + false); + + stream->conn = NULL; + stream->conn_ref = NULL; + } + + return PROTOLAYER_EVENT_PROPAGATE; +} + +static enum protolayer_event_cb_result pl_quic_stream_event_wrap( + enum protolayer_event_type event, void **baton, + struct session2 *session, void *sess_data) +{ + if (event == PROTOLAYER_EVENT_CLOSE + || event == PROTOLAYER_EVENT_FORCE_CLOSE) { + if (session2_is_empty(session)) { + *baton = sess_data; + return PROTOLAYER_EVENT_PROPAGATE; + } return PROTOLAYER_EVENT_CONSUME; } @@ -231,6 +300,7 @@ static void quic_conn_protolayers_init(void) .unwrap = pl_quic_stream_unwrap, .wrap = pl_quic_stream_wrap, .event_unwrap = pl_quic_stream_event_unwrap, + .event_wrap = pl_quic_stream_event_wrap, .request_init = pl_quic_stream_request_init, }; } diff --git a/daemon/worker.c b/daemon/worker.c index f4e7f5c17..63d07a644 100644 --- a/daemon/worker.c +++ b/daemon/worker.c @@ -1955,6 +1955,9 @@ static enum protolayer_event_cb_result pl_dns_stream_connected( return PROTOLAYER_EVENT_PROPAGATE; stream->connected = true; + if (session->proto == KR_PROTO_DOQ_STREAM) { + return PROTOLAYER_EVENT_CONSUME; + } struct sockaddr *peer = session2_get_peer(session); if (session->outgoing && worker_del_tcp_waiting(peer) != 0) { @@ -2030,8 +2033,10 @@ static enum protolayer_event_cb_result pl_dns_stream_disconnected( struct session2 *session, struct pl_dns_stream_sess_data *stream) { struct sockaddr *peer = session2_get_peer(session); - worker_del_tcp_waiting(peer); - worker_del_tcp_connected(peer); + if (session->proto != KR_PROTO_DOQ_STREAM) { + worker_del_tcp_waiting(peer); + worker_del_tcp_connected(peer); + } if (!stream->connected) return PROTOLAYER_EVENT_PROPAGATE;