Adding upstream version 1:9.16.44.upstream/1%9.16.44upstream

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

1 files changed, 1211 insertions, 0 deletions

diff --git a/lib/isc/netmgr/udp.c b/lib/isc/netmgr/udp.c
new file mode 100644
index 0000000..00f9d40
--- /dev/null
+++ b/lib/isc/netmgr/udp.c
@@ -0,0 +1,1211 @@
+/*
+ * Copyright (C) Internet Systems Consortium, Inc. ("ISC")
+ *
+ * SPDX-License-Identifier: MPL-2.0
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, you can obtain one at https://mozilla.org/MPL/2.0/.
+ *
+ * See the COPYRIGHT file distributed with this work for additional
+ * information regarding copyright ownership.
+ */
+
+#include <unistd.h>
+#include <uv.h>
+
+#include <isc/atomic.h>
+#include <isc/barrier.h>
+#include <isc/buffer.h>
+#include <isc/condition.h>
+#include <isc/errno.h>
+#include <isc/magic.h>
+#include <isc/mem.h>
+#include <isc/netmgr.h>
+#include <isc/random.h>
+#include <isc/refcount.h>
+#include <isc/region.h>
+#include <isc/result.h>
+#include <isc/sockaddr.h>
+#include <isc/thread.h>
+#include <isc/util.h>
+
+#include "netmgr-int.h"
+#include "uv-compat.h"
+
+static isc_result_t
+udp_send_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req,
+		isc_sockaddr_t *peer);
+
+static void
+udp_recv_cb(uv_udp_t *handle, ssize_t nrecv, const uv_buf_t *buf,
+	    const struct sockaddr *addr, unsigned flags);
+
+static void
+udp_send_cb(uv_udp_send_t *req, int status);
+
+static void
+udp_close_cb(uv_handle_t *handle);
+
+static void
+read_timer_close_cb(uv_handle_t *handle);
+
+static void
+udp_close_direct(isc_nmsocket_t *sock);
+
+static void
+stop_udp_parent(isc_nmsocket_t *sock);
+static void
+stop_udp_child(isc_nmsocket_t *sock);
+
+static uv_os_sock_t
+isc__nm_udp_lb_socket(isc_nm_t *mgr, sa_family_t sa_family) {
+	isc_result_t result;
+	uv_os_sock_t sock;
+
+	result = isc__nm_socket(sa_family, SOCK_DGRAM, 0, &sock);
+	RUNTIME_CHECK(result == ISC_R_SUCCESS);
+
+	(void)isc__nm_socket_incoming_cpu(sock);
+	(void)isc__nm_socket_disable_pmtud(sock, sa_family);
+
+	result = isc__nm_socket_reuse(sock);
+	RUNTIME_CHECK(result == ISC_R_SUCCESS);
+
+#ifndef _WIN32
+	if (mgr->load_balance_sockets) {
+		result = isc__nm_socket_reuse_lb(sock);
+		RUNTIME_CHECK(result == ISC_R_SUCCESS);
+	}
+#endif
+
+	return (sock);
+}
+
+static void
+start_udp_child(isc_nm_t *mgr, isc_sockaddr_t *iface, isc_nmsocket_t *sock,
+		uv_os_sock_t fd, int tid) {
+	isc_nmsocket_t *csock;
+	isc__netievent_udplisten_t *ievent = NULL;
+
+	csock = &sock->children[tid];
+
+	isc__nmsocket_init(csock, mgr, isc_nm_udpsocket, iface);
+	csock->parent = sock;
+	csock->iface = sock->iface;
+	csock->reading = true;
+	csock->recv_cb = sock->recv_cb;
+	csock->recv_cbarg = sock->recv_cbarg;
+	csock->extrahandlesize = sock->extrahandlesize;
+	csock->tid = tid;
+
+#ifdef _WIN32
+	UNUSED(fd);
+	csock->fd = isc__nm_udp_lb_socket(mgr, iface->type.sa.sa_family);
+#else
+	if (mgr->load_balance_sockets) {
+		UNUSED(fd);
+		csock->fd = isc__nm_udp_lb_socket(mgr,
+						  iface->type.sa.sa_family);
+	} else {
+		csock->fd = dup(fd);
+	}
+#endif
+	REQUIRE(csock->fd >= 0);
+
+	ievent = isc__nm_get_netievent_udplisten(mgr, csock);
+	isc__nm_maybe_enqueue_ievent(&mgr->workers[tid],
+				     (isc__netievent_t *)ievent);
+}
+
+static void
+enqueue_stoplistening(isc_nmsocket_t *sock) {
+	isc__netievent_udpstop_t *ievent =
+		isc__nm_get_netievent_udpstop(sock->mgr, sock);
+	isc__nm_enqueue_ievent(&sock->mgr->workers[sock->tid],
+			       (isc__netievent_t *)ievent);
+}
+
+isc_result_t
+isc_nm_listenudp(isc_nm_t *mgr, isc_sockaddr_t *iface, isc_nm_recv_cb_t cb,
+		 void *cbarg, size_t extrahandlesize, isc_nmsocket_t **sockp) {
+	isc_result_t result = ISC_R_SUCCESS;
+	isc_nmsocket_t *sock = NULL;
+	size_t children_size = 0;
+	REQUIRE(VALID_NM(mgr));
+	uv_os_sock_t fd = -1;
+
+	/*
+	 * We are creating mgr->nworkers duplicated sockets, one
+	 * socket for each worker thread.
+	 */
+	sock = isc_mem_get(mgr->mctx, sizeof(isc_nmsocket_t));
+	isc__nmsocket_init(sock, mgr, isc_nm_udplistener, iface);
+
+	atomic_init(&sock->rchildren, 0);
+#if defined(WIN32)
+	sock->nchildren = 1;
+#else
+	sock->nchildren = mgr->nworkers;
+#endif
+
+	children_size = sock->nchildren * sizeof(sock->children[0]);
+	sock->children = isc_mem_get(mgr->mctx, children_size);
+	memset(sock->children, 0, children_size);
+
+	sock->recv_cb = cb;
+	sock->recv_cbarg = cbarg;
+	sock->extrahandlesize = extrahandlesize;
+	sock->result = ISC_R_UNSET;
+
+	sock->tid = 0;
+	sock->fd = -1;
+
+#ifndef _WIN32
+	if (!mgr->load_balance_sockets) {
+		fd = isc__nm_udp_lb_socket(mgr, iface->type.sa.sa_family);
+	}
+#endif
+
+	isc_barrier_init(&sock->startlistening, sock->nchildren);
+
+	for (size_t i = 0; i < sock->nchildren; i++) {
+		if ((int)i == isc_nm_tid()) {
+			continue;
+		}
+		start_udp_child(mgr, iface, sock, fd, i);
+	}
+
+	if (isc__nm_in_netthread()) {
+		start_udp_child(mgr, iface, sock, fd, isc_nm_tid());
+	}
+
+#ifndef _WIN32
+	if (!mgr->load_balance_sockets) {
+		isc__nm_closesocket(fd);
+	}
+#endif
+
+	LOCK(&sock->lock);
+	while (atomic_load(&sock->rchildren) != sock->nchildren) {
+		WAIT(&sock->cond, &sock->lock);
+	}
+	result = sock->result;
+	atomic_store(&sock->active, true);
+	UNLOCK(&sock->lock);
+
+	INSIST(result != ISC_R_UNSET);
+
+	if (result == ISC_R_SUCCESS) {
+		REQUIRE(atomic_load(&sock->rchildren) == sock->nchildren);
+		*sockp = sock;
+	} else {
+		atomic_store(&sock->active, false);
+		enqueue_stoplistening(sock);
+		isc_nmsocket_close(&sock);
+	}
+
+	return (result);
+}
+
+/*
+ * Asynchronous 'udplisten' call handler: start listening on a UDP socket.
+ */
+void
+isc__nm_async_udplisten(isc__networker_t *worker, isc__netievent_t *ev0) {
+	isc__netievent_udplisten_t *ievent = (isc__netievent_udplisten_t *)ev0;
+	isc_nmsocket_t *sock = NULL;
+	int r, uv_bind_flags = 0;
+	int uv_init_flags = 0;
+	sa_family_t sa_family;
+	isc_result_t result = ISC_R_UNSET;
+	isc_nm_t *mgr = NULL;
+
+	REQUIRE(VALID_NMSOCK(ievent->sock));
+	REQUIRE(ievent->sock->tid == isc_nm_tid());
+	REQUIRE(VALID_NMSOCK(ievent->sock->parent));
+
+	sock = ievent->sock;
+	sa_family = sock->iface.type.sa.sa_family;
+	mgr = sock->mgr;
+
+	REQUIRE(sock->type == isc_nm_udpsocket);
+	REQUIRE(sock->parent != NULL);
+	REQUIRE(sock->tid == isc_nm_tid());
+
+#if HAVE_DECL_UV_UDP_RECVMMSG
+	uv_init_flags |= UV_UDP_RECVMMSG;
+#endif
+	r = uv_udp_init_ex(&worker->loop, &sock->uv_handle.udp, uv_init_flags);
+	UV_RUNTIME_CHECK(uv_udp_init_ex, r);
+	uv_handle_set_data(&sock->uv_handle.handle, sock);
+	/* This keeps the socket alive after everything else is gone */
+	isc__nmsocket_attach(sock, &(isc_nmsocket_t *){ NULL });
+
+	r = uv_timer_init(&worker->loop, &sock->read_timer);
+	UV_RUNTIME_CHECK(uv_timer_init, r);
+	uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
+
+	LOCK(&sock->parent->lock);
+
+	r = uv_udp_open(&sock->uv_handle.udp, sock->fd);
+	if (r < 0) {
+		isc__nm_closesocket(sock->fd);
+		isc__nm_incstats(sock->mgr, sock->statsindex[STATID_OPENFAIL]);
+		goto done;
+	}
+	isc__nm_incstats(sock->mgr, sock->statsindex[STATID_OPEN]);
+
+	if (sa_family == AF_INET6) {
+		uv_bind_flags |= UV_UDP_IPV6ONLY;
+	}
+
+#ifdef _WIN32
+	r = isc_uv_udp_freebind(&sock->uv_handle.udp,
+				&sock->parent->iface.type.sa, uv_bind_flags);
+	if (r < 0) {
+		isc__nm_incstats(sock->mgr, sock->statsindex[STATID_BINDFAIL]);
+		goto done;
+	}
+#else
+	if (mgr->load_balance_sockets) {
+		r = isc_uv_udp_freebind(&sock->uv_handle.udp,
+					&sock->parent->iface.type.sa,
+					uv_bind_flags);
+		if (r < 0) {
+			isc__nm_incstats(sock->mgr,
+					 sock->statsindex[STATID_BINDFAIL]);
+			goto done;
+		}
+	} else {
+		if (sock->parent->fd == -1) {
+			/* This thread is first, bind the socket */
+			r = isc_uv_udp_freebind(&sock->uv_handle.udp,
+						&sock->parent->iface.type.sa,
+						uv_bind_flags);
+			if (r < 0) {
+				isc__nm_incstats(sock->mgr, STATID_BINDFAIL);
+				goto done;
+			}
+			sock->parent->uv_handle.udp.flags =
+				sock->uv_handle.udp.flags;
+			sock->parent->fd = sock->fd;
+		} else {
+			/* The socket is already bound, just copy the flags */
+			sock->uv_handle.udp.flags =
+				sock->parent->uv_handle.udp.flags;
+		}
+	}
+#endif
+
+#ifdef ISC_RECV_BUFFER_SIZE
+	uv_recv_buffer_size(&sock->uv_handle.handle,
+			    &(int){ ISC_RECV_BUFFER_SIZE });
+#endif
+#ifdef ISC_SEND_BUFFER_SIZE
+	uv_send_buffer_size(&sock->uv_handle.handle,
+			    &(int){ ISC_SEND_BUFFER_SIZE });
+#endif
+	r = uv_udp_recv_start(&sock->uv_handle.udp, isc__nm_alloc_cb,
+			      udp_recv_cb);
+	if (r != 0) {
+		isc__nm_incstats(sock->mgr, sock->statsindex[STATID_BINDFAIL]);
+		goto done;
+	}
+
+	atomic_store(&sock->listening, true);
+
+done:
+	result = isc__nm_uverr2result(r);
+	atomic_fetch_add(&sock->parent->rchildren, 1);
+	if (sock->parent->result == ISC_R_UNSET) {
+		sock->parent->result = result;
+	}
+	SIGNAL(&sock->parent->cond);
+	UNLOCK(&sock->parent->lock);
+
+	isc_barrier_wait(&sock->parent->startlistening);
+}
+
+void
+isc__nm_udp_stoplistening(isc_nmsocket_t *sock) {
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->type == isc_nm_udplistener);
+
+	if (!atomic_compare_exchange_strong(&sock->closing, &(bool){ false },
+					    true))
+	{
+		UNREACHABLE();
+	}
+
+	if (!isc__nm_in_netthread()) {
+		enqueue_stoplistening(sock);
+	} else {
+		stop_udp_parent(sock);
+	}
+}
+
+/*
+ * Asynchronous 'udpstop' call handler: stop listening on a UDP socket.
+ */
+void
+isc__nm_async_udpstop(isc__networker_t *worker, isc__netievent_t *ev0) {
+	isc__netievent_udpstop_t *ievent = (isc__netievent_udpstop_t *)ev0;
+	isc_nmsocket_t *sock = ievent->sock;
+
+	UNUSED(worker);
+
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->tid == isc_nm_tid());
+
+	if (sock->parent != NULL) {
+		stop_udp_child(sock);
+		return;
+	}
+
+	stop_udp_parent(sock);
+}
+
+/*
+ * udp_recv_cb handles incoming UDP packet from uv.  The buffer here is
+ * reused for a series of packets, so we need to allocate a new one.
+ * This new one can be reused to send the response then.
+ */
+static void
+udp_recv_cb(uv_udp_t *handle, ssize_t nrecv, const uv_buf_t *buf,
+	    const struct sockaddr *addr, unsigned flags) {
+	isc_nmsocket_t *sock = uv_handle_get_data((uv_handle_t *)handle);
+	isc__nm_uvreq_t *req = NULL;
+	uint32_t maxudp;
+	isc_sockaddr_t sockaddr;
+	isc_result_t result;
+
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->tid == isc_nm_tid());
+	REQUIRE(sock->reading);
+
+	/*
+	 * When using recvmmsg(2), if no errors occur, there will be a final
+	 * callback with nrecv set to 0, addr set to NULL and the buffer
+	 * pointing at the initially allocated data with the UV_UDP_MMSG_CHUNK
+	 * flag cleared and the UV_UDP_MMSG_FREE flag set.
+	 */
+#if HAVE_DECL_UV_UDP_MMSG_FREE
+	if ((flags & UV_UDP_MMSG_FREE) == UV_UDP_MMSG_FREE) {
+		INSIST(nrecv == 0);
+		INSIST(addr == NULL);
+		goto free;
+	}
+#else
+	UNUSED(flags);
+#endif
+
+	/*
+	 * - If we're simulating a firewall blocking UDP packets
+	 *   bigger than 'maxudp' bytes for testing purposes.
+	 */
+	maxudp = atomic_load(&sock->mgr->maxudp);
+	if ((maxudp != 0 && (uint32_t)nrecv > maxudp)) {
+		/*
+		 * We need to keep the read_cb intact in case, so the
+		 * readtimeout_cb can trigger and not crash because of
+		 * missing read_req.
+		 */
+		goto free;
+	}
+
+	/*
+	 * - If addr == NULL, in which case it's the end of stream;
+	 *   we can free the buffer and bail.
+	 */
+	if (addr == NULL) {
+		isc__nm_failed_read_cb(sock, ISC_R_EOF, false);
+		goto free;
+	}
+
+	/*
+	 * - If the socket is no longer active.
+	 */
+	if (!isc__nmsocket_active(sock)) {
+		isc__nm_failed_read_cb(sock, ISC_R_CANCELED, false);
+		goto free;
+	}
+
+	if (nrecv < 0) {
+		isc__nm_failed_read_cb(sock, isc__nm_uverr2result(nrecv),
+				       false);
+		goto free;
+	}
+
+	result = isc_sockaddr_fromsockaddr(&sockaddr, addr);
+	RUNTIME_CHECK(result == ISC_R_SUCCESS);
+
+	req = isc__nm_get_read_req(sock, &sockaddr);
+
+	/*
+	 * The callback will be called synchronously, because result is
+	 * ISC_R_SUCCESS, so we are ok of passing the buf directly.
+	 */
+	req->uvbuf.base = buf->base;
+	req->uvbuf.len = nrecv;
+
+	sock->recv_read = false;
+
+	REQUIRE(!sock->processing);
+	sock->processing = true;
+	isc__nm_readcb(sock, req, ISC_R_SUCCESS);
+	sock->processing = false;
+
+free:
+#if HAVE_DECL_UV_UDP_MMSG_CHUNK
+	/*
+	 * When using recvmmsg(2), chunks will have the UV_UDP_MMSG_CHUNK flag
+	 * set, those must not be freed.
+	 */
+	if ((flags & UV_UDP_MMSG_CHUNK) == UV_UDP_MMSG_CHUNK) {
+		return;
+	}
+#endif
+
+	/*
+	 * When using recvmmsg(2), if a UDP socket error occurs, nrecv will be <
+	 * 0. In either scenario, the callee can now safely free the provided
+	 * buffer.
+	 */
+	if (nrecv < 0) {
+		/*
+		 * The buffer may be a null buffer on error.
+		 */
+		if (buf->base == NULL && buf->len == 0) {
+			return;
+		}
+	}
+
+	isc__nm_free_uvbuf(sock, buf);
+}
+
+/*
+ * Send the data in 'region' to a peer via a UDP socket. We try to find
+ * a proper sibling/child socket so that we won't have to jump to
+ * another thread.
+ */
+void
+isc__nm_udp_send(isc_nmhandle_t *handle, const isc_region_t *region,
+		 isc_nm_cb_t cb, void *cbarg) {
+	isc_nmsocket_t *sock = handle->sock;
+	isc_nmsocket_t *rsock = NULL;
+	isc_sockaddr_t *peer = &handle->peer;
+	isc__nm_uvreq_t *uvreq = NULL;
+	uint32_t maxudp = atomic_load(&sock->mgr->maxudp);
+	int ntid;
+
+	INSIST(sock->type == isc_nm_udpsocket);
+
+	/*
+	 * We're simulating a firewall blocking UDP packets bigger than
+	 * 'maxudp' bytes, for testing purposes.
+	 *
+	 * The client would ordinarily have unreferenced the handle
+	 * in the callback, but that won't happen in this case, so
+	 * we need to do so here.
+	 */
+	if (maxudp != 0 && region->length > maxudp) {
+		isc_nmhandle_detach(&handle);
+		return;
+	}
+
+	if (atomic_load(&sock->client)) {
+		/*
+		 * When we are sending from the client socket, we directly use
+		 * the socket provided.
+		 */
+		rsock = sock;
+		goto send;
+	} else {
+		/*
+		 * When we are sending from the server socket, we either use the
+		 * socket associated with the network thread we are in, or we
+		 * use the thread from the socket associated with the handle.
+		 */
+		INSIST(sock->parent != NULL);
+
+#if defined(WIN32)
+		/* On Windows, we have only a single listening listener */
+		rsock = sock;
+#else
+		if (isc__nm_in_netthread()) {
+			ntid = isc_nm_tid();
+		} else {
+			ntid = sock->tid;
+		}
+		rsock = &sock->parent->children[ntid];
+#endif
+	}
+
+send:
+	uvreq = isc__nm_uvreq_get(rsock->mgr, rsock);
+	uvreq->uvbuf.base = (char *)region->base;
+	uvreq->uvbuf.len = region->length;
+
+	isc_nmhandle_attach(handle, &uvreq->handle);
+
+	uvreq->cb.send = cb;
+	uvreq->cbarg = cbarg;
+
+	if (isc_nm_tid() == rsock->tid) {
+		REQUIRE(rsock->tid == isc_nm_tid());
+		isc__netievent_udpsend_t ievent = { .sock = rsock,
+						    .req = uvreq,
+						    .peer = *peer };
+
+		isc__nm_async_udpsend(NULL, (isc__netievent_t *)&ievent);
+	} else {
+		isc__netievent_udpsend_t *ievent =
+			isc__nm_get_netievent_udpsend(sock->mgr, rsock);
+		ievent->peer = *peer;
+		ievent->req = uvreq;
+
+		isc__nm_enqueue_ievent(&sock->mgr->workers[rsock->tid],
+				       (isc__netievent_t *)ievent);
+	}
+}
+
+/*
+ * Asynchronous 'udpsend' event handler: send a packet on a UDP socket.
+ */
+void
+isc__nm_async_udpsend(isc__networker_t *worker, isc__netievent_t *ev0) {
+	isc_result_t result;
+	isc__netievent_udpsend_t *ievent = (isc__netievent_udpsend_t *)ev0;
+	isc_nmsocket_t *sock = ievent->sock;
+	isc__nm_uvreq_t *uvreq = ievent->req;
+
+	REQUIRE(sock->type == isc_nm_udpsocket);
+	REQUIRE(sock->tid == isc_nm_tid());
+	UNUSED(worker);
+
+	if (isc__nmsocket_closing(sock)) {
+		isc__nm_failed_send_cb(sock, uvreq, ISC_R_CANCELED);
+		return;
+	}
+
+	result = udp_send_direct(sock, uvreq, &ievent->peer);
+	if (result != ISC_R_SUCCESS) {
+		isc__nm_incstats(sock->mgr, sock->statsindex[STATID_SENDFAIL]);
+		isc__nm_failed_send_cb(sock, uvreq, result);
+	}
+}
+
+static void
+udp_send_cb(uv_udp_send_t *req, int status) {
+	isc_result_t result = ISC_R_SUCCESS;
+	isc__nm_uvreq_t *uvreq = uv_handle_get_data((uv_handle_t *)req);
+	isc_nmsocket_t *sock = NULL;
+
+	REQUIRE(VALID_UVREQ(uvreq));
+	REQUIRE(VALID_NMHANDLE(uvreq->handle));
+
+	sock = uvreq->sock;
+
+	REQUIRE(sock->tid == isc_nm_tid());
+
+	if (status < 0) {
+		result = isc__nm_uverr2result(status);
+		isc__nm_incstats(sock->mgr, sock->statsindex[STATID_SENDFAIL]);
+	}
+
+	isc__nm_sendcb(sock, uvreq, result, false);
+}
+
+/*
+ * udp_send_direct sends buf to a peer on a socket. Sock has to be in
+ * the same thread as the callee.
+ */
+static isc_result_t
+udp_send_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req,
+		isc_sockaddr_t *peer) {
+	const struct sockaddr *sa = &peer->type.sa;
+	int r;
+
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(VALID_UVREQ(req));
+	REQUIRE(sock->tid == isc_nm_tid());
+	REQUIRE(sock->type == isc_nm_udpsocket);
+
+	if (isc__nmsocket_closing(sock)) {
+		return (ISC_R_CANCELED);
+	}
+
+#if UV_VERSION_HEX >= UV_VERSION(1, 27, 0)
+	/*
+	 * If we used uv_udp_connect() (and not the shim version for
+	 * older versions of libuv), then the peer address has to be
+	 * set to NULL or else uv_udp_send() could fail or assert,
+	 * depending on the libuv version.
+	 */
+	if (atomic_load(&sock->connected)) {
+		sa = NULL;
+	}
+#endif
+
+	r = uv_udp_send(&req->uv_req.udp_send, &sock->uv_handle.udp,
+			&req->uvbuf, 1, sa, udp_send_cb);
+	if (r < 0) {
+		return (isc__nm_uverr2result(r));
+	}
+
+	return (ISC_R_SUCCESS);
+}
+
+static isc_result_t
+udp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
+	isc__networker_t *worker = NULL;
+	int uv_bind_flags = UV_UDP_REUSEADDR;
+	isc_result_t result = ISC_R_UNSET;
+	int tries = 3;
+	int r;
+
+	REQUIRE(isc__nm_in_netthread());
+	REQUIRE(sock->tid == isc_nm_tid());
+
+	worker = &sock->mgr->workers[isc_nm_tid()];
+
+	atomic_store(&sock->connecting, true);
+
+	r = uv_udp_init(&worker->loop, &sock->uv_handle.udp);
+	UV_RUNTIME_CHECK(uv_udp_init, r);
+	uv_handle_set_data(&sock->uv_handle.handle, sock);
+
+	r = uv_timer_init(&worker->loop, &sock->read_timer);
+	UV_RUNTIME_CHECK(uv_timer_init, r);
+	uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
+
+	r = uv_udp_open(&sock->uv_handle.udp, sock->fd);
+	if (r != 0) {
+		isc__nm_incstats(sock->mgr, sock->statsindex[STATID_OPENFAIL]);
+		goto done;
+	}
+	isc__nm_incstats(sock->mgr, sock->statsindex[STATID_OPEN]);
+
+	if (sock->iface.type.sa.sa_family == AF_INET6) {
+		uv_bind_flags |= UV_UDP_IPV6ONLY;
+	}
+
+	r = uv_udp_bind(&sock->uv_handle.udp, &sock->iface.type.sa,
+			uv_bind_flags);
+	if (r != 0) {
+		isc__nm_incstats(sock->mgr, sock->statsindex[STATID_BINDFAIL]);
+		goto done;
+	}
+
+#ifdef ISC_RECV_BUFFER_SIZE
+	uv_recv_buffer_size(&sock->uv_handle.handle,
+			    &(int){ ISC_RECV_BUFFER_SIZE });
+#endif
+#ifdef ISC_SEND_BUFFER_SIZE
+	uv_send_buffer_size(&sock->uv_handle.handle,
+			    &(int){ ISC_SEND_BUFFER_SIZE });
+#endif
+
+	/*
+	 * On FreeBSD the UDP connect() call sometimes results in a
+	 * spurious transient EADDRINUSE. Try a few more times before
+	 * giving up.
+	 */
+	do {
+		r = isc_uv_udp_connect(&sock->uv_handle.udp,
+				       &req->peer.type.sa);
+	} while (r == UV_EADDRINUSE && --tries > 0);
+	if (r != 0) {
+		isc__nm_incstats(sock->mgr,
+				 sock->statsindex[STATID_CONNECTFAIL]);
+		goto done;
+	}
+	isc__nm_incstats(sock->mgr, sock->statsindex[STATID_CONNECT]);
+
+	atomic_store(&sock->connecting, false);
+	atomic_store(&sock->connected, true);
+
+done:
+	result = isc__nm_uverr2result(r);
+
+	LOCK(&sock->lock);
+	sock->result = result;
+	SIGNAL(&sock->cond);
+	if (!atomic_load(&sock->active)) {
+		WAIT(&sock->scond, &sock->lock);
+	}
+	INSIST(atomic_load(&sock->active));
+	UNLOCK(&sock->lock);
+
+	return (result);
+}
+
+/*
+ * Asynchronous 'udpconnect' call handler: open a new UDP socket and
+ * call the 'open' callback with a handle.
+ */
+void
+isc__nm_async_udpconnect(isc__networker_t *worker, isc__netievent_t *ev0) {
+	isc__netievent_udpconnect_t *ievent =
+		(isc__netievent_udpconnect_t *)ev0;
+	isc_nmsocket_t *sock = ievent->sock;
+	isc__nm_uvreq_t *req = ievent->req;
+	isc_result_t result;
+
+	UNUSED(worker);
+
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->type == isc_nm_udpsocket);
+	REQUIRE(sock->parent == NULL);
+	REQUIRE(sock->tid == isc_nm_tid());
+
+	result = udp_connect_direct(sock, req);
+	if (result != ISC_R_SUCCESS) {
+		atomic_store(&sock->active, false);
+		isc__nm_udp_close(sock);
+		isc__nm_connectcb(sock, req, result, true);
+	} else {
+		/*
+		 * The callback has to be called after the socket has been
+		 * initialized
+		 */
+		isc__nm_connectcb(sock, req, ISC_R_SUCCESS, true);
+	}
+
+	/*
+	 * The sock is now attached to the handle.
+	 */
+	isc__nmsocket_detach(&sock);
+}
+
+void
+isc_nm_udpconnect(isc_nm_t *mgr, isc_sockaddr_t *local, isc_sockaddr_t *peer,
+		  isc_nm_cb_t cb, void *cbarg, unsigned int timeout,
+		  size_t extrahandlesize) {
+	isc_result_t result = ISC_R_SUCCESS;
+	isc_nmsocket_t *sock = NULL;
+	isc__netievent_udpconnect_t *event = NULL;
+	isc__nm_uvreq_t *req = NULL;
+	sa_family_t sa_family;
+
+	REQUIRE(VALID_NM(mgr));
+	REQUIRE(local != NULL);
+	REQUIRE(peer != NULL);
+
+	sa_family = peer->type.sa.sa_family;
+
+	sock = isc_mem_get(mgr->mctx, sizeof(isc_nmsocket_t));
+	isc__nmsocket_init(sock, mgr, isc_nm_udpsocket, local);
+
+	sock->connect_cb = cb;
+	sock->connect_cbarg = cbarg;
+	sock->read_timeout = timeout;
+	sock->extrahandlesize = extrahandlesize;
+	sock->peer = *peer;
+	sock->result = ISC_R_UNSET;
+	atomic_init(&sock->client, true);
+
+	req = isc__nm_uvreq_get(mgr, sock);
+	req->cb.connect = cb;
+	req->cbarg = cbarg;
+	req->peer = *peer;
+	req->local = *local;
+	req->handle = isc__nmhandle_get(sock, &req->peer, &sock->iface);
+
+	result = isc__nm_socket(sa_family, SOCK_DGRAM, 0, &sock->fd);
+	if (result != ISC_R_SUCCESS) {
+		if (isc__nm_in_netthread()) {
+			sock->tid = isc_nm_tid();
+		}
+		isc__nmsocket_clearcb(sock);
+		isc__nm_connectcb(sock, req, result, true);
+		atomic_store(&sock->closed, true);
+		isc__nmsocket_detach(&sock);
+		return;
+	}
+
+	result = isc__nm_socket_reuse(sock->fd);
+	RUNTIME_CHECK(result == ISC_R_SUCCESS ||
+		      result == ISC_R_NOTIMPLEMENTED);
+
+	result = isc__nm_socket_reuse_lb(sock->fd);
+	RUNTIME_CHECK(result == ISC_R_SUCCESS ||
+		      result == ISC_R_NOTIMPLEMENTED);
+
+	(void)isc__nm_socket_incoming_cpu(sock->fd);
+
+	(void)isc__nm_socket_disable_pmtud(sock->fd, sa_family);
+
+	event = isc__nm_get_netievent_udpconnect(mgr, sock, req);
+
+	if (isc__nm_in_netthread()) {
+		atomic_store(&sock->active, true);
+		sock->tid = isc_nm_tid();
+		isc__nm_async_udpconnect(&mgr->workers[sock->tid],
+					 (isc__netievent_t *)event);
+		isc__nm_put_netievent_udpconnect(mgr, event);
+	} else {
+		atomic_init(&sock->active, false);
+		sock->tid = isc_random_uniform(mgr->nworkers);
+		isc__nm_enqueue_ievent(&mgr->workers[sock->tid],
+				       (isc__netievent_t *)event);
+	}
+	LOCK(&sock->lock);
+	while (sock->result == ISC_R_UNSET) {
+		WAIT(&sock->cond, &sock->lock);
+	}
+	atomic_store(&sock->active, true);
+	BROADCAST(&sock->scond);
+	UNLOCK(&sock->lock);
+}
+
+void
+isc__nm_udp_read_cb(uv_udp_t *handle, ssize_t nrecv, const uv_buf_t *buf,
+		    const struct sockaddr *addr, unsigned flags) {
+	isc_nmsocket_t *sock = uv_handle_get_data((uv_handle_t *)handle);
+	REQUIRE(VALID_NMSOCK(sock));
+
+	udp_recv_cb(handle, nrecv, buf, addr, flags);
+	/*
+	 * If a caller calls isc_nm_read() on a listening socket, we can
+	 * get here, but we MUST NOT stop reading from the listener
+	 * socket.  The only difference between listener and connected
+	 * sockets is that the former has sock->parent set and later
+	 * does not.
+	 */
+	if (!sock->parent) {
+		isc__nm_stop_reading(sock);
+	}
+}
+
+void
+isc__nm_udp_failed_read_cb(isc_nmsocket_t *sock, isc_result_t result) {
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(result != ISC_R_SUCCESS);
+
+	if (atomic_load(&sock->client)) {
+		isc__nmsocket_timer_stop(sock);
+		isc__nm_stop_reading(sock);
+
+		if (!sock->recv_read) {
+			goto destroy;
+		}
+		sock->recv_read = false;
+
+		if (sock->recv_cb != NULL) {
+			isc__nm_uvreq_t *req = isc__nm_get_read_req(sock, NULL);
+			isc__nmsocket_clearcb(sock);
+			isc__nm_readcb(sock, req, result);
+		}
+
+	destroy:
+		isc__nmsocket_prep_destroy(sock);
+		return;
+	}
+
+	/*
+	 * For UDP server socket, we don't have child socket via
+	 * "accept", so we:
+	 * - we continue to read
+	 * - we don't clear the callbacks
+	 * - we don't destroy it (only stoplistening could do that)
+	 */
+	if (!sock->recv_read) {
+		return;
+	}
+	sock->recv_read = false;
+
+	if (sock->recv_cb != NULL) {
+		isc__nm_uvreq_t *req = isc__nm_get_read_req(sock, NULL);
+		isc__nm_readcb(sock, req, result);
+	}
+}
+
+/*
+ * Asynchronous 'udpread' call handler: start or resume reading on a
+ * socket; pause reading and call the 'recv' callback after each
+ * datagram.
+ */
+void
+isc__nm_async_udpread(isc__networker_t *worker, isc__netievent_t *ev0) {
+	isc__netievent_udpread_t *ievent = (isc__netievent_udpread_t *)ev0;
+	isc_nmsocket_t *sock = ievent->sock;
+	isc_result_t result;
+
+	UNUSED(worker);
+
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->tid == isc_nm_tid());
+
+	if (isc__nmsocket_closing(sock)) {
+		result = ISC_R_CANCELED;
+	} else {
+		result = isc__nm_start_reading(sock);
+	}
+
+	if (result != ISC_R_SUCCESS) {
+		sock->reading = true;
+		isc__nm_failed_read_cb(sock, result, false);
+		return;
+	}
+
+	isc__nmsocket_timer_start(sock);
+}
+
+void
+isc__nm_udp_read(isc_nmhandle_t *handle, isc_nm_recv_cb_t cb, void *cbarg) {
+	REQUIRE(VALID_NMHANDLE(handle));
+	REQUIRE(VALID_NMSOCK(handle->sock));
+
+	isc_nmsocket_t *sock = handle->sock;
+
+	REQUIRE(sock->type == isc_nm_udpsocket);
+	REQUIRE(sock->statichandle == handle);
+	REQUIRE(sock->tid == isc_nm_tid());
+	REQUIRE(!sock->recv_read);
+
+	sock->recv_cb = cb;
+	sock->recv_cbarg = cbarg;
+	sock->recv_read = true;
+
+	if (!sock->reading && sock->tid == isc_nm_tid()) {
+		isc__netievent_udpread_t ievent = { .sock = sock };
+		isc__nm_async_udpread(NULL, (isc__netievent_t *)&ievent);
+	} else {
+		isc__netievent_udpread_t *ievent =
+			isc__nm_get_netievent_udpread(sock->mgr, sock);
+		isc__nm_enqueue_ievent(&sock->mgr->workers[sock->tid],
+				       (isc__netievent_t *)ievent);
+	}
+}
+
+static void
+udp_stop_cb(uv_handle_t *handle) {
+	isc_nmsocket_t *sock = uv_handle_get_data(handle);
+	uv_handle_set_data(handle, NULL);
+
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->tid == isc_nm_tid());
+	REQUIRE(atomic_load(&sock->closing));
+
+	if (!atomic_compare_exchange_strong(&sock->closed, &(bool){ false },
+					    true))
+	{
+		UNREACHABLE();
+	}
+
+	isc__nm_incstats(sock->mgr, sock->statsindex[STATID_CLOSE]);
+
+	atomic_store(&sock->listening, false);
+
+	isc__nmsocket_detach(&sock);
+}
+
+static void
+udp_close_cb(uv_handle_t *handle) {
+	isc_nmsocket_t *sock = uv_handle_get_data(handle);
+	uv_handle_set_data(handle, NULL);
+
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->tid == isc_nm_tid());
+	REQUIRE(atomic_load(&sock->closing));
+
+	if (!atomic_compare_exchange_strong(&sock->closed, &(bool){ false },
+					    true))
+	{
+		UNREACHABLE();
+	}
+
+	isc__nm_incstats(sock->mgr, sock->statsindex[STATID_CLOSE]);
+
+	if (sock->server != NULL) {
+		isc__nmsocket_detach(&sock->server);
+	}
+
+	atomic_store(&sock->connected, false);
+	atomic_store(&sock->listening, false);
+
+	isc__nmsocket_prep_destroy(sock);
+}
+
+static void
+read_timer_close_cb(uv_handle_t *handle) {
+	isc_nmsocket_t *sock = uv_handle_get_data(handle);
+	uv_handle_set_data(handle, NULL);
+
+	if (sock->parent) {
+		uv_close(&sock->uv_handle.handle, udp_stop_cb);
+	} else {
+		uv_close(&sock->uv_handle.handle, udp_close_cb);
+	}
+}
+
+static void
+stop_udp_child(isc_nmsocket_t *sock) {
+	REQUIRE(sock->type == isc_nm_udpsocket);
+	REQUIRE(sock->tid == isc_nm_tid());
+
+	if (!atomic_compare_exchange_strong(&sock->closing, &(bool){ false },
+					    true))
+	{
+		return;
+	}
+
+	udp_close_direct(sock);
+
+	atomic_fetch_sub(&sock->parent->rchildren, 1);
+
+	isc_barrier_wait(&sock->parent->stoplistening);
+}
+
+static void
+stop_udp_parent(isc_nmsocket_t *sock) {
+	isc_nmsocket_t *csock = NULL;
+
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->tid == isc_nm_tid());
+	REQUIRE(sock->type == isc_nm_udplistener);
+
+	isc_barrier_init(&sock->stoplistening, sock->nchildren);
+
+	for (size_t i = 0; i < sock->nchildren; i++) {
+		csock = &sock->children[i];
+		REQUIRE(VALID_NMSOCK(csock));
+
+		if ((int)i == isc_nm_tid()) {
+			/*
+			 * We need to schedule closing the other sockets first
+			 */
+			continue;
+		}
+
+		atomic_store(&csock->active, false);
+		enqueue_stoplistening(csock);
+	}
+
+	csock = &sock->children[isc_nm_tid()];
+	atomic_store(&csock->active, false);
+	stop_udp_child(csock);
+
+	atomic_store(&sock->closed, true);
+	isc__nmsocket_prep_destroy(sock);
+}
+
+static void
+udp_close_direct(isc_nmsocket_t *sock) {
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->tid == isc_nm_tid());
+
+	uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
+	uv_close((uv_handle_t *)&sock->read_timer, read_timer_close_cb);
+}
+
+void
+isc__nm_async_udpclose(isc__networker_t *worker, isc__netievent_t *ev0) {
+	isc__netievent_udpclose_t *ievent = (isc__netievent_udpclose_t *)ev0;
+	isc_nmsocket_t *sock = ievent->sock;
+
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->tid == isc_nm_tid());
+	UNUSED(worker);
+
+	udp_close_direct(sock);
+}
+
+void
+isc__nm_udp_close(isc_nmsocket_t *sock) {
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->type == isc_nm_udpsocket);
+	REQUIRE(!isc__nmsocket_active(sock));
+
+	if (!atomic_compare_exchange_strong(&sock->closing, &(bool){ false },
+					    true))
+	{
+		return;
+	}
+
+	if (sock->tid == isc_nm_tid()) {
+		udp_close_direct(sock);
+	} else {
+		isc__netievent_udpclose_t *ievent =
+			isc__nm_get_netievent_udpclose(sock->mgr, sock);
+		isc__nm_enqueue_ievent(&sock->mgr->workers[sock->tid],
+				       (isc__netievent_t *)ievent);
+	}
+}
+
+void
+isc__nm_udp_shutdown(isc_nmsocket_t *sock) {
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->tid == isc_nm_tid());
+	REQUIRE(sock->type == isc_nm_udpsocket);
+
+	/*
+	 * If the socket is active, mark it inactive and
+	 * continue. If it isn't active, stop now.
+	 */
+	if (!isc__nmsocket_deactivate(sock)) {
+		return;
+	}
+
+	/*
+	 * If the socket is connecting, the cancel will happen in the
+	 * async_udpconnect() due socket being inactive now.
+	 */
+	if (atomic_load(&sock->connecting)) {
+		return;
+	}
+
+	/*
+	 * When the client detaches the last handle, the
+	 * sock->statichandle would be NULL, in that case, nobody is
+	 * interested in the callback.
+	 */
+	if (sock->statichandle != NULL) {
+		isc__nm_failed_read_cb(sock, ISC_R_CANCELED, false);
+		return;
+	}
+
+	/*
+	 * Otherwise, we just send the socket to abyss...
+	 */
+	if (sock->parent == NULL) {
+		isc__nmsocket_prep_destroy(sock);
+	}
+}
+
+void
+isc__nm_udp_cancelread(isc_nmhandle_t *handle) {
+	isc_nmsocket_t *sock = NULL;
+	isc__netievent_udpcancel_t *ievent = NULL;
+
+	REQUIRE(VALID_NMHANDLE(handle));
+
+	sock = handle->sock;
+
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->type == isc_nm_udpsocket);
+
+	ievent = isc__nm_get_netievent_udpcancel(sock->mgr, sock, handle);
+
+	isc__nm_enqueue_ievent(&sock->mgr->workers[sock->tid],
+			       (isc__netievent_t *)ievent);
+}
+
+void
+isc__nm_async_udpcancel(isc__networker_t *worker, isc__netievent_t *ev0) {
+	isc__netievent_udpcancel_t *ievent = (isc__netievent_udpcancel_t *)ev0;
+	isc_nmsocket_t *sock = NULL;
+
+	UNUSED(worker);
+
+	REQUIRE(VALID_NMSOCK(ievent->sock));
+
+	sock = ievent->sock;
+
+	REQUIRE(sock->tid == isc_nm_tid());
+	REQUIRE(atomic_load(&sock->client));
+
+	isc__nm_failed_read_cb(sock, ISC_R_EOF, false);
+}