/src/haproxy/src/task.c

Source
/*
 * 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/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);
extern void stream_update_timings(struct task *t, uint64_t lat, uint64_t cpu);

DECLARE_TYPED_POOL(pool_head_task,    "task",    struct task, 0, 64);
DECLARE_TYPED_POOL(pool_head_tasklet, "tasklet", struct tasklet, 0, 64);

/* This is the memory pool containing all the signal structs. These
 * struct are used to store each required signal between two tasks.
 */
DECLARE_TYPED_POOL(pool_head_notification, "notification", struct notification);

/* used to detect if the scheduler looks stuck (for warnings) */
static struct {
  int sched_stuck THREAD_ALIGNED();
} sched_ctx[MAX_THREADS];

#if !defined(HA_CAS_IS_8B) && !defined(HA_HAVE_CAS_DW)
__decl_thread(HA_SPINLOCK_T task_state_tid);
#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 = __task_get_current_owner(t->tid);
      if (thr == -1)
        thr = tid;

      /* 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);
      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_QUEUED)) {
      /* 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_QUEUED | 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);
      wake_thread(thr);
      return;
    }
  }
}

/* Do not call this one, please use tasklet_wakeup_here() instead, as this one
 * is the slow path of tasklet_wakeup_here() which performs some preliminary
 * checks and sets TASK_QUEUED before calling this one. It is only permitted to
 * call this function for tasks/tasklets that are not bound (->tid==-1) or that
 * are bound to the local thread (->tid==tid || ->tid==-2-tid).
 */
