MEDIUM: tasks: Redispatch shared tasks when the thread is loaded

Now that there is no longer a shared wake queue, chances are if a shared task
is scheduled, it will always end up on the same thread. In
wake_expired_tasks(), when a task has to be waken up, randomly look to
three other threads, and if the runqueue of the current thread is at least
two time bigger than the runqueue of one of the other threads, then give
that task to that thread, so that our load gets reduced.
If we're giving the task to another thread, then we have to add the
TASK_RUNNING flag until we waked it up, otherwise the other thread could
just run it, if it gets waken up from another path, and free it while
we're still not done with it.
2 times has been chosen somewhat arbitrarily, and may be tweaked at a
later date if deemed not optimal.

diff --git a/include/haproxy/task.h b/include/haproxy/task.h
index 0209d6c20aeeb9d658a96ae2db128400621d91c7..a5b4c691d9efc64458dea4d8d1e6fb12e6ac0b99 100644 (file)
--- a/include/haproxy/task.h
+++ b/include/haproxy/task.h
@@ -333,7 +333,8 @@ static inline void task_drop_running(struct task *t, unsigned int f)
                        new_state |= TASK_QUEUED;
 
 
-               if ((new_state & TASK_QUEUED) || cur_tid >= 0 || task_in_wq(t))
+               if ((new_state & TASK_QUEUED) || cur_tid >= 0 || task_in_wq(t) ||
+                   __task_get_current_owner(cur_tid) != tid)
                        new_tid = cur_tid;
                else
                        new_tid = -1;

diff --git a/src/task.c b/src/task.c
index 0806689d6fa0b438142818a0a3461e20e1e3e15b..1b736d86396946f423b25fdcbe751618f61825f8 100644 (file)
--- a/src/task.c
+++ b/src/task.c
@@ -372,9 +372,53 @@ void wake_expired_tasks()
 
                task = eb32_entry(eb, struct task, wq);
                if (tick_is_expired(task->expire, now_ms)) {
+                       int set_running = 0;
+
                        /* expired task, wake it up */
                        __task_unlink_wq(task);
-                       _task_wakeup(task, TASK_WOKEN_TIMER, 0);
+                       /*
+                        * If it's a shared task, see whether we should hand it
+                        * to a less loaded thread.
+                        */
+                       if (task->tid < 0) {
+                               int attempts = MIN(global.nbthread, 3);
+                               while (attempts-- > 0) {
+                                       uint new_tid = statistical_prng_range(global.nbthread);
+
+                                       if (new_tid == tid)
+                                               continue;
+                                       if (ha_thread_ctx[new_tid].rq_total * 2 < th_ctx->rq_total) {
+                                               int cur_state;
+                                               do {
+                                                       cur_state = _HA_ATOMIC_LOAD(&task->state);
+                                                       /*
+                                                        * Okay the task is already in our runqueue,
+                                                        * or somebody owns the
+                                                        * TASK_RUNNING flag because
+                                                        * it is calling task_schedule(), give up.
+                                                        */
+                                                       if (cur_state & (TASK_QUEUED | TASK_RUNNING))
+                                                               break;
+                                                       /*
+                                                        * Make sure we have TASK_RUNNING set
+                                                        * so that the task don't
+                                                        * immediately run on the
+                                                        * new thread and gets
+                                                        * freed.
+                                                        */
+                                                       if (__task_set_state_and_tid(task, task->tid, -2 - new_tid, cur_state, cur_state | TASK_RUNNING)) {
+                                                               set_running = 1;
+                                                               break;
+                                                       }
+                                               } while (1);
+                                               break;
+                                       }
+                               }
+                       }
+                       if (set_running)
+                               task_drop_running(task, TASK_WOKEN_TIMER);
+                       else
+                               _task_wakeup(task, TASK_WOKEN_TIMER, 0);
                }
                else if (task->expire != eb->key) {
                        /* task is not expired but its key doesn't match so let's