WIP: add udp_cancelread() and udp_recv timeout support

lib/isc/netmgr/netmgr-int.h

@@ -49,6 +49,8 @@
 #define ISC_NETMGR_RECVBUF_SIZE (65536)
 #endif
 
+#define NETMGR_TRACE 1
+
 /*
  * Define NETMGR_TRACE to activate tracing of handles and sockets.
  * This will impair performance but enables us to quickly determine,
@@ -140,6 +142,7 @@ typedef enum isc__netievent_type {
 	netievent_udpread,
 	netievent_udpstop,
 	netievent_udpclose,
+	netievent_udpcancelread,
 
 	netievent_tcpconnect,
 	netievent_tcpsend,
@@ -219,6 +222,7 @@ typedef isc__netievent__socket_t isc__netievent_udplisten_t;
 typedef isc__netievent__socket_t isc__netievent_udpread_t;
 typedef isc__netievent__socket_t isc__netievent_udpstop_t;
 typedef isc__netievent__socket_t isc__netievent_udpclose_t;
+typedef isc__netievent__socket_t isc__netievent_udpcancelread_t;
 typedef isc__netievent__socket_t isc__netievent_tcpstop_t;
 typedef isc__netievent__socket_t isc__netievent_tcpclose_t;
 typedef isc__netievent__socket_t isc__netievent_startread_t;
@@ -718,6 +722,12 @@ isc__nm_udp_close(isc_nmsocket_t *sock);
  * Close a UDP socket.
  */
 
+void
+isc__nm_udp_cancelread(isc_nmhandle_t *handle);
+/*%<
+ * Stop reading on a connected UDP handle.
+ */
+
 void
 isc__nm_udp_stoplistening(isc_nmsocket_t *sock);
 
@@ -734,6 +744,8 @@ void
 isc__nm_async_udpread(isc__networker_t *worker, isc__netievent_t *ev0);
 void
 isc__nm_async_udpclose(isc__networker_t *worker, isc__netievent_t *ev0);
+void
+isc__nm_async_udpcancelread(isc__networker_t *worker, isc__netievent_t *ev0);
 /*%<
  * Callback handlers for asynchronous UDP events (listen, stoplisten, send).
  */

lib/isc/netmgr/netmgr.c

