call_function_init()

IPI(Inter Process Interrupts)를 수신 받아 처리하는 핸들러 함수는 handle_IPI()이다. 이 함수에서 사용하는 여러 IPI 항목 중 다음 2개의 항목은 다른 cpu의 요청에 의해 전달받은 call function data를 통해 이에 등록된 함수를 호출한다. 이러한 기능을 지원하기 위해 call_function_init() 함수를 통해 초기화한다.

  • IPI_CALL_FUNC
    • 요청한 함수들을 동작시킨다.
  • IPI_CALL_FUNC_SINGLE
    • 요청한 함수 하나를 동작시킨다.

 

call_function_init()

kernel/smp.c

void __init call_function_init(void)
{
        void *cpu = (void *)(long)smp_processor_id();
        int i;

        for_each_possible_cpu(i)
                init_llist_head(&per_cpu(call_single_queue, i));

        hotplug_cfd(&hotplug_cfd_notifier, CPU_UP_PREPARE, cpu);
        register_cpu_notifier(&hotplug_cfd_notifier);
}

call function에 사용되는 IPI 기능을 초기화한다.

  • 코드 라인 6~7에서 call function이 담기는 call_single_queue를 cpu 수 만큼초기화한다.
  • 코드 라인 9에서 부트업 중인 현재 cpu는 이미 up되어 있으므로 현재 cpu에 대한 hotplug_cfd() 함수를 호출하여 초기화한다.
  • 코드 라인 10에서 cpu 상태 변화에 따라 호출되도록 cpu notifier 블럭을 추가한다.

 

kernel/smp.c

static DEFINE_PER_CPU_SHARED_ALIGNED(struct llist_head, call_single_queue);

cpu별 call function data 구조체가 담기는 리스트이다.

 

kernel/smp.c

static struct notifier_block hotplug_cfd_notifier = {
        .notifier_call          = hotplug_cfd,
};

cpu 상태가 변경될 때 마다 hotplug_cfd() 함수가 호출될 수 있도록 notifier block을 준비한다.

 

hotplug_cfd()

kernel/smp.c

static int
hotplug_cfd(struct notifier_block *nfb, unsigned long action, void *hcpu)
{
        long cpu = (long)hcpu;
        struct call_function_data *cfd = &per_cpu(cfd_data, cpu);

        switch (action) {
        case CPU_UP_PREPARE:
        case CPU_UP_PREPARE_FROZEN:
                if (!zalloc_cpumask_var_node(&cfd->cpumask, GFP_KERNEL,
                                cpu_to_node(cpu)))
                        return notifier_from_errno(-ENOMEM);
                cfd->csd = alloc_percpu(struct call_single_data);
                if (!cfd->csd) {
                        free_cpumask_var(cfd->cpumask);
                        return notifier_from_errno(-ENOMEM);
                }
                break;

#ifdef CONFIG_HOTPLUG_CPU
        case CPU_UP_CANCELED:
        case CPU_UP_CANCELED_FROZEN:
                /* Fall-through to the CPU_DEAD[_FROZEN] case. */ 

        case CPU_DEAD:
        case CPU_DEAD_FROZEN:
                free_cpumask_var(cfd->cpumask);
                free_percpu(cfd->csd);
                break;

        case CPU_DYING:
        case CPU_DYING_FROZEN:
                /* 
                 * The IPIs for the smp-call-function callbacks queued by other
                 * CPUs might arrive late, either due to hardware latencies or
                 * because this CPU disabled interrupts (inside stop-machine)
                 * before the IPIs were sent. So flush out any pending callbacks
                 * explicitly (without waiting for the IPIs to arrive), to
                 * ensure that the outgoing CPU doesn't go offline with work
                 * still pending.
                 */
                flush_smp_call_function_queue(false);
                break;
#endif
        };

        return NOTIFY_OK;
}

cpu 상태가 변화할 때 마다 호출되는 함수로 cpu가 on/off 될 떄마다 해당 cpu에 대한 cfd_data를 초기화하거나 할당해제 처리한다.

  • 코드 라인 5에서 요청 cpu의 cfd_data를 알아와서 cfd에 대입한다.
  • 코드 라인 7~18에서 cfd의 cpumask를 할당하고 per-cpu로 call_single_data 구조체를 할당받아 cfd의 csd에 할당한다.
  • 코드 라인 20~29에서 시스템이 hotplug cpu를 지원하는 경우 cpu가 off된 경우 cfd에 할당해둔 cpumask와 csd를 할당 해제 한다.
  • 코드 라인 31~44에서 cpu가 off되려고하면 call_single_queue를 다 비우고 그 안에 담겨있었던 call function을 모두 호출하여 처리하고 현재의 모든 irq 작업들을 즉시 수행하게 한다.

 

kernel/smp.c

static DEFINE_PER_CPU_SHARED_ALIGNED(struct call_function_data, cfd_data);

전역 cpu별 call_function_data 구조체로 다른 cpu로 single call function을 요청할 때 요청 정보가 담긴다.

  • 이 정보는 전송 시 다른 cpu에서 요청한 함수의 처리가 완료될 때까지 기다리지 않을 때에만 전달 정보로 사용된다.

 

참고

Interrupts -8- (Workqueue 2)

 

워커

워커 생성

create_worker()

kernel/workqueue.c

/**
 * create_worker - create a new workqueue worker
 * @pool: pool the new worker will belong to
 *
 * Create and start a new worker which is attached to @pool.
 *
 * CONTEXT:
 * Might sleep.  Does GFP_KERNEL allocations.
 *
 * Return:
 * Pointer to the newly created worker.
 */
static struct worker *create_worker(struct worker_pool *pool)
{
        struct worker *worker = NULL;
        int id = -1;
        char id_buf[16];

        /* ID is needed to determine kthread name */
        id = ida_simple_get(&pool->worker_ida, 0, 0, GFP_KERNEL);
        if (id < 0)
                goto fail;

        worker = alloc_worker(pool->node);
        if (!worker)
                goto fail;

        worker->pool = pool;
        worker->id = id;

        if (pool->cpu >= 0)
                snprintf(id_buf, sizeof(id_buf), "%d:%d%s", pool->cpu, id,
                         pool->attrs->nice < 0  ? "H" : "");
        else
                snprintf(id_buf, sizeof(id_buf), "u%d:%d", pool->id, id);

        worker->task = kthread_create_on_node(worker_thread, worker, pool->node,
                                              "kworker/%s", id_buf);
        if (IS_ERR(worker->task))
                goto fail;

        set_user_nice(worker->task, pool->attrs->nice);

        /* prevent userland from meddling with cpumask of workqueue workers */
        worker->task->flags |= PF_NO_SETAFFINITY;

        /* successful, attach the worker to the pool */
        worker_attach_to_pool(worker, pool);

        /* start the newly created worker */
        spin_lock_irq(&pool->lock);
        worker->pool->nr_workers++;
        worker_enter_idle(worker);
        wake_up_process(worker->task);
        spin_unlock_irq(&pool->lock);

        return worker;

fail:
        if (id >= 0)
                ida_simple_remove(&pool->worker_ida, id);
        kfree(worker);
        return NULL;
}

요청한 워커풀에 워커 스레드를 생성한다.

  • 코드 라인 20~22에서 워커풀에서 생성할 새로운 워커를 위해 id를 받아온다.
    • pool->worker_ida는 워커들의 id의 할당 관리를 위해 IDR Radix tree 기반으로 동작한다.
  • 코드 라인 24~29에서 워커 객체를 할당 받은 후 워커풀 및 id를 지정한다.
  • 코드 라인 31~35에서 워커풀의 이름을 지정한다.
    • cpu로 바운드(지정)된 워커풀은 “<cpu>:<worker id>H”과 같이 표현한다.  ‘H’는 스레드의 nice 값이 디폴트 nice보다 작을때(우선순위가 더 높다) H 문자열이 출력된다.
    • 언바운드된 워커풀은 “u<pool id>:<worker id>”와 같이 표현한다.
  • 코드 라인 37~40에서 kthread를 통해 worker_thread() 함수가 호출되는 스레드를 생성하여 워커의 task에 대입한다.
  • 코드 라인 42에서 워커 스레드의 static_prio 값을 워커풀 속성에 있는 nice 값을 우선순위로 변환한 값으로 설정한다.
  • 코드 라인 45에서 워커 스레드에 PF_NO_SETAFFINITY 플래그를 추가하여 다른 cpu로 마이그레이션되지 않도록 막는다.
  • 코드 라인 48에서 워커 스레드를 워커풀에 지정한다.
  • 코드 라인 52에서 워커풀내의 워커 수를 나타내는 nr_workers를 1 증가시킨다.
  • 코드 라인 53~54에서 워커 스레드를 idle 상태로 진입하게 한 후 다시 깨운다.

 

alloc_worker()

kernel/workqueue.c

static struct worker *alloc_worker(int node)
{
        struct worker *worker;

        worker = kzalloc_node(sizeof(*worker), GFP_KERNEL, node);
        if (worker) {
                INIT_LIST_HEAD(&worker->entry);
                INIT_LIST_HEAD(&worker->scheduled);
                INIT_LIST_HEAD(&worker->node);
                /* on creation a worker is in !idle && prep state */
                worker->flags = WORKER_PREP;
        }
        return worker;
}

워커 구조체를 할당받고 내부에서 관리하는 리스트들을 초기화한다.

 

워커를 워커풀에 연결(attach)

worker_attach_to_pool()

kernel/workqueue.c

/**
 * worker_attach_to_pool() - attach a worker to a pool
 * @worker: worker to be attached
 * @pool: the target pool
 *
 * Attach @worker to @pool.  Once attached, the %WORKER_UNBOUND flag and
 * cpu-binding of @worker are kept coordinated with the pool across
 * cpu-[un]hotplugs.
 */
static void worker_attach_to_pool(struct worker *worker,
                                   struct worker_pool *pool)
{
        mutex_lock(&pool->attach_mutex);

        /*
         * set_cpus_allowed_ptr() will fail if the cpumask doesn't have any
         * online CPUs.  It'll be re-applied when any of the CPUs come up.
         */
        set_cpus_allowed_ptr(worker->task, pool->attrs->cpumask);

        /*
         * The pool->attach_mutex ensures %POOL_DISASSOCIATED remains
         * stable across this function.  See the comments above the
         * flag definition for details.
         */
        if (pool->flags & POOL_DISASSOCIATED)
                worker->flags |= WORKER_UNBOUND;

        list_add_tail(&worker->node, &pool->workers);

        mutex_unlock(&pool->attach_mutex);
}

워커를 워커풀에 연결한다.

  • 코드 라인 19에서 워커풀 속성에 있는 cpumask를 워커 스레드에 지정한다.
    • 바운드된 워커풀은 특정 cpu만 마스크비트가 설정되어 있다.
    • 언바운드된 워커풀은 online된 cpu들에 대해 모두 마스크비트가 설정되어 있다.
  • 코드 라인 26~27에서 워커풀이 아직 서비스하지 않는 상태인 경우 워커에 WORKER_UNBOUND 플래그를 설정한다.
  • 코드 라인 29에서 워커풀의 workers 리스트에 워커를 추가한다.

 

워커를 워커풀에서 연결 해제

worker_detach_from_pool()

kernel/workqueue.c

/**
 * worker_detach_from_pool() - detach a worker from its pool
 * @worker: worker which is attached to its pool
 * @pool: the pool @worker is attached to
 *
 * Undo the attaching which had been done in worker_attach_to_pool().  The
 * caller worker shouldn't access to the pool after detached except it has
 * other reference to the pool.
 */
static void worker_detach_from_pool(struct worker *worker,
                                    struct worker_pool *pool)
{
        struct completion *detach_completion = NULL;

        mutex_lock(&pool->attach_mutex);
        list_del(&worker->node);
        if (list_empty(&pool->workers))
                detach_completion = pool->detach_completion;
        mutex_unlock(&pool->attach_mutex);

        /* clear leftover flags without pool->lock after it is detached */
        worker->flags &= ~(WORKER_UNBOUND | WORKER_REBOUND);

        if (detach_completion)
                complete(detach_completion);
}

워커를 워커풀에서 연결해제한다.

  • 코드 라인 16에서 워커풀의 workers 리스트에서 요청한 워커를 제거한다.
  • 코드 라인 17~18에서 워커풀이 빈 경우 워커풀의 detach_completion 상태를 알아온다.
  • 코드 라인 22에서 워커의 WORKER_UNBOUND 및 WORKER_REBOUND 플래그를 제거한다.
  • 코드 라인 24~25에서 detach 작업이 완료될 때 까지 기다린다.

 

워커 Idle 진입

worker_enter_idle()

kernel/workqueue.c

/**
 * worker_enter_idle - enter idle state
 * @worker: worker which is entering idle state
 *
 * @worker is entering idle state.  Update stats and idle timer if
 * necessary.
 *
 * LOCKING:
 * spin_lock_irq(pool->lock).
 */
static void worker_enter_idle(struct worker *worker)
{
        struct worker_pool *pool = worker->pool;

        if (WARN_ON_ONCE(worker->flags & WORKER_IDLE) ||
            WARN_ON_ONCE(!list_empty(&worker->entry) &&
                         (worker->hentry.next || worker->hentry.pprev)))
                return;

        /* can't use worker_set_flags(), also called from create_worker() */
        worker->flags |= WORKER_IDLE;
        pool->nr_idle++;
        worker->last_active = jiffies;

        /* idle_list is LIFO */
        list_add(&worker->entry, &pool->idle_list);

        if (too_many_workers(pool) && !timer_pending(&pool->idle_timer))
                mod_timer(&pool->idle_timer, jiffies + IDLE_WORKER_TIMEOUT);

        /*
         * Sanity check nr_running.  Because wq_unbind_fn() releases
         * pool->lock between setting %WORKER_UNBOUND and zapping
         * nr_running, the warning may trigger spuriously.  Check iff
         * unbind is not in progress.
         */
        WARN_ON_ONCE(!(pool->flags & POOL_DISASSOCIATED) &&
                     pool->nr_workers == pool->nr_idle &&
                     atomic_read(&pool->nr_running));
}

요청한 워커 스레드를 idle 상태로 설정하고 idle 카운터를 증가시킨다. 필요 시 idle 카운터를 동작시킨다.

  • 코드 라인 15~18에서 워커가 이미 idle 상태이거나  워커가 워커풀의 idle_list에서 대기중면서 busy_hash[] 리스트에서도 존재하는 경우 경고 메시지를 1번 출력 후 함수를 빠져나간다.
  • 코드 라인 21~23에서 워커에 WORKER_IDLE 플래그를 설정하고 idle 카운터를 1 증가시킨다. last_active에 현재 시각으로 갱신한다.
  • 코드 라인 26에서 워커풀의 idle 리스트에 워커를 추가한다.
  • 코드 라인 28~29에서 워커풀에 워커들이 너무 많이 쉬고 있으면 idle 타이머를 현재 시각 기준으로  IDLE_WORKER_TIMEOUT(5분) 후에 동작하도록 설정한다.
  • 코드 라인 37~39에서 워커풀이 서비스 중이고 워커들이 모두 idle 상태인데 동작 중인 워커들이 있다고 하는 경우 경고메시지를 출력한다.
    • nr_workers(워커풀에 등록된 워커 수) = nr_running(동작중인 워커 수) + nr_idle(idle 상태인 워커 수)

 

too_many_workers()

kernel/workqueue.c

/* Do we have too many workers and should some go away? */
static bool too_many_workers(struct worker_pool *pool)
{
        bool managing = mutex_is_locked(&pool->manager_arb);
        int nr_idle = pool->nr_idle + managing; /* manager is considered idle */
        int nr_busy = pool->nr_workers - nr_idle;

        return nr_idle > 2 && (nr_idle - 2) * MAX_IDLE_WORKERS_RATIO >= nr_busy;
}

워커풀에 idle 워커들이 busy 워커들의 일정 비율 이상 즉, idle 워커 수가 많은지 여부를 반환한다.

  • 코드 라인 4~6에서 워커풀에 등록된 워커들에서 idle 워커를 제외한 수를 알아온다.
  • 코드 라인 8에서 idle 워커에서 기본 idle 워커 수 2를 뺀 나머지 idle 워커 수가 busy 워커보다 일정 비율(기본 1/4배) 이상인 경우 놀고 있는 idle 워커 수가 많다고 판단하여 true를 반환한다.
    • 조건: (idle 워커 – 2) * MAX_IDLE_WORKERS_RATIO(4) > busy 워커
    • 이 함수가 true가 되는 조건 예)
      • idle=3, busy=0~4
      • idle=4, busy=0~8
      • idle=5, busy=0~12

 

다음 그림은 idle 워커가 너무 많은지 여부를 판단한다.

 

Idle 워커  타임아웃 -> idle 워커 수 줄이기

idle_worker_timeout()

kernel/workqueue.c

static void idle_worker_timeout(unsigned long __pool)
{
        struct worker_pool *pool = (void *)__pool;

        spin_lock_irq(&pool->lock);

        while (too_many_workers(pool)) {
                struct worker *worker;
                unsigned long expires;

                /* idle_list is kept in LIFO order, check the last one */
                worker = list_entry(pool->idle_list.prev, struct worker, entry);
                expires = worker->last_active + IDLE_WORKER_TIMEOUT;

                if (time_before(jiffies, expires)) {
                        mod_timer(&pool->idle_timer, expires);
                        break;
                }

                destroy_worker(worker);
        }

        spin_unlock_irq(&pool->lock);
}

idle 타이머가 만료 시 호출되며 busy 워커에 비해 너무 많은 idle 워커 비율(기본 1/4)가 있다고 판단하면 일정 비율만큼 idle 워커를 줄인다.

  • 코드 라인 7에서 워커풀에 idle 워커들이 busy 워커들의 일정 비율 이상 즉, idle 워커 수가 많은 경우에 한하여 순회한다.
  • 코드 라인 12~21에서 idle_list에서 하나의 워커를 가져와서 워커의 최종 사용 시각이 현재 시각에 비해 IDLE_WORKER_TIMEOUT(5분)을 지나지 않은 경우 만료 시각을 다시 재설정하고 함수를 빠져간다. 이미 지정된 시간을 초과한 경우 워커를 제거하고 계속 루프를 돈다.

 

 

워커 소멸

destroy_worker()

kernel/workqueue.c

/**
 * destroy_worker - destroy a workqueue worker
 * @worker: worker to be destroyed
 *
 * Destroy @worker and adjust @pool stats accordingly.  The worker should
 * be idle.
 *
 * CONTEXT:
 * spin_lock_irq(pool->lock).
 */
static void destroy_worker(struct worker *worker)
{
        struct worker_pool *pool = worker->pool;

        lockdep_assert_held(&pool->lock);

        /* sanity check frenzy */
        if (WARN_ON(worker->current_work) ||
            WARN_ON(!list_empty(&worker->scheduled)) ||
            WARN_ON(!(worker->flags & WORKER_IDLE)))
                return;

        pool->nr_workers--;
        pool->nr_idle--;

        list_del_init(&worker->entry);
        worker->flags |= WORKER_DIE;
        wake_up_process(worker->task);
}

워커를 소멸시킨다. 즉 워커 스레드를 소멸시킨다.

  • 코드 라인 18~21에서 요청한 워크가 동작 중이거나 이 워커에 스케줄된 워크가 있거나 idle 플래그가 없는 경우 경고 메시지를 출력하고 함수를 빠져나간다.
  • 코드 라인 23에서 워커풀에 등록된 워커 수인 nr_workers를 1 감소시킨다.
  • 코드 라인 24에서 워커풀에서 idle 중인 워커 수를 나타내는 nr_idle을 1 감소시킨다.
  • 코드 라인 26에서 워커를 워커풀의 idle_list 에서 제거한다.
  • 코드 라인 27~28에서 워커에 WORKER_DIE 플래그를 설정하고 worker에 연결된 태스크를 깨워 소멸 처리한다.
    • 워커풀에서 워커 id를 반납하고 워커풀에서 디태치하며 워커를 제거한다.
    • 제거중인 상황에서의 태스크명을 “kworker/dying”으로 변경한다.

 

워크가 동작했던 워커 찾기

find_worker_executing_work()

kernel/workqueue.c

/**
 * find_worker_executing_work - find worker which is executing a work
 * @pool: pool of interest
 * @work: work to find worker for
 *
 * Find a worker which is executing @work on @pool by searching
 * @pool->busy_hash which is keyed by the address of @work.  For a worker
 * to match, its current execution should match the address of @work and
 * its work function.  This is to avoid unwanted dependency between
 * unrelated work executions through a work item being recycled while still
 * being executed.
 *
 * This is a bit tricky.  A work item may be freed once its execution
 * starts and nothing prevents the freed area from being recycled for
 * another work item.  If the same work item address ends up being reused
 * before the original execution finishes, workqueue will identify the
 * recycled work item as currently executing and make it wait until the
 * current execution finishes, introducing an unwanted dependency.
 *
 * This function checks the work item address and work function to avoid
 * false positives.  Note that this isn't complete as one may construct a
 * work function which can introduce dependency onto itself through a
 * recycled work item.  Well, if somebody wants to shoot oneself in the
 * foot that badly, there's only so much we can do, and if such deadlock
 * actually occurs, it should be easy to locate the culprit work function.
 *
 * CONTEXT:
 * spin_lock_irq(pool->lock).
 *
 * Return:
 * Pointer to worker which is executing @work if found, %NULL
 * otherwise.
 */
static struct worker *find_worker_executing_work(struct worker_pool *pool,
                                                 struct work_struct *work)
{
        struct worker *worker;

        hash_for_each_possible(pool->busy_hash, worker, hentry,
                               (unsigned long)work)
                if (worker->current_work == work &&
                    worker->current_func == work->func)
                        return worker;

        return NULL;
}

동작중인 워커들에서 워크가 동작하는 워커를 찾아온다. 없으면 null을 반환한다.

  • busy_hash 리스트를 순회하며 요청한 워크가 현재 동작중인 워커를 찾는다.

 

need_more_worker()

kernel/workqueue.c

/*
 * Need to wake up a worker?  Called from anything but currently
 * running workers.
 *
 * Note that, because unbound workers never contribute to nr_running, this
 * function will always return %true for unbound pools as long as the
 * worklist isn't empty.
 */
static bool need_more_worker(struct worker_pool *pool)
{
        return !list_empty(&pool->worklist) && __need_more_worker(pool);
}

워커가 더 필요한지 여부를 알아온다.

  • 워커풀에 작업이 대기되어 있고 워커가 더 필요한 경우 true를 반환한다.

 

__need_more_worker()

kernel/workqueue.c

/*
 * Policy functions.  These define the policies on how the global worker
 * pools are managed.  Unless noted otherwise, these functions assume that
 * they're being called with pool->lock held.
 */

static bool __need_more_worker(struct worker_pool *pool)
{
        return !atomic_read(&pool->nr_running);
}

워커풀에서 동작 중인 워커가 없는지 여부를 반환한다. (1=현재 동작중인 워커가 없다.)

 

wake_up_worker()

kernel/workqueue.c

/**
 * wake_up_worker - wake up an idle worker
 * @pool: worker pool to wake worker from
 *
 * Wake up the first idle worker of @pool.
 *
 * CONTEXT:
 * spin_lock_irq(pool->lock).
 */
static void wake_up_worker(struct worker_pool *pool)
{
        struct worker *worker = first_idle_worker(pool);

        if (likely(worker))
                wake_up_process(worker->task);
}

요청한 워커풀의 첫 번째  idle 워커를 깨운다.

 

first_idle_worker()

kernel/workqueue.c

/* Return the first idle worker.  Safe with preemption disabled */
static struct worker *first_idle_worker(struct worker_pool *pool)
{
        if (unlikely(list_empty(&pool->idle_list)))
                return NULL;

        return list_first_entry(&pool->idle_list, struct worker, entry);
}

요청한 워커풀에서 첫 idle 워커를 반환한다.

  • idle_list에서 있는 첫 번째 워커를 반환한다.

 

워커 스레드 동작

worker_thread()

kernel/workqueue.c

/**
 * worker_thread - the worker thread function
 * @__worker: self
 *
 * The worker thread function.  All workers belong to a worker_pool -
 * either a per-cpu one or dynamic unbound one.  These workers process all
 * work items regardless of their specific target workqueue.  The only
 * exception is work items which belong to workqueues with a rescuer which
 * will be explained in rescuer_thread().
 *
 * Return: 0
 */
