upstream/haproxy
1
0
Fork 0
mirror of https://github.com/haproxy/haproxy.git synced 2026-06-13 19:00:25 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

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

Browse source 
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.
This commit is contained in:
Olivier Houchard 2026-06-09 13:01:09 +02:00 committed by Olivier Houchard
parent aaee6c463c
commit b9aa1c0e64
2 changed files with 47 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
include/haproxy/task.h
View file

@ -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;

46
src/task.c
View file

@ -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