diff options
Diffstat (limited to '')
| -rw-r--r-- | src/task.c | 1044 |
1 files changed, 1044 insertions, 0 deletions
diff --git a/src/task.c b/src/task.c new file mode 100644 index 0000000..a926f4c --- /dev/null +++ b/src/task.c @@ -0,0 +1,1044 @@ +/* + * Task management functions. + * + * Copyright 2000-2009 Willy Tarreau <w@1wt.eu> + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public License + * as published by the Free Software Foundation; either version + * 2 of the License, or (at your option) any later version. + * + */ + +#include <string.h> + +#include <import/eb32sctree.h> +#include <import/eb32tree.h> + +#include <haproxy/api.h> +#include <haproxy/activity.h> +#include <haproxy/cfgparse.h> +#include <haproxy/clock.h> +#include <haproxy/fd.h> +#include <haproxy/list.h> +#include <haproxy/pool.h> +#include <haproxy/task.h> +#include <haproxy/tools.h> + +extern struct task *process_stream(struct task *t, void *context, unsigned int state); + +DECLARE_POOL(pool_head_task, "task", sizeof(struct task)); +DECLARE_POOL(pool_head_tasklet, "tasklet", sizeof(struct tasklet)); + +/* This is the memory pool containing all the signal structs. These + * struct are used to store each required signal between two tasks. + */ +DECLARE_POOL(pool_head_notification, "notification", sizeof(struct notification)); + +volatile unsigned long global_tasks_mask = 0; /* Mask of threads with tasks in the global runqueue */ +unsigned int niced_tasks = 0; /* number of niced tasks in the run queue */ + +__decl_aligned_spinlock(rq_lock); /* spin lock related to run queue */ +__decl_aligned_rwlock(wq_lock); /* RW lock related to the wait queue */ + +#ifdef USE_THREAD +struct eb_root timers; /* sorted timers tree, global, accessed under wq_lock */ +struct eb_root rqueue; /* tree constituting the global run queue, accessed under rq_lock */ +unsigned int grq_total; /* total number of entries in the global run queue, atomic */ +static unsigned int global_rqueue_ticks; /* insertion count in the grq, use rq_lock */ +#endif + + + +/* Flags the task <t> for immediate destruction and puts it into its first + * thread's shared tasklet list if not yet queued/running. This will bypass + * the priority scheduling and make the task show up as fast as possible in + * the other thread's queue. Note that this operation isn't idempotent and is + * not supposed to be run on the same task from multiple threads at once. It's + * the caller's responsibility to make sure it is the only one able to kill the + * task. + */ +void task_kill(struct task *t) +{ + unsigned int state = t->state; + unsigned int thr; + + BUG_ON(state & TASK_KILLED); + + while (1) { + while (state & (TASK_RUNNING | TASK_QUEUED)) { + /* task already in the queue and about to be executed, + * or even currently running. Just add the flag and be + * done with it, the process loop will detect it and kill + * it. The CAS will fail if we arrive too late. + */ + if (_HA_ATOMIC_CAS(&t->state, &state, state | TASK_KILLED)) + return; + } + + /* We'll have to wake it up, but we must also secure it so that + * it doesn't vanish under us. TASK_QUEUED guarantees nobody will + * add past us. + */ + if (_HA_ATOMIC_CAS(&t->state, &state, state | TASK_QUEUED | TASK_KILLED)) { + /* Bypass the tree and go directly into the shared tasklet list. + * Note: that's a task so it must be accounted for as such. Pick + * the task's first thread for the job. + */ + thr = my_ffsl(t->thread_mask) - 1; + + /* Beware: tasks that have never run don't have their ->list empty yet! */ + MT_LIST_APPEND(&ha_thread_ctx[thr].shared_tasklet_list, + list_to_mt_list(&((struct tasklet *)t)->list)); + _HA_ATOMIC_INC(&ha_thread_ctx[thr].rq_total); + _HA_ATOMIC_INC(&ha_thread_ctx[thr].tasks_in_list); + if (sleeping_thread_mask & (1UL << thr)) { + _HA_ATOMIC_AND(&sleeping_thread_mask, ~(1UL << thr)); + wake_thread(thr); + } + return; + } + } +} + +/* Equivalent of task_kill for tasklets. Mark the tasklet <t> for destruction. + * It will be deleted on the next scheduler invocation. This function is + * thread-safe : a thread can kill a tasklet of another thread. + */ +void tasklet_kill(struct tasklet *t) +{ + unsigned int state = t->state; + unsigned int thr; + + BUG_ON(state & TASK_KILLED); + + while (1) { + while (state & (TASK_IN_LIST)) { + /* Tasklet already in the list ready to be executed. Add + * the killed flag and wait for the process loop to + * detect it. + */ + if (_HA_ATOMIC_CAS(&t->state, &state, state | TASK_KILLED)) + return; + } + + /* Mark the tasklet as killed and wake the thread to process it + * as soon as possible. + */ + if (_HA_ATOMIC_CAS(&t->state, &state, state | TASK_IN_LIST | TASK_KILLED)) { + thr = t->tid >= 0 ? t->tid : tid; + MT_LIST_APPEND(&ha_thread_ctx[thr].shared_tasklet_list, + list_to_mt_list(&t->list)); + _HA_ATOMIC_INC(&ha_thread_ctx[thr].rq_total); + if (sleeping_thread_mask & (1UL << thr)) { + _HA_ATOMIC_AND(&sleeping_thread_mask, ~(1UL << thr)); + wake_thread(thr); + } + return; + } + } +} + +/* Do not call this one, please use tasklet_wakeup_on() instead, as this one is + * the slow path of tasklet_wakeup_on() which performs some preliminary checks + * and sets TASK_IN_LIST before calling this one. A negative <thr> designates + * the current thread. + */ +void __tasklet_wakeup_on(struct tasklet *tl, int thr) +{ + if (likely(thr < 0)) { + /* this tasklet runs on the caller thread */ + if (tl->state & TASK_HEAVY) { + LIST_APPEND(&th_ctx->tasklets[TL_HEAVY], &tl->list); + th_ctx->tl_class_mask |= 1 << TL_HEAVY; + } + else if (tl->state & TASK_SELF_WAKING) { + LIST_APPEND(&th_ctx->tasklets[TL_BULK], &tl->list); + th_ctx->tl_class_mask |= 1 << TL_BULK; + } + else if ((struct task *)tl == th_ctx->current) { + _HA_ATOMIC_OR(&tl->state, TASK_SELF_WAKING); + LIST_APPEND(&th_ctx->tasklets[TL_BULK], &tl->list); + th_ctx->tl_class_mask |= 1 << TL_BULK; + } + else if (th_ctx->current_queue < 0) { + LIST_APPEND(&th_ctx->tasklets[TL_URGENT], &tl->list); + th_ctx->tl_class_mask |= 1 << TL_URGENT; + } + else { + LIST_APPEND(&th_ctx->tasklets[th_ctx->current_queue], &tl->list); + th_ctx->tl_class_mask |= 1 << th_ctx->current_queue; + } + _HA_ATOMIC_INC(&th_ctx->rq_total); + } else { + /* this tasklet runs on a specific thread. */ + MT_LIST_APPEND(&ha_thread_ctx[thr].shared_tasklet_list, list_to_mt_list(&tl->list)); + _HA_ATOMIC_INC(&ha_thread_ctx[thr].rq_total); + if (sleeping_thread_mask & (1UL << thr)) { + _HA_ATOMIC_AND(&sleeping_thread_mask, ~(1UL << thr)); + wake_thread(thr); + } + } +} + +/* Do not call this one, please use tasklet_wakeup_after_on() instead, as this one is + * the slow path of tasklet_wakeup_after() which performs some preliminary checks + * and sets TASK_IN_LIST before calling this one. + */ +struct list *__tasklet_wakeup_after(struct list *head, struct tasklet *tl) +{ + BUG_ON(tid != tl->tid); + /* this tasklet runs on the caller thread */ + if (!head) { + if (tl->state & TASK_HEAVY) { + LIST_INSERT(&th_ctx->tasklets[TL_HEAVY], &tl->list); + th_ctx->tl_class_mask |= 1 << TL_HEAVY; + } + else if (tl->state & TASK_SELF_WAKING) { + LIST_INSERT(&th_ctx->tasklets[TL_BULK], &tl->list); + th_ctx->tl_class_mask |= 1 << TL_BULK; + } + else if ((struct task *)tl == th_ctx->current) { + _HA_ATOMIC_OR(&tl->state, TASK_SELF_WAKING); + LIST_INSERT(&th_ctx->tasklets[TL_BULK], &tl->list); + th_ctx->tl_class_mask |= 1 << TL_BULK; + } + else if (th_ctx->current_queue < 0) { + LIST_INSERT(&th_ctx->tasklets[TL_URGENT], &tl->list); + th_ctx->tl_class_mask |= 1 << TL_URGENT; + } + else { + LIST_INSERT(&th_ctx->tasklets[th_ctx->current_queue], &tl->list); + th_ctx->tl_class_mask |= 1 << th_ctx->current_queue; + } + } + else { + LIST_APPEND(head, &tl->list); + } + _HA_ATOMIC_INC(&th_ctx->rq_total); + return &tl->list; +} + +/* Puts the task <t> in run queue at a position depending on t->nice. <t> is + * returned. The nice value assigns boosts in 32th of the run queue size. A + * nice value of -1024 sets the task to -tasks_run_queue*32, while a nice value + * of 1024 sets the task to tasks_run_queue*32. The state flags are cleared, so + * the caller will have to set its flags after this call. + * The task must not already be in the run queue. If unsure, use the safer + * task_wakeup() function. + */ +void __task_wakeup(struct task *t) +{ + struct eb_root *root = &th_ctx->rqueue; + +#ifdef USE_THREAD + if (t->thread_mask != tid_bit && global.nbthread != 1) { + root = &rqueue; + + _HA_ATOMIC_INC(&grq_total); + HA_SPIN_LOCK(TASK_RQ_LOCK, &rq_lock); + + global_tasks_mask |= t->thread_mask; + t->rq.key = ++global_rqueue_ticks; + __ha_barrier_store(); + } else +#endif + { + _HA_ATOMIC_INC(&th_ctx->rq_total); + t->rq.key = ++th_ctx->rqueue_ticks; + } + + if (likely(t->nice)) { + int offset; + + _HA_ATOMIC_INC(&niced_tasks); + offset = t->nice * (int)global.tune.runqueue_depth; + t->rq.key += offset; + } + + if (task_profiling_mask & tid_bit) + t->wake_date = now_mono_time(); + + eb32sc_insert(root, &t->rq, t->thread_mask); + +#ifdef USE_THREAD + if (root == &rqueue) { + _HA_ATOMIC_OR(&t->state, TASK_GLOBAL); + HA_SPIN_UNLOCK(TASK_RQ_LOCK, &rq_lock); + + /* If all threads that are supposed to handle this task are sleeping, + * wake one. + */ + if ((((t->thread_mask & all_threads_mask) & sleeping_thread_mask) == + (t->thread_mask & all_threads_mask))) { + unsigned long m = (t->thread_mask & all_threads_mask) &~ tid_bit; + + m = (m & (m - 1)) ^ m; // keep lowest bit set + _HA_ATOMIC_AND(&sleeping_thread_mask, ~m); + wake_thread(my_ffsl(m) - 1); + } + } +#endif + return; +} + +/* + * __task_queue() + * + * Inserts a task into wait queue <wq> at the position given by its expiration + * date. It does not matter if the task was already in the wait queue or not, + * as it will be unlinked. The task MUST NOT have an infinite expiration timer. + * Last, tasks must not be queued further than the end of the tree, which is + * between <now_ms> and <now_ms> + 2^31 ms (now+24days in 32bit). + * + * This function should not be used directly, it is meant to be called by the + * inline version of task_queue() which performs a few cheap preliminary tests + * before deciding to call __task_queue(). Moreover this function doesn't care + * at all about locking so the caller must be careful when deciding whether to + * lock or not around this call. + */ +void __task_queue(struct task *task, struct eb_root *wq) +{ +#ifdef USE_THREAD + BUG_ON((wq == &timers && !(task->state & TASK_SHARED_WQ)) || + (wq == &th_ctx->timers && (task->state & TASK_SHARED_WQ)) || + (wq != &timers && wq != &th_ctx->timers)); +#endif + /* if this happens the process is doomed anyway, so better catch it now + * so that we have the caller in the stack. + */ + BUG_ON(task->expire == TICK_ETERNITY); + + if (likely(task_in_wq(task))) + __task_unlink_wq(task); + + /* the task is not in the queue now */ + task->wq.key = task->expire; +#ifdef DEBUG_CHECK_INVALID_EXPIRATION_DATES + if (tick_is_lt(task->wq.key, now_ms)) + /* we're queuing too far away or in the past (most likely) */ + return; +#endif + + eb32_insert(wq, &task->wq); +} + +/* + * Extract all expired timers from the timer queue, and wakes up all + * associated tasks. + */ +void wake_expired_tasks() +{ + struct thread_ctx * const tt = th_ctx; // thread's tasks + int max_processed = global.tune.runqueue_depth; + struct task *task; + struct eb32_node *eb; + __decl_thread(int key); + + while (max_processed-- > 0) { + lookup_next_local: + eb = eb32_lookup_ge(&tt->timers, now_ms - TIMER_LOOK_BACK); + if (!eb) { + /* we might have reached the end of the tree, typically because + * <now_ms> is in the first half and we're first scanning the last + * half. Let's loop back to the beginning of the tree now. + */ + eb = eb32_first(&tt->timers); + if (likely(!eb)) + break; + } + + /* It is possible that this task was left at an earlier place in the + * tree because a recent call to task_queue() has not moved it. This + * happens when the new expiration date is later than the old one. + * Since it is very unlikely that we reach a timeout anyway, it's a + * lot cheaper to proceed like this because we almost never update + * the tree. We may also find disabled expiration dates there. Since + * we have detached the task from the tree, we simply call task_queue + * to take care of this. Note that we might occasionally requeue it at + * the same place, before <eb>, so we have to check if this happens, + * and adjust <eb>, otherwise we may skip it which is not what we want. + * We may also not requeue the task (and not point eb at it) if its + * expiration time is not set. We also make sure we leave the real + * expiration date for the next task in the queue so that when calling + * next_timer_expiry() we're guaranteed to see the next real date and + * not the next apparent date. This is in order to avoid useless + * wakeups. + */ + + task = eb32_entry(eb, struct task, wq); + if (tick_is_expired(task->expire, now_ms)) { + /* expired task, wake it up */ + __task_unlink_wq(task); + task_wakeup(task, TASK_WOKEN_TIMER); + } + else if (task->expire != eb->key) { + /* task is not expired but its key doesn't match so let's + * update it and skip to next apparently expired task. + */ + __task_unlink_wq(task); + if (tick_isset(task->expire)) + __task_queue(task, &tt->timers); + } + else { + /* task not expired and correctly placed. It may not be eternal. */ + BUG_ON(task->expire == TICK_ETERNITY); + break; + } + } + +#ifdef USE_THREAD + if (eb_is_empty(&timers)) + goto leave; + + HA_RWLOCK_RDLOCK(TASK_WQ_LOCK, &wq_lock); + eb = eb32_lookup_ge(&timers, now_ms - TIMER_LOOK_BACK); + if (!eb) { + eb = eb32_first(&timers); + if (likely(!eb)) { + HA_RWLOCK_RDUNLOCK(TASK_WQ_LOCK, &wq_lock); + goto leave; + } + } + key = eb->key; + + if (tick_is_lt(now_ms, key)) { + HA_RWLOCK_RDUNLOCK(TASK_WQ_LOCK, &wq_lock); + goto leave; + } + + /* There's really something of interest here, let's visit the queue */ + + if (HA_RWLOCK_TRYRDTOSK(TASK_WQ_LOCK, &wq_lock)) { + /* if we failed to grab the lock it means another thread is + * already doing the same here, so let it do the job. + */ + HA_RWLOCK_RDUNLOCK(TASK_WQ_LOCK, &wq_lock); + goto leave; + } + + while (1) { + lookup_next: + if (max_processed-- <= 0) + break; + eb = eb32_lookup_ge(&timers, now_ms - TIMER_LOOK_BACK); + if (!eb) { + /* we might have reached the end of the tree, typically because + * <now_ms> is in the first half and we're first scanning the last + * half. Let's loop back to the beginning of the tree now. + */ + eb = eb32_first(&timers); + if (likely(!eb)) + break; + } + + task = eb32_entry(eb, struct task, wq); + + /* Check for any competing run of the task (quite rare but may + * involve a dangerous concurrent access on task->expire). In + * order to protect against this, we'll take an exclusive access + * on TASK_RUNNING before checking/touching task->expire. If the + * task is already RUNNING on another thread, it will deal by + * itself with the requeuing so we must not do anything and + * simply quit the loop for now, because we cannot wait with the + * WQ lock held as this would prevent the running thread from + * requeuing the task. One annoying effect of holding RUNNING + * here is that a concurrent task_wakeup() will refrain from + * waking it up. This forces us to check for a wakeup after + * releasing the flag. + */ + if (HA_ATOMIC_FETCH_OR(&task->state, TASK_RUNNING) & TASK_RUNNING) + break; + + if (tick_is_expired(task->expire, now_ms)) { + /* expired task, wake it up */ + HA_RWLOCK_SKTOWR(TASK_WQ_LOCK, &wq_lock); + __task_unlink_wq(task); + HA_RWLOCK_WRTOSK(TASK_WQ_LOCK, &wq_lock); + task_drop_running(task, TASK_WOKEN_TIMER); + } + else if (task->expire != eb->key) { + /* task is not expired but its key doesn't match so let's + * update it and skip to next apparently expired task. + */ + HA_RWLOCK_SKTOWR(TASK_WQ_LOCK, &wq_lock); + __task_unlink_wq(task); + if (tick_isset(task->expire)) + __task_queue(task, &timers); + HA_RWLOCK_WRTOSK(TASK_WQ_LOCK, &wq_lock); + task_drop_running(task, 0); + goto lookup_next; + } + else { + /* task not expired and correctly placed. It may not be eternal. */ + BUG_ON(task->expire == TICK_ETERNITY); + task_drop_running(task, 0); + break; + } + } + + HA_RWLOCK_SKUNLOCK(TASK_WQ_LOCK, &wq_lock); +#endif +leave: + return; +} + +/* Checks the next timer for the current thread by looking into its own timer + * list and the global one. It may return TICK_ETERNITY if no timer is present. + * Note that the next timer might very well be slightly in the past. + */ +int next_timer_expiry() +{ + struct thread_ctx * const tt = th_ctx; // thread's tasks + struct eb32_node *eb; + int ret = TICK_ETERNITY; + __decl_thread(int key = TICK_ETERNITY); + + /* first check in the thread-local timers */ + eb = eb32_lookup_ge(&tt->timers, now_ms - TIMER_LOOK_BACK); + if (!eb) { + /* we might have reached the end of the tree, typically because + * <now_ms> is in the first half and we're first scanning the last + * half. Let's loop back to the beginning of the tree now. + */ + eb = eb32_first(&tt->timers); + } + + if (eb) + ret = eb->key; + +#ifdef USE_THREAD + if (!eb_is_empty(&timers)) { + HA_RWLOCK_RDLOCK(TASK_WQ_LOCK, &wq_lock); + eb = eb32_lookup_ge(&timers, now_ms - TIMER_LOOK_BACK); + if (!eb) + eb = eb32_first(&timers); + if (eb) + key = eb->key; + HA_RWLOCK_RDUNLOCK(TASK_WQ_LOCK, &wq_lock); + if (eb) + ret = tick_first(ret, key); + } +#endif + return ret; +} + +/* Walks over tasklet lists th_ctx->tasklets[0..TL_CLASSES-1] and run at most + * budget[TL_*] of them. Returns the number of entries effectively processed + * (tasks and tasklets merged). The count of tasks in the list for the current + * thread is adjusted. + */ +unsigned int run_tasks_from_lists(unsigned int budgets[]) +{ + struct task *(*process)(struct task *t, void *ctx, unsigned int state); + struct list *tl_queues = th_ctx->tasklets; + struct task *t; + uint8_t budget_mask = (1 << TL_CLASSES) - 1; + struct sched_activity *profile_entry = NULL; + unsigned int done = 0; + unsigned int queue; + unsigned int state; + void *ctx; + + for (queue = 0; queue < TL_CLASSES;) { + th_ctx->current_queue = queue; + + /* global.tune.sched.low-latency is set */ + if (global.tune.options & GTUNE_SCHED_LOW_LATENCY) { + if (unlikely(th_ctx->tl_class_mask & budget_mask & ((1 << queue) - 1))) { + /* a lower queue index has tasks again and still has a + * budget to run them. Let's switch to it now. + */ + queue = (th_ctx->tl_class_mask & 1) ? 0 : + (th_ctx->tl_class_mask & 2) ? 1 : 2; + continue; + } + + if (unlikely(queue > TL_URGENT && + budget_mask & (1 << TL_URGENT) && + !MT_LIST_ISEMPTY(&th_ctx->shared_tasklet_list))) { + /* an urgent tasklet arrived from another thread */ + break; + } + + if (unlikely(queue > TL_NORMAL && + budget_mask & (1 << TL_NORMAL) && + (!eb_is_empty(&th_ctx->rqueue) || + (global_tasks_mask & tid_bit)))) { + /* a task was woken up by a bulk tasklet or another thread */ + break; + } + } + + if (LIST_ISEMPTY(&tl_queues[queue])) { + th_ctx->tl_class_mask &= ~(1 << queue); + queue++; + continue; + } + + if (!budgets[queue]) { + budget_mask &= ~(1 << queue); + queue++; + continue; + } + + budgets[queue]--; + activity[tid].ctxsw++; + + t = (struct task *)LIST_ELEM(tl_queues[queue].n, struct tasklet *, list); + ctx = t->context; + process = t->process; + t->calls++; + th_ctx->current = t; + th_ctx->flags &= ~TH_FL_STUCK; // this thread is still running + + _HA_ATOMIC_DEC(&th_ctx->rq_total); + + if (t->state & TASK_F_TASKLET) { + LIST_DEL_INIT(&((struct tasklet *)t)->list); + __ha_barrier_store(); + + th_ctx->sched_wake_date = ((struct tasklet *)t)->wake_date; + if (th_ctx->sched_wake_date) { + uint32_t now_ns = now_mono_time(); + uint32_t lat = now_ns - th_ctx->sched_wake_date; + + ((struct tasklet *)t)->wake_date = 0; + th_ctx->sched_call_date = now_ns; + profile_entry = sched_activity_entry(sched_activity, t->process); + th_ctx->sched_profile_entry = profile_entry; + HA_ATOMIC_ADD(&profile_entry->lat_time, lat); + HA_ATOMIC_INC(&profile_entry->calls); + } + + state = _HA_ATOMIC_FETCH_AND(&t->state, TASK_PERSISTENT); + __ha_barrier_atomic_store(); + + if (likely(!(state & TASK_KILLED))) { + process(t, ctx, state); + } + else { + done++; + th_ctx->current = NULL; + pool_free(pool_head_tasklet, t); + __ha_barrier_store(); + continue; + } + + if (th_ctx->sched_wake_date) + HA_ATOMIC_ADD(&profile_entry->cpu_time, (uint32_t)(now_mono_time() - th_ctx->sched_call_date)); + + done++; + th_ctx->current = NULL; + __ha_barrier_store(); + continue; + } + + LIST_DEL_INIT(&((struct tasklet *)t)->list); + __ha_barrier_store(); + + th_ctx->sched_wake_date = t->wake_date; + if (unlikely(t->wake_date)) { + uint32_t now_ns = now_mono_time(); + uint32_t lat = now_ns - t->wake_date; + + t->lat_time += lat; + t->wake_date = 0; + th_ctx->sched_call_date = now_ns; + profile_entry = sched_activity_entry(sched_activity, t->process); + th_ctx->sched_profile_entry = profile_entry; + HA_ATOMIC_ADD(&profile_entry->lat_time, lat); + HA_ATOMIC_INC(&profile_entry->calls); + } + + __ha_barrier_store(); + + /* We must be the exclusive owner of the TASK_RUNNING bit, and + * have to be careful that the task is not being manipulated on + * another thread finding it expired in wake_expired_tasks(). + * The TASK_RUNNING bit will be set during these operations, + * they are extremely rare and do not last long so the best to + * do here is to wait. + */ + state = _HA_ATOMIC_LOAD(&t->state); + do { + while (unlikely(state & TASK_RUNNING)) { + __ha_cpu_relax(); + state = _HA_ATOMIC_LOAD(&t->state); + } + } while (!_HA_ATOMIC_CAS(&t->state, &state, (state & TASK_PERSISTENT) | TASK_RUNNING)); + + __ha_barrier_atomic_store(); + + /* OK then this is a regular task */ + + _HA_ATOMIC_DEC(&ha_thread_ctx[tid].tasks_in_list); + + /* Note for below: if TASK_KILLED arrived before we've read the state, we + * directly free the task. Otherwise it will be seen after processing and + * it's freed on the exit path. + */ + if (likely(!(state & TASK_KILLED) && process == process_stream)) + t = process_stream(t, ctx, state); + else if (!(state & TASK_KILLED) && process != NULL) + t = process(t, ctx, state); + else { + task_unlink_wq(t); + __task_free(t); + th_ctx->current = NULL; + __ha_barrier_store(); + /* We don't want max_processed to be decremented if + * we're just freeing a destroyed task, we should only + * do so if we really ran a task. + */ + continue; + } + th_ctx->current = NULL; + __ha_barrier_store(); + + /* stats are only registered for non-zero wake dates */ + if (unlikely(th_ctx->sched_wake_date)) { + uint32_t cpu = (uint32_t)now_mono_time() - th_ctx->sched_call_date; + + if (t) + t->cpu_time += cpu; + HA_ATOMIC_ADD(&profile_entry->cpu_time, cpu); + } + + /* If there is a pending state we have to wake up the task + * immediately, else we defer it into wait queue + */ + if (t != NULL) { + state = _HA_ATOMIC_LOAD(&t->state); + if (unlikely(state & TASK_KILLED)) { + task_unlink_wq(t); + __task_free(t); + } + else { + task_queue(t); + task_drop_running(t, 0); + } + } + done++; + } + th_ctx->current_queue = -1; + + return done; +} + +/* The run queue is chronologically sorted in a tree. An insertion counter is + * used to assign a position to each task. This counter may be combined with + * other variables (eg: nice value) to set the final position in the tree. The + * counter may wrap without a problem, of course. We then limit the number of + * tasks processed to 200 in any case, so that general latency remains low and + * so that task positions have a chance to be considered. The function scans + * both the global and local run queues and picks the most urgent task between + * the two. We need to grab the global runqueue lock to touch it so it's taken + * on the very first access to the global run queue and is released as soon as + * it reaches the end. + * + * The function adjusts <next> if a new event is closer. + */ +void process_runnable_tasks() +{ + struct thread_ctx * const tt = th_ctx; + struct eb32sc_node *lrq; // next local run queue entry + struct eb32sc_node *grq; // next global run queue entry + struct task *t; + const unsigned int default_weights[TL_CLASSES] = { + [TL_URGENT] = 64, // ~50% of CPU bandwidth for I/O + [TL_NORMAL] = 48, // ~37% of CPU bandwidth for tasks + [TL_BULK] = 16, // ~13% of CPU bandwidth for self-wakers + [TL_HEAVY] = 1, // never more than 1 heavy task at once + }; + unsigned int max[TL_CLASSES]; // max to be run per class + unsigned int max_total; // sum of max above + struct mt_list *tmp_list; + unsigned int queue; + int max_processed; + int lpicked, gpicked; + int heavy_queued = 0; + int budget; + + th_ctx->flags &= ~TH_FL_STUCK; // this thread is still running + + if (!thread_has_tasks()) { + activity[tid].empty_rq++; + return; + } + + max_processed = global.tune.runqueue_depth; + + if (likely(niced_tasks)) + max_processed = (max_processed + 3) / 4; + + if (max_processed < th_ctx->rq_total && th_ctx->rq_total <= 2*max_processed) { + /* If the run queue exceeds the budget by up to 50%, let's cut it + * into two identical halves to improve latency. + */ + max_processed = th_ctx->rq_total / 2; + } + + not_done_yet: + max[TL_URGENT] = max[TL_NORMAL] = max[TL_BULK] = 0; + + /* urgent tasklets list gets a default weight of ~50% */ + if ((tt->tl_class_mask & (1 << TL_URGENT)) || + !MT_LIST_ISEMPTY(&tt->shared_tasklet_list)) + max[TL_URGENT] = default_weights[TL_URGENT]; + + /* normal tasklets list gets a default weight of ~37% */ + if ((tt->tl_class_mask & (1 << TL_NORMAL)) || + !eb_is_empty(&th_ctx->rqueue) || (global_tasks_mask & tid_bit)) + max[TL_NORMAL] = default_weights[TL_NORMAL]; + + /* bulk tasklets list gets a default weight of ~13% */ + if ((tt->tl_class_mask & (1 << TL_BULK))) + max[TL_BULK] = default_weights[TL_BULK]; + + /* heavy tasks are processed only once and never refilled in a + * call round. That budget is not lost either as we don't reset + * it unless consumed. + */ + if (!heavy_queued) { + if ((tt->tl_class_mask & (1 << TL_HEAVY))) + max[TL_HEAVY] = default_weights[TL_HEAVY]; + else + max[TL_HEAVY] = 0; + heavy_queued = 1; + } + + /* Now compute a fair share of the weights. Total may slightly exceed + * 100% due to rounding, this is not a problem. Note that while in + * theory the sum cannot be NULL as we cannot get there without tasklets + * to process, in practice it seldom happens when multiple writers + * conflict and rollback on MT_LIST_TRY_APPEND(shared_tasklet_list), causing + * a first MT_LIST_ISEMPTY() to succeed for thread_has_task() and the + * one above to finally fail. This is extremely rare and not a problem. + */ + max_total = max[TL_URGENT] + max[TL_NORMAL] + max[TL_BULK] + max[TL_HEAVY]; + if (!max_total) + goto leave; + + for (queue = 0; queue < TL_CLASSES; queue++) + max[queue] = ((unsigned)max_processed * max[queue] + max_total - 1) / max_total; + + /* The heavy queue must never process more than very few tasks at once + * anyway. We set the limit to 1 if running on low_latency scheduling, + * given that we know that other values can have an impact on latency + * (~500us end-to-end connection achieved at 130kcps in SSL), 1 + one + * per 1024 tasks if there is at least one non-heavy task while still + * respecting the ratios above, or 1 + one per 128 tasks if only heavy + * tasks are present. This allows to drain excess SSL handshakes more + * efficiently if the queue becomes congested. + */ + if (max[TL_HEAVY] > 1) { + if (global.tune.options & GTUNE_SCHED_LOW_LATENCY) + budget = 1; + else if (tt->tl_class_mask & ~(1 << TL_HEAVY)) + budget = 1 + tt->rq_total / 1024; + else + budget = 1 + tt->rq_total / 128; + + if (max[TL_HEAVY] > budget) + max[TL_HEAVY] = budget; + } + + lrq = grq = NULL; + + /* pick up to max[TL_NORMAL] regular tasks from prio-ordered run queues */ + /* Note: the grq lock is always held when grq is not null */ + lpicked = gpicked = 0; + budget = max[TL_NORMAL] - tt->tasks_in_list; + while (lpicked + gpicked < budget) { + if ((global_tasks_mask & tid_bit) && !grq) { +#ifdef USE_THREAD + HA_SPIN_LOCK(TASK_RQ_LOCK, &rq_lock); + grq = eb32sc_lookup_ge(&rqueue, global_rqueue_ticks - TIMER_LOOK_BACK, tid_bit); + if (unlikely(!grq)) { + grq = eb32sc_first(&rqueue, tid_bit); + if (!grq) { + global_tasks_mask &= ~tid_bit; + HA_SPIN_UNLOCK(TASK_RQ_LOCK, &rq_lock); + } + } +#endif + } + + /* If a global task is available for this thread, it's in grq + * now and the global RQ is locked. + */ + + if (!lrq) { + lrq = eb32sc_lookup_ge(&tt->rqueue, tt->rqueue_ticks - TIMER_LOOK_BACK, tid_bit); + if (unlikely(!lrq)) + lrq = eb32sc_first(&tt->rqueue, tid_bit); + } + + if (!lrq && !grq) + break; + + if (likely(!grq || (lrq && (int)(lrq->key - grq->key) <= 0))) { + t = eb32sc_entry(lrq, struct task, rq); + lrq = eb32sc_next(lrq, tid_bit); + eb32sc_delete(&t->rq); + lpicked++; + } +#ifdef USE_THREAD + else { + t = eb32sc_entry(grq, struct task, rq); + grq = eb32sc_next(grq, tid_bit); + _HA_ATOMIC_AND(&t->state, ~TASK_GLOBAL); + eb32sc_delete(&t->rq); + + if (unlikely(!grq)) { + grq = eb32sc_first(&rqueue, tid_bit); + if (!grq) { + global_tasks_mask &= ~tid_bit; + HA_SPIN_UNLOCK(TASK_RQ_LOCK, &rq_lock); + } + } + gpicked++; + } +#endif + if (t->nice) + _HA_ATOMIC_DEC(&niced_tasks); + + /* Add it to the local task list */ + LIST_APPEND(&tt->tasklets[TL_NORMAL], &((struct tasklet *)t)->list); + } + + /* release the rqueue lock */ + if (grq) { + HA_SPIN_UNLOCK(TASK_RQ_LOCK, &rq_lock); + grq = NULL; + } + + if (lpicked + gpicked) { + tt->tl_class_mask |= 1 << TL_NORMAL; + _HA_ATOMIC_ADD(&tt->tasks_in_list, lpicked + gpicked); +#ifdef USE_THREAD + if (gpicked) { + _HA_ATOMIC_SUB(&grq_total, gpicked); + _HA_ATOMIC_ADD(&tt->rq_total, gpicked); + } +#endif + activity[tid].tasksw += lpicked + gpicked; + } + + /* Merge the list of tasklets waken up by other threads to the + * main list. + */ + tmp_list = MT_LIST_BEHEAD(&tt->shared_tasklet_list); + if (tmp_list) { + LIST_SPLICE_END_DETACHED(&tt->tasklets[TL_URGENT], (struct list *)tmp_list); + if (!LIST_ISEMPTY(&tt->tasklets[TL_URGENT])) + tt->tl_class_mask |= 1 << TL_URGENT; + } + + /* execute tasklets in each queue */ + max_processed -= run_tasks_from_lists(max); + + /* some tasks may have woken other ones up */ + if (max_processed > 0 && thread_has_tasks()) + goto not_done_yet; + + leave: + if (tt->tl_class_mask) + activity[tid].long_rq++; +} + +/* + * Delete every tasks before running the master polling loop + */ +void mworker_cleantasks() +{ + struct task *t; + int i; + struct eb32_node *tmp_wq = NULL; + struct eb32sc_node *tmp_rq = NULL; + +#ifdef USE_THREAD + /* cleanup the global run queue */ + tmp_rq = eb32sc_first(&rqueue, MAX_THREADS_MASK); + while (tmp_rq) { + t = eb32sc_entry(tmp_rq, struct task, rq); + tmp_rq = eb32sc_next(tmp_rq, MAX_THREADS_MASK); + task_destroy(t); + } + /* cleanup the timers queue */ + tmp_wq = eb32_first(&timers); + while (tmp_wq) { + t = eb32_entry(tmp_wq, struct task, wq); + tmp_wq = eb32_next(tmp_wq); + task_destroy(t); + } +#endif + /* clean the per thread run queue */ + for (i = 0; i < global.nbthread; i++) { + tmp_rq = eb32sc_first(&ha_thread_ctx[i].rqueue, MAX_THREADS_MASK); + while (tmp_rq) { + t = eb32sc_entry(tmp_rq, struct task, rq); + tmp_rq = eb32sc_next(tmp_rq, MAX_THREADS_MASK); + task_destroy(t); + } + /* cleanup the per thread timers queue */ + tmp_wq = eb32_first(&ha_thread_ctx[i].timers); + while (tmp_wq) { + t = eb32_entry(tmp_wq, struct task, wq); + tmp_wq = eb32_next(tmp_wq); + task_destroy(t); + } + } +} + +/* perform minimal intializations */ +static void init_task() +{ + int i, q; + +#ifdef USE_THREAD + memset(&timers, 0, sizeof(timers)); + memset(&rqueue, 0, sizeof(rqueue)); +#endif + for (i = 0; i < MAX_THREADS; i++) { + for (q = 0; q < TL_CLASSES; q++) + LIST_INIT(&ha_thread_ctx[i].tasklets[q]); + MT_LIST_INIT(&ha_thread_ctx[i].shared_tasklet_list); + } +} + +/* config parser for global "tune.sched.low-latency", accepts "on" or "off" */ +static int cfg_parse_tune_sched_low_latency(char **args, int section_type, struct proxy *curpx, + const struct proxy *defpx, const char *file, int line, + char **err) +{ + if (too_many_args(1, args, err, NULL)) + return -1; + + if (strcmp(args[1], "on") == 0) + global.tune.options |= GTUNE_SCHED_LOW_LATENCY; + else if (strcmp(args[1], "off") == 0) + global.tune.options &= ~GTUNE_SCHED_LOW_LATENCY; + else { + memprintf(err, "'%s' expects either 'on' or 'off' but got '%s'.", args[0], args[1]); + return -1; + } + return 0; +} + +/* config keyword parsers */ +static struct cfg_kw_list cfg_kws = {ILH, { + { CFG_GLOBAL, "tune.sched.low-latency", cfg_parse_tune_sched_low_latency }, + { 0, NULL, NULL } +}}; + +INITCALL1(STG_REGISTER, cfg_register_keywords, &cfg_kws); +INITCALL0(STG_PREPARE, init_task); + +/* + * Local variables: + * c-indent-level: 8 + * c-basic-offset: 8 + * End: + */ |