@@ -622,6 +622,9 @@ process_queue(isc__networker_t *worker, isc_queue_t *queue) {
 		case netievent_udpclose:
 			isc__nm_async_udpclose(worker, ievent);
 			break;
+		case netievent_udpcancelread:
+			isc__nm_async_udpcancelread(worker, ievent);
+			break;
 
 		case netievent_tcpconnect:
 			isc__nm_async_tcpconnect(worker, ievent);
@@ -1455,6 +1458,9 @@ isc_nm_cancelread(isc_nmhandle_t *handle) {
 	REQUIRE(VALID_NMHANDLE(handle));
 
 	switch (handle->sock->type) {
+	case isc_nm_udpsocket:
+		isc__nm_udp_cancelread(handle);
+		break;
 	case isc_nm_tcpsocket:
 		isc__nm_tcp_cancelread(handle);
 		break;

lib/isc/netmgr/tcp.c

@@ -1252,6 +1252,8 @@ isc__nm_async_tcpcancel(isc__networker_t *worker, isc__netievent_t *ev0) {
 	REQUIRE(worker->id == sock->tid);
 	REQUIRE(sock->tid == isc_nm_tid());
 
+	uv_read_stop(&sock->uv_handle.stream);
+
 	if (atomic_load(&sock->client)) {
 		isc_nm_recv_cb_t cb;
 		void *cbarg = NULL;

lib/isc/netmgr/udp.c

@@ -604,6 +604,9 @@ isc__nm_async_udpconnect(isc__networker_t *worker, isc__netievent_t *ev0) {
 	REQUIRE(sock->parent == NULL);
 	REQUIRE(sock->tid == isc_nm_tid());
 
+	/* FIXME: this needs it's own timeout */
+	sock->read_timeout = sock->mgr->init;
+
 	uv_udp_init(&worker->loop, &sock->uv_handle.udp);
 	uv_handle_set_data(&sock->uv_handle.handle, NULL);
 	uv_handle_set_data(&sock->uv_handle.handle, sock);
@@ -758,6 +761,29 @@ udp_read_cb(uv_udp_t *handle, ssize_t nrecv, const uv_buf_t *buf,
 	uv_udp_recv_stop(&sock->uv_handle.udp);
 }
 
+static void
+readtimeout_cb(uv_timer_t *handle) {
+	isc_nmsocket_t *sock = uv_handle_get_data((uv_handle_t *)handle);
+	isc_nm_recv_cb_t cb;
+	void *cbarg = NULL;
+
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->tid == isc_nm_tid());
+
+	/*
+	 * Timeout; stop reading and process whatever we have.
+	 */
+	uv_udp_recv_stop(&sock->uv_handle.udp);
+
+	cb = sock->recv_cb;
+	cbarg = sock->recv_cbarg;
+	isc__nmsocket_clearcb(sock);
+
+	if (cb != NULL) {
+		cb(sock->statichandle, ISC_R_TIMEDOUT, NULL, cbarg);
+	}
+}
+
 /*
  * Asynchronous 'udpread' call handler: start or resume reading on a socket;
  * pause reading and call the 'recv' callback after each datagram.
@@ -767,7 +793,17 @@ 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;
 
-	UNUSED(worker);
+	REQUIRE(worker->id == isc_nm_tid());
+	if (sock->read_timeout != 0) {
+		if (!sock->timer_initialized) {
+			uv_timer_init(&worker->loop, &sock->timer);
+			uv_handle_set_data((uv_handle_t *)&sock->timer, sock);
+			sock->timer_initialized = true;
+		}
+		uv_timer_start(&sock->timer, readtimeout_cb, sock->read_timeout,
+			       0);
+		sock->timer_running = true;
+	}
 
 	uv_udp_recv_start(&sock->uv_handle.udp, udp_alloc_cb, udp_read_cb);
 }
@@ -842,3 +878,42 @@ isc__nm_udp_close(isc_nmsocket_t *sock) {
 				       (isc__netievent_t *)ievent);
 	}
 }
+
+void
+isc__nm_async_udpcancelread(isc__networker_t *worker, isc__netievent_t *ev0) {
+	isc__netievent_udpcancelread_t *ievent =
+		(isc__netievent_udpcancelread_t *)ev0;
+	isc_nmsocket_t *sock = ievent->sock;
+
+	REQUIRE(worker->id == ievent->sock->tid);
+
+	uv_udp_recv_stop(&sock->uv_handle.udp);
+
+	if (sock->recv_cb) {
+		sock->recv_cb(NULL, ISC_R_EOF, NULL, sock->recv_cbarg);
+	}
+}
+
+void
+isc__nm_udp_cancelread(isc_nmhandle_t *handle) {
+	isc_nmsocket_t *sock = NULL;
+	REQUIRE(VALID_NMHANDLE(handle));
+
+	sock = handle->sock;
+
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(sock->type == isc_nm_udpsocket);
+
+	isc__netievent_udpcancelread_t *ievent =
+		isc__nm_get_ievent(sock->mgr, netievent_udpcancelread);
+	ievent->sock = sock;
+
+	if (sock->tid == isc_nm_tid()) {
+		isc__nm_async_udpcancelread(&sock->mgr->workers[sock->tid],
+					    (isc__netievent_t *)ievent);
+		isc__nm_put_ievent(sock->mgr, ievent);
+	} else {
+		isc__nm_enqueue_ievent(&sock->mgr->workers[sock->tid],
+				       (isc__netievent_t *)ievent);
+	}
+}