static int worker_thread(void *__worker)
{
        struct worker *worker = __worker;
        struct worker_pool *pool = worker->pool;

        /* tell the scheduler that this is a workqueue worker */
        worker->task->flags |= PF_WQ_WORKER;
woke_up:
        spin_lock_irq(&pool->lock);

        /* am I supposed to die? */
        if (unlikely(worker->flags & WORKER_DIE)) {
                spin_unlock_irq(&pool->lock);
                WARN_ON_ONCE(!list_empty(&worker->entry));
                worker->task->flags &= ~PF_WQ_WORKER;

                set_task_comm(worker->task, "kworker/dying");
                ida_simple_remove(&pool->worker_ida, worker->id);
                worker_detach_from_pool(worker, pool);
                kfree(worker);
                return 0;
        }

        worker_leave_idle(worker);
recheck:
        /* no more worker necessary? */
        if (!need_more_worker(pool))
                goto sleep;

        /* do we need to manage? */
        if (unlikely(!may_start_working(pool)) && manage_workers(worker))
                goto recheck;

        /*
         * ->scheduled list can only be filled while a worker is
         * preparing to process a work or actually processing it.
         * Make sure nobody diddled with it while I was sleeping.
         */
        WARN_ON_ONCE(!list_empty(&worker->scheduled));
  • 코드 라인 19에서 스케줄러로 하여금 이 태스크가 워커 스레드인 것을 알 수 있게 PF_WQ_WORKER 플래그를 설정한다.
  • 코드 라인 24~34에서 워커 스레드의 종료 요청 플래그가 설정된 경우 워커를 워커풀에서 제거하고 할당 해제 후 종료한다.
  • 코드 라인 36에서 워커 스레드를 idle 상태로 둔다.
  • 코드 라인 39~40에서 추가 적인 워커가 필요하지 않으면 sleep으로 이동한다.
    • 동작중인 워커가 하나도 없으면 더 이상 필요한 워커가 없다고 판단한다.
  • 코드 라인 43~44에서 워커풀에 idle 워커가 없으면 워커를 생성한다. 만일 워커를 생성하기 위한 락 획득 시도가 실패하는 경우 다시 recheck 레이블로 이동하여 다시 시도한다.

 

        /*
         * Finish PREP stage.  We're guaranteed to have at least one idle
         * worker or that someone else has already assumed the manager
         * role.  This is where @worker starts participating in concurrency
         * management if applicable and concurrency management is restored
         * after being rebound.  See rebind_workers() for details.
         */
        worker_clr_flags(worker, WORKER_PREP | WORKER_REBOUND);

        do {
                struct work_struct *work =
                        list_first_entry(&pool->worklist,
                                         struct work_struct, entry);

                if (likely(!(*work_data_bits(work) & WORK_STRUCT_LINKED))) {
                        /* optimization path, not strictly necessary */
                        process_one_work(worker, work);
                        if (unlikely(!list_empty(&worker->scheduled)))
                                process_scheduled_works(worker);
                } else {
                        move_linked_works(work, &worker->scheduled, NULL);
                        process_scheduled_works(worker);
                }
        } while (keep_working(pool));

        worker_set_flags(worker, WORKER_PREP);
sleep:
        /*
         * pool->lock is held and there's no work to process and no need to
         * manage, sleep.  Workers are woken up only while holding
         * pool->lock or from local cpu, so setting the current state
         * before releasing pool->lock is enough to prevent losing any
         * event.
         */
        worker_enter_idle(worker);
        __set_current_state(TASK_INTERRUPTIBLE);
        spin_unlock_irq(&pool->lock);
        schedule();
        goto woke_up;
}
  • 코드 라인 8에서 워커에서 WORKER_PREP 및 WORKER_REBOUND  플래그를 제거한다.
  • 코드 라인 10~13에서 워커풀에 있는 워크리스트에서 처리할 워크를 가져온다.
  • 코드 라인 15~19에서 높은 확률로 워크에서 linked라는 플래그가 없는 경우 하나의 워크를 처리한다. 그 후 낮은 확률로 워커의 scheduled리스트에 있는 워크도 모두 처리한다.
  • 코드 라인 20~23에서 lined라는 플래그가 있는 경우 다음 연결된 워크를 모두 처리한다.
  • 코드 라인 24에서 워커풀의 워크리스트가 다 처리되어 empty될 때까지 루프를 반복한다.
  • 코드 라인 26에서 워커를 다시 prep 상태로 변경한다.
  • 코드 라인 35~36에서 워커를 idle 상태로 설정하고 태스크 상태는 슬립 중 wakeup할 수 있도록 인터럽터블로 변경한다.
  • 코드 라인 38~39에서 스케줄 함수를 통해 슬립한다. 그 후 깨어나면 다시 처음 루틴부터 시작하도록 woke_up 레이블로 이동한다.

 

다음 그림은 워커 스레드가 워크를 처리하는 과정을 보여준다.

 

manage_workers()

kernel/workqueue.c

/**
 * manage_workers - manage worker pool
 * @worker: self
 *
 * Assume the manager role and manage the worker pool @worker belongs
 * to.  At any given time, there can be only zero or one manager per
 * pool.  The exclusion is handled automatically by this function.
 *
 * The caller can safely start processing works on false return.  On
 * true return, it's guaranteed that need_to_create_worker() is false
 * and may_start_working() is true.
 *
 * CONTEXT:             
 * spin_lock_irq(pool->lock) which may be released and regrabbed
 * multiple times.  Does GFP_KERNEL allocations.
 *
 * Return:
 * %false if the pool doesn't need management and the caller can safely
 * start processing works, %true if management function was performed and
 * the conditions that the caller verified before calling the function may
 * no longer be true.
 */
static bool manage_workers(struct worker *worker)
{
        struct worker_pool *pool = worker->pool;

        /*
         * Anyone who successfully grabs manager_arb wins the arbitration
         * and becomes the manager.  mutex_trylock() on pool->manager_arb
         * failure while holding pool->lock reliably indicates that someone
         * else is managing the pool and the worker which failed trylock
         * can proceed to executing work items.  This means that anyone
         * grabbing manager_arb is responsible for actually performing
         * manager duties.  If manager_arb is grabbed and released without
         * actual management, the pool may stall indefinitely.
         */
        if (!mutex_trylock(&pool->manager_arb))
                return false;

        maybe_create_worker(pool);

        mutex_unlock(&pool->manager_arb);
        return true;
}

필요한 만큼 요청한 워커풀에 워커를 생성한다. 만일 워커를 생성하기 위해 락 획득 시도가 실패하는 경우 false를 반환한다.

 

maybe_create_worker()

kernel/workqueue.c

/**
 * maybe_create_worker - create a new worker if necessary
 * @pool: pool to create a new worker for
 *
 * Create a new worker for @pool if necessary.  @pool is guaranteed to
 * have at least one idle worker on return from this function.  If
 * creating a new worker takes longer than MAYDAY_INTERVAL, mayday is
 * sent to all rescuers with works scheduled on @pool to resolve
 * possible allocation deadlock.
 *
 * On return, need_to_create_worker() is guaranteed to be %false and
 * may_start_working() %true.
 *
 * LOCKING:
 * spin_lock_irq(pool->lock) which may be released and regrabbed
 * multiple times.  Does GFP_KERNEL allocations.  Called only from
 * manager.
 */
static void maybe_create_worker(struct worker_pool *pool)
__releases(&pool->lock)
__acquires(&pool->lock)
{
restart:
        spin_unlock_irq(&pool->lock);

        /* if we don't make progress in MAYDAY_INITIAL_TIMEOUT, call for help */
        mod_timer(&pool->mayday_timer, jiffies + MAYDAY_INITIAL_TIMEOUT);

        while (true) {
                if (create_worker(pool) || !need_to_create_worker(pool))
                        break;

                schedule_timeout_interruptible(CREATE_COOLDOWN);

                if (!need_to_create_worker(pool))
                        break;
        }

        del_timer_sync(&pool->mayday_timer);
        spin_lock_irq(&pool->lock);
        /*
         * This is necessary even after a new worker was just successfully
         * created as @pool->lock was dropped and the new worker might have
         * already become busy.
         */
        if (need_to_create_worker(pool))
                goto restart;
}

필요한 만큼 요청한 워커풀에 워커를 생성한다.

  • 코드 라인 27에서 mayday 타이머를 현재 시각 기준으로 만료 시각을 다시 조정한다.
  • 코드 라인 29~37에서 반복하며 필요한 수 만큼 워커를 생성한다.
  • 코드 라인 39에서 mayday 타이머를 제거한다.
  • 코드 라인 46~47에서 다시 한 번 워커를 추가해야 하는지 확인하여 재시도할지 결정한다.

 

need_to_create_worker()

kernel/workqueue.c

/* Do we need a new worker?  Called from manager. */
static bool need_to_create_worker(struct worker_pool *pool)
{
        return need_more_worker(pool) && !may_start_working(pool);
}

요청한 워커풀에 워커가 더 필요하지만 준비된 idle 워커가 없어서 곧바로 동작할 수 없는 경우 워커를 만들 필요가 있는지 여부를 반환한다.

 

연결된(스케줄된) 워크들 처리

move_linked_works()

kernel/workqueue.c

/**                               
 * move_linked_works - move linked works to a list
 * @work: start of series of works to be scheduled
 * @head: target list to append @work to
 * @nextp: out paramter for nested worklist walking
 *
 * Schedule linked works starting from @work to @head.  Work series to
 * be scheduled starts at @work and includes any consecutive work with
 * WORK_STRUCT_LINKED set in its predecessor.
 *      
 * If @nextp is not NULL, it's updated to point to the next work of
 * the last scheduled work.  This allows move_linked_works() to be
 * nested inside outer list_for_each_entry_safe().
 *
 * CONTEXT:
 * spin_lock_irq(pool->lock).
 */
static void move_linked_works(struct work_struct *work, struct list_head *head,
                              struct work_struct **nextp)
{       
        struct work_struct *n;

        /*
         * Linked worklist will always end before the end of the list,
         * use NULL for list head.
         */
        list_for_each_entry_safe_from(work, n, NULL, entry) {
                list_move_tail(&work->entry, head);
                if (!(*work_data_bits(work) & WORK_STRUCT_LINKED))
                        break;
        }

        /*
         * If we're already inside safe list traversal and have moved
         * multiple works to the scheduled queue, the next position
         * needs to be updated.
         */
        if (nextp)
                *nextp = n;
}

연결된 워크들을 두 번째 head 리스트로 옮긴다.

  • 코드 라인 27~31에서 워크들을 순회하며 head로 옮긴다. 만일 워크에 linked 속성 플래그가 없는 경우 루프를 벗어난다.
  • 코드 라인 38~39에서 출력 인수 nextp에 처리된 워크를 반환한다.

 

process_scheduled_works()

kernel/workqueue.c

/**
 * process_scheduled_works - process scheduled works
 * @worker: self
 *
 * Process all scheduled works.  Please note that the scheduled list
 * may change while processing a work, so this function repeatedly
 * fetches a work from the top and executes it.
 *
 * CONTEXT:
 * spin_lock_irq(pool->lock) which may be released and regrabbed
 * multiple times.
 */
static void process_scheduled_works(struct worker *worker)
{
        while (!list_empty(&worker->scheduled)) {
                struct work_struct *work = list_first_entry(&worker->scheduled,
                                                struct work_struct, entry);
                process_one_work(worker, work);
        }
}

모든 스케줄된 워크를 처리한다.

  • 코드 라인 15~17에서 워커의 스케줄드 리스트에 있는 워크들 수 만큼 순회하며 첫 워크를 가져온다.
  • 코드 라인 18에서 가져온 워크를 처리하낟.

 

하나의 워크 처리

process_one_work()

워커에서 워크를 하나 처리한다.

kernel/workqueue.c – 1/4

/**
 * process_one_work - process single work
 * @worker: self
 * @work: work to process
 *
 * Process @work.  This function contains all the logics necessary to
 * process a single work including synchronization against and
 * interaction with other workers on the same cpu, queueing and
 * flushing.  As long as context requirement is met, any worker can
 * call this function to process a work.
 *
 * CONTEXT:
 * spin_lock_irq(pool->lock) which is released and regrabbed.
 */
static void process_one_work(struct worker *worker, struct work_struct *work)
__releases(&pool->lock)
__acquires(&pool->lock)
{
        struct pool_workqueue *pwq = get_work_pwq(work);
        struct worker_pool *pool = worker->pool;
        bool cpu_intensive = pwq->wq->flags & WQ_CPU_INTENSIVE;
        int work_color;
        struct worker *collision;
#ifdef CONFIG_LOCKDEP
        /*
         * It is permissible to free the struct work_struct from
         * inside the function that is called from it, this we need to
         * take into account for lockdep too.  To avoid bogus "held
         * lock freed" warnings as well as problems when looking into
         * work->lockdep_map, make a copy and use that here.
         */
        struct lockdep_map lockdep_map;

        lockdep_copy_map(&lockdep_map, &work->lockdep_map);
#endif
        /* ensure we're on the correct CPU */
        WARN_ON_ONCE(!(pool->flags & POOL_DISASSOCIATED) &&
                     raw_smp_processor_id() != pool->cpu);
  • 코드 라인 21에서 워크큐에 cpu intensive 플래그가 사용되었는지 여부를 알아온다.

 

kernel/workqueue.c – 2/4

.       /*
         * A single work shouldn't be executed concurrently by
         * multiple workers on a single cpu.  Check whether anyone is
         * already processing the work.  If so, defer the work to the
         * currently executing one.
         */
        collision = find_worker_executing_work(pool, work);
        if (unlikely(collision)) {
                move_linked_works(work, &collision->scheduled, NULL);
                return;
        }

        /* claim and dequeue */
        debug_work_deactivate(work);
        hash_add(pool->busy_hash, &worker->hentry, (unsigned long)work);
        worker->current_work = work;
        worker->current_func = work->func;
        worker->current_pwq = pwq;
        work_color = get_work_color(work);

        list_del_init(&work->entry);

        /*
         * CPU intensive works don't participate in concurrency management.
         * They're the scheduler's responsibility.  This takes @worker out
         * of concurrency management and the next code block will chain
         * execution of the pending work items.
         */
        if (unlikely(cpu_intensive))
                worker_set_flags(worker, WORKER_CPU_INTENSIVE);

        /*
         * Wake up another worker if necessary.  The condition is always
         * false for normal per-cpu workers since nr_running would always
         * be >= 1 at this point.  This is used to chain execution of the
         * pending work items for WORKER_NOT_RUNNING workers such as the
         * UNBOUND and CPU_INTENSIVE ones.
         */
        if (need_more_worker(pool))
                wake_up_worker(pool);
  • 코드 라인 7~11에서 워터풀내에서 요청한 워크가 이미 처리중인 경우 처리중인 워커의 스케줄드 리스트에 추가하고 함수를 빠져나간다.
    • 같은 워크 요청은 하나의 cpu에서 여러 개의 워커에서 동작못하게 막는다.
  • 코드 라인 15~18에서 busy_hash에 work를 키로 인덱스를 결정하고 워커를 추가한다. 추가할 때 워커에 처리중인 함수와 워크 및 풀워크큐를 지정한다.
  • 코드 라인 19에서 현재 작업의 워크 컬러를 알아온다.
  • 코드 라인 21에서 워크를 기존 리스트에서 제거한다.
  • 코드 라인 29~30에서 cpu intensive 워크큐를 이용하는 경우 워커에 cpu intensive 플래그를 설정한다.
  • 코드 라인 40~41에서 워커가 더 필요한 경우 대기중인 첫 번째 idle 워커를 깨운다.

 

kernel/workqueue.c – 3/4

.       /*
         * Record the last pool and clear PENDING which should be the last
         * update to @work.  Also, do this inside @pool->lock so that
         * PENDING and queued state changes happen together while IRQ is
         * disabled.
         */
        set_work_pool_and_clear_pending(work, pool->id);

        spin_unlock_irq(&pool->lock);

        lock_map_acquire_read(&pwq->wq->lockdep_map);
        lock_map_acquire(&lockdep_map);
        trace_workqueue_execute_start(work);
        worker->current_func(work);
        /*
         * While we must be careful to not use "work" after this, the trace
         * point will only record its address.
         */
        trace_workqueue_execute_end(work);
        lock_map_release(&lockdep_map);
        lock_map_release(&pwq->wq->lockdep_map);

        if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
                pr_err("BUG: workqueue leaked lock or atomic: %s/0x%08x/%d\n"
                       "     last function: %pf\n",
                       current->comm, preempt_count(), task_pid_nr(current),
                       worker->current_func);
                debug_show_held_locks(current);
                dump_stack();
        }
  • 코드 라인 7에서 워크에 워커풀 id를 기록하고 pending 비트를 포함한 나머지 플래그들도 모두 지운다.
  • 코드 라인 14에서 워커에 지정된 처리 함수를 호출한다. (워크에 지정된 함수)
  • 코드 라인 23~30에서 PREEMPT_ACTIVE를 뺀 preempt 카운터가 0이 아니면 즉, preempt가 disable된 경우 에러 메시지를 출력하고 스택을 덤프한다.

 

kernel/workqueue.c – 4/4

        /*
         * The following prevents a kworker from hogging CPU on !PREEMPT
         * kernels, where a requeueing work item waiting for something to
         * happen could deadlock with stop_machine as such work item could
         * indefinitely requeue itself while all other CPUs are trapped in
         * stop_machine. At the same time, report a quiescent RCU state so
         * the same condition doesn't freeze RCU.
         */
        cond_resched_rcu_qs();

        spin_lock_irq(&pool->lock);

        /* clear cpu intensive status */
        if (unlikely(cpu_intensive))
                worker_clr_flags(worker, WORKER_CPU_INTENSIVE);

        /* we're done with it, release */
        hash_del(&worker->hentry);
        worker->current_work = NULL;
        worker->current_func = NULL;
        worker->current_pwq = NULL;
        worker->desc_valid = false;
        pwq_dec_nr_in_flight(pwq, work_color);
}
  • 코드 라인 9에서 리스케줄 요청이 있는 경우 스케줄을 진행하고 rcu quiesecent state를 등록한다.
  • 코드 라인 14에서 워커에서 cpu intensive 플래그를 제거한다.
  • 코드 라인 17에서 busy_hash에서 워커를 제거한다.
  • 코드 라인 18~21에서 워커에 설정한 워크 정보를 초기화한다.
  • 코드 라인 22에서 워크 컬러에 해당하는 현재 처리 중인 워크 수를 1 감소시킨다.

 

워크

static 워크 생성

컴파일 타임 주요 선언 및 초기 매크로

  • DECLARE_WORK()
  • INIT_WORK()

 

컴파일 타임 기타 선언 및 초기 매크로

  • DECLARE_DELAYED_WORK()
  • DECLARE_DEFERABLE_WORK()
  • INIT_WORK_ONSTACK()
  • INIT_DELAYED_WORK()
  • INIT_DELAYED_WORK_ONSTACK()
  • INIT_DEFERRABLE_WORK()
  • INIT_DEFERRABLE_WORK_ONSTACK()

 

DECLARE_WORK()

include/linux/workqueue.h

#define DECLARE_WORK(n, f)                                              \
        struct work_struct n = __WORK_INITIALIZER(n, f)

컴파일 타임에 워크를 static 하게 선언하고 워크 함수를 지정한다.

 

 

#define __WORK_INITIALIZER(n, f) {                                      \
        .data = WORK_DATA_STATIC_INIT(),                                \
        .entry  = { &(n).entry, &(n).entry },                           \
        .func = (f),                                                    \
        __WORK_INIT_LOCKDEP_MAP(#n, &(n))                               \
        }

워크 구조체의 주요 멤버를 설정하고 초기화한다.

  • data 멤버의 초기 설정 값으로 아직 워크풀이 지정되지 않았고 컴파일 타임에 static하게 생성되었음을 알린다.
  • entry 멤버는 아직 워크풀에 등록되지 않았으므로 자기 자신을 가리키게 초기화한다.
  • func 멤버는 실행될 함수를 지정한다.

 

#define WORK_DATA_STATIC_INIT() \
        ATOMIC_LONG_INIT(WORK_STRUCT_NO_POOL | WORK_STRUCT_STATIC)

워크의 data 멤버의 초기 설정 값으로 아직 워크풀이 지정되지 않았고 컴파일 타임에 static하게 생성되었음을 알린다.

  • WORK_STRUCT_NO_POOL 값은 플래그들이 있는 lsb 몇 비트를 제외하고 pool id에 해당하는 모든 비트가 1로 설정된다.
  • WORK_STRUCT_STATIC 플래그는 CONFIG_DEBUG_OBJECT 커널 옵션을 사용하는 경우에만 bit4에 해당하는 플래그가 추가되고 1로 설정된다.

 

dynamic 워크 생성

INIT_WORK()

include/linux/workqueue.h

#define INIT_WORK(_work, _func)                                         \
        __INIT_WORK((_work), (_func), 0)

요청한 워크에 워크 함수를 지정하고 초기화한다.  (런타임에 사용)

 

 

#define __INIT_WORK(_work, _func, _onstack)                             \
        do {                                                            \
                __init_work((_work), _onstack);                         \
                (_work)->data = (atomic_long_t) WORK_DATA_INIT();       \
                INIT_LIST_HEAD(&(_work)->entry);                        \
                (_work)->func = (_func);                                \
        } while (0)
#endif

런타임에 요청한 워크에 함수를 지정하고 초기화한다.

  • data 멤버에는초기 설정 값으로 아직 워크풀이 지정되지 않았음을 알린다.
  • entry 멤버는 아직 워크풀에 등록되지 않았으므로 자기 자신을 가리키게 초기화한다.
  • func 멤버는 실행될 함수를 지정한다.

 

#define WORK_DATA_INIT()        ATOMIC_LONG_INIT(WORK_STRUCT_NO_POOL)

워크의 data 멤버 초기값으로 pool id 필드를 WORK_STRUCT_NO_POOL로 설정하여 아직 워크풀이 지정되지 않았음을 알린다.

  • WORK_STRUCT_NO_POOL 값은 플래그들이 있는 lsb 몇 비트를 제외하고 pool id에 해당하는 모든 비트가 1로 설정된다.

 

워크를 워크큐에 엔큐

queue_work()

include/linux/workqueue.h

/**
 * queue_work - queue work on a workqueue
 * @wq: workqueue to use
 * @work: work to queue
 *
 * Returns %false if @work was already on a queue, %true otherwise.
 *
 * We queue the work to the CPU on which it was submitted, but if the CPU dies
 * it can be processed by another CPU.
 */
static inline bool queue_work(struct workqueue_struct *wq,
                              struct work_struct *work)
{
        return queue_work_on(WORK_CPU_UNBOUND, wq, work);
}

워크큐에 워크를 엔큐한다. 가능하면 현재 cpu에 작업을 시키도록 요청한다. 이미 워크큐에 등록되어 있는 상태이면 실패를 반환한다.

 

queue_work_on()

kernel/workqueue.c

/**
 * queue_work_on - queue work on specific cpu
 * @cpu: CPU number to execute work on
 * @wq: workqueue to use
 * @work: work to queue
 *
 * We queue the work to a specific CPU, the caller must ensure it
 * can't go away.
 *
 * Return: %false if @work was already on a queue, %true otherwise.
 */
bool queue_work_on(int cpu, struct workqueue_struct *wq,
                   struct work_struct *work)
{
        bool ret = false;
        unsigned long flags;

        local_irq_save(flags);

        if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work))) {
                __queue_work(cpu, wq, work);
                ret = true;   
        }
        
        local_irq_restore(flags);
        return ret;
}
EXPORT_SYMBOL(queue_work_on);

워크가 이미 등록된 상태가 아니면 워크큐에 워크를 엔큐한다. 가능하면 현재 cpu에 작업을 시키도록 요청한다. 이미 워크큐에 등록되어 있는 상태이면 실패를 반환한다.

  • 워크에 pending 플래그를 보고 워크가 이미 워크큐에 엔큐되어 있다는 것을 알 수 있다.

 

__queue_work()

워크를 워크큐에 엔큐한다.

kernel/workqueue.c – 1/2

static void __queue_work(int cpu, struct workqueue_struct *wq,
                         struct work_struct *work)
{
        struct pool_workqueue *pwq;
        struct worker_pool *last_pool;
        struct list_head *worklist;
        unsigned int work_flags;
        unsigned int req_cpu = cpu;

        /*
         * While a work item is PENDING && off queue, a task trying to
         * steal the PENDING will busy-loop waiting for it to either get
         * queued or lose PENDING.  Grabbing PENDING and queueing should
         * happen with IRQ disabled.
         */
        WARN_ON_ONCE(!irqs_disabled());

        debug_work_activate(work);

        /* if draining, only works from the same workqueue are allowed */
        if (unlikely(wq->flags & __WQ_DRAINING) &&
            WARN_ON_ONCE(!is_chained_work(wq)))
                return;
retry:
        if (req_cpu == WORK_CPU_UNBOUND)
                cpu = raw_smp_processor_id();

        /* pwq which will be used unless @work is executing elsewhere */
        if (!(wq->flags & WQ_UNBOUND))
                pwq = per_cpu_ptr(wq->cpu_pwqs, cpu);
        else
                pwq = unbound_pwq_by_node(wq, cpu_to_node(cpu));