void __tasklet_wakeup_here(struct tasklet *tl)
{
  BUG_ON_HOT(tl->tid != -1 && tl->tid != tid && tl->tid != -2 - tid);

  if (_HA_ATOMIC_LOAD(&th_ctx->flags) & TH_FL_TASK_PROFILING)
    tl->wake_date = now_mono_time();

  /* 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 (unlikely(tl->state & TASK_RT)) {
    LIST_APPEND(&th_ctx->tasklets[TL_RT], &tl->list);
    th_ctx->tl_class_mask |= 1 << TL_RT;
  }
  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 && !(tl->state & TASK_WOKEN_ANY)) {
    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[TL_NORMAL], &tl->list);
    th_ctx->tl_class_mask |= 1 << TL_NORMAL;
  }
  _HA_ATOMIC_INC(&th_ctx->rq_total);
}

/* 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_QUEUED before calling this one.
 */
void __tasklet_wakeup_on(struct tasklet *tl, int thr)
{
  BUG_ON_HOT(thr < 0);

  if (_HA_ATOMIC_LOAD(&ha_thread_ctx[thr].flags) & TH_FL_TASK_PROFILING)
    tl->wake_date = now_mono_time();

  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);
  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_QUEUED before calling this one.
 */
struct list *__tasklet_wakeup_after(struct list *head, struct tasklet *tl)
{
  BUG_ON(tl->tid >= 0 && tid != tl->tid);

  /* this tasklet runs on the caller thread */
  if (!head) {
    __tasklet_wakeup_here(tl);
  }
  else {
    if (_HA_ATOMIC_LOAD(&th_ctx->flags) & TH_FL_TASK_PROFILING)
      tl->wake_date = now_mono_time();
    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;
  /*
   * At this point the task tid should always be set to the relevant
   * thread, so we can just use __task_get_current_owner();
   */
  int thr __maybe_unused = __task_get_current_owner(t->tid);

  BUG_ON(t->tid == -1);

  if (unlikely(_HA_ATOMIC_LOAD(&t->state) & TASK_RT)) {
    /* real-time tasks must be super rare; they are woken up as tasklets. */
    if (thr < 0 || thr == tid)
      __tasklet_wakeup_here((struct tasklet *)t);
    else
      __tasklet_wakeup_on((struct tasklet *)t, thr);
    return;
  }

#ifdef USE_THREAD
  if (thr != tid) {
    root = &ha_thread_ctx[thr].rqueue_shared;

    _HA_ATOMIC_INC(&ha_thread_ctx[thr].rq_total);
    HA_SPIN_LOCK(TASK_RQ_LOCK, &ha_thread_ctx[thr].rqsh_lock);

    t->rq.key = _HA_ATOMIC_ADD_FETCH(&ha_thread_ctx[thr].rqueue_ticks, 1);
    __ha_barrier_store();
  } else
#endif
  {
    _HA_ATOMIC_INC(&th_ctx->rq_total);
    t->rq.key = _HA_ATOMIC_ADD_FETCH(&th_ctx->rqueue_ticks, 1);
  }

  if (likely(t->nice)) {
    int offset;

    _HA_ATOMIC_INC(&ha_thread_info[thr].tg_ctx->niced_tasks);
    offset = t->nice * (int)global.tune.runqueue_depth;
    t->rq.key += offset;
  }

  if (_HA_ATOMIC_LOAD(&th_ctx->flags) & TH_FL_TASK_PROFILING)
    t->wake_date = now_mono_time();

  eb32_insert(root, &t->rq);

#ifdef USE_THREAD
  if (thr != tid) {
    HA_SPIN_UNLOCK(TASK_RQ_LOCK, &ha_thread_ctx[thr].rqsh_lock);

    /* If all threads that are supposed to handle this task are sleeping,
     * wake one.
     */
    wake_thread(thr);
  }
#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)
{
  int old_state, new_state;
  int old_tid;
  int cur_owner;

  /* 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);

  do {
    new_state = old_state = _HA_ATOMIC_LOAD(&task->state);
    if (old_state & TASK_KILLED)
      return;
    old_tid = _HA_ATOMIC_LOAD(&task->tid);
    cur_owner = __task_get_current_owner(old_tid);
    if (old_tid != -1 && cur_owner != tid)
      new_state |= TASK_WOKEN_WQ;
  } while (!(__task_set_state_and_tid(task, old_tid, __task_get_new_tid_field(old_tid), old_state, new_state)));

  if (cur_owner != tid && cur_owner != -1) {
    /*
     * If the task has already been woken up to be added in the
     * wait queue, nothing left to do, the target thread will
     * eventually do the right thing.
     */
    if (!(old_state & TASK_WOKEN_WQ))
      _task_wakeup(task, 0, NULL);
    return;
  }

  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(&th_ctx->timers, &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;

  while (1) {
    if (max_processed-- <= 0)
      goto leave;

    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)) {
      int set_running = 0;

      /* expired task, wake it up */
      __task_unlink_wq(task);
      /*
       * If it's a shared task, see whether we should hand it
       * to a less loaded thread.
       */
      if (unlikely(task->tid < 0) && global.nbthread > 1) {
        int attempts = MIN(global.nbthread, 3);
        while (attempts-- > 0) {
          uint new_tid = statistical_prng_range(global.nbthread);

          if (new_tid == tid)
            continue;

          ASSUME(new_tid < MAX_THREADS);
          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
       * update it and skip to next apparently expired task.
       */
      __task_unlink_wq(task);
      if (tick_isset(task->expire))
        __task_queue(task);
    }
    else {
      /* task not expired and correctly placed. It may not be eternal. */
      BUG_ON(task->expire == TICK_ETERNITY);
      break;
    }
  }
leave:
  return;
}

/* Checks the next timer for the current thread by looking into its own timer
 * list. 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;

  /* 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;

  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 :
          (th_ctx->tl_class_mask & 4) ? 2 : 3;
        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) || !eb_is_empty(&th_ctx->rqueue_shared)))) {
        /* 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;
    }

    t = (struct task *)LIST_ELEM(tl_queues[queue].n, struct tasklet *, list);

    /* check if this task has already run during this loop */
    if ((uint16_t)t->last_run == (uint16_t)activity[tid].loops) {
      budget_mask &= ~(1 << queue);
      queue++;
      continue;
    }
    t->last_run = activity[tid].loops;
    ctx = t->context;
    process = t->process;
    t->calls++;

    budgets[queue]--;
    activity[tid].ctxsw++;

    th_ctx->lock_wait_total = 0;
    th_ctx->mem_wait_total = 0;
    th_ctx->locked_total = 0;
    th_ctx->sched_wake_date = t->wake_date;
    if (th_ctx->sched_wake_date || (t->state & TASK_F_WANTS_TIME)) {
      /* take the most accurate clock we have, either
       * mono_time() or last now_ns (monotonic but only
       * incremented once per poll loop).
       */
      th_ctx->sched_call_date = now_mono_time();
      if (unlikely(!th_ctx->sched_call_date))
        th_ctx->sched_call_date = now_ns;
    }

    if (th_ctx->sched_wake_date) {
      t->wake_date = 0;
      profile_entry = sched_activity_entry(sched_activity, t->process, t->caller);
      th_ctx->sched_profile_entry = profile_entry;
      HA_ATOMIC_ADD(&profile_entry->lat_time, (uint32_t)(th_ctx->sched_call_date - th_ctx->sched_wake_date));
      HA_ATOMIC_INC(&profile_entry->calls);
    }

    __ha_barrier_store();

    th_ctx->current = t;
    _HA_ATOMIC_AND(&th_ctx->flags, ~TH_FL_STUCK); // this thread is still running

    _HA_ATOMIC_DEC(&th_ctx->rq_total);
    LIST_DEL_INIT(&((struct tasklet *)t)->list);
    __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();

    /* keep the task counter up to date */
    if (!(state & TASK_F_TASKLET))
      _HA_ATOMIC_DEC(&ha_thread_ctx[tid].tasks_in_list);

    /* From this point, we know that the task or tasklet was properly
     * dequeued, flagged and accounted for. Let's now check if it was
     * killed. If TASK_KILLED arrived before we've read the state, we
     * directly free the task/tasklet. Otherwise for tasks it will be
     * seen after processing and it's freed on the exit path.
     */

    if (unlikely((state & TASK_KILLED) || process == NULL)) {
      /* Task or tasklet has been killed, let's remove it */
      if (state & TASK_F_TASKLET)
        pool_free(pool_head_tasklet, t);
      else {
        task_unlink_wq(t);
        __task_free(t);
      }
      /* 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.
       */
      goto next;
    }

    if (state & TASK_WOKEN_WQ) {
      /* We should add this task to our wait queue */
      task_queue(t);
      /*
       * If this is the only reason the task got scheduled,
       * then we don't actually have ot run it.
       */
      if ((state & TASK_WOKEN_ANY) == TASK_WOKEN_WQ) {
        task_drop_running(t, 0);
        goto next;
      }
      state &= ~TASK_WOKEN_WQ;
    }
    /* OK now the task or tasklet is well alive and is going to be run */
    if (state & TASK_F_TASKLET) {
      /* this is a tasklet */

      t = EXEC_CTX_WITH_RET(EXEC_CTX_MAKE(TH_EX_CTX_TASK, process),
                            process(t, ctx, state));
      if (t != NULL)
        _HA_ATOMIC_AND(&t->state, ~TASK_RUNNING);
    } else {
      /* This is a regular task */

      if (process == process_stream)
        t = EXEC_CTX_WITH_RET(EXEC_CTX_MAKE(TH_EX_CTX_TASK, process_stream),
                              process_stream(t, ctx, state));
      else
        t = EXEC_CTX_WITH_RET(EXEC_CTX_MAKE(TH_EX_CTX_TASK, process),
                                    process(t, ctx, state));

      /* 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 {
          if (__task_get_current_owner(t->tid) == tid)
            task_queue(t);
          task_drop_running(t, 0);
        }
      }
    }
    done++;
  next:
    th_ctx->current = NULL;
    sched_ctx[tid].sched_stuck = 0; // scheduler is not stuck (don't warn)
    __ha_barrier_store();

    /* stats are only registered for non-zero wake dates */
    if (unlikely(th_ctx->sched_wake_date)) {
      HA_ATOMIC_ADD(&profile_entry->cpu_time, (uint32_t)(now_mono_time() - th_ctx->sched_call_date));
      if (th_ctx->lock_wait_total)
        HA_ATOMIC_ADD(&profile_entry->lkw_time, th_ctx->lock_wait_total);
      if (th_ctx->mem_wait_total)
        HA_ATOMIC_ADD(&profile_entry->mem_time, th_ctx->mem_wait_total);
      if (th_ctx->locked_total)
        HA_ATOMIC_ADD(&profile_entry->lkd_time, th_ctx->locked_total);
    }
  }
  th_ctx->current_queue = -1;
  th_ctx->sched_wake_date = TICK_ETERNITY;

  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 eb32_node *lrq; // next local run queue entry
  struct eb32_node *grq; // next global run queue entry
  struct task *t;
  const unsigned int default_weights[TL_CLASSES] = {
    [TL_RT]     = 1,  // never more than 1 RT task at once
    [TL_URGENT] = 64, // ~50% of CPU bandwidth for I/O
    [TL_NORMAL] = 60, // ~47% of CPU bandwidth for tasks
    [TL_BULK]   = 4,  // ~3% 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 rt_queued = 0;
  int heavy_queued = 0;
  int budget, done;

  _HA_ATOMIC_AND(&th_ctx->flags, ~TH_FL_STUCK); // this thread is still running

  swrate_add_peak_local(&th_ctx->rq_tot_peak, RQ_LOAD_SAMPLES, th_ctx->rq_total);

  if (!thread_has_tasks()) {
    activity[tid].empty_rq++;
    return;
  }

  max_processed = global.tune.runqueue_depth;

  if (likely(tg_ctx->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;

  /* RT tasklets list may be processed at most once */
  if (!rt_queued) {
    if ((tt->tl_class_mask & (1 << TL_RT))) {
      max[TL_RT] = default_weights[TL_RT];
      rt_queued = 1;
    }
    else
      max[TL_RT] = 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 ~47% */
  if ((tt->tl_class_mask & (1 << TL_NORMAL)) ||
      !eb_is_empty(&th_ctx->rqueue) || !eb_is_empty(&th_ctx->rqueue_shared))
    max[TL_NORMAL] = default_weights[TL_NORMAL];

  /* bulk tasklets list gets a default weight of ~3% */
  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_RT] + 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 RT queue must never process more than one task at once */
  if (max[TL_RT] > 1)
    max[TL_RT] = 1;

  /* 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 && (!rt_queued || !(global.tune.options & GTUNE_SCHED_LOW_LATENCY))) {
    if (!eb_is_empty(&th_ctx->rqueue_shared) && !grq) {
#ifdef USE_THREAD
      HA_SPIN_LOCK(TASK_RQ_LOCK, &th_ctx->rqsh_lock);
      grq = eb32_lookup_ge(&th_ctx->rqueue_shared, _HA_ATOMIC_LOAD(&tt->rqueue_ticks) - TIMER_LOOK_BACK);
      if (unlikely(!grq)) {
        grq = eb32_first(&th_ctx->rqueue_shared);
        if (!grq)
          HA_SPIN_UNLOCK(TASK_RQ_LOCK, &th_ctx->rqsh_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 = eb32_lookup_ge(&tt->rqueue, _HA_ATOMIC_LOAD(&tt->rqueue_ticks) - TIMER_LOOK_BACK);
      if (unlikely(!lrq))
        lrq = eb32_first(&tt->rqueue);
    }

    if (!lrq && !grq)
      break;

    if (likely(!grq || (lrq && (int)(lrq->key - grq->key) <= 0))) {
      t = eb32_entry(lrq, struct task, rq);
      lrq = eb32_next(lrq);
      eb32_delete(&t->rq);
      lpicked++;
    }
#ifdef USE_THREAD
    else {
      t = eb32_entry(grq, struct task, rq);
      grq = eb32_next(grq);
      eb32_delete(&t->rq);

      if (unlikely(!grq)) {
        grq = eb32_first(&th_ctx->rqueue_shared);
        if (!grq)
          HA_SPIN_UNLOCK(TASK_RQ_LOCK, &th_ctx->rqsh_lock);
      }
      gpicked++;
    }
#endif
    if (t->nice)
      _HA_ATOMIC_DEC(&tg_ctx->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, &th_ctx->rqsh_lock);
    grq = NULL;
  }

  if (lpicked + gpicked) {
    tt->tl_class_mask |= 1 << TL_NORMAL;
    _HA_ATOMIC_ADD(&tt->tasks_in_list, lpicked + gpicked);
    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 */
  done = run_tasks_from_lists(max);
  max_processed -= done;

  /* some tasks may have woken other ones up */
  if (done && max_processed > 0 && !rt_queued && thread_has_tasks())
    goto not_done_yet;

 leave:
  if (tt->tl_class_mask)
    activity[tid].long_rq++;
}

/* Pings the scheduler to verify that tasks continue running for thread <thr>.
 * Returns 1 if the scheduler made progress since last call, 0 if it looks
 * stuck. It marks it as stuck for next visit.
 */
int is_sched_alive(int thr)
{
  return !HA_ATOMIC_XCHG(&sched_ctx[thr].sched_stuck, 1);
}

/*
 * Delete every tasks before running the master polling loop
 */
void mworker_cleantasks()
{
  struct task *t;
  int i;
  struct eb32_node *tmp_wq = NULL;
  struct eb32_node *tmp_rq = NULL;

#ifdef USE_THREAD
  /* cleanup the global run queue */
  tmp_rq = eb32_first(&th_ctx->rqueue_shared);
  while (tmp_rq) {
    t = eb32_entry(tmp_rq, struct task, rq);
    tmp_rq = eb32_next(tmp_rq);
    task_destroy(t);
  }
#endif
  /* clean the per thread run queue */
  for (i = 0; i < global.nbthread; i++) {
    tmp_rq = eb32_first(&ha_thread_ctx[i].rqueue);
    while (tmp_rq) {
      t = eb32_entry(tmp_rq, struct task, rq);
      tmp_rq = eb32_next(tmp_rq);
      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 initializations */
static void init_task()
{
  int i, q;

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

Coverage Report

Created: 2026-09-01 07:17