Split fast and slow task queues

Change the taskmgr (and thus netmgr) in a way that it supports fast and
slow task queues.  The fast queue is used for incoming DNS traffic and
it will pass the processing to the slow queue for sending outgoing DNS
messages and processing resolver messages.

In the future, more tasks might get moved to the slow queues, so the
cached and authoritative DNS traffic can be handled without being slowed
down by operations that take longer time to process.

(cherry picked from commit 1b3b0cef224e7a9e8279c5cfe2f7e188e3777cc7)

diff --git a/lib/dns/resolver.c b/lib/dns/resolver.c
index a97aaa85e9299c4710ea1669739356920118f72a..09526246899fce4b4d57ad5fd1c839c47293e9f7 100644 (file)
--- a/lib/dns/resolver.c
+++ b/lib/dns/resolver.c
@@ -10634,8 +10634,8 @@ dns_resolver_create(dns_view_t *view, isc_taskmgr_t *taskmgr,
                 * Since we have a pool of tasks we bind them to task queues
                 * to spread the load evenly
                 */
-               result = isc_task_create_bound(taskmgr, 0,
-                                              &res->buckets[i].task, i);
+               result = isc_task_create_bound(
+                       taskmgr, 0, &res->buckets[i].task, ISC_NM_TASK_SLOW(i));
                if (result != ISC_R_SUCCESS) {
                        isc_mutex_destroy(&res->buckets[i].lock);
                        goto cleanup_buckets;

diff --git a/lib/isc/include/isc/netmgr.h b/lib/isc/include/isc/netmgr.h
index f1747be0583842d1a3d23dc68e1b16268208235e..efeb5f3d0006fad14185c5823ac14afadf83b594 100644 (file)
--- a/lib/isc/include/isc/netmgr.h
+++ b/lib/isc/include/isc/netmgr.h
@@ -479,6 +479,9 @@ isc_nm_tcpdnsconnect(isc_nm_t *mgr, isc_sockaddr_t *local, isc_sockaddr_t *peer,
  * 'cb'.
  */
 
+#define ISC_NM_TASK_SLOW_OFFSET -2
+#define ISC_NM_TASK_SLOW(i)    (ISC_NM_TASK_SLOW_OFFSET - 1 - i)
+
 void
 isc_nm_task_enqueue(isc_nm_t *mgr, isc_task_t *task, int threadid);
 /*%<

diff --git a/lib/isc/netmgr/netmgr-int.h b/lib/isc/netmgr/netmgr-int.h
index 05fde1a791b0a512c845f8a7de5a79204b1e3f44..c3a176260b93a3a0d7060a2f4c263e7771df847b 100644 (file)
--- a/lib/isc/netmgr/netmgr-int.h
+++ b/lib/isc/netmgr/netmgr-int.h
@@ -655,6 +655,7 @@ struct isc_nm {
        isc_refcount_t references;
        isc_mem_t *mctx;
        int nworkers;
+       int nlisteners;
        isc_mutex_t lock;
        isc_condition_t wkstatecond;
        isc_condition_t wkpausecond;

diff --git a/lib/isc/netmgr/netmgr.c b/lib/isc/netmgr/netmgr.c
index 6f42ec90ffb7e38235ca3bd9e2c82efc72fdcb64..7bff1cc9b3ec260b2f389424243e5130bf5f0e95 100644 (file)
--- a/lib/isc/netmgr/netmgr.c
+++ b/lib/isc/netmgr/netmgr.c
@@ -231,12 +231,12 @@ isc__nm_winsock_destroy(void) {
 #endif /* WIN32 */
 
 static void
-isc__nm_threadpool_initialize(uint32_t workers) {
+isc__nm_threadpool_initialize(uint32_t nworkers) {
        char buf[11];
        int r = uv_os_getenv("UV_THREADPOOL_SIZE", buf,
                             &(size_t){ sizeof(buf) });
        if (r == UV_ENOENT) {
-               snprintf(buf, sizeof(buf), "%" PRIu32, workers);
+               snprintf(buf, sizeof(buf), "%" PRIu32, nworkers);
                uv_os_setenv("UV_THREADPOOL_SIZE", buf);
        }
 }
@@ -254,11 +254,11 @@ isc__nm_threadpool_initialize(uint32_t workers) {
 #endif
 
 void
-isc__netmgr_create(isc_mem_t *mctx, uint32_t workers, isc_nm_t **netmgrp) {
+isc__netmgr_create(isc_mem_t *mctx, uint32_t nworkers, isc_nm_t **netmgrp) {
        isc_nm_t *mgr = NULL;
        char name[32];
 
-       REQUIRE(workers > 0);
+       REQUIRE(nworkers > 0);
 
 #ifdef MAXIMAL_UV_VERSION
        if (uv_version() > MAXIMAL_UV_VERSION) {
@@ -282,10 +282,13 @@ isc__netmgr_create(isc_mem_t *mctx, uint32_t workers, isc_nm_t **netmgrp) {
        isc__nm_winsock_initialize();
 #endif /* WIN32 */
 
-       isc__nm_threadpool_initialize(workers);
+       isc__nm_threadpool_initialize(nworkers);
 
        mgr = isc_mem_get(mctx, sizeof(*mgr));
-       *mgr = (isc_nm_t){ .nworkers = workers };
+       *mgr = (isc_nm_t){
+               .nworkers = nworkers * 2,
+               .nlisteners = nworkers,
+       };
 
        isc_mem_attach(mctx, &mgr->mctx);
        isc_mutex_init(&mgr->lock);
@@ -316,11 +319,12 @@ isc__netmgr_create(isc_mem_t *mctx, uint32_t workers, isc_nm_t **netmgrp) {
        atomic_init(&mgr->keepalive, 30000);
        atomic_init(&mgr->advertised, 30000);
 
-       isc_barrier_init(&mgr->pausing, workers);
-       isc_barrier_init(&mgr->resuming, workers);
+       isc_barrier_init(&mgr->pausing, mgr->nworkers);
+       isc_barrier_init(&mgr->resuming, mgr->nworkers);
 
-       mgr->workers = isc_mem_get(mctx, workers * sizeof(isc__networker_t));
-       for (size_t i = 0; i < workers; i++) {
+       mgr->workers = isc_mem_get(mctx,
+                                  mgr->nworkers * sizeof(isc__networker_t));
+       for (int i = 0; i < mgr->nworkers; i++) {
                isc__networker_t *worker = &mgr->workers[i];
                int r;
 
@@ -354,7 +358,7 @@ isc__netmgr_create(isc_mem_t *mctx, uint32_t workers, isc_nm_t **netmgrp) {
                mgr->workers_running++;
                isc_thread_create(nm_thread, &mgr->workers[i], &worker->thread);
 
-               snprintf(name, sizeof(name), "isc-net-%04zu", i);
+               snprintf(name, sizeof(name), "isc-net-%04d", i);
                isc_thread_setname(worker->thread, name);
        }
 
@@ -840,9 +844,15 @@ isc_nm_task_enqueue(isc_nm_t *nm, isc_task_t *task, int threadid) {
        isc__networker_t *worker = NULL;
 
        if (threadid == -1) {
-               tid = (int)isc_random_uniform(nm->nworkers);
+               tid = (int)isc_random_uniform(nm->nlisteners);
+       } else if (threadid == ISC_NM_TASK_SLOW_OFFSET) {
+               tid = nm->nlisteners +
+                     (int)isc_random_uniform(nm->nworkers - nm->nlisteners);
+       } else if (threadid < ISC_NM_TASK_SLOW_OFFSET) {
+               tid = nm->nlisteners + (ISC_NM_TASK_SLOW(threadid) %
+                                       (nm->nworkers - nm->nlisteners));
        } else {
-               tid = threadid % nm->nworkers;
+               tid = threadid % nm->nlisteners;
        }
 
        worker = &nm->workers[tid];

diff --git a/lib/isc/netmgr/tcp.c b/lib/isc/netmgr/tcp.c
index 821d6c4568ee34bc291283b4d4d018fdc38493a4..1666318453c9cc7b8649cc53f816c63fe51a458b 100644 (file)
--- a/lib/isc/netmgr/tcp.c
+++ b/lib/isc/netmgr/tcp.c
@@ -323,7 +323,7 @@ isc_nm_tcpconnect(isc_nm_t *mgr, isc_sockaddr_t *local, isc_sockaddr_t *peer,
                        isc__nm_connectcb(sock, req, result, false);
                } else {
                        isc__nmsocket_clearcb(sock);
-                       sock->tid = isc_random_uniform(mgr->nworkers);
+                       sock->tid = isc_random_uniform(mgr->nlisteners);
                        isc__nm_connectcb(sock, req, result, true);
                }
                atomic_store(&sock->closed, true);
@@ -341,7 +341,7 @@ isc_nm_tcpconnect(isc_nm_t *mgr, isc_sockaddr_t *local, isc_sockaddr_t *peer,
                isc__nm_put_netievent_tcpconnect(mgr, ievent);
        } else {
                atomic_init(&sock->active, false);
-               sock->tid = isc_random_uniform(mgr->nworkers);
+               sock->tid = isc_random_uniform(mgr->nlisteners);
                isc__nm_enqueue_ievent(&mgr->workers[sock->tid],
                                       (isc__netievent_t *)ievent);
        }
@@ -445,7 +445,7 @@ isc_nm_listentcp(isc_nm_t *mgr, isc_sockaddr_t *iface,
 #if defined(WIN32)
        sock->nchildren = 1;
 #else
-       sock->nchildren = mgr->nworkers;
+       sock->nchildren = mgr->nlisteners;
 #endif
        children_size = sock->nchildren * sizeof(sock->children[0]);
        sock->children = isc_mem_get(mgr->mctx, children_size);

diff --git a/lib/isc/netmgr/tcpdns.c b/lib/isc/netmgr/tcpdns.c
index bd593ebf4f722ee7a461b65f6d051d5e45b49d1a..037d74c607b94b29e5e3a4c9c08068b6339bc2f8 100644 (file)
--- a/lib/isc/netmgr/tcpdns.c
+++ b/lib/isc/netmgr/tcpdns.c
@@ -303,7 +303,7 @@ isc_nm_tcpdnsconnect(isc_nm_t *mgr, isc_sockaddr_t *local, isc_sockaddr_t *peer,
                isc__nm_put_netievent_tcpdnsconnect(mgr, ievent);
        } else {
                atomic_init(&sock->active, false);
-               sock->tid = isc_random_uniform(mgr->nworkers);
+               sock->tid = isc_random_uniform(mgr->nlisteners);
                isc__nm_enqueue_ievent(&mgr->workers[sock->tid],
                                       (isc__netievent_t *)ievent);
        }
@@ -410,7 +410,7 @@ isc_nm_listentcpdns(isc_nm_t *mgr, isc_sockaddr_t *iface,
 #if defined(WIN32)
        sock->nchildren = 1;
 #else
-       sock->nchildren = mgr->nworkers;
+       sock->nchildren = mgr->nlisteners;
 #endif
        children_size = sock->nchildren * sizeof(sock->children[0]);
        sock->children = isc_mem_get(mgr->mctx, children_size);

diff --git a/lib/isc/netmgr/udp.c b/lib/isc/netmgr/udp.c
index 00f9d40884f83e495888323336a1cc081302f36a..bc59fca54364afba53ac158c2016b1460f2e7cc3 100644 (file)
--- a/lib/isc/netmgr/udp.c
+++ b/lib/isc/netmgr/udp.c
@@ -136,7 +136,7 @@ isc_nm_listenudp(isc_nm_t *mgr, isc_sockaddr_t *iface, isc_nm_recv_cb_t cb,
        uv_os_sock_t fd = -1;
 
        /*
-        * We are creating mgr->nworkers duplicated sockets, one
+        * We are creating mgr->nlisteners duplicated sockets, one
         * socket for each worker thread.
         */
        sock = isc_mem_get(mgr->mctx, sizeof(isc_nmsocket_t));
@@ -146,7 +146,7 @@ isc_nm_listenudp(isc_nm_t *mgr, isc_sockaddr_t *iface, isc_nm_recv_cb_t cb,
 #if defined(WIN32)
        sock->nchildren = 1;
 #else
-       sock->nchildren = mgr->nworkers;
+       sock->nchildren = mgr->nlisteners;
 #endif
 
        children_size = sock->nchildren * sizeof(sock->children[0]);
@@ -847,7 +847,7 @@ isc_nm_udpconnect(isc_nm_t *mgr, isc_sockaddr_t *local, isc_sockaddr_t *peer,
                isc__nm_put_netievent_udpconnect(mgr, event);
        } else {
                atomic_init(&sock->active, false);
-               sock->tid = isc_random_uniform(mgr->nworkers);
+               sock->tid = isc_random_uniform(mgr->nlisteners);
                isc__nm_enqueue_ievent(&mgr->workers[sock->tid],
                                       (isc__netievent_t *)event);
        }