diff options
Diffstat (limited to 'lib/isc/task.c')
| -rw-r--r-- | lib/isc/task.c | 1486 |
1 files changed, 1486 insertions, 0 deletions
diff --git a/lib/isc/task.c b/lib/isc/task.c new file mode 100644 index 0000000..3d8deaf --- /dev/null +++ b/lib/isc/task.c @@ -0,0 +1,1486 @@ +/* + * 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 */ + +/* + * XXXRTH Need to document the states a task can be in, and the rules + * for changing states. + */ + +#include <stdbool.h> +#include <unistd.h> + +#include <isc/app.h> +#include <isc/atomic.h> +#include <isc/condition.h> +#include <isc/event.h> +#include <isc/log.h> +#include <isc/magic.h> +#include <isc/mem.h> +#include <isc/once.h> +#include <isc/platform.h> +#include <isc/print.h> +#include <isc/random.h> +#include <isc/refcount.h> +#include <isc/string.h> +#include <isc/task.h> +#include <isc/thread.h> +#include <isc/time.h> +#include <isc/util.h> + +#ifdef HAVE_LIBXML2 +#include <libxml/xmlwriter.h> +#define ISC_XMLCHAR (const xmlChar *) +#endif /* HAVE_LIBXML2 */ + +#ifdef HAVE_JSON_C +#include <json_object.h> +#endif /* HAVE_JSON_C */ + +#include "task_p.h" + +/* + * Task manager is built around 'as little locking as possible' concept. + * Each thread has his own queue of tasks to be run, if a task is in running + * state it will stay on the runner it's currently on, if a task is in idle + * state it can be woken up on a specific runner with isc_task_sendto - that + * helps with data locality on CPU. + * + * To make load even some tasks (from task pools) are bound to specific + * queues using isc_task_create_bound. This way load balancing between + * CPUs/queues happens on the higher layer. + */ + +#ifdef ISC_TASK_TRACE +#define XTRACE(m) \ + fprintf(stderr, "task %p thread %zu: %s\n", task, isc_tid_v, (m)) +#define XTTRACE(t, m) \ + fprintf(stderr, "task %p thread %zu: %s\n", (t), isc_tid_v, (m)) +#define XTHREADTRACE(m) fprintf(stderr, "thread %zu: %s\n", isc_tid_v, (m)) +#else /* ifdef ISC_TASK_TRACE */ +#define XTRACE(m) +#define XTTRACE(t, m) +#define XTHREADTRACE(m) +#endif /* ifdef ISC_TASK_TRACE */ + +/*** + *** Types. + ***/ + +typedef enum { + task_state_idle, /* not doing anything, events queue empty */ + task_state_ready, /* waiting in worker's queue */ + task_state_paused, /* not running, paused */ + task_state_pausing, /* running, waiting to be paused */ + task_state_running, /* actively processing events */ + task_state_done /* shutting down, no events or references */ +} task_state_t; + +#if defined(HAVE_LIBXML2) || defined(HAVE_JSON_C) +static const char *statenames[] = { + "idle", "ready", "paused", "pausing", "running", "done", +}; +#endif /* if defined(HAVE_LIBXML2) || defined(HAVE_JSON_C) */ + +#define TASK_MAGIC ISC_MAGIC('T', 'A', 'S', 'K') +#define VALID_TASK(t) ISC_MAGIC_VALID(t, TASK_MAGIC) + +struct isc_task { + /* Not locked. */ + unsigned int magic; + isc_taskmgr_t *manager; + isc_mutex_t lock; + /* Locked by task lock. */ + int threadid; + task_state_t state; + int pause_cnt; + isc_refcount_t references; + isc_refcount_t running; + isc_eventlist_t events; + isc_eventlist_t on_shutdown; + unsigned int nevents; + unsigned int quantum; + isc_stdtime_t now; + isc_time_t tnow; + char name[16]; + void *tag; + bool bound; + /* Protected by atomics */ + atomic_bool shuttingdown; + atomic_bool privileged; + /* Locked by task manager lock. */ + LINK(isc_task_t) link; +}; + +#define TASK_SHUTTINGDOWN(t) (atomic_load_acquire(&(t)->shuttingdown)) +#define TASK_PRIVILEGED(t) (atomic_load_acquire(&(t)->privileged)) + +#define TASK_MANAGER_MAGIC ISC_MAGIC('T', 'S', 'K', 'M') +#define VALID_MANAGER(m) ISC_MAGIC_VALID(m, TASK_MANAGER_MAGIC) + +struct isc_taskmgr { + /* Not locked. */ + unsigned int magic; + isc_refcount_t references; + isc_mem_t *mctx; + isc_mutex_t lock; + atomic_uint_fast32_t tasks_count; + isc_nm_t *netmgr; + + /* Locked by task manager lock. */ + unsigned int default_quantum; + LIST(isc_task_t) tasks; + atomic_uint_fast32_t mode; + atomic_bool exclusive_req; + bool exiting; + isc_task_t *excl; +}; + +#define DEFAULT_DEFAULT_QUANTUM 25 + +/*% + * The following are intended for internal use (indicated by "isc__" + * prefix) but are not declared as static, allowing direct access from + * unit tests etc. + */ + +bool +isc_task_purgeevent(isc_task_t *task, isc_event_t *event); +void +isc_taskmgr_setexcltask(isc_taskmgr_t *mgr, isc_task_t *task); +isc_result_t +isc_taskmgr_excltask(isc_taskmgr_t *mgr, isc_task_t **taskp); + +/*** + *** Tasks. + ***/ + +static void +task_finished(isc_task_t *task) { + isc_taskmgr_t *manager = task->manager; + isc_mem_t *mctx = manager->mctx; + REQUIRE(EMPTY(task->events)); + REQUIRE(task->nevents == 0); + REQUIRE(EMPTY(task->on_shutdown)); + REQUIRE(task->state == task_state_done); + + XTRACE("task_finished"); + + isc_refcount_destroy(&task->running); + isc_refcount_destroy(&task->references); + + LOCK(&manager->lock); + UNLINK(manager->tasks, task, link); + atomic_fetch_sub(&manager->tasks_count, 1); + UNLOCK(&manager->lock); + + isc_mutex_destroy(&task->lock); + task->magic = 0; + isc_mem_put(mctx, task, sizeof(*task)); + + isc_taskmgr_detach(&manager); +} + +isc_result_t +isc_task_create(isc_taskmgr_t *manager, unsigned int quantum, + isc_task_t **taskp) { + return (isc_task_create_bound(manager, quantum, taskp, -1)); +} + +isc_result_t +isc_task_create_bound(isc_taskmgr_t *manager, unsigned int quantum, + isc_task_t **taskp, int threadid) { + isc_task_t *task = NULL; + bool exiting; + + REQUIRE(VALID_MANAGER(manager)); + REQUIRE(taskp != NULL && *taskp == NULL); + + XTRACE("isc_task_create"); + + task = isc_mem_get(manager->mctx, sizeof(*task)); + *task = (isc_task_t){ 0 }; + + isc_taskmgr_attach(manager, &task->manager); + + if (threadid == -1) { + /* + * Task is not pinned to a queue, it's threadid will be + * chosen when first task will be sent to it - either + * randomly or specified by isc_task_sendto. + */ + task->bound = false; + task->threadid = -1; + } else { + /* + * Task is pinned to a queue, it'll always be run + * by a specific thread. + */ + task->bound = true; + task->threadid = threadid; + } + + isc_mutex_init(&task->lock); + task->state = task_state_idle; + task->pause_cnt = 0; + + isc_refcount_init(&task->references, 1); + isc_refcount_init(&task->running, 0); + INIT_LIST(task->events); + INIT_LIST(task->on_shutdown); + task->nevents = 0; + task->quantum = (quantum > 0) ? quantum : manager->default_quantum; + atomic_init(&task->shuttingdown, false); + atomic_init(&task->privileged, false); + task->now = 0; + isc_time_settoepoch(&task->tnow); + memset(task->name, 0, sizeof(task->name)); + task->tag = NULL; + INIT_LINK(task, link); + task->magic = TASK_MAGIC; + + LOCK(&manager->lock); + exiting = manager->exiting; + if (!exiting) { + APPEND(manager->tasks, task, link); + atomic_fetch_add(&manager->tasks_count, 1); + } + UNLOCK(&manager->lock); + + if (exiting) { + isc_refcount_destroy(&task->running); + isc_refcount_decrement(&task->references); + isc_refcount_destroy(&task->references); + isc_mutex_destroy(&task->lock); + isc_taskmgr_detach(&task->manager); + isc_mem_put(manager->mctx, task, sizeof(*task)); + return (ISC_R_SHUTTINGDOWN); + } + + *taskp = task; + + return (ISC_R_SUCCESS); +} + +void +isc_task_attach(isc_task_t *source, isc_task_t **targetp) { + /* + * Attach *targetp to source. + */ + + REQUIRE(VALID_TASK(source)); + REQUIRE(targetp != NULL && *targetp == NULL); + + XTTRACE(source, "isc_task_attach"); + + isc_refcount_increment(&source->references); + + *targetp = source; +} + +static bool +task_shutdown(isc_task_t *task) { + bool was_idle = false; + isc_event_t *event, *prev; + + /* + * Caller must be holding the task's lock. + */ + + XTRACE("task_shutdown"); + + if (atomic_compare_exchange_strong(&task->shuttingdown, + &(bool){ false }, true)) + { + XTRACE("shutting down"); + if (task->state == task_state_idle) { + INSIST(EMPTY(task->events)); + task->state = task_state_ready; + was_idle = true; + } + INSIST(task->state == task_state_ready || + task->state == task_state_paused || + task->state == task_state_pausing || + task->state == task_state_running); + + /* + * Note that we post shutdown events LIFO. + */ + for (event = TAIL(task->on_shutdown); event != NULL; + event = prev) + { + prev = PREV(event, ev_link); + DEQUEUE(task->on_shutdown, event, ev_link); + ENQUEUE(task->events, event, ev_link); + task->nevents++; + } + } + + return (was_idle); +} + +/* + * Moves a task onto the appropriate run queue. + * + * Caller must NOT hold queue lock. + */ +static void +task_ready(isc_task_t *task) { + isc_taskmgr_t *manager = task->manager; + REQUIRE(VALID_MANAGER(manager)); + + XTRACE("task_ready"); + + isc_refcount_increment0(&task->running); + LOCK(&task->lock); + isc_nm_task_enqueue(manager->netmgr, task, task->threadid); + UNLOCK(&task->lock); +} + +void +isc_task_ready(isc_task_t *task) { + task_ready(task); +} + +static bool +task_detach(isc_task_t *task) { + /* + * Caller must be holding the task lock. + */ + + XTRACE("detach"); + + if (isc_refcount_decrement(&task->references) == 1 && + task->state == task_state_idle) + { + INSIST(EMPTY(task->events)); + /* + * There are no references to this task, and no + * pending events. We could try to optimize and + * either initiate shutdown or clean up the task, + * depending on its state, but it's easier to just + * make the task ready and allow run() or the event + * loop to deal with shutting down and termination. + */ + task->state = task_state_ready; + return (true); + } + + return (false); +} + +void +isc_task_detach(isc_task_t **taskp) { + isc_task_t *task; + bool was_idle; + + /* + * Detach *taskp from its task. + */ + + REQUIRE(taskp != NULL); + task = *taskp; + REQUIRE(VALID_TASK(task)); + + XTRACE("isc_task_detach"); + + LOCK(&task->lock); + was_idle = task_detach(task); + UNLOCK(&task->lock); + + if (was_idle) { + task_ready(task); + } + + *taskp = NULL; +} + +static bool +task_send(isc_task_t *task, isc_event_t **eventp, int c) { + bool was_idle = false; + isc_event_t *event; + + /* + * Caller must be holding the task lock. + */ + + REQUIRE(eventp != NULL); + event = *eventp; + *eventp = NULL; + REQUIRE(event != NULL); + REQUIRE(event->ev_type > 0); + REQUIRE(task->state != task_state_done); + REQUIRE(!ISC_LINK_LINKED(event, ev_ratelink)); + + XTRACE("task_send"); + + if (task->bound) { + c = task->threadid; + } else if (c < 0) { + c = -1; + } + + if (task->state == task_state_idle) { + was_idle = true; + task->threadid = c; + INSIST(EMPTY(task->events)); + task->state = task_state_ready; + } + INSIST(task->state == task_state_ready || + task->state == task_state_running || + task->state == task_state_paused || + task->state == task_state_pausing); + ENQUEUE(task->events, event, ev_link); + task->nevents++; + + return (was_idle); +} + +void +isc_task_send(isc_task_t *task, isc_event_t **eventp) { + isc_task_sendto(task, eventp, -1); +} + +void +isc_task_sendanddetach(isc_task_t **taskp, isc_event_t **eventp) { + isc_task_sendtoanddetach(taskp, eventp, -1); +} + +void +isc_task_sendto(isc_task_t *task, isc_event_t **eventp, int c) { + bool was_idle; + + /* + * Send '*event' to 'task'. + */ + + REQUIRE(VALID_TASK(task)); + XTRACE("isc_task_send"); + + /* + * We're trying hard to hold locks for as short a time as possible. + * We're also trying to hold as few locks as possible. This is why + * some processing is deferred until after the lock is released. + */ + LOCK(&task->lock); + was_idle = task_send(task, eventp, c); + UNLOCK(&task->lock); + + if (was_idle) { + /* + * We need to add this task to the ready queue. + * + * We've waited until now to do it because making a task + * ready requires locking the manager. If we tried to do + * this while holding the task lock, we could deadlock. + * + * We've changed the state to ready, so no one else will + * be trying to add this task to the ready queue. The + * only way to leave the ready state is by executing the + * task. It thus doesn't matter if events are added, + * removed, or a shutdown is started in the interval + * between the time we released the task lock, and the time + * we add the task to the ready queue. + */ + task_ready(task); + } +} + +void +isc_task_sendtoanddetach(isc_task_t **taskp, isc_event_t **eventp, int c) { + bool idle1, idle2; + isc_task_t *task; + + /* + * Send '*event' to '*taskp' and then detach '*taskp' from its + * task. + */ + + REQUIRE(taskp != NULL); + task = *taskp; + REQUIRE(VALID_TASK(task)); + XTRACE("isc_task_sendanddetach"); + + LOCK(&task->lock); + idle1 = task_send(task, eventp, c); + idle2 = task_detach(task); + UNLOCK(&task->lock); + + /* + * If idle1, then idle2 shouldn't be true as well since we're holding + * the task lock, and thus the task cannot switch from ready back to + * idle. + */ + INSIST(!(idle1 && idle2)); + + if (idle1 || idle2) { + task_ready(task); + } + + *taskp = NULL; +} + +#define PURGE_OK(event) (((event)->ev_attributes & ISC_EVENTATTR_NOPURGE) == 0) + +static unsigned int +dequeue_events(isc_task_t *task, void *sender, isc_eventtype_t first, + isc_eventtype_t last, void *tag, isc_eventlist_t *events, + bool purging) { + isc_event_t *event, *next_event; + unsigned int count = 0; + + REQUIRE(VALID_TASK(task)); + REQUIRE(last >= first); + + XTRACE("dequeue_events"); + + /* + * Events matching 'sender', whose type is >= first and <= last, and + * whose tag is 'tag' will be dequeued. If 'purging', matching events + * which are marked as unpurgable will not be dequeued. + * + * sender == NULL means "any sender", and tag == NULL means "any tag". + */ + + LOCK(&task->lock); + + for (event = HEAD(task->events); event != NULL; event = next_event) { + next_event = NEXT(event, ev_link); + if (event->ev_type >= first && event->ev_type <= last && + (sender == NULL || event->ev_sender == sender) && + (tag == NULL || event->ev_tag == tag) && + (!purging || PURGE_OK(event))) + { + DEQUEUE(task->events, event, ev_link); + task->nevents--; + ENQUEUE(*events, event, ev_link); + count++; + } + } + + UNLOCK(&task->lock); + + return (count); +} + +unsigned int +isc_task_purgerange(isc_task_t *task, void *sender, isc_eventtype_t first, + isc_eventtype_t last, void *tag) { + unsigned int count; + isc_eventlist_t events; + isc_event_t *event, *next_event; + REQUIRE(VALID_TASK(task)); + + /* + * Purge events from a task's event queue. + */ + + XTRACE("isc_task_purgerange"); + + ISC_LIST_INIT(events); + + count = dequeue_events(task, sender, first, last, tag, &events, true); + + for (event = HEAD(events); event != NULL; event = next_event) { + next_event = NEXT(event, ev_link); + ISC_LIST_UNLINK(events, event, ev_link); + isc_event_free(&event); + } + + /* + * Note that purging never changes the state of the task. + */ + + return (count); +} + +unsigned int +isc_task_purge(isc_task_t *task, void *sender, isc_eventtype_t type, + void *tag) { + /* + * Purge events from a task's event queue. + */ + REQUIRE(VALID_TASK(task)); + + XTRACE("isc_task_purge"); + + return (isc_task_purgerange(task, sender, type, type, tag)); +} + +bool +isc_task_purgeevent(isc_task_t *task, isc_event_t *event) { + bool found = false; + + /* + * Purge 'event' from a task's event queue. + */ + + REQUIRE(VALID_TASK(task)); + + /* + * If 'event' is on the task's event queue, it will be purged, + * unless it is marked as unpurgeable. 'event' does not have to be + * on the task's event queue; in fact, it can even be an invalid + * pointer. Purging only occurs if the event is actually on the task's + * event queue. + * + * Purging never changes the state of the task. + */ + + LOCK(&task->lock); + if (ISC_LINK_LINKED(event, ev_link)) { + DEQUEUE(task->events, event, ev_link); + task->nevents--; + found = true; + } + UNLOCK(&task->lock); + + if (!found) { + return (false); + } + + isc_event_free(&event); + + return (true); +} + +unsigned int +isc_task_unsendrange(isc_task_t *task, void *sender, isc_eventtype_t first, + isc_eventtype_t last, void *tag, isc_eventlist_t *events) { + /* + * Remove events from a task's event queue. + */ + REQUIRE(VALID_TASK(task)); + + XTRACE("isc_task_unsendrange"); + + return (dequeue_events(task, sender, first, last, tag, events, false)); +} + +unsigned int +isc_task_unsend(isc_task_t *task, void *sender, isc_eventtype_t type, void *tag, + isc_eventlist_t *events) { + /* + * Remove events from a task's event queue. + */ + + XTRACE("isc_task_unsend"); + + return (dequeue_events(task, sender, type, type, tag, events, false)); +} + +isc_result_t +isc_task_onshutdown(isc_task_t *task, isc_taskaction_t action, void *arg) { + bool disallowed = false; + isc_result_t result = ISC_R_SUCCESS; + isc_event_t *event; + + /* + * Send a shutdown event with action 'action' and argument 'arg' when + * 'task' is shutdown. + */ + + REQUIRE(VALID_TASK(task)); + REQUIRE(action != NULL); + + event = isc_event_allocate(task->manager->mctx, NULL, + ISC_TASKEVENT_SHUTDOWN, action, arg, + sizeof(*event)); + + if (TASK_SHUTTINGDOWN(task)) { + disallowed = true; + result = ISC_R_SHUTTINGDOWN; + } else { + LOCK(&task->lock); + ENQUEUE(task->on_shutdown, event, ev_link); + UNLOCK(&task->lock); + } + + if (disallowed) { + isc_mem_put(task->manager->mctx, event, sizeof(*event)); + } + + return (result); +} + +void +isc_task_shutdown(isc_task_t *task) { + bool was_idle; + + /* + * Shutdown 'task'. + */ + + REQUIRE(VALID_TASK(task)); + + LOCK(&task->lock); + was_idle = task_shutdown(task); + UNLOCK(&task->lock); + + if (was_idle) { + task_ready(task); + } +} + +void +isc_task_destroy(isc_task_t **taskp) { + /* + * Destroy '*taskp'. + */ + + REQUIRE(taskp != NULL); + + isc_task_shutdown(*taskp); + isc_task_detach(taskp); +} + +void +isc_task_setname(isc_task_t *task, const char *name, void *tag) { + /* + * Name 'task'. + */ + + REQUIRE(VALID_TASK(task)); + + LOCK(&task->lock); + strlcpy(task->name, name, sizeof(task->name)); + task->tag = tag; + UNLOCK(&task->lock); +} + +const char * +isc_task_getname(isc_task_t *task) { + REQUIRE(VALID_TASK(task)); + + return (task->name); +} + +void * +isc_task_gettag(isc_task_t *task) { + REQUIRE(VALID_TASK(task)); + + return (task->tag); +} + +void +isc_task_getcurrenttime(isc_task_t *task, isc_stdtime_t *t) { + REQUIRE(VALID_TASK(task)); + REQUIRE(t != NULL); + + LOCK(&task->lock); + *t = task->now; + UNLOCK(&task->lock); +} + +void +isc_task_getcurrenttimex(isc_task_t *task, isc_time_t *t) { + REQUIRE(VALID_TASK(task)); + REQUIRE(t != NULL); + + LOCK(&task->lock); + *t = task->tnow; + UNLOCK(&task->lock); +} + +isc_nm_t * +isc_task_getnetmgr(isc_task_t *task) { + REQUIRE(VALID_TASK(task)); + + return (task->manager->netmgr); +} + +void +isc_task_setquantum(isc_task_t *task, unsigned int quantum) { + REQUIRE(VALID_TASK(task)); + + LOCK(&task->lock); + task->quantum = (quantum > 0) ? quantum + : task->manager->default_quantum; + UNLOCK(&task->lock); +} + +/*** + *** Task Manager. + ***/ + +static isc_result_t +task_run(isc_task_t *task) { + unsigned int dispatch_count = 0; + bool finished = false; + isc_event_t *event = NULL; + isc_result_t result = ISC_R_SUCCESS; + uint32_t quantum; + + REQUIRE(VALID_TASK(task)); + + LOCK(&task->lock); + quantum = task->quantum; + + /* + * It is possible because that we have a paused task in the queue - it + * might have been paused in the meantime and we never hold both queue + * and task lock to avoid deadlocks, just bail then. + */ + if (task->state != task_state_ready) { + goto done; + } + + INSIST(task->state == task_state_ready); + task->state = task_state_running; + XTRACE("running"); + XTRACE(task->name); + TIME_NOW(&task->tnow); + task->now = isc_time_seconds(&task->tnow); + + while (true) { + if (!EMPTY(task->events)) { + event = HEAD(task->events); + DEQUEUE(task->events, event, ev_link); + task->nevents--; + + /* + * Execute the event action. + */ + XTRACE("execute action"); + XTRACE(task->name); + if (event->ev_action != NULL) { + UNLOCK(&task->lock); + (event->ev_action)(task, event); + LOCK(&task->lock); + } + XTRACE("execution complete"); + dispatch_count++; + } + + if (isc_refcount_current(&task->references) == 0 && + EMPTY(task->events) && !TASK_SHUTTINGDOWN(task)) + { + /* + * There are no references and no pending events for + * this task, which means it will not become runnable + * again via an external action (such as sending an + * event or detaching). + * + * We initiate shutdown to prevent it from becoming a + * zombie. + * + * We do this here instead of in the "if + * EMPTY(task->events)" block below because: + * + * If we post no shutdown events, we want the task + * to finish. + * + * If we did post shutdown events, will still want + * the task's quantum to be applied. + */ + INSIST(!task_shutdown(task)); + } + + if (EMPTY(task->events)) { + /* + * Nothing else to do for this task right now. + */ + XTRACE("empty"); + if (isc_refcount_current(&task->references) == 0 && + TASK_SHUTTINGDOWN(task)) + { + /* + * The task is done. + */ + XTRACE("done"); + task->state = task_state_done; + } else { + if (task->state == task_state_running) { + XTRACE("idling"); + task->state = task_state_idle; + } else if (task->state == task_state_pausing) { + XTRACE("pausing"); + task->state = task_state_paused; + } + } + break; + } else if (task->state == task_state_pausing) { + /* + * We got a pause request on this task, stop working on + * it and switch the state to paused. + */ + XTRACE("pausing"); + task->state = task_state_paused; + break; + } else if (dispatch_count >= quantum) { + /* + * Our quantum has expired, but there is more work to be + * done. We'll requeue it to the ready queue later. + * + * We don't check quantum until dispatching at least one + * event, so the minimum quantum is one. + */ + XTRACE("quantum"); + task->state = task_state_ready; + result = ISC_R_QUOTA; + break; + } + } + +done: + if (isc_refcount_decrement(&task->running) == 1 && + task->state == task_state_done) + { + finished = true; + } + UNLOCK(&task->lock); + + if (finished) { + task_finished(task); + } + + return (result); +} + +isc_result_t +isc_task_run(isc_task_t *task) { + return (task_run(task)); +} + +static void +manager_free(isc_taskmgr_t *manager) { + isc_refcount_destroy(&manager->references); + isc_nm_detach(&manager->netmgr); + + isc_mutex_destroy(&manager->lock); + manager->magic = 0; + isc_mem_putanddetach(&manager->mctx, manager, sizeof(*manager)); +} + +void +isc_taskmgr_attach(isc_taskmgr_t *source, isc_taskmgr_t **targetp) { + REQUIRE(VALID_MANAGER(source)); + REQUIRE(targetp != NULL && *targetp == NULL); + + isc_refcount_increment(&source->references); + + *targetp = source; +} + +void +isc_taskmgr_detach(isc_taskmgr_t **managerp) { + REQUIRE(managerp != NULL); + REQUIRE(VALID_MANAGER(*managerp)); + + isc_taskmgr_t *manager = *managerp; + *managerp = NULL; + + if (isc_refcount_decrement(&manager->references) == 1) { + manager_free(manager); + } +} + +isc_result_t +isc__taskmgr_create(isc_mem_t *mctx, unsigned int default_quantum, isc_nm_t *nm, + isc_taskmgr_t **managerp) { + isc_taskmgr_t *manager; + + /* + * Create a new task manager. + */ + + REQUIRE(managerp != NULL && *managerp == NULL); + REQUIRE(nm != NULL); + + manager = isc_mem_get(mctx, sizeof(*manager)); + *manager = (isc_taskmgr_t){ .magic = TASK_MANAGER_MAGIC }; + + isc_mutex_init(&manager->lock); + + if (default_quantum == 0) { + default_quantum = DEFAULT_DEFAULT_QUANTUM; + } + manager->default_quantum = default_quantum; + + if (nm != NULL) { + isc_nm_attach(nm, &manager->netmgr); + } + + INIT_LIST(manager->tasks); + atomic_init(&manager->mode, isc_taskmgrmode_normal); + atomic_init(&manager->exclusive_req, false); + atomic_init(&manager->tasks_count, 0); + + isc_mem_attach(mctx, &manager->mctx); + + isc_refcount_init(&manager->references, 1); + + *managerp = manager; + + return (ISC_R_SUCCESS); +} + +void +isc__taskmgr_shutdown(isc_taskmgr_t *manager) { + isc_task_t *task; + + REQUIRE(VALID_MANAGER(manager)); + + XTHREADTRACE("isc_taskmgr_shutdown"); + /* + * Only one non-worker thread may ever call this routine. + * If a worker thread wants to initiate shutdown of the + * task manager, it should ask some non-worker thread to call + * isc_taskmgr_destroy(), e.g. by signalling a condition variable + * that the startup thread is sleeping on. + */ + + /* + * Unlike elsewhere, we're going to hold this lock a long time. + * We need to do so, because otherwise the list of tasks could + * change while we were traversing it. + * + * This is also the only function where we will hold both the + * task manager lock and a task lock at the same time. + */ + LOCK(&manager->lock); + if (manager->excl != NULL) { + isc_task_detach((isc_task_t **)&manager->excl); + } + + /* + * Make sure we only get called once. + */ + INSIST(manager->exiting == false); + manager->exiting = true; + + /* + * Post shutdown event(s) to every task (if they haven't already been + * posted). + */ + for (task = HEAD(manager->tasks); task != NULL; task = NEXT(task, link)) + { + bool was_idle; + + LOCK(&task->lock); + was_idle = task_shutdown(task); + if (was_idle) { + task->threadid = 0; + } + UNLOCK(&task->lock); + + if (was_idle) { + task_ready(task); + } + } + + UNLOCK(&manager->lock); +} + +void +isc__taskmgr_destroy(isc_taskmgr_t **managerp) { + REQUIRE(managerp != NULL && VALID_MANAGER(*managerp)); + XTHREADTRACE("isc_taskmgr_destroy"); + +#ifdef ISC_TASK_TRACE + int counter = 0; + while (isc_refcount_current(&(*managerp)->references) > 1 && + counter++ < 1000) + { + usleep(10 * 1000); + } + INSIST(counter < 1000); +#else + while (isc_refcount_current(&(*managerp)->references) > 1) { + usleep(10 * 1000); + } +#endif + + isc_taskmgr_detach(managerp); +} + +void +isc_taskmgr_setexcltask(isc_taskmgr_t *mgr, isc_task_t *task) { + REQUIRE(VALID_MANAGER(mgr)); + REQUIRE(VALID_TASK(task)); + + LOCK(&task->lock); + REQUIRE(task->threadid == 0); + UNLOCK(&task->lock); + + LOCK(&mgr->lock); + if (mgr->excl != NULL) { + isc_task_detach(&mgr->excl); + } + isc_task_attach(task, &mgr->excl); + UNLOCK(&mgr->lock); +} + +isc_result_t +isc_taskmgr_excltask(isc_taskmgr_t *mgr, isc_task_t **taskp) { + isc_result_t result; + + REQUIRE(VALID_MANAGER(mgr)); + REQUIRE(taskp != NULL && *taskp == NULL); + + LOCK(&mgr->lock); + if (mgr->excl != NULL) { + isc_task_attach(mgr->excl, taskp); + result = ISC_R_SUCCESS; + } else if (mgr->exiting) { + result = ISC_R_SHUTTINGDOWN; + } else { + result = ISC_R_NOTFOUND; + } + UNLOCK(&mgr->lock); + + return (result); +} + +isc_result_t +isc_task_beginexclusive(isc_task_t *task) { + isc_taskmgr_t *manager; + + REQUIRE(VALID_TASK(task)); + + manager = task->manager; + + REQUIRE(task->state == task_state_running); + + LOCK(&manager->lock); + REQUIRE(task == manager->excl || + (manager->exiting && manager->excl == NULL)); + UNLOCK(&manager->lock); + + if (!atomic_compare_exchange_strong(&manager->exclusive_req, + &(bool){ false }, true)) + { + return (ISC_R_LOCKBUSY); + } + + if (isc_log_wouldlog(isc_lctx, ISC_LOG_DEBUG(1))) { + isc_log_write(isc_lctx, ISC_LOGCATEGORY_GENERAL, + ISC_LOGMODULE_OTHER, ISC_LOG_DEBUG(1), + "exclusive task mode: %s", "starting"); + } + + isc_nm_pause(manager->netmgr); + + if (isc_log_wouldlog(isc_lctx, ISC_LOG_DEBUG(1))) { + isc_log_write(isc_lctx, ISC_LOGCATEGORY_GENERAL, + ISC_LOGMODULE_OTHER, ISC_LOG_DEBUG(1), + "exclusive task mode: %s", "started"); + } + + return (ISC_R_SUCCESS); +} + +void +isc_task_endexclusive(isc_task_t *task) { + isc_taskmgr_t *manager; + + REQUIRE(VALID_TASK(task)); + REQUIRE(task->state == task_state_running); + manager = task->manager; + + if (isc_log_wouldlog(isc_lctx, ISC_LOG_DEBUG(1))) { + isc_log_write(isc_lctx, ISC_LOGCATEGORY_GENERAL, + ISC_LOGMODULE_OTHER, ISC_LOG_DEBUG(1), + "exclusive task mode: %s", "ending"); + } + + isc_nm_resume(manager->netmgr); + + if (isc_log_wouldlog(isc_lctx, ISC_LOG_DEBUG(1))) { + isc_log_write(isc_lctx, ISC_LOGCATEGORY_GENERAL, + ISC_LOGMODULE_OTHER, ISC_LOG_DEBUG(1), + "exclusive task mode: %s", "ended"); + } + + REQUIRE(atomic_compare_exchange_strong(&manager->exclusive_req, + &(bool){ true }, false)); +} + +void +isc_task_pause(isc_task_t *task) { + REQUIRE(VALID_TASK(task)); + + LOCK(&task->lock); + task->pause_cnt++; + if (task->pause_cnt > 1) { + /* + * Someone already paused this task, just increase + * the number of pausing clients. + */ + UNLOCK(&task->lock); + return; + } + + INSIST(task->state == task_state_idle || + task->state == task_state_ready || + task->state == task_state_running); + if (task->state == task_state_running) { + task->state = task_state_pausing; + } else { + task->state = task_state_paused; + } + UNLOCK(&task->lock); +} + +void +isc_task_unpause(isc_task_t *task) { + bool was_idle = false; + + REQUIRE(VALID_TASK(task)); + + LOCK(&task->lock); + task->pause_cnt--; + INSIST(task->pause_cnt >= 0); + if (task->pause_cnt > 0) { + UNLOCK(&task->lock); + return; + } + + INSIST(task->state == task_state_paused || + task->state == task_state_pausing); + /* If the task was pausing we can't reschedule it */ + if (task->state == task_state_pausing) { + task->state = task_state_running; + } else { + task->state = task_state_idle; + } + if (task->state == task_state_idle && !EMPTY(task->events)) { + task->state = task_state_ready; + was_idle = true; + } + UNLOCK(&task->lock); + + if (was_idle) { + task_ready(task); + } +} + +void +isc_taskmgr_setmode(isc_taskmgr_t *manager, isc_taskmgrmode_t mode) { + atomic_store(&manager->mode, mode); +} + +isc_taskmgrmode_t +isc_taskmgr_mode(isc_taskmgr_t *manager) { + return (atomic_load(&manager->mode)); +} + +void +isc_task_setprivilege(isc_task_t *task, bool priv) { + REQUIRE(VALID_TASK(task)); + + atomic_store_release(&task->privileged, priv); +} + +bool +isc_task_getprivilege(isc_task_t *task) { + REQUIRE(VALID_TASK(task)); + + return (TASK_PRIVILEGED(task)); +} + +bool +isc_task_privileged(isc_task_t *task) { + REQUIRE(VALID_TASK(task)); + + return (isc_taskmgr_mode(task->manager) && TASK_PRIVILEGED(task)); +} + +bool +isc_task_exiting(isc_task_t *task) { + REQUIRE(VALID_TASK(task)); + + return (TASK_SHUTTINGDOWN(task)); +} + +#ifdef HAVE_LIBXML2 +#define TRY0(a) \ + do { \ + xmlrc = (a); \ + if (xmlrc < 0) \ + goto error; \ + } while (0) +int +isc_taskmgr_renderxml(isc_taskmgr_t *mgr, void *writer0) { + isc_task_t *task = NULL; + int xmlrc; + xmlTextWriterPtr writer = (xmlTextWriterPtr)writer0; + + LOCK(&mgr->lock); + + /* + * Write out the thread-model, and some details about each depending + * on which type is enabled. + */ + TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "thread-model")); + TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "type")); + TRY0(xmlTextWriterWriteString(writer, ISC_XMLCHAR "threaded")); + TRY0(xmlTextWriterEndElement(writer)); /* type */ + + TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "default-quantum")); + TRY0(xmlTextWriterWriteFormatString(writer, "%d", + mgr->default_quantum)); + TRY0(xmlTextWriterEndElement(writer)); /* default-quantum */ + + TRY0(xmlTextWriterEndElement(writer)); /* thread-model */ + + TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "tasks")); + task = ISC_LIST_HEAD(mgr->tasks); + while (task != NULL) { + LOCK(&task->lock); + TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "task")); + + if (task->name[0] != 0) { + TRY0(xmlTextWriterStartElement(writer, + ISC_XMLCHAR "name")); + TRY0(xmlTextWriterWriteFormatString(writer, "%s", + task->name)); + TRY0(xmlTextWriterEndElement(writer)); /* name */ + } + + TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "reference" + "s")); + TRY0(xmlTextWriterWriteFormatString( + writer, "%" PRIuFAST32, + isc_refcount_current(&task->references))); + TRY0(xmlTextWriterEndElement(writer)); /* references */ + + TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "id")); + TRY0(xmlTextWriterWriteFormatString(writer, "%p", task)); + TRY0(xmlTextWriterEndElement(writer)); /* id */ + + TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "state")); + TRY0(xmlTextWriterWriteFormatString(writer, "%s", + statenames[task->state])); + TRY0(xmlTextWriterEndElement(writer)); /* state */ + + TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "quantum")); + TRY0(xmlTextWriterWriteFormatString(writer, "%d", + task->quantum)); + TRY0(xmlTextWriterEndElement(writer)); /* quantum */ + + TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "events")); + TRY0(xmlTextWriterWriteFormatString(writer, "%d", + task->nevents)); + TRY0(xmlTextWriterEndElement(writer)); /* events */ + + TRY0(xmlTextWriterEndElement(writer)); + + UNLOCK(&task->lock); + task = ISC_LIST_NEXT(task, link); + } + TRY0(xmlTextWriterEndElement(writer)); /* tasks */ + +error: + if (task != NULL) { + UNLOCK(&task->lock); + } + UNLOCK(&mgr->lock); + + return (xmlrc); +} +#endif /* HAVE_LIBXML2 */ + +#ifdef HAVE_JSON_C +#define CHECKMEM(m) \ + do { \ + if (m == NULL) { \ + result = ISC_R_NOMEMORY; \ + goto error; \ + } \ + } while (0) + +isc_result_t +isc_taskmgr_renderjson(isc_taskmgr_t *mgr, void *tasks0) { + isc_result_t result = ISC_R_SUCCESS; + isc_task_t *task = NULL; + json_object *obj = NULL, *array = NULL, *taskobj = NULL; + json_object *tasks = (json_object *)tasks0; + + LOCK(&mgr->lock); + + /* + * Write out the thread-model, and some details about each depending + * on which type is enabled. + */ + obj = json_object_new_string("threaded"); + CHECKMEM(obj); + json_object_object_add(tasks, "thread-model", obj); + + obj = json_object_new_int(mgr->default_quantum); + CHECKMEM(obj); + json_object_object_add(tasks, "default-quantum", obj); + + array = json_object_new_array(); + CHECKMEM(array); + + for (task = ISC_LIST_HEAD(mgr->tasks); task != NULL; + task = ISC_LIST_NEXT(task, link)) + { + char buf[255]; + + LOCK(&task->lock); + + taskobj = json_object_new_object(); + CHECKMEM(taskobj); + json_object_array_add(array, taskobj); + + snprintf(buf, sizeof(buf), "%p", task); + obj = json_object_new_string(buf); + CHECKMEM(obj); + json_object_object_add(taskobj, "id", obj); + + if (task->name[0] != 0) { + obj = json_object_new_string(task->name); + CHECKMEM(obj); + json_object_object_add(taskobj, "name", obj); + } + + obj = json_object_new_int( + isc_refcount_current(&task->references)); + CHECKMEM(obj); + json_object_object_add(taskobj, "references", obj); + + obj = json_object_new_string(statenames[task->state]); + CHECKMEM(obj); + json_object_object_add(taskobj, "state", obj); + + obj = json_object_new_int(task->quantum); + CHECKMEM(obj); + json_object_object_add(taskobj, "quantum", obj); + + obj = json_object_new_int(task->nevents); + CHECKMEM(obj); + json_object_object_add(taskobj, "events", obj); + + UNLOCK(&task->lock); + } + + json_object_object_add(tasks, "tasks", array); + array = NULL; + result = ISC_R_SUCCESS; + +error: + if (array != NULL) { + json_object_put(array); + } + + if (task != NULL) { + UNLOCK(&task->lock); + } + UNLOCK(&mgr->lock); + + return (result); +} +#endif /* ifdef HAVE_JSON_C */ |