Add TCP, TCPDNS and TLSDNS write timer

When the outgoing TCP write buffers are full because the other party is
not reading the data, the uv_write() could wait indefinitely on the
uv_loop and never calling the callback.  Add a new write timer that uses
the `tcp-idle-timeout` value to interrupt the TCP connection when we are
not able to send data for defined period of time.

(cherry picked from commit 408b3621696e39ac6dfe58be75fad168a37b31ff)

diff --git a/lib/isc/netmgr/netmgr-int.h b/lib/isc/netmgr/netmgr-int.h
index c4b8ca54762b65c5816dbef43a93c5c26087582f..3871d70939d7527e9c5b3254905fce957cc088c5 100644 (file)
--- a/lib/isc/netmgr/netmgr-int.h
+++ b/lib/isc/netmgr/netmgr-int.h
@@ -767,6 +767,13 @@ struct isc_nmsocket {
        uint64_t read_timeout;
        uint64_t connect_timeout;
 
+       /*%
+        * TCP write timeout timer.
+        */
+       uv_timer_t write_timer;
+       uint64_t write_timeout;
+       int64_t writes;
+
        /*% outer socket is for 'wrapped' sockets - e.g. tcpdns in tcp */
        isc_nmsocket_t *outer;
 
@@ -1588,11 +1595,24 @@ void
 isc__nm_failed_read_cb(isc_nmsocket_t *sock, isc_result_t result, bool async);
 
 void
-isc__nmsocket_connecttimeout_cb(uv_timer_t *timer);
+isc__nm_accept_connection_log(isc_result_t result, bool can_log_quota);
 
+/*
+ * Timeout callbacks
+ */
 void
-isc__nm_accept_connection_log(isc_result_t result, bool can_log_quota);
+isc__nmsocket_connecttimeout_cb(uv_timer_t *timer);
+void
+isc__nmsocket_readtimeout_cb(uv_timer_t *timer);
+void
+isc__nmsocket_writetimeout_cb(uv_timer_t *timer);
 
+/*%<
+ *
+ * Maximum number of simultaneous handles in flight supported for a single
+ * connected TCPDNS socket. This value was chosen arbitrarily, and may be
+ * changed in the future.
+ */
 #define STREAM_CLIENTS_PER_CONN 23
 
 #define UV_RUNTIME_CHECK(func, ret)                                           \

diff --git a/lib/isc/netmgr/netmgr.c b/lib/isc/netmgr/netmgr.c
index b2883892449b6990f87e754846ece3fa48c255c1..6bc8c6433767dd54ac430cfc4a8afa262a2af109 100644 (file)
--- a/lib/isc/netmgr/netmgr.c
+++ b/lib/isc/netmgr/netmgr.c
@@ -1969,7 +1969,21 @@ isc__nm_accept_connection_log(isc_result_t result, bool can_log_quota) {
                      isc_result_totext(result));
 }
 
