Convert netmgr connect, read and send callbacks to isc_job

The netmgr connect, read and send callbacks can now only be executed on
the same loop, convert it from asynchronous netievent queue event to
more direct isc_job.

diff --git a/lib/isc/netmgr/netmgr-int.h b/lib/isc/netmgr/netmgr-int.h
index 5db189b3ba6e5054bdb89cb3a8f9ab61e1da67fb..3432604a2d731d1045250d1e0e57b3575376ac9a 100644 (file)
--- a/lib/isc/netmgr/netmgr-int.h
+++ b/lib/isc/netmgr/netmgr-int.h
@@ -266,10 +266,6 @@ typedef enum isc__netievent_type {
        netievent_streamdnsread,
        netievent_streamdnscancel,
 
-       netievent_connectcb,
-       netievent_readcb,
-       netievent_sendcb,
-
        netievent_settlsctx,
        netievent_sockstop, /* for multilayer sockets */
 
@@ -311,6 +307,7 @@ struct isc__nm_uvreq {
        void *cbarg;           /* callback argument */
        isc_nm_timer_t *timer; /* TCP write timer */
        int connect_tries;     /* connect retries */
+       isc_result_t result;
 
        union {
                uv_handle_t handle;
@@ -1177,18 +1174,9 @@ void
 isc__nm_connectcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
                  isc_result_t eresult, bool async);
 
-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
- */
-
 void
 isc__nm_readcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
               isc_result_t eresult, bool async);
-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.
@@ -1738,8 +1726,6 @@ NETIEVENT_SOCKET_HTTP_EPS_TYPE(httpendpoints);
 
 NETIEVENT_SOCKET_REQ_TYPE(tlssend);
 
-NETIEVENT_SOCKET_REQ_RESULT_TYPE(connectcb);
-NETIEVENT_SOCKET_REQ_RESULT_TYPE(readcb);
 NETIEVENT_SOCKET_REQ_RESULT_TYPE(sendcb);
 
 NETIEVENT_SOCKET_HANDLE_TYPE(detach);
@@ -1771,8 +1757,6 @@ NETIEVENT_SOCKET_HTTP_EPS_DECL(httpendpoints);
 
 NETIEVENT_SOCKET_REQ_DECL(tlssend);
 
-NETIEVENT_SOCKET_REQ_RESULT_DECL(connectcb);
-NETIEVENT_SOCKET_REQ_RESULT_DECL(readcb);
 NETIEVENT_SOCKET_REQ_RESULT_DECL(sendcb);
 
 NETIEVENT_SOCKET_HANDLE_DECL(udpcancel);

diff --git a/lib/isc/netmgr/netmgr.c b/lib/isc/netmgr/netmgr.c
index cf9601e5be418e5d3a41bb2cae107c13bd62193a..ecbc7b3c720658c64ce578bf4c9ddd9320975b48 100644 (file)
--- a/lib/isc/netmgr/netmgr.c
+++ b/lib/isc/netmgr/netmgr.c
@@ -466,10 +466,6 @@ process_netievent(void *arg) {
                NETIEVENT_CASE(settlsctx);
                NETIEVENT_CASE(sockstop);
 
-               NETIEVENT_CASE(connectcb);
-               NETIEVENT_CASE(readcb);
-               NETIEVENT_CASE(sendcb);
-
                NETIEVENT_CASE(detach);
        default:
                UNREACHABLE();
@@ -511,9 +507,6 @@ NETIEVENT_SOCKET_HTTP_EPS_DEF(httpendpoints);
 #endif /* HAVE_LIBNGHTTP2 */
 
 NETIEVENT_SOCKET_REQ_DEF(tlssend);
-NETIEVENT_SOCKET_REQ_RESULT_DEF(connectcb);
-NETIEVENT_SOCKET_REQ_RESULT_DEF(readcb);
-NETIEVENT_SOCKET_REQ_RESULT_DEF(sendcb);
 
 NETIEVENT_SOCKET_DEF(detach);
 
@@ -2005,84 +1998,44 @@ isc__nm_async_sockstop(isc__networker_t *worker, isc__netievent_t *ev0) {
        isc_barrier_wait(&listener->stop_barrier);
 }
 
-void
-isc__nm_connectcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
-                 isc_result_t eresult, bool async) {
-       REQUIRE(VALID_NMSOCK(sock));
-       REQUIRE(VALID_UVREQ(uvreq));
-       REQUIRE(VALID_NMHANDLE(uvreq->handle));
-
-       if (!async) {
-               isc__netievent_connectcb_t ievent = { .sock = sock,
-                                                     .req = uvreq,
-                                                     .result = eresult };
-               isc__nm_async_connectcb(NULL, (isc__netievent_t *)&ievent);
-       } else {
-               isc__netievent_connectcb_t *ievent =
-                       isc__nm_get_netievent_connectcb(sock->worker, sock,
-                                                       uvreq, eresult);
-               isc__nm_enqueue_ievent(sock->worker,
-                                      (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);
+static void
+isc__nm_connectcb_job(void *arg) {
+       isc__nm_uvreq_t *uvreq = arg;
+       isc_result_t eresult = uvreq->result;
 
-       REQUIRE(VALID_NMSOCK(sock));
        REQUIRE(VALID_UVREQ(uvreq));
        REQUIRE(VALID_NMHANDLE(uvreq->handle));
-       REQUIRE(ievent->sock->tid == isc_tid());
        REQUIRE(uvreq->cb.connect != NULL);
 
        uvreq->cb.connect(uvreq->handle, eresult, uvreq->cbarg);
-
-       isc__nm_uvreq_put(&uvreq, sock);
+       isc__nm_uvreq_put(&uvreq, uvreq->handle->sock);
 }
 
 void
-isc__nm_readcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
-              isc_result_t eresult, bool async) {
+isc__nm_connectcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
+                 isc_result_t eresult, bool async) {
        REQUIRE(VALID_NMSOCK(sock));
        REQUIRE(VALID_UVREQ(uvreq));
        REQUIRE(VALID_NMHANDLE(uvreq->handle));
 
-       if (!async) {
-               isc__netievent_readcb_t ievent = { .type = netievent_readcb,
-                                                  .sock = sock,
-                                                  .req = uvreq,
-                                                  .result = eresult };
+       uvreq->result = eresult;
 
-               isc__nm_async_readcb(NULL, (isc__netievent_t *)&ievent);
+       if (!async) {
+               isc__nm_connectcb_job(uvreq);
                return;
        }
 
-       isc__netievent_readcb_t *ievent = isc__nm_get_netievent_readcb(
-               sock->worker, sock, uvreq, eresult);
-       isc__nm_enqueue_ievent(sock->worker, (isc__netievent_t *)ievent);
+       isc_job_run(sock->worker->netmgr->loopmgr, isc__nm_connectcb_job,
+                   uvreq);
 }
 
-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;
+static void
+isc__nm_readcb_job(void *arg) {
+       isc__nm_uvreq_t *uvreq = arg;
+       isc_result_t eresult = uvreq->result;
+       isc_nmsocket_t *sock = uvreq->handle->sock;
        isc_region_t region;
 
-       UNUSED(worker);
-
-       REQUIRE(VALID_NMSOCK(sock));
-       REQUIRE(VALID_UVREQ(uvreq));
-       REQUIRE(VALID_NMHANDLE(uvreq->handle));
-       REQUIRE(sock->tid == isc_tid());
-
        region.base = (unsigned char *)uvreq->uvbuf.base;
        region.length = uvreq->uvbuf.len;
 
@@ -2092,42 +2045,46 @@ isc__nm_async_readcb(isc__networker_t *worker, isc__netievent_t *ev0) {
 }
 
 void
-isc__nm_sendcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
+isc__nm_readcb(isc_nmsocket_t *sock, isc__nm_uvreq_t *uvreq,
               isc_result_t eresult, bool async) {
        REQUIRE(VALID_NMSOCK(sock));
        REQUIRE(VALID_UVREQ(uvreq));
        REQUIRE(VALID_NMHANDLE(uvreq->handle));
+       uvreq->result = eresult;
 
        if (!async) {
-               isc__netievent_sendcb_t ievent = { .sock = sock,
-                                                  .req = uvreq,
-                                                  .result = eresult };
-               isc__nm_async_sendcb(NULL, (isc__netievent_t *)&ievent);
+               isc__nm_readcb_job(uvreq);
                return;
        }
 
-       isc__netievent_sendcb_t *ievent = isc__nm_get_netievent_sendcb(
-               sock->worker, sock, uvreq, eresult);
-       isc__nm_enqueue_ievent(sock->worker, (isc__netievent_t *)ievent);
+       isc_job_run(sock->worker->netmgr->loopmgr, isc__nm_readcb_job, uvreq);
 }
 
-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;
+static void
+isc__nm_sendcb_job(void *arg) {
+       isc__nm_uvreq_t *uvreq = arg;
+       isc_result_t eresult = uvreq->result;
+       isc_nmsocket_t *sock = uvreq->handle->sock;
 
-       UNUSED(worker);
+       uvreq->cb.send(uvreq->handle, eresult, 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, bool async) {
        REQUIRE(VALID_NMSOCK(sock));
        REQUIRE(VALID_UVREQ(uvreq));
        REQUIRE(VALID_NMHANDLE(uvreq->handle));
-       REQUIRE(sock->tid == isc_tid());
+       uvreq->result = eresult;
 
-       uvreq->cb.send(uvreq->handle, eresult, uvreq->cbarg);
+       if (!async) {
+               isc__nm_sendcb_job(uvreq);
+               return;
+       }
 
-       isc__nm_uvreq_put(&uvreq, sock);
+       isc_job_run(sock->worker->netmgr->loopmgr, isc__nm_sendcb_job, uvreq);
 }
 
 void