Compare commits

...
Author SHA1 Message Date
Witold Kręcicki 5b11668ad8 xxx tmp 2020-02-10 11:35:42 +01:00
Witold Kręcicki f689db69db Revert "Test: initial LOWAC capacity at 100k"
This reverts commit 0e52562e4eea06a4c1764ece09bc1bffcae51d32.
2018-11-24 22:38:09 +00:00
Witold Kręcicki 7b276d4062 Test: initial LOWAC capacity at 100k 2018-11-24 22:38:09 +00:00
Witold Kręcicki c5949ff9d1 Reset affinity of LOWAC worker 2018-11-24 22:38:09 +00:00
Witold Kręcicki 57ab5d3a14 Check if we have a view before LOWAC-caching 2018-11-24 22:38:09 +00:00
Witold Kręcicki 496923f71f Use memcpy instead of byte-by-byte in dns_name_fromwire 2018-11-24 22:38:09 +00:00
Witold Kręcicki 32472059d3 Minor fixes 2018-11-24 22:38:09 +00:00
Witold Kręcicki 0dcdbdfbdd Don't overuse isc_time_now(), use local time 2018-11-24 22:38:09 +00:00
Witold Kręcicki 3b6985b783 Fix atomic_ macros 2018-11-24 22:38:09 +00:00
Witold Kręcicki 96ba0486f3 Make stats inc/dec inlined macros 2018-11-24 22:38:09 +00:00
Witold Kręcicki 7132ffffad Revert "Use memory_order_acq_rel"
This reverts commit 673792d111965cf93c79a39005024b55ad4d8ba8.
2018-11-24 22:38:09 +00:00
Witold Kręcicki 48853a75ff WiP 2018-11-24 22:38:09 +00:00
Witold Kręcicki df0c5bc4cf Allocate memory for blob: 2018-11-24 22:38:09 +00:00
Witold Kręcicki cd6f768a16 Longer cache time 2018-11-24 22:38:09 +00:00
Witold Kręcicki 7a6b2ffddd Always enqueue removal after replacement 2018-11-24 22:38:09 +00:00
Witold Kręcicki 758045a2fd Fixes 2018-11-24 22:38:09 +00:00
Witold Kręcicki 0e037f8873 Mark unused memory 2018-11-24 22:38:09 +00:00
Witold Kręcicki 7c68a70bf9 Use memory_order_acq_rel 2018-11-24 22:38:09 +00:00
Witold Kręcicki f9d4a96021 Add some magic checks 2018-11-24 22:38:09 +00:00
Witold Kręcicki 9b5bdf4e89 Fixes 2018-11-24 22:38:09 +00:00
Witold Kręcicki 8c9ae61062 Revert "Use common isc_time"
This reverts commit 1c5ba6ba8af5fb7e65456ef8b846afdc874b77a2.
2018-11-24 22:38:09 +00:00
Witold Kręcicki f7390b39f8 Iterator fix 2018-11-24 22:38:09 +00:00
Witold Kręcicki 14ec374173 Use common isc_time 2018-11-24 22:38:09 +00:00
Witold Kręcicki 1fec6d70e6 Cleanup better 2018-11-24 22:38:09 +00:00
Witold Kręcicki 96b76abece More fiddling 2018-11-24 22:38:09 +00:00
Witold Kręcicki 0a73f7bb27 Some fiddling 2018-11-24 22:38:09 +00:00
Witold Kręcicki aaaa727cb3 Cleaning up 2018-11-24 22:38:09 +00:00
Witold Kręcicki 55a435f091 Cleaning up 2018-11-24 22:38:09 +00:00
Witold Kręcicki 609e505bc1 Test: ck_fifo 2018-11-24 22:38:09 +00:00
Witold Kręcicki 8ffd188a26 PoC LOWAC 2018-11-24 22:38:09 +00:00
Witold Kręcicki 965ef669bc Get rid of tasks-ready and tasks-running 2018-11-24 22:32:51 +00:00
Witold Kręcicki c42f920449 Set quantum in dispatch 2018-11-24 22:32:40 +00:00
Witold Kręcicki 95037b96a3 Add tasks_waiting field in taskqueue 2018-11-24 22:32:36 +00:00
Witold Kręcicki 09d0f57296 Remove configurable default taskmgr quantum, it was always default anyway 2018-11-24 22:32:31 +00:00
Witold Kręcicki 5098bd7776 pseudosend 2018-11-24 09:39:42 +00:00
46 changed files with 803 additions and 178 deletions
+1 -1
View File
@@ -1623,7 +1623,7 @@ main(int argc, char *argv[]) {
fatal("failed to create mctx");
CHECK(isc_appctx_create(mctx, &actx));
CHECK(isc_taskmgr_createinctx(mctx, actx, 1, 0, &taskmgr));
CHECK(isc_taskmgr_createinctx(mctx, actx, 1, &taskmgr));
CHECK(isc_socketmgr_createinctx(mctx, actx, &socketmgr));
CHECK(isc_timermgr_createinctx(mctx, actx, &timermgr));
+1 -1
View File
@@ -1363,7 +1363,7 @@ setup_libs(void) {
isc_log_setdebuglevel(lctx, 0);
result = isc_taskmgr_create(mctx, 1, 0, &taskmgr);
result = isc_taskmgr_create(mctx, 1, &taskmgr);
check_result(result, "isc_taskmgr_create");
result = isc_task_create(taskmgr, 0, &global_task);
+1 -1
View File
@@ -3853,7 +3853,7 @@ main(int argc, char *argv[]) {
print_time(outfp);
print_version(outfp);
result = isc_taskmgr_create(mctx, ntasks, 0, &taskmgr);
result = isc_taskmgr_create(mctx, ntasks, &taskmgr);
if (result != ISC_R_SUCCESS)
fatal("failed to create task manager: %s",
isc_result_totext(result));
+1 -1
View File
@@ -803,7 +803,7 @@ create_managers(void) {
"using %u UDP listener%s per interface",
named_g_udpdisp, named_g_udpdisp == 1 ? "" : "s");
result = isc_taskmgr_create(named_g_mctx, named_g_cpus, 0,
result = isc_taskmgr_create(named_g_mctx, named_g_cpus,
&named_g_taskmgr);
if (result != ISC_R_SUCCESS) {
UNEXPECTED_ERROR(__FILE__, __LINE__,
+1 -1
View File
@@ -885,7 +885,7 @@ setup_system(void) {
result = isc_timermgr_create(gmctx, &timermgr);
check_result(result, "dns_timermgr_create");
result = isc_taskmgr_create(gmctx, 1, 0, &taskmgr);
result = isc_taskmgr_create(gmctx, 1, &taskmgr);
check_result(result, "isc_taskmgr_create");
result = isc_task_create(taskmgr, 0, &global_task);
+1 -1
View File
@@ -934,7 +934,7 @@ main(int argc, char **argv) {
DO("create memory context", isc_mem_create(0, 0, &rndc_mctx));
DO("create socket manager", isc_socketmgr_create(rndc_mctx, &socketmgr));
DO("create task manager", isc_taskmgr_create(rndc_mctx, 1, 0, &taskmgr));
DO("create task manager", isc_taskmgr_create(rndc_mctx, 1, &taskmgr));
DO("create task", isc_task_create(taskmgr, 0, &task));
DO("create logging context", isc_log_create(rndc_mctx, &log, &logconfig));
+1 -1
View File
@@ -146,7 +146,7 @@ create_managers(void) {
isc_result_t result;
taskmgr = NULL;
result = isc_taskmgr_create(mctx, 5, 0, &taskmgr);
result = isc_taskmgr_create(mctx, 5, &taskmgr);
check_result(result, "isc_taskmgr_create");
timermgr = NULL;
+1 -1
View File
@@ -114,7 +114,7 @@ main(int argc, char *argv[]) {
}
taskmgr = NULL;
RUNTIME_CHECK(isc_taskmgr_create(mctx, workers, 0, &taskmgr)
RUNTIME_CHECK(isc_taskmgr_create(mctx, workers, &taskmgr)
== ISC_R_SUCCESS);
task = NULL;
RUNTIME_CHECK(isc_task_create(taskmgr, 0, &task)
+1 -1
View File
@@ -228,7 +228,7 @@ main(int argc, char *argv[]) {
}
taskmgr = NULL;
RUNTIME_CHECK(isc_taskmgr_create(mctx, workers, 0, &taskmgr) ==
RUNTIME_CHECK(isc_taskmgr_create(mctx, workers, &taskmgr) ==
ISC_R_SUCCESS);
task = NULL;
RUNTIME_CHECK(isc_task_create(taskmgr, 0, &task) ==
+1 -1
View File
@@ -472,7 +472,7 @@ main(int argc, char *argv[]) {
RUNCHECK(dst_lib_init(mctx, NULL));
taskmgr = NULL;
RUNCHECK(isc_taskmgr_create(mctx, 1, 0, &taskmgr));
RUNCHECK(isc_taskmgr_create(mctx, 1, &taskmgr));
task = NULL;
RUNCHECK(isc_task_create(taskmgr, 0, &task));
timermgr = NULL;
+1 -2
View File
@@ -104,8 +104,7 @@ main(int argc, char *argv[]) {
isc_interval_set(&linterval, 1, 0);
RUNTIME_CHECK(isc_mem_create(0, 0, &mctx) == ISC_R_SUCCESS);
RUNTIME_CHECK(isc_taskmgr_create(mctx, 3, 0, &taskmgr) ==
ISC_R_SUCCESS);
RUNTIME_CHECK(isc_taskmgr_create(mctx, 3, &taskmgr) == ISC_R_SUCCESS);
RUNTIME_CHECK(isc_timermgr_create(mctx, &timermgr) ==
ISC_R_SUCCESS);
RUNTIME_CHECK(isc_task_create(taskmgr, 0, &g_task) ==
+1 -1
View File
@@ -183,7 +183,7 @@ main(int argc, char *argv[]) {
RUNTIME_CHECK(isc_mem_create(0, 0, &mctx) == ISC_R_SUCCESS);
mctx2 = NULL;
RUNTIME_CHECK(isc_mem_create(0, 0, &mctx2) == ISC_R_SUCCESS);
RUNTIME_CHECK(isc_taskmgr_create(mctx, workers, 0, &task_manager) ==
RUNTIME_CHECK(isc_taskmgr_create(mctx, workers, &task_manager) ==
ISC_R_SUCCESS);
RUNTIME_CHECK(isc_timermgr_create(mctx, &timer_manager) ==
ISC_R_SUCCESS);
+1 -2
View File
@@ -228,8 +228,7 @@ main(int argc, char *argv[]) {
dst_result_register();
taskmgr = NULL;
RUNTIME_CHECK(isc_taskmgr_create(mctx, 2, 0, &taskmgr) ==
ISC_R_SUCCESS);
RUNTIME_CHECK(isc_taskmgr_create(mctx, 2, &taskmgr) == ISC_R_SUCCESS);
task1 = NULL;
RUNTIME_CHECK(isc_task_create(taskmgr, 0, &task1) == ISC_R_SUCCESS);
+1 -1
View File
@@ -297,7 +297,7 @@ main(int argc, char *argv[]) {
* The task manager is independent (other than memory context)
*/
manager = NULL;
RUNTIME_CHECK(isc_taskmgr_create(mctx, workers, 0, &manager) ==
RUNTIME_CHECK(isc_taskmgr_create(mctx, workers, &manager) ==
ISC_R_SUCCESS);
/*
+1 -1
View File
@@ -81,7 +81,7 @@ main(int argc, char *argv[]) {
RUNTIME_CHECK(isc_mem_create(0, 0, &mctx) == ISC_R_SUCCESS);
RUNTIME_CHECK(isc_taskmgr_create(mctx, workers, 0, &manager) ==
RUNTIME_CHECK(isc_taskmgr_create(mctx, workers, &manager) ==
ISC_R_SUCCESS);
RUNTIME_CHECK(isc_task_create(manager, 0, &t1) == ISC_R_SUCCESS);
+1 -1
View File
@@ -110,7 +110,7 @@ main(int argc, char *argv[]) {
printf("%u workers\n", workers);
RUNTIME_CHECK(isc_mem_create(0, 0, &mctx1) == ISC_R_SUCCESS);
RUNTIME_CHECK(isc_taskmgr_create(mctx1, workers, 0, &manager) ==
RUNTIME_CHECK(isc_taskmgr_create(mctx1, workers, &manager) ==
ISC_R_SUCCESS);
RUNTIME_CHECK(isc_timermgr_create(mctx1, &timgr) == ISC_R_SUCCESS);
+1 -1
View File
@@ -284,7 +284,7 @@ main(int argc, char **argv) {
RUNTIME_CHECK(isc_app_start() == ISC_R_SUCCESS);
RUNTIME_CHECK(isc_mem_create(0, 0, &mctx) == ISC_R_SUCCESS);
RUNTIME_CHECK(isc_taskmgr_create(mctx, 2, 0, &taskmgr) ==
RUNTIME_CHECK(isc_taskmgr_create(mctx, 2, &taskmgr) ==
ISC_R_SUCCESS);
RUNTIME_CHECK(isc_timermgr_create(mctx, &timermgr) == ISC_R_SUCCESS);
RUNTIME_CHECK(isc_socketmgr_create(mctx, &socketmgr) == ISC_R_SUCCESS);
+1 -1
View File
@@ -279,7 +279,7 @@ main(int argc, char *argv[]) {
RUNCHECK(dst_lib_init(mctx, NULL));
taskmgr = NULL;
RUNCHECK(isc_taskmgr_create(mctx, 1, 0, &taskmgr));
RUNCHECK(isc_taskmgr_create(mctx, 1, &taskmgr));
task = NULL;
RUNCHECK(isc_task_create(taskmgr, 0, &task));
timermgr = NULL;
+1 -1
View File
@@ -239,7 +239,7 @@ main(int argc, char *argv[]) {
RUNCHECK(dst_lib_init(mctx, NULL));
taskmgr = NULL;
RUNCHECK(isc_taskmgr_create(mctx, 1, 0, &taskmgr));
RUNCHECK(isc_taskmgr_create(mctx, 1, &taskmgr));
task = NULL;
RUNCHECK(isc_task_create(taskmgr, 0, &task));
timermgr = NULL;
+1 -1
View File
@@ -178,7 +178,7 @@ main(int argc, char **argv) {
RUNCHECK(dst_lib_init(mctx, NULL));
taskmgr = NULL;
RUNCHECK(isc_taskmgr_create(mctx, 1, 0, &taskmgr));
RUNCHECK(isc_taskmgr_create(mctx, 1, &taskmgr));
task = NULL;
RUNCHECK(isc_task_create(taskmgr, 0, &task));
timermgr = NULL;
+1 -1
View File
@@ -1981,7 +1981,7 @@ main(int argc, char *argv[]) {
fatal("can't choose between IPv4 and IPv6");
taskmgr = NULL;
RUNCHECK(isc_taskmgr_create(mctx, 1, 0, &taskmgr));
RUNCHECK(isc_taskmgr_create(mctx, 1, &taskmgr));
task = NULL;
RUNCHECK(isc_task_create(taskmgr, 0, &task));
timermgr = NULL;
Vendored
+2 -2
View File
@@ -15115,8 +15115,8 @@ ac_compiler_gnu=$ac_cv_c_compiler_gnu
LIBS="$PTHREAD_LIBS $LIBS"
CFLAGS="$CFLAGS $PTHREAD_CFLAGS"
LIBS="$PTHREAD_LIBS -lck $LIBS"
CFLAGS="$CFLAGS $PTHREAD_CFLAGS -I/usr/include/ck"
CC="$PTHREAD_CC"
#
+2 -2
View File
@@ -657,8 +657,8 @@ AC_CHECK_FUNCS(arc4random arc4random_buf arc4random_uniform getrandom)
AX_PTHREAD
LIBS="$PTHREAD_LIBS $LIBS"
CFLAGS="$CFLAGS $PTHREAD_CFLAGS"
LIBS="$PTHREAD_LIBS -lck $LIBS"
CFLAGS="$CFLAGS $PTHREAD_CFLAGS -I/usr/include/ck"
CC="$PTHREAD_CC"
#
+3 -3
View File
@@ -37,7 +37,7 @@ ISCLIBS = ../../lib/isc/libisc.@A@ @OPENSSL_LIBS@
ISCDEPLIBS = ../../lib/isc/libisc.@A@
LIBS = @LIBS@
LIBS = @LIBS@ -lck
# Alphabetically
@@ -73,7 +73,7 @@ DNSOBJS = acl.@O@ adb.@O@ badcache.@O@ byaddr.@O@ \
stats.@O@ tcpmsg.@O@ time.@O@ timer.@O@ tkey.@O@ \
tsec.@O@ tsig.@O@ ttl.@O@ update.@O@ validator.@O@ \
version.@O@ view.@O@ xfrin.@O@ zone.@O@ zonekey.@O@ \
zoneverify.@O@ zt.@O@
zoneverify.@O@ zt.@O@ lowac.@O@
PORTDNSOBJS = client.@O@ ecdb.@O@
OBJS= @DNSTAPOBJS@ ${DNSOBJS} ${OTHEROBJS} ${DSTOBJS} \
@@ -108,7 +108,7 @@ DNSSRCS = acl.c adb.c badcache. byaddr.c \
stats.c tcpmsg.c time.c timer.c tkey.c \
tsec.c tsig.c ttl.c update.c validator.c \
version.c view.c xfrin.c zone.c zoneverify.c \
zonekey.c zt.c ${OTHERSRCS}
zonekey.c zt.c ${OTHERSRCS} lowac.c
PORTDNSSRCS = client.c ecdb.c
SRCS = ${DSTSRCS} ${DNSSRCS} ${PORTDNSSRCS} @DNSTAPSRCS@ @GEOIPLINKSRCS@
+1 -1
View File
@@ -379,7 +379,7 @@ dns_client_create(dns_client_t **clientp, unsigned int options) {
result = isc_app_ctxstart(actx);
if (result != ISC_R_SUCCESS)
goto cleanup;
result = isc_taskmgr_createinctx(mctx, actx, 1, 0, &taskmgr);
result = isc_taskmgr_createinctx(mctx, actx, 1, &taskmgr);
if (result != ISC_R_SUCCESS)
goto cleanup;
result = isc_socketmgr_createinctx(mctx, actx, &socketmgr);
+17
View File
@@ -0,0 +1,17 @@
#include <isc/mem.h>
#include <dns/name.h>
typedef struct dns_lowac_entry dns_lowac_entry_t;
typedef struct dns_lowac dns_lowac_t;
dns_lowac_t*
dns_lowac_create(isc_mem_t *mctx);
void
dns_lowac_destroy(dns_lowac_t *lowac);
isc_result_t
dns_lowac_put(dns_lowac_t *lowac, dns_name_t *name, char* packet, int size);
isc_result_t
dns_lowac_get(dns_lowac_t *lowac, dns_name_t *name, unsigned char* blob, int* blobsize, bool tcp);
+3
View File
@@ -75,6 +75,7 @@
#include <dns/rpz.h>
#include <dns/types.h>
#include <dns/zt.h>
#include <dns/lowac.h>
ISC_LANG_BEGINDECLS
@@ -238,6 +239,8 @@ struct dns_view {
dns_dtenv_t *dtenv; /* Dnstap environment */
dns_dtmsgtype_t dttypes; /* Dnstap message types
to log */
dns_lowac_t* lowac;
};
#define DNS_VIEW_MAGIC ISC_MAGIC('V','i','e','w')
+490
View File
@@ -0,0 +1,490 @@
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <isc/thread.h>
#include <isc/mem.h>
#include <isc/hash.h>
#include <isc/time.h>
#include <isc/util.h>
#include <isc/random.h>
#include <isc/refcount.h>
#include <dns/name.h>
#include <dns/lowac.h>
#include <stdlib.h>
#include <ck_ht.h>
#include <ck_fifo.h>
#include <ck_queue.h>
#define fprintf(...) {}
#define MAXIMUM_PKT_SIZE 1024
#define LENTRY_MAGIC ISC_MAGIC('L', 'o', 'w', 'E')
#define VALID_LENTRY(c) ISC_MAGIC_VALID(c, LENTRY_MAGIC)
struct dns_lowac_entry {
unsigned int magic;
dns_name_t name;
isc_region_t key;
uint16_t flags;
int64_t expire;
ck_ht_hash_t hash;
unsigned char *blob;
uint32_t blobsize;
isc_refcount_t refcount;
/*
* Set to 'true' when we enqueue this entry on remq.
* Nothing serious will happen if we enqueue the entry
* on remq multiple times as we're refcounting.
* It's not locked in any way as it can only change
* from false to true during runtime.
*/
bool remq_enqueued;
bool inht;
isc_refcount_t rqrefcount;
atomic_int_fast64_t lastusage;
};
struct dns_lowac {
isc_mem_t * mctx;
isc_thread_t thread;
int expiry; /* Time after which records expire */
bool running; /* We're not shut down */
bool accepting; /* We're accepting packets into cache */
ck_ht_t ht; /* Cache hash table */
ck_ht_iterator_t htit; /* Hashtable maintenance iterator */
ck_fifo_mpmc_t inq; /* Packet input queue */
ck_fifo_mpmc_t remq; /* Removal queue */
atomic_int_fast64_t now; /* Epoch time, updated at every loop */
int count;
};
static void *
rthread(void*d);
static void *
ht_malloc(size_t r)
{
return(malloc(r));
}
static void
ht_free(void *p, size_t b, bool r)
{
(void)b;
(void)r;
free(p);
return;
}
static void
ht_hash_wrapper(struct ck_ht_hash *h, const void *key, size_t length,
uint64_t seed)
{
(void)seed;
h->value = isc_hash_function(key, length, false, NULL);
return;
}
static struct ck_malloc my_allocator = {
.malloc = ht_malloc,
.free = ht_free
};
static void
free_entry(dns_lowac_t *lowac, dns_lowac_entry_t *entry) {
entry->magic = 0xffffffff;
dns_name_free(&entry->name, lowac->mctx);
isc_mem_put(lowac->mctx, entry->blob, entry->blobsize);
isc_mem_put(lowac->mctx, entry, sizeof(*entry));
}
static bool
dequeue_input_entry(dns_lowac_t*lowac) {
dns_lowac_entry_t *entry = NULL;
ck_fifo_mpmc_entry_t *garbage = NULL;
if (ck_fifo_mpmc_dequeue(&lowac->inq, &entry,
&garbage) == true) {
REQUIRE(VALID_LENTRY(entry));
isc_mem_put(lowac->mctx, garbage,
sizeof(*garbage));
ck_ht_entry_t htentry;
ck_ht_entry_t oldhtentry;
ck_ht_entry_set(&htentry, entry->hash, entry->key.base,
entry->key.length, entry);
ck_ht_entry_set(&oldhtentry, entry->hash, entry->key.base,
entry->key.length, NULL);
if (ck_ht_get_spmc(&lowac->ht, entry->hash,
&oldhtentry) == false) {
/* We don't have this entry in hashtable */
entry->inht = true;
if (ck_ht_put_spmc(&lowac->ht, entry->hash,
&htentry) == false) {
abort();
printf("oddness but might happen\n");
/* We can do it safely as it never was in
* hashtable */
RUNTIME_CHECK(isc_refcount_decrement(&entry->
refcount) ==
1);
free_entry(lowac, entry);
} else {
fprintf(stderr, "XXXInsert %p %d\n", entry, lowac->count);
lowac->count++;
}
} else {
dns_lowac_entry_t *oldentry =
ck_ht_entry_value(&oldhtentry);
/* Note: set without remove doesn't seem to work
* properly */
oldentry->inht = false;
RUNTIME_CHECK(ck_ht_remove_spmc(&lowac->ht,
entry->hash,
&oldhtentry) == true);
entry->inht = true;
RUNTIME_CHECK(ck_ht_set_spmc(&lowac->ht, entry->hash,
&htentry) == true);
lowac->count++;
/*
* We don't need to increment refcount since we
* have not decremented it
* when removing from hashtable
*/
ck_fifo_mpmc_entry_t *qentry =
isc_mem_get(lowac->mctx,
(sizeof(
ck_fifo_mpmc_entry_t)));
oldentry->remq_enqueued = true;
isc_refcount_increment(&oldentry->rqrefcount);
fprintf(stderr, "XXXenq %p %d\n", oldentry, __LINE__);
ck_fifo_mpmc_enqueue(&lowac->remq, qentry,
oldentry);
}
return (true);
}
return (false);
}
static int
expire_entries(dns_lowac_t *lowac) {
int iterated = 0;
ck_ht_entry_t *htitentry = NULL;
bool iterok = ck_ht_next(&lowac->ht, &lowac->htit, &htitentry);
if (!iterok) {
ck_ht_iterator_init(&lowac->htit);
iterok = ck_ht_next(&lowac->ht, &lowac->htit, &htitentry);
}
while (iterok && htitentry != NULL && iterated < 2048) {
dns_lowac_entry_t *entry = ck_ht_entry_value(htitentry);
REQUIRE(VALID_LENTRY(entry));
if (entry->expire < atomic_load(&lowac->now)) {
if (!ck_ht_remove_spmc(&lowac->ht, entry->hash, htitentry)) {
abort();
/*
* Very unlikely, but can happen when we're between generations.
* TODO verify that it's ok at all, maybe we're using the iterator wrong?
*/
isc_refcount_increment(&entry->refcount);
fprintf(stderr, "INCREF %d %p\n", __LINE__, entry);
} else {
dns_lowac_entry_t *ent2 = ck_ht_entry_value(htitentry);
RUNTIME_CHECK(ent2 == entry);
}
entry->inht = false;
/*
* We don't need to increment refcount since we have
* not decremented it when removing from hashtable.
*/
ck_fifo_mpmc_entry_t *qentry =
isc_mem_get(lowac->mctx,
(sizeof(
ck_fifo_mpmc_entry_t)));
entry->remq_enqueued = true;
isc_refcount_increment(&entry->rqrefcount);
fprintf(stderr, "XXXenq %p %d\n", entry, __LINE__);
ck_fifo_mpmc_enqueue(&lowac->remq, qentry, entry);
}
htitentry = NULL;
iterok = ck_ht_next(&lowac->ht, &lowac->htit, &htitentry);
iterated++;
}
if (!iterok) {
ck_ht_iterator_init(&lowac->htit);
}
return (iterated);
}
/*
* Iterate over remq.
* If the entry is in HT - remove it, requeue.
* If the entry is not in HT but it is referenced - requeue.
* If the entry is not referenced - free.
*/
static int
cleanup_entries(dns_lowac_t *lowac) {
ck_ht_entry_t htentry;
dns_lowac_entry_t *entry;
ck_fifo_mpmc_entry_t *qentry = NULL;
int removed = 0;
int max = 100;
while (--max > 0 &&
ck_fifo_mpmc_dequeue(&lowac->remq, &entry, &qentry) == true)
{
REQUIRE(VALID_LENTRY(entry));
int rc = isc_refcount_decrement(&entry->refcount);
int rqrc = isc_refcount_decrement(&entry->rqrefcount);
fprintf(stderr, "DECREF %d %p %d\n", __LINE__, entry, lowac->count);
if (entry->inht) {
/*
* Remove entry from the hashtable, we'll remove the
* entry itself in the next iteration.
*/
ck_ht_entry_set(&htentry, entry->hash, entry->key.base,
entry->key.length, entry);
RUNTIME_CHECK(ck_ht_remove_spmc(&lowac->ht,
entry->hash,
&htentry) == true);
entry->inht = false;
isc_refcount_increment(&entry->rqrefcount);
/* We don't increment refcount - it was at 2 (remq + ht), it stays at 1 (remq) */
fprintf(stderr, "INCREF %d %p\n", __LINE__, entry);
fprintf(stderr, "XXXenq %p %d\n", entry, __LINE__);
ck_fifo_mpmc_enqueue(&lowac->remq, qentry, entry);
} else {
if (rc > 1) {
/* Requeue if still in use and there's no other request in queue */
/* TODO that's completely wrong, if we have two instances in the queue they will always have rc > 1 !!! */
if (rqrc == 0) {
int p1 = isc_refcount_increment(&entry->rqrefcount);
int p2 = isc_refcount_increment(&entry->refcount);
(void) p1;
(void) p2;
fprintf(stderr, "INCREF %d %p %d %d\n", __LINE__, entry, p1, p2);
fprintf(stderr, "XXXenq %p %d\n", entry, __LINE__);
ck_fifo_mpmc_enqueue(&lowac->remq, qentry,
entry);
} else {
isc_mem_put(lowac->mctx, qentry,
sizeof(*qentry));
}
fprintf(stderr, "IGNREF %d %p\n", __LINE__, entry);
} else {
isc_mem_put(lowac->mctx, qentry,
sizeof(*qentry));
removed++;
free_entry(lowac, entry);
fprintf(stderr, "XXXFree %p %d\n", entry, lowac->count);
lowac->count--;
}
}
}
return (removed);
}
static void *
rthread(void *d) {
dns_lowac_t *lowac = (dns_lowac_t*) d;
isc_thread_resetaffinity();
ck_ht_iterator_init(&lowac->htit);
while (lowac->running) {
isc_time_t now;
isc_time_now(&now);
atomic_store(&lowac->now, now.seconds);
bool dequeued = dequeue_input_entry(lowac);
if (!dequeued) {
expire_entries(lowac);
int removed = cleanup_entries(lowac);
if (removed < 256) {
usleep(1000);
} else {
RUNTIME_CHECK(ck_ht_gc(&lowac->ht, 128,
isc_random32()));
}
}
}
return (NULL);
}
dns_lowac_t*
dns_lowac_create(isc_mem_t *mctx) {
dns_lowac_t *lowac = isc_mem_get(mctx, sizeof(dns_lowac_t));
lowac->expiry = 600;
lowac->count = 0;
lowac->running = true;
if (ck_ht_init(&lowac->ht,
CK_HT_MODE_BYTESTRING | CK_HT_WORKLOAD_DELETE,
ht_hash_wrapper,
&my_allocator, 32, isc_random32()) == false) {
abort();
}
ck_fifo_mpmc_entry_t *inq_stub =
isc_mem_get(mctx, sizeof(ck_fifo_mpmc_entry_t));
ck_fifo_mpmc_entry_t *remq_stub =
isc_mem_get(mctx, sizeof(ck_fifo_mpmc_entry_t));
ck_fifo_mpmc_init(&lowac->inq, inq_stub);
ck_fifo_mpmc_init(&lowac->remq, remq_stub);
lowac->mctx = NULL;
isc_mem_attach(mctx, &lowac->mctx);
isc_thread_create(rthread, lowac, &lowac->thread);
isc_thread_setname(lowac->thread, "lowac-worker");
return (lowac);
}
void
dns_lowac_destroy(dns_lowac_t *lowac) {
lowac->running = false;
isc_thread_join(lowac->thread, NULL);
dns_lowac_entry_t *entry;
ck_fifo_mpmc_entry_t *fentry;
/* Entries in inq can only be in inq, it's safe to free them */
while (ck_fifo_mpmc_dequeue(&lowac->inq,
&entry,
&fentry))
{
isc_mem_put(lowac->mctx, fentry, sizeof(*fentry));
free_entry(lowac, entry);
}
/*
* Entries in remq can be freed only if they're no longer referenced
* by either multiple occurences in queue or HT
*/
while (ck_fifo_mpmc_dequeue(&lowac->remq,
&entry,
&fentry))
{
if (isc_refcount_decrement(&entry->refcount) == 1) {
fprintf(stderr, "DECREF %d %p\n", __LINE__, entry);
free_entry(lowac, entry);
} else {
fprintf(stderr, "Non-unreferenced entry in remq %p\n", entry);
}
isc_mem_put(lowac->mctx, fentry, sizeof(*fentry));
}
/* Finally we free rest of entries from HT */
while (ck_ht_count(&lowac->ht) > 0) {
ck_ht_gc(&lowac->ht, 0, 0);
ck_ht_iterator_t htit = CK_HT_ITERATOR_INITIALIZER;
ck_ht_entry_t *htitentry = NULL;
while (ck_ht_next(&lowac->ht, &htit, &htitentry)) {
entry = ck_ht_entry_value(htitentry);
ck_ht_remove_spmc(&lowac->ht, entry->hash, htitentry);
free_entry(lowac, entry);
}
}
ck_fifo_mpmc_deinit(&lowac->inq, &fentry);
isc_mem_put(lowac->mctx, fentry, sizeof(*fentry));
ck_fifo_mpmc_deinit(&lowac->remq, &fentry);
isc_mem_put(lowac->mctx, fentry, sizeof(*fentry));
ck_ht_destroy(&lowac->ht);
isc_mem_putanddetach(&lowac->mctx, lowac, sizeof(*lowac));
}
isc_result_t
dns_lowac_put(dns_lowac_t *lowac, dns_name_t *name, char*packet, int size) {
if (!lowac->running) {
return (ISC_R_SHUTTINGDOWN);
}
if (size > MAXIMUM_PKT_SIZE) {
return (ISC_R_FAILURE);
}
dns_lowac_entry_t *entry = isc_mem_get(lowac->mctx, sizeof(*entry));
dns_name_init(&entry->name, NULL);
dns_name_dup(name, lowac->mctx, &entry->name);
entry->blob = isc_mem_get(lowac->mctx, size);
memcpy(entry->blob, packet, size);
entry->blobsize = size;
isc_refcount_init(&entry->refcount, 1);
isc_refcount_init(&entry->rqrefcount, 0);
entry->expire = atomic_load(&lowac->now) + lowac->expiry;
entry->remq_enqueued = false;
entry->inht = false;
entry->magic = LENTRY_MAGIC;
dns_name_toregion(&entry->name, &entry->key);
ck_ht_hash(&entry->hash, &lowac->ht, entry->key.base,
entry->key.length);
ck_fifo_mpmc_entry_t *qentry =
isc_mem_get(lowac->mctx, (sizeof(ck_fifo_mpmc_entry_t)));
ck_fifo_mpmc_enqueue(&lowac->inq, qentry, entry);
return (ISC_R_SUCCESS);
}
isc_result_t
dns_lowac_get(dns_lowac_t *lowac, dns_name_t *name, unsigned char *blob,
int *blobsize, bool tcp) {
ck_ht_entry_t htentry;
ck_ht_hash_t h;
isc_region_t r;
dns_name_toregion(name, &r);
ck_ht_hash(&h, &lowac->ht, r.base, r.length);
ck_ht_entry_key_set(&htentry, r.base, r.length);
if (ck_ht_get_spmc(&lowac->ht, h, &htentry) == false) {
return (ISC_R_NOTFOUND);
} else {
dns_lowac_entry_t *entry = ck_ht_entry_value(&htentry);
REQUIRE(VALID_LENTRY(entry));
int oldrc = isc_refcount_increment(&entry->refcount);
fprintf(stderr, "INCREF %d %p\n", __LINE__, entry, oldrc);
RUNTIME_CHECK(oldrc > 0);
if (entry->remq_enqueued || !entry->inht) {
/* This entry is being removed, bail */
isc_refcount_decrement(&entry->refcount);
fprintf(stderr, "DECREF %d %p\n", __LINE__, entry);
return (ISC_R_NOTFOUND);
}
if (entry->expire < atomic_load(&lowac->now)) {
ck_fifo_mpmc_entry_t *qentry =
isc_mem_get(lowac->mctx,
(sizeof(ck_fifo_mpmc_entry_t)));
entry->remq_enqueued = true;
isc_refcount_increment(&entry->rqrefcount);
fprintf(stderr, "XXXenq %p %d\n", entry, __LINE__);
ck_fifo_mpmc_enqueue(&lowac->remq, qentry, entry);
return (ISC_R_NOTFOUND);
}
if (tcp) {
blob[0] = entry->blobsize >> 8;
blob[1] = entry->blobsize;
memcpy(blob + 2, entry->blob, entry->blobsize);
*blobsize = entry->blobsize + 2;
} else {
memcpy(blob, entry->blob, entry->blobsize);
*blobsize = entry->blobsize;
}
isc_refcount_decrement(&entry->refcount);
fprintf(stderr, "DECREF %d %p\n", __LINE__, entry);
return (ISC_R_SUCCESS);
}
return (ISC_R_FAILURE);
}
+15 -1
View File
@@ -1835,7 +1835,21 @@ dns_name_fromwire(dns_name_t *name, isc_buffer_t *source,
if (c == 0)
done = true;
n = c;
state = fw_ordinary;
if (downcase) {
state = fw_ordinary;
} else {
current += n;
if (current >= source->active) {
return (ISC_R_UNEXPECTEDEND);
}
memcpy(ndata, cdata, n);
cdata += n;
ndata += n;
if (!seen_pointer) {
cused+=n;
}
state = fw_start;
}
} else if (c >= 128 && c < 192) {
/*
* 14 bit local compression pointer.
+1 -1
View File
@@ -109,7 +109,7 @@ create_managers(void) {
isc_result_t result;
ncpus = isc_os_ncpus();
CHECK(isc_taskmgr_create(mctx, ncpus, 0, &taskmgr));
CHECK(isc_taskmgr_create(mctx, ncpus, &taskmgr));
CHECK(isc_timermgr_create(mctx, &timermgr));
CHECK(isc_socketmgr_create(mctx, &socketmgr));
CHECK(isc_task_create(taskmgr, 0, &maintask));
+5
View File
@@ -288,6 +288,8 @@ dns_view_create(isc_mem_t *mctx, dns_rdataclass_t rdclass,
view, NULL, NULL, NULL);
view->viewlist = NULL;
view->magic = DNS_VIEW_MAGIC;
/* TODO we might destroy the view (and LOWAC) when we still need it on shutdown! */
view->lowac = dns_lowac_create(view->mctx);
*viewp = view;
@@ -627,6 +629,9 @@ view_flushanddetach(dns_view_t **viewp, bool flush) {
if (view->catzs != NULL) {
dns_catz_catzs_detach(&view->catzs);
}
if (view->lowac != NULL) {
dns_lowac_destroy(view->lowac);
}
done = all_done(view);
UNLOCK(&view->lock);
+1 -1
View File
@@ -104,7 +104,7 @@ ctxs_init(isc_mem_t **mctxp, isc_appctx_t **actxp,
if (result != ISC_R_SUCCESS)
goto fail;
result = isc_taskmgr_createinctx(*mctxp, *actxp, 1, 0, taskmgrp);
result = isc_taskmgr_createinctx(*mctxp, *actxp, 1, taskmgrp);
if (result != ISC_R_SUCCESS)
goto fail;
+44 -4
View File
@@ -17,10 +17,47 @@
#include <inttypes.h>
#include <isc/atomic.h>
#include <isc/types.h>
#include <isc/magic.h>
#include <isc/util.h>
#include <isc/rwlock.h>
ISC_LANG_BEGINDECLS
typedef atomic_int_fast64_t isc_stat_t;
#define ISC_STATS_MAGIC ISC_MAGIC('S', 't', 'a', 't')
#define ISC_STATS_VALID(x) ISC_MAGIC_VALID(x, ISC_STATS_MAGIC)
struct isc_stats {
/*% Unlocked */
unsigned int magic;
isc_mem_t *mctx;
int ncounters;
isc_mutex_t lock;
unsigned int references; /* locked by lock */
/*%
* Locked by counterlock or unlocked if efficient rwlock is not
* available.
*/
isc_stat_t *counters;
/*%
* We don't want to lock the counters while we are dumping, so we first
* copy the current counter values into a local array. This buffer
* will be used as the copy destination. It's allocated on creation
* of the stats structure so that the dump operation won't fail due
* to memory allocation failure.
* XXX: this approach is weird for non-threaded build because the
* additional memory and the copy overhead could be avoided. We prefer
* simplicity here, however, under the assumption that this function
* should be only rarely called.
*/
uint64_t *copiedcounters;
};
/*%<
* Flag(s) for isc_stats_dump().
*/
@@ -78,8 +115,9 @@ isc_stats_ncounters(isc_stats_t *stats);
*
*/
void
isc_stats_increment(isc_stats_t *stats, isc_statscounter_t counter);
#define isc_stats_increment(stats, counter) do { \
atomic_fetch_add_explicit(&stats->counters[counter], 1, \
memory_order_relaxed); } while(0)
/*%<
* Increment the counter-th counter of stats.
*
@@ -90,8 +128,10 @@ isc_stats_increment(isc_stats_t *stats, isc_statscounter_t counter);
* on creation.
*/
void
isc_stats_decrement(isc_stats_t *stats, isc_statscounter_t counter);
#define isc_stats_decrement(stats, counter) do { \
atomic_fetch_sub_explicit(&stats->counters[counter], 1, \
memory_order_relaxed); } while(0)
/*%<
* Decrement the counter-th counter of stats.
*
+4 -4
View File
@@ -134,11 +134,11 @@ typedef uint_fast64_t atomic_uint_fast64_t;
#define atomic_load(obj) \
atomic_load_explicit(obj, memory_order_seq_cst)
#define atomic_store(obj) \
atomic_store_explicit(obj, memory_order_seq_cst)
#define atomic_fetch_add(obj) \
#define atomic_store(obj, arg) \
atomic_store_explicit(obj, arg, memory_order_seq_cst)
#define atomic_fetch_add(obj, arg) \
atomic_fetch_add_explicit(obj, arg, memory_order_seq_cst)
#define atomic_fetch_sub(obj) \
#define atomic_fetch_sub(obj, arg) \
atomic_fetch_sub_explicit(obj, arg, memory_order_seq_cst)
#define atomic_compare_exchange_strong(obj, expected, desired) \
atomic_compare_exchange_strong_explicit(obj, expected, desired, memory_order_seq_cst, memory_order_seq_cst)
+2 -7
View File
@@ -631,11 +631,10 @@ isc_task_privilege(isc_task_t *task);
isc_result_t
isc_taskmgr_createinctx(isc_mem_t *mctx, isc_appctx_t *actx,
unsigned int workers, unsigned int default_quantum,
isc_taskmgr_t **managerp);
unsigned int workers, isc_taskmgr_t **managerp);
isc_result_t
isc_taskmgr_create(isc_mem_t *mctx, unsigned int workers,
unsigned int default_quantum, isc_taskmgr_t **managerp);
isc_taskmgr_t **managerp);
/*%<
* Create a new task manager. isc_taskmgr_createinctx() also associates
* the new manager with the specified application context.
@@ -648,10 +647,6 @@ isc_taskmgr_create(isc_mem_t *mctx, unsigned int workers,
* create 'workers' threads, but if at least one thread creation
* succeeds, isc_taskmgr_create() may return ISC_R_SUCCESS.
*
*\li If 'default_quantum' is non-zero, then it will be used as the default
* quantum value when tasks are created. If zero, then an implementation
* defined default quantum will be used.
*
* Requires:
*
*\li 'mctx' is a valid memory context.
+4
View File
@@ -44,9 +44,13 @@ isc_thread_yield(void);
void
isc_thread_setname(isc_thread_t thread, const char *name);
isc_result_t
isc_thread_setaffinity(int cpu);
isc_result_t
isc_thread_resetaffinity(void);
/* XXX We could do fancier error handling... */
#define isc_thread_join(t, rp) \
+32
View File
@@ -111,6 +111,38 @@ isc_thread_yield(void) {
#endif
}
isc_result_t
isc_thread_resetaffinity() {
#if defined(HAVE_CPUSET_SETAFFINITY)
cpuset_t cpuset;
CPU_ZERO(&cpuset);
for (int cpu=0; cpu < 32; cpu++) {
CPU_SET(cpu, &cpuset);
}
if (cpuset_setaffinity(CPU_LEVEL_WHICH, CPU_WHICH_TID, -1,
&cpuset, sizeof(cpuset)) != 0) {
return (ISC_R_FAILURE);
}
#elif defined(HAVE_PTHREAD_SETAFFINITY_NP)
cpu_set_t set;
CPU_ZERO(&set);
for (int cpu=0; cpu < 32; cpu++) {
CPU_SET(cpu, &set);
}
if (pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t),
&set) != 0) {
return (ISC_R_FAILURE);
}
#elif defined(HAVE_PROCESSOR_BIND)
if (processor_bind(P_LWPID, P_MYID, PBIND_NONE, NULL) != 0) {
return (ISC_R_FAILURE);
}
#else
UNUSED(cpu);
#endif
return (ISC_R_SUCCESS);
}
isc_result_t
isc_thread_setaffinity(int cpu) {
#if defined(HAVE_CPUSET_SETAFFINITY)
-54
View File
@@ -19,47 +19,10 @@
#include <isc/atomic.h>
#include <isc/buffer.h>
#include <isc/magic.h>
#include <isc/mem.h>
#include <isc/platform.h>
#include <isc/print.h>
#include <isc/rwlock.h>
#include <isc/stats.h>
#include <isc/util.h>
#define ISC_STATS_MAGIC ISC_MAGIC('S', 't', 'a', 't')
#define ISC_STATS_VALID(x) ISC_MAGIC_VALID(x, ISC_STATS_MAGIC)
typedef atomic_int_fast64_t isc_stat_t;
struct isc_stats {
/*% Unlocked */
unsigned int magic;
isc_mem_t *mctx;
int ncounters;
isc_mutex_t lock;
unsigned int references; /* locked by lock */
/*%
* Locked by counterlock or unlocked if efficient rwlock is not
* available.
*/
isc_stat_t *counters;
/*%
* We don't want to lock the counters while we are dumping, so we first
* copy the current counter values into a local array. This buffer
* will be used as the copy destination. It's allocated on creation
* of the stats structure so that the dump operation won't fail due
* to memory allocation failure.
* XXX: this approach is weird for non-threaded build because the
* additional memory and the copy overhead could be avoided. We prefer
* simplicity here, however, under the assumption that this function
* should be only rarely called.
*/
uint64_t *copiedcounters;
};
static isc_result_t
create_stats(isc_mem_t *mctx, int ncounters, isc_stats_t **statsp) {
@@ -159,23 +122,6 @@ isc_stats_create(isc_mem_t *mctx, isc_stats_t **statsp, int ncounters) {
return (create_stats(mctx, ncounters, statsp));
}
void
isc_stats_increment(isc_stats_t *stats, isc_statscounter_t counter) {
REQUIRE(ISC_STATS_VALID(stats));
REQUIRE(counter < stats->ncounters);
atomic_fetch_add_explicit(&stats->counters[counter], 1,
memory_order_relaxed);
}
void
isc_stats_decrement(isc_stats_t *stats, isc_statscounter_t counter) {
REQUIRE(ISC_STATS_VALID(stats));
REQUIRE(counter < stats->ncounters);
atomic_fetch_sub_explicit(&stats->counters[counter], 1,
memory_order_relaxed);
}
void
isc_stats_dump(isc_stats_t *stats, isc_stats_dumper_t dump_fn,
+25 -64
View File
@@ -134,6 +134,8 @@ struct isc__taskqueue {
isc_condition_t work_available;
isc_thread_t thread;
unsigned int threadid;
unsigned int tasks_waiting;
bool running;
isc__taskmgr_t *manager;
};
@@ -145,13 +147,10 @@ struct isc__taskmgr {
isc_mutex_t halt_lock;
isc_condition_t halt_cond;
unsigned int workers;
atomic_uint_fast32_t tasks_running;
atomic_uint_fast32_t tasks_ready;
atomic_uint_fast32_t curq;
isc__taskqueue_t *queues;
/* Locked by task manager lock. */
unsigned int default_quantum;
LIST(isc__task_t) tasks;
isc_taskmgrmode_t mode;
bool pause_requested;
@@ -176,7 +175,7 @@ void
isc__taskmgr_resume(isc_taskmgr_t *manager0);
#define DEFAULT_DEFAULT_QUANTUM 25
#define DEFAULT_QUANTUM 25
#define FINISHED(m) ((m)->exiting && EMPTY((m)->tasks))
/*%
@@ -292,7 +291,7 @@ isc_task_create_bound(isc_taskmgr_t *manager0, unsigned int quantum,
INIT_LIST(task->events);
INIT_LIST(task->on_shutdown);
task->nevents = 0;
task->quantum = (quantum > 0) ? quantum : manager->default_quantum;
task->quantum = quantum;
task->flags = 0;
task->now = 0;
isc_time_settoepoch(&task->tnow);
@@ -932,6 +931,8 @@ pop_readyq(isc__taskmgr_t *manager, int c) {
if (task != NULL) {
DEQUEUE(manager->queues[c].ready_tasks, task, ready_link);
INSIST(manager->queues[c].tasks_waiting > 0);
manager->queues[c].tasks_waiting--;
if (ISC_LINK_LINKED(task, ready_priority_link)) {
DEQUEUE(manager->queues[c].ready_priority_tasks, task,
ready_priority_link);
@@ -950,12 +951,11 @@ pop_readyq(isc__taskmgr_t *manager, int c) {
static inline void
push_readyq(isc__taskmgr_t *manager, isc__task_t *task, int c) {
ENQUEUE(manager->queues[c].ready_tasks, task, ready_link);
manager->queues[c].tasks_waiting++;
if ((task->flags & TASK_F_PRIVILEGED) != 0) {
ENQUEUE(manager->queues[c].ready_priority_tasks, task,
ready_priority_link);
}
atomic_fetch_add_explicit(&manager->tasks_ready, 1,
memory_order_acquire);
}
static void
@@ -1095,6 +1095,7 @@ dispatch(isc__taskmgr_t *manager, unsigned int threadid) {
task = pop_readyq(manager, threadid);
if (task != NULL) {
unsigned int dispatch_count = 0;
unsigned int quantum = (task->quantum > 0) ? task->quantum : DEFAULT_QUANTUM;
bool done = false;
bool requeue = false;
bool finished = false;
@@ -1107,12 +1108,8 @@ dispatch(isc__taskmgr_t *manager, unsigned int threadid) {
* have a task to do. We must reacquire the queue
* lock before exiting the 'if (task != NULL)' block.
*/
manager->queues[threadid].running = true;
UNLOCK(&manager->queues[threadid].lock);
RUNTIME_CHECK(
atomic_fetch_sub_explicit(&manager->tasks_ready,
1, memory_order_release) > 0);
atomic_fetch_add_explicit(&manager->tasks_running, 1,
memory_order_acquire);
LOCK(&task->lock);
INSIST(task->state == task_state_ready);
@@ -1201,7 +1198,7 @@ dispatch(isc__taskmgr_t *manager, unsigned int threadid) {
} else
task->state = task_state_idle;
done = true;
} else if (dispatch_count >= task->quantum) {
} else if (dispatch_count >= quantum) {
/*
* Our quantum has expired, but
* there is more work to be done.
@@ -1226,10 +1223,8 @@ dispatch(isc__taskmgr_t *manager, unsigned int threadid) {
if (finished)
task_finished(task);
RUNTIME_CHECK(
atomic_fetch_sub_explicit(&manager->tasks_running,
1, memory_order_release) > 0);
LOCK(&manager->queues[threadid].lock);
manager->queues[threadid].running = false;
if (requeue) {
/*
* We know we're awake, so we don't have
@@ -1256,13 +1251,14 @@ dispatch(isc__taskmgr_t *manager, unsigned int threadid) {
/*
* If we are in privileged execution mode and there are no
* tasks remaining on the current ready queue, then
* we're stuck. Automatically drop privileges at that
* point and continue with the regular ready queue.
* tasks remaining on the current ready queue, we need to check
* if maybe we need to drop from privileged to normal mode.
* This might seem too heavy with all the locking but
* privileged mode is used only during startup and dropped
* once - after that this code is never executed.
*/
if (manager->mode != isc_taskmgrmode_normal &&
atomic_load_explicit(&manager->tasks_running,
memory_order_acquire) == 0)
empty_readyq(manager, threadid))
{
UNLOCK(&manager->queues[threadid].lock);
LOCK(&manager->lock);
@@ -1273,16 +1269,14 @@ dispatch(isc__taskmgr_t *manager, unsigned int threadid) {
* we'll end up in a deadlock over queue locks.
*
*/
if (manager->mode != isc_taskmgrmode_normal &&
atomic_load_explicit(&manager->tasks_running,
memory_order_acquire) == 0)
{
if (manager->mode != isc_taskmgrmode_normal) {
bool empty = true;
unsigned int i;
for (i = 0; i < manager->workers && empty; i++)
{
LOCK(&manager->queues[i].lock);
empty &= empty_readyq(manager, i);
empty &= empty_readyq(manager, i) &&
!manager->queues[i].running;
UNLOCK(&manager->queues[i].lock);
}
if (empty) {
@@ -1343,7 +1337,7 @@ manager_free(isc__taskmgr_t *manager) {
isc_result_t
isc_taskmgr_create(isc_mem_t *mctx, unsigned int workers,
unsigned int default_quantum, isc_taskmgr_t **managerp)
isc_taskmgr_t **managerp)
{
unsigned int i;
isc__taskmgr_t *manager;
@@ -1369,16 +1363,10 @@ isc_taskmgr_create(isc_mem_t *mctx, unsigned int workers,
manager->workers = workers;
if (default_quantum == 0) {
default_quantum = DEFAULT_DEFAULT_QUANTUM;
}
manager->default_quantum = default_quantum;
INIT_LIST(manager->tasks);
manager->queues = isc_mem_get(mctx, workers * sizeof(isc__taskqueue_t));
RUNTIME_CHECK(manager->queues != NULL);
manager->tasks_running = 0;
manager->tasks_ready = 0;
manager->curq = 0;
manager->exiting = false;
manager->excl = NULL;
@@ -1400,6 +1388,8 @@ isc_taskmgr_create(isc_mem_t *mctx, unsigned int workers,
manager->queues[i].manager = manager;
manager->queues[i].threadid = i;
manager->queues[i].tasks_waiting = 0;
manager->queues[i].running = false;
RUNTIME_CHECK(isc_thread_create(run, &manager->queues[i],
&manager->queues[i].thread)
== ISC_R_SUCCESS);
@@ -1708,21 +1698,6 @@ isc_taskmgr_renderxml(isc_taskmgr_t *mgr0, xmlTextWriterPtr writer) {
TRY0(xmlTextWriterWriteFormatString(writer, "%d", mgr->workers));
TRY0(xmlTextWriterEndElement(writer)); /* worker-threads */
TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "default-quantum"));
TRY0(xmlTextWriterWriteFormatString(writer, "%d",
mgr->default_quantum));
TRY0(xmlTextWriterEndElement(writer)); /* default-quantum */
TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "tasks-running"));
TRY0(xmlTextWriterWriteFormatString(writer, "%d",
(int) mgr->tasks_running));
TRY0(xmlTextWriterEndElement(writer)); /* tasks-running */
TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "tasks-ready"));
TRY0(xmlTextWriterWriteFormatString(writer, "%d",
(int) mgr->tasks_ready));
TRY0(xmlTextWriterEndElement(writer)); /* tasks-ready */
TRY0(xmlTextWriterEndElement(writer)); /* thread-model */
TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "tasks"));
@@ -1809,18 +1784,6 @@ isc_taskmgr_renderjson(isc_taskmgr_t *mgr0, json_object *tasks) {
CHECKMEM(obj);
json_object_object_add(tasks, "worker-threads", obj);
obj = json_object_new_int(mgr->default_quantum);
CHECKMEM(obj);
json_object_object_add(tasks, "default-quantum", obj);
obj = json_object_new_int(mgr->tasks_running);
CHECKMEM(obj);
json_object_object_add(tasks, "tasks-running", obj);
obj = json_object_new_int(mgr->tasks_ready);
CHECKMEM(obj);
json_object_object_add(tasks, "tasks-ready", obj);
array = json_object_new_array();
CHECKMEM(array);
@@ -1885,13 +1848,11 @@ isc_taskmgr_renderjson(isc_taskmgr_t *mgr0, json_object *tasks) {
isc_result_t
isc_taskmgr_createinctx(isc_mem_t *mctx, isc_appctx_t *actx,
unsigned int workers, unsigned int default_quantum,
isc_taskmgr_t **managerp)
unsigned int workers, isc_taskmgr_t **managerp)
{
isc_result_t result;
result = isc_taskmgr_create(mctx, workers, default_quantum,
managerp);
result = isc_taskmgr_create(mctx, workers, managerp);
if (result == ISC_R_SUCCESS)
isc_appctx_settaskmgr(actx, *managerp);
+1 -1
View File
@@ -86,7 +86,7 @@ create_managers(unsigned int workers) {
workers = atoi(p);
}
CHECK(isc_taskmgr_create(mctx, workers, 0, &taskmgr));
CHECK(isc_taskmgr_create(mctx, workers, &taskmgr));
CHECK(isc_task_create(taskmgr, 0, &maintask));
isc_taskmgr_setexcltask(taskmgr, maintask);
+1 -1
View File
@@ -712,7 +712,7 @@ manytasks(void **state) {
result = isc_mem_create(0, 0, &mctx);
assert_int_equal(result, ISC_R_SUCCESS);
result = isc_taskmgr_create(mctx, 4, 0, &taskmgr);
result = isc_taskmgr_create(mctx, 4, &taskmgr);
assert_int_equal(result, ISC_R_SUCCESS);
done = false;
+125 -5
View File
@@ -924,7 +924,7 @@ client_allocsendbuf(ns_client_t *client, isc_buffer_t *buffer,
}
static isc_result_t
client_sendpkg(ns_client_t *client, isc_buffer_t *buffer) {
client_sendpkg(ns_client_t *client, isc_buffer_t *buffer, bool lowac) {
struct in6_pktinfo *pktinfo;
isc_result_t result;
isc_region_t r;
@@ -989,7 +989,9 @@ client_sendpkg(ns_client_t *client, isc_buffer_t *buffer) {
client->sendevent->attributes |= ISC_SOCKEVENTATTR_USEMINMTU;
CTRACE("sendto");
if (lowac && client->view != NULL) {
dns_lowac_put(client->view->lowac, ISC_LIST_HEAD(client->message->sections[0]), (char*) r.base, r.length);
}
result = isc_socket_sendto2(sock, &r, client->task,
address, pktinfo,
client->sendevent, sockflags);
@@ -1037,7 +1039,7 @@ ns_client_sendraw(ns_client_t *client, dns_message_t *message) {
r.base[0] = (client->message->id >> 8) & 0xff;
r.base[1] = client->message->id & 0xff;
result = client_sendpkg(client, &buffer);
result = client_sendpkg(client, &buffer, true);
if (result == ISC_R_SUCCESS)
return;
@@ -1049,6 +1051,51 @@ ns_client_sendraw(ns_client_t *client, dns_message_t *message) {
ns_client_next(client, result);
}
#ifdef PSEUDOSEND
static void
client_pseudosend(ns_client_t *client, uint16_t id) {
isc_result_t result;
unsigned char *data;
isc_buffer_t buffer;
isc_buffer_t tcpbuffer;
isc_region_t r;
unsigned char sendbuf[SEND_BUFFER_SIZE];
result = client_allocsendbuf(client, &buffer, &tcpbuffer, 0,
sendbuf, &data);
if (result != ISC_R_SUCCESS) {
goto done;
}
isc_buffer_putuint16(&buffer, id);
isc_buffer_putuint8(&buffer, 0x81);
isc_buffer_putuint8(&buffer, 0x80);
isc_buffer_putuint16(&buffer, 0);
isc_buffer_putuint16(&buffer, 0);
isc_buffer_putuint16(&buffer, 0);
isc_buffer_putuint16(&buffer, 0);
if (client->sendcb != NULL) {
client->sendcb(&buffer);
} else if (TCP_CLIENT(client)) {
isc_buffer_usedregion(&buffer, &r);
isc_buffer_putuint16(&tcpbuffer, (uint16_t) r.length);
isc_buffer_add(&tcpbuffer, r.length);
result = client_sendpkg(client, &tcpbuffer);
} else {
result = client_sendpkg(client, &buffer);
}
if (result == ISC_R_SUCCESS) {
return;
}
done:
if (client->tcpbuf != NULL) {
isc_mem_put(client->mctx, client->tcpbuf, TCP_BUFFER_SIZE);
client->tcpbuf = NULL;
}
ns_client_next(client, result);
}
#endif
static void
client_send(ns_client_t *client) {
isc_result_t result;
@@ -1257,7 +1304,7 @@ client_send(ns_client_t *client) {
/* don't count the 2-octet length header */
respsize = isc_buffer_usedlength(&tcpbuffer) - 2;
result = client_sendpkg(client, &tcpbuffer);
result = client_sendpkg(client, &tcpbuffer, true);
switch (isc_sockaddr_pf(&client->peeraddr)) {
case AF_INET:
@@ -1288,7 +1335,7 @@ client_send(ns_client_t *client) {
#endif /* HAVE_DNSTAP */
respsize = isc_buffer_usedlength(&buffer);
result = client_sendpkg(client, &buffer);
result = client_sendpkg(client, &buffer, true);
switch (isc_sockaddr_pf(&client->peeraddr)) {
case AF_INET:
@@ -2226,6 +2273,60 @@ process_opt(ns_client_t *client, dns_rdataset_t *opt) {
return (result);
}
static isc_result_t
lowac_checksend(ns_client_t *client) {
isc_result_t result;
unsigned char *data;
isc_buffer_t buffer;
isc_buffer_t tcpbuffer;
isc_region_t r;
unsigned char sendbuf[SEND_BUFFER_SIZE];
result = client_allocsendbuf(client, &buffer, &tcpbuffer, 0,
sendbuf, &data);
if (result != ISC_R_SUCCESS) {
return (result);
}
int blen;
result = dns_lowac_get(client->view->lowac, ISC_LIST_HEAD(client->message->sections[0]), data, &blen, TCP_CLIENT(client));
if (result != ISC_R_SUCCESS) {
goto done;
}
if (TCP_CLIENT(client)) {
data[2] = (client->message->id >> 8) & 0xff;
data[3] = client->message->id & 0xff;
} else {
data[0] = (client->message->id >> 8) & 0xff;
data[1] = client->message->id & 0xff;
}
isc_buffer_add(&buffer, blen);
/*
isc_buffer_putuint16(&buffer, client->message->id);
isc_buffer_putuint8(&buffer, 0x81);
isc_buffer_putuint8(&buffer, 0x80);
isc_buffer_putuint16(&buffer, 0);
isc_buffer_putuint16(&buffer, 0);
isc_buffer_putuint16(&buffer, 0);
isc_buffer_putuint16(&buffer, 0);
*/
if (client->sendcb != NULL) {
client->sendcb(&buffer);
} else if (TCP_CLIENT(client)) {
isc_buffer_usedregion(&buffer, &r);
isc_buffer_putuint16(&tcpbuffer, (uint16_t) r.length);
isc_buffer_add(&tcpbuffer, r.length);
result = client_sendpkg(client, &tcpbuffer, false);
} else {
result = client_sendpkg(client, &buffer, false);
}
done:
if (client->tcpbuf != NULL) {
isc_mem_put(client->mctx, client->tcpbuf, TCP_BUFFER_SIZE);
client->tcpbuf = NULL;
}
return result;
}
/*
* Handle an incoming request event from the socket (UDP case)
* or tcpmsg (TCP case).
@@ -2335,6 +2436,20 @@ ns__client_request(isc_task_t *task, isc_event_t *event) {
return;
}
#ifdef PSEUDOSEND
result = dns_message_peekheader(buffer, &id, &flags);
if (result != ISC_R_SUCCESS) {
/*
* There isn't enough header to determine whether
* this was a request or a response. Drop it.
*/
ns_client_next(client, result);
return;
}
client_pseudosend(client, id);
return;
#endif
isc_netaddr_fromsockaddr(&netaddr, &client->peeraddr);
#if NS_CLIENT_DROPPORT
@@ -2674,6 +2789,11 @@ ns__client_request(isc_task_t *task, isc_event_t *event) {
NS_LOGMODULE_CLIENT, ISC_LOG_DEBUG(5),
"using view '%s'", client->view->name);
result = lowac_checksend(client);
if (result == ISC_R_SUCCESS) {
return;
}
/*
* Check for a signature. We log bad signatures regardless of
* whether they ultimately cause the request to be rejected or
+1 -1
View File
@@ -182,7 +182,7 @@ create_managers(void) {
isc_event_t *event = NULL;
ncpus = isc_os_ncpus();
CHECK(isc_taskmgr_create(mctx, ncpus, 0, &taskmgr));
CHECK(isc_taskmgr_create(mctx, ncpus, &taskmgr));
CHECK(isc_task_create(taskmgr, 0, &maintask));
isc_taskmgr_setexcltask(taskmgr, maintask);
CHECK(isc_task_onshutdown(maintask, shutdown_managers, NULL));
+1 -1
View File
@@ -226,7 +226,7 @@ ctxs_init(isc_mem_t **mctxp, isc_appctx_t **actxp,
if (result != ISC_R_SUCCESS)
goto fail;
result = isc_taskmgr_createinctx(*mctxp, *actxp, 1, 0, taskmgrp);
result = isc_taskmgr_createinctx(*mctxp, *actxp, 1, taskmgrp);
if (result != ISC_R_SUCCESS)
goto fail;
+1 -1
View File
@@ -381,7 +381,7 @@ main(int argc, char *argv[]) {
result = isc_app_ctxstart(actx);
if (result != ISC_R_SUCCESS)
goto cleanup;
result = isc_taskmgr_createinctx(mctx, actx, 1, 0, &taskmgr);
result = isc_taskmgr_createinctx(mctx, actx, 1, &taskmgr);
if (result != ISC_R_SUCCESS)
goto cleanup;
result = isc_socketmgr_createinctx(mctx, actx, &socketmgr);
+1 -1
View File
@@ -108,7 +108,7 @@ ctxs_init(isc_mem_t **mctxp, isc_appctx_t **actxp,
if (result != ISC_R_SUCCESS)
goto fail;
result = isc_taskmgr_createinctx(*mctxp, *actxp, 1, 0, taskmgrp);
result = isc_taskmgr_createinctx(*mctxp, *actxp, 1, taskmgrp);
if (result != ISC_R_SUCCESS)
goto fail;