Make netmgr tcpdns send calls asynchronous.

isc__nm_tcpdns_send() was not asynchronous and accessed socket
internal fields in an unsafe manner, which could lead to a race
condition and subsequent crash. Fix it by moving the whole tcpdns
processing to a proper netmgr thread.

diff --git a/lib/isc/netmgr/netmgr-int.h b/lib/isc/netmgr/netmgr-int.h
index b15e6fac2908f10f05da9ebb51fc2d62537f4565..5c598ae295cc93c670bb14b2042681e5824187b7 100644 (file)
--- a/lib/isc/netmgr/netmgr-int.h
+++ b/lib/isc/netmgr/netmgr-int.h
@@ -136,7 +136,9 @@ typedef enum isc__netievent_type {
        netievent_tcpaccept,
        netievent_tcpstop,
        netievent_tcpclose,
+
        netievent_tcpdnsclose,
+       netievent_tcpdnssend,
 
        netievent_closecb,
        netievent_shutdown,
@@ -227,6 +229,7 @@ typedef struct isc__netievent__socket_req {
 typedef isc__netievent__socket_req_t isc__netievent_tcpconnect_t;
 typedef isc__netievent__socket_req_t isc__netievent_tcplisten_t;
 typedef isc__netievent__socket_req_t isc__netievent_tcpsend_t;
+typedef isc__netievent__socket_req_t isc__netievent_tcpdnssend_t;
 
 typedef struct isc__netievent__socket_streaminfo_quota {
        isc__netievent_type type;
@@ -746,6 +749,9 @@ isc__nm_tcpdns_stoplistening(isc_nmsocket_t *sock);
 void
 isc__nm_async_tcpdnsclose(isc__networker_t *worker, isc__netievent_t *ev0);
 
+void
+isc__nm_async_tcpdnssend(isc__networker_t *worker, isc__netievent_t *ev0);
+
 #define isc__nm_uverr2result(x) \
        isc___nm_uverr2result(x, true, __FILE__, __LINE__)
 isc_result_t

diff --git a/lib/isc/netmgr/netmgr.c b/lib/isc/netmgr/netmgr.c
index 228e72dd3fc5684b2d9a2ef5268be24d9cc45945..ac981a8613727be3ea0d651425ccf84477a3c99b 100644 (file)
--- a/lib/isc/netmgr/netmgr.c
+++ b/lib/isc/netmgr/netmgr.c
@@ -621,6 +621,9 @@ process_queue(isc__networker_t *worker, isc_queue_t *queue) {
                case netievent_tcpsend:
                        isc__nm_async_tcpsend(worker, ievent);
                        break;
+               case netievent_tcpdnssend:
+                       isc__nm_async_tcpdnssend(worker, ievent);
+                       break;
                case netievent_tcpstop:
                        isc__nm_async_tcpstop(worker, ievent);
                        break;

diff --git a/lib/isc/netmgr/tcpdns.c b/lib/isc/netmgr/tcpdns.c
index 7e3d1728dd2f30e54eb4548b3d55e0a6f9f0bacc..cfd9b3f873e125e67ddbeda856d99600be1293d5 100644 (file)
--- a/lib/isc/netmgr/tcpdns.c
+++ b/lib/isc/netmgr/tcpdns.c
@@ -365,15 +365,6 @@ isc_nm_tcpdns_keepalive(isc_nmhandle_t *handle) {
        atomic_store(&handle->sock->outer->keepalive, true);
 }
 
-typedef struct tcpsend {
-       isc_mem_t *mctx;
-       isc_nmhandle_t *handle;
-       isc_region_t region;
-       isc_nmhandle_t *orighandle;
-       isc_nm_cb_t cb;
-       void *cbarg;
-} tcpsend_t;
-
 static void
 resume_processing(void *arg) {
        isc_nmsocket_t *sock = (isc_nmsocket_t *)arg;
@@ -445,15 +436,40 @@ resume_processing(void *arg) {
 
 static void
 tcpdnssend_cb(isc_nmhandle_t *handle, isc_result_t result, void *cbarg) {
-       tcpsend_t *ts = (tcpsend_t *)cbarg;
+       isc__nm_uvreq_t *req = (isc__nm_uvreq_t *)cbarg;
 
        UNUSED(handle);
 
-       ts->cb(ts->orighandle, result, ts->cbarg);
-       isc_mem_put(ts->mctx, ts->region.base, ts->region.length);
+       req->cb.send(req->handle, result, req->cbarg);
+       isc_mem_put(req->sock->mgr->mctx, req->uvbuf.base, req->uvbuf.len);
+       isc__nm_uvreq_put(&req, req->handle->sock);
+}
+
+void
+isc__nm_async_tcpdnssend(isc__networker_t *worker, isc__netievent_t *ev0) {
+       isc_result_t result;
+       isc__netievent_tcpdnssend_t *ievent =
+               (isc__netievent_tcpdnssend_t *)ev0;
+       isc__nm_uvreq_t *req = ievent->req;
+       isc_nmsocket_t *sock = ievent->sock;
+
+       REQUIRE(worker->id == sock->tid);
 
-       isc_nmhandle_unref(ts->orighandle);
-       isc_mem_putanddetach(&ts->mctx, ts, sizeof(*ts));
+       result = ISC_R_NOTCONNECTED;
+       if (atomic_load(&sock->active)) {
+               isc_region_t r;
+
+               r.base = (unsigned char *)req->uvbuf.base;
+               r.length = req->uvbuf.len;
+               result = isc__nm_tcp_send(sock->outer->tcphandle, &r,
+                                         tcpdnssend_cb, req);
+       }
+
+       if (result != ISC_R_SUCCESS) {
+               req->cb.send(req->handle, result, req->cbarg);
+               isc_mem_put(sock->mgr->mctx, req->uvbuf.base, req->uvbuf.len);
+               isc__nm_uvreq_put(&req, sock);
+       }
 }
 
 /*
@@ -462,7 +478,7 @@ tcpdnssend_cb(isc_nmhandle_t *handle, isc_result_t result, void *cbarg) {
 isc_result_t
 isc__nm_tcpdns_send(isc_nmhandle_t *handle, isc_region_t *region,
                    isc_nm_cb_t cb, void *cbarg) {
-       tcpsend_t *t = NULL;
+       isc__nm_uvreq_t *uvreq = NULL;
 
        REQUIRE(VALID_NMHANDLE(handle));
 
@@ -471,36 +487,46 @@ isc__nm_tcpdns_send(isc_nmhandle_t *handle, isc_region_t *region,
        REQUIRE(VALID_NMSOCK(sock));
        REQUIRE(sock->type == isc_nm_tcpdnssocket);
 
-       if (sock->outer == NULL) {
-               /* The socket is closed */
-               return (ISC_R_NOTCONNECTED);
-       }
+       uvreq = isc__nm_uvreq_get(sock->mgr, sock);
+       uvreq->handle = handle;
+       isc_nmhandle_ref(uvreq->handle);
+       uvreq->cb.send = cb;
+       uvreq->cbarg = cbarg;
 
-       t = isc_mem_get(sock->mgr->mctx, sizeof(*t));
-       *t = (tcpsend_t){
-               .cb = cb,
-               .cbarg = cbarg,
-               .handle = handle->sock->outer->tcphandle,
-       };
+       uvreq->uvbuf.base = isc_mem_get(sock->mgr->mctx, region->length + 2);
+       uvreq->uvbuf.len = region->length + 2;
+       *(uint16_t *)uvreq->uvbuf.base = htons(region->length);
+       memmove(uvreq->uvbuf.base + 2, region->base, region->length);
 
-       isc_mem_attach(sock->mgr->mctx, &t->mctx);
-       t->orighandle = handle;
-       isc_nmhandle_ref(t->orighandle);
+       if (sock->tid == isc_nm_tid()) {
+               isc_region_t r;
+
+               r.base = (unsigned char *)uvreq->uvbuf.base;
+               r.length = uvreq->uvbuf.len;
 
-       t->region = (isc_region_t){ .base = isc_mem_get(t->mctx,
-                                                       region->length + 2),
-                                   .length = region->length + 2 };
+               return (isc__nm_tcp_send(sock->outer->tcphandle, &r,
+                                        tcpdnssend_cb, uvreq));
+       } else {
+               isc__netievent_tcpdnssend_t *ievent = NULL;
 
-       *(uint16_t *)t->region.base = htons(region->length);
-       memmove(t->region.base + 2, region->base, region->length);
+               ievent = isc__nm_get_ievent(sock->mgr, netievent_tcpdnssend);
+               ievent->req = uvreq;
+               ievent->sock = sock;
 
-       return (isc_nm_send(t->handle, &t->region, tcpdnssend_cb, t));
+               isc__nm_enqueue_ievent(&sock->mgr->workers[sock->tid],
+                                      (isc__netievent_t *)ievent);
+
+               return (ISC_R_SUCCESS);
+       }
+
+       return (ISC_R_UNEXPECTED);
 }
 
 static void
 tcpdns_close_direct(isc_nmsocket_t *sock) {
        REQUIRE(sock->tid == isc_nm_tid());
        /* We don't need atomics here, it's all in single network thread */
+
        if (sock->timer_initialized) {
                /*
                 * We need to fire the timer callback to clean it up,