mirror of
https://gitlab.isc.org/isc-projects/bind9
synced 2025-09-04 16:45:24 +00:00
Update netmgr, tasks, and applications to use isc_loopmgr
Previously: * applications were using isc_app as the base unit for running the application and signal handling. * networking was handled in the netmgr layer, which would start a number of threads, each with a uv_loop event loop. * task/event handling was done in the isc_task unit, which used netmgr event loops to run the isc_event calls. In this refactoring: * the network manager now uses isc_loop instead of maintaining its own worker threads and event loops. * the taskmgr that manages isc_task instances now also uses isc_loopmgr, and every isc_task runs on a specific isc_loop bound to the specific thread. * applications have been updated as necessary to use the new API. * new ISC_LOOP_TEST macros have been added to enable unit tests to run isc_loop event loops. unit tests have been updated to use this where needed.
This commit is contained in:
@@ -72,45 +72,37 @@ accept_connection(isc_nmsocket_t *ssock, isc_quota_t *quota);
|
||||
static void
|
||||
quota_accept_cb(isc_quota_t *quota, void *sock0);
|
||||
|
||||
static void
|
||||
stop_tcpdns_parent(isc_nmsocket_t *sock);
|
||||
static void
|
||||
stop_tcpdns_child(isc_nmsocket_t *sock);
|
||||
|
||||
static isc_result_t
|
||||
tcpdns_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
|
||||
isc__networker_t *worker = NULL;
|
||||
isc_result_t result = ISC_R_UNSET;
|
||||
int r;
|
||||
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(VALID_UVREQ(req));
|
||||
|
||||
REQUIRE(isc__nm_in_netthread());
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
|
||||
worker = &sock->mgr->workers[sock->tid];
|
||||
worker = sock->worker;
|
||||
|
||||
atomic_store(&sock->connecting, true);
|
||||
|
||||
r = uv_tcp_init(&worker->loop, &sock->uv_handle.tcp);
|
||||
r = uv_tcp_init(&worker->loop->loop, &sock->uv_handle.tcp);
|
||||
UV_RUNTIME_CHECK(uv_tcp_init, r);
|
||||
uv_handle_set_data(&sock->uv_handle.handle, sock);
|
||||
|
||||
r = uv_timer_init(&worker->loop, &sock->read_timer);
|
||||
r = uv_timer_init(&worker->loop->loop, &sock->read_timer);
|
||||
UV_RUNTIME_CHECK(uv_timer_init, r);
|
||||
uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
|
||||
|
||||
if (isc__nm_closing(sock)) {
|
||||
result = ISC_R_SHUTTINGDOWN;
|
||||
goto error;
|
||||
if (isc__nm_closing(worker)) {
|
||||
return (ISC_R_SHUTTINGDOWN);
|
||||
}
|
||||
|
||||
r = uv_tcp_open(&sock->uv_handle.tcp, sock->fd);
|
||||
if (r != 0) {
|
||||
isc__nm_closesocket(sock->fd);
|
||||
isc__nm_incstats(sock, STATID_OPENFAIL);
|
||||
goto done;
|
||||
return (isc_uverr2result(r));
|
||||
}
|
||||
isc__nm_incstats(sock, STATID_OPEN);
|
||||
|
||||
@@ -122,18 +114,19 @@ tcpdns_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
|
||||
*/
|
||||
if (r != 0 && r != UV_EINVAL) {
|
||||
isc__nm_incstats(sock, STATID_BINDFAIL);
|
||||
goto done;
|
||||
return (isc_uverr2result(r));
|
||||
}
|
||||
}
|
||||
|
||||
isc__nm_set_network_buffers(sock->mgr, &sock->uv_handle.handle);
|
||||
isc__nm_set_network_buffers(sock->worker->netmgr,
|
||||
&sock->uv_handle.handle);
|
||||
|
||||
uv_handle_set_data(&req->uv_req.handle, req);
|
||||
r = uv_tcp_connect(&req->uv_req.connect, &sock->uv_handle.tcp,
|
||||
&req->peer.type.sa, tcpdns_connect_cb);
|
||||
if (r != 0) {
|
||||
isc__nm_incstats(sock, STATID_CONNECTFAIL);
|
||||
goto done;
|
||||
return (isc_uverr2result(r));
|
||||
}
|
||||
|
||||
uv_handle_set_data((uv_handle_t *)&sock->read_timer,
|
||||
@@ -142,19 +135,7 @@ tcpdns_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
|
||||
|
||||
atomic_store(&sock->connected, true);
|
||||
|
||||
done:
|
||||
result = isc_uverr2result(r);
|
||||
error:
|
||||
LOCK(&sock->lock);
|
||||
sock->result = result;
|
||||
SIGNAL(&sock->cond);
|
||||
if (!atomic_load(&sock->active)) {
|
||||
WAIT(&sock->scond, &sock->lock);
|
||||
}
|
||||
INSIST(atomic_load(&sock->active));
|
||||
UNLOCK(&sock->lock);
|
||||
|
||||
return (result);
|
||||
return (ISC_R_SUCCESS);
|
||||
}
|
||||
|
||||
void
|
||||
@@ -170,7 +151,7 @@ isc__nm_async_tcpdnsconnect(isc__networker_t *worker, isc__netievent_t *ev0) {
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->type == isc_nm_tcpdnssocket);
|
||||
REQUIRE(sock->parent == NULL);
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
|
||||
result = tcpdns_connect_direct(sock, req);
|
||||
if (result != ISC_R_SUCCESS) {
|
||||
@@ -192,10 +173,13 @@ tcpdns_connect_cb(uv_connect_t *uvreq, int status) {
|
||||
isc__nm_uvreq_t *req = NULL;
|
||||
isc_nmsocket_t *sock = uv_handle_get_data((uv_handle_t *)uvreq->handle);
|
||||
struct sockaddr_storage ss;
|
||||
isc__networker_t *worker = NULL;
|
||||
int r;
|
||||
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
|
||||
worker = sock->worker;
|
||||
|
||||
req = uv_handle_get_data((uv_handle_t *)uvreq);
|
||||
|
||||
@@ -205,7 +189,7 @@ tcpdns_connect_cb(uv_connect_t *uvreq, int status) {
|
||||
if (atomic_load(&sock->timedout)) {
|
||||
result = ISC_R_TIMEDOUT;
|
||||
goto error;
|
||||
} else if (isc__nm_closing(sock)) {
|
||||
} else if (isc__nm_closing(worker)) {
|
||||
/* Network manager shutting down */
|
||||
result = ISC_R_SHUTTINGDOWN;
|
||||
goto error;
|
||||
@@ -271,6 +255,7 @@ isc_nm_tcpdnsconnect(isc_nm_t *mgr, isc_sockaddr_t *local, isc_sockaddr_t *peer,
|
||||
isc__netievent_tcpdnsconnect_t *ievent = NULL;
|
||||
isc__nm_uvreq_t *req = NULL;
|
||||
sa_family_t sa_family;
|
||||
isc__networker_t *worker = &mgr->workers[isc_tid()];
|
||||
|
||||
REQUIRE(VALID_NM(mgr));
|
||||
REQUIRE(local != NULL);
|
||||
@@ -278,14 +263,13 @@ isc_nm_tcpdnsconnect(isc_nm_t *mgr, isc_sockaddr_t *local, isc_sockaddr_t *peer,
|
||||
|
||||
sa_family = peer->type.sa.sa_family;
|
||||
|
||||
sock = isc_mem_get(mgr->mctx, sizeof(*sock));
|
||||
isc__nmsocket_init(sock, mgr, isc_nm_tcpdnssocket, local);
|
||||
sock = isc_mem_get(worker->mctx, sizeof(*sock));
|
||||
isc__nmsocket_init(sock, worker, isc_nm_tcpdnssocket, local);
|
||||
|
||||
sock->connect_timeout = timeout;
|
||||
sock->result = ISC_R_UNSET;
|
||||
atomic_init(&sock->client, true);
|
||||
|
||||
req = isc__nm_uvreq_get(mgr, sock);
|
||||
req = isc__nm_uvreq_get(worker, sock);
|
||||
req->cb.connect = cb;
|
||||
req->cbarg = cbarg;
|
||||
req->peer = *peer;
|
||||
@@ -294,9 +278,6 @@ isc_nm_tcpdnsconnect(isc_nm_t *mgr, isc_sockaddr_t *local, isc_sockaddr_t *peer,
|
||||
|
||||
result = isc__nm_socket(sa_family, SOCK_STREAM, 0, &sock->fd);
|
||||
if (result != ISC_R_SUCCESS) {
|
||||
if (isc__nm_in_netthread()) {
|
||||
sock->tid = isc_nm_tid();
|
||||
}
|
||||
isc__nmsocket_clearcb(sock);
|
||||
isc__nm_connectcb(sock, req, result, true);
|
||||
atomic_store(&sock->closed, true);
|
||||
@@ -311,28 +292,13 @@ isc_nm_tcpdnsconnect(isc_nm_t *mgr, isc_sockaddr_t *local, isc_sockaddr_t *peer,
|
||||
result = isc__nm_socket_connectiontimeout(sock->fd, 120 * 1000);
|
||||
RUNTIME_CHECK(result == ISC_R_SUCCESS);
|
||||
|
||||
ievent = isc__nm_get_netievent_tcpdnsconnect(mgr, sock, req);
|
||||
ievent = isc__nm_get_netievent_tcpdnsconnect(sock->worker, sock, req);
|
||||
|
||||
atomic_store(&sock->active, true);
|
||||
isc__nm_async_tcpdnsconnect(sock->worker, (isc__netievent_t *)ievent);
|
||||
isc__nm_put_netievent_tcpdnsconnect(sock->worker, ievent);
|
||||
|
||||
if (isc__nm_in_netthread()) {
|
||||
atomic_store(&sock->active, true);
|
||||
sock->tid = isc_nm_tid();
|
||||
isc__nm_async_tcpdnsconnect(&mgr->workers[sock->tid],
|
||||
(isc__netievent_t *)ievent);
|
||||
isc__nm_put_netievent_tcpdnsconnect(mgr, ievent);
|
||||
} else {
|
||||
atomic_init(&sock->active, false);
|
||||
sock->tid = isc_random_uniform(mgr->nworkers);
|
||||
isc__nm_enqueue_ievent(&mgr->workers[sock->tid],
|
||||
(isc__netievent_t *)ievent);
|
||||
}
|
||||
|
||||
LOCK(&sock->lock);
|
||||
while (sock->result == ISC_R_UNSET) {
|
||||
WAIT(&sock->cond, &sock->lock);
|
||||
}
|
||||
atomic_store(&sock->active, true);
|
||||
BROADCAST(&sock->scond);
|
||||
UNLOCK(&sock->lock);
|
||||
}
|
||||
|
||||
static uv_os_sock_t
|
||||
@@ -359,28 +325,20 @@ isc__nm_tcpdns_lb_socket(isc_nm_t *mgr, sa_family_t sa_family) {
|
||||
return (sock);
|
||||
}
|
||||
|
||||
static void
|
||||
enqueue_stoplistening(isc_nmsocket_t *sock) {
|
||||
isc__netievent_tcpdnsstop_t *ievent =
|
||||
isc__nm_get_netievent_tcpdnsstop(sock->mgr, sock);
|
||||
isc__nm_enqueue_ievent(&sock->mgr->workers[sock->tid],
|
||||
(isc__netievent_t *)ievent);
|
||||
}
|
||||
|
||||
static void
|
||||
start_tcpdns_child(isc_nm_t *mgr, isc_sockaddr_t *iface, isc_nmsocket_t *sock,
|
||||
uv_os_sock_t fd, int tid) {
|
||||
isc__netievent_tcpdnslisten_t *ievent = NULL;
|
||||
isc_nmsocket_t *csock = &sock->children[tid];
|
||||
isc__networker_t *worker = &mgr->workers[tid];
|
||||
|
||||
isc__nmsocket_init(csock, mgr, isc_nm_tcpdnssocket, iface);
|
||||
isc__nmsocket_init(csock, worker, isc_nm_tcpdnssocket, iface);
|
||||
csock->parent = sock;
|
||||
csock->accept_cb = sock->accept_cb;
|
||||
csock->accept_cbarg = sock->accept_cbarg;
|
||||
csock->recv_cb = sock->recv_cb;
|
||||
csock->recv_cbarg = sock->recv_cbarg;
|
||||
csock->backlog = sock->backlog;
|
||||
csock->tid = tid;
|
||||
/*
|
||||
* We don't attach to quota, just assign - to avoid
|
||||
* increasing quota unnecessarily.
|
||||
@@ -389,7 +347,7 @@ start_tcpdns_child(isc_nm_t *mgr, isc_sockaddr_t *iface, isc_nmsocket_t *sock,
|
||||
isc_quota_cb_init(&csock->quotacb, quota_accept_cb, csock);
|
||||
|
||||
if (mgr->load_balance_sockets) {
|
||||
UNUSED(fd);
|
||||
REQUIRE(fd == -1);
|
||||
csock->fd = isc__nm_tcpdns_lb_socket(mgr,
|
||||
iface->type.sa.sa_family);
|
||||
} else {
|
||||
@@ -397,34 +355,48 @@ start_tcpdns_child(isc_nm_t *mgr, isc_sockaddr_t *iface, isc_nmsocket_t *sock,
|
||||
}
|
||||
REQUIRE(csock->fd >= 0);
|
||||
|
||||
ievent = isc__nm_get_netievent_tcpdnslisten(mgr, csock);
|
||||
isc__nm_maybe_enqueue_ievent(&mgr->workers[tid],
|
||||
(isc__netievent_t *)ievent);
|
||||
ievent = isc__nm_get_netievent_tcpdnslisten(csock->worker, csock);
|
||||
|
||||
if (tid == 0) {
|
||||
isc__nm_process_ievent(csock->worker,
|
||||
(isc__netievent_t *)ievent);
|
||||
} else {
|
||||
isc__nm_enqueue_ievent(csock->worker,
|
||||
(isc__netievent_t *)ievent);
|
||||
}
|
||||
}
|
||||
|
||||
isc_result_t
|
||||
isc_nm_listentcpdns(isc_nm_t *mgr, uint32_t workers, isc_sockaddr_t *iface,
|
||||
isc_nm_recv_cb_t recv_cb, void *recv_cbarg,
|
||||
isc_nm_accept_cb_t accept_cb, void *accept_cbarg,
|
||||
int backlog, isc_quota_t *quota, isc_nmsocket_t **sockp) {
|
||||
isc_result_t result = ISC_R_SUCCESS;
|
||||
isc_nmsocket_t *sock = NULL;
|
||||
size_t children_size = 0;
|
||||
uv_os_sock_t fd = -1;
|
||||
isc_result_t result = ISC_R_UNSET;
|
||||
isc__networker_t *worker = &mgr->workers[0];
|
||||
|
||||
REQUIRE(VALID_NM(mgr));
|
||||
REQUIRE(isc_tid() == 0);
|
||||
|
||||
sock = isc_mem_get(mgr->mctx, sizeof(*sock));
|
||||
isc__nmsocket_init(sock, mgr, isc_nm_tcpdnslistener, iface);
|
||||
if (workers == 0) {
|
||||
workers = mgr->nloops;
|
||||
}
|
||||
REQUIRE(workers <= mgr->nloops);
|
||||
|
||||
sock = isc_mem_get(worker->mctx, sizeof(*sock));
|
||||
isc__nmsocket_init(sock, worker, isc_nm_tcpdnslistener, iface);
|
||||
|
||||
atomic_init(&sock->rchildren, 0);
|
||||
sock->nchildren = (workers == ISC_NM_LISTEN_ALL)
|
||||
? (uint32_t)mgr->nworkers
|
||||
: workers;
|
||||
sock->nchildren = (workers == ISC_NM_LISTEN_ALL) ? (uint32_t)mgr->nloops
|
||||
: workers;
|
||||
children_size = sock->nchildren * sizeof(sock->children[0]);
|
||||
sock->children = isc_mem_get(mgr->mctx, children_size);
|
||||
sock->children = isc_mem_get(worker->mctx, children_size);
|
||||
memset(sock->children, 0, children_size);
|
||||
|
||||
sock->result = ISC_R_UNSET;
|
||||
isc_barrier_init(&sock->barrier, sock->nchildren);
|
||||
|
||||
sock->accept_cb = accept_cb;
|
||||
sock->accept_cbarg = accept_cbarg;
|
||||
sock->recv_cb = recv_cb;
|
||||
@@ -432,50 +404,38 @@ isc_nm_listentcpdns(isc_nm_t *mgr, uint32_t workers, isc_sockaddr_t *iface,
|
||||
sock->backlog = backlog;
|
||||
sock->pquota = quota;
|
||||
|
||||
sock->tid = 0;
|
||||
sock->fd = -1;
|
||||
|
||||
if (!mgr->load_balance_sockets) {
|
||||
fd = isc__nm_tcpdns_lb_socket(mgr, iface->type.sa.sa_family);
|
||||
}
|
||||
|
||||
isc_barrier_init(&sock->startlistening, sock->nchildren);
|
||||
|
||||
for (size_t i = 0; i < sock->nchildren; i++) {
|
||||
if ((int)i == isc_nm_tid()) {
|
||||
continue;
|
||||
}
|
||||
for (size_t i = 1; i < sock->nchildren; i++) {
|
||||
start_tcpdns_child(mgr, iface, sock, fd, i);
|
||||
}
|
||||
|
||||
if (isc__nm_in_netthread()) {
|
||||
start_tcpdns_child(mgr, iface, sock, fd, isc_nm_tid());
|
||||
}
|
||||
start_tcpdns_child(mgr, iface, sock, fd, 0);
|
||||
|
||||
if (!mgr->load_balance_sockets) {
|
||||
isc__nm_closesocket(fd);
|
||||
}
|
||||
|
||||
LOCK(&sock->lock);
|
||||
while (atomic_load(&sock->rchildren) != sock->nchildren) {
|
||||
WAIT(&sock->cond, &sock->lock);
|
||||
}
|
||||
result = sock->result;
|
||||
atomic_store(&sock->active, true);
|
||||
UNLOCK(&sock->lock);
|
||||
|
||||
INSIST(result != ISC_R_UNSET);
|
||||
|
||||
if (result == ISC_R_SUCCESS) {
|
||||
REQUIRE(atomic_load(&sock->rchildren) == sock->nchildren);
|
||||
*sockp = sock;
|
||||
} else {
|
||||
atomic_store(&sock->active, true);
|
||||
|
||||
if (result != ISC_R_SUCCESS) {
|
||||
atomic_store(&sock->active, false);
|
||||
enqueue_stoplistening(sock);
|
||||
isc__nm_tcpdns_stoplistening(sock);
|
||||
isc_nmsocket_close(&sock);
|
||||
|
||||
return (result);
|
||||
}
|
||||
|
||||
return (result);
|
||||
REQUIRE(atomic_load(&sock->rchildren) == sock->nchildren);
|
||||
*sockp = sock;
|
||||
return (ISC_R_SUCCESS);
|
||||
}
|
||||
|
||||
void
|
||||
@@ -490,32 +450,30 @@ isc__nm_async_tcpdnslisten(isc__networker_t *worker, isc__netievent_t *ev0) {
|
||||
isc_nm_t *mgr = NULL;
|
||||
|
||||
REQUIRE(VALID_NMSOCK(ievent->sock));
|
||||
REQUIRE(ievent->sock->tid == isc_nm_tid());
|
||||
REQUIRE(ievent->sock->tid == isc_tid());
|
||||
REQUIRE(VALID_NMSOCK(ievent->sock->parent));
|
||||
|
||||
sock = ievent->sock;
|
||||
sa_family = sock->iface.type.sa.sa_family;
|
||||
mgr = sock->mgr;
|
||||
mgr = sock->worker->netmgr;
|
||||
|
||||
REQUIRE(sock->type == isc_nm_tcpdnssocket);
|
||||
REQUIRE(sock->parent != NULL);
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
|
||||
(void)isc__nm_socket_min_mtu(sock->fd, sa_family);
|
||||
(void)isc__nm_socket_tcp_maxseg(sock->fd, NM_MAXSEG);
|
||||
|
||||
r = uv_tcp_init(&worker->loop, &sock->uv_handle.tcp);
|
||||
r = uv_tcp_init(&worker->loop->loop, &sock->uv_handle.tcp);
|
||||
UV_RUNTIME_CHECK(uv_tcp_init, r);
|
||||
uv_handle_set_data(&sock->uv_handle.handle, sock);
|
||||
/* This keeps the socket alive after everything else is gone */
|
||||
isc__nmsocket_attach(sock, &(isc_nmsocket_t *){ NULL });
|
||||
|
||||
r = uv_timer_init(&worker->loop, &sock->read_timer);
|
||||
r = uv_timer_init(&worker->loop->loop, &sock->read_timer);
|
||||
UV_RUNTIME_CHECK(uv_timer_init, r);
|
||||
uv_handle_set_data((uv_handle_t *)&sock->read_timer, sock);
|
||||
|
||||
LOCK(&sock->parent->lock);
|
||||
|
||||
r = uv_tcp_open(&sock->uv_handle.tcp, sock->fd);
|
||||
if (r < 0) {
|
||||
isc__nm_closesocket(sock->fd);
|
||||
@@ -536,11 +494,13 @@ isc__nm_async_tcpdnslisten(isc__networker_t *worker, isc__netievent_t *ev0) {
|
||||
goto done;
|
||||
}
|
||||
} else {
|
||||
LOCK(&sock->parent->lock);
|
||||
if (sock->parent->fd == -1) {
|
||||
r = isc__nm_tcp_freebind(&sock->uv_handle.tcp,
|
||||
&sock->iface.type.sa, flags);
|
||||
if (r < 0) {
|
||||
isc__nm_incstats(sock, STATID_BINDFAIL);
|
||||
UNLOCK(&sock->parent->lock);
|
||||
goto done;
|
||||
}
|
||||
sock->parent->uv_handle.tcp.flags =
|
||||
@@ -551,9 +511,11 @@ isc__nm_async_tcpdnslisten(isc__networker_t *worker, isc__netievent_t *ev0) {
|
||||
sock->uv_handle.tcp.flags =
|
||||
sock->parent->uv_handle.tcp.flags;
|
||||
}
|
||||
UNLOCK(&sock->parent->lock);
|
||||
}
|
||||
|
||||
isc__nm_set_network_buffers(sock->mgr, &sock->uv_handle.handle);
|
||||
isc__nm_set_network_buffers(sock->worker->netmgr,
|
||||
&sock->uv_handle.handle);
|
||||
|
||||
/*
|
||||
* The callback will run in the same thread uv_listen() was called
|
||||
@@ -574,18 +536,22 @@ isc__nm_async_tcpdnslisten(isc__networker_t *worker, isc__netievent_t *ev0) {
|
||||
|
||||
done:
|
||||
result = isc_uverr2result(r);
|
||||
atomic_fetch_add(&sock->parent->rchildren, 1);
|
||||
|
||||
if (result != ISC_R_SUCCESS) {
|
||||
sock->pquota = NULL;
|
||||
}
|
||||
|
||||
atomic_fetch_add(&sock->parent->rchildren, 1);
|
||||
LOCK(&sock->parent->lock);
|
||||
if (sock->parent->result == ISC_R_UNSET) {
|
||||
sock->parent->result = result;
|
||||
} else {
|
||||
REQUIRE(sock->parent->result == result);
|
||||
}
|
||||
SIGNAL(&sock->parent->cond);
|
||||
UNLOCK(&sock->parent->lock);
|
||||
|
||||
isc_barrier_wait(&sock->parent->startlistening);
|
||||
REQUIRE(!worker->loop->paused);
|
||||
isc_barrier_wait(&sock->parent->barrier);
|
||||
}
|
||||
|
||||
static void
|
||||
@@ -600,7 +566,7 @@ tcpdns_connection_cb(uv_stream_t *server, int status) {
|
||||
}
|
||||
|
||||
REQUIRE(VALID_NMSOCK(ssock));
|
||||
REQUIRE(ssock->tid == isc_nm_tid());
|
||||
REQUIRE(ssock->tid == isc_tid());
|
||||
|
||||
if (isc__nmsocket_closing(ssock)) {
|
||||
result = ISC_R_CANCELED;
|
||||
@@ -621,21 +587,48 @@ done:
|
||||
isc__nm_accept_connection_log(result, can_log_tcpdns_quota());
|
||||
}
|
||||
|
||||
static void
|
||||
stop_tcpdns_child(isc_nmsocket_t *sock, uint32_t tid) {
|
||||
isc_nmsocket_t *csock = NULL;
|
||||
isc__netievent_tcpstop_t *ievent = NULL;
|
||||
|
||||
csock = &sock->children[tid];
|
||||
REQUIRE(VALID_NMSOCK(csock));
|
||||
|
||||
atomic_store(&csock->active, false);
|
||||
ievent = isc__nm_get_netievent_tcpdnsstop(csock->worker, csock);
|
||||
|
||||
if (tid == 0) {
|
||||
isc__nm_process_ievent(csock->worker,
|
||||
(isc__netievent_t *)ievent);
|
||||
} else {
|
||||
isc__nm_enqueue_ievent(csock->worker,
|
||||
(isc__netievent_t *)ievent);
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
stop_tcpdns_parent(isc_nmsocket_t *sock) {
|
||||
/* Stop the parent */
|
||||
atomic_store(&sock->closed, true);
|
||||
isc__nmsocket_prep_destroy(sock);
|
||||
}
|
||||
|
||||
void
|
||||
isc__nm_tcpdns_stoplistening(isc_nmsocket_t *sock) {
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->type == isc_nm_tcpdnslistener);
|
||||
|
||||
if (!atomic_compare_exchange_strong(&sock->closing, &(bool){ false },
|
||||
true)) {
|
||||
UNREACHABLE();
|
||||
RUNTIME_CHECK(atomic_compare_exchange_strong(&sock->closing,
|
||||
&(bool){ false }, true));
|
||||
|
||||
for (size_t i = 1; i < sock->nchildren; i++) {
|
||||
stop_tcpdns_child(sock, i);
|
||||
}
|
||||
|
||||
if (!isc__nm_in_netthread()) {
|
||||
enqueue_stoplistening(sock);
|
||||
} else {
|
||||
stop_tcpdns_parent(sock);
|
||||
}
|
||||
stop_tcpdns_child(sock, 0);
|
||||
|
||||
stop_tcpdns_parent(sock);
|
||||
}
|
||||
|
||||
void
|
||||
@@ -647,14 +640,18 @@ isc__nm_async_tcpdnsstop(isc__networker_t *worker, isc__netievent_t *ev0) {
|
||||
UNUSED(worker);
|
||||
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
REQUIRE(sock->parent != NULL);
|
||||
|
||||
if (sock->parent != NULL) {
|
||||
stop_tcpdns_child(sock);
|
||||
return;
|
||||
}
|
||||
RUNTIME_CHECK(atomic_compare_exchange_strong(&sock->closing,
|
||||
&(bool){ false }, true));
|
||||
|
||||
stop_tcpdns_parent(sock);
|
||||
tcpdns_close_direct(sock);
|
||||
|
||||
(void)atomic_fetch_sub(&sock->parent->rchildren, 1);
|
||||
|
||||
REQUIRE(!worker->loop->paused);
|
||||
isc_barrier_wait(&sock->parent->barrier);
|
||||
}
|
||||
|
||||
void
|
||||
@@ -695,6 +692,7 @@ isc__nm_tcpdns_read(isc_nmhandle_t *handle, isc_nm_recv_cb_t cb, void *cbarg) {
|
||||
|
||||
isc_nmsocket_t *sock = handle->sock;
|
||||
isc__netievent_tcpdnsread_t *ievent = NULL;
|
||||
isc_nm_t *netmgr = sock->worker->netmgr;
|
||||
|
||||
REQUIRE(sock->type == isc_nm_tcpdnssocket);
|
||||
REQUIRE(sock->statichandle == handle);
|
||||
@@ -703,13 +701,12 @@ isc__nm_tcpdns_read(isc_nmhandle_t *handle, isc_nm_recv_cb_t cb, void *cbarg) {
|
||||
sock->recv_cbarg = cbarg;
|
||||
sock->recv_read = true;
|
||||
if (sock->read_timeout == 0) {
|
||||
sock->read_timeout =
|
||||
(atomic_load(&sock->keepalive)
|
||||
? atomic_load(&sock->mgr->keepalive)
|
||||
: atomic_load(&sock->mgr->idle));
|
||||
sock->read_timeout = (atomic_load(&sock->keepalive)
|
||||
? atomic_load(&netmgr->keepalive)
|
||||
: atomic_load(&netmgr->idle));
|
||||
}
|
||||
|
||||
ievent = isc__nm_get_netievent_tcpdnsread(sock->mgr, sock);
|
||||
ievent = isc__nm_get_netievent_tcpdnsread(sock->worker, sock);
|
||||
|
||||
/*
|
||||
* This MUST be done asynchronously, no matter which thread we're
|
||||
@@ -717,8 +714,7 @@ isc__nm_tcpdns_read(isc_nmhandle_t *handle, isc_nm_recv_cb_t cb, void *cbarg) {
|
||||
* isc_nm_read() again; if we tried to do that synchronously
|
||||
* we'd clash in processbuffer() and grow the stack indefinitely.
|
||||
*/
|
||||
isc__nm_enqueue_ievent(&sock->mgr->workers[sock->tid],
|
||||
(isc__netievent_t *)ievent);
|
||||
isc__nm_enqueue_ievent(sock->worker, (isc__netievent_t *)ievent);
|
||||
|
||||
return;
|
||||
}
|
||||
@@ -733,7 +729,7 @@ isc__nm_async_tcpdnsread(isc__networker_t *worker, isc__netievent_t *ev0) {
|
||||
UNUSED(worker);
|
||||
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
|
||||
if (isc__nmsocket_closing(sock)) {
|
||||
result = ISC_R_CANCELED;
|
||||
@@ -763,7 +759,7 @@ isc__nm_tcpdns_processbuffer(isc_nmsocket_t *sock) {
|
||||
isc_nmhandle_t *handle = NULL;
|
||||
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
|
||||
if (isc__nmsocket_closing(sock)) {
|
||||
return (ISC_R_CANCELED);
|
||||
@@ -857,7 +853,7 @@ isc__nm_tcpdns_read_cb(uv_stream_t *stream, ssize_t nread,
|
||||
isc_result_t result;
|
||||
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
REQUIRE(atomic_load(&sock->reading));
|
||||
REQUIRE(buf != NULL);
|
||||
|
||||
@@ -894,7 +890,7 @@ isc__nm_tcpdns_read_cb(uv_stream_t *stream, ssize_t nread,
|
||||
sock->buf_len += len;
|
||||
|
||||
if (!atomic_load(&sock->client)) {
|
||||
sock->read_timeout = atomic_load(&sock->mgr->idle);
|
||||
sock->read_timeout = atomic_load(&sock->worker->netmgr->idle);
|
||||
}
|
||||
|
||||
result = isc__nm_process_sock_buffer(sock);
|
||||
@@ -925,9 +921,8 @@ quota_accept_cb(isc_quota_t *quota, void *sock0) {
|
||||
*/
|
||||
|
||||
isc__netievent_tcpdnsaccept_t *ievent =
|
||||
isc__nm_get_netievent_tcpdnsaccept(sock->mgr, sock, quota);
|
||||
isc__nm_maybe_enqueue_ievent(&sock->mgr->workers[sock->tid],
|
||||
(isc__netievent_t *)ievent);
|
||||
isc__nm_get_netievent_tcpdnsaccept(sock->worker, sock, quota);
|
||||
isc__nm_maybe_enqueue_ievent(sock->worker, (isc__netievent_t *)ievent);
|
||||
}
|
||||
|
||||
/*
|
||||
@@ -942,7 +937,7 @@ isc__nm_async_tcpdnsaccept(isc__networker_t *worker, isc__netievent_t *ev0) {
|
||||
UNUSED(worker);
|
||||
|
||||
REQUIRE(VALID_NMSOCK(ievent->sock));
|
||||
REQUIRE(ievent->sock->tid == isc_nm_tid());
|
||||
REQUIRE(ievent->sock->tid == isc_tid());
|
||||
|
||||
result = accept_connection(ievent->sock, ievent->quota);
|
||||
isc__nm_accept_connection_log(result, can_log_tcpdns_quota());
|
||||
@@ -960,7 +955,7 @@ accept_connection(isc_nmsocket_t *ssock, isc_quota_t *quota) {
|
||||
isc_nmhandle_t *handle = NULL;
|
||||
|
||||
REQUIRE(VALID_NMSOCK(ssock));
|
||||
REQUIRE(ssock->tid == isc_nm_tid());
|
||||
REQUIRE(ssock->tid == isc_tid());
|
||||
|
||||
if (isc__nmsocket_closing(ssock)) {
|
||||
if (quota != NULL) {
|
||||
@@ -971,23 +966,22 @@ accept_connection(isc_nmsocket_t *ssock, isc_quota_t *quota) {
|
||||
|
||||
REQUIRE(ssock->accept_cb != NULL);
|
||||
|
||||
csock = isc_mem_get(ssock->mgr->mctx, sizeof(isc_nmsocket_t));
|
||||
isc__nmsocket_init(csock, ssock->mgr, isc_nm_tcpdnssocket,
|
||||
csock = isc_mem_get(ssock->worker->mctx, sizeof(isc_nmsocket_t));
|
||||
isc__nmsocket_init(csock, ssock->worker, isc_nm_tcpdnssocket,
|
||||
&ssock->iface);
|
||||
csock->tid = ssock->tid;
|
||||
isc__nmsocket_attach(ssock, &csock->server);
|
||||
csock->recv_cb = ssock->recv_cb;
|
||||
csock->recv_cbarg = ssock->recv_cbarg;
|
||||
csock->quota = quota;
|
||||
atomic_init(&csock->accepting, true);
|
||||
|
||||
worker = &csock->mgr->workers[csock->tid];
|
||||
worker = csock->worker;
|
||||
|
||||
r = uv_tcp_init(&worker->loop, &csock->uv_handle.tcp);
|
||||
r = uv_tcp_init(&worker->loop->loop, &csock->uv_handle.tcp);
|
||||
UV_RUNTIME_CHECK(uv_tcp_init, r);
|
||||
uv_handle_set_data(&csock->uv_handle.handle, csock);
|
||||
|
||||
r = uv_timer_init(&worker->loop, &csock->read_timer);
|
||||
r = uv_timer_init(&worker->loop->loop, &csock->read_timer);
|
||||
UV_RUNTIME_CHECK(uv_timer_init, r);
|
||||
uv_handle_set_data((uv_handle_t *)&csock->read_timer, csock);
|
||||
|
||||
@@ -1041,7 +1035,7 @@ accept_connection(isc_nmsocket_t *ssock, isc_quota_t *quota) {
|
||||
|
||||
isc__nm_incstats(csock, STATID_ACCEPT);
|
||||
|
||||
csock->read_timeout = atomic_load(&csock->mgr->init);
|
||||
csock->read_timeout = atomic_load(&csock->worker->netmgr->init);
|
||||
|
||||
csock->closehandle_cb = isc__nm_resume_processing;
|
||||
|
||||
@@ -1062,9 +1056,10 @@ accept_connection(isc_nmsocket_t *ssock, isc_quota_t *quota) {
|
||||
* The initial timer has been set, update the read timeout for the next
|
||||
* reads.
|
||||
*/
|
||||
csock->read_timeout = (atomic_load(&csock->keepalive)
|
||||
? atomic_load(&csock->mgr->keepalive)
|
||||
: atomic_load(&csock->mgr->idle));
|
||||
csock->read_timeout =
|
||||
(atomic_load(&csock->keepalive)
|
||||
? atomic_load(&csock->worker->netmgr->keepalive)
|
||||
: atomic_load(&csock->worker->netmgr->idle));
|
||||
|
||||
isc_nmhandle_detach(&handle);
|
||||
|
||||
@@ -1102,7 +1097,7 @@ isc__nm_tcpdns_send(isc_nmhandle_t *handle, isc_region_t *region,
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->type == isc_nm_tcpdnssocket);
|
||||
|
||||
uvreq = isc__nm_uvreq_get(sock->mgr, sock);
|
||||
uvreq = isc__nm_uvreq_get(sock->worker, sock);
|
||||
*(uint16_t *)uvreq->tcplen = htons(region->length);
|
||||
uvreq->uvbuf.base = (char *)region->base;
|
||||
uvreq->uvbuf.len = region->length;
|
||||
@@ -1112,9 +1107,8 @@ isc__nm_tcpdns_send(isc_nmhandle_t *handle, isc_region_t *region,
|
||||
uvreq->cb.send = cb;
|
||||
uvreq->cbarg = cbarg;
|
||||
|
||||
ievent = isc__nm_get_netievent_tcpdnssend(sock->mgr, sock, uvreq);
|
||||
isc__nm_maybe_enqueue_ievent(&sock->mgr->workers[sock->tid],
|
||||
(isc__netievent_t *)ievent);
|
||||
ievent = isc__nm_get_netievent_tcpdnssend(sock->worker, sock, uvreq);
|
||||
isc__nm_maybe_enqueue_ievent(sock->worker, (isc__netievent_t *)ievent);
|
||||
|
||||
return;
|
||||
}
|
||||
@@ -1158,7 +1152,7 @@ isc__nm_async_tcpdnssend(isc__networker_t *worker, isc__netievent_t *ev0) {
|
||||
REQUIRE(VALID_UVREQ(ievent->req));
|
||||
REQUIRE(VALID_NMSOCK(ievent->sock));
|
||||
REQUIRE(ievent->sock->type == isc_nm_tcpdnssocket);
|
||||
REQUIRE(ievent->sock->tid == isc_nm_tid());
|
||||
REQUIRE(ievent->sock->tid == isc_tid());
|
||||
|
||||
sock = ievent->sock;
|
||||
uvreq = ievent->req;
|
||||
@@ -1166,8 +1160,8 @@ isc__nm_async_tcpdnssend(isc__networker_t *worker, isc__netievent_t *ev0) {
|
||||
if (sock->write_timeout == 0) {
|
||||
sock->write_timeout =
|
||||
(atomic_load(&sock->keepalive)
|
||||
? atomic_load(&sock->mgr->keepalive)
|
||||
: atomic_load(&sock->mgr->idle));
|
||||
? atomic_load(&sock->worker->netmgr->keepalive)
|
||||
: atomic_load(&sock->worker->netmgr->idle));
|
||||
}
|
||||
|
||||
uv_buf_t bufs[2] = { { .base = uvreq->tcplen, .len = 2 },
|
||||
@@ -1228,7 +1222,7 @@ tcpdns_stop_cb(uv_handle_t *handle) {
|
||||
isc_nmsocket_t *sock = uv_handle_get_data(handle);
|
||||
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
REQUIRE(atomic_load(&sock->closing));
|
||||
|
||||
uv_handle_set_data(handle, NULL);
|
||||
@@ -1248,7 +1242,7 @@ tcpdns_stop_cb(uv_handle_t *handle) {
|
||||
static void
|
||||
tcpdns_close_sock(isc_nmsocket_t *sock) {
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
REQUIRE(atomic_load(&sock->closing));
|
||||
|
||||
if (!atomic_compare_exchange_strong(&sock->closed, &(bool){ false },
|
||||
@@ -1292,60 +1286,10 @@ read_timer_close_cb(uv_handle_t *timer) {
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
stop_tcpdns_child(isc_nmsocket_t *sock) {
|
||||
REQUIRE(sock->type == isc_nm_tcpdnssocket);
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
|
||||
if (!atomic_compare_exchange_strong(&sock->closing, &(bool){ false },
|
||||
true)) {
|
||||
return;
|
||||
}
|
||||
|
||||
tcpdns_close_direct(sock);
|
||||
|
||||
atomic_fetch_sub(&sock->parent->rchildren, 1);
|
||||
|
||||
isc_barrier_wait(&sock->parent->stoplistening);
|
||||
}
|
||||
|
||||
static void
|
||||
stop_tcpdns_parent(isc_nmsocket_t *sock) {
|
||||
isc_nmsocket_t *csock = NULL;
|
||||
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->type == isc_nm_tcpdnslistener);
|
||||
|
||||
isc_barrier_init(&sock->stoplistening, sock->nchildren);
|
||||
|
||||
for (size_t i = 0; i < sock->nchildren; i++) {
|
||||
csock = &sock->children[i];
|
||||
REQUIRE(VALID_NMSOCK(csock));
|
||||
|
||||
if ((int)i == isc_nm_tid()) {
|
||||
/*
|
||||
* We need to schedule closing the other sockets first
|
||||
*/
|
||||
continue;
|
||||
}
|
||||
|
||||
atomic_store(&csock->active, false);
|
||||
enqueue_stoplistening(csock);
|
||||
}
|
||||
|
||||
csock = &sock->children[isc_nm_tid()];
|
||||
atomic_store(&csock->active, false);
|
||||
stop_tcpdns_child(csock);
|
||||
|
||||
atomic_store(&sock->closed, true);
|
||||
isc__nmsocket_prep_destroy(sock);
|
||||
}
|
||||
|
||||
static void
|
||||
tcpdns_close_direct(isc_nmsocket_t *sock) {
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
REQUIRE(atomic_load(&sock->closing));
|
||||
|
||||
if (sock->quota != NULL) {
|
||||
@@ -1374,16 +1318,16 @@ isc__nm_tcpdns_close(isc_nmsocket_t *sock) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (sock->tid == isc_nm_tid()) {
|
||||
if (sock->tid == isc_tid()) {
|
||||
tcpdns_close_direct(sock);
|
||||
} else {
|
||||
/*
|
||||
* We need to create an event and pass it using async channel
|
||||
*/
|
||||
isc__netievent_tcpdnsclose_t *ievent =
|
||||
isc__nm_get_netievent_tcpdnsclose(sock->mgr, sock);
|
||||
isc__nm_get_netievent_tcpdnsclose(sock->worker, sock);
|
||||
|
||||
isc__nm_enqueue_ievent(&sock->mgr->workers[sock->tid],
|
||||
isc__nm_enqueue_ievent(sock->worker,
|
||||
(isc__netievent_t *)ievent);
|
||||
}
|
||||
}
|
||||
@@ -1397,7 +1341,7 @@ isc__nm_async_tcpdnsclose(isc__networker_t *worker, isc__netievent_t *ev0) {
|
||||
UNUSED(worker);
|
||||
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
|
||||
tcpdns_close_direct(sock);
|
||||
}
|
||||
@@ -1408,8 +1352,7 @@ tcpdns_close_connect_cb(uv_handle_t *handle) {
|
||||
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
|
||||
REQUIRE(isc__nm_in_netthread());
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
|
||||
isc__nmsocket_prep_destroy(sock);
|
||||
isc__nmsocket_detach(&sock);
|
||||
@@ -1417,10 +1360,14 @@ tcpdns_close_connect_cb(uv_handle_t *handle) {
|
||||
|
||||
void
|
||||
isc__nm_tcpdns_shutdown(isc_nmsocket_t *sock) {
|
||||
isc__networker_t *worker = NULL;
|
||||
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
REQUIRE(sock->type == isc_nm_tcpdnssocket);
|
||||
|
||||
worker = sock->worker;
|
||||
|
||||
/*
|
||||
* If the socket is active, mark it inactive and
|
||||
* continue. If it isn't active, stop now.
|
||||
@@ -1441,7 +1388,7 @@ isc__nm_tcpdns_shutdown(isc_nmsocket_t *sock) {
|
||||
}
|
||||
|
||||
if (sock->statichandle != NULL) {
|
||||
if (isc__nm_closing(sock)) {
|
||||
if (isc__nm_closing(worker)) {
|
||||
isc__nm_failed_read_cb(sock, ISC_R_SHUTTINGDOWN, false);
|
||||
} else {
|
||||
isc__nm_failed_read_cb(sock, ISC_R_CANCELED, false);
|
||||
@@ -1469,9 +1416,8 @@ isc__nm_tcpdns_cancelread(isc_nmhandle_t *handle) {
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->type == isc_nm_tcpdnssocket);
|
||||
|
||||
ievent = isc__nm_get_netievent_tcpdnscancel(sock->mgr, sock, handle);
|
||||
isc__nm_enqueue_ievent(&sock->mgr->workers[sock->tid],
|
||||
(isc__netievent_t *)ievent);
|
||||
ievent = isc__nm_get_netievent_tcpdnscancel(sock->worker, sock, handle);
|
||||
isc__nm_enqueue_ievent(sock->worker, (isc__netievent_t *)ievent);
|
||||
}
|
||||
|
||||
void
|
||||
@@ -1483,7 +1429,7 @@ isc__nm_async_tcpdnscancel(isc__networker_t *worker, isc__netievent_t *ev0) {
|
||||
UNUSED(worker);
|
||||
|
||||
REQUIRE(VALID_NMSOCK(sock));
|
||||
REQUIRE(sock->tid == isc_nm_tid());
|
||||
REQUIRE(sock->tid == isc_tid());
|
||||
|
||||
isc__nm_failed_read_cb(sock, ISC_R_EOF, false);
|
||||
}
|
||||
|
Reference in New Issue
Block a user