上一篇文章对工作队列原理以及核心数据结构做了简单介绍，本文重点介绍下workqueue的创建以及worker的管理。
一、工作队列的创建（__alloc_workqueue_key）
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 = ;
va_list args;
struct workqueue_struct *wq;
struct pool_workqueue *pwq; /* allocate wq and format name */
if (flags & WQ_UNBOUND)
tbl_size = wq_numa_tbl_len * sizeof(wq->numa_pwq_tbl[]);
/*分配workqueue_struct结构*/
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, );
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, );
INIT_LIST_HEAD(&wq->list);
if (alloc_and_link_pwqs(wq) < )
goto err_free_wq;
/*
* 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();
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;
}
该函数主要任务就是通过kzalloc分配一个workqueue_struct结构，然后格式化一个名称，对workqueue进行简单初始化；’接着就调用 和pwd建立关系。我们暂且不考虑WQ_MEM_RECLAIM的情况，那么该函数主要就完成这两个功能。所有的workqueue会链接成一个链表，链表头是 一个全局静态变量
static LIST_HEAD(workqueues); /* PL: list of all workqueues */
本函数比较重要的就是和pwq建立关系了
static int alloc_and_link_pwqs(struct workqueue_struct *wq)
{
bool highpri = wq->flags & WQ_HIGHPRI;
int cpu;
/*如果是普通的work_queue*/
if (!(wq->flags & WQ_UNBOUND)) {
/*为每个CPU 分配pool_workqueue--pwq*/
wq->cpu_pwqs = alloc_percpu(struct pool_workqueue);
if (!wq->cpu_pwqs)
return -ENOMEM;
/*把pwd和wq链接*/
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 ;
} else {
return apply_workqueue_attrs(wq, unbound_std_wq_attrs[highpri]);
}
}
这里先知考虑普通的workqueue，不考虑WQ_UNBOUND。函数通过alloc_percpu为workqueue分配了pool_workqueue变量，然后通过for_each_possible_cpu，对每个CPU进行处理，实际上就是把对应的pool_workqueue和worker_pool通过init_pwq关联起来。如上一篇文章所描述的，worker_pool分为两种：普通的和高优先级的。普通的为第0项，而高优先级的为第一项。建立关联后在通过link_pwq把pwq接入wq的链表中。
二、worker的创建
在创建好workqueue和对应的pwq以及worker_pool后，需要显示的为worker_pool创建worker。核心函数为create_and_start_worker
static int create_and_start_worker(struct worker_pool *pool)
{
struct worker *worker; mutex_lock(&pool->manager_mutex);
/*创建一个属于pool的worker，实际上是创建一个线程*/
worker = create_worker(pool);
if (worker) {
spin_lock_irq(&pool->lock);
/*启动worker，即唤醒线程*/
start_worker(worker);
spin_unlock_irq(&pool->lock);
} mutex_unlock(&pool->manager_mutex);
return worker ? : -ENOMEM;
}
注意这里是针对worker_pool创建worker，所以worker_pool作为参数传递进来，而具体执行创建任务的是create_worker函数，且由于有专门的worker manager，故这里给worker_pool增加worker需要加锁。
create_worker函数其实也不复杂，核心任务主要包含以下几个步骤：
- 通过alloc_worker分配一个worker结构，并执行简单的初始化
- 在worker和worker_pool之间建立联系
- 通过kthread_create_on_node创建工作线程，处理函数为worker_thread
- 设置线程优先级
初始状态下是为每个worker_pool创建一个worker。创建好之后通过start_worker启动worker
static void start_worker(struct worker *worker)
{
worker->flags |= WORKER_STARTED;
worker->pool->nr_workers++;
worker_enter_idle(worker);
wake_up_process(worker->task);
}
该函数较简单，首先就更新worker状态为WORKER_STARTED，增加pool中worker统计量；然后通过worker_enter_idle标记worker目前处于idle状态；最后通过wake_up_process唤醒worker。我们看下中间设置idle状态的过程
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 start_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));
}
该函数会设置WORKER_IDLE，递增pool的nr_idle计数，然后更新last_active为当前jiffies。接着把worker挂入pool的idle_list的链表头.默认状态下，一个worker在idle状态停留的最长时IDLE_WORKER_TIMEOUT，超过这个时间就要启用管理工作。这里的last_active便是纪录进入idle状态的时间，
三、worker的管理
系统中会根据实际对worker的需要，动态的增删worker。针对idle worker，worker_pool中有个定时器idle_timer，其处理函数为idle_worker_timeout，看下该处理函数
static void idle_worker_timeout(unsigned long __pool)
{
struct worker_pool *pool = (void *)__pool; spin_lock_irq(&pool->lock); if (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;
/*idleworker每次最多保持idle状态IDLE_WORKER_TIMEOU，当定时器到期时进行检查，如果还未到最长时间，则延迟定时器，否则
*对pool设置管理标志，唤醒线程进行管理
*/
if (time_before(jiffies, expires))
mod_timer(&pool->idle_timer, expires);//重置到期时间
else {
/* it's been idle for too long, wake up manager */
pool->flags |= POOL_MANAGE_WORKERS;
wake_up_worker(pool);
}
} spin_unlock_irq(&pool->lock);
}
该函数主要是针对系统中出现太多worker的情况进行处理，如何判定worker太多呢？too_many_workers去完成，具体就是 nr_idle > 2 && (nr_idle - 2) * MAX_IDLE_WORKERS_RATIO >= nr_busy决定，其中MAX_IDLE_WORKERS_RATIO为4。当的确idle worker太多了的时候，取最先挂入idle链表中的worker，判定其处于idle状态的时间是否超时，即超过IDLE_WORKER_TIMEOUT，如果没有超时，则通过mod_timer修改定时器到期时间为该定时器对应的最长idle时间，否则设置pool的POOL_MANAGE_WORKERS状态，唤醒pool中的first worker执行管理工作。在worker的处理函数worker_thread中，通过need_more_worker判断当前是否需要更多的worker，如果不需要，则直接goto到sleep
sleep:
if (unlikely(need_to_manage_workers(pool)) && manage_workers(worker))
goto recheck;
need_to_manage_workers就是判断POOL_MANAGE_WORKERS，如果设置了该标志则返回真。 管理worker的核心在manage_workers，其中只有两个关键函数
ret |= maybe_destroy_workers(pool);
ret |= maybe_create_worker(pool);
我们只看maybe_destroy_workers
static bool maybe_destroy_workers(struct worker_pool *pool)
{
bool ret = false; while (too_many_workers(pool)) {
struct worker *worker;
unsigned long expires; 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;
}
/*删除最先挂入链表的worker*/
destroy_worker(worker);
ret = true;
}
return ret;
}
该函数中会在此通过too_many_workers判断是否有太多worker，如果是，则再次取最后一个worker，检查idle时间，如果没有超时，则修改定时器到期时间，否则通过destroy_worker销毁worker。为什么要这样判断呢？通过对代码的分析，我感觉manage_work不仅负责删除，还负责增加worker，定时器主要是针对idle worker即目的是销毁多余的worker，但是执行管理任务的工作集成到了worker_thread中，因此就worker_thread而言，有可能需要增加、有可能需要删除、还有可能不需要管理。因此这里需要再次判断。
四、work的添加
static inline bool schedule_work(struct work_struct *work)
{
return queue_work(system_wq, work);
} static inline bool queue_work(struct workqueue_struct *wq,
struct work_struct *work)
{
return queue_work_on(WORK_CPU_UNBOUND, wq, work);
}
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;
}
因此主体就是__queue_work，其中一个核心工作就是调用了insert_work
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();
/*如果需要更多，则唤醒,主要是判断当前是否有正在运行的worker*/
if (__need_more_worker(pool))
wake_up_worker(pool);
}
函数首先调用set_work_pwq把pwd写入到work的data字段，然后把work加入到worker_pool维护的work链表中，在最后判断现在是否需要更多worker，如果需要，则执行唤醒操作。当然是针对当前worker_pool，且唤醒的是worker_pool的第一个worker。其实在queue_work中，为避免work重入，在选定worker_pool的时候会判断该work是否仍在其他worker_pool上运行，如果是，就把该work挂入对应worker_pool的work_list上，
以马内利
参考资料：
LInux 3.10.1源码