        /*
         * If @work was previously on a different pool, it might still be
         * running there, in which case the work needs to be queued on that
         * pool to guarantee non-reentrancy.
         */
        last_pool = get_work_pool(work);
        if (last_pool && last_pool != pwq->pool) {
                struct worker *worker;

                spin_lock(&last_pool->lock);

                worker = find_worker_executing_work(last_pool, work);

                if (worker && worker->current_pwq->wq == wq) {
                        pwq = worker->current_pwq;
                } else {
                        /* meh... not running there, queue here */
                        spin_unlock(&last_pool->lock);
                        spin_lock(&pwq->pool->lock);
                }
        } else {
                spin_lock(&pwq->pool->lock);
        }
  • 코드 라인 21~23에서 draining 플래그가 있는 워크큐이면서 워크큐의 워커가 현재 실행중이 아니면 경고 메시지를 출력하고 함수를 빠져나간다.
  • 코드 라인 25~26에서 WORK_CPU_UNBOUND cpu로 요청한 경우 일단 현재 cpu로 설정한다.
  • 코드 라인 29~32에서 unbound 워크큐가 아닌 경우 요청 cpu에 대한 풀워크큐를 알아온다. unbound 워크큐인 경우 현재 cpu 노드의 풀워크큐를 알아온다.
    • wq->cpu_pwqs:
      • cpu 바운드 워크큐인 경우 cpu별로 사용
    •  wq->numa_pwq_tbl[node]:
      • unbound 워크큐이면 노드별로 사용
  • 코드 라인 39~45에서 워크가 마지막으로 동작했었던 워커풀이 조금전에 알아온 풀워크큐가 가리키는 워커풀과 다른 경우 워크가 동작했었던 마지막 풀에서 요청한 워크가 현재 동작중인 워커를 알아온다.
  • 코드 라인 47~53에서 워커가 현재 동작중인 풀워크큐의 워커인 경우 워커에서 현재 동작하는 풀워크큐를 그대로 사용한다.
    • 워크의 재진입을 허용하지 않게 하는 것을 보장하기 위해 기존 워크가 수행되었던 워커의 풀워크큐를 그대로 사용한다.

kernel/workqueue.c – 2/2

        /*
         * pwq is determined and locked.  For unbound pools, we could have
         * raced with pwq release and it could already be dead.  If its
         * refcnt is zero, repeat pwq selection.  Note that pwqs never die
         * without another pwq replacing it in the numa_pwq_tbl or while
         * work items are executing on it, so the retrying is guaranteed to
         * make forward-progress.
         */
        if (unlikely(!pwq->refcnt)) {
                if (wq->flags & WQ_UNBOUND) {
                        spin_unlock(&pwq->pool->lock);
                        cpu_relax();
                        goto retry;
                }
                /* oops */
                WARN_ONCE(true, "workqueue: per-cpu pwq for %s on cpu%d has 0 refcnt",
                          wq->name, cpu);
        }

        /* pwq determined, queue */
        trace_workqueue_queue_work(req_cpu, pwq, work);

        if (WARN_ON(!list_empty(&work->entry))) {
                spin_unlock(&pwq->pool->lock);
                return;
        }

        pwq->nr_in_flight[pwq->work_color]++;
        work_flags = work_color_to_flags(pwq->work_color);

        if (likely(pwq->nr_active < pwq->max_active)) {
                trace_workqueue_activate_work(work);
                pwq->nr_active++;
                worklist = &pwq->pool->worklist;
        } else {
                work_flags |= WORK_STRUCT_DELAYED;
                worklist = &pwq->delayed_works;
        }

        insert_work(pwq, work, worklist, work_flags);

        spin_unlock(&pwq->pool->lock);
}
  • 코드 라인 9~18에서 풀워크큐의 참조 카운터가 0인 경우 경고 메시지를 출력한다. 단 unbound 워크큐인경우 retry 한다.
    • 풀워크큐가 race 상황에서 release되었을 수도 있다. 그런 경우 다시 풀워크큐를 알아온다.
  • 코드 라인 23~26에서 워크가 아무 곳에도 등록되지 않은 경우 함수를 빠져나간다.
  • 코드 라인 28~29에서 워크 컬러에 해당하는 처리중인 워크 수를 증가시키고 워크 컬러 값을 플래그 값으로 변환한다.
  • 코드 라인 31~40에서 풀워크큐에 등록된 워크의 수가 최대 제한에 걸리지 않았으면 워커 수 카운터인 nr_active를 1 증가시키고 풀워크큐가 가리키는 워커풀의 워크리스트에 워크를 추가한다.  만일 최대 제한을 초과한 경우 워크에 delay 플래그를 추가한 후 풀워크큐의 delayed_works 리스트에 추가한다.
    • 최대 제한 수는 512이며 unbound 워크큐인 경우 512와 cpu*4 중 큰 수를 사용한다.

 

insert_work()

kernel/workqueue.c

/**
 * insert_work - insert a work into a pool
 * @pwq: pwq @work belongs to
 * @work: work to insert
 * @head: insertion point
 * @extra_flags: extra WORK_STRUCT_* flags to set
 *
 * Insert @work which belongs to @pwq after @head.  @extra_flags is or'd to
 * work_struct flags.
 *
 * CONTEXT:
 * spin_lock_irq(pool->lock).
 */
static void insert_work(struct pool_workqueue *pwq, struct work_struct *work,
                        struct list_head *head, unsigned int extra_flags)
{
        struct worker_pool *pool = pwq->pool;

        /* we own @work, set data and link */
        set_work_pwq(work, pwq, extra_flags);
        list_add_tail(&work->entry, head);
        get_pwq(pwq);

        /*
         * Ensure either wq_worker_sleeping() sees the above
         * list_add_tail() or we see zero nr_running to avoid workers lying
         * around lazily while there are works to be processed.
         */
        smp_mb();

        if (__need_more_worker(pool))
                wake_up_worker(pool);
}

요청한 리스트에 워크를 추가한다.

  • 코드 라인 20에서 워크 데이터가 풀워크큐를 가리키게 하고 요청한 플래그들을 추가한다.
  • 코드 라인 21에서 요청한 리스트의 후미에 워크를 추가한다.
  • 코드 라인 22에서 풀워크큐의 참조카운터를 1 증가시킨다.
  • 코드 라인 31~32에서 워커풀에서 동작중인 워커가 없는 경우 idle 워커를 깨운다.

 

set_work_pwq()

kernel/workqueue.c

static void set_work_pwq(struct work_struct *work, struct pool_workqueue *pwq,
                         unsigned long extra_flags)
{
        set_work_data(work, (unsigned long)pwq,
                      WORK_STRUCT_PENDING | WORK_STRUCT_PWQ | extra_flags);
}

워크에 풀워크큐를 지정하고 요청한 플래그 이외에도 기본적으로 WORK_STRUCT_PENDING 및 WORK_STRUCT_PWQ 를 추가한다.

 

set_work_data()

kernel/workqueue.c

/*
 * While queued, %WORK_STRUCT_PWQ is set and non flag bits of a work's data
 * contain the pointer to the queued pwq.  Once execution starts, the flag
 * is cleared and the high bits contain OFFQ flags and pool ID.
 *
 * set_work_pwq(), set_work_pool_and_clear_pending(), mark_work_canceling()
 * and clear_work_data() can be used to set the pwq, pool or clear
 * work->data.  These functions should only be called while the work is
 * owned - ie. while the PENDING bit is set.
 *
 * get_work_pool() and get_work_pwq() can be used to obtain the pool or pwq
 * corresponding to a work.  Pool is available once the work has been
 * queued anywhere after initialization until it is sync canceled.  pwq is
 * available only while the work item is queued.
 *
 * %WORK_OFFQ_CANCELING is used to mark a work item which is being
 * canceled.  While being canceled, a work item may have its PENDING set
 * but stay off timer and worklist for arbitrarily long and nobody should
 * try to steal the PENDING bit.
 */
static inline void set_work_data(struct work_struct *work, unsigned long data,
                                 unsigned long flags)
{
        WARN_ON_ONCE(!work_pending(work));
        atomic_long_set(&work->data, data | flags | work_static(work));
}

워크에 data(풀워크큐 또는 pool id)와 플래그를 더해 설정한다.

 

구조체

workqueue_struct 구조체

kernel/workqueue.c

/*
 * The externally visible workqueue.  It relays the issued work items to
 * the appropriate worker_pool through its pool_workqueues.
 */
struct workqueue_struct {
        struct list_head        pwqs;           /* WR: all pwqs of this wq */
        struct list_head        list;           /* PL: list of all workqueues */

        struct mutex            mutex;          /* protects this wq */
        int                     work_color;     /* WQ: current work color */
        int                     flush_color;    /* WQ: current flush color */
        atomic_t                nr_pwqs_to_flush; /* flush in progress */
        struct wq_flusher       *first_flusher; /* WQ: first flusher */
        struct list_head        flusher_queue;  /* WQ: flush waiters */
        struct list_head        flusher_overflow; /* WQ: flush overflow list */

        struct list_head        maydays;        /* MD: pwqs requesting rescue */
        struct worker           *rescuer;       /* I: rescue worker */

        int                     nr_drainers;    /* WQ: drain in progress */
        int                     saved_max_active; /* WQ: saved pwq max_active */

        struct workqueue_attrs  *unbound_attrs; /* WQ: only for unbound wqs */
        struct pool_workqueue   *dfl_pwq;       /* WQ: only for unbound wqs */

#ifdef CONFIG_SYSFS
        struct wq_device        *wq_dev;        /* I: for sysfs interface */
#endif
#ifdef CONFIG_LOCKDEP
        struct lockdep_map      lockdep_map;
#endif
        char                    name[WQ_NAME_LEN]; /* I: workqueue name */

        /* hot fields used during command issue, aligned to cacheline */
        unsigned int            flags ____cacheline_aligned; /* WQ: WQ_* flags */
        struct pool_workqueue __percpu *cpu_pwqs; /* I: per-cpu pwqs */
        struct pool_workqueue __rcu *numa_pwq_tbl[]; /* FR: unbound pwqs indexed by node */
};
  • pwqs
    • 워크큐에 소속된 풀워크큐들
  • list
    • 모든 워크큐가 전역 workqueues 리스트에 연결될 떄 사용하는 list 노드이다.
  • work_color
    • 현재 워크 컬러
  • flush_color
    • 현재 플러시 컬러
  • nr_pwqs_to_flush
    • 플러시될 풀워크큐 수로 플러시가 진행중일 때 사용된다.
  • *first_flusher
    • 처음 플러셔(플러시 요청)
  • flusher_queue
    • 플러셔 리스트에 플러시 요청이 쌓이며 하나씩 처리하기 위해 first_flusher로 옮긴다.
    • 플러시가 완료되면 first_flusher는 null이 되고 flusher_queue 리스트도 비게된다.
  • flusher_overflow
    • 플러시 오버플로 리스트로 플러시 컬러 공간이 부족할 때 플러시 요청을 이 리스트에 추가한다.
  • maydays
    • 구조 요청한 풀워크큐 리스트
  • nr_drainers
    • 진행중인 drainer 수로 워크큐를 비워달라고 요청받으면 1 증가되고 드레이닝이 시작하고 완료되면 1 감소된다.
  • saved_max_active
    • 등록할 수 있는 최대 active 워크 수로 이 수를 초과하는 워크의 경우 워커풀로 배포하지 않고 풀워크큐의 delayed_works 리스트에서 대기하게 한다.
  • *unbound_attrs
    • 언바운드 워크큐 속성
  • *dfl_pwq
    • 디폴트 언바운드 풀워크큐를 가리키며 cpu on/off 시 numa_pwq_tbl[]의 이용이 힘들 때 잠시 fall-back용으로 사용한다.
  • wq_dev
    • sysfs 인터페이스
  • name[]
    • 워크큐명
  • flags
    • 플래그들
  • *cpu_pwqs
    • per-cpu 풀워크큐들
  • *numa_pwq_tbl[]
    • 노드별 언바운드 풀워크큐

 

workqueue_attrs 구조체

include/linux/workqueue.h

/*      
 * A struct for workqueue attributes.  This can be used to change
 * attributes of an unbound workqueue.
 *      
 * Unlike other fields, ->no_numa isn't a property of a worker_pool.  It
 * only modifies how apply_workqueue_attrs() select pools and thus doesn't
 * participate in pool hash calculations or equality comparisons.
 */
struct workqueue_attrs {
        int                     nice;           /* nice level */
        cpumask_var_t           cpumask;        /* allowed CPUs */
        bool                    no_numa;        /* disable NUMA affinity */
};
  • nice
    • nice 우선순위
  • cpumask
    • 허락된 cpu들
  • no_numa
    • NUMA 노드 정보를 disable

 

wq_flusher 구조체

kernel/workqueue.c

/*
 * Structure used to wait for workqueue flush.
 */
struct wq_flusher {
        struct list_head        list;           /* WQ: list of flushers */
        int                     flush_color;    /* WQ: flush color waiting for */
        struct completion       done;           /* flush completion */
};
  • list
    • 플러셔(플러시 요청)를 플러셔 리스트에 추가할 때 사용되는 연결 노드
  • flush_color
    • 플러시 컬러
  • done
    • 플러시 완료(completion) 대기를 위해 사용

 

pool_workqueue 구조체

kernel/workqueue.c

/*
 * The per-pool workqueue.  While queued, the lower WORK_STRUCT_FLAG_BITS
 * of work_struct->data are used for flags and the remaining high bits
 * point to the pwq; thus, pwqs need to be aligned at two's power of the
 * number of flag bits.
 */
struct pool_workqueue {
        struct worker_pool      *pool;          /* I: the associated pool */
        struct workqueue_struct *wq;            /* I: the owning workqueue */
        int                     work_color;     /* L: current color */
        int                     flush_color;    /* L: flushing color */
        int                     refcnt;         /* L: reference count */
        int                     nr_in_flight[WORK_NR_COLORS];
                                                /* L: nr of in_flight works */
        int                     nr_active;      /* L: nr of active works */
        int                     max_active;     /* L: max active works */
        struct list_head        delayed_works;  /* L: delayed works */
        struct list_head        pwqs_node;      /* WR: node on wq->pwqs */
        struct list_head        mayday_node;    /* MD: node on wq->maydays */

        /*
         * Release of unbound pwq is punted to system_wq.  See put_pwq()
         * and pwq_unbound_release_workfn() for details.  pool_workqueue
         * itself is also sched-RCU protected so that the first pwq can be
         * determined without grabbing wq->mutex.
         */
        struct work_struct      unbound_release_work;
        struct rcu_head         rcu;
} __aligned(1 << WORK_STRUCT_FLAG_BITS);
  • *pool
    • 연결된 워커풀
  • *wq
    • 워크큐
  • work_color
    • 워크 컬러로 새 워크마다 이 컬러의 워크 컬러를 사용한다.
    • 플러시 요청이 발생한 경우 다음 워크 컬러를 선택한다. (0 ~ 14이내에서 순환 증가)
  • flush_color
    • 플러시 컬러로 플러시 요청이 있는 경우 현재 사용한 워크 컬러를 지정한다. 그리고 이 플러시 컬러에 해당하는 워크들의 작업이 끝날때까지 기다린다. (플러시)
  • refcnt
    • 참조카운터
  • nr_in_flight[]
    • 컬러별 현재 워커가 처리중인 워크 수를 담는다.
  • nr_active
    • active 워크 수
    • 지연된 워크는 포함되지 않는다.
  • max_active
    • 동시 처리 제한(최대 active 워크 수)
  • delayed_works
    • 동시 처리 제한(max_active)을 초과한 워크 요청이 대기하는 리스트이다.
    • suspend PM 기능이 동작하는 경우 모든 인입되는 워크도 이 곳에서 대기된다.
  • pwqs_node
    • 노드별 풀워크큐
  • mayday_node
    • 워크큐의 maydays 리스트에 추가될 때 연결되는 리스트 노드
  • unbound_release_work
    • 풀워크큐가 내장하여 사용하는 워크이다. 언바운드 워커풀을 제거할 때 사용한다.
    • pwq_unbound_release_workfn() 함수를 호출하여 워커풀을 제거하는데 모든 워커풀이 제거된 워크큐도 함께 제거된다.
  • rcu
    • 풀워크큐 구조체를 rcu 방식으로 풀워크큐 slab 캐시로 free할 때 사용하는 rcu 노드이다.

 

worker_pool 구조체

kernel/workqueue.c

/*
 * Structure fields follow one of the following exclusion rules.
 *
 * I: Modifiable by initialization/destruction paths and read-only for
 *    everyone else.
 *
 * P: Preemption protected.  Disabling preemption is enough and should
 *    only be modified and accessed from the local cpu.
 *
 * L: pool->lock protected.  Access with pool->lock held.
 *
 * X: During normal operation, modification requires pool->lock and should
 *    be done only from local cpu.  Either disabling preemption on local
 *    cpu or grabbing pool->lock is enough for read access.  If
 *    POOL_DISASSOCIATED is set, it's identical to L.
 *
 * A: pool->attach_mutex protected.
 *
 * PL: wq_pool_mutex protected.
 *
 * PR: wq_pool_mutex protected for writes.  Sched-RCU protected for reads.
 *
 * WQ: wq->mutex protected.
 *
 * WR: wq->mutex protected for writes.  Sched-RCU protected for reads.
 *
 * MD: wq_mayday_lock protected.
 */

/* struct worker is defined in workqueue_internal.h */

 

struct worker_pool {
        spinlock_t              lock;           /* the pool lock */
        int                     cpu;            /* I: the associated cpu */
        int                     node;           /* I: the associated node ID */
        int                     id;             /* I: pool ID */
        unsigned int            flags;          /* X: flags */

        struct list_head        worklist;       /* L: list of pending works */
        int                     nr_workers;     /* L: total number of workers */

        /* nr_idle includes the ones off idle_list for rebinding */
        int                     nr_idle;        /* L: currently idle ones */

        struct list_head        idle_list;      /* X: list of idle workers */
        struct timer_list       idle_timer;     /* L: worker idle timeout */
        struct timer_list       mayday_timer;   /* L: SOS timer for workers */

        /* a workers is either on busy_hash or idle_list, or the manager */
        DECLARE_HASHTABLE(busy_hash, BUSY_WORKER_HASH_ORDER);
                                                /* L: hash of busy workers */

        /* see manage_workers() for details on the two manager mutexes */
        struct mutex            manager_arb;    /* manager arbitration */
        struct mutex            attach_mutex;   /* attach/detach exclusion */
        struct list_head        workers;        /* A: attached workers */
        struct completion       *detach_completion; /* all workers detached */

        struct ida              worker_ida;     /* worker IDs for task name */

        struct workqueue_attrs  *attrs;         /* I: worker attributes */
        struct hlist_node       hash_node;      /* PL: unbound_pool_hash node */
        int                     refcnt;         /* PL: refcnt for unbound pools */

        /*
         * The current concurrency level.  As it's likely to be accessed
         * from other CPUs during try_to_wake_up(), put it in a separate
         * cacheline.
         */
        atomic_t                nr_running ____cacheline_aligned_in_smp;

        /*
         * Destruction of pool is sched-RCU protected to allow dereferences
         * from get_work_pool().
         */
        struct rcu_head         rcu;
} ____cacheline_aligned_in_smp;
  • cpu
    • cpu 번호
  • node
    • 노드번호
  • id
    • 워커풀 id
  • flags
    • 워커풀의 플래그
  • worklist
    • 처리할 워크가 담기는 리스트
  • nr_workers
    • 워커풀에 등록된 워커 수
  • nr_idle
    • 워커풀에 대기중인 워커 수
  • idle_list
    • idle 워커 리스트
  • idle_timer
    • idle 타이머
  • mayday_timer
    • mayday 타이머
    • 0.1초 간격으로 워크들의 데드락 상황을 파악하여 rescuer_thread를 깨운다.
  • busy_hash[]
    • busy 워커들이 있는 해시 리스트
  • workers
    • 연결된 워커들 리스트
  • *detach_completion
    • 모든 워커들을 detach할 때 사용
  • worker_ida
    • 워커들 id를 발급하는 IDA 트리
  • attrs
    • 워크큐 속성(nice 및 cpumask)
    • 언바운드 워크큐는 같은 속성을 사용하는 경우 워커풀을 공유하여 사용한다.
  • hash_node
    • 언바운드 풀 해시 노드
  • refcount
    • 참조 카운터
  • nr_running
    • 워크를 처리중인 워커 수
    • 다른 cpu들에서 try_to_wake_up() 함수를 통해 atomic하게 접근되는 변수로 이용되고 캐시 라인 바이트 수만큼 정렬되어 사용된다.
  • rcu
    • 워커풀을 소멸시킬 때 rcu를 사용하여 수행한다.

 

work_struct 구조체

include/linux/workqueue.h

struct work_struct {
        atomic_long_t data;
        struct list_head entry;
        work_func_t func;
#ifdef CONFIG_LOCKDEP
        struct lockdep_map lockdep_map;
#endif
};
  • data
    • 풀워크큐 주소 또는 pool id가 저장되고 하위 플래그들이 구성된다. (본문참고)
  • entry
    • 리스트에 연결될 때 사용하는 리스트 노드
  • func
    • 작업 호출 함수

 

delayed_work 구조체

include/linux/workqueue.h

struct delayed_work {
        struct work_struct work;
        struct timer_list timer;

        /* target workqueue and CPU ->timer uses to queue ->work */
        struct workqueue_struct *wq; 
        int cpu;                    
};
  • work
    • 워크 구조체가 포함된다.
  • timer
    • 지연 타이머
  • *wq
    • 타이머가 만료되어 워크가 등록될 워크큐를 가리킨다.
  • cpu
    • 워크가 동작할 cpu 지정

 

참고

 

Interrupts -7- (Workqueue 1)

 

워크큐는 인터럽트 처리 중 bottom-half 처리를 위해 만들어졌다.  Device의 top-half 부분의 인터럽트 처리를 수행한 후 bottom-half 부분의 작업을 워크큐에 의뢰하면 워커스레드가 이를 처리한다.

  • 워크큐의 우선 순위는일반 태스크와 스케줄 경쟁하도록 디폴트 우선 순위인 nice 를 사용한다.
  • 단 highpri 워크큐는 softirq와 같은 nice -20의 우선 순위를 사용하여 일반 태스크들 중 가장 우선 순위가 높게 처리된다.
  • softirq 및 tasklet 같은 IRQ용 bottom-half 서브 시스템과는 다르게 처리 루틴에서 sleep 가능하다.

 

다음 그림은 인터럽트 처리를 워크큐에 의뢰하여 워커 스레드가 해당 work를 처리하는 과정을 보여준다.

 

워크큐의 구성원들 및 특징에 대해 알아본다.

  • 워크큐
    • 시스템이 기본 생성하는 시스템 워크큐는 초기 커널에서는 bound 및 unbound 2개의 시스템 워크큐가 있었고, 점점 늘어나 커널 v3.11 ~ v4.12 현재까지 7개를 사용하고 있다.
    • 시스템 워크큐의  특징
      • system_wq
        • Multi-CPU, multi-threaded용으로 오래 걸리지 않는 작업에 사용한다.
      • system_highpri_wq
        • 가장 높은 nice 우선 순위 -20으로 워커들이 워크를 처리한다.
      • system_long_wq
        • system_wq과 유사한 환경에서 조금 더 오래 걸리는 작업에 사용한다.
      • system_unbound_wq
        • 특정 cpu에 연결되지 않고 동일한 작업이 2개 이상 동시 처리하지 않는 성격의 작업에 사용한다.
        • 한 번에 하나씩만 처리되고, max_active 제한까지 스레드 수가 커질 수 있어서 요청한 작업은 즉각 동작한다.
      • system_freezable
        • system_wq과 유사한 환경이지만 suspend되는 경우 freeze 가능한 작업에 사용한다.
        • 작업 중 suspend 요청이 오면 다른 워크큐는 마저 다 처리하지만 이 워크큐는 그대로 freeze하고 복구(resume)한 후에 동작을 계속한다.
      • system_power_efficient_wq
        • 절전 목적으로 성능을 약간 희생하여도 되는 작업에 사용한다.
      • system_freezable_power_efficient_wq
        • 절전 목적으로 성능을 약간 희생하여도 되고 suspend 시 freeze되어도 되는 작업에 사용한다.
    • 워크큐는 1개 이상의 풀별 워크큐(pwq)를 지정하는데 워크큐의 성격에따라 사용되는 풀 수가 다르다.
      • cpu bound된 워크큐는 cpu 수 만큼
      • unbound된 워크큐는 UMA 시스템에서 1개
      • unbound된 워크큐는 NUMA 시스템에서 노드 수만큼
  • 풀별 워크큐(pwq)
    • 지연된 작업들이 대기되어 관리된다.
    • 기존 cwq가 풀워크큐로 이름이 변경되었다.
    • 풀워크큐는 1개 이상의 워커풀을 지정하여 사용한다.
  • 워커풀
    • 컴파일 타임에 생성되는 standard 워커풀은 cpu 수 만큼 NR_STD_WORKER_POOLS(2) 세트
  • 워커
    • idle 상태에 있다가 워크가 주어지면 워커스레드를 깨워 워크를 처리한다.
  • 워크
    • bottom-half 처리할 함수가 워크큐에 등록된 후 적절한 워커스레드에 전달되어 워크에 등록된 함수가 수행된다.
    • delayed 워크인 경우 곧바로 워커풀에 전달하여 처리하지 않고 타이머를 먼저 동작시킨다. 타이머가 만료되면 그 때 워크큐에 등록한다.

 

