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.
/* Copyright (C) CZ.NIC, z.s.p.o. <knot-resolver@labs.nic.cz>
* SPDX-License-Identifier: GPL-3.0-or-later
*/
+
#include <ngtcp2/ngtcp2.h>
+#include <uv.h>
#include "contrib/openbsd/siphash.h"
#include "lib/dnssec.h"
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)
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);
#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),
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,
*/
#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 <gnutls/gnutls.h>
#include <libknot/wire.h>
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();
}
msg, sizeof(msg) - 1);
return NGTCP2_ERR_CALLBACK_FAILURE;
}
+ session2_inc_refs(stream->h.session);
queue_push(conn->pending_unwrap, stream);
}
++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;
}
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;
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)
.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();
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,
}
mm_free(&ctx->pool, ctx->payload.wire_buf);
- ctx->payload.wire_buf = save;
+ ctx->payload.wire_buf = save_wb;
return ret;
}
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,
}
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);
}
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 };
&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);
}
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();
}
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",
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)
{
.unwrap = pl_quic_conn_unwrap,
.wrap = pl_quic_conn_wrap,
.event_unwrap = pl_quic_conn_event_unwrap,
+ .event_wrap = pl_quic_conn_event_wrap,
};
}
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,
#include "quic_conn.h"
#include "quic_demux.h"
#include "lib/dnssec.h"
+#include "session2.h"
+#include "worker.h"
#include <stdlib.h>
/* Toggle sending retry for new connections. This is a way to validate the
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,
* 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,
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);
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;
}
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)
{
.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,
};
}
#include "lib/resolve.h"
#include "quic_common.h"
+#include "quic_conn.h"
+#include "session2.h"
+#include <stdint.h>
#include "quic_stream.h"
/* forward declaration */
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)
{
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,
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;
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)
{
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);
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;
}
.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,
};
}
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) {
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;
projects / thirdparty / knot-resolver.git / commitdiff
