static void
async_cb(uv_async_t *handle);
static bool
-process_queue(isc__networker_t *worker, isc_queue_t *queue);
+process_queue(isc__networker_t *worker, isc_queue_t *queue,
+ unsigned int *quantump);
static bool
-process_priority_queue(isc__networker_t *worker);
-static void
-process_privilege_queue(isc__networker_t *worker);
-static void
-process_tasks_queue(isc__networker_t *worker);
-static void
-process_normal_queue(isc__networker_t *worker);
+process_priority_queue(isc__networker_t *worker, unsigned int *quantump);
+static bool
+process_privilege_queue(isc__networker_t *worker, unsigned int *quantump);
+static bool
+process_task_queue(isc__networker_t *worker, unsigned int *quantump);
+static bool
+process_normal_queue(isc__networker_t *worker, unsigned int *quantump);
static void
isc__nm_async_stop(isc__networker_t *worker, isc__netievent_t *ev0);
*worker = (isc__networker_t){
.mgr = mgr,
.id = i,
+ .quantum = ISC_NETMGR_QUANTUM_DEFAULT,
};
r = uv_loop_init(&worker->loop);
isc_condition_init(&worker->cond);
worker->ievents = isc_queue_new(mgr->mctx, 128);
- worker->ievents_priv = isc_queue_new(mgr->mctx, 128);
worker->ievents_task = isc_queue_new(mgr->mctx, 128);
+ worker->ievents_priv = isc_queue_new(mgr->mctx, 128);
worker->ievents_prio = isc_queue_new(mgr->mctx, 128);
worker->recvbuf = isc_mem_get(mctx, ISC_NETMGR_RECVBUF_SIZE);
worker->sendbuf = isc_mem_get(mctx, ISC_NETMGR_SENDBUF_SIZE);
while (worker->paused) {
WAIT(&worker->cond, &worker->lock);
UNLOCK(&worker->lock);
- (void)process_priority_queue(worker);
+ (void)process_priority_queue(
+ worker, &(unsigned int){ UINT_MAX });
LOCK(&worker->lock);
}
UNLOCK(&worker->lock);
/*
- * All workers must run the privileged event
+ * All workers must drain the privileged event
* queue before we resume from pause.
*/
- process_privilege_queue(worker);
+ (void)process_privilege_queue(
+ worker, &(unsigned int){ UINT_MAX });
LOCK(&mgr->lock);
while (atomic_load(&mgr->paused)) {
}
INSIST(!worker->finished);
-
- /*
- * We've fully resumed from pause. Drain the normal
- * asynchronous event queues before resuming the uv_run()
- * loop. (This is not strictly necessary, it just ensures
- * that all pending events are processed before another
- * pause can slip in.)
- */
- process_tasks_queue(worker);
- process_normal_queue(worker);
}
/*
* (they may include shutdown events) but do not process
* the netmgr event queue.
*/
- process_privilege_queue(worker);
- process_tasks_queue(worker);
+ (void)process_privilege_queue(worker, &(unsigned int){ UINT_MAX });
+ (void)process_task_queue(worker, &(unsigned int){ UINT_MAX });
LOCK(&mgr->lock);
mgr->workers_running--;
return ((isc_threadresult_t)0);
}
+static bool
+process_all_queues(isc__networker_t *worker, unsigned int quantum) {
+ /*
+ * The queue processing functions will return false when the
+ * system is pausing or stopping, or if we have completed
+ * 'quantum' events.
+ *
+ * We don't want to proceed to a new queue until the previous one
+ * has been fully drained, so whenever one queue is interrupted,
+ * we skip all the later ones.
+ */
+ return (process_priority_queue(worker, &quantum) &&
+ process_privilege_queue(worker, &quantum) &&
+ process_task_queue(worker, &quantum) &&
+ process_normal_queue(worker, &quantum));
+}
+
/*
* async_cb() is a universal callback for 'async' events sent to event loop.
* It's the only way to safely pass data to the libuv event loop. We use a
static void
async_cb(uv_async_t *handle) {
isc__networker_t *worker = (isc__networker_t *)handle->loop->data;
+ unsigned int quantum = worker->quantum;
- /*
- * process_priority_queue() returns false when pausing or stopping,
- * so we don't want to process the other queues in that case.
- */
- if (!process_priority_queue(worker)) {
- return;
+ if (!process_all_queues(worker, quantum)) {
+ /* If we didn't process all the events, we need to enqueue
+ * async_cb to be run in the next iteration of the uv_loop
+ */
+ uv_async_send(handle);
}
-
- process_privilege_queue(worker);
- process_tasks_queue(worker);
- process_normal_queue(worker);
}
static void
switch (result) {
case ISC_R_QUOTA:
- isc_nm_task_enqueue(worker->mgr, (isc_task_t *)ievent->task,
- isc_nm_tid());
+ isc_task_ready(ievent->task);
return;
case ISC_R_SUCCESS:
return;
}
static bool
-process_priority_queue(isc__networker_t *worker) {
- return (process_queue(worker, worker->ievents_prio));
+process_priority_queue(isc__networker_t *worker, unsigned int *quantump) {
+ return (process_queue(worker, worker->ievents_prio, quantump));
}
-static void
-process_privilege_queue(isc__networker_t *worker) {
- (void)process_queue(worker, worker->ievents_priv);
+static bool
+process_privilege_queue(isc__networker_t *worker, unsigned int *quantump) {
+ return (process_queue(worker, worker->ievents_priv, quantump));
}
-static void
-process_tasks_queue(isc__networker_t *worker) {
- (void)process_queue(worker, worker->ievents_task);
+static bool
+process_task_queue(isc__networker_t *worker, unsigned int *quantump) {
+ return (process_queue(worker, worker->ievents_task, quantump));
}
-static void
-process_normal_queue(isc__networker_t *worker) {
- (void)process_queue(worker, worker->ievents);
+static bool
+process_normal_queue(isc__networker_t *worker, unsigned int *quantump) {
+ return (process_queue(worker, worker->ievents, quantump));
}
/*
}
static bool
-process_queue(isc__networker_t *worker, isc_queue_t *queue) {
- isc__netievent_t *ievent = NULL;
+process_queue(isc__networker_t *worker, isc_queue_t *queue,
+ unsigned int *quantump) {
+ while (*quantump > 0) {
+ isc__netievent_t *ievent =
+ (isc__netievent_t *)isc_queue_dequeue(queue);
+
+ (*quantump)--;
+
+ if (ievent == NULL) {
+ /* We fully drained this queue */
+ return (true);
+ }
- while ((ievent = (isc__netievent_t *)isc_queue_dequeue(queue)) != NULL)
- {
if (!process_netievent(worker, ievent)) {
+ /* Netievent told us to stop */
return (false);
}
}
- return (true);
+
+ /* No more quantum */
+ return (false);
}
void *
*/
#include <stdbool.h>
+#include <unistd.h>
#include <isc/app.h>
#include <isc/atomic.h>
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;
void *tag;
bool bound;
/* Protected by atomics */
- atomic_uint_fast32_t flags;
+ atomic_bool shuttingdown;
+ atomic_bool privileged;
/* Locked by task manager lock. */
LINK(isc_task_t) link;
};
-#define TASK_F_SHUTTINGDOWN 0x01
-#define TASK_F_PRIVILEGED 0x02
-
-#define TASK_SHUTTINGDOWN(t) \
- ((atomic_load_acquire(&(t)->flags) & TASK_F_SHUTTINGDOWN) != 0)
-#define TASK_PRIVILEGED(t) \
- ((atomic_load_acquire(&(t)->flags) & TASK_F_PRIVILEGED) != 0)
-
-#define TASK_FLAG_SET(t, f) atomic_fetch_or_release(&(t)->flags, (f))
-#define TASK_FLAG_CLR(t, f) atomic_fetch_and_release(&(t)->flags, ~(f))
+#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)
isc_refcount_t references;
isc_mem_t *mctx;
isc_mutex_t lock;
- atomic_uint_fast32_t tasks_running;
- atomic_uint_fast32_t tasks_ready;
atomic_uint_fast32_t tasks_count;
- isc_nm_t *nm;
+ isc_nm_t *netmgr;
/* Locked by task manager lock. */
unsigned int default_quantum;
atomic_bool exclusive_req;
atomic_bool exiting;
- /* Locked by halt_lock */
- unsigned int halted;
-
/*
* Multiple threads can read/write 'excl' at the same time, so we need
* to protect the access. We can't use 'lock' since isc_task_detach()
};
#define DEFAULT_DEFAULT_QUANTUM 25
-#define FINISHED(m) \
- (atomic_load_relaxed(&((m)->exiting)) && \
- atomic_load(&(m)->tasks_count) == 0)
/*%
* The following are intended for internal use (indicated by "isc__"
XTRACE("task_finished");
+ isc_refcount_destroy(&task->running);
isc_refcount_destroy(&task->references);
LOCK(&manager->lock);
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->flags, 0);
+ 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));
XTRACE("task_shutdown");
- if (!TASK_SHUTTINGDOWN(task)) {
+ if (atomic_compare_exchange_strong(&task->shuttingdown,
+ &(bool){ false }, true)) {
XTRACE("shutting down");
- TASK_FLAG_SET(task, TASK_F_SHUTTINGDOWN);
if (task->state == task_state_idle) {
INSIST(EMPTY(task->events));
task->state = task_state_ready;
REQUIRE(VALID_MANAGER(manager));
XTRACE("task_ready");
- isc_nm_task_enqueue(manager->nm, task, task->threadid);
+
+ isc_refcount_increment0(&task->running);
+ isc_nm_task_enqueue(manager->netmgr, task, task->threadid);
+}
+
+void
+isc_task_ready(isc_task_t *task) {
+ task_ready(task);
}
static inline bool
* and task lock to avoid deadlocks, just bail then.
*/
if (task->state != task_state_ready) {
- UNLOCK(&task->lock);
- return (ISC_R_SUCCESS);
+ goto done;
}
INSIST(task->state == task_state_ready);
* The task is done.
*/
XTRACE("done");
- finished = true;
task->state = task_state_done;
} else {
if (task->state == task_state_running) {
break;
}
}
+
+done:
+ if (isc_refcount_decrement(&task->running) == 1 &&
+ task->state == task_state_done)
+ {
+ finished = true;
+ }
UNLOCK(&task->lock);
if (finished) {
static void
manager_free(isc_taskmgr_t *manager) {
isc_refcount_destroy(&manager->references);
- isc_nm_detach(&manager->nm);
+ isc_nm_detach(&manager->netmgr);
isc_mutex_destroy(&manager->lock);
isc_mutex_destroy(&manager->excl_lock);
manager->default_quantum = default_quantum;
if (nm != NULL) {
- isc_nm_attach(nm, &manager->nm);
+ isc_nm_attach(nm, &manager->netmgr);
}
INIT_LIST(manager->tasks);
- atomic_init(&manager->tasks_count, 0);
- atomic_init(&manager->tasks_running, 0);
- atomic_init(&manager->tasks_ready, 0);
atomic_init(&manager->exiting, false);
atomic_store_relaxed(&manager->exclusive_req, false);
XTHREADTRACE("isc_taskmgr_destroy");
+#ifdef ISC_TASK_TRACE
+ int counter = 0;
+ while (isc_refcount_current(&manager->references) > 1 &&
+ counter++ < 1000) {
+ usleep(10 * 1000);
+ }
+#else
+ while (isc_refcount_current(&manager->references) > 1) {
+ usleep(10 * 1000);
+ }
+#endif
+
REQUIRE(isc_refcount_decrement(&manager->references) == 1);
manager_free(manager);
}
return (ISC_R_LOCKBUSY);
}
- isc_nm_pause(manager->nm);
+ isc_nm_pause(manager->netmgr);
return (ISC_R_SUCCESS);
}
REQUIRE(task->state == task_state_running);
manager = task->manager;
- isc_nm_resume(manager->nm);
+ isc_nm_resume(manager->netmgr);
REQUIRE(atomic_compare_exchange_strong(&manager->exclusive_req,
&(bool){ true }, false));
}
void
isc_task_setprivilege(isc_task_t *task, bool priv) {
REQUIRE(VALID_TASK(task));
- uint_fast32_t oldflags, newflags;
-
- oldflags = atomic_load_acquire(&task->flags);
- do {
- if (priv) {
- newflags = oldflags | TASK_F_PRIVILEGED;
- } else {
- newflags = oldflags & ~TASK_F_PRIVILEGED;
- }
- if (newflags == oldflags) {
- return;
- }
- } while (!atomic_compare_exchange_weak_acq_rel(&task->flags, &oldflags,
- newflags));
+
+ atomic_store_release(&task->privileged, priv);
}
bool
mgr->default_quantum));
TRY0(xmlTextWriterEndElement(writer)); /* default-quantum */
- TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "tasks-count"));
- TRY0(xmlTextWriterWriteFormatString(
- writer, "%d", (int)atomic_load_relaxed(&mgr->tasks_count)));
- TRY0(xmlTextWriterEndElement(writer)); /* tasks-count */
-
- TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "tasks-running"));
- TRY0(xmlTextWriterWriteFormatString(
- writer, "%d", (int)atomic_load_relaxed(&mgr->tasks_running)));
- TRY0(xmlTextWriterEndElement(writer)); /* tasks-running */
-
- TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "tasks-ready"));
- TRY0(xmlTextWriterWriteFormatString(
- writer, "%d", (int)atomic_load_relaxed(&mgr->tasks_ready)));
- TRY0(xmlTextWriterEndElement(writer)); /* tasks-ready */
-
TRY0(xmlTextWriterEndElement(writer)); /* thread-model */
TRY0(xmlTextWriterStartElement(writer, ISC_XMLCHAR "tasks"));
CHECKMEM(obj);
json_object_object_add(tasks, "default-quantum", obj);
- obj = json_object_new_int(atomic_load_relaxed(&mgr->tasks_count));
- CHECKMEM(obj);
- json_object_object_add(tasks, "tasks-count", obj);
-
- obj = json_object_new_int(atomic_load_relaxed(&mgr->tasks_running));
- CHECKMEM(obj);
- json_object_object_add(tasks, "tasks-running", obj);
-
- obj = json_object_new_int(atomic_load_relaxed(&mgr->tasks_ready));
- CHECKMEM(obj);
- json_object_object_add(tasks, "tasks-ready", obj);
-
array = json_object_new_array();
CHECKMEM(array);
projects / thirdparty / bind9.git / commitdiff