다음 그림은 워크큐와 관련된 컴포넌트간의 관계를 보여준다.

 

History

리눅스 workqueue 서브시스템은 interrupt bottom-half 처리를 위해  준비되었고 다음과 같이 진행되어 왔다.

  • 커널 v2.5.41에서 처음 소개
  • 커널 v2.6.20에서 delayed work API가 추가
  • 커널 v2.6.36에서 오늘날 사용되는 형태의 CMWQ(concurrency managed workqueue)가 소개
  • 커널 v3.8에서 워커풀 추가
  • 커널 v3.9에서 cwq 등이 없어지고 pool workqueue로 변경

 

다음 그림은 처음 소개된 워크큐의 모습이다.

 

다음 그림은 CMWQ(Concurrency Managed WorkQueue) 방식으로 구현이 변경되었을 때의 모습이다.

 

다음 그림은 워커풀이 추가되었을 때의 모습이다.

 

다음 그림은 최근 커널에서 워크큐가 구현된 모습을 보여준다.

 

워크큐별 플래그

워크큐에서 사용하는 플래그들을 알아본다.

  • WQ_UNBOUND
    • cpu를 배정하지 않고 현재 cpu에 워크를 배정하게 한다.
    • cpu별로 풀워크큐를 만들지 않고 누마 노드별 풀워크큐를 만든다.
  • WQ_FREEZABLE
    • 시스템이 절전을 위해 suspend 상태로 빠지는 경우 플러싱을 하도록한다.
    • 플러시가 진행되면 기존 워크들이 다 처리되고 새로운 워크들을 받아들이지 않는다.
  • WQ_MEM_RECLAIM
    • 메모리 부족으로 회수가 진행되는 중에도 반드시 동작해야 하는 경우 사용한다.
    • idle 워커가 부족하여 새로운 워커를 할당 받아 동작해야 하는 경우 메모리 부족으로 인해 새로운 워커를 할당받지 못하는 이러한 비상 상황에 사용하기 위해 이 플래그가 사용되어 만들어둔 rescue 워커에게 워크 실행을 의뢰한다.
  • WQ_HIGH_PRI
    • 다른 워크보다 더 빠른 우선 순위를 갖고 처리할 수 있게 한다.
    • 이 큐는 FIFO 처리를 보장한다.
  • WQ_CPU_INTENSIVE
    • 동시성 관리(concurrent management)에 참여하지 않게 한다.
    • 워크들이 진입한 순서대로 빠르게 처리하는 WQ_HIGH_PRI와 다르게 순서와 상관없이 빠르게 처리한다.
  • WQ_SYSFS
    • sysfs 인터페이스를 제공한다.
  • WQ_POWER_EFFICIENT
    • 절전 옵션에 따라 성능을 희생해도 되는 작업을 위해 사용된다.
    • CONFIG_WQ_POWER_EFFICIENT_DEFAULT 커널 옵션을 사용하거나 “power_efficient” 모듈 파라메터를 설정하는 경우 절전을 위해 언바운드 큐로 동작한다.
  • __WQ_ORDERED
    • 워크풀에서 하나의 워커만 처리하도록 제한하여 워크의 FIFO 처리 순서를 보장한다.

 

컴파일 타임 시스템 생성 워크큐들

max_active는 워커풀에서 동시 처리 가능 워크 수이다.

그 외 런타임 생성 워크큐들

 

워크큐

워크큐 초기화

init_workqueues()

시스템용 워크큐들과 워커를 생성하고 초기화한다.

kernel/workqueue.c – 1/2

static int __init init_workqueues(void)
{
        int std_nice[NR_STD_WORKER_POOLS] = { 0, HIGHPRI_NICE_LEVEL };
        int i, cpu;

        WARN_ON(__alignof__(struct pool_workqueue) < __alignof__(long long));

        pwq_cache = KMEM_CACHE(pool_workqueue, SLAB_PANIC);

        cpu_notifier(workqueue_cpu_up_callback, CPU_PRI_WORKQUEUE_UP);
        hotcpu_notifier(workqueue_cpu_down_callback, CPU_PRI_WORKQUEUE_DOWN);

        wq_numa_init();

        /* initialize CPU pools */
        for_each_possible_cpu(cpu) {
                struct worker_pool *pool;

                i = 0;
                for_each_cpu_worker_pool(pool, cpu) {
                        BUG_ON(init_worker_pool(pool));
                        pool->cpu = cpu;
                        cpumask_copy(pool->attrs->cpumask, cpumask_of(cpu));
                        pool->attrs->nice = std_nice[i++];
                        pool->node = cpu_to_node(cpu);

                        /* alloc pool ID */
                        mutex_lock(&wq_pool_mutex);
                        BUG_ON(worker_pool_assign_id(pool));
                        mutex_unlock(&wq_pool_mutex);
                }
        }

        /* create the initial worker */
        for_each_online_cpu(cpu) {
                struct worker_pool *pool;

                for_each_cpu_worker_pool(pool, cpu) {
                        pool->flags &= ~POOL_DISASSOCIATED;
                        BUG_ON(!create_worker(pool));
                }
        }
  • 코드 라인 3에서 표준 워커풀에서 사용할  nice 우선 순위를 배열로 정의한다. 표준 워커풀은 컴파일 타임에 생성된 기본 2개를 사용하고 디폴트 nice 0과 최고 우선 순위의 nice -20을 사용한다.
  • 코드 라인 8에서 pool_workqueue 구조체에 사용할 객체를 slab 캐시로부터 할당받아온다.
  • 코드 라인 10에서 cpu notifier 리스트에 workqueue_cpu_up_callback() 함수를 CPU_PRI_WORKQUEUE_UP(5) 우선 순위로 등록한다.
    • 우선 순위 값이 클 수록 선두에 등록된다.
  • 코드 라인 11에서 hot cpu notifier 리스트에 workqueue_cpu_down_callback() 함수를 CPU_PRI_WORKQUEUE_DOWN(-5) 우선 순위로 등록한다.
  • 코드 라인 13에서 NUMA 시스템인 경우 possible cpumask 배열과 워크큐 속성을 할당받고 초기화한다.
  • 코드 라인 16~32에서 cpu에 해당하는 기본 워커풀 수 만큼 순회하면서 컴파일 타임에 만들어진 기본 2개의 워커풀을 초기화한다.
    • 워커풀은 컴파일 타임에 cpu 수 만큼 각각 NR_STD_WORKER_POOLS(2)개가 생성된다.
    • 워커풀의 cpu, 워커풀 속성의 cpumask, 워커풀 속성의 nice 우선 순위, 워커풀의 노드를 초기화한다.
    • worker_pool_idr 트리에서 워커풀의 id를 할당받아 워크풀에 대입한다.
  • 코드 라인 35~42에서 모든 워커풀들의 속성에서 POOL_DISASSOCIATED 속성을 제거하고 각 워커풀에서 동작할 워커를 생성한다.

 

kernel/workqueue.c – 2/2

        /* create default unbound and ordered wq attrs */
        for (i = 0; i < NR_STD_WORKER_POOLS; i++) {
                struct workqueue_attrs *attrs;

                BUG_ON(!(attrs = alloc_workqueue_attrs(GFP_KERNEL)));
                attrs->nice = std_nice[i];
                unbound_std_wq_attrs[i] = attrs;

                /*
                 * An ordered wq should have only one pwq as ordering is
                 * guaranteed by max_active which is enforced by pwqs.
                 * Turn off NUMA so that dfl_pwq is used for all nodes.
                 */
                BUG_ON(!(attrs = alloc_workqueue_attrs(GFP_KERNEL)));
                attrs->nice = std_nice[i];
                attrs->no_numa = true;
                ordered_wq_attrs[i] = attrs;
        }

        system_wq = alloc_workqueue("events", 0, 0);
        system_highpri_wq = alloc_workqueue("events_highpri", WQ_HIGHPRI, 0);
        system_long_wq = alloc_workqueue("events_long", 0, 0);
        system_unbound_wq = alloc_workqueue("events_unbound", WQ_UNBOUND,
                                            WQ_UNBOUND_MAX_ACTIVE);
        system_freezable_wq = alloc_workqueue("events_freezable",
                                              WQ_FREEZABLE, 0);
        system_power_efficient_wq = alloc_workqueue("events_power_efficient",
                                              WQ_POWER_EFFICIENT, 0);
        system_freezable_power_efficient_wq = alloc_workqueue("events_freezable_power_efficient",
                                              WQ_FREEZABLE | WQ_POWER_EFFICIENT,
                                              0);
        BUG_ON(!system_wq || !system_highpri_wq || !system_long_wq ||
               !system_unbound_wq || !system_freezable_wq ||
               !system_power_efficient_wq ||
               !system_freezable_power_efficient_wq);
        return 0;
}
early_initcall(init_workqueues);
  • 코드 라인 2~18에서 언바운드 표준 워크큐 속성과 순차실행용(ordered) 워크큐 속성을 할당받고 초기화한다.
    • 순차실행용 워크큐는 pwq를 1개만 생성하여 작업이 순서대로 실행되는 것을 보장한다.
    • NUMA 시스템을 사용하지 않는 경우 모든 노드들은 오직 dfl_pwq를 사용한다.
  • 코드 라인 20에서 “events”용 워크큐를 생성한다.
  • 코드 라인 21에서 “events_highpri”용 워크큐를 생성할 때 워크큐에 WQ_HIGHPRI 플래그를 사용하여 high priority 용 워크큐임을 지정한다.
  • 코드 라인 22에서 “events_long”용 워크큐를 생성한다.
  • 코드 라인 23~24에서 “events_unbound”용 워크큐를 생성할 때 워크큐에 WQ_UNBOUND 플래그를 사용하여 최대 WQ_UNBOUND_MAX_ACTIVE(512) 수와 cpu * 4 중 큰 수의 값으로 워커 생성을 제한한다.
  • 코드 라인 25~26에서 “events_freezable”용 워크큐를 생성할 때 워크큐에 WQ_FREEZABLE 플래그를 사용하여 suspend 처리 시 freeze 처리가 가능하도록 한다.
  • 코드 라인 27~28에서 “events_power_efficient”용 워크큐를 생성할 때 워크큐에 WQ_POWER_EFFICENT 플래그를 사용하여 성능보다 전력 효율 우선으로 처리를 하도록 지정한다.
  • 코드 라인 29~31에서 “events_freezable_power_efficient”용 워크큐를 생성할 때 WQ_FREEZABLE 및 WQ_POWER_EFFICENT 플래그를 사용하여 성능보다 전력 효율 우선으로 처리를 하도록 지정한다. 추가적으로 suspend 시 처리되지 않고 얼린채로 남아있는 것을 허용한다.

 

누마에 필요한 설정 준비

wq_numa_init()

kernel/workqueue.c

static void __init wq_numa_init(void)
{
        cpumask_var_t *tbl;
        int node, cpu;

        if (num_possible_nodes() <= 1)
                return;

        if (wq_disable_numa) {
                pr_info("workqueue: NUMA affinity support disabled\n");
                return;
        }

        wq_update_unbound_numa_attrs_buf = alloc_workqueue_attrs(GFP_KERNEL);
        BUG_ON(!wq_update_unbound_numa_attrs_buf);

        /*
         * We want masks of possible CPUs of each node which isn't readily
         * available.  Build one from cpu_to_node() which should have been
         * fully initialized by now.
         */
        tbl = kzalloc(nr_node_ids * sizeof(tbl[0]), GFP_KERNEL);
        BUG_ON(!tbl);

        for_each_node(node)
                BUG_ON(!zalloc_cpumask_var_node(&tbl[node], GFP_KERNEL,
                                node_online(node) ? node : NUMA_NO_NODE));

        for_each_possible_cpu(cpu) {
                node = cpu_to_node(cpu);
                if (WARN_ON(node == NUMA_NO_NODE)) {
                        pr_warn("workqueue: NUMA node mapping not available for cpu%d, disabling NUMA support\n", cpu);
                        /* happens iff arch is bonkers, let's just proceed */
                        return;
                }
                cpumask_set_cpu(cpu, tbl[node]);
        }               
                        
        wq_numa_possible_cpumask = tbl;
        wq_numa_enabled = true;
}

워크큐를 위해 NUMA 시스템인 경우 possible cpumask 배열과 워크큐 속성을 할당받고 초기화한다.

  • 코드 라인 6~7에서 멀티 노드가 아닌 경우 함수를 빠져나간다.
  • 코드 라인 9~12에서 커널 파라메터를 통해 “workqueue.disable_numa.disable_numa”가 사용되면 함수를 빠져나간다.
    • “/module/workqueue/parameters/disable_numa” 파일
  • 코드 라인 14에서 전역 wq_update_unbound_numa_attrs_buf 변수에 워크큐 속성을 할당받아 저장한다.
    • wq_update_unbound_numa() 함수에서 언바운드 워크에 대해 이 속성을 사용한다.
  • 코드 라인 25~27에서 노드 수만큼 포인터 변수 어레이를 할당받는다. 그런 후 노드 수만큼 순회하며 cpumask를 각 누마 노드에서 할당받아온다.
    • CONFIG_CPUMASK_OFFSTACK 커널 옵션을 사용하는 경우 cpumask를 별도로 할당받고 그렇지 않은 경우 NR_CPUS 수 만큼 컴파일 타임에 만들어진 것을 사용한다.
  • 코드 라인 29~37에서 모든 possible cpu 수만큼 순회를 하며 해당 노드 테이블에 해당하는 cpu에 bit를 설정한다.
  • 코드 라인 39에서 이렇게 cpumask가 설정된 테이블을 wq_numa_possible_cpumask에 대입한다.
  • 코드 라인 40에서 워크큐용 numa 설정이 enable되었음을 알린다.

 

 

 

워크큐 생성

alloc_workqueue()

include/linux/workqueue.h

