diff --git a/lib/isc/netmgr/netmgr-int.h b/lib/isc/netmgr/netmgr-int.h index a17304418e..884dff6e7b 100644 --- a/lib/isc/netmgr/netmgr-int.h +++ b/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). */ diff --git a/lib/isc/netmgr/netmgr.c b/lib/isc/netmgr/netmgr.c index fd0dff0455..1b692c5cd9 100644 --- a/lib/isc/netmgr/netmgr.c +++ b/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; diff --git a/lib/isc/netmgr/tcp.c b/lib/isc/netmgr/tcp.c index 302c31f0c5..e34ecd722e 100644 --- a/lib/isc/netmgr/tcp.c +++ b/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; diff --git a/lib/isc/netmgr/udp.c b/lib/isc/netmgr/udp.c index 103245c9f9..31ea706112 100644 --- a/lib/isc/netmgr/udp.c +++ b/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); + } +}