diff --git a/bin/dig/nslookup.c b/bin/dig/nslookup.c index 5687f96422..e2fa6d807f 100644 --- a/bin/dig/nslookup.c +++ b/bin/dig/nslookup.c @@ -889,7 +889,8 @@ static void start_next_command(void); static void -process_next_command(void *arg ISC_ATTR_UNUSED) { +process_next_command(void *arg ISC_ATTR_UNUSED, + isc_result_t result ISC_ATTR_UNUSED) { isc_loop_t *loop = isc_loop_main(loopmgr); if (cmdline == NULL) { in_use = false; @@ -916,11 +917,11 @@ start_next_command(void) { isc_loopmgr_pause(loopmgr); if (interactive) { - isc_work_enqueue(loop, readline_next_command, + isc_work_enqueue(loop, ISC_WORKLANE_FAST, readline_next_command, process_next_command, loop); } else { - isc_work_enqueue(loop, fgets_next_command, process_next_command, - loop); + isc_work_enqueue(loop, ISC_WORKLANE_FAST, fgets_next_command, + process_next_command, loop); } isc_loopmgr_resume(loopmgr); } diff --git a/configure.ac b/configure.ac index 2278a5d763..3f6c53e1c6 100644 --- a/configure.ac +++ b/configure.ac @@ -250,6 +250,8 @@ AC_DEFINE_UNQUOTED([RCU_VERSION], ["$RCU_VERSION"], [Compile-time Userspace-RCU CFLAGS="$CFLAGS $LIBURCU_CFLAGS" LIBS="$LIBS $LIBURCU_LIBS" +AC_CHECK_HEADERS([urcu/assert.h urcu/uatomic.h urcu/pointer.h]) + # # Userspace-RCU inlining doesn't work for rcu_deference() with some combination # of C compiler and library version. diff --git a/lib/dns/catz.c b/lib/dns/catz.c index 8736617f3e..0c5a2eec68 100644 --- a/lib/dns/catz.c +++ b/lib/dns/catz.c @@ -30,6 +30,7 @@ #include #include +#include #include #include #include @@ -115,7 +116,7 @@ dns__catz_timer_stop(void *arg); static void dns__catz_update_cb(void *data); static void -dns__catz_done_cb(void *data); +dns__catz_done_cb(void *data, isc_result_t result); static isc_result_t catz_process_zones_entry(dns_catz_zone_t *catz, dns_rdataset_t *value, @@ -2122,8 +2123,8 @@ dns__catz_timer_cb(void *arg) { ISC_LOG_INFO, "catz: %s: reload start", domain); dns_catz_zone_ref(catz); - isc_work_enqueue(catz->loop, dns__catz_update_cb, dns__catz_done_cb, - catz); + isc_work_enqueue(catz->loop, ISC_WORKLANE_SLOW, dns__catz_update_cb, + dns__catz_done_cb, catz); exit: isc_timer_destroy(&catz->updatetimer); @@ -2517,7 +2518,7 @@ exit: } static void -dns__catz_done_cb(void *data) { +dns__catz_done_cb(void *data, isc_result_t result ISC_ATTR_UNUSED) { dns_catz_zone_t *catz = (dns_catz_zone_t *)data; char dname[DNS_NAME_FORMATSIZE]; diff --git a/lib/dns/master.c b/lib/dns/master.c index d7514121a1..27b6023c94 100644 --- a/lib/dns/master.c +++ b/lib/dns/master.c @@ -2693,7 +2693,7 @@ master_load_callback(void *arg) { } static void -master_load_done(void *arg) { +master_load_done(void *arg, isc_result_t result ISC_ATTR_UNUSED) { dns_loadctx_t *lctx = arg; isc_async_run(lctx->loop, master_load_callback, lctx); @@ -2729,7 +2729,8 @@ dns_master_loadfileasync(const char *master_file, dns_name_t *top, dns_loadctx_ref(lctx); isc_loop_attach(loop, &lctx->loop); - isc_work_enqueue(isc_loop(), master_load, master_load_done, lctx); + isc_work_enqueue(isc_loop(), ISC_WORKLANE_SLOW, master_load, + master_load_done, lctx); *lctxp = lctx; diff --git a/lib/dns/masterdump.c b/lib/dns/masterdump.c index 5721e63e5b..d5618b6491 100644 --- a/lib/dns/masterdump.c +++ b/lib/dns/masterdump.c @@ -1536,7 +1536,7 @@ master_dump_callback(void *data) { } static void -master_dump_done(void *data) { +master_dump_done(void *data, isc_result_t result ISC_ATTR_UNUSED) { dns_dumpctx_t *dctx = data; isc_async_run(dctx->loop, master_dump_callback, dctx); @@ -1795,7 +1795,8 @@ dns_master_dumptostreamasync(isc_mem_t *mctx, dns_db_t *db, dns_dumpctx_ref(dctx); isc_loop_attach(loop, &dctx->loop); - isc_work_enqueue(isc_loop(), master_dump, master_dump_done, dctx); + isc_work_enqueue(isc_loop(), ISC_WORKLANE_SLOW, master_dump, + master_dump_done, dctx); *dctxp = dctx; @@ -1893,7 +1894,8 @@ dns_master_dumpasync(isc_mem_t *mctx, dns_db_t *db, dns_dbversion_t *version, dns_dumpctx_ref(dctx); isc_loop_attach(loop, &dctx->loop); - isc_work_enqueue(isc_loop(), master_dump, master_dump_done, dctx); + isc_work_enqueue(isc_loop(), ISC_WORKLANE_SLOW, master_dump, + master_dump_done, dctx); *dctxp = dctx; diff --git a/lib/dns/message.c b/lib/dns/message.c index 91fb29bdd4..0f86b89d6e 100644 --- a/lib/dns/message.c +++ b/lib/dns/message.c @@ -25,7 +25,6 @@ #include #include #include -#include #include #include #include @@ -3181,23 +3180,22 @@ dns_message_dumpsig(dns_message_t *msg, char *txt1) { #endif /* ifdef SKAN_MSG_DEBUG */ static void -checksig_done(void *arg); +checksig_done(void *arg, isc_result_t result); static void checksig_run(void *arg) { checksig_ctx_t *chsigctx = arg; chsigctx->result = dns_message_checksig(chsigctx->msg, chsigctx->view); - - isc_async_run(chsigctx->loop, checksig_done, chsigctx); } static void -checksig_done(void *arg) { +checksig_done(void *arg, isc_result_t result ISC_ATTR_UNUSED) { checksig_ctx_t *chsigctx = arg; dns_message_t *msg = chsigctx->msg; - chsigctx->cb(chsigctx->cbarg, chsigctx->result); + chsigctx->cb(chsigctx->cbarg, + (result != ISC_R_SUCCESS) ? result : chsigctx->result); dns_view_detach(&chsigctx->view); isc_loop_detach(&chsigctx->loop); @@ -3224,7 +3222,8 @@ dns_message_checksig_async(dns_message_t *msg, dns_view_t *view, dns_view_attach(view, &chsigctx->view); dns_message_clonebuffer(msg); - isc_helper_run(loop, checksig_run, chsigctx); + isc_work_enqueue(loop, ISC_WORKLANE_FAST, checksig_run, checksig_done, + chsigctx); return DNS_R_WAIT; } diff --git a/lib/dns/rpz.c b/lib/dns/rpz.c index 335dd37e0b..7ab69d4060 100644 --- a/lib/dns/rpz.c +++ b/lib/dns/rpz.c @@ -1744,7 +1744,7 @@ dns__rpz_timer_stop(void *arg) { } static void -update_rpz_done_cb(void *data) { +update_rpz_done_cb(void *data, isc_result_t result ISC_ATTR_UNUSED) { dns_rpz_zone_t *rpz = (dns_rpz_zone_t *)data; char dname[DNS_NAME_FORMATSIZE]; @@ -2050,7 +2050,8 @@ dns__rpz_timer_cb(void *arg) { ISC_LOG_INFO, "rpz: %s: reload start", domain); dns_rpz_zones_ref(rpz->rpzs); - isc_work_enqueue(rpz->loop, update_rpz_cb, update_rpz_done_cb, rpz); + isc_work_enqueue(rpz->loop, ISC_WORKLANE_SLOW, update_rpz_cb, + update_rpz_done_cb, rpz); isc_timer_destroy(&rpz->updatetimer); rpz->loop = NULL; diff --git a/lib/dns/validator.c b/lib/dns/validator.c index 0fafb68c82..a8d272b7d1 100644 --- a/lib/dns/validator.c +++ b/lib/dns/validator.c @@ -18,7 +18,6 @@ #include #include #include -#include #include #include #include @@ -134,7 +133,7 @@ validate_async_done(dns_validator_t *val, isc_result_t result); static isc_result_t validate_async_run(dns_validator_t *val, isc_job_cb cb); static isc_result_t -validate_helper_run(dns_validator_t *val, isc_job_cb cb); +validate_work_enqueue(dns_validator_t *val, isc_job_cb cb); static void validate_dnskey(void *arg); @@ -501,8 +500,8 @@ fetch_callback_dnskey(void *arg) { if (eresult == ISC_R_SUCCESS && rdataset->trust >= dns_trust_secure) { - result = validate_helper_run(val, - resume_answer_with_key); + result = validate_work_enqueue(val, + resume_answer_with_key); } else { result = validate_async_run(val, resume_answer); } @@ -678,8 +677,8 @@ validator_callback_dnskey(void *arg) { * Only extract the dst key if the keyset is secure. */ if (val->frdataset.trust >= dns_trust_secure) { - result = validate_helper_run(val, - resume_answer_with_key); + result = validate_work_enqueue(val, + resume_answer_with_key); } else { result = validate_async_run(val, resume_answer); } @@ -1292,7 +1291,8 @@ seek_dnskey(dns_validator_t *val) { dns_rdataset_disassociate(&val->fsigrdataset); } - return validate_helper_run(val, resume_answer_with_key); + return validate_work_enqueue(val, + resume_answer_with_key); } break; @@ -1766,7 +1766,7 @@ validate_answer_signing_key_done(void *arg) { val->result = ISC_R_CANCELED; } else if (val->key != NULL) { /* Process with next key if we selected one */ - (void)validate_helper_run(val, validate_answer_signing_key); + (void)validate_work_enqueue(val, validate_answer_signing_key); return; } @@ -1834,7 +1834,7 @@ validate_answer_process(void *arg) { goto next_key; } - (void)validate_helper_run(val, validate_answer_signing_key); + (void)validate_work_enqueue(val, validate_answer_signing_key); return; next_key: @@ -1958,10 +1958,15 @@ validate_async_run(dns_validator_t *val, isc_job_cb cb) { return DNS_R_WAIT; } +static void +null_done(void *arg ISC_ATTR_UNUSED, isc_result_t result ISC_ATTR_UNUSED) { + /* no-op for now */ +} + static isc_result_t -validate_helper_run(dns_validator_t *val, isc_job_cb cb) { +validate_work_enqueue(dns_validator_t *val, isc_job_cb cb) { val->attributes |= VALATTR_OFFLOADED; - isc_helper_run(val->loop, cb, val); + isc_work_enqueue(val->loop, ISC_WORKLANE_FAST, cb, null_done, val); return DNS_R_WAIT; } @@ -2249,7 +2254,7 @@ validate_dnskey_dsset_next_done(void *arg) { break; default: /* Continue validation until we have success or no more data */ - (void)validate_helper_run(val, validate_dnskey_dsset_next); + (void)validate_work_enqueue(val, validate_dnskey_dsset_next); return; } @@ -2276,8 +2281,8 @@ validate_dnskey_dsset_first(dns_validator_t *val) { /* continue async run */ result = validate_dnskey_dsset(val); if (result != ISC_R_SUCCESS) { - (void)validate_helper_run(val, - validate_dnskey_dsset_next); + (void)validate_work_enqueue(val, + validate_dnskey_dsset_next); return; } } diff --git a/lib/dns/xfrin.c b/lib/dns/xfrin.c index 8ee2cda9d7..56d12fa642 100644 --- a/lib/dns/xfrin.c +++ b/lib/dns/xfrin.c @@ -369,12 +369,13 @@ cleanup: } static void -axfr_apply_done(void *arg) { +axfr_apply_done(void *arg, isc_result_t eresult) { xfrin_work_t *work = arg; REQUIRE(VALID_XFRIN_WORK(work)); dns_xfrin_t *xfr = work->xfr; - isc_result_t result = work->result; + isc_result_t result = (eresult == ISC_R_SUCCESS) ? work->result + : eresult; REQUIRE(VALID_XFRIN(xfr)); @@ -417,7 +418,8 @@ axfr_commit(dns_xfrin_t *xfr) { .xfr = dns_xfrin_ref(xfr), }; xfr->diff_running = true; - isc_work_enqueue(xfr->loop, axfr_apply, axfr_apply_done, work); + isc_work_enqueue(xfr->loop, ISC_WORKLANE_SLOW, axfr_apply, + axfr_apply_done, work); } static isc_result_t @@ -602,14 +604,15 @@ ixfr_apply(void *arg) { } static void -ixfr_apply_done(void *arg) { +ixfr_apply_done(void *arg, isc_result_t eresult) { xfrin_work_t *work = arg; REQUIRE(VALID_XFRIN_WORK(work)); dns_xfrin_t *xfr = work->xfr; REQUIRE(VALID_XFRIN(xfr)); - isc_result_t result = work->result; + isc_result_t result = (eresult == ISC_R_SUCCESS) ? work->result + : eresult; if (atomic_load(&xfr->shuttingdown)) { result = ISC_R_SHUTTINGDOWN; @@ -621,7 +624,8 @@ ixfr_apply_done(void *arg) { if (!xfr->retry_axfr && !cds_wfcq_empty(&xfr->diff_head, &xfr->diff_tail)) { - isc_work_enqueue(xfr->loop, ixfr_apply, ixfr_apply_done, work); + isc_work_enqueue(xfr->loop, ISC_WORKLANE_SLOW, ixfr_apply, + ixfr_apply_done, work); return; } @@ -701,7 +705,8 @@ ixfr_commit(dns_xfrin_t *xfr) { .xfr = dns_xfrin_ref(xfr), }; xfr->diff_running = true; - isc_work_enqueue(xfr->loop, ixfr_apply, ixfr_apply_done, work); + isc_work_enqueue(xfr->loop, ISC_WORKLANE_SLOW, ixfr_apply, + ixfr_apply_done, work); } cleanup: diff --git a/lib/isc/Makefile.am b/lib/isc/Makefile.am index c0769ec527..39208e7962 100644 --- a/lib/isc/Makefile.am +++ b/lib/isc/Makefile.am @@ -31,7 +31,6 @@ libisc_la_HEADERS = \ include/isc/hash.h \ include/isc/hashmap.h \ include/isc/heap.h \ - include/isc/helper.h \ include/isc/hex.h \ include/isc/histo.h \ include/isc/hmac.h \ @@ -136,7 +135,6 @@ libisc_la_SOURCES = \ hash.c \ hashmap.c \ heap.c \ - helper.c \ hex.c \ histo.c \ hmac.c \ diff --git a/lib/isc/helper.c b/lib/isc/helper.c deleted file mode 100644 index f5a83cc833..0000000000 --- a/lib/isc/helper.c +++ /dev/null @@ -1,68 +0,0 @@ -/* - * Copyright (C) Internet Systems Consortium, Inc. ("ISC") - * - * SPDX-License-Identifier: MPL-2.0 - * - * This Source Code Form is subject to the terms of the Mozilla Public - * License, v. 2.0. If a copy of the MPL was not distributed with this - * file, you can obtain one at https://mozilla.org/MPL/2.0/. - * - * See the COPYRIGHT file distributed with this work for additional - * information regarding copyright ownership. - */ - -#include -#include -#include - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include "async_p.h" -#include "job_p.h" -#include "loop_p.h" - -void -isc_helper_run(isc_loop_t *loop, isc_job_cb cb, void *cbarg) { - REQUIRE(VALID_LOOP(loop)); - REQUIRE(cb != NULL); - - isc_loop_t *helper = &loop->loopmgr->helpers[loop->tid]; - - isc_job_t *job = isc_mem_get(helper->mctx, sizeof(*job)); - *job = (isc_job_t){ - .cb = cb, - .cbarg = cbarg, - }; - - cds_wfcq_node_init(&job->wfcq_node); - - /* - * cds_wfcq_enqueue() is non-blocking and enqueues the job to async - * queue. - * - * The function returns 'false' in case the queue was empty - in such - * case we need to trigger the async callback. - */ - if (!cds_wfcq_enqueue(&helper->async_jobs.head, - &helper->async_jobs.tail, &job->wfcq_node)) - { - int r = uv_async_send(&helper->async_trigger); - UV_RUNTIME_CHECK(uv_async_send, r); - } -} diff --git a/lib/isc/httpd.c b/lib/isc/httpd.c index 809cb30728..051f92f500 100644 --- a/lib/isc/httpd.c +++ b/lib/isc/httpd.c @@ -875,7 +875,7 @@ prepare_response(void *arg) { } static void -prepare_response_done(void *arg) { +prepare_response_done(void *arg, isc_result_t result) { isc_region_t r; isc_httpd_sendreq_t *req = arg; isc_httpd_t *httpd = req->httpd; @@ -885,6 +885,11 @@ prepare_response_done(void *arg) { */ isc_buffer_usedregion(req->sendbuffer, &r); + if (result != ISC_R_SUCCESS) { + httpd_senddone(httpd->handle, result, req); + return; + } + isc_nm_send(httpd->handle, &r, httpd_senddone, req); } @@ -945,8 +950,8 @@ httpd_request(isc_nmhandle_t *handle, isc_result_t eresult, isc_httpd_sendreq_t *req = isc__httpd_sendreq_new(httpd); isc_nmhandle_ref(handle); - isc_work_enqueue(isc_loop(), prepare_response, prepare_response_done, - req); + isc_work_enqueue(isc_loop(), ISC_WORKLANE_SLOW, prepare_response, + prepare_response_done, req); return; close_readhandle: diff --git a/lib/isc/include/isc/helper.h b/lib/isc/include/isc/helper.h deleted file mode 100644 index fa13f0baf2..0000000000 --- a/lib/isc/include/isc/helper.h +++ /dev/null @@ -1,45 +0,0 @@ -/* - * Copyright (C) Internet Systems Consortium, Inc. ("ISC") - * - * SPDX-License-Identifier: MPL-2.0 - * - * This Source Code Form is subject to the terms of the Mozilla Public - * License, v. 2.0. If a copy of the MPL was not distributed with this - * file, you can obtain one at https://mozilla.org/MPL/2.0/. - * - * See the COPYRIGHT file distributed with this work for additional - * information regarding copyright ownership. - */ - -/*! \file isc/helper.h */ - -#pragma once - -#include - -#include -#include -#include -#include -#include - -ISC_LANG_BEGINDECLS - -void -isc_helper_run(isc_loop_t *loop, isc_job_cb cb, void *cbarg); -/*%< - * Schedule the job callback 'cb' to be run on the 'loop' event loop. - * - * Requires: - * - *\li 'loop' is a valid isc event loop - *\li 'cb' is a callback function, must be non-NULL - *\li 'cbarg' is passed to the 'cb' as the only argument, may be NULL - */ - -#define isc_helper_current(cb, cbarg) isc_async_run(isc_loop(), cb, cbarg) -/*%< - * Helper macro to run the job on the current loop - */ - -ISC_LANG_ENDDECLS diff --git a/lib/isc/include/isc/loop.h b/lib/isc/include/isc/loop.h index fe3545115a..6505000a0b 100644 --- a/lib/isc/include/isc/loop.h +++ b/lib/isc/include/isc/loop.h @@ -20,6 +20,8 @@ #include #include #include +#include +#include typedef void (*isc_job_cb)(void *); @@ -225,4 +227,5 @@ isc_loop_shuttingdown(isc_loop_t *loop); * * \li 'loop' is a valid loop and the loop tid matches the current tid. */ + ISC_LANG_ENDDECLS diff --git a/lib/isc/include/isc/queue.h b/lib/isc/include/isc/queue.h index 02f94d866d..1996ba195c 100644 --- a/lib/isc/include/isc/queue.h +++ b/lib/isc/include/isc/queue.h @@ -24,7 +24,7 @@ typedef struct isc_queue { sizeof(struct __cds_wfcq_head)]; struct cds_wfcq_tail tail; uint8_t __padding_tail[ISC_OS_CACHELINE_SIZE - - sizeof(struct __cds_wfcq_head)]; + sizeof(struct cds_wfcq_tail)]; } isc_queue_t; typedef struct cds_wfcq_node isc_queue_node_t; diff --git a/lib/isc/include/isc/urcu.h b/lib/isc/include/isc/urcu.h index 6747dcc6d7..a78200a93a 100644 --- a/lib/isc/include/isc/urcu.h +++ b/lib/isc/include/isc/urcu.h @@ -27,9 +27,20 @@ #include #endif -#include - +#if HAVE_URCU_ASSERT_H +#include +#endif +#if HAVE_URCU_UATOMIC_H +#include +#endif #include +#include +#include +#if HAVE_URCU_POINTER_H +#include +#else +#include +#endif #include #include #include @@ -167,4 +178,17 @@ #undef _CMM_STORE_SHARED #define _CMM_STORE_SHARED(x, v) CMM_STORE_SHARED(x, v) +#endif /* __SANITIZE_THREAD */ + +#if !defined(uatomic_load) || !defined(uatomic_store) +#define uatomic_load(ptr, mo) uatomic_read(ptr) +#define uatomic_store(ptr, v, mo) uatomic_set(ptr, v) + +#define CMM_RELAXED __ATOMIC_RELAXED +#define CMM_CONSUME __ATOMIC_CONSUME +#define CMM_ACQUIRE __ATOMIC_ACQUIRE +#define CMM_RELEASE __ATOMIC_RELEASE +#define CMM_ACQ_REL __ATOMIC_ACQ_REL +#define CMM_SEQ_CST __ATOMIC_SEQ_CST + #endif diff --git a/lib/isc/include/isc/work.h b/lib/isc/include/isc/work.h index df196e95ea..b84a98a85b 100644 --- a/lib/isc/include/isc/work.h +++ b/lib/isc/include/isc/work.h @@ -11,31 +11,118 @@ * information regarding copyright ownership. */ +/*! \file isc/work.h + * \brief Offload work from an event loop onto a dedicated worker thread. + * + * Each isc event loop has one worker thread per lane (see isc_worklane_t). + * isc_work_enqueue() runs a callback on the worker thread bound to the calling + * loop's lane and, when it finishes, runs a second callback back on that loop. + * The handle it returns can be used to cancel a task that has not started + * running yet. + */ + #pragma once -#include - #include -#include +#include + +typedef enum isc_worklane { + ISC_WORKLANE_FAST = 0, /*%< short, bounded tasks (e.g. crypto) */ + ISC_WORKLANE_SLOW, /*%< blocking/long tasks (e.g. zone dump) */ + ISC_WORKLANE_COUNT, +} isc_worklane_t; +/*%< + * Selects which per-loop worker thread runs an enqueued task. Keeping long, + * blocking SLOW work (disk I/O, zone dump/load) on its own lane stops it from + * holding up short FAST tasks behind it. + */ typedef void (*isc_work_cb)(void *arg); -typedef void (*isc_after_work_cb)(void *arg); +typedef void (*isc_work_done_cb)(void *arg, isc_result_t result); typedef struct isc_work isc_work_t; ISC_LANG_BEGINDECLS -void -isc_work_enqueue(isc_loop_t *loop, isc_work_cb work_cb, - isc_after_work_cb after_work_cb, void *cbarg); +isc_work_t * +isc_work_enqueue(isc_loop_t *loop, isc_worklane_t lane, isc_work_cb cb, + isc_work_done_cb done_cb, void *cbarg); /*%< - * Schedules work to be handled by the libuv thread pool (see uv_work_t). - * The function specified in `work_cb` will be run by a thread in the - * thread pool; when complete, the `after_work_cb` function will run - * in 'loop' to inform the caller that the work was completed. + * Schedule 'cb' to run on the worker thread bound to 'loop' and 'lane'. When + * 'cb' returns, 'done_cb' is scheduled back on 'loop' with the result of the + * work: ISC_R_SUCCESS normally, or ISC_R_CANCELED if the task was canceled + * before it started (see isc_work_cancel()). + * + * Returns a handle that may be passed to isc_work_cancel(). The handle is + * owned by 'loop' and stays valid until 'after_cb' has run; it must not be used + * afterwards. * * Requires: - * \li 'loop' is a valid event loop. - * \li 'work_cb' and 'after_work_cb' are not NULL. + * + *\li 'loop' is a valid isc event loop. + *\li 'cb' is non-NULL. + *\li 'after_cb' is non-NULL. + *\li 'cbarg' is passed to both callbacks, may be NULL. + */ + +bool +isc_work_cancel(isc_work_t *work); +/*%< + * Try to cancel 'work' before its 'cb' starts running. If the task is still + * queued it is marked canceled and 'cb' will not run; if it is already running + * or has finished this has no effect. Either way the 'after_cb' passed to + * isc_work_enqueue() still runs on the origin loop, with ISC_R_CANCELED when + * the cancel succeeded. Nothing is freed here. + * + * Returns: + * + *\li true the task was still queued; 'cb' will not run. + *\li false the task is already running or done (uv_cancel() semantics). + * + * Requires: + * + *\li 'work' is a handle from isc_work_enqueue() whose 'after_cb' has not run. + */ + +/* private */ + +typedef struct isc__workthread isc__workthread_t; + +isc__workthread_t * +isc__workthread_create(isc_loop_t *loop, isc_worklane_t lane); +/*%< + * Create one worker thread for 'lane' with its own dispatch queue (the loop + * manager creates one per loop). Used by the loop manager; not for general + * use. + */ + +void +isc__workthread_shutdown(isc__workthread_t *thread); +/*%< + * Begin shutdown of 'thread': set its SHUTDOWN flag (after which new enqueues + * run inline on the caller instead of queuing), then, after an RCU grace period + * that fences any in-flight enqueue, wake the worker so it drains its queue and + * exits. Does not join the worker; see isc__workthread_destroy(). Idempotent. + */ + +void +isc__workthread_pause(isc__workthread_t *thread); +/*%< + * Quiesce 'thread' for isc_loopmgr_pause(): mark it paused and block until it + * has parked. A no-op if the worker is already shutting down. Must be paired + * with isc__workthread_resume() and called from the worker's owning loop. + */ + +void +isc__workthread_resume(isc__workthread_t *thread); +/*%< + * Release a worker previously parked by isc__workthread_pause(). + */ + +void +isc__workthread_destroy(isc__workthread_t **threadp); +/*%< + * Join the worker thread, then free '*threadp' and set it to NULL. Must be + * called after isc__workthread_shutdown(). */ ISC_LANG_ENDDECLS diff --git a/lib/isc/loop.c b/lib/isc/loop.c index dafeeb8b50..141d954cb7 100644 --- a/lib/isc/loop.c +++ b/lib/isc/loop.c @@ -92,6 +92,11 @@ static void pause_loop(isc_loop_t *loop) { isc_loopmgr_t *loopmgr = loop->loopmgr; + /* Quiesce this loop's own workers before going to the barrier. */ + for (isc_worklane_t lane = 0; lane < ISC_WORKLANE_COUNT; lane++) { + isc__workthread_pause(loop->workthreads[lane]); + } + rcu_thread_offline(); loop->paused = true; @@ -106,6 +111,10 @@ resume_loop(isc_loop_t *loop) { loop->paused = false; rcu_thread_online(); + + for (isc_worklane_t lane = 0; lane < ISC_WORKLANE_COUNT; lane++) { + isc__workthread_resume(loop->workthreads[lane]); + } } static void @@ -178,6 +187,10 @@ shutdown_cb(uv_async_t *handle) { isc_signal_destroy(&loopmgr->sigint); } + for (isc_worklane_t lane = 0; lane < ISC_WORKLANE_COUNT; lane++) { + isc__workthread_shutdown(loop->workthreads[lane]); + } + enum cds_wfcq_ret ret = __cds_wfcq_splice_blocking( &loop->async_jobs.head, &loop->async_jobs.tail, &loop->teardown_jobs.head, &loop->teardown_jobs.tail); @@ -251,16 +264,6 @@ quiescent_cb(uv_prepare_t *handle) { #endif } -static void -helper_close(isc_loop_t *loop) { - int r = uv_loop_close(&loop->loop); - UV_RUNTIME_CHECK(uv_loop_close, r); - - INSIST(cds_wfcq_empty(&loop->async_jobs.head, &loop->async_jobs.tail)); - - isc_mem_detach(&loop->mctx); -} - static void loop_close(isc_loop_t *loop) { int r = uv_loop_close(&loop->loop); @@ -274,32 +277,10 @@ loop_close(isc_loop_t *loop) { isc_mem_detach(&loop->mctx); } -static void * -helper_thread(void *arg) { - isc_loop_t *helper = (isc_loop_t *)arg; - - int r = uv_prepare_start(&helper->quiescent, quiescent_cb); - UV_RUNTIME_CHECK(uv_prepare_start, r); - - isc_barrier_wait(&helper->loopmgr->starting); - - r = uv_run(&helper->loop, UV_RUN_DEFAULT); - UV_RUNTIME_CHECK(uv_run, r); - - /* Invalidate the helper early */ - helper->magic = 0; - - isc_barrier_wait(&helper->loopmgr->stopping); - - return NULL; -} - static void * loop_thread(void *arg) { isc_loop_t *loop = (isc_loop_t *)arg; isc_loopmgr_t *loopmgr = loop->loopmgr; - isc_loop_t *helper = &loopmgr->helpers[loop->tid]; - char name[32]; /* Initialize the thread_local variables*/ REQUIRE(isc__loop_local == NULL || isc__loop_local == loop); @@ -307,14 +288,13 @@ loop_thread(void *arg) { isc__tid_init(loop->tid); - /* Start the helper thread */ - isc_thread_create(helper_thread, helper, &helper->thread); - snprintf(name, sizeof(name), "isc-helper-%04" PRIu32, loop->tid); - isc_thread_setname(helper->thread, name); - int r = uv_prepare_start(&loop->quiescent, quiescent_cb); UV_RUNTIME_CHECK(uv_prepare_start, r); + for (isc_worklane_t lane = 0; lane < ISC_WORKLANE_COUNT; lane++) { + loop->workthreads[lane] = isc__workthread_create(loop, lane); + } + isc_barrier_wait(&loopmgr->starting); enum cds_wfcq_ret ret = __cds_wfcq_splice_blocking( @@ -333,12 +313,12 @@ loop_thread(void *arg) { /* Invalidate the loop early */ loop->magic = 0; - /* Shutdown the helper thread */ - r = uv_async_send(&helper->shutdown_trigger); - UV_RUNTIME_CHECK(uv_async_send, r); - isc_barrier_wait(&loopmgr->stopping); + for (isc_worklane_t lane = 0; lane < ISC_WORKLANE_COUNT; lane++) { + isc__workthread_destroy(&loop->workthreads[lane]); + } + return NULL; } @@ -346,17 +326,6 @@ loop_thread(void *arg) { * Public */ -static void -threadpool_initialize(uint32_t workers) { - 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); - uv_os_setenv("UV_THREADPOOL_SIZE", buf); - } -} - static void loop_destroy(isc_loop_t *loop) { int r = uv_async_send(&loop->destroy_trigger); @@ -376,7 +345,6 @@ isc_loopmgr_create(isc_mem_t *mctx, uint32_t nloops, isc_loopmgr_t **loopmgrp) { REQUIRE(loopmgrp != NULL && *loopmgrp == NULL); REQUIRE(nloops > 0); - threadpool_initialize(nloops); isc__tid_initcount(nloops); loopmgr = isc_mem_get(mctx, sizeof(*loopmgr)); @@ -386,11 +354,13 @@ isc_loopmgr_create(isc_mem_t *mctx, uint32_t nloops, isc_loopmgr_t **loopmgrp) { isc_mem_attach(mctx, &loopmgr->mctx); - /* We need to double the number for loops and helpers */ - isc_barrier_init(&loopmgr->pausing, loopmgr->nloops * 2); - isc_barrier_init(&loopmgr->resuming, loopmgr->nloops * 2); - isc_barrier_init(&loopmgr->starting, loopmgr->nloops * 2); - isc_barrier_init(&loopmgr->stopping, loopmgr->nloops * 2); + /* We need to double the number for loops */ + isc_barrier_init(&loopmgr->pausing, loopmgr->nloops); + isc_barrier_init(&loopmgr->resuming, loopmgr->nloops); + isc_barrier_init(&loopmgr->starting, + loopmgr->nloops * (1 + ISC_WORKLANE_COUNT)); + isc_barrier_init(&loopmgr->stopping, + loopmgr->nloops * (1 + ISC_WORKLANE_COUNT)); loopmgr->loops = isc_mem_cget(loopmgr->mctx, loopmgr->nloops, sizeof(loopmgr->loops[0])); @@ -399,13 +369,6 @@ isc_loopmgr_create(isc_mem_t *mctx, uint32_t nloops, isc_loopmgr_t **loopmgrp) { loop_init(loop, loopmgr, i, "loop"); } - loopmgr->helpers = isc_mem_cget(loopmgr->mctx, loopmgr->nloops, - sizeof(loopmgr->helpers[0])); - for (size_t i = 0; i < loopmgr->nloops; i++) { - isc_loop_t *loop = &loopmgr->helpers[i]; - loop_init(loop, loopmgr, i, "helper"); - } - loopmgr->sigint = isc_signal_new(loopmgr, isc__loopmgr_signal, loopmgr, SIGINT); loopmgr->sigterm = isc_signal_new(loopmgr, isc__loopmgr_signal, loopmgr, @@ -525,13 +488,6 @@ isc_loopmgr_pause(isc_loopmgr_t *loopmgr) { "loop exclusive mode: starting"); } - for (size_t i = 0; i < loopmgr->nloops; i++) { - isc_loop_t *helper = &loopmgr->helpers[i]; - - int r = uv_async_send(&helper->pause_trigger); - UV_RUNTIME_CHECK(uv_async_send, r); - } - for (size_t i = 0; i < loopmgr->nloops; i++) { isc_loop_t *loop = &loopmgr->loops[i]; @@ -589,12 +545,6 @@ isc_loopmgr_destroy(isc_loopmgr_t **loopmgrp) { RUNTIME_CHECK(atomic_compare_exchange_strong(&loopmgr->running, &(bool){ true }, false)); - /* Wait for all helpers to finish */ - for (size_t i = 0; i < loopmgr->nloops; i++) { - isc_loop_t *helper = &loopmgr->helpers[i]; - isc_thread_join(helper->thread, NULL); - } - /* First wait for all loops to finish */ for (size_t i = 1; i < loopmgr->nloops; i++) { isc_loop_t *loop = &loopmgr->loops[i]; @@ -603,13 +553,6 @@ isc_loopmgr_destroy(isc_loopmgr_t **loopmgrp) { loopmgr->magic = 0; - for (size_t i = 0; i < loopmgr->nloops; i++) { - isc_loop_t *helper = &loopmgr->helpers[i]; - helper_close(helper); - } - isc_mem_cput(loopmgr->mctx, loopmgr->helpers, loopmgr->nloops, - sizeof(loopmgr->helpers[0])); - for (size_t i = 0; i < loopmgr->nloops; i++) { isc_loop_t *loop = &loopmgr->loops[i]; loop_close(loop); @@ -697,3 +640,21 @@ isc_loop_shuttingdown(isc_loop_t *loop) { return loop->shuttingdown; } + +isc__workthread_t * +isc__loopmgr_workthread(isc_loop_t *loop, isc_worklane_t lane) { + REQUIRE(VALID_LOOP(loop)); + REQUIRE(lane < ISC_WORKLANE_COUNT); + + return loop->workthreads[lane]; +} + +void +isc__loopmgr_starting(isc_loop_t *loop) { + isc_barrier_wait(&loop->loopmgr->starting); +} + +void +isc__loopmgr_stopping(isc_loop_t *loop) { + isc_barrier_wait(&loop->loopmgr->stopping); +} diff --git a/lib/isc/loop_p.h b/lib/isc/loop_p.h index a822004e88..35ec4bb689 100644 --- a/lib/isc/loop_p.h +++ b/lib/isc/loop_p.h @@ -76,6 +76,9 @@ struct isc_loop { /* safe memory reclamation */ uv_prepare_t quiescent; + + /* thread pools */ + isc__workthread_t *workthreads[ISC_WORKLANE_COUNT]; }; /* @@ -110,7 +113,6 @@ struct isc_loopmgr { /* per-thread objects */ isc_loop_t *loops; - isc_loop_t *helpers; }; /* @@ -134,18 +136,15 @@ struct isc_signal { #define JOB_MAGIC ISC_MAGIC('J', 'O', 'B', ' ') #define VALID_JOB(t) ISC_MAGIC_VALID(t, JOB_MAGIC) -/* - * Work to be offloaded to an external thread. - */ -struct isc_work { - uv_work_t work; - isc_loop_t *loop; - isc_work_cb work_cb; - isc_after_work_cb after_work_cb; - void *cbarg; -}; - #define DEFAULT_LOOP(loopmgr) (&(loopmgr)->loops[0]) #define CURRENT_LOOP(loopmgr) (&(loopmgr)->loops[isc_tid()]) #define LOOP(loopmgr, tid) (&(loopmgr)->loops[tid]) #define ON_LOOP(loop) ((loop) == CURRENT_LOOP((loop)->loopmgr)) + +isc__workthread_t * +isc__loopmgr_workthread(isc_loop_t *loop, isc_worklane_t lane); + +void +isc__loopmgr_starting(isc_loop_t *loop); +void +isc__loopmgr_stopping(isc_loop_t *loop); diff --git a/lib/isc/work.c b/lib/isc/work.c index 0b7cdf5743..4e2c159d20 100644 --- a/lib/isc/work.c +++ b/lib/isc/work.c @@ -11,69 +11,434 @@ * information regarding copyright ownership. */ -#include +#include +#include +#include -#include +#include #include #include +#include +#include +#include #include +#include #include #include #include "loop_p.h" +#define WORK_MAGIC ISC_MAGIC('W', 'o', 'r', 'k') +#define VALID_WORK(t) ISC_MAGIC_VALID(t, WORK_MAGIC) +#define WORKTHREAD_MAGIC ISC_MAGIC('W', 'k', 'T', 'h') +#define VALID_WORKTHREAD(t) ISC_MAGIC_VALID(t, WORKTHREAD_MAGIC) + +enum waitstate { + /* The value a sleeping worker blocks on in FUTEX_WAIT. */ + THREAD_WAITING = 0, + /* Any non-zero bit keeps FUTEX_WAIT from blocking. */ + THREAD_WAKEUP = (1 << 0), + THREAD_RUNNING = (1 << 1), + THREAD_SHUTDOWN = (1 << 2), + THREAD_PAUSE = (1 << 3), /* request from the owning loop */ + THREAD_PAUSED = (1 << 4), /* ack from the worker */ +}; + +/* Sticky bits a paused worker must not drop. */ +#define THREAD_STICKY (THREAD_SHUTDOWN | THREAD_PAUSE | THREAD_PAUSED) + +enum workstate { + WORK_QUEUED = 0, + WORK_RUNNING, + WORK_CANCELED, +}; + +struct isc_work { + unsigned int magic; + uint32_t state; /* enum workstate */ + isc_work_cb cb; /* runs on a worker thread */ + isc_work_done_cb done_cb; /* runs on the origin loop */ + void *cbarg; + isc_loop_t *loop; /* origin loop, referenced */ + struct cds_wfcq_node node; /* dispatch queue linkage */ +}; + +typedef struct isc__workthread { + union { + struct { + unsigned int magic; + isc_worklane_t lane; + isc_loop_t *loop; + isc_thread_t thread; + struct __cds_wfcq_head qhead; + int32_t state; /* enum waitstate */ + }; + uint8_t __padding0[ISC_OS_CACHELINE_SIZE]; + }; + union { + struct cds_wfcq_tail qtail; + uint8_t __padding1[ISC_OS_CACHELINE_SIZE]; + }; +} isc__workthread_t; + +STATIC_ASSERT(ISC_OS_CACHELINE_SIZE >= sizeof(struct cds_wfcq_tail), + "ISC_OS_CACHELINE_SIZE smaller than sizeof(struct " + "cds_wfcq_tail)"); +STATIC_ASSERT(offsetof(isc__workthread_t, qtail) == ISC_OS_CACHELINE_SIZE, + "isc__workthread_t.qtail not on second cacheline"); +STATIC_ASSERT(sizeof(isc__workthread_t) == 2 * ISC_OS_CACHELINE_SIZE, + "isc__workthread_t is not two cachelines"); + static void -isc__work_cb(uv_work_t *req) { - isc_work_t *work = uv_req_get_data((uv_req_t *)req); +workthread_wake(isc__workthread_t *thread) { + cmm_smp_mb(); + if ((uatomic_load(&thread->state, CMM_RELAXED) & THREAD_RUNNING) != 0) { + /* Actively running; it will notice the queue on its own. */ + return; + } - isc__iterated_hash_initialize(); - - rcu_register_thread(); - - work->work_cb(work->cbarg); - - rcu_unregister_thread(); - - isc__iterated_hash_shutdown(); + uatomic_or(&thread->state, THREAD_WAKEUP); + if (futex_noasync(&thread->state, FUTEX_WAKE, 1, NULL, NULL, 0) < 0) { + FATAL_ERROR("futex_noasync(FUTEX_WAKE): %s", strerror(errno)); + } } static void -isc__after_work_cb(uv_work_t *req, int status) { - isc_work_t *work = uv_req_get_data((uv_req_t *)req); +workthread_slumber(isc__workthread_t *thread) { + rcu_thread_offline(); + while (futex_noasync(&thread->state, FUTEX_WAIT, THREAD_WAITING, NULL, + NULL, 0) != 0) + { + if (errno == EWOULDBLOCK) { + break; + } else if (errno != EINTR) { + FATAL_ERROR("futex_noasync(FUTEX_WAIT): %s", + strerror(errno)); + } + /* Or retry if interrupted by signal. */ + } + rcu_thread_online(); +} + +static void +workthread_sleep(isc__workthread_t *thread) { + /* + * Drop to WAITING while keeping a pending SHUTDOWN/PAUSE sticky, so the + * FUTEX_WAIT below refuses to block once either is signalled. + */ + uatomic_and(&thread->state, THREAD_STICKY); + cmm_smp_mb(); + + /* + * The queue is the one wake condition that can't live in 'state', so + * recheck it under the fence; SHUTDOWN and WAKEUP are handled by + * FUTEX_WAIT's own value check. + */ + if (cds_wfcq_empty(&thread->qhead, &thread->qtail)) { + workthread_slumber(thread); + } + + /* Tell the waker we are running (keeping any sticky SHUTDOWN/PAUSE). */ + uatomic_or(&thread->state, THREAD_RUNNING); +} + +/* + * Acknowledge a pause request: publish PAUSED (dropping RUNNING/WAKEUP) and + * wake the waiting pauser. A new pause clears PAUSED, so the worker re-acks + * and the pauser only ever observes an ack set for its own request, never a + * stale one from the previous pause generation. + */ +static void +workthread_ack_pause(isc__workthread_t *thread) { + int32_t old, next; + do { + old = uatomic_load(&thread->state, CMM_RELAXED); + next = (old & THREAD_STICKY) | THREAD_PAUSED; + } while (uatomic_cmpxchg(&thread->state, old, next) != old); + + (void)futex_noasync(&thread->state, FUTEX_WAKE, INT_MAX, NULL, NULL, 0); +} + +/* + * Honour a pause: pause until the owning loop clears PAUSE (resume). A fresh + * pause clears PAUSED (see isc__workthread_pause), so (re-)ack whenever PAUSED + * is gone — the pauser only proceeds on an ack set for *its* request, never a + * stale one from the previous generation. Stays RCU-offline while paused so + * it can't hold up an exclusive-mode grace period. + */ +static void +workthread_pause(isc__workthread_t *thread) { + rcu_thread_offline(); + + while (true) { + int32_t old = uatomic_load(&thread->state, CMM_ACQUIRE); + if ((old & (THREAD_PAUSE | THREAD_SHUTDOWN)) != THREAD_PAUSE) { + break; + } + if ((old & THREAD_PAUSED) == 0) { + workthread_ack_pause(thread); + continue; + } + (void)futex_noasync(&thread->state, FUTEX_WAIT, old, NULL, NULL, + 0); + } + + uatomic_and(&thread->state, ~THREAD_PAUSED); + rcu_thread_online(); +} + +static void +work_done(void *arg) { + isc_work_t *work = arg; isc_loop_t *loop = work->loop; + isc_result_t result = (uatomic_load(&work->state, CMM_ACQUIRE) != + WORK_CANCELED) + ? ISC_R_SUCCESS + : ISC_R_CANCELED; - UV_RUNTIME_CHECK(uv_after_work_cb, status); + work->done_cb(work->cbarg, result); - work->after_work_cb(work->cbarg); + work->magic = 0; + isc_mem_put(work->loop->mctx, work, sizeof(*work)); + isc_loop_unref(loop); +} - isc_mem_put(loop->mctx, work, sizeof(*work)); +static void +work_run(void *arg) { + isc_work_t *work = arg; + /* + * The CAS *is* the tombstone check: whoever moves the item out + * of WORK_QUEUED first — this worker or isc_work_cancel() — + * decides whether the callback runs. uatomic_cmpxchg returns the + * prior state, so WORK_QUEUED means we won the race. + */ + uint32_t prev = uatomic_cmpxchg(&work->state, WORK_QUEUED, + WORK_RUNNING); + switch (prev) { + case WORK_QUEUED: + work->cb(work->cbarg); + break; + case WORK_CANCELED: + break; + default: + UNREACHABLE(); + } - isc_loop_detach(&loop); + /* Completion always routes back to the origin loop. */ + isc_async_run(work->loop, work_done, work); +} + +static void * +workthread_thread(void *arg) { + isc__workthread_t *thread = arg; + + isc__loopmgr_starting(thread->loop); + + while (true) { + /* + * Honour a pause before touching the queue (gated on !SHUTDOWN + * so a shutting-down worker exits instead of pausing). + */ + int32_t state = uatomic_load(&thread->state, CMM_ACQUIRE); + if ((state & (THREAD_PAUSE | THREAD_SHUTDOWN)) == THREAD_PAUSE) + { + workthread_pause(thread); + continue; + } + + struct cds_wfcq_node *node; + node = __cds_wfcq_dequeue_blocking(&thread->qhead, + &thread->qtail); + + if (node == NULL) { + /* + * Only exit the loop if there's nothing to do. + */ + if ((uatomic_load(&thread->state, CMM_ACQUIRE) & + THREAD_SHUTDOWN) != 0) + { + synchronize_rcu(); + if (!cds_wfcq_empty(&thread->qhead, + &thread->qtail)) + { + continue; + } + break; + } + + workthread_sleep(thread); + + continue; + } + + isc_work_t *work = caa_container_of(node, isc_work_t, node); + work_run(work); + } + + isc__loopmgr_stopping(thread->loop); + + return NULL; +} + +isc_work_t * +isc_work_enqueue(isc_loop_t *loop, isc_worklane_t lane, isc_work_cb cb, + isc_work_done_cb done_cb, void *cbarg) { + REQUIRE(loop == isc_loop()); + + isc__workthread_t *thread = isc__loopmgr_workthread(loop, lane); + + isc_work_t *work = isc_mem_get(loop->mctx, sizeof(*work)); + *work = (isc_work_t){ + .magic = WORK_MAGIC, + .cb = cb, + .done_cb = done_cb, + .cbarg = cbarg, + .loop = isc_loop_ref(loop), + .state = WORK_QUEUED, + }; + + rcu_read_lock(); + if ((uatomic_load(&thread->state, CMM_ACQUIRE) & THREAD_SHUTDOWN) != 0) + { + rcu_read_unlock(); + + /* + * We are shutting down, so immedaitely run task instead of + * adding more in the queue. (The worker is running the + * remaining enqueue tasks and shutdown after, see + * workthread_thread().) + */ + isc_async_run(loop, work_run, work); + } else { + (void)cds_wfcq_enqueue(&thread->qhead, &thread->qtail, + &work->node); + rcu_read_unlock(); + + if ((uatomic_load(&thread->state, CMM_ACQUIRE) & + THREAD_RUNNING) == 0) + { + workthread_wake(thread); + } + } + + return work; +} + +bool +isc_work_cancel(isc_work_t *work) { + REQUIRE(VALID_WORK(work)); + + /* + * Tombstone: QUEUED -> CANCELED. The node stays in the queue + * (no interior unlink in a singly-linked lock-free queue) and + * is discarded by whichever worker dequeues it; after_cb still + * fires with ISC_R_CANCELED. Nothing is freed here. False + * means the callback is running or done — uv_cancel semantics. + */ + return uatomic_cmpxchg(&work->state, WORK_QUEUED, WORK_CANCELED) == + WORK_QUEUED; +} + +isc__workthread_t * +isc__workthread_create(isc_loop_t *loop, isc_worklane_t lane) { + isc__workthread_t *thread = isc_mem_get(loop->mctx, sizeof(*thread)); + + *thread = (isc__workthread_t){ + .lane = lane, + .magic = WORKTHREAD_MAGIC, + .state = THREAD_WAITING, + .loop = loop, + }; + + __cds_wfcq_init(&thread->qhead, &thread->qtail); + + isc_thread_create(workthread_thread, thread, &thread->thread); + + return thread; } void -isc_work_enqueue(isc_loop_t *loop, isc_work_cb work_cb, - isc_after_work_cb after_work_cb, void *cbarg) { - isc_work_t *work = NULL; - int r; +isc__workthread_shutdown(isc__workthread_t *thread) { + REQUIRE(VALID_WORKTHREAD(thread)); - REQUIRE(VALID_LOOP(loop)); - REQUIRE(isc_loop() == loop); - REQUIRE(work_cb != NULL); - REQUIRE(after_work_cb != NULL); + /* + * Not called while the worker is paused by isc__workthread_pause(): + * shutdown callbacks run from uv loops, and loopmgr pause keeps every + * loop out of uv_run() until resume, so PAUSE and SHUTDOWN never + * coexist on a worker (the SHUTDOWN checks in the pause path are only + * a belt-and-braces exit if that ever changed). + */ - work = isc_mem_get(loop->mctx, sizeof(*work)); - *work = (isc_work_t){ - .work_cb = work_cb, - .after_work_cb = after_work_cb, - .cbarg = cbarg, - }; + /* Set the sticky SHUTDOWN bit once; bail if already shutting down. */ + int32_t old; + do { + old = uatomic_load(&thread->state, CMM_RELAXED); + if ((old & THREAD_SHUTDOWN) != 0) { + return; + } + } while (uatomic_cmpxchg(&thread->state, old, old | THREAD_SHUTDOWN) != + old); - isc_loop_attach(loop, &work->loop); + /* Fence in-flight enqueues (which touch the queue) before draining. */ + synchronize_rcu(); - uv_req_set_data((uv_req_t *)&work->work, work); - - r = uv_queue_work(&loop->loop, &work->work, isc__work_cb, - isc__after_work_cb); - UV_RUNTIME_CHECK(uv_queue_work, r); + workthread_wake(thread); +} + +void +isc__workthread_destroy(isc__workthread_t **threadp) { + REQUIRE(threadp != NULL && VALID_WORKTHREAD(*threadp)); + isc__workthread_t *thread = MOVE_OWNERSHIP(*threadp); + + isc_thread_join(thread->thread, NULL); + + INSIST(cds_wfcq_empty(&thread->qhead, &thread->qtail)); + + thread->magic = 0; + isc_mem_put(thread->loop->mctx, thread, sizeof(*thread)); +} + +void +isc__workthread_pause(isc__workthread_t *thread) { + REQUIRE(VALID_WORKTHREAD(thread)); + + /* + * Request a pause, but only if not already shutting down — a + * shutting-down worker heads for the stopping barrier and must never + * be waited on here (that'd be a deadlock). Clearing PAUSED as we set + * PAUSE invalidates any ack left over from the previous generation, so + * the wait below can only succeed on an ack for this request. + */ + int32_t old; + do { + old = uatomic_load(&thread->state, CMM_RELAXED); + if ((old & THREAD_SHUTDOWN) != 0) { + return; + } + } while (uatomic_cmpxchg(&thread->state, old, + (old | THREAD_PAUSE) & ~THREAD_PAUSED) != old); + + workthread_wake(thread); + + /* + * Wait for the worker to acknowledge (PAUSED, form workthread_thread() + * calling workthread_pause()) or for shutdown. + */ + while (true) { + old = uatomic_load(&thread->state, CMM_ACQUIRE); + if ((old & (THREAD_PAUSED | THREAD_SHUTDOWN)) != 0) { + return; + } + (void)futex_noasync(&thread->state, FUTEX_WAIT, old, NULL, NULL, + 0); + } +} + +void +isc__workthread_resume(isc__workthread_t *thread) { + REQUIRE(VALID_WORKTHREAD(thread)); + + /* Clear the request and wake the paused worker. */ + uatomic_and(&thread->state, ~THREAD_PAUSE); + (void)futex_noasync(&thread->state, FUTEX_WAKE, INT_MAX, NULL, NULL, 0); } diff --git a/tests/isc/work_test.c b/tests/isc/work_test.c index 5e0184cd52..6eec31f900 100644 --- a/tests/isc/work_test.c +++ b/tests/isc/work_test.c @@ -48,16 +48,15 @@ work_cb(void *arg) { } static void -after_work_cb(void *arg) { - UNUSED(arg); - +after_work_cb(void *arg ISC_ATTR_UNUSED, isc_result_t result ISC_ATTR_UNUSED) { assert_int_equal(atomic_load(&scheduled), 1); isc_loopmgr_shutdown(loopmgr); } static void work_enqueue_cb(void *arg ISC_ATTR_UNUSED) { - isc_work_enqueue(isc_loop(), work_cb, after_work_cb, NULL); + isc_work_enqueue(isc_loop(), ISC_WORKLANE_FAST, work_cb, after_work_cb, + NULL); } ISC_RUN_TEST_IMPL(isc_work_enqueue) {