#define alloc_workqueue(fmt, flags, max_active, args...)                \
        __alloc_workqueue_key((fmt), (flags), (max_active),             \
                              NULL, NULL, ##args)

워크큐를 fmt 이름으로 생성하고 최대 워크 수를 설정한다.

 

__alloc_workqueue_key()

kernel/workqueue.c

struct workqueue_struct *__alloc_workqueue_key(const char *fmt,
                                               unsigned int flags,
                                               int max_active,
                                               struct lock_class_key *key,
                                               const char *lock_name, ...)
{
        size_t tbl_size = 0;
        va_list args;
        struct workqueue_struct *wq;
        struct pool_workqueue *pwq;

        /* see the comment above the definition of WQ_POWER_EFFICIENT */
        if ((flags & WQ_POWER_EFFICIENT) && wq_power_efficient)
                flags |= WQ_UNBOUND;

        /* allocate wq and format name */
        if (flags & WQ_UNBOUND)
                tbl_size = nr_node_ids * sizeof(wq->numa_pwq_tbl[0]);

        wq = kzalloc(sizeof(*wq) + tbl_size, GFP_KERNEL);
        if (!wq)
                return NULL;

        if (flags & WQ_UNBOUND) {
                wq->unbound_attrs = alloc_workqueue_attrs(GFP_KERNEL);
                if (!wq->unbound_attrs)
                        goto err_free_wq;
        }

        va_start(args, lock_name);
        vsnprintf(wq->name, sizeof(wq->name), fmt, args);
        va_end(args);

        max_active = max_active ?: WQ_DFL_ACTIVE;
        max_active = wq_clamp_max_active(max_active, flags, wq->name);

        /* init wq */
        wq->flags = flags;
        wq->saved_max_active = max_active;
        mutex_init(&wq->mutex);
        atomic_set(&wq->nr_pwqs_to_flush, 0);
        INIT_LIST_HEAD(&wq->pwqs);
        INIT_LIST_HEAD(&wq->flusher_queue);
        INIT_LIST_HEAD(&wq->flusher_overflow);
        INIT_LIST_HEAD(&wq->maydays);

        lockdep_init_map(&wq->lockdep_map, lock_name, key, 0);
        INIT_LIST_HEAD(&wq->list);

        if (alloc_and_link_pwqs(wq) < 0)
                goto err_free_wq;
  • 코드 라인 13~14에서 power_efficient용 워크큐를 요청한 경우 WO_UNBOUND 플래그를 추가하여 unbound 워크큐로 만들어 절전을 목적으로 성능을 약간 희생시킨다.
    • “module/workqueue/parameters/power_efficient” 파일 설정 (커널 파라메터를 통해 모듈 변수의 값을 조절할 수도 있다.)
  • 코드 라인 17~28에서 워크큐와 워크큐 속성을 할당한다. 언바운드 워크큐인 경우 워크큐에 누마 노드별 풀워크큐 포인터 용도의 어레이를 추가 할당한다.
  • 코드 라인 30~32에서 워크큐 이름을 설정한다.
  • 코드 라인 34~39에서 워크큐에 플래그와 max_active 설정을 한다. max_active가 0으로 요청된 경우 WQ_DFL_ACTIVE(256)으로 설정한다.
    • 최대 max_active는 512개로 제한된다. 언바운드 워크큐인 경우 cpu가 많은 경우 cpu x 4개 한 수가 512를 초과한 경우라도 설정할 수 있다.
  • 코드 라인 41~48에서 각각의 리스트들과 초기값들을 설정한다.
  • 코드 라인 50~51에서 풀워크큐를 할당하고 워크큐에 연결한다.

 

        /*
         * Workqueues which may be used during memory reclaim should
         * have a rescuer to guarantee forward progress.
         */
        if (flags & WQ_MEM_RECLAIM) {
                struct worker *rescuer;

                rescuer = alloc_worker(NUMA_NO_NODE);
                if (!rescuer)
                        goto err_destroy;

                rescuer->rescue_wq = wq;
                rescuer->task = kthread_create(rescuer_thread, rescuer, "%s",
                                               wq->name);
                if (IS_ERR(rescuer->task)) {
                        kfree(rescuer);
                        goto err_destroy;
                }

                wq->rescuer = rescuer;
                rescuer->task->flags |= PF_NO_SETAFFINITY;
                wake_up_process(rescuer->task);
        }

        if ((wq->flags & WQ_SYSFS) && workqueue_sysfs_register(wq))
                goto err_destroy;

        /*
         * wq_pool_mutex protects global freeze state and workqueues list.
         * Grab it, adjust max_active and add the new @wq to workqueues
         * list.
         */
        mutex_lock(&wq_pool_mutex);

        mutex_lock(&wq->mutex);
        for_each_pwq(pwq, wq)
                pwq_adjust_max_active(pwq);
        mutex_unlock(&wq->mutex);

        list_add(&wq->list, &workqueues);

        mutex_unlock(&wq_pool_mutex);

        return wq;

err_free_wq:
        free_workqueue_attrs(wq->unbound_attrs);
        kfree(wq);
        return NULL;
err_destroy:
        destroy_workqueue(wq);
        return NULL;
}
EXPORT_SYMBOL_GPL(__alloc_workqueue_key);
  • 코드 라인 5~23에서 메모리 회수시에도 동작을 보장하도록 rescuer_thread를 워커를 만들고 실행시킨다. 그리고 이 이를 워크큐에 특별한 워커로 설정해둔다.
  • 코드 라인 25~26에서 sysfs로 노출시키라는 요청을 받은 경우 워크큐를 sysfs에 등록한다.
    • 등록한 워크는 “sys/bus/workqueue/devices” 디렉토리에 생성된다.
    • 커널에서 이 요청을 사용한 디바이스는 mm/backing-dev.c – default_bdi_init() 함수에 구현되어 있다.
  • 코드 라인 36~37에서 풀워크큐 수 만큼 순회하며 max_active를 조정한다.
  • 코드 라인 40에서 워크큐를 전역 워크큐 리스트에 추가한다.

 

다음 그림은 바운드용 워크큐를 생성했을 때의 모습을 보여준다.  워커풀은 생성되지 않고 컴파일 타임에 생성되었던 워커풀과 연결된다.

 

다음 그림은 언바운드용 워크큐를 생성했을 때의 모습을 보여준다. 워커풀은 기존 언바운드용 해시 워커풀에 같은 워크큐 속성을 가진 워커풀이 있으면 같이 공유하여 사용하고 그렇지 않은 경우 생성을 해서 사용한다.

 

wq_clamp_max_active()

kernel/workqueue.c

static int wq_clamp_max_active(int max_active, unsigned int f lags,
                               const char *name)
{
        int lim = flags & WQ_UNBOUND ? WQ_UNBOUND_MAX_ACTIVE : WQ_MAX_ACTIVE;

        if (max_active < 1 || max_active > lim)
                pr_warn("workqueue: max_active %d requested for %s is out of range, clamping between %d and %d\n",
                        max_active, name, 1, lim);

        return clamp_val(max_active, 1, lim);
}

max_active는 1~512개로 제한된다. 언바운드 워크큐인 경우에 한하여 cpu가 많은 경우 cpu x 4개 한 수가 512를 초과한 경우라도 설정할 수 있다.

 

워크큐 삭제

destroy_workqueue()

워크큐에 등록된 모든 워크들을 처리하고 워크큐를 삭제할 때 소속된 워커풀도 삭제한다.

kernel/workqueue.c – 1/2

/**
 * destroy_workqueue - safely terminate a workqueue
 * @wq: target workqueue
 *
 * Safely destroy a workqueue. All work currently pending will be done first.
 */
void destroy_workqueue(struct workqueue_struct *wq)
{
        struct pool_workqueue *pwq;
        int node;

        /* drain it before proceeding with destruction */
        drain_workqueue(wq);

        /* sanity checks */
        mutex_lock(&wq->mutex);
        for_each_pwq(pwq, wq) {
                int i;

                for (i = 0; i < WORK_NR_COLORS; i++) {
                        if (WARN_ON(pwq->nr_in_flight[i])) {
                                mutex_unlock(&wq->mutex);
                                return;
                        }
                }

                if (WARN_ON((pwq != wq->dfl_pwq) && (pwq->refcnt > 1)) ||
                    WARN_ON(pwq->nr_active) ||
                    WARN_ON(!list_empty(&pwq->delayed_works))) {
                        mutex_unlock(&wq->mutex);
                        return;
                }
        }
        mutex_unlock(&wq->mutex);
  • 코드 라인 13에서 이미 등록된 워크들에 대한 드레인(플러시) 처리를 수행한다. 드레인 처리 중에는 워크류로 새로운 워크의 등록 요청을 허용하지 않는다.
  • 코드 라인 17~25에서 워크큐에서 사용하는 풀워크큐를 순회하며 처리중인 워크가 있으면 경고 메시지를 출력하고 함수를 빠져나간다.
  • 코드 라인 27~32에서 처리중인 워크가 있거나 delayed 워크들이 있는 경우 경고 메시지를 출력하고 함수를 빠져나간다.

 

kernel/workqueue.c – 1/2

        /*
         * wq list is used to freeze wq, remove from list after
         * flushing is complete in case freeze races us.
         */
        mutex_lock(&wq_pool_mutex);
        list_del_init(&wq->list);
        mutex_unlock(&wq_pool_mutex);

        workqueue_sysfs_unregister(wq);

        if (wq->rescuer) {
                kthread_stop(wq->rescuer->task);
                kfree(wq->rescuer);
                wq->rescuer = NULL;
        }

        if (!(wq->flags & WQ_UNBOUND)) {
                /*
                 * The base ref is never dropped on per-cpu pwqs.  Directly
                 * free the pwqs and wq.
                 */
                free_percpu(wq->cpu_pwqs);
                kfree(wq);
        } else {
                /*
                 * We're the sole accessor of @wq at this point.  Directly
                 * access numa_pwq_tbl[] and dfl_pwq to put the base refs.
                 * @wq will be freed when the last pwq is released.
                 */
                for_each_node(node) {
                        pwq = rcu_access_pointer(wq->numa_pwq_tbl[node]);
                        RCU_INIT_POINTER(wq->numa_pwq_tbl[node], NULL);
                        put_pwq_unlocked(pwq);
                }

                /*
                 * Put dfl_pwq.  @wq may be freed any time after dfl_pwq is
                 * put.  Don't access it afterwards.
                 */
                pwq = wq->dfl_pwq;
                wq->dfl_pwq = NULL;
                put_pwq_unlocked(pwq);
        }
}
EXPORT_SYMBOL_GPL(destroy_workqueue);
  • 코드 라인 6에서 전역 워크큐 리스트에서 요청한 워크큐를 제거한다.
  • 코드 라인 9에서 sysfs 등록된 경우 sysfs의 관련 워크큐 항목을 제거한다.
  • 코드 라인 11~15에서 rescu 워커가 등록되어 있는 워크큐인 경우 해당 스레드를 정지시키고 제거한다.
  • 코드 라인 17~23에서 언바운드 워크큐가 아닌 경우 cpu별 풀워크큐 및 워크큐를 제거한다.
  • 코드 라인 24~43에서 언바운드 워크큐인 경우 노드별 풀워크큐를 제거하고 가리키는 테이블도 rcu 방식으로 제거한다. 디폴트 풀워크큐에 대한 참조 카운터를 1 감소시키고 연결을 제거한다.

 

워크큐 드레인

drain_workqueue()

kernel/workqueue.c

/**
 * drain_workqueue - drain a workqueue
 * @wq: workqueue to drain
 *
 * Wait until the workqueue becomes empty.  While draining is in progress,
 * only chain queueing is allowed.  IOW, only currently pending or running
 * work items on @wq can queue further work items on it.  @wq is flushed
 * repeatedly until it becomes empty.  The number of flushing is detemined
 * by the depth of chaining and should be relatively short.  Whine if it
 * takes too long.
 */
void drain_workqueue(struct workqueue_struct *wq)
{
        unsigned int flush_cnt = 0;
        struct pool_workqueue *pwq;

        /*
         * __queue_work() needs to test whether there are drainers, is much
         * hotter than drain_workqueue() and already looks at @wq->flags.
         * Use __WQ_DRAINING so that queue doesn't have to check nr_drainers.
         */
        mutex_lock(&wq->mutex);
        if (!wq->nr_drainers++)
                wq->flags |= __WQ_DRAINING;
        mutex_unlock(&wq->mutex);
reflush:
        flush_workqueue(wq);

        mutex_lock(&wq->mutex);

        for_each_pwq(pwq, wq) {
                bool drained;

                spin_lock_irq(&pwq->pool->lock);
                drained = !pwq->nr_active && list_empty(&pwq->delayed_works);
                spin_unlock_irq(&pwq->pool->lock);

                if (drained)
                        continue;

                if (++flush_cnt == 10 ||
                    (flush_cnt % 100 == 0 && flush_cnt <= 1000))
                        pr_warn("workqueue %s: drain_workqueue() isn't complete after %u tries\n",
                                wq->name, flush_cnt);

                mutex_unlock(&wq->mutex);
                goto reflush;
        }

        if (!--wq->nr_drainers)
                wq->flags &= ~__WQ_DRAINING;
        mutex_unlock(&wq->mutex);
}
EXPORT_SYMBOL_GPL(drain_workqueue);

워크큐에 있는 워크를 모두 처리하여 비울때까지 기다린다. 드레인 처리 중에는 현재 워크큐로 요청하는 워크들의 등록 요청은 모두 거절된다.

  • 코드 라인 23~24에서 드레이너 수가 0인 경우 워크큐 플래그에 __WQ_DRAINING을 추가한다. 드레이너 수를 1 증가시킨다.
  • 코드 라인 27에서 워크큐를 플러시한다.
  • 코드 라인 31~39에서 워크큐에 소속된 풀워크큐들에 대해 순회하며 처리할 워크가 없는 경우 skip 한다.
  • 코드 라인 41~44에서 플러시 카운터를 1증가시키고 플러시 카운터가 10이거나 1000번 미만에 대해 100번마다 경고 메시지를 출력한다.
  • 코드 라인 47에서 다시 reflush 레이블로 반복한다.
  • 코드 라인 50~51에서 드레이너 수를 1 감소시키고 0이된 경우 워크큐 플래그에서 __WQ_DRAINING을 제거한다.

 

워크큐 플러시

워크 컬러 & 플러시 컬러

워크큐를 비우는데 사용한 기존 동기화 코드를 모두 지우고 동기화 없이 빠르게 플러시를 할 수 있는 방법으로 대체되었다. 다음 과정을 살펴보자.

  • 워크들이 등록될 때 마다 워크 컬러값을 부여한다.
  • 플러시 요청이 있는 경우 플러시 컬러에 워크 컬러를 대입하고 다음 워크 컬러로 증가(0 ~ 14 순환)시킨다.
    • 마지막 컬러 번호 WORK_NO_COLOR(15)는 사용하지 않는다.
  • 플러시(기존 워크 컬러) 컬러값에 해당하는 워크들이 모두 처리되도록 기다린다.

 

flush_workqueue()

kernel/workqueue.c – 1/3

/**
 * flush_workqueue - ensure that any scheduled work has run to completion.
 * @wq: workqueue to flush
 *
 * This function sleeps until all work items which were queued on entry
 * have finished execution, but it is not livelocked by new incoming ones.
 */
void flush_workqueue(struct workqueue_struct *wq)
{
        struct wq_flusher this_flusher = {
                .list = LIST_HEAD_INIT(this_flusher.list),
                .flush_color = -1,
                .done = COMPLETION_INITIALIZER_ONSTACK(this_flusher.done),
        };
        int next_color;

        lock_map_acquire(&wq->lockdep_map);
        lock_map_release(&wq->lockdep_map);

        mutex_lock(&wq->mutex);

        /*
         * Start-to-wait phase
         */
        next_color = work_next_color(wq->work_color);

        if (next_color != wq->flush_color) {
                /*
                 * Color space is not full.  The current work_color
                 * becomes our flush_color and work_color is advanced
                 * by one.
                 */
                WARN_ON_ONCE(!list_empty(&wq->flusher_overflow));
                this_flusher.flush_color = wq->work_color;
                wq->work_color = next_color;

                if (!wq->first_flusher) {
                        /* no flush in progress, become the first flusher */
                        WARN_ON_ONCE(wq->flush_color != this_flusher.flush_color);

                        wq->first_flusher = &this_flusher;

                        if (!flush_workqueue_prep_pwqs(wq, wq->flush_color,
                                                       wq->work_color)) {
                                /* nothing to flush, done */
                                wq->flush_color = next_color;
                                wq->first_flusher = NULL;
                                goto out_unlock;
                        }
                } else {
                        /* wait in queue */
                        WARN_ON_ONCE(wq->flush_color == this_flusher.flush_color);
                        list_add_tail(&this_flusher.list, &wq->flusher_queue);
                        flush_workqueue_prep_pwqs(wq, -1, wq->work_color);
                }
        } else {
                /*
                 * Oops, color space is full, wait on overflow queue.
                 * The next flush completion will assign us
                 * flush_color and transfer to flusher_queue.
                 */
                list_add_tail(&this_flusher.list, &wq->flusher_overflow);
        }

        mutex_unlock(&wq->mutex);

        wait_for_completion(&this_flusher.done);
  • 코드 라인 10~14에서 워크큐 플러셔 구조체를 준비한다.
  • 코드 라인 25에서 다음 워크큐 컬러를 가져온다.
    • 0 ~ 14번의 컬러 범위이다.
  • 코드 라인 27에서 다음 컬러가 워크큐의 플러시 컬러와 다른 경우 컬러 스페이스가 풀되지 않았음을 의미한다.
  • 코드 라인 34~35에서 요청한 워크큐의 워크 컬러는 현재 플러셔의 플러시 컬러에 백업한 후 다음 컬러를 지정한다.
  • 코드 라인 37~41에서 요청한 워크큐의 first_flusher가 지정되지 않은 경우 지금 진행되는 플러셔를 대입한다.
  • 코드 라인 43~49에서 워크큐 플러시를 위해 풀워크큐들도 준비하는데 완료되어 wakeup stage를 갈 필요 없을 경우 워크큐의 플러시 컬러를 다음 컬러로 대입하고 first_flusher에 null을 대입하고 함수를 빠져나간다.
  • 코드 라인 50~55에서 요청한 워크큐에 first_flusher가 있는 경우워크큐의 flusher_queue 리스트에 현재 플러시 요청을 추가하고 워크큐 플러시를 위해 풀워크큐들도 준비한다. 이 때 플러시컬러는 지정하지 않는다.
  • 코드 라인 56~63에서 다음 컬러 공간이 부족하면 워크큐의 flusher_overflow 리스트에 현재 플러쉬 요청을 추가한다.
    • 플러시 컬러가 한 바퀴를 돌아 현재 진행하는 워커 컬러번호가 되면 오버플로우된다.
  • 코드 라인 67에서 현재 플러시 요청이 완료될 때까지 기다린다. 다음 두 가지 함수에서 완료를 체크하여 이곳 first 플러셔를 깨운다.
    • flush_workqueue_prep_pwqs() 함수
      • 플워크큐들에서 요청한 플러시 컬러로 처리중인 워크가 없으면 완료처리한다.
    • pwq_dec_nr_in_flight() 함수
      • 마지막 워크가 처리된 경우 완료처리한다.
    • 이 함수의 2nd phase
      • 같은 플러시 번호를 사용하는 경우 같이 처리되어 완료된다.

 

kernel/workqueue.c – 2/3

.       /*
         * Wake-up-and-cascade phase
         *
         * First flushers are responsible for cascading flushes and
         * handling overflow.  Non-first flushers can simply return.
         */
        if (wq->first_flusher != &this_flusher)
                return;

        mutex_lock(&wq->mutex);

        /* we might have raced, check again with mutex held */
        if (wq->first_flusher != &this_flusher)
                goto out_unlock;

        wq->first_flusher = NULL;

        WARN_ON_ONCE(!list_empty(&this_flusher.list));
        WARN_ON_ONCE(wq->flush_color != this_flusher.flush_color);

        while (true) {
                struct wq_flusher *next, *tmp;

                /* complete all the flushers sharing the current flush color */
                list_for_each_entry_safe(next, tmp, &wq->flusher_queue, list) {
                        if (next->flush_color != wq->flush_color)
                                break;
                        list_del_init(&next->list);
                        complete(&next->done);
                }

                WARN_ON_ONCE(!list_empty(&wq->flusher_overflow) &&
                             wq->flush_color != work_next_color(wq->work_color));

                /* this flush_color is finished, advance by one */
                wq->flush_color = work_next_color(wq->flush_color);
  • 코드 라인 7~8에서 first 플러셔가 완료 요청을 받은 경우가 아니면 함수를 빠져나간다.
  • 코드 라인 13~14에서 다시 한 번 락을 건 후 first 플러셔가 완료 요청을 받은 경우가 아니면 함수를 빠져나간다.
  • 코드 라인 16에서 워크큐의 first 플러셔에 null을 대입한다.
  • 코드 라인 21~30에서 대기중인 플러셔들 중에 같은 플러시 번호로 대기중인 플러셔가 있으면 그 플러셔를 제거하고 완료 요청을 보낸다.
  • 코드 라인 36에서 워크큐에 다음 플러시 컬러를 준비한다.

 

kernel/workqueue.c – 3/3

                /* one color has been freed, handle overflow queue */
                if (!list_empty(&wq->flusher_overflow)) {
                        /*
                         * Assign the same color to all overflowed
                         * flushers, advance work_color and append to
                         * flusher_queue.  This is the start-to-wait
                         * phase for these overflowed flushers.
                         */
                        list_for_each_entry(tmp, &wq->flusher_overflow, list)
                                tmp->flush_color = wq->work_color;

                        wq->work_color = work_next_color(wq->work_color);

                        list_splice_tail_init(&wq->flusher_overflow,
                                              &wq->flusher_queue);
                        flush_workqueue_prep_pwqs(wq, -1, wq->work_color);
                }

                if (list_empty(&wq->flusher_queue)) {
                        WARN_ON_ONCE(wq->flush_color != wq->work_color);
                        break;
                }

                /*
                 * Need to flush more colors.  Make the next flusher
                 * the new first flusher and arm pwqs.
                 */
                WARN_ON_ONCE(wq->flush_color == wq->work_color);
                WARN_ON_ONCE(wq->flush_color != next->flush_color);

                list_del_init(&next->list);
                wq->first_flusher = next;

                if (flush_workqueue_prep_pwqs(wq, wq->flush_color, -1))
                        break;

                /*
                 * Meh... this color is already done, clear first
                 * flusher and repeat cascading.
                 */
                wq->first_flusher = NULL;
        }

out_unlock:
        mutex_unlock(&wq->mutex);
}
EXPORT_SYMBOL_GPL(flush_workqueue);
  • 코드 라인 2~12에서 flusher_overflow 리스트에 플러시 요청이 있는 경우 순회하며 플러시 컬러를 모두 워크큐의 워크 컬러로 치환하고 워크큐는 다음 워크컬러로 갱신한다.
  • 코드 라인 14~17에서 오버플로우 리스트에 연결된 플러시 요청들을 flusher_quque 리스트에 옮긴다. 그런 후 워크큐 플러시를 위해 풀워크큐들도 준비하는데 플러시 컬러는 지정하지 않는다.
  • 코드 라인 19~22에서 처리할 플러시 요청이 없는 경우 루프를 벗어난다.
  • 코드 라인 31~32에서 현재 처리한 플러시 요청을 리스트에서 제거하고 워크큐의 first_flusher로 옮겨 지정한다.
  • 코드 라인 34~35에서 워크큐 플러시를 위해 풀워크큐들도 준비하는데 워크큐 컬러는 지정하지 않는다. 처리 후 true이면 함수를 빠져나간다.
  • 코드 라인 41~42에서 워크큐의 first_flusher에 null을 대입한 후 계속 루프를 돈다.

 

다음 그림은 워크큐 플러시 처리과정을 보여준다.

  • 항상 첫 번째 플러시 요청자인 1st 플러셔에 대한 요청을 처리한다.
  • 2 개 이상의 플러시 요청이 있을 경우 두 번째 요청자 부터 (2nd 플러셔 부터) 플러셔 큐에서 대기시킨다.

 

풀워크큐

pool_workqueue 참조

get_pwq()

kernel/workqueue.c

/**
 * get_pwq - get an extra reference on the specified pool_workqueue
 * @pwq: pool_workqueue to get
 *
 * Obtain an extra reference on @pwq.  The caller should guarantee that
 * @pwq has positive refcnt and be holding the matching pool->lock.
 */
static void get_pwq(struct pool_workqueue *pwq)
{
        lockdep_assert_held(&pwq->pool->lock);
        WARN_ON_ONCE(pwq->refcnt <= 0);
        pwq->refcnt++;
}

pwq 참조 카운터를 증가시킨다.

 

put_pwq()

kernel/workqueue.c

/**
 * put_pwq - put a pool_workqueue reference
 * @pwq: pool_workqueue to put
 *
 * Drop a reference of @pwq.  If its refcnt reaches zero, schedule its
 * destruction.  The caller should be holding the matching pool->lock.
 */
static void put_pwq(struct pool_workqueue *pwq)
{
        lockdep_assert_held(&pwq->pool->lock);
        if (likely(--pwq->refcnt))
                return;
        if (WARN_ON_ONCE(!(pwq->wq->flags & WQ_UNBOUND)))
                return;
        /*
         * @pwq can't be released under pool->lock, bounce to
         * pwq_unbound_release_workfn().  This never recurses on the same
         * pool->lock as this path is taken only for unbound workqueues and
         * the release work item is scheduled on a per-cpu workqueue.  To
         * avoid lockdep warning, unbound pool->locks are given lockdep
         * subclass of 1 in get_unbound_pool().
         */
        schedule_work(&pwq->unbound_release_work);
}

pwq 참조 카운터를 감소시킨다. 만일 참조 카운터가 0이되면 unbound_release_work 작업을 스케줄한다.

  • pwq_unbound_release_workfn() 함수를 실행시켜 pwq를 제거한다.

 

pwq_unbound_release_workfn()

kernel/workqueue.c

/*
 * Scheduled on system_wq by put_pwq() when an unbound pwq hits zero refcnt
 * and needs to be destroyed.
 */
static void pwq_unbound_release_workfn(struct work_struct *work)
{
        struct pool_workqueue *pwq = container_of(work, struct pool_workqueue,
                                                  unbound_release_work);
        struct workqueue_struct *wq = pwq->wq;
        struct worker_pool *pool = pwq->pool;
        bool is_last;

        if (WARN_ON_ONCE(!(wq->flags & WQ_UNBOUND)))
                return;

        mutex_lock(&wq->mutex);
        list_del_rcu(&pwq->pwqs_node);
        is_last = list_empty(&wq->pwqs);
        mutex_unlock(&wq->mutex);

        mutex_lock(&wq_pool_mutex);
        put_unbound_pool(pool);
        mutex_unlock(&wq_pool_mutex);

        call_rcu_sched(&pwq->rcu, rcu_free_pwq);

        /*
         * If we're the last pwq going away, @wq is already dead and no one
         * is gonna access it anymore.  Free it.
         */
        if (is_last) {
                free_workqueue_attrs(wq->unbound_attrs);
                kfree(wq);
        }
}

언바운드 풀워크큐를 제거한다. 만일 워크큐에 등록된 풀워크큐가 하나도 없는 경우 워크큐도 제거한다.

  • 코드 라인 13~14에서 unbound 워크큐와 연결된 풀워크큐인 경우 경고 메시지를 출력한 후 함수를 빠져나간다.
  • 코드 라인 17에서 워크큐에 등록된 풀워크큐를 rcu를 사용하여 제거요청을 한다.
  • 코드 라인 18에서 삭제한 풀워크큐가 워크큐에서 마지막이었던 경우 true, 아니면 false를 대입한다.
  • 코드 라인 22에서 언바운드 워크풀의 참조카운터를 1 감소시키고 0이되면 언바운드 워커풀을 소멸시킨다.
  • 코드 라인 25에서 rcu를 사용하여 풀워크큐에 사용한 pool_workqueue 객체를 pwq 캐시에서 할당 해제한다.
  • 코드 라인 31~34에서 제거한 풀워크큐가 마지막 이었으면 워크큐 속성과 워크큐 마저도 할당 해제한다.

 

flush_workqueue_prep_pwqs()

kernel/workqueue.c

/**
 * flush_workqueue_prep_pwqs - prepare pwqs for workqueue flushing
 * @wq: workqueue being flushed
 * @flush_color: new flush color, < 0 for no-op
 * @work_color: new work color, < 0 for no-op
 *
 * Prepare pwqs for workqueue flushing.
 *
 * If @flush_color is non-negative, flush_color on all pwqs should be
 * -1.  If no pwq has in-flight commands at the specified color, all
 * pwq->flush_color's stay at -1 and %false is returned.  If any pwq
 * has in flight commands, its pwq->flush_color is set to
 * @flush_color, @wq->nr_pwqs_to_flush is updated accordingly, pwq
 * wakeup logic is armed and %true is returned.
 *
 * The caller should have initialized @wq->first_flusher prior to
 * calling this function with non-negative @flush_color.  If
 * @flush_color is negative, no flush color update is done and %false
 * is returned.
 *
 * If @work_color is non-negative, all pwqs should have the same
 * work_color which is previous to @work_color and all will be
 * advanced to @work_color.
 *
 * CONTEXT:
 * mutex_lock(wq->mutex).
 *
 * Return:
 * %true if @flush_color >= 0 and there's something to flush.  %false
 * otherwise.
 */

 

static bool flush_workqueue_prep_pwqs(struct workqueue_struct *wq,
                                      int flush_color, int work_color)
{
        bool wait = false;
        struct pool_workqueue *pwq;

        if (flush_color >= 0) {
                WARN_ON_ONCE(atomic_read(&wq->nr_pwqs_to_flush));
                atomic_set(&wq->nr_pwqs_to_flush, 1);
        }

        for_each_pwq(pwq, wq) {
                struct worker_pool *pool = pwq->pool;

                spin_lock_irq(&pool->lock);

                if (flush_color >= 0) {
                        WARN_ON_ONCE(pwq->flush_color != -1);

                        if (pwq->nr_in_flight[flush_color]) {
                                pwq->flush_color = flush_color;
                                atomic_inc(&wq->nr_pwqs_to_flush);
                                wait = true;
                        }
                }

                if (work_color >= 0) {
                        WARN_ON_ONCE(work_color != work_next_color(pwq->work_color));
                        pwq->work_color = work_color;
                }

                spin_unlock_irq(&pool->lock);
        }

        if (flush_color >= 0 && atomic_dec_and_test(&wq->nr_pwqs_to_flush))
                complete(&wq->first_flusher->done);

        return wait;
}

워크큐 플러싱을 위해 소속된 풀워크큐들 플러시 준비를 한다.

  • 코드 라인 7~10에서 플러시 컬러가 지정된 경우 워크큐가 플러시중임을 알 수 있도록 nr_pwqs_to_flush를 1로 설정한다.
  • 코드 라인 12~13에서 워크큐에 등록된 모든 풀워크큐를 순회하며 연결된 워커풀을 가져온다.
  • 코드 라인 17~25에서 플러시 컬러가 지정된 경우 플러시 컬러에 해당하는 nr_in_flight가 있을 때 풀워크큐의 플러시 컬러를 갱신한다. 그런 후 워크큐의 nr_pwqs_to_flush도 증가시킨다.
  • 코드 라인 27~30에서 워크 컬러가 지정된 경우 풀워크큐의 워크컬러도 갱신한다.
  • 코드 라인 35~36에서 플러시 컬러가 지정된 경우 nr_pwqs_to_flush를 감소시키고 그 전 값이 0인 경우 워크큐의 첫 플러싱이 완료될 때가지 기다린다.

 

pwq_adjust_max_active()

kernel/workqueue.c

/**
 * pwq_adjust_max_active - update a pwq's max_active to the current setting
 * @pwq: target pool_workqueue
 *
 * If @pwq isn't freezing, set @pwq->max_active to the associated
 * workqueue's saved_max_active and activate delayed work items
 * accordingly.  If @pwq is freezing, clear @pwq->max_active to zero.
 */
static void pwq_adjust_max_active(struct pool_workqueue *pwq)
{
        struct workqueue_struct *wq = pwq->wq;
        bool freezable = wq->flags & WQ_FREEZABLE;

        /* for @wq->saved_max_active */
        lockdep_assert_held(&wq->mutex);

        /* fast exit for non-freezable wqs */
        if (!freezable && pwq->max_active == wq->saved_max_active)
                return;

        spin_lock_irq(&pwq->pool->lock);

        /*
         * During [un]freezing, the caller is responsible for ensuring that
         * this function is called at least once after @workqueue_freezing
         * is updated and visible.
         */
        if (!freezable || !workqueue_freezing) {
                pwq->max_active = wq->saved_max_active;

                while (!list_empty(&pwq->delayed_works) &&
                       pwq->nr_active < pwq->max_active)
                        pwq_activate_first_delayed(pwq);

                /*
                 * Need to kick a worker after thawed or an unbound wq's
                 * max_active is bumped.  It's a slow path.  Do it always.
                 */
                wake_up_worker(pwq->pool);
        } else {
                pwq->max_active = 0;
        }

        spin_unlock_irq(&pwq->pool->lock);
}

풀워크큐의 max_active를 조정한다. (max_active는 최대 동시 처리할 수 있는 워크 수이다.)

  • 코드 라인 12에서 WQ_FREEZABLE 플래그가 사용된 워크큐인지 여부를 알아온다.
  • 코드 라인 18~19에서 freezable 설정되지 않았으면서 워크큐와 워커풀이 동일한 max_active 설정이 된 경우 함수를 빠져나간다.
  • 코드 라인 28~29에서 freezable 설정되지 않았거나 절전이 시작되어(suspend로 인해) 워크큐가 현재 프리징이 진행중인 경우가 아니면 워크큐의 saved_max_active를 그대로 풀워크큐에도 사용한다.
  • 코드 라인 31~39에서 delayed 리스트에 워크들이 있으면서 풀워크큐의 동시 처리수(max_active)가 여유가 있으면 첫 번째 delayed 리스트의 워크를 워커풀로 옮긴 후 워커풀의 워커를 꺠운다. (max_active 제한 및 suspend 처리로 인해 지연된 작업을 워커풀에 옮겨 처리하게 한다)
  • 코드 라인 40~42에서 freezable 기능으로 인해 풀워크큐의 max_active는 0으로 설정된다.
    • 풀워크큐의 max_active를 0으로 처리하는 경우 모든 워크 요청은 곧바로 처리하지 않고 delayed 리스트에 등록한다.

 

풀워크큐들 할당 및 연결

alloc_and_link_pwqs()

kernel/workqueue.c

static int alloc_and_link_pwqs(struct workqueue_struct *wq)
{
        bool highpri = wq->flags & WQ_HIGHPRI;
        int cpu, ret;

        if (!(wq->flags & WQ_UNBOUND)) {
                wq->cpu_pwqs = alloc_percpu(struct pool_workqueue);
                if (!wq->cpu_pwqs)
                        return -ENOMEM;

                for_each_possible_cpu(cpu) {
                        struct pool_workqueue *pwq =
                                per_cpu_ptr(wq->cpu_pwqs, cpu);
                        struct worker_pool *cpu_pools =
                                per_cpu(cpu_worker_pools, cpu);

                        init_pwq(pwq, wq, &cpu_pools[highpri]);

                        mutex_lock(&wq->mutex);
                        link_pwq(pwq);
                        mutex_unlock(&wq->mutex);
                }
                return 0;
        } else if (wq->flags & __WQ_ORDERED) {
                ret = apply_workqueue_attrs(wq, ordered_wq_attrs[highpri]);
                /* there should only be single pwq for ordering guarantee */
                WARN(!ret && (wq->pwqs.next != &wq->dfl_pwq->pwqs_node ||
                              wq->pwqs.prev != &wq->dfl_pwq->pwqs_node),
                     "ordering guarantee broken for workqueue %s\n", wq->name);
                return ret;
        } else {
                return apply_workqueue_attrs(wq, unbound_std_wq_attrs[highpri]);
        }
}

풀워크큐들을 할당하고 워커풀에 연결한다.

  • bound용인 경우 cpu 수 만큼 풀워크큐를 할당하고 워커풀에 연결한다.
  • unbound용인 경우 node 수 만큼 풀워크큐를 할당하고 언바운드 해시 워커풀에서 같은 속성의 워커풀을 가져와서 공유한다. 단 같은 속성을 사용하는 워커풀이 없는 경우 워커풀과 소속된 워커를 새로 할당하여 사용한다.

 

  • 코드 라인 3에서 워크큐가 WQ_HIGHPRI 속성을 가지고 있는지 확인한다.
  • 코드 라인 6~23에서 언바운드용 워크큐가 아니면 cpu 수 만큼 풀워크큐를 할당받은 후 컴파일 타임에 만들어진 cpu별 워커풀과 연결하고 생성한 풀워크큐를 초기화하고 워커풀과 연결한 후 정상적으로 함수를 빠져나간다.
  • 코드 라인 24~30에서 워크큐가 __WQ_ORDERED 플래그가 설정된 경우 워크들이 병렬처리되면 안된다. 따라서 1개의 풀워크큐와 워커풀이 연결되게 한다.
  • 코드 라인 31~33에서 그 밖의 언바운드용 워크큐인 경우 노드 수 만큼 풀워크큐를 할당받은 후 언바운드 해시 워커풀에서 같은 속성의 워커풀과 연결한다. 만일 같은 속성의 워커풀이 없는 경우 새로운 워커풀과 워커풀용 워커를 할당하여 사용한다.

 

다음 그림은 cpu 수 만큼 bound용 풀워크큐를 할당하고 cpu별 워커풀을 연결하는 모습을 보여준다.

 

다음 그림은 노드 수 만큼 풀워크큐를 할당하고 unbound용 해시 워커풀에서 같은 속성을 사용하는 워커풀에 연결되는 모습을 보여준다.

 

풀워크큐 초기화

init_pwq()

kernel/workqueue.c

/* initialize newly alloced @pwq which is associated with @wq and @pool */
static void init_pwq(struct pool_workqueue *pwq, struct workqueue_struct *wq,
                     struct worker_pool *pool)
{
        BUG_ON((unsigned long)pwq & WORK_STRUCT_FLAG_MASK);

        memset(pwq, 0, sizeof(*pwq));

        pwq->pool = pool;
        pwq->wq = wq;
        pwq->flush_color = -1;
        pwq->refcnt = 1;
        INIT_LIST_HEAD(&pwq->delayed_works);
        INIT_LIST_HEAD(&pwq->pwqs_node);
        INIT_LIST_HEAD(&pwq->mayday_node);
        INIT_WORK(&pwq->unbound_release_work, pwq_unbound_release_workfn);
}

풀워크큐를 초기화한다.

  • 코드 라인 7에서 풀워크큐 구조체 내용을 모두 0으로 초기화한다.
  • 코드 라인 9~10에서 풀워크큐에 요청한 워커풀 및 워크큐를 대입한다.
  • 코드 라인 11에서 플러시 컬러를 -1로 지정하여 초기화한다.
  • 코드 라인 12에서 이 풀워크큐의 참조 카운터를 1로 하여 사용중으로 한다.
  • 코드 라인 13~15에서 각 리스트를 초기화한다.
  • 코드 라인 16에서 풀워크큐를 삭제하는 시스템 전용의 특수한 워크를 풀워크큐에 등록한다.
    • 마지막 풀워크큐가 삭제되면 워크큐도 삭제한다.

 

풀워크큐를 워크큐에 연결

link_pwq()

kernel/workqueue.c

/* sync @pwq with the current state of its associated wq and link it */
static void link_pwq(struct pool_workqueue *pwq)
{
        struct workqueue_struct *wq = pwq->wq;

        lockdep_assert_held(&wq->mutex);

        /* may be called multiple times, ignore if already linked */
        if (!list_empty(&pwq->pwqs_node))
                return;

        /* set the matching work_color */
        pwq->work_color = wq->work_color;
        
        /* sync max_active to the current setting */
        pwq_adjust_max_active(pwq);

        /* link in @pwq */
        list_add_rcu(&pwq->pwqs_node, &wq->pwqs);
}

풀워크큐를 워크큐에 연결한다.

  • 코드 라인 9~10에서 풀워크큐가 이미 워크큐의 풀워크큐 리스트에 등록된 경우 함수를 빠져나간다.
  • 코드 라인 13에서 워크큐의 워크 컬러 값을 풀워크큐에 대입한다.
  • 코드 라인 16에서 워크큐의 saved_max_active를 반영하여 풀워크큐의 max_active를 조정한다.
  • 코드 라인 19에서 rcu 방식으로 워크큐의 풀워크큐 리스트에 풀워크큐를 추가한다.

 

언바운드 풀워크큐 할당

alloc_unbound_pwq()

kernel/workqueue.c

/* obtain a pool matching @attr and create a pwq associating the pool and @wq */
static struct pool_workqueue *alloc_unbound_pwq(struct workqueue_struct *wq,
                                        const struct workqueue_attrs *attrs)
{
        struct worker_pool *pool;
        struct pool_workqueue *pwq;

        lockdep_assert_held(&wq_pool_mutex);

        pool = get_unbound_pool(attrs);
        if (!pool)
                return NULL;

        pwq = kmem_cache_alloc_node(pwq_cache, GFP_KERNEL, pool->node);
        if (!pwq) {
                put_unbound_pool(pool);
                return NULL;
        }

        init_pwq(pwq, wq, pool);
        return pwq;
}

언바운드 풀워크큐를 할당한다. (워커풀 및 초기 워커까지 준비한다.)

  • 코드 라인 10~12에서 언바운드 워커풀을 얻어온다. (기존에 같은 속성을 가진 워커풀을 공유하거나 새로 할당 받아온다)
  • 코드 라인 14~18에서 풀워크큐를 할당받아온다.
  • 코드 라인 20~21에서 풀워크큐를 초기화한 후 반환한다.

 

워커풀

워커풀 초기화

init_worker_pool()

kernel/workqueue.c

/**
 * init_worker_pool - initialize a newly zalloc'd worker_pool
 * @pool: worker_pool to initialize
 *
 * Initiailize a newly zalloc'd @pool.  It also allocates @pool->attrs.
 *
 * Return: 0 on success, -errno on failure.  Even on failure, all fields
 * inside @pool proper are initialized and put_unbound_pool() can be called
 * on @pool safely to release it.
 */
static int init_worker_pool(struct worker_pool *pool)
{
        spin_lock_init(&pool->lock);
        pool->id = -1;
        pool->cpu = -1;
        pool->node = NUMA_NO_NODE;
        pool->flags |= POOL_DISASSOCIATED;
        INIT_LIST_HEAD(&pool->worklist);
        INIT_LIST_HEAD(&pool->idle_list);
        hash_init(pool->busy_hash);

        init_timer_deferrable(&pool->idle_timer);
        pool->idle_timer.function = idle_worker_timeout;
        pool->idle_timer.data = (unsigned long)pool;

        setup_timer(&pool->mayday_timer, pool_mayday_timeout,
                    (unsigned long)pool);

        mutex_init(&pool->manager_arb);
        mutex_init(&pool->attach_mutex);
        INIT_LIST_HEAD(&pool->workers);

        ida_init(&pool->worker_ida);
        INIT_HLIST_NODE(&pool->hash_node);
        pool->refcnt = 1;

        /* shouldn't fail above this point */
        pool->attrs = alloc_workqueue_attrs(GFP_KERNEL);
        if (!pool->attrs)
                return -ENOMEM;
        return 0;
}

워커풀들을 초기화한다.

  • 코드 라인 14에서 워커풀의 id를 -1로 설정한다.
  • 코드 라인 15에서 연결된 cpu가 없는 -1로 설정한다.
  • 코드 라인 16에서 누마 적용되지 않은 상태(-1)로 설정한다.
  • 코드 라인 17에서 어떠한 워커와 연결되지 않은 상태로 설정한다.
  • 코드 라인 18~19에서 worklist와 idle_list 리스트를 초기화한다.
  • 코드 라인 20에서 해시용 busy_hash 리스트들을 초기화한다.
  • 코드 라인 22~24에서 idle_timer를 유예 타이머로 초기화한다.
    • 만료 시 idle_worker_timerout() 함수가 호출된다.
  • 코드 라인 26~27에서 mayday_timer를 초기화한다.
    • 만료 시 pool_mayday_timerout() 함수가 호출된다.
  • 코드 라인 31에서 워크들이 등록될 worker 리스트를 초기화한다.
  • 코드 라인 33에서 워커들을 ida radix 트리 구조로 관리하기 위해 worker_ida를 초기화한다.
  • 코드 라인 34에서 hash_node를 초기화한다.
  • 코드 라인 35에서 워커풀의 참조 카운터를 1로 설정한다.
  • 코드 라인 38~40에서 워크큐 속성을 할당받아 풀에 지정한다.

 

for_each_cpu_worker_pool()

kernel/workqueue.c

#define for_each_cpu_worker_pool(pool, cpu)                             \
        for ((pool) = &per_cpu(cpu_worker_pools, cpu)[0];               \
             (pool) < &per_cpu(cpu_worker_pools, cpu)[NR_STD_WORKER_POOLS]; \
             (pool)++)

요청한 cpu에 해당하는 워커풀을 순회한다.

  • cpu 워커풀은 컴파일 타임에 NR_STD_WORKER_POOLS(2) 만큼 생성된다.

 

워커풀 속성

alloc_workqueue_attrs()

kernel/workqueue.c

/**
 * alloc_workqueue_attrs - allocate a workqueue_attrs
 * @gfp_mask: allocation mask to use
 *
 * Allocate a new workqueue_attrs, initialize with default settings and
 * return it.
 *
 * Return: The allocated new workqueue_attr on success. %NULL on failure.
 */
struct workqueue_attrs *alloc_workqueue_attrs(gfp_t gfp_mask)
{
        struct workqueue_attrs *attrs;

        attrs = kzalloc(sizeof(*attrs), gfp_mask);
        if (!attrs)
                goto fail;
        if (!alloc_cpumask_var(&attrs->cpumask, gfp_mask))
                goto fail;

        cpumask_copy(attrs->cpumask, cpu_possible_mask);
        return attrs;
fail:
        free_workqueue_attrs(attrs);
        return NULL;
}

워크큐 속성을 할당받아 반환한다.

  • 코드 라인 14~16에서 워크큐 속성을 할당받아온다.
  • 코드 라인 17~18에서 워크큐 속성의 cpumask도 할당받거나 준비한다.
  • 코드 라인 20~21에서 possible cpu들에 대한 cpumask를 속성에 복사한다.

 

워커풀 id 할당

worker_pool_assign_id()

kernel/workqueue.c

/**
 * worker_pool_assign_id - allocate ID and assing it to @pool
 * @pool: the pool pointer of interest
 *
 * Returns 0 if ID in [0, WORK_OFFQ_POOL_NONE) is allocated and assigned
 * successfully, -errno on failure.
 */
static int worker_pool_assign_id(struct worker_pool *pool)
{
        int ret;

        lockdep_assert_held(&wq_pool_mutex); 

        ret = idr_alloc(&worker_pool_idr, pool, 0, WORK_OFFQ_POOL_NONE,
                        GFP_KERNEL);
        if (ret >= 0) { 
                pool->id = ret;
                return 0; 
        }
        return ret;
}

IDR Radix 트리인 전역 worker_pool_idr에서 워커풀의 id를 할당받은 후 워커풀의 id에 지정한다. 성공한 경우 0을 반환한다.

 

워크가 등록된 워커풀 찾기

get_work_pool()

kernel/workqueue.c

/**
 * get_work_pool - return the worker_pool a given work was associated with
 * @work: the work item of interest
 *
 * Pools are created and destroyed under wq_pool_mutex, and allows read
 * access under sched-RCU read lock.  As such, this function should be
 * called under wq_pool_mutex or with preemption disabled.
 *
 * All fields of the returned pool are accessible as long as the above
 * mentioned locking is in effect.  If the returned pool needs to be used
 * beyond the critical section, the caller is responsible for ensuring the
 * returned pool is and stays online.
 *
 * Return: The worker_pool @work was last associated with.  %NULL if none.
 */
static struct worker_pool *get_work_pool(struct work_struct *work)
{
        unsigned long data = atomic_long_read(&work->data);
        int pool_id;

        assert_rcu_or_pool_mutex();

        if (data & WORK_STRUCT_PWQ)
                return ((struct pool_workqueue *)
                        (data & WORK_STRUCT_WQ_DATA_MASK))->pool;

        pool_id = data >> WORK_OFFQ_POOL_SHIFT;
        if (pool_id == WORK_OFFQ_POOL_NONE)
                return NULL;

        return idr_find(&worker_pool_idr, pool_id);
}

요청한 워크가 등록된 워크풀을 찾아 반환한다.

  • 코드 라인 18~25에서 data 멤버를 통해 풀별 워크큐를 알아낸 후 연결된 워커풀을 반환한다.
  • 코드 라인 27~29에서 풀 id를 알아오고 그 값이 지정되지 않은 경우 null을 반환한다.
  • 코드 라인 31에서 풀 id로 워커풀을 찾아 반환한다.
    • 전역 IDR 트리인 worker_pool_idr에서 pool_id로 검색하여 등록된 worker_pool을 알아온다.

 

다음 그림은 get_work_pool() 함수가 2 가지의 방법을 사용해서 워커풀을 알아오는 모습을 보여준다.

 

다음 그림에서 워크의 data 멤버는 아래 그림과 같은 정보들을 표현하며 다음의 2 가지 정보 중 하나를 가지고 있다.

  • WORK_STRUCT_PWQ_BIT가 설정되지 않은 경우 WORK_OFFQ_POOL_SHIFT 만큼 우측으로 쉬프트한 값을 pool id 값으로 사용한다.
  • WORK_STRUCT_PWQ_BIT가 설정된 경우 pool_workqueue 구조체 주소를 지정한다.

 

언바운드 워커풀

get_unbound_pool()

kernel/workqueue.c

/**
 * get_unbound_pool - get a worker_pool with the specified attributes
 * @attrs: the attributes of the worker_pool to get
 *
 * Obtain a worker_pool which has the same attributes as @attrs, bump the
 * reference count and return it. If there already is a matching
 * worker_pool, it will be used; otherwise, this function attempts to
 * create a new one.
 *
 * Should be called with wq_pool_mutex held.
 *
 * Return: On success, a worker_pool with the same attributes as @attrs.
 * On failure, %NULL.
 */
static struct worker_pool *get_unbound_pool(const struct workqueue_attrs *attrs)
{
        u32 hash = wqattrs_hash(attrs);
        struct worker_pool *pool;
        int node;

        lockdep_assert_held(&wq_pool_mutex);

        /* do we already have a matching pool? */ 
        hash_for_each_possible(unbound_pool_hash, pool, hash_node, hash) {
                if (wqattrs_equal(pool->attrs, attrs)) {
                        pool->refcnt++;
                        return pool;
                }
        }

        /* nope, create a new one */
        pool = kzalloc(sizeof(*pool), GFP_KERNEL);
        if (!pool || init_worker_pool(pool) < 0)
                goto fail;

        lockdep_set_subclass(&pool->lock, 1);   /* see put_pwq() */
        copy_workqueue_attrs(pool->attrs, attrs);

        /*
         * no_numa isn't a worker_pool attribute, always clear it.  See
         * 'struct workqueue_attrs' comments for detail.
         */
        pool->attrs->no_numa = false;

        /* if cpumask is contained inside a NUMA node, we belong to that node */
        if (wq_numa_enabled) {
                for_each_node(node) {
                        if (cpumask_subset(pool->attrs->cpumask,
                                           wq_numa_possible_cpumask[node])) {
                                pool->node = node;
                                break;
                        }
                }
        }

        if (worker_pool_assign_id(pool) < 0)
                goto fail;

        /* create and start the initial worker */
        if (!create_worker(pool))
                goto fail;

        /* install */
        hash_add(unbound_pool_hash, &pool->hash_node, hash);

        return pool;
fail:
        if (pool)
                put_unbound_pool(pool);
        return NULL;
}

1언바운드 워커풀

  • 코드 라인 17에서 요청한 속성 값으로 hash 인덱스를 알아온다.
  • 코드 라인 24~29에서 언바운드 해시 워커풀의 hash에 해당하는 인덱스의 리스트를 순회하며 워커풀에 등록된 워크큐 속성이 요청한 워크큐 속성과 동일(nice와 cpumask 비교)한 경우 그 워커풀의 참조 카운터를 1 증가시키고 반환한다. 워크큐 속성이 같은 경우 새로운 워커풀을 생성하지 않고 기존 워커풀을 공유한다.
  • 코드 라인 32~34에서 워커풀을 할당 받은 후 초기화한다.
  • 코드 라인 37에서 요청한 워크큐 속성을 워커풀에 복사한다. (nice, cpumask, no_numa)
  • 코드 라인 43에서 워커풀의 속성 중 no_numa는 워커풀의 속성이 아니다 항상 생성 후 먼저 false로 클리어한다.
  • 코드 라인 46~54에서 wq_numa_enabled 변수가 설정된 경우 cpumask 정보에 누마 노드 정보도 포함되었음을 의미하며 이러한 경우 노드를 순회하며 워커풀의 cpumask가 wq_numa_possible_cpumask[]에 포함된 경우 워커풀에 노드 번호를 지정하고 루프를 빠져나온다.
  • 코드 라인 56~57에서 IDR Radix 트리인 전역 worker_pool_idr에서 워커풀의 id를 할당받은 후 워커풀의 id에 지정한다.
  • 코드 라인 60~61에서 워커풀내에 초기 워커를 생성한다.
  • 코드 라인 64에서 언바운드 해시 워커풀의 해시에 생성한 워커풀을 추가하고 반환한다.

 

put_unbound_pool()

kernel/workqueue.c

/**
 * put_unbound_pool - put a worker_pool
 * @pool: worker_pool to put
 *
 * Put @pool.  If its refcnt reaches zero, it gets destroyed in sched-RCU
 * safe manner.  get_unbound_pool() calls this function on its failure path
 * and this function should be able to release pools which went through,
 * successfully or not, init_worker_pool().
 *
 * Should be called with wq_pool_mutex held.
 */
static void put_unbound_pool(struct worker_pool *pool)
{
        DECLARE_COMPLETION_ONSTACK(detach_completion);
        struct worker *worker;

        lockdep_assert_held(&wq_pool_mutex);

        if (--pool->refcnt)
                return;

        /* sanity checks */
        if (WARN_ON(!(pool->cpu < 0)) ||
            WARN_ON(!list_empty(&pool->worklist)))
                return;

        /* release id and unhash */
        if (pool->id >= 0)
                idr_remove(&worker_pool_idr, pool->id);
        hash_del(&pool->hash_node);

        /*
         * Become the manager and destroy all workers.  Grabbing
         * manager_arb prevents @pool's workers from blocking on
         * attach_mutex.
         */
        mutex_lock(&pool->manager_arb);

        spin_lock_irq(&pool->lock);
        while ((worker = first_idle_worker(pool)))
                destroy_worker(worker);
        WARN_ON(pool->nr_workers || pool->nr_idle);
        spin_unlock_irq(&pool->lock);

        mutex_lock(&pool->attach_mutex);
        if (!list_empty(&pool->workers))
                pool->detach_completion = &detach_completion;
        mutex_unlock(&pool->attach_mutex);

        if (pool->detach_completion)
                wait_for_completion(pool->detach_completion);

        mutex_unlock(&pool->manager_arb);

        /* shut down the timers */
        del_timer_sync(&pool->idle_timer);
        del_timer_sync(&pool->mayday_timer);

        /* sched-RCU protected to allow dereferences from get_work_pool() */
        call_rcu_sched(&pool->rcu, rcu_free_pool);
}

언바운드 워커풀의 사용을 해제한다.

  • 코드 라인 19~20에서 워커풀의 참조카운터를 1 감소시키고 0이 아니면 함수를 빠져나간다.
  • 코드 라인 23~25에서 워커풀의 cpu가 지정되어 있지 않거나 워커풀에 워크리스트가 비어 있지 않은 경우 경고 메시지 출력과 함께 함수를 빠져나간다.
  • 코드 라인 28~30에서 worker_pool_idr 트리에서 워커풀의 id를 제거하고 언바운드 해시 워커풀에서도 제거한다.
  • 코드 라인 40~41에서 idle 워커들을 제거한다.
  • 코드 라인 46~51에서 워커풀에 워커리스트가 비어 있지 않은 경우 워커풀의 워커들이 모두 제거될 때까지 대기한다.
  • 코드 라인 56~57에서 idle 타이머와 mayday 타이머를 제거한다.
  • 코드 라인 60에서 rcu 방식으로 rcu_free_pool() 함수를 호출하여 워커풀을 제거한다.

 

may_start_working()

kernel/workqueue.c

/* Can I start working?  Called from busy but !running workers. */
static bool may_start_working(struct worker_pool *pool)
{
        return pool->nr_idle;
}

요청한 워커풀에 idle 워커가 준비되어 작업을 처리할 수 있는지 여부를 알아온다. (idle 워커 수를 반환한다.)

 

rcu_free_pool()

kernel/workqueue.c

static void rcu_free_pool(struct rcu_head *rcu)
{
        struct worker_pool *pool = container_of(rcu, struct worker_pool, rcu);

        ida_destroy(&pool->worker_ida);
        free_workqueue_attrs(pool->attrs);
        kfree(pool);
}

워커풀을 제거한다.

  • 코드 라인 5에서 워커풀의 worker_ida 트리를 제거한다.
  • 코드 라인 6~7에서 워커풀의 워크큐 속성 및 워커풀을 할당 해제한다.

 

참고

 

cpu_startup_entry()

 

Idle-task

Idle 태스크는 코어 별로 존재하는데 별도로 생성하지 않고 각 코어의 커널 부트업 프로세스 과정이 끝난 후 이 프로세스를 이용하여 Idle 태스크로 활용하도록 Idle-task 스케줄러에 연결한다.

  • idle 진입 시 폴링 커널 옵션을 사용하는 경우 cpu 절전 기능을 사용하지 않고 busy-wait을 한다.
    • CONFIG_GENERIC_IDLE_POLL_SETUP 커널 옵션 사용시 폴링한다.
  • idle 진입 시 폴링 커널 옵션을 사용하지 않는 경우 아키텍처의 idle 방법을 사용한다.
    • cpu idle 드라이버를 사용하는 경우 아래와 같이 동작하고 그렇지 않은 경우 해당 아키텍처의 기본 동작을 사용한다. armv7의 기본 동작은 wfi 이다.

 

 

cpu idle 드라이버

cpu idle 드라이버는 멀티 사용을 지원한다. cpu ilde을 사용하지 않는 경우 armv7 아키텍처는 wfi 기반을 사용한다.

  • arm 빅 드라이버 & 리틀 드라이버
    • ARMv7에서 cpu 개별 idle 진입 시 wfi 기반을 사용한다.
    • 빅/리틀 클러스터 idle 진입 시 deep-state는 “C1″으로 클러스터 단위로 전원을 내리기 위해 PM(suspend/resume) 기능을 동작시키고 timer도 stop 된다.
    • “arm,idle-state” – drivers/cpuidle/cpuidle-arm.c
    • 머신 compatible 명이 “samsung,exynos4210″인 경우 – drivers/cpuidle/cpuidle-exynos.c
  • arm64 클러스터 드라이버
    • ARMv8에서 cpu 개별 idle 진입 시 wfi 기반을 사용한다.
    • 클러스터 idle 진입 시 클러스터 단위로 전원을 내리기 위해 PM(suspend/resume) 기능을 동작시키고 timer도 stop 된다.
    • “arm,idle-state” – drivers/cpuidle/cpuidle-arm64.c
  • mips 및 ppc도 드라이버가 있지만 생략한다.

 

cpu_startup_entry()

kernel/sched/idle.c

void cpu_startup_entry(enum cpuhp_state state) 
{
        /*
         * This #ifdef needs to die, but it's too late in the cycle to
         * make this generic (arm and sh have never invoked the canary
         * init for the non boot cpus!). Will be fixed in 3.11
         */
#ifdef CONFIG_X86
        /*
         * If we're the non-boot CPU, nothing set the stack canary up
         * for us. The boot CPU already has it initialized but no harm
         * in doing it again. This is a good place for updating it, as
         * we wont ever return from this function (so the invalid
         * canaries already on the stack wont ever trigger).
         */
        boot_init_stack_canary();
#endif
        arch_cpu_idle_prepare();
        cpu_idle_loop();
}

각 코어의 커널 부트업 프로세스 과정이 끝난 후 진입되어 첫 태스크인 idle-task가 동작하고 idle 루프를 수행한다.

  • 코드 라인 18에서 idle 태스크를 처음 수행할 때 아키텍처 관련 준비할 루틴을 수행한다.
    • arm 아키텍처에서는 현재 cpu의 fiq를 enable한다.
  • 코드 라인 19에서 idle 루프를 수행한다.

 

arch_cpu_idle_prepare()

arch/arm/kernel/process.c

void arch_cpu_idle_prepare(void)
{
        local_fiq_enable();
}

현재 cpu의 fiq를 enable한다.

 

IDLE 루프

cpu_idle_loop()

idle 루프를 수행하며 규칙적으로 깨어나거나 다른 프로세서에 의해 깨어날 때 마다 preemption 여부를 체크하여 필요 시 다른 태스크로의 수행을 넘겨주기 위해 preemption 된다.

kernel/sched/idle.c

/*
 * Generic idle loop implementation
 *
 * Called with polling cleared.
 */
static void cpu_idle_loop(void)
{
        while (1) {
                /*
                 * If the arch has a polling bit, we maintain an invariant:
                 *
                 * Our polling bit is clear if we're not scheduled (i.e. if
                 * rq->curr != rq->idle).  This means that, if rq->idle has
                 * the polling bit set, then setting need_resched is
                 * guaranteed to cause the cpu to reschedule.
                 */

                __current_set_polling();
                tick_nohz_idle_enter();

                while (!need_resched()) {
                        check_pgt_cache();
                        rmb();

                        if (cpu_is_offline(smp_processor_id()))
                                arch_cpu_idle_dead();

                        local_irq_disable();
                        arch_cpu_idle_enter();

                        /*
                         * In poll mode we reenable interrupts and spin.
                         *
                         * Also if we detected in the wakeup from idle
                         * path that the tick broadcast device expired
                         * for us, we don't want to go deep idle as we
                         * know that the IPI is going to arrive right
                         * away
                         */
                        if (cpu_idle_force_poll || tick_check_broadcast_expired())
                                cpu_idle_poll();
                        else
                                cpuidle_idle_call();

                        arch_cpu_idle_exit();
                }
  • 코드 라인 18에서 무한 루프를 돌며 polling 중으로 설정한다.
  • 코드 라인 19에서 nohz idle을 지원하는 경우 tick을 발생하지 못하게 nohz idle 모드로 진입한다.
  • 코드 라인 21에서 리스케줄 요청이 없는 경우 루프를 돈다.
  • 코드 라인 25~26에서 cpu가 offline 상태인 경우 die 상태로 진입한다.
  • 코드 라인 28~29에서 현재 cpu의 irq를 disable하고 idle 시작에 대한 아키텍처 관련 처리를 수행한다.
    • arm 아키텍처에서는 idle 시작에 대한 led 밝기 처리를 변경한다.
  • 코드 라인 40~41에서 idle 상태에서 폴링이 필요하거나 현재 cpu에 대해 tick 브로드캐스트 체크가 필요한 경우 idle 폴링을 한다.
    • idle 폴링은 인터럽트를 enable하고 cpu의 전원을 끄지 않은 상태에서 리스케줄링 요청 플래그만 계속 감시한다.
  • 코드 라인 42~43에서 그렇지 않은 경우 절전 기능을 위해 deep state idle 진입을 위해 cpuidle 드라이버에 요청한다.
  • 코드 라인 45에서 idle 종료에 대한 아키텍처 관련 처리를 수행한다.
    • arm 아키텍처에서는 idle 종료에 대한 led 밝기 처리를 변경한다.

 

                /*
                 * Since we fell out of the loop above, we know
                 * TIF_NEED_RESCHED must be set, propagate it into
                 * PREEMPT_NEED_RESCHED.
                 *
                 * This is required because for polling idle loops we will
                 * not have had an IPI to fold the state for us.
                 */
                preempt_set_need_resched();
                tick_nohz_idle_exit();
                __current_clr_polling();

                /*
                 * We promise to call sched_ttwu_pending and reschedule
                 * if need_resched is set while polling is set.  That
                 * means that clearing polling needs to be visible
                 * before doing these things.
                 */
                smp_mb__after_atomic();

                sched_ttwu_pending();
                schedule_preempt_disabled();
        }
}
  • 코드 라인 9에서 리스케줄 요청이 있는 경우 이 루틴에 도착하게 된다. preemption 설정을 하는데 arm 아키텍처는 아무 일도 수행하지 않는다.
  • 코드 라인 10에서 nohz idle 모드를 벗어난다.
  • 코드 라인 11에서 폴링 플래그를 제거한다.
  • 코드 라인 21에서 대기중인 태스크를 깨운다.
  • 코드 라인 22에서 리스케줄한다.

 

 

 

IDLE 폴링

“nohlt”, “hlt” 커널 파라메터 설정

“nohlt”, “hlt” 설정 시 다음 정수 변수에 설정이 반영된다.

kernel/sched/idle.c

static int __read_mostly cpu_idle_force_poll;

 

cpu_idle_poll_setup() & cpu_idle_nopoll_setup()

kernel/sched/idle.c

#ifdef CONFIG_GENERIC_IDLE_POLL_SETUP
static int __init cpu_idle_poll_setup(char *__unused)
{
        cpu_idle_force_poll = 1;
        return 1;
}
__setup("nohlt", cpu_idle_poll_setup);

static int __init cpu_idle_nopoll_setup(char *__unused)
{
        cpu_idle_force_poll = 0;
        return 1;
}
__setup("hlt", cpu_idle_nopoll_setup);
#endif

“hlt” 커널 파라메터를 사용하는 경우 cpu_idle_force_poll=1로 설정하여 폴링을 하게한다.

“nohlt” 커널 파라메터를 사용하는 경우 cpu_idle_force_poll=0로 설정하여 폴링을 하지 못하게한다. (기본값)

 

Idle 폴링 수행

cpu_idle_poll()

kernel/sched/idle.c

static inline int cpu_idle_poll(void)
{
        rcu_idle_enter();
        trace_cpu_idle_rcuidle(0, smp_processor_id());
        local_irq_enable();
        while (!tif_need_resched() &&
                (cpu_idle_force_poll || tick_check_broadcast_expired()))
                cpu_relax();
        trace_cpu_idle_rcuidle(PWR_EVENT_EXIT, smp_processor_id());
        rcu_idle_exit();
        return 1;
}

idle 상태에 진입해 있는 중이며, 리스케줄 요청이 있을 때까지 계속 루프를 돌며 조사한다.

 

__current_set_polling()

include/linux/sched.h

static inline void __current_set_polling(void)
{
        set_thread_flag(TIF_POLLING_NRFLAG);
}

현재 스레드에 폴링 플래그를 설정한다.

 

__current_clr_polling()

include/linux/sched.h

static inline void __current_clr_polling(void)
{         
        clear_thread_flag(TIF_POLLING_NRFLAG);
}

현재 스레드에서 폴링 플래그를 클리어한다.

 

절전용 Idle 콜 호출

cpuidle_idle_call()

kernel/sched/idle.c -1/3-

/**
 * cpuidle_idle_call - the main idle function
 *
 * NOTE: no locks or semaphores should be used here
 *
 * On archs that support TIF_POLLING_NRFLAG, is called with polling
 * set, and it returns with polling set.  If it ever stops polling, it
 * must clear the polling bit.
 */
static void cpuidle_idle_call(void)
{
        struct cpuidle_device *dev = __this_cpu_read(cpuidle_devices);
        struct cpuidle_driver *drv = cpuidle_get_cpu_driver(dev);
        int next_state, entered_state;
        unsigned int broadcast;
        bool reflect;

        /*
         * Check if the idle task must be rescheduled. If it is the
         * case, exit the function after re-enabling the local irq.
         */
        if (need_resched()) {
                local_irq_enable();
                return;
        }

        /*
         * During the idle period, stop measuring the disabled irqs
         * critical sections latencies
         */
        stop_critical_timings();

        /*
         * Tell the RCU framework we are entering an idle section,
         * so no more rcu read side critical sections and one more
         * step to the grace period
         */
        rcu_idle_enter();

        if (cpuidle_not_available(drv, dev))
                goto use_default;
  • 코드 라인 22~25에서 리스케줄 요청이 있는 경우 현재 cpu의 irq를 enable하고 함수를 빠져나간다.
  • 코드 라인 31에서 irq 및 preempt 트레이싱을 위한 타이밍 측정을 잠시 멈춘다.
  • 코드 라인 38에서 idle 진입 전에 rcu를 위해 수행할 일들을 처리한다.
  • 코드 라인 40~41에서 cpu idle 드라이버가 없는 경우 아키텍처 기본 처리 루틴을 호출한다.
    • armv7에서는 특정 머신 코드를 호출하거나 특정 머신 코드가 없는 대부분의 경우 cpu_do_idle() 함수에서 wfi 동작을 수행한다.

 

kernel/sched/idle.c -2/3-

        /*
         * Suspend-to-idle ("freeze") is a system state in which all user space
         * has been frozen, all I/O devices have been suspended and the only
         * activity happens here and in iterrupts (if any).  In that case bypass
         * the cpuidle governor and go stratight for the deepest idle state
         * available.  Possibly also suspend the local tick and the entire
         * timekeeping to prevent timer interrupts from kicking us out of idle
         * until a proper wakeup interrupt happens.
         */
        if (idle_should_freeze()) {
                entered_state = cpuidle_enter_freeze(drv, dev);
                if (entered_state >= 0) {
                        local_irq_enable();
                        goto exit_idle;
                }

                reflect = false;
                next_state = cpuidle_find_deepest_state(drv, dev);
        } else {
                reflect = true;
                /*
                 * Ask the cpuidle framework to choose a convenient idle state.
                 */
                next_state = cpuidle_select(drv, dev);
        }
        /* Fall back to the default arch idle method on errors. */
        if (next_state < 0)
                goto use_default;

        /*
         * The idle task must be scheduled, it is pointless to
         * go to idle, just update no idle residency and get
         * out of this function
         */
        if (current_clr_polling_and_test()) {
                dev->last_residency = 0;
                entered_state = next_state;
                local_irq_enable();
                goto exit_idle;
        }
  • 코드 라인 10~11에서 suspend로 인한 idle로 진입하는 경우 freeze 한다.
  • 코드 라인 12~15에서 resume에 의해 깨어난 후에 cpu idle 드라이버를 사용하여 suspend 한 경우 현재 cpu의 irq를 enable 시킨 후  exit_idle 레이블로 이동한다.
  • 코드 라인 17~18에서 cpu idle 드라이버를 사용하여 진입하지 못한 경우 reflect에 false를 대입한 후 다음 이용할 적절한 0 이상의 state 인덱스를 알아온다.
  • 코드 라인 19~25에서 reflect에 false를 대입한 후 cpu idle 프레임워크가 적절한 state 인덱스를 결정한다.
  • 코드 라인 27~28에서 만일 결정된 state가 없으면 use_default 레이블로 이동하여 아키텍처 자체 idle 처리를 수행하게 한다.
  • 코드 라인 35~40에서 폴링 플래그를 클리어한 후 리스케줄 요청이 있는 경우 현재 cpu의 irq를 enable하고 exit_idle 레이블로 이동한다.

 

kernel/sched/idle.c -3/3-

        broadcast = drv->states[next_state].flags & CPUIDLE_FLAG_TIMER_STOP;

        /*
         * Tell the time framework to switch to a broadcast timer
         * because our local timer will be shutdown. If a local timer
         * is used from another cpu as a broadcast timer, this call may
         * fail if it is not available
         */
        if (broadcast &&
            clockevents_notify(CLOCK_EVT_NOTIFY_BROADCAST_ENTER, &dev->cpu))
                goto use_default;

        /* Take note of the planned idle state. */
        idle_set_state(this_rq(), &drv->states[next_state]);

        /*
         * Enter the idle state previously returned by the governor decision.
         * This function will block until an interrupt occurs and will take
         * care of re-enabling the local interrupts
         */
        entered_state = cpuidle_enter(drv, dev, next_state);

        /* The cpu is no longer idle or about to enter idle. */
        idle_set_state(this_rq(), NULL);

        if (broadcast)
                clockevents_notify(CLOCK_EVT_NOTIFY_BROADCAST_EXIT, &dev->cpu);

        /*
         * Give the governor an opportunity to reflect on the outcome
         */
        if (reflect)
                cpuidle_reflect(dev, entered_state);

exit_idle:
        __current_set_polling();

        /*
         * It is up to the idle functions to reenable local interrupts
         */
        if (WARN_ON_ONCE(irqs_disabled()))
                local_irq_enable();

        rcu_idle_exit();
        start_critical_timings();
        return;

use_default:
        /*
         * We can't use the cpuidle framework, let's use the default
         * idle routine.
         */
        if (current_clr_polling_and_test())
                local_irq_enable();
        else
                arch_cpu_idle();

        goto exit_idle;
}
  • 코드 라인 1에서 cpu idle 드라이버에서 동작시킬 state 인덱스의 플래그레 timer-stop 플래그가 있는지 여부를 알아온다.
    • timer를 사용할 수 없는 경우 deep-idle 상태에서 빠져나오려면 브로드캐스팅을 해서 알려야 한다.
  • 코드 라인 9~11에서 브로드캐스트가 필요한 상태에서 클럭이벤트 드라이버를 통해 broadcast enter에 대한 통지가 성공한 경우 use_default 레이블로 이동하여 아키텍처 기본 idle 루틴을 사용하게 한다.
    • 로컬 타이머가 셧다운되어 사용하지 못하는 상황에서는 브로드 캐스트 타이머를 사용해야 한다.
  • 코드 라인 14에서 런큐의 idle_state에 동작 시킬 cpu idle driver의 state를 대입한다.
  • 코드 라인 21에서 cpu idle 드라이버를 통해 idle에 진입한다.
  • 코드 라인 24에서런큐의 idle_state에 다시 null을 대입한다.
  • 코드 라인 26에서 브로드캐스트가 필요한 상태였었던 경우 클럭이벤트 드라이버를 통해 broadcast exit에 대한 통지를 수행한다.
  • 코드 라인 31~32에서 cpu idle 드라이버 진입 시 프레임웍이 골라준 state 인덱스를 reflect에 반영한다.
  • 코드 라인 35에서 폴링 플래그를 클리어한다.
  • 코드 라인 43에서 idle에서 빠져나온 후 rcu를 위해 수행할 일들을 처리한다.
  • 코드 라인 44에서 irq 및 preempt 트레이싱을 위해 타이밍 측정을 다시 재계한다.

 

ARM 아키텍처 관련

arch_cpu_idle_prepare()

arch/arm/kernel/process.c

void arch_cpu_idle_prepare(void)
{
        local_fiq_enable();
}

현재 cpu의 fiq를 enable 한다.

 

arch_cpu_idle_dead()

arch/arm/kernel/process.c

#ifdef CONFIG_HOTPLUG_CPU
void arch_cpu_idle_dead(void)
{
        cpu_die();
}
#endif

hotplug cpu 커널 옵션을 사용하는 경우 die 처리를 한다.

 

arch_cpu_idle_enter()

arch/arm/kernel/process.c

void arch_cpu_idle_enter(void)
{
        ledtrig_cpu(CPU_LED_IDLE_START);
#ifdef CONFIG_PL310_ERRATA_769419
        wmb(); 
#endif
}

arm 아키텍처에서 cpu idle 진입 시 led 장치의 트리거 on을 위해 호출한다.

 

arch_cpu_idle_exit()

arch/arm/kernel/process.c

void arch_cpu_idle_exit(void)
{
        ledtrig_cpu(CPU_LED_IDLE_END);
}

arm 아키텍처에서 cpu idle 처리 완료 후 led 장치의 트리거 off를 위해 호출한다.

 

arch_cpu_idle()

arch/arm/kernel/process.c

/*      
 * Called from the core idle loop.
 */
void arch_cpu_idle(void)
{
        if (arm_pm_idle)
                arm_pm_idle();
        else
                cpu_do_idle();
        local_irq_enable(); 
}

arm cpu가 idle 대기 모드로 진입하도록 한다.

  • 코드 6~7에서 (*arm_pm_idle) 후크 함수에 연결된 함수가 있으면 호출한다.
  • 코드 8~9에서 없으면 cpu_do_idle() 함수를 호출한다.
    • armv7 아키텍처에서는 cpu_v7_do_idle() 함수를 호출하여  인터럽트가 발생하기 전까지 저전력 상태에서 대기한다.

 

cpu_v7_do_idle()

arch/arm/mm/proc-v7.S

/*
 *      cpu_v7_do_idle()
 *
 *      Idle the processor (eg, wait for interrupt).
 *
 *      IRQs are already disabled.
 */
ENTRY(cpu_v7_do_idle)
        dsb                                     @ WFI may enter a low-power mode
        wfi
        ret     lr
ENDPROC(cpu_v7_do_idle)

인터럽트가 발생하기 전까지 저전력 상태에서 대기한다. 인터럽트에 의해 다시 재개된다.

 

Tick nohz idle

nohz idle로 진입할 때

tick_nohz_idle_enter()

kernel/time/tick-sched.c

/**
 * tick_nohz_idle_enter - stop the idle tick from the idle task
 *
 * When the next event is more than a tick into the future, stop the idle tick
 * Called when we start the idle loop.
 *
 * The arch is responsible of calling:
 *
 * - rcu_idle_enter() after its last use of RCU before the CPU is put
 *  to sleep.
 * - rcu_idle_exit() before the first use of RCU after the CPU is woken up.
 */
void tick_nohz_idle_enter(void)
{
        struct tick_sched *ts;

        WARN_ON_ONCE(irqs_disabled());

        /*
         * Update the idle state in the scheduler domain hierarchy
         * when tick_nohz_stop_sched_tick() is called from the idle loop.
         * State will be updated to busy during the first busy tick after
         * exiting idle.
         */             
        set_cpu_sd_state_idle();

        local_irq_disable();

        ts = this_cpu_ptr(&tick_cpu_sched);
        ts->inidle = 1;
        __tick_nohz_idle_enter(ts);

        local_irq_enable();
}

nohz idle에 진입함을 설정한다.

  • 코드 라인 25에서 스케줄 도메인에 현재 cpu가 idle 상태로 진입했음을 설정한다.
  • 코드 라인 29~30에서 tick_sched에 idle로 진입했음을 설정한다.
  • 코드 라인 31에서 nohz idle 상태로 진입한다.

 

set_cpu_sd_state_idle()

kernel/sched/fair.c

void set_cpu_sd_state_idle(void)
{
        struct sched_domain *sd;
        int cpu = smp_processor_id();

        rcu_read_lock();
        sd = rcu_dereference(per_cpu(sd_busy, cpu));
        
        if (!sd || sd->nohz_idle)
                goto unlock;
        sd->nohz_idle = 1;

        atomic_dec(&sd->groups->sgc->nr_busy_cpus);
unlock:
        rcu_read_unlock();
}

busy 스케줄 도메인에 nohz idle 상태를 설정한다.

  • 코드 라인 9~10에서 현재 cpu에 대한 busy 스케줄링 도메인이 없거나 nohz_idle 설정이 없으면 함수를 빠져나온다.
  • 코드 라인 11에서 현재 cpu에 대한 busy 스케줄링 도메인의 nohz_idle에 1을 설정하여 현재 cpu가 nohz idle 상태에 진입했음을 알린다.
  • 코드 라인 13에서 역시 스케줄링 도메인의 그룹 캐파에 busy cpu 수를 감소시킨다.

 

__tick_nohz_idle_enter()

kernel/time/tick-sched.c

static void __tick_nohz_idle_enter(struct tick_sched *ts)
{
        ktime_t now, expires;
        int cpu = smp_processor_id();

        now = tick_nohz_start_idle(ts);

        if (can_stop_idle_tick(cpu, ts)) {
                int was_stopped = ts->tick_stopped;

                ts->idle_calls++;

                expires = tick_nohz_stop_sched_tick(ts, now, cpu);
                if (expires.tv64 > 0LL) {
                        ts->idle_sleeps++;
                        ts->idle_expires = expires;
                }
                        
                if (!was_stopped && ts->tick_stopped)
                        ts->idle_jiffies = ts->last_jiffies;
        }       
}

nohz idle 상태로 진입한다.

  • 코드 라인 6에서 현재 시각으로 idle 시작 시간을 설정하고 알아온다.
  • 코드 라인 8에서 정상적으로 idle 처리를 위해 tick을 정지시켜야하는 경우
  • 코드 라인 9~11에서 기존 틱 정지 상태를 알아와서 was_stopped에 담고 idle 처리 카운터를 1 증가시킨다.
  • 코드 라인 13에서 정규 스케줄 틱을 정지 시키고 nohz 모드로 진입한다.  여러 이유로 타이머 프로그램을 한 경우 만료 시각을 반환한다.
  • 코드 라인 14~17에서 만료 시각이 설정된 경우 idle 슬립 카운터를 1 증가킨다.
  • 코드 라인 19~20에서 틱 운용 중에 정지 요청이 발생한 경우 idle_jiffies에 last_jiffies를 대입한다.

 

can_stop_idle_tick()

kernel/time/tick-sched.c

static bool can_stop_idle_tick(int cpu, struct tick_sched *ts)
{
        /*
         * If this cpu is offline and it is the one which updates
         * jiffies, then give up the assignment and let it be taken by
         * the cpu which runs the tick timer next. If we don't drop
         * this here the jiffies might be stale and do_timer() never
         * invoked.
         */
        if (unlikely(!cpu_online(cpu))) {
                if (cpu == tick_do_timer_cpu)
                        tick_do_timer_cpu = TICK_DO_TIMER_NONE;
                return false;
        }

        if (unlikely(ts->nohz_mode == NOHZ_MODE_INACTIVE)) {
                ts->sleep_length = (ktime_t) { .tv64 = NSEC_PER_SEC/HZ };
                return false;
        }

        if (need_resched())
                return false;

        if (unlikely(local_softirq_pending() && cpu_online(cpu))) {
                static int ratelimit;

                if (ratelimit < 10 &&
                    (local_softirq_pending() & SOFTIRQ_STOP_IDLE_MASK)) {
                        pr_warn("NOHZ: local_softirq_pending %02x\n",
                                (unsigned int) local_softirq_pending());
                        ratelimit++;
                }
                return false;
        }

        if (tick_nohz_full_enabled()) {
                /*
                 * Keep the tick alive to guarantee timekeeping progression
                 * if there are full dynticks CPUs around
                 */
                if (tick_do_timer_cpu == cpu)
                        return false;
                /*
                 * Boot safety: make sure the timekeeping duty has been
                 * assigned before entering dyntick-idle mode,
                 */
                if (tick_do_timer_cpu == TICK_DO_TIMER_NONE)
                        return false;
        }

        return true;
}

idle 처리를 위해 tick을 정지시켜야하는지 여부를 반환한다.

  • 코드 라인 10~14에서 낮은 확률로 요청한 cpu가 online이 아니면 false를 반환한다. 요청하는 cpu가 틱을 발생하는 cpu인 경우 tick_do_timer_cpu에 TICK_DO_TIMER_NONE(-1)을 대입하여 틱을 발생하는 cpu가 없다는 것을 설정하게 한다.
  • 코드 라인 16~19에서 앚은 확률로 nohz 모드가 설정되지 않은 상태인 경우 sleep_length에 1초에 해당하는 jiffies 값을 대입하고 false를 반환한다.
  • 코드 라인 21~22에서 리스케줄 요청이 있는 경우 false를 반환한다.
  • 코드 라인 24~34에서 낮은 확률로 현재 cpu에서 softirq 펜딩중인 경우 경고 메시지를 출력하고 false를 반환한다. 경고 메시지는 10회 미만일때 까지만 출력한다.
  • 코드 라인 36~49에서 nohz full이 enable된 경우이고 틱 타이머가 현재 cpu이거나 설정되지 않은 경우 false를 반환한다.
  • 코드 라인 51에서 그외의 경우 true를 반환한다.

 

tick_nohz_start_idle()

kernel/time/tick-sched.c

static ktime_t tick_nohz_start_idle(struct tick_sched *ts)
{
        ktime_t now = ktime_get();

        ts->idle_entrytime = now;
        ts->idle_active = 1;
        sched_clock_idle_sleep_event();
        return now;
}

틱 스케줄에게 nohz idle 진입함을 설정하고 현재 시각을 반환한다.

 

tick_nohz_stop_sched_tick()

정규 스케줄 틱을 정지 시키고 nohz 모드로 진입한다. lazy된 rcu 처리나 다른 타이머 요청이 있는 경우의 타이머에 대해서는 동작시킨다.(nohz full인 경우 1초 단위) 타이머를 동작시키는 경우 만료 시각을 반환한다.

kernel/time/tick-sched.c

static ktime_t tick_nohz_stop_sched_tick(struct tick_sched *ts,
                                         ktime_t now, int cpu)
{
        unsigned long seq, last_jiffies, next_jiffies, delta_jiffies;
        ktime_t last_update, expires, ret = { .tv64 = 0 };
        unsigned long rcu_delta_jiffies;
        struct clock_event_device *dev = __this_cpu_read(tick_cpu_device.evtdev);
        u64 time_delta;

        time_delta = timekeeping_max_deferment();

        /* Read jiffies and the time when jiffies were updated last */
        do {
                seq = read_seqbegin(&jiffies_lock);
                last_update = last_jiffies_update;
                last_jiffies = jiffies;
        } while (read_seqretry(&jiffies_lock, seq));

        if (rcu_needs_cpu(&rcu_delta_jiffies) ||
            arch_needs_cpu() || irq_work_needs_cpu()) {
                next_jiffies = last_jiffies + 1;
                delta_jiffies = 1;
        } else {
                /* Get the next timer wheel timer */
                next_jiffies = get_next_timer_interrupt(last_jiffies);
                delta_jiffies = next_jiffies - last_jiffies;
                if (rcu_delta_jiffies < delta_jiffies) {
                        next_jiffies = last_jiffies + rcu_delta_jiffies;
                        delta_jiffies = rcu_delta_jiffies;
                }
        }
  • 코드 라인 10에서 타임 키핑 클럭에서 유예된 최대 시간을 알아온다.
  • 코드 라인 13~17에서 jiffies를 읽어와서 last_jiffies에 대입한다.
  • 코드 라인 19~22에서 idle 모드에 진입하기 전에 현재 cpu에서 처리할 rcu 콜백들을 처리하거나 지연처리하고 0을 반환하거나 lazy rcu 용 지연된 jiffies 틱을 알아왔거나 irq work가 아직 처리되지 않은 것이 남아있는 경우 다음 jiffies에 last_jiffies+1을 한다.
    • arch_needs_cpu() 함수는 대부분의 아키텍처에서는 0을 반환하고 s390 아키텍처만 CIF_NOHZ_DELAY 플래그가 설정된 경우만 1이된다.
  • 코드 라인 23~31에서 지연되어 처리할 시간이 필요 없는 경우 다음 lowres 및 hrtimer에서 jiffies 단위의 가장 빠른 만료시각을 알아와서 next_jiffies에 대입한다. delta_jiffies에는 마지막 last_jiffies와의 차이 시간을 구한다. 만일 delta 기간이 rcu 처리에 필요한 틱보다 더 큰 경우 rcu에 필요한 기간도 더한다.

 

        /*
         * Do not stop the tick, if we are only one off (or less)
         * or if the cpu is required for RCU:
         */
        if (!ts->tick_stopped && delta_jiffies <= 1)
                goto out;

        /* Schedule the tick, if we are at least one jiffie off */
        if ((long)delta_jiffies >= 1) {

                /*
                 * If this cpu is the one which updates jiffies, then
                 * give up the assignment and let it be taken by the
                 * cpu which runs the tick timer next, which might be
                 * this cpu as well. If we don't drop this here the
                 * jiffies might be stale and do_timer() never
                 * invoked. Keep track of the fact that it was the one
                 * which had the do_timer() duty last. If this cpu is
                 * the one which had the do_timer() duty last, we
                 * limit the sleep time to the timekeeping
                 * max_deferement value which we retrieved
                 * above. Otherwise we can sleep as long as we want.
                 */
                if (cpu == tick_do_timer_cpu) {
                        tick_do_timer_cpu = TICK_DO_TIMER_NONE;
                        ts->do_timer_last = 1;
                } else if (tick_do_timer_cpu != TICK_DO_TIMER_NONE) {
                        time_delta = KTIME_MAX;
                        ts->do_timer_last = 0;
                } else if (!ts->do_timer_last) {
                        time_delta = KTIME_MAX;
                }

#ifdef CONFIG_NO_HZ_FULL
                if (!ts->inidle) {
                        time_delta = min(time_delta,
                                         scheduler_tick_max_deferment());
                }
#endif
  • 코드 라인 5~6에서 이미 틱이 스탑되었었던 경우이면서 delta 틱이 1이하인 경우 처리를 중단하고 함수를 빠져나간다.
  • 코드 라인 9에서 delta 틱이 1 이상인 경우
  • 코드 라인 24~26에서 요청한 cpu가 현재 틱을 생성하는 cpu인 경우 tick_do_timer_cpu에 none을 대입하고 do_timer_last에 1을 대입한다.
  • 코드 라인 27~29에서 그렇지 않고 현재 틱을 생성하는 cpu가 지정된 경우 time_delta에 KTIME_MAX를 대입하고 do_timer_last에 0을 대입한다.
  • 코드 라인 30~32에서 그렇지 않고 do_timer_last가 0인 경우 time_delta에 KTIME_MAX를 대입한다.
  • 코드 라인 35~38에서 nohz full을 지원하는 시스템에서 inidle이 0인 경우 time_delta와 nohz full이 필요한 시간에서 작은 시간으로 time_delta를 구한다.
    • nohz full은 완벽하게 nohz로 아직 동작하지 않는다. uptime, CFS vruntime, 로드 밸런싱, … 등 아직 이전이 완료되지 않아서 다음 스케줄링에 필요한 틱에 1초를 더해 나노단위로 반환한다.

 

                /*
                 * calculate the expiry time for the next timer wheel
                 * timer. delta_jiffies >= NEXT_TIMER_MAX_DELTA signals
                 * that there is no timer pending or at least extremely
                 * far into the future (12 days for HZ=1000). In this
                 * case we set the expiry to the end of time.
                 */
                if (likely(delta_jiffies < NEXT_TIMER_MAX_DELTA)) {
                        /*
                         * Calculate the time delta for the next timer event.
                         * If the time delta exceeds the maximum time delta
                         * permitted by the current clocksource then adjust
                         * the time delta accordingly to ensure the
                         * clocksource does not wrap.
                         */
                        time_delta = min_t(u64, time_delta,
                                           tick_period.tv64 * delta_jiffies);
                }

                if (time_delta < KTIME_MAX)
                        expires = ktime_add_ns(last_update, time_delta);
                else
                        expires.tv64 = KTIME_MAX;

                /* Skip reprogram of event if its not changed */
                if (ts->tick_stopped && ktime_equal(expires, dev->next_event))
                        goto out;

                ret = expires;

                /*
                 * nohz_stop_sched_tick can be called several times before
                 * the nohz_restart_sched_tick is called. This happens when
                 * interrupts arrive which do not cause a reschedule. In the
                 * first call we save the current tick time, so we can restart
                 * the scheduler tick in nohz_restart_sched_tick.
                 */
                if (!ts->tick_stopped) {
                        nohz_balance_enter_idle(cpu);
                        calc_load_enter_idle();

                        ts->last_tick = hrtimer_get_expires(&ts->sched_timer);
                        ts->tick_stopped = 1;
                        trace_tick_stop(1, " ");
                }
  • 코드 라인 8~18에서 delta_jiffies가 1G-1 틱 미만인 경우 time_delta와 delta_jiffies를 나노초로 변환한 시간 중 작은 기간을 time_delta로 한다.
  • 코드 라인 20~23에서 time_delta가 KTIME_MAX보다 작은 경우 만료시각으로 last_update에 time_delta를 추가한다. 그렇지 않은 경우 최대 만료 시각을 담는다.
  • 코드 라인 26~27에서 틱이 이미 정지된 상태이고 이미 동일한 만료 시각으로 설정된 경우는 변경 사항이 없는 경우이므로 함수를 빠져나간다.
  • 코드 라인 29에서 반환 값으로 만료 시각을 대입한다.
  • 코드 라인 38~45에서 요청 cpu의 틱이 동작 중인 경우 로드밸런스를 위해 틱이 정지되어 idle 상태로 진입할 것이라 설정하고,  nohz 모드 진입으로 calc_load_idle도 갱신한다. 또한 last_tick을 스케줄 타이머의 만료시각으로 갱신하고 tick_stopped에 1을 대입하여 틱이 정지된 상태임을 알린다.

 

                /*
                 * If the expiration time == KTIME_MAX, then
                 * in this case we simply stop the tick timer.
                 */
                 if (unlikely(expires.tv64 == KTIME_MAX)) {
                        if (ts->nohz_mode == NOHZ_MODE_HIGHRES)
                                hrtimer_cancel(&ts->sched_timer);
                        goto out;
                }

                if (ts->nohz_mode == NOHZ_MODE_HIGHRES) {
                        hrtimer_start(&ts->sched_timer, expires,
                                      HRTIMER_MODE_ABS_PINNED);
                        /* Check, if the timer was already in the past */
                        if (hrtimer_active(&ts->sched_timer))
                                goto out;
                } else if (!tick_program_event(expires, 0))
                                goto out;
                /*
                 * We are past the event already. So we crossed a
                 * jiffie boundary. Update jiffies and raise the
                 * softirq.
                 */
                tick_do_update_jiffies64(ktime_get());
        }
        raise_softirq_irqoff(TIMER_SOFTIRQ);
out:
        ts->next_jiffies = next_jiffies;
        ts->last_jiffies = last_jiffies;
        ts->sleep_length = ktime_sub(dev->next_event, now);

        return ret;
}
  • 코드 라인 5~9에서 만료 시각이 최대치로 설정된 경우 nohz 모드가 high-resolution을 사용하는 타이머인 경우 스케줄 타이머를 정지하고 함수를 빠져나간다.
  • 코드 라인 11~16에서 nohz 모드가 high-resolution을 사용하는 타이머인 경우 만료 시각으로 틱 스케줄 타이머를 가동시킨다. 만일 잘 가동된 경우 함수를 빠져나간다.
  • 코드 라인 17~18에서 만료 시각으로 틱 프로그램을 하고 성공(0)한 경우 함수를 빠져나간다.
  • 코드 라인 24에서 jiffies 값을 갱신한다.
  • 코드 라인 26에서 lowres 타이머용 softirq를 트리거한다.
  • 코드 라인 28~32에서 틱의 next_jiffies, last_jiffies, sleep_length를 갱신하고 ret를 반환한다.

 

nohz_balance_enter_idle()

kernel/sched/fair.c

/*
 * This routine will record that the cpu is going idle with tick stopped.
 * This info will be used in performing idle load balancing in the future.
 */
void nohz_balance_enter_idle(int cpu)
{
        /* 
         * If this cpu is going down, then nothing needs to be done.
         */
        if (!cpu_active(cpu))
                return;

        if (test_bit(NOHZ_TICK_STOPPED, nohz_flags(cpu)))
                return;

        /*
         * If we're a completely isolated CPU, we don't play.
         */
        if (on_null_domain(cpu_rq(cpu)))
                return;

        cpumask_set_cpu(cpu, nohz.idle_cpus_mask);
        atomic_inc(&nohz.nr_cpus);
        set_bit(NOHZ_TICK_STOPPED, nohz_flags(cpu));
}

로드밸런스를 위해 틱이 정지되어 idle 상태로 진입할 것이라 설정한다.

 

  • 코드 라인 10~11에서 요청한 cpu가 active cpu가 아닌 경우 함수를 빠져나간다.
  • 코드 라인 13~14에서 런큐 nohz_flags에 이미 NOHZ_TICK_STOPPED 플래그가 설정된 경우 함수를 빠져나간다.
    • 이미 nohz idle 진입했음을 알아낸다.
  • 코드 라인 19~20에서 요청한 cpu의 런큐에 스케줄 도메인이 없는 경우 함수를 빠져나간다.
  • 코드 라인 22에서 nohz.idle_cpus_mask 비트맵에 요청한 cpu에 해당하는 비트를 설정하여 idle 상태임을 알린다.
  • 코드 라인 23에서 nohz.nr_cpus를 1 증가시켜 nohz idle 상태로 진입한 cpu의 수를 나타내게한다.
  • 코드 라인 24에서 런큐 nohz_flags에 NOHZ_TICK_STOPPED 플래그를 설정한다.

 

calc_load_enter_idle()

kernel/sched/proc.c

void calc_load_enter_idle(void)
{
        struct rq *this_rq = this_rq();
        long delta;     

        /*
         * We're going into NOHZ mode, if there's any pending delta, fold it
         * into the pending idle delta.
         */     
        delta = calc_load_fold_active(this_rq);
        if (delta) {
                int idx = calc_load_write_idx();
                atomic_long_add(delta, &calc_load_idle[idx]);
        }               
}

nohz idle 진입 시 로드값을 갱신한다.

  • 코드 라인 10에서 요청한 런큐의 기존 active 태스크 수와 현재 산출한 active 태스크 수의 차이를 산출한다.
  • 코드 라인 11~14에서 그 차이가 발생하는 경우 로드 계산에 사용할 인덱스를 알아와서 그 인덱스로 calc_load_idle[]에 delta를 추가한다.

 

nohz idle에서 벗어날 때

tick_nohz_idle_exit()

kernel/time/tick-sched.c

/**
 * tick_nohz_idle_exit - restart the idle tick from the idle task
 *
 * Restart the idle tick when the CPU is woken up from idle
 * This also exit the RCU extended quiescent state. The CPU
 * can use RCU again after this function is called.
 */
void tick_nohz_idle_exit(void)
{
        struct tick_sched *ts = this_cpu_ptr(&tick_cpu_sched);
        ktime_t now;

        local_irq_disable();

        WARN_ON_ONCE(!ts->inidle);

        ts->inidle = 0;

        if (ts->idle_active || ts->tick_stopped)
                now = ktime_get();

        if (ts->idle_active)
                tick_nohz_stop_idle(ts, now);

        if (ts->tick_stopped) { 
                tick_nohz_restart_sched_tick(ts, now);
                tick_nohz_account_idle_ticks(ts);
        }

        local_irq_enable();
}

nohz idle이 종료되었음을 설정하고 틱을 리스타트한다.

  • 코드 라인 17에서  tick_sched에 idle에서 벗어났음을 설정한다.
  • 코드 라인 19~20에서 idle_active 또는 tick_stopped가 설정된 경우 현재 시각을 알아온다.
  • 코드 라인 22~23에서 idle_active가 설정된 경우 현재 시각으로 nohz가 stop 되었음을 알린다.
  • 코드 라인 25에서 tick_stopped가 설정된 경우 현재 시각을 기준으로 nohz 틱을 리스타트하고 측정과 관련하여 갱신한다.

 

tick_nohz_stop_idle()

kernel/time/tick-sched.c

static void tick_nohz_stop_idle(struct tick_sched *ts, ktime_t now)
{
        update_ts_time_stats(smp_processor_id(), ts, now, NULL);
        ts->idle_active = 0;

        sched_clock_idle_wakeup_event(0);
}

 

요청한 틱 스케줄러의 idle_active에 0을 대입하여 nohz가 stop 되었음을 나타낸다.

  • sched_clock_idle_wakeup_event()는 unstable 클럭을 사용하는 경우에 수행되며 arm 아키텍처에서는 아무것도 수행하지 않는다.

 

tick_nohz_restart_sched_tick()

kernel/time/tick-sched.c

static void tick_nohz_restart_sched_tick(struct tick_sched *ts, ktime_t now)
{
        /* Update jiffies first */
        tick_do_update_jiffies64(now);
        update_cpu_load_nohz();

        calc_load_exit_idle();
        touch_softlockup_watchdog();
        /*
         * Cancel the scheduled timer and restore the tick
         */
        ts->tick_stopped  = 0;
        ts->idle_exittime = now;

        tick_nohz_restart(ts, now);
}

 

nohz 틱을 다시 리스타트 한다.

  • 코드 라인 4에서 jiffes를 갱신한다.
  • 코드 라인 5에서 cpu load를 갱신한다.
  • 코드 라인 7에서 idle을 빠져나가는 것에 대해 로드 밸런스 체크 주기를 갱신한다.
  • 코드 라인 12~15에서 tick_stopped에 0을 대입하고 idle 끝시각을 현재 시각으로 갱신하고 틱 nohz를 다시 시작한다.

 

tick_nohz_restart()

kernel/time/tick-sched.c

static void tick_nohz_restart(struct tick_sched *ts, ktime_t now)
{
        hrtimer_cancel(&ts->sched_timer);
        hrtimer_set_expires(&ts->sched_timer, ts->last_tick);

        while (1) {
                /* Forward the time to expire in the future */
                hrtimer_forward(&ts->sched_timer, now, tick_period);

                if (ts->nohz_mode == NOHZ_MODE_HIGHRES) {
                        hrtimer_start_expires(&ts->sched_timer,
                                              HRTIMER_MODE_ABS_PINNED);
                        /* Check, if the timer was already in the past */
                        if (hrtimer_active(&ts->sched_timer))
                                break;
                } else {
                        if (!tick_program_event(
                                hrtimer_get_expires(&ts->sched_timer), 0))
                                break;
                }
                /* Reread time and update jiffies */
                now = ktime_get();
                tick_do_update_jiffies64(now);
        }
}

nohz 틱을 다시 리프로그램한다.

  • 코드 라인 3에서 기존 스케줄 틱을 정지시킨다.
  • 코드 라인 4에서 last_tick으로 스케줄 타이머의 만료 시각을 다시 설정한다.
  • 코드 라인 6에서 루프를 돌며 tick_period 기간으로 스케줄 타이머를 동작 시킨다.
  • 코드 라인 8~13에서 nohz_mode가 NOHZ_MODE_HIGHRES로 설정된 경우 스케줄 타이머를 동작시킨다. 만일 정상적으로 가동된 경우 함수를 빠져나간다.
  • 코드 라인 14~18에서 다른 모드로 설정된 경우 틱 이벤트를 다시 프로그램하고 함수를 빠져나간다.
  • 코드 라인 20~21에서 리프로그램이 실패한 경우 jiffies를 갱신하고 루프를 다시 돈다.

 

 

nohz full 킥

kernel/time/tick-sched.c

static DEFINE_PER_CPU(struct irq_work, nohz_full_kick_work) = {
        .func = nohz_full_kick_work_func,
};

nohz_full_kick_work_func() 함수를 호출하는 per-cpu nohz_full_kick_work라는 이름의 irq 워크큐를 선언한다.

 

tick_nohz_full_kick()

kernel/time/tick-sched.c

/*
 * Kick this CPU if it's full dynticks in order to force it to
 * re-evaluate its dependency on the tick and restart it if necessary.
 * This kick, unlike tick_nohz_full_kick_cpu() and tick_nohz_full_kick_all(),
 * is NMI safe.
 */
void tick_nohz_full_kick(void)
{
        if (!tick_nohz_full_cpu(smp_processor_id()))
                return;

        irq_work_queue(this_cpu_ptr(&nohz_full_kick_work));
}

현재 cpu가 nohz full 모드로 동작하는 경우 nohz_full_kick_work를 irq 워크큐에 추가한 후 ipi call을 호출한다.

 

tick_nohz_full_kick_cpu()

kernel/time/tick-sched.c

/*
 * Kick the CPU if it's full dynticks in order to force it to
 * re-evaluate its dependency on the tick and restart it if necessary.
 */
void tick_nohz_full_kick_cpu(int cpu)
{
        if (!tick_nohz_full_cpu(cpu))
                return;

        irq_work_queue_on(&per_cpu(nohz_full_kick_work, cpu), cpu);
}

요청한 cpu가 nohz full 모드로 동작하는 경우 nohz_full_kick_work를 irq 워크큐에 추가한 후 ipi call을 호출한다.

 

nohz_full_kick_work_func()

kernel/time/tick-sched.c

static void nohz_full_kick_work_func(struct irq_work *work)
{
        __tick_nohz_full_check();
}

ipi 호출되었다. nohz full 상태를 체크하여 필요한 경우 틱을 재개시킨다.

 

__tick_nohz_full_check()

kernel/time/tick-sched.c

/*
 * Re-evaluate the need for the tick on the current CPU
 * and restart it if necessary.
 */
void __tick_nohz_full_check(void)
{
        struct tick_sched *ts = this_cpu_ptr(&tick_cpu_sched);

        if (tick_nohz_full_cpu(smp_processor_id())) {
                if (ts->tick_stopped && !is_idle_task(current)) {
                        if (!can_stop_full_tick())
                                tick_nohz_restart_sched_tick(ts, ktime_get());
                }
        }
}

nohz full 상태를 체크하여 필요한 경우 틱을 재개시킨다.

  • 코드 라인 9~에서 현재 cpu가 nohz full 상태, tick 정지, 현재 태스크가 idle 상태가 아니고 nohz full을 멈출 수 없는 경우 틱을 정지시킨 후 다시 재개시킨다.

 

참고

 

 

Scheduler -10- (Idle Scheduer)

 

Idle-task Scheduler

어떠한 스케줄러보다 우선 순위가 낮아 다른 스케줄러에서 태스크가 동작해야 하는 상황이되면 언제나 preemption 된다. idle 태스크는 core 마다 1개씩 지정되어 있어 다른 cpu로의 migration이 필요 없다.

  • 각 core의 부트업 처리가 완료되면 cpu_startup_entry() 함수를 호출하여 idle 태스크로의 역할을 수행한다.
    • 부트업 프로세스가 사용하는 메모리 환경을 그대로 사용한다. (init_mm 공유)

 

select_task_rq_idle()

kernel/sched/idle_task.c

select_task_rq_idle(struct task_struct *p, int cpu, int sd_flag, int flags)
{
        return task_cpu(p); /* IDLE tasks as never migrated */
}

migration을 허용하지 않는다.

 

check_preempt_curr_idle()

kernel/sched/idle_task.c

/*
 * Idle tasks are unconditionally rescheduled:
 */
static void check_preempt_curr_idle(struct rq *rq, struct task_struct *p, int flags)
{
        resched_curr(rq);
}

항상 preemption 한다.

 

dequeue_task_idle()

kernel/sched/idle_task.c

/*
 * It is not legal to sleep in the idle task - print a warning
 * message if some code attempts to do it:
 */
static void
dequeue_task_idle(struct rq *rq, struct task_struct *p, int flags)
{
        raw_spin_unlock_irq(&rq->lock);
        printk(KERN_ERR "bad: scheduling from the idle thread!\n");
        dump_stack();
        raw_spin_lock_irq(&rq->lock);
}

idle 태스크는 엔큐 및 디큐 개념이 없다. 따라서 호출되는 경우 경고 메시지를 출력한다.

 

task_tick_idle()

kernel/sched/idle_task.c

static void task_tick_idle(struct rq *rq, struct task_struct *curr, int queued)
{
}

task 틱마다 아무런 동작도 하지 않는다.

 

update_curr_idle()

kernel/sched/idle_task.c

static void update_curr_idle(struct rq *rq)
{
}

아무런 동작도 하지 않는다.

 

switched_to_idle()

kernel/sched/idle_task.c

static void switched_to_idle(struct rq *rq, struct task_struct *p)
{
        BUG(); 
}

설계상 이 idle 태스크로의 스위칭은 불가능하다.

 

prio_changed_idle()

kernel/sched/idle_task.c

static void
prio_changed_idle(struct rq *rq, struct task_struct *p, int oldprio)
{
        BUG(); 
}

idle 태스크 자체가 가장 낮은 순위의 태스크로 별도로 우선 순위를 변경할 수 없다.

 

get_rr_interval_idle()

kernel/sched/idle_task.c

static unsigned int
get_rr_interval_idle(struct rq *rq, struct task_struct *task)
{
        return 0;
}

idle 태스크는 preemption되지 않는 경우 계속 수행되어야 하므로 idle 스케줄러는 인터벌을 사용하지 않는다.

 

 

다음 idle 태스크 선택

pick_next_task_idle()

kernel/sched/idle_task.c

static struct task_struct *
pick_next_task_idle(struct rq *rq, struct task_struct *prev)
{
        put_prev_task(rq, prev);

        schedstat_inc(rq, sched_goidle);
        return rq->idle;
}

기존 태스크에 대한 런타임 계산 등을 끝내고 idle 태스크를 반환한다.

 

put_prev_task_idle()

kernel/sched/idle_task.c

static void put_prev_task_idle(struct rq *rq, struct task_struct *prev)
{
        idle_exit_fair(rq);
        rq_last_tick_reset(rq);
}

idle 태스크의 실행 시각이 끝난 경우 이에 대한 처리를 수행한다.

  • 코드 라인 3에서 런큐의 러너블 로드 평균을 갱신한다.
  • 코드 라인 4에서 nohz full 을 허용한 시스템인 경우 last_sched_tick에 현재 시각(jiffies)를 기록한다.

 

Idle-task 스케줄러 OPS

kernel/sched/idle_task.c

/*
 * Simple, special scheduling class for the per-CPU idle tasks:
 */
const struct sched_class idle_sched_class = {
        /* .next is NULL */
        /* no enqueue/yield_task for idle tasks */

        /* dequeue is not valid, we print a debug message there: */
        .dequeue_task           = dequeue_task_idle,

        .check_preempt_curr     = check_preempt_curr_idle,

        .pick_next_task         = pick_next_task_idle,
        .put_prev_task          = put_prev_task_idle,

#ifdef CONFIG_SMP
        .select_task_rq         = select_task_rq_idle,
#endif

        .set_curr_task          = set_curr_task_idle,
        .task_tick              = task_tick_idle,

        .get_rr_interval        = get_rr_interval_idle,

        .prio_changed           = prio_changed_idle,
        .switched_to            = switched_to_idle,
        .update_curr            = update_curr_idle,
};

 

참고