Commit ff2bc789 authored by Ondřej Surý's avatar Ondřej Surý
Browse files

Merge branch '2255-dig-crashed-in-tcp_connected-on-openbsd' into 'main'

Turn all the callback to be always asynchronous

Closes #2255

See merge request !4386
parents fece7a48 a49d8856
Pipeline #57128 canceled with stages
in 16 minutes and 2 seconds
......@@ -172,6 +172,11 @@ typedef enum isc__netievent_type {
netievent_stop,
netievent_pause,
netievent_connectcb,
netievent_acceptcb,
netievent_readcb,
netievent_sendcb,
netievent_prio = 0xff, /* event type values higher than this
* will be treated as high-priority
* events, which can be processed
......@@ -187,6 +192,7 @@ typedef union {
isc_nm_recv_cb_t recv;
isc_nm_cb_t send;
isc_nm_cb_t connect;
isc_nm_accept_cb_t accept;
} isc__nm_cb_t;
/*
......@@ -260,6 +266,18 @@ 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_req_result {
isc__netievent_type type;
isc_nmsocket_t *sock;
isc__nm_uvreq_t *req;
isc_result_t result;
} isc__netievent__socket_req_result_t;
typedef isc__netievent__socket_req_result_t isc__netievent_connectcb_t;
typedef isc__netievent__socket_req_result_t isc__netievent_acceptcb_t;
typedef isc__netievent__socket_req_result_t isc__netievent_readcb_t;
typedef isc__netievent__socket_req_result_t isc__netievent_sendcb_t;
typedef struct isc__netievent__socket_streaminfo_quota {
isc__netievent_type type;
isc_nmsocket_t *sock;
......@@ -746,6 +764,48 @@ isc__nmsocket_clearcb(isc_nmsocket_t *sock);
* Clear the recv and accept callbacks in 'sock'.
*/
void
isc__nm_connectcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
isc_result_t eresult);
void
isc__nm_async_connectcb(isc__networker_t *worker, isc__netievent_t *ev0);
/*%<
* Issue a connect callback on the socket, used to call the callback
* on failed conditions when the event can't be scheduled on the uv loop.
*/
void
isc__nm_acceptcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
isc_result_t eresult);
void
isc__nm_async_acceptcb(isc__networker_t *worker, isc__netievent_t *ev0);
/*%<
* Issue a accept callback on the socket, used to call the callback
* on failed conditions when the event can't be scheduled on the uv loop.
*/
void
isc__nm_readcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
isc_result_t eresult);
void
isc__nm_async_readcb(isc__networker_t *worker, isc__netievent_t *ev0);
/*%<
* Issue a read callback on the socket, used to call the callback
* on failed conditions when the event can't be scheduled on the uv loop.
*
*/
void
isc__nm_sendcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
isc_result_t eresult);
void
isc__nm_async_sendcb(isc__networker_t *worker, isc__netievent_t *ev0);
/*%<
* Issue a write callback on the socket, used to call the callback
* on failed conditions when the event can't be scheduled on the uv loop.
*/
void
isc__nm_async_closecb(isc__networker_t *worker, isc__netievent_t *ev0);
/*%<
......
......@@ -700,9 +700,22 @@ process_queue(isc__networker_t *worker, isc_queue_t *queue) {
isc__nm_async_tls_do_bio(worker, ievent);
break;
case netievent_connectcb:
isc__nm_async_connectcb(worker, ievent);
break;
case netievent_acceptcb:
isc__nm_async_acceptcb(worker, ievent);
break;
case netievent_readcb:
isc__nm_async_readcb(worker, ievent);
break;
case netievent_sendcb:
isc__nm_async_sendcb(worker, ievent);
break;
case netievent_closecb:
isc__nm_async_closecb(worker, ievent);
break;
case netievent_detach:
isc__nm_async_detach(worker, ievent);
break;
......@@ -1644,18 +1657,195 @@ isc_nm_stoplistening(isc_nmsocket_t *sock) {
}
}
void
isc__nm_connectcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
isc_result_t eresult) {
isc__netievent_connectcb_t *ievent =
isc__nm_get_ievent(sock->mgr, netievent_connectcb);
REQUIRE(VALID_NMSOCK(sock));
REQUIRE(VALID_UVREQ(uvreq));
REQUIRE(VALID_NMHANDLE(uvreq->handle));
ievent->sock = sock;
ievent->req = uvreq;
ievent->result = eresult;
if (eresult == ISC_R_SUCCESS) {
isc__nm_async_connectcb(&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);
}
}
void
isc__nm_async_connectcb(isc__networker_t *worker, isc__netievent_t *ev0) {
isc__netievent_connectcb_t *ievent = (isc__netievent_connectcb_t *)ev0;
isc_nmsocket_t *sock = ievent->sock;
isc__nm_uvreq_t *uvreq = ievent->req;
isc_result_t eresult = ievent->result;
UNUSED(worker);
REQUIRE(VALID_NMSOCK(sock));
REQUIRE(VALID_UVREQ(uvreq));
REQUIRE(VALID_NMHANDLE(uvreq->handle));
REQUIRE(ievent->sock->tid == isc_nm_tid());
REQUIRE(uvreq->cb.connect != NULL);
uvreq->cb.connect(uvreq->handle, eresult, uvreq->cbarg);
isc__nm_uvreq_put(&uvreq, sock);
}
void
isc__nm_acceptcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
isc_result_t eresult) {
isc__netievent_acceptcb_t *ievent =
isc__nm_get_ievent(sock->mgr, netievent_acceptcb);
REQUIRE(VALID_NMSOCK(sock));
REQUIRE(VALID_UVREQ(uvreq));
REQUIRE(VALID_NMHANDLE(uvreq->handle));
ievent->sock = sock;
ievent->req = uvreq;
ievent->result = eresult;
if (eresult == ISC_R_SUCCESS) {
isc__nm_async_acceptcb(&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);
}
}
void
isc__nm_async_acceptcb(isc__networker_t *worker, isc__netievent_t *ev0) {
isc__netievent_acceptcb_t *ievent = (isc__netievent_acceptcb_t *)ev0;
isc_nmsocket_t *sock = ievent->sock;
isc__nm_uvreq_t *uvreq = ievent->req;
isc_result_t eresult = ievent->result;
UNUSED(worker);
REQUIRE(VALID_NMSOCK(sock));
REQUIRE(VALID_UVREQ(uvreq));
REQUIRE(VALID_NMHANDLE(uvreq->handle));
REQUIRE(sock->tid == isc_nm_tid());
REQUIRE(uvreq->cb.accept != NULL);
uvreq->cb.accept(uvreq->handle, eresult, uvreq->cbarg);
isc__nm_uvreq_put(&uvreq, sock);
}
void
isc__nm_readcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
isc_result_t eresult) {
isc__netievent_readcb_t *ievent = isc__nm_get_ievent(sock->mgr,
netievent_readcb);
REQUIRE(VALID_NMSOCK(sock));
REQUIRE(VALID_UVREQ(uvreq));
REQUIRE(VALID_NMHANDLE(uvreq->handle));
ievent->sock = sock;
ievent->req = uvreq;
ievent->result = eresult;
if (eresult == ISC_R_SUCCESS) {
isc__nm_async_readcb(&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);
}
}
void
isc__nm_async_readcb(isc__networker_t *worker, isc__netievent_t *ev0) {
isc__netievent_readcb_t *ievent = (isc__netievent_readcb_t *)ev0;
isc_nmsocket_t *sock = ievent->sock;
isc__nm_uvreq_t *uvreq = ievent->req;
isc_result_t eresult = ievent->result;
isc_region_t region = { .base = (unsigned char *)uvreq->uvbuf.base,
.length = uvreq->uvbuf.len };
UNUSED(worker);
REQUIRE(VALID_NMSOCK(sock));
REQUIRE(VALID_UVREQ(uvreq));
REQUIRE(VALID_NMHANDLE(uvreq->handle));
REQUIRE(sock->tid == isc_nm_tid());
uvreq->cb.recv(uvreq->handle, eresult, &region, uvreq->cbarg);
isc__nm_uvreq_put(&uvreq, sock);
}
void
isc__nm_sendcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
isc_result_t eresult) {
isc__netievent_sendcb_t *ievent = isc__nm_get_ievent(sock->mgr,
netievent_sendcb);
REQUIRE(VALID_NMSOCK(sock));
REQUIRE(VALID_UVREQ(uvreq));
REQUIRE(VALID_NMHANDLE(uvreq->handle));
ievent->sock = sock;
ievent->req = uvreq;
ievent->result = eresult;
if (eresult == ISC_R_SUCCESS) {
isc__nm_async_sendcb(&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);
}
}
void
isc__nm_async_sendcb(isc__networker_t *worker, isc__netievent_t *ev0) {
isc__netievent_sendcb_t *ievent = (isc__netievent_sendcb_t *)ev0;
isc_nmsocket_t *sock = ievent->sock;
isc__nm_uvreq_t *uvreq = ievent->req;
isc_result_t eresult = ievent->result;
UNUSED(worker);
REQUIRE(VALID_NMSOCK(sock));
REQUIRE(VALID_UVREQ(uvreq));
REQUIRE(VALID_NMHANDLE(uvreq->handle));
REQUIRE(sock->tid == isc_nm_tid());
uvreq->cb.send(uvreq->handle, eresult, uvreq->cbarg);
isc__nm_uvreq_put(&uvreq, sock);
}
void
isc__nm_async_closecb(isc__networker_t *worker, isc__netievent_t *ev0) {
isc__netievent_closecb_t *ievent = (isc__netievent_closecb_t *)ev0;
isc_nmsocket_t *sock = ievent->sock;
REQUIRE(VALID_NMSOCK(ievent->sock));
REQUIRE(ievent->sock->tid == isc_nm_tid());
REQUIRE(ievent->sock->closehandle_cb != NULL);
REQUIRE(sock->tid == isc_nm_tid());
REQUIRE(sock->closehandle_cb != NULL);
UNUSED(worker);
ievent->sock->closehandle_cb(ievent->sock);
isc__nmsocket_detach(&ievent->sock);
ievent->sock->closehandle_cb(sock);
isc__nmsocket_detach(&sock);
}
void
......
......@@ -49,7 +49,7 @@ can_log_tcp_quota(void) {
return (false);
}
static int
static isc_result_t
tcp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req);
static void
......@@ -80,6 +80,17 @@ quota_accept_cb(isc_quota_t *quota, void *sock0);
static void
failed_accept_cb(isc_nmsocket_t *sock, isc_result_t eresult);
static void
failed_send_cb(isc_nmsocket_t *sock, isc__nm_uvreq_t *req,
isc_result_t eresult);
static bool
inactive(isc_nmsocket_t *sock) {
return (!isc__nmsocket_active(sock) ||
atomic_load(&sock->mgr->closing) ||
(sock->server != NULL && !isc__nmsocket_active(sock->server)));
}
static void
failed_accept_cb(isc_nmsocket_t *sock, isc_result_t eresult) {
/*
......@@ -126,20 +137,17 @@ failed_connect_cb(isc_nmsocket_t *sock, isc__nm_uvreq_t *req,
}
if (!atomic_load(&sock->connecting)) {
isc__nm_uvreq_put(&req, sock);
return;
}
atomic_store(&sock->connecting, false);
isc__nmsocket_clearcb(sock);
if (req->cb.connect != NULL) {
req->cb.connect(NULL, eresult, req->cbarg);
isc__nm_connectcb(sock, req, eresult);
} else {
isc__nm_uvreq_put(&req, sock);
}
req->cb.connect = NULL;
req->cbarg = NULL;
isc__nm_uvreq_put(&req, sock);
isc__nmsocket_detach(&sock);
}
static void
......@@ -147,16 +155,22 @@ connecttimeout_cb(uv_timer_t *handle) {
isc__nm_uvreq_t *req = uv_handle_get_data((uv_handle_t *)handle);
isc_nmsocket_t *sock = req->sock;
REQUIRE(VALID_UVREQ(req));
REQUIRE(VALID_NMHANDLE(req->handle));
REQUIRE(sock->tid == isc_nm_tid());
failed_connect_cb(sock, req, ISC_R_TIMEDOUT);
isc__nmsocket_detach(&sock);
}
static int
static isc_result_t
tcp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
isc__networker_t *worker = NULL;
int r;
REQUIRE(VALID_NMSOCK(sock));
REQUIRE(VALID_UVREQ(req));
REQUIRE(isc__nm_in_netthread());
REQUIRE(sock->tid == isc_nm_tid());
......@@ -169,11 +183,8 @@ tcp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
isc__nm_incstats(sock->mgr, sock->statsindex[STATID_OPENFAIL]);
atomic_store(&sock->closing, true);
atomic_store(&sock->closed, true);
atomic_store(&sock->result, isc__nm_uverr2result(r));
atomic_store(&sock->connect_error, true);
failed_connect_cb(sock, req, isc__nm_uverr2result(r));
atomic_store(&sock->active, false);
return (r);
return (isc__nm_uverr2result(r));
}
if (req->local.length != 0) {
......@@ -181,12 +192,9 @@ tcp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
if (r != 0) {
isc__nm_incstats(sock->mgr,
sock->statsindex[STATID_BINDFAIL]);
atomic_store(&sock->result, isc__nm_uverr2result(r));
atomic_store(&sock->connect_error, true);
failed_connect_cb(sock, req, isc__nm_uverr2result(r));
atomic_store(&sock->active, false);
isc__nm_tcp_close(sock);
return (r);
return (isc__nm_uverr2result(r));
}
}
......@@ -203,12 +211,9 @@ tcp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
if (r != 0) {
isc__nm_incstats(sock->mgr,
sock->statsindex[STATID_CONNECTFAIL]);
atomic_store(&sock->result, isc__nm_uverr2result(r));
atomic_store(&sock->connect_error, true);
failed_connect_cb(sock, req, isc__nm_uverr2result(r));
atomic_store(&sock->active, false);
isc__nm_tcp_close(sock);
return (r);
return (isc__nm_uverr2result(r));
}
isc__nm_incstats(sock->mgr, sock->statsindex[STATID_CONNECT]);
......@@ -216,7 +221,7 @@ tcp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
0);
sock->timer_running = true;
return (0);
return (ISC_R_SUCCESS);
}
void
......@@ -225,23 +230,29 @@ isc__nm_async_tcpconnect(isc__networker_t *worker, isc__netievent_t *ev0) {
(isc__netievent_tcpconnect_t *)ev0;
isc_nmsocket_t *sock = ievent->sock;
isc__nm_uvreq_t *req = ievent->req;
int r;
isc_result_t result = ISC_R_SUCCESS;
UNUSED(worker);
REQUIRE(VALID_NMSOCK(sock));
REQUIRE(sock->type == isc_nm_tcpsocket);
REQUIRE(sock->iface != NULL);
REQUIRE(sock->parent == NULL);
REQUIRE(sock->tid == isc_nm_tid());
UNUSED(worker);
r = tcp_connect_direct(sock, req);
if (r != 0) {
LOCK(&sock->lock);
SIGNAL(&sock->cond);
UNLOCK(&sock->lock);
return;
req->handle = isc__nmhandle_get(sock, &req->peer, &sock->iface->addr);
result = tcp_connect_direct(sock, req);
atomic_store(&sock->result, result);
if (result == ISC_R_SUCCESS) {
atomic_store(&sock->connected, true);
/* uvreq will be freed in tcp_connect_cb */
/* socket will be detached in tcp_connect_cb */
} else {
atomic_store(&sock->connect_error, true);
isc__nm_uvreq_put(&req, sock);
isc__nmsocket_detach(&ievent->sock);
}
atomic_store(&sock->connected, true);
LOCK(&sock->lock);
SIGNAL(&sock->cond);
UNLOCK(&sock->lock);
......@@ -250,28 +261,32 @@ isc__nm_async_tcpconnect(isc__networker_t *worker, isc__netievent_t *ev0) {
static void
tcp_connect_cb(uv_connect_t *uvreq, int status) {
isc_result_t result;
isc__nm_uvreq_t *req = uv_handle_get_data((uv_handle_t *)uvreq);
isc__nm_uvreq_t *req = NULL;
isc_nmsocket_t *sock = uv_handle_get_data((uv_handle_t *)uvreq->handle);
struct sockaddr_storage ss;
isc_nmhandle_t *handle = NULL;
int r;
REQUIRE(VALID_NMSOCK(sock));
REQUIRE(sock->tid == isc_nm_tid());
if (sock->timer_running) {
uv_timer_stop(&sock->timer);
sock->timer_running = false;
}
/* We timed out */
if (!atomic_load(&sock->connecting)) {
return;
}
req = uv_handle_get_data((uv_handle_t *)uvreq);
REQUIRE(VALID_UVREQ(req));
REQUIRE(VALID_NMHANDLE(req->handle));
if (sock->timer_running) {
uv_timer_stop(&sock->timer);
sock->timer_running = false;
}
if (status != 0) {
failed_connect_cb(sock, req, isc__nm_uverr2result(status));
isc__nmsocket_detach(&sock);
return;
}
......@@ -280,6 +295,7 @@ tcp_connect_cb(uv_connect_t *uvreq, int status) {
&(int){ sizeof(ss) });
if (r != 0) {
failed_connect_cb(sock, req, isc__nm_uverr2result(r));
isc__nmsocket_detach(&sock);
return;
}
......@@ -288,21 +304,12 @@ tcp_connect_cb(uv_connect_t *uvreq, int status) {
result = isc_sockaddr_fromsockaddr(&sock->peer, (struct sockaddr *)&ss);
RUNTIME_CHECK(result == ISC_R_SUCCESS);
handle = isc__nmhandle_get(sock, NULL, NULL);
req->cb.connect(handle, ISC_R_SUCCESS, req->cbarg);
isc__nm_uvreq_put(&req, sock);
isc__nm_connectcb(sock, req, ISC_R_SUCCESS);
/*
* The sock is now attached to the handle.
*/
isc__nmsocket_detach(&sock);
/*
* The connect callback should have attached to the handle.
* If it didn't, the socket will be closed now.
*/
isc_nmhandle_detach(&handle);
}
isc_result_t
......@@ -310,7 +317,7 @@ isc_nm_tcpconnect(isc_nm_t *mgr, isc_nmiface_t *local, isc_nmiface_t *peer,
isc_nm_cb_t cb, void *cbarg, unsigned int timeout,
size_t extrahandlesize) {
isc_result_t result = ISC_R_SUCCESS;
isc_nmsocket_t *nsock = NULL, *tmp = NULL;
isc_nmsocket_t *sock = NULL, *tmp = NULL;
isc__netievent_tcpconnect_t *ievent = NULL;
isc__nm_uvreq_t *req = NULL;
......@@ -318,50 +325,50 @@ isc_nm_tcpconnect(isc_nm_t *mgr, isc_nmiface_t *local, isc_nmiface_t *peer,
REQUIRE(local != NULL);
REQUIRE(peer != NULL);
nsock = isc_mem_get(mgr->mctx, sizeof(*nsock));
isc__nmsocket_init(nsock, mgr, isc_nm_tcpsocket, local);
sock = isc_mem_get(mgr->mctx, sizeof(*sock));
isc__nmsocket_init(sock, mgr, isc_nm_tcpsocket, local);
nsock->extrahandlesize = extrahandlesize;
nsock->connect_timeout = timeout;
sock->extrahandlesize = extrahandlesize;
sock->connect_timeout = timeout;
atomic_init(&nsock->result, ISC_R_SUCCESS);
atomic_init(&nsock->client, true);
atomic_init(&sock->result, ISC_R_SUCCESS);
atomic_init(&sock->client, true);
req = isc__nm_uvreq_get(mgr, nsock);
req = isc__nm_uvreq_get(mgr, sock);
req->cb.connect = cb;
req->cbarg = cbarg;
req->peer = peer->addr;
req->local = local->addr;
ievent = isc__nm_get_ievent(mgr, netievent_tcpconnect);
ievent->sock = nsock;
ievent->sock = sock;
ievent->req = req;
/*
* Async callbacks can dereference the socket in the meantime,
* we need to hold an additional reference to it.
*/
isc__nmsocket_attach(nsock, &tmp);
isc__nmsocket_attach(sock, &tmp);
if (isc__nm_in_netthread()) {
nsock->tid = isc_nm_tid();
isc__nm_async_tcpconnect(&mgr->workers[nsock->tid],
sock->tid = isc_nm_tid();
isc__nm_async_tcpconnect(&mgr->workers[sock->tid],
(isc__netievent_t *)ievent);
isc__nm_put_ievent(mgr, ievent);
} else {
nsock->tid = isc_random_uniform(mgr->nworkers);
isc__nm_enqueue_ievent(&mgr->workers[nsock->tid],
sock->tid = isc_random_uniform(mgr->nworkers);
isc__nm_enqueue_ievent(&mgr->workers[sock->tid],
(isc__netievent_t *)ievent);
LOCK(&nsock->lock);
while (!atomic_load(&nsock->connected) &&
!atomic_load(&nsock->connect_error)) {
WAIT(&nsock->cond, &nsock->lock);
LOCK(&sock->lock);
while (!atomic_load(&sock->connected) &&
!atomic_load(&sock->connect_error)) {
WAIT(&sock->cond, &sock->lock);
}
UNLOCK(&nsock->lock);