-static void
+void
+isc__nmsocket_writetimeout_cb(uv_timer_t *timer) {
+       isc_nmsocket_t *sock = uv_handle_get_data((uv_handle_t *)timer);
+
+       int r = uv_timer_stop(&sock->write_timer);
+       UV_RUNTIME_CHECK(uv_timer_stop, r);
+
+       /* The shutdown will be handled in the respective close functions */
+       r = uv_tcp_close_reset(&sock->uv_handle.tcp, NULL);
+       UV_RUNTIME_CHECK(uv_tcp_close_reset, r);
+
+       isc__nmsocket_shutdown(sock);
+}
+
+void
 isc__nmsocket_readtimeout_cb(uv_timer_t *timer) {
        isc_nmsocket_t *sock = uv_handle_get_data((uv_handle_t *)timer);
 
@@ -2317,6 +2331,8 @@ isc_nmhandle_keepalive(isc_nmhandle_t *handle, bool value) {
                atomic_store(&sock->keepalive, value);
                sock->read_timeout = value ? atomic_load(&sock->mgr->keepalive)
                                           : atomic_load(&sock->mgr->idle);
+               sock->write_timeout = value ? atomic_load(&sock->mgr->keepalive)
+                                           : atomic_load(&sock->mgr->idle);
                break;
        default:
                /*

diff --git a/lib/isc/netmgr/tcp.c b/lib/isc/netmgr/tcp.c
index 52cc32f03976c1f942eabee231faf2dfcfbe6b5c..e562ef2d694c207dd1576879cf81219d113b0590 100644 (file)
--- a/lib/isc/netmgr/tcp.c
+++ b/lib/isc/netmgr/tcp.c
@@ -144,6 +144,10 @@ tcp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
        UV_RUNTIME_CHECK(uv_timer_init, r);
        uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
 
+       r = uv_timer_init(&worker->loop, &sock->write_timer);
+       UV_RUNTIME_CHECK(uv_timer_init, r);
+       uv_handle_set_data((uv_handle_t *)&sock->write_timer, sock);
+
        r = uv_tcp_open(&sock->uv_handle.tcp, sock->fd);
        if (r != 0) {
                isc__nm_closesocket(sock->fd);
@@ -532,6 +536,10 @@ isc__nm_async_tcplisten(isc__networker_t *worker, isc__netievent_t *ev0) {
        UV_RUNTIME_CHECK(uv_timer_init, r);
        uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
 
+       r = uv_timer_init(&worker->loop, &sock->write_timer);
+       UV_RUNTIME_CHECK(uv_timer_init, r);
+       uv_handle_set_data((uv_handle_t *)&sock->write_timer, sock);
+
        LOCK(&sock->parent->lock);
 
        r = uv_tcp_open(&sock->uv_handle.tcp, sock->fd);
@@ -981,6 +989,10 @@ accept_connection(isc_nmsocket_t *ssock, isc_quota_t *quota) {
        UV_RUNTIME_CHECK(uv_timer_init, r);
        uv_handle_set_data((uv_handle_t *)&csock->read_timer, csock);
 
+       r = uv_timer_init(&worker->loop, &csock->write_timer);
+       UV_RUNTIME_CHECK(uv_timer_init, r);
+       uv_handle_set_data((uv_handle_t *)&csock->write_timer, csock);
+
        r = uv_accept(&ssock->uv_handle.stream, &csock->uv_handle.stream);
        if (r != 0) {
                result = isc__nm_uverr2result(r);
@@ -1074,6 +1086,13 @@ isc__nm_tcp_send(isc_nmhandle_t *handle, const isc_region_t *region,
        uvreq->cb.send = cb;
        uvreq->cbarg = cbarg;
 
+       if (sock->write_timeout == 0) {
+               sock->write_timeout =
+                       (atomic_load(&sock->keepalive)
+                                ? atomic_load(&sock->mgr->keepalive)
+                                : atomic_load(&sock->mgr->idle));
+       }
+
        ievent = isc__nm_get_netievent_tcpsend(sock->mgr, sock, uvreq);
        isc__nm_maybe_enqueue_ievent(&sock->mgr->workers[sock->tid],
                                     (isc__netievent_t *)ievent);
@@ -1084,11 +1103,17 @@ isc__nm_tcp_send(isc_nmhandle_t *handle, const isc_region_t *region,
 static void
 tcp_send_cb(uv_write_t *req, int status) {
        isc__nm_uvreq_t *uvreq = (isc__nm_uvreq_t *)req->data;
+
        REQUIRE(VALID_UVREQ(uvreq));
        REQUIRE(VALID_NMHANDLE(uvreq->handle));
 
        isc_nmsocket_t *sock = uvreq->sock;
 
+       if (--sock->writes == 0) {
+               int r = uv_timer_stop(&sock->write_timer);
+               UV_RUNTIME_CHECK(uv_timer_stop, r);
+       }
+
        if (status < 0) {
                isc__nm_incstats(sock->mgr, sock->statsindex[STATID_SENDFAIL]);
                failed_send_cb(sock, uvreq, isc__nm_uverr2result(status));
@@ -1132,6 +1157,11 @@ tcp_send_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
                return (ISC_R_CANCELED);
        }
 
+       r = uv_timer_start(&sock->write_timer, isc__nmsocket_writetimeout_cb,
+                          sock->write_timeout, 0);
+       UV_RUNTIME_CHECK(uv_timer_start, r);
+       RUNTIME_CHECK(sock->writes++ >= 0);
+
        r = uv_write(&req->uv_req.write, &sock->uv_handle.stream, &req->uvbuf,
                     1, tcp_send_cb);
        if (r < 0) {
@@ -1195,7 +1225,7 @@ tcp_close_cb(uv_handle_t *handle) {
 }
 
 static void
-timer_close_cb(uv_handle_t *handle) {
+read_timer_close_cb(uv_handle_t *handle) {
        isc_nmsocket_t *sock = uv_handle_get_data(handle);
        uv_handle_set_data(handle, NULL);
 
@@ -1208,6 +1238,17 @@ timer_close_cb(uv_handle_t *handle) {
        }
 }
 
+static void
+write_timer_close_cb(uv_handle_t *timer) {
+       isc_nmsocket_t *sock = uv_handle_get_data(timer);
+       uv_handle_set_data(timer, NULL);
+
+       REQUIRE(VALID_NMSOCK(sock));
+
+       uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
+       uv_close((uv_handle_t *)&sock->read_timer, read_timer_close_cb);
+}
+
 static void
 stop_tcp_child(isc_nmsocket_t *sock) {
        REQUIRE(sock->type == isc_nm_tcpsocket);
@@ -1260,6 +1301,8 @@ stop_tcp_parent(isc_nmsocket_t *sock) {
 
 static void
 tcp_close_direct(isc_nmsocket_t *sock) {
+       int r;
+
        REQUIRE(VALID_NMSOCK(sock));
        REQUIRE(sock->tid == isc_nm_tid());
        REQUIRE(atomic_load(&sock->closing));
@@ -1281,8 +1324,10 @@ tcp_close_direct(isc_nmsocket_t *sock) {
        isc__nmsocket_timer_stop(sock);
        isc__nm_stop_reading(sock);
 
-       uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
-       uv_close((uv_handle_t *)&sock->read_timer, timer_close_cb);
+       r = uv_timer_stop(&sock->write_timer);
+       UV_RUNTIME_CHECK(uv_timer_stop, r);
+       uv_handle_set_data((uv_handle_t *)&sock->write_timer, sock);
+       uv_close((uv_handle_t *)&sock->write_timer, write_timer_close_cb);
 }
 
 void

diff --git a/lib/isc/netmgr/tcpdns.c b/lib/isc/netmgr/tcpdns.c
index 8f4ecfb9aca6291755e61f897c3a865c2c7e2bb0..61a8e6b710b3fc05ea0de1f3a9531750df7eb699 100644 (file)
--- a/lib/isc/netmgr/tcpdns.c
+++ b/lib/isc/netmgr/tcpdns.c
@@ -37,13 +37,6 @@
 #include "netmgr-int.h"
 #include "uv-compat.h"
 
-/*%<
- *
- * Maximum number of simultaneous handles in flight supported for a single
- * connected TCPDNS socket. This value was chosen arbitrarily, and may be
- * changed in the future.
- */
-
 static atomic_uint_fast32_t last_tcpdnsquota_log = ATOMIC_VAR_INIT(0);
 
 static bool
@@ -109,6 +102,10 @@ tcpdns_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
        UV_RUNTIME_CHECK(uv_timer_init, r);
        uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
 
+       r = uv_timer_init(&worker->loop, &sock->write_timer);
+       UV_RUNTIME_CHECK(uv_timer_init, r);
+       uv_handle_set_data((uv_handle_t *)&sock->write_timer, sock);
+
        if (isc__nm_closing(sock)) {
                result = ISC_R_CANCELED;
                goto error;
@@ -502,6 +499,10 @@ isc__nm_async_tcpdnslisten(isc__networker_t *worker, isc__netievent_t *ev0) {
        UV_RUNTIME_CHECK(uv_timer_init, r);
        uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
 
+       r = uv_timer_init(&worker->loop, &sock->write_timer);
+       UV_RUNTIME_CHECK(uv_timer_init, r);
+       uv_handle_set_data((uv_handle_t *)&sock->write_timer, sock);
+
        LOCK(&sock->parent->lock);
 
        r = uv_tcp_open(&sock->uv_handle.tcp, sock->fd);
@@ -956,6 +957,10 @@ accept_connection(isc_nmsocket_t *ssock, isc_quota_t *quota) {
        UV_RUNTIME_CHECK(uv_timer_init, r);
        uv_handle_set_data((uv_handle_t *)&csock->read_timer, csock);
 
+       r = uv_timer_init(&worker->loop, &csock->write_timer);
+       UV_RUNTIME_CHECK(uv_timer_init, r);
+       uv_handle_set_data((uv_handle_t *)&csock->write_timer, csock);
+
        r = uv_accept(&ssock->uv_handle.stream, &csock->uv_handle.stream);
        if (r != 0) {
                result = isc__nm_uverr2result(r);
@@ -1070,6 +1075,13 @@ isc__nm_tcpdns_send(isc_nmhandle_t *handle, isc_region_t *region,
        uvreq->cb.send = cb;
        uvreq->cbarg = cbarg;
 
+       if (sock->write_timeout == 0) {
+               sock->write_timeout =
+                       (atomic_load(&sock->keepalive)
+                                ? atomic_load(&sock->mgr->keepalive)
+                                : atomic_load(&sock->mgr->idle));
+       }
+
        ievent = isc__nm_get_netievent_tcpdnssend(sock->mgr, sock, uvreq);
        isc__nm_maybe_enqueue_ievent(&sock->mgr->workers[sock->tid],
                                     (isc__netievent_t *)ievent);
@@ -1087,6 +1099,11 @@ tcpdns_send_cb(uv_write_t *req, int status) {
 
        sock = uvreq->sock;
 
+       if (--sock->writes == 0) {
+               int r = uv_timer_stop(&sock->write_timer);
+               UV_RUNTIME_CHECK(uv_timer_stop, r);
+       }
+
        if (status < 0) {
                isc__nm_incstats(sock->mgr, sock->statsindex[STATID_SENDFAIL]);
                isc__nm_failed_send_cb(sock, uvreq,
@@ -1151,6 +1168,11 @@ isc__nm_async_tcpdnssend(isc__networker_t *worker, isc__netievent_t *ev0) {
                goto fail;
        }
 
+       r = uv_timer_start(&sock->write_timer, isc__nmsocket_writetimeout_cb,
+                          sock->write_timeout, 0);
+       UV_RUNTIME_CHECK(uv_timer_start, r);
+       RUNTIME_CHECK(sock->writes++ >= 0);
+
        r = uv_write(&uvreq->uv_req.write, &sock->uv_handle.stream, bufs, nbufs,
                     tcpdns_send_cb);
        if (r < 0) {
@@ -1222,7 +1244,7 @@ tcpdns_close_cb(uv_handle_t *handle) {
 }
 
 static void
-timer_close_cb(uv_handle_t *timer) {
+read_timer_close_cb(uv_handle_t *timer) {
        isc_nmsocket_t *sock = uv_handle_get_data(timer);
        uv_handle_set_data(timer, NULL);
 
@@ -1237,6 +1259,17 @@ timer_close_cb(uv_handle_t *timer) {
        }
 }
 
+static void
+write_timer_close_cb(uv_handle_t *timer) {
+       isc_nmsocket_t *sock = uv_handle_get_data(timer);
+       uv_handle_set_data(timer, NULL);
+
+       REQUIRE(VALID_NMSOCK(sock));
+
+       uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
+       uv_close((uv_handle_t *)&sock->read_timer, read_timer_close_cb);
+}
+
 static void
 stop_tcpdns_child(isc_nmsocket_t *sock) {
        REQUIRE(sock->type == isc_nm_tcpdnssocket);
@@ -1289,6 +1322,7 @@ stop_tcpdns_parent(isc_nmsocket_t *sock) {
 
 static void
 tcpdns_close_direct(isc_nmsocket_t *sock) {
+       int r;
        REQUIRE(VALID_NMSOCK(sock));
        REQUIRE(sock->tid == isc_nm_tid());
        REQUIRE(atomic_load(&sock->closing));
@@ -1304,8 +1338,10 @@ tcpdns_close_direct(isc_nmsocket_t *sock) {
        isc__nmsocket_timer_stop(sock);
        isc__nm_stop_reading(sock);
 
-       uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
-       uv_close((uv_handle_t *)&sock->read_timer, timer_close_cb);
+       r = uv_timer_stop(&sock->write_timer);
+       UV_RUNTIME_CHECK(uv_timer_stop, r);
+       uv_handle_set_data((uv_handle_t *)&sock->write_timer, sock);
+       uv_close((uv_handle_t *)&sock->write_timer, write_timer_close_cb);
 }
 
 void

diff --git a/lib/isc/netmgr/udp.c b/lib/isc/netmgr/udp.c
index 8119ac79c61ee4114405f072351d542f3ab8864c..77de1cdafdf55e03dae705e6c0fc77b6e4a517c1 100644 (file)
--- a/lib/isc/netmgr/udp.c
+++ b/lib/isc/netmgr/udp.c
@@ -48,7 +48,10 @@ static void
 udp_close_cb(uv_handle_t *handle);
 
 static void
-timer_close_cb(uv_handle_t *handle);
+read_timer_close_cb(uv_handle_t *handle);
+
+static void
+write_timer_close_cb(uv_handle_t *handle);
 
 static void
 udp_close_direct(isc_nmsocket_t *sock);
@@ -232,6 +235,10 @@ isc__nm_async_udplisten(isc__networker_t *worker, isc__netievent_t *ev0) {
        UV_RUNTIME_CHECK(uv_timer_init, r);
        uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
 
+       r = uv_timer_init(&worker->loop, &sock->write_timer);
+       UV_RUNTIME_CHECK(uv_timer_init, r);
+       uv_handle_set_data((uv_handle_t *)&sock->write_timer, sock);
+
        LOCK(&sock->parent->lock);
 
        r = uv_udp_open(&sock->uv_handle.udp, sock->fd);
@@ -654,6 +661,10 @@ udp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
        UV_RUNTIME_CHECK(uv_timer_init, r);
        uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
 
+       r = uv_timer_init(&worker->loop, &sock->write_timer);
+       UV_RUNTIME_CHECK(uv_timer_init, r);
+       uv_handle_set_data((uv_handle_t *)&sock->write_timer, sock);
+
        r = uv_udp_open(&sock->uv_handle.udp, sock->fd);
        if (r != 0) {
                isc__nm_incstats(sock->mgr, sock->statsindex[STATID_OPENFAIL]);
@@ -998,7 +1009,7 @@ udp_close_cb(uv_handle_t *handle) {
 }
 
 static void
-timer_close_cb(uv_handle_t *handle) {
+read_timer_close_cb(uv_handle_t *handle) {
        isc_nmsocket_t *sock = uv_handle_get_data(handle);
        uv_handle_set_data(handle, NULL);
 
@@ -1009,6 +1020,17 @@ timer_close_cb(uv_handle_t *handle) {
        }
 }
 
+static void
+write_timer_close_cb(uv_handle_t *timer) {
+       isc_nmsocket_t *sock = uv_handle_get_data(timer);
+       uv_handle_set_data(timer, NULL);
+
+       REQUIRE(VALID_NMSOCK(sock));
+
+       uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
+       uv_close((uv_handle_t *)&sock->read_timer, read_timer_close_cb);
+}
+
 static void
 stop_udp_child(isc_nmsocket_t *sock) {
        REQUIRE(sock->type == isc_nm_udpsocket);
@@ -1061,10 +1083,14 @@ stop_udp_parent(isc_nmsocket_t *sock) {
 
 static void
 udp_close_direct(isc_nmsocket_t *sock) {
+       int r;
        REQUIRE(VALID_NMSOCK(sock));
        REQUIRE(sock->tid == isc_nm_tid());
 
-       uv_close((uv_handle_t *)&sock->read_timer, timer_close_cb);
+       r = uv_timer_stop(&sock->write_timer);
+       UV_RUNTIME_CHECK(uv_timer_stop, r);
+       uv_handle_set_data((uv_handle_t *)&sock->write_timer, sock);
+       uv_close((uv_handle_t *)&sock->write_timer, write_timer_close_cb);
 }
 
 void