Adding upstream version 2.6.12.upstream/2.6.12upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

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:
+ */