src/knot/query/requestor.h
src/knot/server/dthreads.c
src/knot/server/dthreads.h
+src/knot/server/handler.c
+src/knot/server/handler.h
src/knot/server/proxyv2.c
src/knot/server/proxyv2.h
src/knot/server/quic-handler.c
tests/tap/float.h
tests/tap/macros.h
tests/tap/runtests.c
-tests/utils/test_cert.c
tests/utils/test_lookup.c
knot/journal/serialization.h \
knot/server/dthreads.c \
knot/server/dthreads.h \
+ knot/server/handler.c \
+ knot/server/handler.h \
knot/server/proxyv2.c \
knot/server/proxyv2.h \
knot/server/server.c \
--- /dev/null
+/* Copyright (C) 2023 CZ.NIC, z.s.p.o. <knot-dns@labs.nic.cz>
+
+ 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 3 of the License, or
+ (at your option) any later version.
+
+ This program 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 General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "knot/server/handler.h"
+
+#include "contrib/time.h"
+#include "contrib/ucw/mempool.h"
+#include "knot/common/log.h"
+#include "knot/server/proxyv2.h"
+
+void handle_query(knotd_qdata_params_t *params, knot_layer_t *layer,
+ const struct iovec *payload, struct sockaddr_storage *proxied_remote)
+{
+ knot_layer_begin(layer, params);
+
+ knot_pkt_t *query = knot_pkt_new(payload->iov_base, payload->iov_len, layer->mm);
+ int ret = knot_pkt_parse(query, 0);
+ if (ret != KNOT_EOK && query->parsed > 0) { // parsing failed (e.g. 2x OPT)
+ if (params->proto == KNOTD_QUERY_PROTO_UDP &&
+ proxyv2_header_strip(&query, params->remote, proxied_remote) == KNOT_EOK) {
+ assert(proxied_remote);
+ params->remote = proxied_remote;
+ } else {
+ query->parsed--; // artificially decreasing "parsed" leads to FORMERR
+ }
+ }
+
+ knot_layer_consume(layer, query);
+}
+
+void handle_finish(knot_layer_t *layer)
+{
+ knot_layer_finish(layer);
+
+ // Flush per-query memory (including query and answer packets).
+ mp_flush(layer->mm->ctx);
+}
+
+void handle_udp_reply(knotd_qdata_params_t *params, knot_layer_t *layer,
+ struct iovec *rx, struct iovec *tx,
+ struct sockaddr_storage *proxied_remote)
+{
+ handle_query(params, layer, rx, proxied_remote);
+
+ knot_pkt_t *ans = knot_pkt_new(tx->iov_base, tx->iov_len, layer->mm);
+
+ while (active_state(layer->state)) {
+ knot_layer_produce(layer, ans);
+ }
+
+ // Send response only if finished successfully.
+ if (layer->state == KNOT_STATE_DONE) {
+ tx->iov_len = ans->size;
+ } else {
+ tx->iov_len = 0;
+ }
+
+ handle_finish(layer);
+}
+
+#ifdef ENABLE_QUIC
+static void handle_quic_stream(knot_xquic_conn_t *conn, int64_t stream_id, struct iovec *inbuf,
+ knot_layer_t *layer, knotd_qdata_params_t *params, uint8_t *ans_buf,
+ size_t ans_buf_size)
+{
+ // Consume the query.
+ handle_query(params, layer, inbuf, NULL);
+
+ // Process the reply.
+ knot_pkt_t *ans = knot_pkt_new(ans_buf, ans_buf_size, layer->mm);
+ while (active_state(layer->state)) {
+ knot_layer_produce(layer, ans);
+ if (!send_state(layer->state)) {
+ continue;
+ }
+ if (knot_xquic_stream_add_data(conn, stream_id, ans->wire, ans->size) == NULL) {
+ break;
+ }
+ }
+
+ handle_finish(layer);
+}
+
+void handle_quic_streams(knot_xquic_conn_t *conn, knotd_qdata_params_t *params,
+ knot_layer_t *layer, void *msg)
+{
+ uint8_t ans_buf[KNOT_WIRE_MAX_PKTSIZE];
+
+ int64_t stream_id;
+ knot_xquic_stream_t *stream;
+
+ while (conn != NULL && (stream = knot_xquic_stream_get_process(conn, &stream_id)) != NULL) {
+ assert(stream->inbuf_fin != NULL);
+ assert(stream->inbuf_fin->iov_len > 0);
+ if (msg) {
+#ifdef ENABLE_XDP
+ params_xdp_update(params, KNOTD_QUERY_PROTO_QUIC, msg,
+ knot_xquic_conn_rtt(conn), conn->tls_session);
+#endif // ENABLE_XDP
+ } else {
+ params_update(params, knot_xquic_conn_rtt(conn), conn->tls_session);
+ }
+ handle_quic_stream(conn, stream_id, stream->inbuf_fin, layer, params,
+ ans_buf, sizeof(ans_buf));
+ free(stream->inbuf_fin);
+ stream->inbuf_fin = NULL;
+ }
+}
+#endif // ENABLE_QUIC
+
+void log_swept(knot_sweep_stats_t *stats, bool tcp)
+{
+ struct timespec now = time_now();
+ uint64_t sec = now.tv_sec + now.tv_nsec / 1000000000;
+ if (sec - stats->last_log <= 9 || (stats->total == 0)) {
+ return;
+ }
+
+ const char *proto = tcp ? "TCP" : "QUIC";
+
+ uint32_t timedout = stats->counters[KNOT_SWEEP_CTR_TIMEOUT];
+ uint32_t limit_conn = stats->counters[KNOT_SWEEP_CTR_LIMIT_CONN];
+ uint32_t limit_ibuf = stats->counters[KNOT_SWEEP_CTR_LIMIT_IBUF];
+ uint32_t limit_obuf = stats->counters[KNOT_SWEEP_CTR_LIMIT_OBUF];
+
+ if (tcp || stats->total != timedout) {
+ log_notice("%s, connection sweep, closed %u, count limit %u, inbuf limit %u, outbuf limit %u",
+ proto, timedout, limit_conn, limit_ibuf, limit_obuf);
+ } else {
+ log_debug("%s, timed out connections %u", proto, timedout);
+ }
+
+ knot_sweep_stats_reset(stats);
+ stats->last_log = sec;
+}
--- /dev/null
+/* Copyright (C) 2023 CZ.NIC, z.s.p.o. <knot-dns@labs.nic.cz>
+
+ 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 3 of the License, or
+ (at your option) any later version.
+
+ This program 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 General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include "knot/include/module.h"
+#include "knot/query/layer.h"
+#include "knot/server/server.h"
+#include "libknot/xdp/tcp_iobuf.h"
+
+#ifdef ENABLE_QUIC
+#include "libknot/quic/quic.h"
+#endif // ENABLE_QUIC
+
+#ifdef ENABLE_XDP
+#include "libknot/xdp.h"
+#endif // ENABLE_XDP
+
+#define QUIC_MAX_SEND_PER_RECV 4
+#define QUIC_IBUFS_PER_CONN 512 /* Heuristic value: this means that e.g. for 100k allowed
+ QUIC conns, we will limit total size of input buffers to 50 MiB. */
+
+inline static knotd_qdata_params_t params_init(knotd_query_proto_t proto,
+ const void *remote, const void *local,
+ int sock, server_t *server,
+ unsigned thread_id)
+
+{
+ knotd_qdata_params_t params = {
+ .proto = proto,
+ .remote = (const struct sockaddr_storage *)remote,
+ .local = (const struct sockaddr_storage *)local,
+ .socket = sock,
+ .server = server,
+ .thread_id = thread_id
+ };
+
+ return params;
+}
+
+inline static void params_update(knotd_qdata_params_t *params, uint32_t rtt,
+ struct gnutls_session_int *session)
+{
+ params->measured_rtt = rtt;
+ params->session = session;
+}
+
+#ifdef ENABLE_XDP
+inline static knotd_qdata_params_t params_xdp_init(int sock, server_t *server,
+ unsigned thread_id)
+{
+ knotd_qdata_params_t params = {
+ .socket = sock,
+ .server = server,
+ .thread_id = thread_id
+ };
+
+ return params;
+}
+
+inline static void params_xdp_update(knotd_qdata_params_t *params,
+ knotd_query_proto_t proto,
+ struct knot_xdp_msg *msg,
+ uint32_t rtt,
+ struct gnutls_session_int *session)
+{
+ params->proto = proto;
+ params->remote = (struct sockaddr_storage *)&msg->ip_from;
+ params->local = (struct sockaddr_storage *)&msg->ip_to;
+ params->xdp_msg = msg;
+ params->measured_rtt = rtt;
+ params->session = session;
+}
+#endif // ENABLE_XDP
+
+inline static bool active_state(int state)
+{
+ return (state == KNOT_STATE_PRODUCE || state == KNOT_STATE_FAIL);
+}
+
+inline static bool send_state(int state)
+{
+ return (state != KNOT_STATE_FAIL && state != KNOT_STATE_NOOP);
+}
+
+void handle_query(knotd_qdata_params_t *params, knot_layer_t *layer,
+ const struct iovec *payload, struct sockaddr_storage *proxied_remote);
+
+void handle_finish(knot_layer_t *layer);
+
+void handle_udp_reply(knotd_qdata_params_t *params, knot_layer_t *layer,
+ struct iovec *rx, struct iovec *tx,
+ struct sockaddr_storage *proxied_remote);
+
+#ifdef ENABLE_QUIC
+void handle_quic_streams(knot_xquic_conn_t *conn, knotd_qdata_params_t *params,
+ knot_layer_t *layer, void *msg);
+#endif // ENABLE_QUIC
+
+void log_swept(knot_sweep_stats_t *stats, bool tcp);
#include <netinet/in.h>
#include <string.h>
-#include <time.h>
#include "contrib/macros.h"
-#include "contrib/ucw/mempool.h"
-#include "knot/common/log.h"
+#include "knot/server/handler.h"
#include "knot/server/quic-handler.h"
#include "knot/server/server.h"
#include "libknot/quic/quic.h"
-#include "libknot/xdp/eth.h"
#include "libknot/xdp/tcp_iobuf.h"
-#define QUIC_MAX_SEND_PER_RECV 4 // NOTE: also in xdp-handler.c
-#define QUIC_IBUFS_PER_CONN 512 /* Heuristic value: this means that e.g. for 100k allowed
- QUIC conns, we will limit total size of input buffers to 50 MiB. */
-
-static bool quic_active_state(int state)
-{
- return (state == KNOT_STATE_PRODUCE || state == KNOT_STATE_FAIL);
-}
-
-static bool quic_send_state(int state)
-{
- return (state != KNOT_STATE_FAIL && state != KNOT_STATE_NOOP);
-}
-
-static void log_closed(knot_sweep_stats_t *stats)
-{
- struct timespec now = time_now();
- uint64_t sec = now.tv_sec + now.tv_nsec / 1000000000;
- if (sec - stats->last_log <= 9 || (stats->total == 0)) {
- return;
- }
-
- const char *proto = "QUIC";
-
- uint32_t timedout = stats->counters[KNOT_SWEEP_CTR_TIMEOUT];
- uint32_t limit_conn = stats->counters[KNOT_SWEEP_CTR_LIMIT_CONN];
- uint32_t limit_ibuf = stats->counters[KNOT_SWEEP_CTR_LIMIT_IBUF];
- uint32_t limit_obuf = stats->counters[KNOT_SWEEP_CTR_LIMIT_OBUF];
-
- if (stats->total != timedout) {
- log_notice("%s, connection sweep, closed %u, count limit %u, inbuf limit %u, outbuf limit %u",
- proto, timedout, limit_conn, limit_ibuf, limit_obuf);
- } else {
- log_debug("%s, timed out connections %u", proto, timedout);
- }
-
- knot_sweep_stats_reset(stats);
- stats->last_log = sec;
-}
-
-static void handle_quic_init(knotd_qdata_params_t *params, knot_layer_t *layer,
- const struct sockaddr_storage *remote,
- const struct sockaddr_storage *local,
- const struct iovec *payload)
-{
- params->remote = remote;
- params->local = local;
- params->xdp_msg = NULL;
-
- knot_layer_begin(layer, params);
-
- knot_pkt_t *query = knot_pkt_new(payload->iov_base, payload->iov_len, layer->mm);
- int ret = knot_pkt_parse(query, 0);
- if (ret != KNOT_EOK && query->parsed > 0) { // parsing failed (e.g. 2x OPT)
- query->parsed--; // artificially decreasing "parsed" leads to FORMERR
- }
- knot_layer_consume(layer, query);
-}
-
-static void handle_quic_finish(knot_layer_t *layer)
-{
- knot_layer_finish(layer);
-
- // Flush per-query memory (including query and answer packets).
- mp_flush(layer->mm->ctx);
-}
-
-static void handle_quic_stream(knot_xquic_conn_t *conn, int64_t stream_id, struct iovec *inbuf,
- knot_layer_t *layer, knotd_qdata_params_t *params, uint8_t *ans_buf,
- size_t ans_buf_size, const struct sockaddr_storage *remote,
- const struct sockaddr_storage *local)
-{
- // Consume the query.
- handle_quic_init(params, layer, remote, local, inbuf);
- params->measured_rtt = knot_xquic_conn_rtt(conn);
-
- // Process the reply.
- knot_pkt_t *ans = knot_pkt_new(ans_buf, ans_buf_size, layer->mm);
- while (quic_active_state(layer->state)) {
- knot_layer_produce(layer, ans);
- if (!quic_send_state(layer->state)) {
- continue;
- }
- if (knot_xquic_stream_add_data(conn, stream_id, ans->wire, ans->size) == NULL) {
- break;
- }
- }
-
- handle_quic_finish(layer);
-}
-
static int uq_alloc_reply(knot_quic_reply_t *r)
{
r->out_payload->iov_len = KNOT_WIRE_MAX_PKTSIZE;
.send_reply = uq_send_reply,
.free_reply = uq_free_reply
};
- knot_xquic_conn_t *conn = NULL;
+ knot_xquic_conn_t *conn = NULL;
(void)knot_quic_handle(table, &rpl, idle_close, &conn);
- int64_t stream_id;
- knot_xquic_stream_t *stream;
-
- while (conn != NULL && (stream = knot_xquic_stream_get_process(conn, &stream_id)) != NULL) {
- assert(stream->inbuf_fin != NULL);
- assert(stream->inbuf_fin->iov_len > 0);
- params->session = conn->tls_session;
- handle_quic_stream(conn, stream_id, stream->inbuf_fin, layer, params,
- tx->iov_base, tx->iov_len, params->remote, params->local); // NOTE: tx is used here just as temporary buffer
- free(stream->inbuf_fin);
- stream->inbuf_fin = NULL;
- }
+ handle_quic_streams(conn, params, layer, NULL);
(void)knot_quic_send(table, conn, &rpl, QUIC_MAX_SEND_PER_RECV, false);
void quic_sweep(knot_xquic_table_t *table, knot_sweep_stats_t *stats)
{
(void)knot_xquic_table_sweep(table, stats);
- log_closed(stats);
+ log_swept(stats, false);
}
void *quic_make_table(struct server *server)
{
conf_t *pconf = conf();
- size_t udp_pl = MIN(pconf->cache.srv_udp_max_payload_ipv4, pconf->cache.srv_udp_max_payload_ipv6);
+ size_t udp_pl = MIN(pconf->cache.srv_udp_max_payload_ipv4,
+ pconf->cache.srv_udp_max_payload_ipv6);
- size_t quic_max_conns = pconf->cache.srv_quic_max_clients / pconf->cache.srv_udp_threads;
+ size_t quic_max_conns = pconf->cache.srv_quic_max_clients /
+ pconf->cache.srv_udp_threads;
size_t quic_max_inbufs= quic_max_conns * QUIC_IBUFS_PER_CONN;
size_t quic_max_obufs = pconf->cache.srv_quic_obuf_max_size;
- return knot_xquic_table_new(quic_max_conns, quic_max_inbufs, quic_max_obufs, udp_pl, server->quic_creds);
+ return knot_xquic_table_new(quic_max_conns, quic_max_inbufs, quic_max_obufs,
+ udp_pl, server->quic_creds);
}
void quic_unmake_table(knot_xquic_table_t *table)
#include <sys/uio.h>
#endif // HAVE_SYS_UIO_H
+#include "knot/server/handler.h"
#include "knot/server/server.h"
#include "knot/server/tcp-handler.h"
#include "knot/common/log.h"
return FDSET_SWEEP;
}
-static bool tcp_active_state(int state)
-{
- return (state == KNOT_STATE_PRODUCE || state == KNOT_STATE_FAIL);
-}
-
-static bool tcp_send_state(int state)
-{
- return (state != KNOT_STATE_FAIL && state != KNOT_STATE_NOOP);
-}
-
static void tcp_log_error(const struct sockaddr_storage *ss, const char *operation, int ret)
{
/* Don't log ECONN as it usually means client closed the connection. */
const sockaddr_t *local, struct iovec *rx, struct iovec *tx)
{
/* Create query processing parameter. */
- knotd_qdata_params_t params = {
- .proto = KNOTD_QUERY_PROTO_TCP,
- .remote = (const struct sockaddr_storage *)remote,
- .local = (const struct sockaddr_storage *)local,
- .socket = fd,
- .server = tcp->server,
- .thread_id = tcp->thread_id
- };
+ knotd_qdata_params_t params = params_init(KNOTD_QUERY_PROTO_TCP, remote, local,
+ fd, tcp->server, tcp->thread_id);
rx->iov_len = KNOT_WIRE_MAX_PKTSIZE;
tx->iov_len = KNOT_WIRE_MAX_PKTSIZE;
return KNOT_EOF;
}
- /* Initialize processing layer. */
- knot_layer_begin(&tcp->layer, ¶ms);
-
- /* Create packets. */
- knot_pkt_t *ans = knot_pkt_new(tx->iov_base, tx->iov_len, tcp->layer.mm);
- knot_pkt_t *query = knot_pkt_new(rx->iov_base, rx->iov_len, tcp->layer.mm);
-
- /* Input packet. */
- int ret = knot_pkt_parse(query, 0);
- if (ret != KNOT_EOK && query->parsed > 0) { // parsing failed (e.g. 2x OPT)
- query->parsed--; // artificially decreasing "parsed" leads to FORMERR
- }
- knot_layer_consume(&tcp->layer, query);
+ handle_query(¶ms, &tcp->layer, rx, NULL);
/* Resolve until NOOP or finished. */
- while (tcp_active_state(tcp->layer.state)) {
+ knot_pkt_t *ans = knot_pkt_new(tx->iov_base, tx->iov_len, tcp->layer.mm);
+ while (active_state(tcp->layer.state)) {
knot_layer_produce(&tcp->layer, ans);
/* Send, if response generation passed and wasn't ignored. */
- if (ans->size > 0 && tcp_send_state(tcp->layer.state)) {
+ if (ans->size > 0 && send_state(tcp->layer.state)) {
int sent = net_dns_tcp_send(fd, ans->wire, ans->size,
tcp->io_timeout, NULL);
if (sent != ans->size) {
tcp_log_error(params.remote, "send", sent);
- ret = KNOT_EOF;
- break;
+ handle_finish(&tcp->layer);
+ return KNOT_EOF;
}
}
}
- /* Reset after processing. */
- knot_layer_finish(&tcp->layer);
-
- /* Flush per-query memory (including query and answer packets). */
- mp_flush(tcp->layer.mm->ctx);
+ handle_finish(&tcp->layer);
- return ret;
+ return KNOT_EOK;
}
static void tcp_event_accept(tcp_context_t *tcp, unsigned i, const iface_t *iface)
#include "knot/common/fdset.h"
#include "knot/nameserver/process_query.h"
#include "knot/query/layer.h"
-#include "knot/server/proxyv2.h"
+#include "knot/server/handler.h"
#include "knot/server/server.h"
#ifdef ENABLE_QUIC
#include "knot/server/quic-handler.h"
#endif // ENABLE_QUIC
} udp_context_t;
-static bool udp_state_active(int state)
-{
- return (state == KNOT_STATE_PRODUCE || state == KNOT_STATE_FAIL);
-}
-
-#define PARAMS_INIT(is_quic, fd, remote_ss, local_ss, udp) \
- knotd_qdata_params_t params = { \
- .proto = is_quic ? KNOTD_QUERY_PROTO_QUIC : KNOTD_QUERY_PROTO_UDP, \
- .remote = (const struct sockaddr_storage *)remote_ss, \
- .local = (const struct sockaddr_storage *)local_ss, \
- .socket = fd, \
- .server = udp->server, \
- .thread_id = udp->thread_id \
- };
-
static void udp_handler(udp_context_t *udp, knotd_qdata_params_t *params,
struct iovec *rx, struct iovec *tx)
{
+ // Prepare a reply.
struct sockaddr_storage proxied_remote;
-
- /* Start query processing. */
- knot_layer_begin(&udp->layer, params);
-
- /* Create packets. */
- knot_pkt_t *query = knot_pkt_new(rx->iov_base, rx->iov_len, udp->layer.mm);
- knot_pkt_t *ans = knot_pkt_new(tx->iov_base, tx->iov_len, udp->layer.mm);
-
- /* Input packet. */
- int ret = knot_pkt_parse(query, 0);
- if (ret != KNOT_EOK && query->parsed > 0) {
- ret = proxyv2_header_strip(&query, params->remote, &proxied_remote);
- if (ret == KNOT_EOK) {
- params->remote = &proxied_remote;
- } else {
- query->parsed--; // artificially decreasing "parsed" leads to FORMERR
- }
- }
- knot_layer_consume(&udp->layer, query);
-
- /* Process answer. */
- while (udp_state_active(udp->layer.state)) {
- knot_layer_produce(&udp->layer, ans);
- }
-
- /* Send response only if finished successfully. */
- if (udp->layer.state == KNOT_STATE_DONE) {
- tx->iov_len = ans->size;
- } else {
- tx->iov_len = 0;
- }
-
- /* Reset after processing. */
- knot_layer_finish(&udp->layer);
-
- /* Flush per-query memory (including query and answer packets). */
- mp_flush(udp->layer.mm->ctx);
+ handle_udp_reply(params, &udp->layer, rx, tx, &proxied_remote);
}
typedef struct {
&ctx->local, iface);
/* Process received pkt. */
- PARAMS_INIT(iface->quic, rq->fd, &rq->addr, local, ctx);
+ knotd_qdata_params_t params = params_init(
+ iface->quic ? KNOTD_QUERY_PROTO_QUIC : KNOTD_QUERY_PROTO_UDP,
+ &rq->addr, local, rq->fd, ctx->server, ctx->thread_id);
if (iface->quic) {
#ifdef ENABLE_QUIC
quic_handler(¶ms, &ctx->layer, ctx->quic_idle_close,
/* Update output message control buffer. */
const sockaddr_t *local = udp_pktinfo_handle(rx, tx, &ctx->local, iface);
- PARAMS_INIT(iface->quic, rq->fd, &rq->addrs[i], local, ctx);
+ knotd_qdata_params_t params = params_init(
+ iface->quic ? KNOTD_QUERY_PROTO_QUIC : KNOTD_QUERY_PROTO_UDP,
+ &rq->addrs[i], local, rq->fd, ctx->server, ctx->thread_id);
if (iface->quic) {
#ifdef ENABLE_QUIC
quic_handler(¶ms, &ctx->layer, ctx->quic_idle_close,
#include <stdlib.h>
#include <urcu.h>
+#include "knot/server/handler.h"
#include "knot/server/xdp-handler.h"
#include "knot/common/log.h"
-#include "knot/server/proxyv2.h"
#include "knot/server/server.h"
-#include "contrib/sockaddr.h"
-#include "contrib/time.h"
-#include "contrib/ucw/mempool.h"
-#include "libknot/endian.h"
#include "libknot/error.h"
#ifdef ENABLE_QUIC
#include "libknot/quic/quic.h"
#include "libknot/xdp/tcp.h"
#include "libknot/xdp/tcp_iobuf.h"
-#define QUIC_MAX_SEND_PER_RECV 4
-#define QUIC_IBUFS_PER_CONN 512 /* Heuristic value: this means that e.g. for 100k allowed
- QUIC conns, we will limit total size of input buffers to 50 MiB. */
-
typedef struct xdp_handle_ctx {
knot_xdp_socket_t *sock;
knot_xdp_msg_t msg_recv[XDP_BATCHLEN];
knot_sweep_stats_t tcp_closed;
} xdp_handle_ctx_t;
-static bool udp_state_active(int state)
-{
- return (state == KNOT_STATE_PRODUCE || state == KNOT_STATE_FAIL);
-}
-
-static bool tcp_active_state(int state)
-{
- return (state == KNOT_STATE_PRODUCE || state == KNOT_STATE_FAIL);
-}
-
-static bool tcp_send_state(int state)
-{
- return (state != KNOT_STATE_FAIL && state != KNOT_STATE_NOOP);
-}
-
-static void log_closed(knot_sweep_stats_t *stats, bool tcp)
-{
- struct timespec now = time_now();
- uint64_t sec = now.tv_sec + now.tv_nsec / 1000000000;
- if (sec - stats->last_log <= 9 || (stats->total == 0)) {
- return;
- }
-
- const char *proto = tcp ? "TCP" : "QUIC";
-
- uint32_t timedout = stats->counters[KNOT_SWEEP_CTR_TIMEOUT];
- uint32_t limit_conn = stats->counters[KNOT_SWEEP_CTR_LIMIT_CONN];
- uint32_t limit_ibuf = stats->counters[KNOT_SWEEP_CTR_LIMIT_IBUF];
- uint32_t limit_obuf = stats->counters[KNOT_SWEEP_CTR_LIMIT_OBUF];
-
- if (tcp || stats->total != timedout) {
- log_notice("%s, connection sweep, closed %u, count limit %u, inbuf limit %u, outbuf limit %u",
- proto, timedout, limit_conn, limit_ibuf, limit_obuf);
- } else {
- log_debug("%s, timed out connections %u", proto, timedout);
- }
-
- knot_sweep_stats_reset(stats);
- stats->last_log = sec;
-}
-
void xdp_handle_reconfigure(xdp_handle_ctx_t *ctx)
{
rcu_read_lock();
}
#endif // ENABLE_QUIC
-xdp_handle_ctx_t *xdp_handle_init(struct server *server, knot_xdp_socket_t *xdp_sock)
+xdp_handle_ctx_t *xdp_handle_init(server_t *server, knot_xdp_socket_t *xdp_sock)
{
xdp_handle_ctx_t *ctx = calloc(1, sizeof(*ctx));
if (ctx == NULL) {
return ret == KNOT_EOK ? ctx->msg_recv_count : ret;
}
-static void handle_init(knotd_qdata_params_t *params, knot_layer_t *layer,
- knotd_query_proto_t proto, const knot_xdp_msg_t *msg,
- const struct iovec *payload, struct sockaddr_storage *proxied_remote)
-{
- params->proto = proto;
- params->remote = (struct sockaddr_storage *)&msg->ip_from;
- params->local = (struct sockaddr_storage *)&msg->ip_to;
- params->xdp_msg = msg;
-
- knot_layer_begin(layer, params);
-
- knot_pkt_t *query = knot_pkt_new(payload->iov_base, payload->iov_len, layer->mm);
- int ret = knot_pkt_parse(query, 0);
- if (ret != KNOT_EOK && query->parsed > 0) { // parsing failed (e.g. 2x OPT)
- if (params->proto == KNOTD_QUERY_PROTO_UDP &&
- proxyv2_header_strip(&query, params->remote, proxied_remote) == KNOT_EOK) {
- assert(proxied_remote);
- params->remote = proxied_remote;
- } else {
- query->parsed--; // artificially decreasing "parsed" leads to FORMERR
- }
- }
- knot_layer_consume(layer, query);
-}
-
-static void handle_finish(knot_layer_t *layer)
-{
- knot_layer_finish(layer);
-
- // Flush per-query memory (including query and answer packets).
- mp_flush(layer->mm->ctx);
-}
-
static void handle_udp(xdp_handle_ctx_t *ctx, knot_layer_t *layer,
knotd_qdata_params_t *params)
{
}
ctx->msg_udp_count++;
- // Consume the query.
- handle_init(params, layer, KNOTD_QUERY_PROTO_UDP, msg_recv, &msg_recv->payload,
- &proxied_remote);
-
- // Process the reply.
- knot_pkt_t *ans = knot_pkt_new(msg_send->payload.iov_base,
- msg_send->payload.iov_len, layer->mm);
- while (udp_state_active(layer->state)) {
- knot_layer_produce(layer, ans);
- }
- if (layer->state == KNOT_STATE_DONE) {
- msg_send->payload.iov_len = ans->size;
- } else {
- // If not success, don't send any reply.
- msg_send->payload.iov_len = 0;
- }
-
- // Reset the processing.
- handle_finish(layer);
+ // Prepare a reply.
+ params_xdp_update(params, KNOTD_QUERY_PROTO_UDP, msg_recv, 0, NULL);
+ handle_udp_reply(params, layer, &msg_recv->payload, &msg_send->payload,
+ &proxied_remote);
}
}
// Process all complete DNS queries in one TCP stream.
for (size_t j = 0; j < rl->inbufs_count; j++) {
// Consume the query.
- handle_init(params, layer, KNOTD_QUERY_PROTO_TCP, rl->msg, &rl->inbufs[j], NULL);
- params->measured_rtt = rl->conn->establish_rtt;
+ params_xdp_update(params, KNOTD_QUERY_PROTO_TCP, ctx->msg_recv,
+ rl->conn->establish_rtt, NULL);
+ handle_query(params, layer, &rl->inbufs[j], NULL);
// Process the reply.
knot_pkt_t *ans = knot_pkt_new(ans_buf, sizeof(ans_buf), layer->mm);
- while (tcp_active_state(layer->state)) {
+ while (active_state(layer->state)) {
knot_layer_produce(layer, ans);
- if (!tcp_send_state(layer->state)) {
+ if (!send_state(layer->state)) {
continue;
}
}
}
-#ifdef ENABLE_QUIC
-static void handle_quic_stream(knot_xquic_conn_t *conn, int64_t stream_id, struct iovec *inbuf,
- knot_layer_t *layer, knotd_qdata_params_t *params, uint8_t *ans_buf,
- size_t ans_buf_size, const knot_xdp_msg_t *xdp_msg)
-{
- // Consume the query.
- handle_init(params, layer, KNOTD_QUERY_PROTO_QUIC, xdp_msg, inbuf, NULL);
- params->measured_rtt = knot_xquic_conn_rtt(conn);
-
- // Process the reply.
- knot_pkt_t *ans = knot_pkt_new(ans_buf, ans_buf_size, layer->mm);
- while (tcp_active_state(layer->state)) {
- knot_layer_produce(layer, ans);
- if (!tcp_send_state(layer->state)) {
- continue;
- }
- if (knot_xquic_stream_add_data(conn, stream_id, ans->wire, ans->size) == NULL) {
- break;
- }
- }
-
- handle_finish(layer);
-}
-#endif // ENABLE_QUIC
-
static void handle_quic(xdp_handle_ctx_t *ctx, knot_layer_t *layer,
knotd_qdata_params_t *params)
{
return;
}
- uint8_t ans_buf[KNOT_WIRE_MAX_PKTSIZE];
-
for (uint32_t i = 0; i < ctx->msg_recv_count; i++) {
knot_xdp_msg_t *msg_recv = &ctx->msg_recv[i];
ctx->quic_relays[i] = NULL;
(void)knot_quic_handle(ctx->quic_table, reply, ctx->quic_idle_close,
&ctx->quic_relays[i]);
+ knot_xquic_conn_t *conn = ctx->quic_relays[i];
- knot_xquic_conn_t *rl = ctx->quic_relays[i];
-
- int64_t stream_id;
- knot_xquic_stream_t *stream;
-
- while (rl != NULL && (stream = knot_xquic_stream_get_process(rl, &stream_id)) != NULL) {
- assert(stream->inbuf_fin != NULL);
- assert(stream->inbuf_fin->iov_len > 0);
- params->session = rl->tls_session;
- handle_quic_stream(rl, stream_id, stream->inbuf_fin, layer, params,
- ans_buf, sizeof(ans_buf), &ctx->msg_recv[i]);
- free(stream->inbuf_fin);
- stream->inbuf_fin = NULL;
- }
+ handle_quic_streams(conn, params, layer, &ctx->msg_recv[i]);
}
#else
(void)(ctx);
{
assert(ctx->msg_recv_count > 0);
- knotd_qdata_params_t params = {
- .socket = knot_xdp_socket_fd(ctx->sock),
- .server = server,
- .thread_id = thread_id,
- };
+ knotd_qdata_params_t params = params_xdp_init(
+ knot_xdp_socket_fd(ctx->sock), server, thread_id);
knot_xdp_send_prepare(ctx->sock);
#ifdef ENABLE_QUIC
if (ctx->quic_table != NULL) {
knot_xquic_table_sweep(ctx->quic_table, &ctx->quic_closed);
- log_closed(&ctx->quic_closed, false);
+ log_swept(&ctx->quic_closed, false);
}
#endif // ENABLE_QUIC
(void)knot_xdp_send_finish(ctx->sock);
} while (ret == KNOT_EOK && prev_total < ctx->tcp_closed.total);
- log_closed(&ctx->tcp_closed, true);
+ log_swept(&ctx->tcp_closed, true);
}
#endif // ENABLE_XDP
static void udp_stdin_handle(udp_context_t *ctx, _unused_ const iface_t *iface, void *d)
{
udp_stdin_t *rq = (udp_stdin_t *)d;
- PARAMS_INIT(false, STDIN_FILENO, &rq->addr, &iface->addr, ctx);
+ knotd_qdata_params_t params = params_init(KNOTD_QUERY_PROTO_UDP, &rq->addr,
+ &iface->addr, STDIN_FILENO, NULL, 0);
udp_handler(ctx, ¶ms, &rq->iov[RX], &rq->iov[TX]);
}
projects / thirdparty / knot-dns.git / commitdiff
