Adding upstream version 3.2.1.upstream/3.2.1upstream

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

1 files changed, 494 insertions, 0 deletions

diff --git a/daemon/io.c b/daemon/io.c
new file mode 100644
index 0000000..f183d34
--- /dev/null
+++ b/daemon/io.c
@@ -0,0 +1,494 @@
+/*  Copyright (C) 2014-2017 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 <string.h>
+#include <libknot/errcode.h>
+#include <contrib/ucw/lib.h>
+#include <contrib/ucw/mempool.h>
+#include <assert.h>
+
+#include "daemon/io.h"
+#include "daemon/network.h"
+#include "daemon/worker.h"
+#include "daemon/tls.h"
+#include "daemon/session.h"
+
+#define negotiate_bufsize(func, handle, bufsize_want) do { \
+    int bufsize = 0; func(handle, &bufsize); \
+	if (bufsize < bufsize_want) { \
+		bufsize = bufsize_want; \
+		func(handle, &bufsize); \
+	} \
+} while (0)
+
+static void check_bufsize(uv_handle_t* handle)
+{
+	return; /* TODO: resurrect after https://github.com/libuv/libuv/issues/419 */
+	/* We want to buffer at least N waves in advance.
+	 * This is magic presuming we can pull in a whole recvmmsg width in one wave.
+	 * Linux will double this the bufsize wanted.
+	 */
+	const int bufsize_want = 2 * sizeof( ((struct worker_ctx *)NULL)->wire_buf ) ;
+	negotiate_bufsize(uv_recv_buffer_size, handle, bufsize_want);
+	negotiate_bufsize(uv_send_buffer_size, handle, bufsize_want);
+}
+
+#undef negotiate_bufsize
+
+static void handle_getbuf(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf)
+{
+	/* UDP sessions use worker buffer for wire data,
+	 * TCP sessions use session buffer for wire data
+	 * (see session_set_handle()).
+	 * TLS sessions use buffer from TLS context.
+	 * The content of the worker buffer is
+	 * guaranteed to be unchanged only for the duration of
+	 * udp_read() and tcp_read().
+	 */
+	struct session *s = handle->data;
+	if (!session_flags(s)->has_tls) {
+		buf->base = (char *) session_wirebuf_get_free_start(s);
+		buf->len = session_wirebuf_get_free_size(s);
+	} else {
+		struct tls_common_ctx *ctx = session_tls_get_common_ctx(s);
+		buf->base = (char *) ctx->recv_buf;
+		buf->len = sizeof(ctx->recv_buf);
+	}
+}
+
+void udp_recv(uv_udp_t *handle, ssize_t nread, const uv_buf_t *buf,
+	const struct sockaddr *addr, unsigned flags)
+{
+	uv_loop_t *loop = handle->loop;
+	struct worker_ctx *worker = loop->data;
+	struct session *s = handle->data;
+	if (session_flags(s)->closing) {
+		return;
+	}
+	if (nread <= 0) {
+		if (nread < 0) { /* Error response, notify resolver */
+			worker_submit(s, NULL);
+		} /* nread == 0 is for freeing buffers, we don't need to do this */
+		return;
+	}
+	if (addr->sa_family == AF_UNSPEC) {
+		return;
+	}
+	struct sockaddr *peer = session_get_peer(s);
+	if (session_flags(s)->outgoing) {
+		assert(peer->sa_family != AF_UNSPEC);
+		if (kr_sockaddr_cmp(peer, addr) != 0) {
+			return;
+		}
+	} else {
+		memcpy(peer, addr, kr_sockaddr_len(addr));
+	}
+	ssize_t consumed = session_wirebuf_consume(s, (const uint8_t *)buf->base,
+						   nread);
+	assert(consumed == nread); (void)consumed;
+	session_wirebuf_process(s);
+	session_wirebuf_discard(s);
+	mp_flush(worker->pkt_pool.ctx);
+}
+
+static int udp_bind_finalize(uv_handle_t *handle)
+{
+	check_bufsize(handle);
+	/* Handle is already created, just create context. */
+	struct session *s = session_new(handle, false);
+	assert(s);
+	session_flags(s)->outgoing = false;
+	return io_start_read(handle);
+}
+
+int udp_bind(uv_udp_t *handle, struct sockaddr *addr)
+{
+	unsigned flags = UV_UDP_REUSEADDR;
+	if (addr->sa_family == AF_INET6) {
+		flags |= UV_UDP_IPV6ONLY;
+	}
+	int ret = uv_udp_bind(handle, addr, flags);
+	if (ret != 0) {
+		return ret;
+	}
+	return udp_bind_finalize((uv_handle_t *)handle);
+}
+
+int udp_bindfd(uv_udp_t *handle, int fd)
+{
+	if (!handle) {
+		return kr_error(EINVAL);
+	}
+
+	int ret = uv_udp_open(handle, (uv_os_sock_t) fd);
+	if (ret != 0) {
+		return ret;
+	}
+	return udp_bind_finalize((uv_handle_t *)handle);
+}
+
+void tcp_timeout_trigger(uv_timer_t *timer)
+{
+	struct session *s = timer->data;
+
+	assert(!session_flags(s)->closing);
+
+	struct worker_ctx *worker = timer->loop->data;
+
+	if (!session_tasklist_is_empty(s)) {
+		int finalized = session_tasklist_finalize_expired(s);
+		worker->stats.timeout += finalized;
+		/* session_tasklist_finalize_expired() may call worker_task_finalize().
+		 * If session is a source session and there were IO errors,
+		 * worker_task_finalize() can filnalize all tasks and close session. */
+		if (session_flags(s)->closing) {
+			return;
+		}
+
+	}
+	if (!session_tasklist_is_empty(s)) {
+		uv_timer_stop(timer);
+		session_timer_start(s, tcp_timeout_trigger,
+				    KR_RESOLVE_TIME_LIMIT / 2,
+				    KR_RESOLVE_TIME_LIMIT / 2);
+	} else {
+		/* Normally it should not happen,
+		 * but better to check if there anything in this list. */
+		while (!session_waitinglist_is_empty(s)) {
+			struct qr_task *t = session_waitinglist_pop(s, false);
+			worker_task_finalize(t, KR_STATE_FAIL);
+			worker_task_unref(t);
+			worker->stats.timeout += 1;
+			if (session_flags(s)->closing) {
+				return;
+			}
+		}
+		const struct engine *engine = worker->engine;
+		const struct network *net = &engine->net;
+		uint64_t idle_in_timeout = net->tcp.in_idle_timeout;
+		uint64_t last_activity = session_last_activity(s);
+		uint64_t idle_time = kr_now() - last_activity;
+		if (idle_time < idle_in_timeout) {
+			idle_in_timeout -= idle_time;
+			uv_timer_stop(timer);
+			session_timer_start(s, tcp_timeout_trigger,
+					    idle_in_timeout, idle_in_timeout);
+		} else {
+			struct sockaddr *peer = session_get_peer(s);
+			char *peer_str = kr_straddr(peer);
+			kr_log_verbose("[io] => closing connection to '%s'\n",
+				       peer_str ? peer_str : "");
+			if (session_flags(s)->outgoing) {
+				worker_del_tcp_waiting(worker, peer);
+				worker_del_tcp_connected(worker, peer);
+			}
+			session_close(s);
+		}
+	}
+}
+
+static void tcp_recv(uv_stream_t *handle, ssize_t nread, const uv_buf_t *buf)
+{
+	struct session *s = handle->data;
+	assert(s && session_get_handle(s) == (uv_handle_t *)handle &&
+	       handle->type == UV_TCP);	
+
+	if (session_flags(s)->closing) {
+		return;
+	}
+
+	/* nread might be 0, which does not indicate an error or EOF.
+	 * This is equivalent to EAGAIN or EWOULDBLOCK under read(2). */
+	if (nread == 0) {
+		return;
+	}
+
+	if (nread < 0 || !buf->base) {
+		if (kr_verbose_status) {
+			struct sockaddr *peer = session_get_peer(s);
+			char *peer_str = kr_straddr(peer);
+			kr_log_verbose("[io] => connection to '%s' closed by peer (%s)\n",
+				       peer_str ? peer_str : "",
+				       uv_strerror(nread));
+		}
+		worker_end_tcp(s);
+		return;
+	}
+
+	ssize_t consumed = 0;
+	const uint8_t *data = (const uint8_t *)buf->base;
+	ssize_t data_len = nread;
+	if (session_flags(s)->has_tls) {
+		/* buf->base points to start of the tls receive buffer.
+		   Decode data free space in session wire buffer. */
+		consumed = tls_process_input_data(s, (const uint8_t *)buf->base, nread);
+		if (consumed < 0) {
+			if (kr_verbose_status) {
+				struct sockaddr *peer = session_get_peer(s);
+				char *peer_str = kr_straddr(peer);
+				kr_log_verbose("[io] => connection to '%s': "
+					       "error processing TLS data, close\n",
+					       peer_str ? peer_str : "");
+			}
+			worker_end_tcp(s);
+			return;
+		} else if (consumed == 0) {
+			return;
+		}
+		data = session_wirebuf_get_free_start(s);
+		data_len = consumed;
+	}
+
+	/* data points to start of the free space in session wire buffer.
+	   Simple increase internal counter. */
+	consumed = session_wirebuf_consume(s, data, data_len);
+	assert(consumed == data_len);
+
+	int ret = session_wirebuf_process(s);
+	if (ret < 0) {
+		/* An error has occurred, close the session. */
+		worker_end_tcp(s);
+	}
+	session_wirebuf_compress(s);
+	struct worker_ctx *worker = handle->loop->data;
+	mp_flush(worker->pkt_pool.ctx);
+}
+
+static void _tcp_accept(uv_stream_t *master, int status, bool tls)
+{
+ 	if (status != 0) {
+		return;
+	}
+
+	struct worker_ctx *worker = (struct worker_ctx *)master->loop->data;
+	uv_tcp_t *client = malloc(sizeof(uv_tcp_t));
+	if (!client) {
+		return;
+	}
+	int res = io_create(master->loop, (uv_handle_t *)client,
+			    SOCK_STREAM, AF_UNSPEC, tls);
+	if (res) {
+		if (res == UV_EMFILE) {
+			worker->too_many_open = true;
+			worker->rconcurrent_highwatermark = worker->stats.rconcurrent;
+		}
+		/* Since res isn't OK struct session wasn't allocated \ borrowed.
+		 * We must release client handle only.
+		 */
+		free(client);
+		return;
+	}
+
+	/* struct session was allocated \ borrowed from memory pool. */
+	struct session *s = client->data;
+	assert(session_flags(s)->outgoing == false);
+	assert(session_flags(s)->has_tls == tls);
+
+	if (uv_accept(master, (uv_stream_t *)client) != 0) {
+		/* close session, close underlying uv handles and
+		 * deallocate (or return to memory pool) memory. */
+		session_close(s);
+		return;
+	}
+
+	/* Set deadlines for TCP connection and start reading.
+	 * It will re-check every half of a request time limit if the connection
+	 * is idle and should be terminated, this is an educated guess. */
+	struct sockaddr *peer = session_get_peer(s);
+	int peer_len = sizeof(union inaddr);
+	int ret = uv_tcp_getpeername(client, peer, &peer_len);
+	if (ret || peer->sa_family == AF_UNSPEC) {
+		session_close(s);
+		return;
+	}
+
+	const struct engine *engine = worker->engine;
+	const struct network *net = &engine->net;
+	uint64_t idle_in_timeout = net->tcp.in_idle_timeout;
+
+	uint64_t timeout = KR_CONN_RTT_MAX / 2;
+	if (tls) {
+		timeout += TLS_MAX_HANDSHAKE_TIME;
+		struct tls_ctx_t *ctx = session_tls_get_server_ctx(s);
+		if (!ctx) {
+			ctx = tls_new(worker);
+			if (!ctx) {
+				session_close(s);
+				return;
+			}
+			ctx->c.session = s;
+			ctx->c.handshake_state = TLS_HS_IN_PROGRESS;
+			session_tls_set_server_ctx(s, ctx);
+		}
+	}
+	session_timer_start(s, tcp_timeout_trigger, timeout, idle_in_timeout);
+	io_start_read((uv_handle_t *)client);
+}
+
+static void tcp_accept(uv_stream_t *master, int status)
+{
+	_tcp_accept(master, status, false);
+}
+
+static void tls_accept(uv_stream_t *master, int status)
+{
+	_tcp_accept(master, status, true);
+}
+
+static int set_tcp_option(uv_handle_t *handle, int option, int val)
+{
+	uv_os_fd_t fd = 0;
+	if (uv_fileno(handle, &fd) == 0) {
+		return setsockopt(fd, IPPROTO_TCP, option, &val, sizeof(val));
+	}
+	return 0; /* N/A */
+}
+
+static int tcp_bind_finalize(uv_handle_t *handle)
+{
+	/* TCP_FASTOPEN enables 1 RTT connection resumptions. */
+#ifdef TCP_FASTOPEN
+# ifdef __linux__
+	(void) set_tcp_option(handle, TCP_FASTOPEN, 16); /* Accepts queue length hint */
+# else
+	(void) set_tcp_option(handle, TCP_FASTOPEN, 1);  /* Accepts on/off */
+# endif
+#endif
+
+	handle->data = NULL;
+	return 0;
+}
+
+static int _tcp_bind(uv_tcp_t *handle, struct sockaddr *addr, uv_connection_cb connection, int tcp_backlog)
+{
+	unsigned flags = 0;
+	if (addr->sa_family == AF_INET6) {
+		flags |= UV_TCP_IPV6ONLY;
+	}
+
+	int ret = uv_tcp_bind(handle, addr, flags);
+	if (ret != 0) {
+		return ret;
+	}
+
+	/* TCP_DEFER_ACCEPT delays accepting connections until there is readable data. */
+#ifdef TCP_DEFER_ACCEPT
+	if (set_tcp_option((uv_handle_t *)handle, TCP_DEFER_ACCEPT, KR_CONN_RTT_MAX/1000) != 0) {
+		kr_log_info("[ io ] tcp_bind (defer_accept): %s\n", strerror(errno));
+	}
+#endif
+
+	ret = uv_listen((uv_stream_t *)handle, tcp_backlog, connection);
+	if (ret != 0) {
+		return ret;
+	}
+
+	return tcp_bind_finalize((uv_handle_t *)handle);
+}
+
+int tcp_bind(uv_tcp_t *handle, struct sockaddr *addr, int tcp_backlog)
+{
+	return _tcp_bind(handle, addr, tcp_accept, tcp_backlog);
+}
+
+int tcp_bind_tls(uv_tcp_t *handle, struct sockaddr *addr, int tcp_backlog)
+{
+	return _tcp_bind(handle, addr, tls_accept, tcp_backlog);
+}
+
+static int _tcp_bindfd(uv_tcp_t *handle, int fd, uv_connection_cb connection, int tcp_backlog)
+{
+	if (!handle) {
+		return kr_error(EINVAL);
+	}
+
+	int ret = uv_tcp_open(handle, (uv_os_sock_t) fd);
+	if (ret != 0) {
+		return ret;
+	}
+
+	ret = uv_listen((uv_stream_t *)handle, tcp_backlog, connection);
+	if (ret != 0) {
+		return ret;
+	}
+	return tcp_bind_finalize((uv_handle_t *)handle);
+}
+
+int tcp_bindfd(uv_tcp_t *handle, int fd, int tcp_backlog)
+{
+	return _tcp_bindfd(handle, fd, tcp_accept, tcp_backlog);
+}
+
+int tcp_bindfd_tls(uv_tcp_t *handle, int fd, int tcp_backlog)
+{
+	return _tcp_bindfd(handle, fd, tls_accept, tcp_backlog);
+}
+
+int io_create(uv_loop_t *loop, uv_handle_t *handle, int type, unsigned family, bool has_tls)
+{
+	int ret = -1;
+	if (type == SOCK_DGRAM) {
+		ret = uv_udp_init(loop, (uv_udp_t *)handle);
+	} else if (type == SOCK_STREAM) {
+		ret = uv_tcp_init_ex(loop, (uv_tcp_t *)handle, family);
+		uv_tcp_nodelay((uv_tcp_t *)handle, 1);
+	}
+	if (ret != 0) {
+		return ret;
+	}
+	struct session *s = session_new(handle, has_tls);
+	if (s == NULL) {
+		ret = -1;
+	}
+	return ret;
+}
+
+void io_deinit(uv_handle_t *handle)
+{
+	if (!handle) {
+		return;
+	}
+	session_free(handle->data);
+	handle->data = NULL;
+}
+
+void io_free(uv_handle_t *handle)
+{
+	io_deinit(handle);
+	free(handle);
+}
+
+int io_start_read(uv_handle_t *handle)
+{
+	switch (handle->type) {
+	case UV_UDP:
+		return uv_udp_recv_start((uv_udp_t *)handle, &handle_getbuf, &udp_recv);
+	case UV_TCP:
+		return uv_read_start((uv_stream_t *)handle, &handle_getbuf, &tcp_recv);
+	default:
+		assert(!EINVAL);
+		return kr_error(EINVAL);
+	}
+}
+
+int io_stop_read(uv_handle_t *handle)
+{
+	if (handle->type == UV_UDP) {
+		return uv_udp_recv_stop((uv_udp_t *)handle);
+	} else {
+		return uv_read_stop((uv_stream_t *)handle);
+	}
+}