Commit 591b79b5 authored by Evan Hunt's avatar Evan Hunt
Browse files

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 tcpdns processing
to a proper netmgr thread.
parent 1cf65cd8
Pipeline #45308 failed with stages
in 23 minutes and 51 seconds
......@@ -127,7 +127,9 @@ typedef enum isc__netievent_type {
netievent_tcpaccept,
netievent_tcpstop,
netievent_tcpclose,
netievent_tcpdnsclose,
netievent_tcpdnssend,
netievent_closecb,
netievent_shutdown,
......@@ -219,6 +221,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;
......@@ -772,6 +775,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
......
......@@ -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;
......
......@@ -389,15 +389,6 @@ isc_nm_tcpdns_keepalive(isc_nmhandle_t *handle) {
atomic_store(&handle->sock->outerhandle->sock->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;
......@@ -474,15 +465,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);
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);
}
ts->cb(ts->orighandle, result, ts->cbarg);
isc_mem_put(ts->mctx, ts->region.base, ts->region.length);
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;
isc_nmhandle_unref(ts->orighandle);
isc_mem_putanddetach(&ts->mctx, ts, sizeof(*ts));
REQUIRE(worker->id == sock->tid);
isc_nmhandle_unref(handle);
result = ISC_R_NOTCONNECTED;
if (atomic_load(&sock->active) && sock->outerhandle != NULL) {
isc_region_t r;
r.base = (unsigned char *)req->uvbuf.base;
r.length = req->uvbuf.len;
result = isc__nm_tcp_send(sock->outerhandle, &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);
}
}
/*
......@@ -491,7 +507,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));
......@@ -500,31 +516,39 @@ isc__nm_tcpdns_send(isc_nmhandle_t *handle, isc_region_t *region,
REQUIRE(VALID_NMSOCK(sock));
REQUIRE(sock->type == isc_nm_tcpdnssocket);
if (sock->outerhandle == 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;
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);
if (sock->tid == isc_nm_tid()) {
isc_region_t r;
t = isc_mem_get(sock->mgr->mctx, sizeof(*t));
*t = (tcpsend_t){
.cb = cb,
.cbarg = cbarg,
.handle = handle->sock->outerhandle,
};
r.base = (unsigned char *)uvreq->uvbuf.base;
r.length = uvreq->uvbuf.len;
isc_mem_attach(sock->mgr->mctx, &t->mctx);
t->orighandle = handle;
isc_nmhandle_ref(t->orighandle);
isc_nmhandle_ref(t->handle);
return (isc__nm_tcp_send(sock->outerhandle, &r, tcpdnssend_cb,
uvreq));
} else {
isc__netievent_tcpdnssend_t *ievent = NULL;
t->region = (isc_region_t){ .base = isc_mem_get(t->mctx,
region->length + 2),
.length = region->length + 2 };
ievent = isc__nm_get_ievent(sock->mgr, netievent_tcpdnssend);
ievent->req = uvreq;
ievent->sock = sock;
*(uint16_t *)t->region.base = htons(region->length);
memmove(t->region.base + 2, region->base, region->length);
isc__nm_enqueue_ievent(&sock->mgr->workers[sock->tid],
(isc__netievent_t *)ievent);
return (ISC_R_SUCCESS);
}
return (isc_nm_send(t->handle, &t->region, tcpdnssend_cb, t));
return (ISC_R_UNEXPECTED);
}
static void
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment