Adding upstream version 5.6.0.upstream/5.6.0upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 145 insertions, 0 deletions

diff --git a/daemon/udp_queue.c b/daemon/udp_queue.c
new file mode 100644
index 0000000..7460e04
--- /dev/null
+++ b/daemon/udp_queue.c
@@ -0,0 +1,145 @@
+/*  Copyright (C) CZ.NIC, z.s.p.o. <knot-resolver@labs.nic.cz>
+ *  SPDX-License-Identifier: GPL-3.0-or-later
+ */
+
+#include "kresconfig.h"
+#include "daemon/udp_queue.h"
+
+#include "daemon/worker.h"
+#include "lib/generic/array.h"
+#include "lib/utils.h"
+
+struct qr_task;
+
+#include <sys/socket.h>
+
+
+#if !ENABLE_SENDMMSG
+int udp_queue_init_global(uv_loop_t *loop)
+{
+	return 0;
+}
+/* Appease the linker in case this unused call isn't optimized out. */
+void udp_queue_push(int fd, struct kr_request *req, struct qr_task *task)
+{
+	abort();
+}
+#else
+
+/* LATER: it might be useful to have this configurable during runtime,
+ * but the structures below would have to change a little (broken up). */
+#define UDP_QUEUE_LEN 64
+
+/** A queue of up to UDP_QUEUE_LEN messages, meant for the same socket. */
+typedef struct {
+	int len; /**< The number of messages in the queue: 0..UDP_QUEUE_LEN */
+	struct mmsghdr msgvec[UDP_QUEUE_LEN]; /**< Parameter for sendmmsg() */
+	struct {
+		struct qr_task *task; /**< Links for completion callbacks. */
+		struct iovec msg_iov[1]; /**< storage for .msgvec[i].msg_iov */
+	} items[UDP_QUEUE_LEN];
+} udp_queue_t;
+
+static udp_queue_t * udp_queue_create()
+{
+	udp_queue_t *q = calloc(1, sizeof(*q));
+	kr_require(q != NULL);
+
+	for (int i = 0; i < UDP_QUEUE_LEN; ++i) {
+		struct msghdr *mhi = &q->msgvec[i].msg_hdr;
+		/* These shall remain always the same. */
+		mhi->msg_iov = q->items[i].msg_iov;
+		mhi->msg_iovlen = 1;
+		/* msg_name and msg_namelen will be per-call,
+		 * and the rest is OK to remain zeroed all the time. */
+	}
+	return q;
+}
+
+/** Global state for udp_queue_*.  Note: we never free the pointed-to memory. */
+struct {
+	/** Singleton map: fd -> udp_queue_t, as a simple array of pointers. */
+	udp_queue_t **udp_queues;
+	int udp_queues_len;
+
+	/** List of FD numbers that might have a non-empty queue. */
+	array_t(int) waiting_fds;
+
+	uv_check_t check_handle;
+} static state = {0};
+
+/** Empty the given queue.  The queue is assumed to exist (but may be empty). */
+static void udp_queue_send(int fd)
+{
+	udp_queue_t *const q = state.udp_queues[fd];
+	if (!q->len) return;
+	int sent_len = sendmmsg(fd, q->msgvec, q->len, 0);
+	/* ATM we don't really do anything about failures. */
+	int err = sent_len < 0 ? errno : EAGAIN /* unknown error, really */;
+	for (int i = 0; i < q->len; ++i) {
+		qr_task_on_send(q->items[i].task, NULL, i < sent_len ? 0 : err);
+		worker_task_unref(q->items[i].task);
+	}
+	q->len = 0;
+}
+
+/** Periodical callback to send all queued packets. */
+static void udp_queue_check(uv_check_t *handle)
+{
+	for (int i = 0; i < state.waiting_fds.len; ++i) {
+		udp_queue_send(state.waiting_fds.at[i]);
+	}
+	state.waiting_fds.len = 0;
+}
+
+int udp_queue_init_global(uv_loop_t *loop)
+{
+	int ret = uv_check_init(loop, &state.check_handle);
+	if (!ret) ret = uv_check_start(&state.check_handle, udp_queue_check);
+	return ret;
+}
+
+void udp_queue_push(int fd, struct kr_request *req, struct qr_task *task)
+{
+	if (fd < 0) {
+		kr_log_error(SYSTEM, "ERROR: called udp_queue_push(fd = %d, ...)\n", fd);
+		abort();
+	}
+	worker_task_ref(task);
+	/* Get a valid correct queue. */
+	if (fd >= state.udp_queues_len) {
+		const int new_len = fd + 1;
+		state.udp_queues = realloc(state.udp_queues,
+					sizeof(state.udp_queues[0]) * new_len);
+		if (!state.udp_queues) abort();
+		memset(state.udp_queues + state.udp_queues_len, 0,
+			sizeof(state.udp_queues[0]) * (new_len - state.udp_queues_len));
+		state.udp_queues_len = new_len;
+	}
+	if (unlikely(state.udp_queues[fd] == NULL))
+		state.udp_queues[fd] = udp_queue_create();
+	udp_queue_t *const q = state.udp_queues[fd];
+
+	/* Append to the queue */
+	struct sockaddr *sa = (struct sockaddr *)/*const-cast*/req->qsource.comm_addr;
+	q->msgvec[q->len].msg_hdr.msg_name = sa;
+	q->msgvec[q->len].msg_hdr.msg_namelen = kr_sockaddr_len(sa);
+	q->items[q->len].task = task;
+	q->items[q->len].msg_iov[0] = (struct iovec){
+		.iov_base = req->answer->wire,
+		.iov_len  = req->answer->size,
+	};
+	if (q->len == 0)
+		array_push(state.waiting_fds, fd);
+	++(q->len);
+
+	if (q->len >= UDP_QUEUE_LEN) {
+		kr_assert(q->len == UDP_QUEUE_LEN);
+		udp_queue_send(fd);
+		/* We don't need to search state.waiting_fds;
+		 * anyway, it's more efficient to let the hook do that. */
+	}
+}
+
+#endif
+