Read the fucking source code!
--By 鲁迅A picture is worth a thousand words.
--By 高尔基说明:html
Workqueue
工做队列是利用内核线程来异步执行工做任务的通用机制;Workqueue
工做队列能够用做中断处理的Bottom-half
机制,利用进程上下文来执行中断处理中耗时的任务,所以它容许睡眠,而Softirq
和Tasklet
在处理任务时不能睡眠;来一张概述图:node
workqueue
的调度或入队接口后,经过创建好的连接关系图逐级找到合适的worker
,最终完成工做任务的执行;此处应有图:linux
work_struct
:工做队列调度的最小单位,work item
;workqueue_struct
:工做队列,work item
都挂入到工做队列中;worker
:work item
的处理者,每一个worker
对应一个内核线程;worker_pool
:worker
池(内核线程池),是一个共享资源池,提供不一样的worker
来对work item
进行处理;pool_workqueue
:充当桥梁纽带的做用,用于链接workqueue
和worker_pool
,创建连接关系;下边看看细节吧:api
struct work_struct
用来描述work
,初始化一个work
并添加到工做队列后,将会将其传递到合适的内核线程来进行处理,它是用于调度的最小单位。缓存
关键字段描述以下:数据结构
struct work_struct { atomic_long_t data; //低比特存放状态位,高比特存放worker_pool的ID或者pool_workqueue的指针 struct list_head entry; //用于添加到其余队列上 work_func_t func; //工做任务的处理函数,在内核线程中回调 #ifdef CONFIG_LOCKDEP struct lockdep_map lockdep_map; #endif };
图片说明下data
字段:并发
内核中工做队列分为两种:异步
worker
建立的内核线程绑定到特定的CPU上运行;WQ_UNBOUND
标志,内核线程能够在处理器间迁移;内核默认建立了一些工做队列(用户也能够建立):函数
system_mq
:若是work item
执行时间较短,使用本队列,调用schedule[_delayed]_work[_on]()
接口就是添加到本队列中;system_highpri_mq
:高优先级工做队列,以nice值-20来运行;system_long_wq
:若是work item
执行时间较长,使用本队列;system_unbound_wq
:该工做队列的内核线程不绑定到特定的处理器上;system_freezable_wq
:该工做队列用于在Suspend时可冻结的work item
;system_power_efficient_wq
:该工做队列用于节能目的而选择牺牲性能的work item
;system_freezable_power_efficient_wq
:该工做队列用于节能或Suspend时可冻结目的的work item
;struct workqueue_struct
关键字段介绍以下:工具
struct workqueue_struct { struct list_head pwqs; /* WR: all pwqs of this wq */ //全部的pool_workqueue都添加到本链表中 struct list_head list; /* PR: list of all workqueues */ //用于将工做队列添加到全局链表workqueues中 struct list_head maydays; /* MD: pwqs requesting rescue */ //rescue状态下的pool_workqueue添加到本链表中 struct worker *rescuer; /* I: rescue worker */ //rescuer内核线程,用于处理内存紧张时建立工做线程失败的状况 struct pool_workqueue *dfl_pwq; /* PW: only for unbound wqs */ 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 */ //Per-CPU都建立pool_workqueue struct pool_workqueue __rcu *numa_pwq_tbl[]; /* PWR: unbound pwqs indexed by node */ //Per-Node建立pool_workqueue ... };
worker
对应一个内核线程,用于对work item
的处理;worker
根据工做状态,能够添加到worker_pool
的空闲链表或忙碌列表中;worker
处于空闲状态时并接收到工做处理请求,将唤醒内核线程来处理;worker_pool
中由一个初始的空闲工做线程建立的,并根据须要动态建立和销毁;关键字段描述以下:
struct worker { /* on idle list while idle, on busy hash table while busy */ union { struct list_head entry; /* L: while idle */ //用于添加到worker_pool的空闲链表中 struct hlist_node hentry; /* L: while busy */ //用于添加到worker_pool的忙碌列表中 }; struct work_struct *current_work; /* L: work being processed */ //当前正在处理的work work_func_t current_func; /* L: current_work's fn */ //当前正在执行的work回调函数 struct pool_workqueue *current_pwq; /* L: current_work's pwq */ //指向当前work所属的pool_workqueue struct list_head scheduled; /* L: scheduled works */ //全部被调度执行的work都将添加到该链表中 /* 64 bytes boundary on 64bit, 32 on 32bit */ struct task_struct *task; /* I: worker task */ //指向内核线程 struct worker_pool *pool; /* I: the associated pool */ //该worker所属的worker_pool /* L: for rescuers */ struct list_head node; /* A: anchored at pool->workers */ //添加到worker_pool->workers链表中 /* A: runs through worker->node */ ... };
worker_pool
是一个资源池,管理多个worker
,也就是管理多个内核线程;worker_pool
是Per-CPU建立,每一个CPU都有两个worker_pool
,对应不一样的优先级,nice值分别为0和-20;worker_pool
建立后会添加到unbound_pool_hash
哈希表中;worker_pool
管理一个空闲链表和一个忙碌列表,其中忙碌列表由哈希管理;关键字段描述以下:
struct worker_pool { spinlock_t lock; /* the pool lock */ int cpu; /* I: the associated cpu */ //绑定到CPU的workqueue,表明CPU ID int node; /* I: the associated node ID */ //非绑定类型的workqueue,表明内存Node ID int id; /* I: pool ID */ unsigned int flags; /* X: flags */ unsigned long watchdog_ts; /* L: watchdog timestamp */ struct list_head worklist; /* L: list of pending works */ //pending状态的work添加到本链表 int nr_workers; /* L: total number of workers */ //worker的数量 /* 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 */ //处于IDLE状态的worker添加到本链表 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); //工做状态的worker添加到本哈希表中 /* L: hash of busy workers */ /* see manage_workers() for details on the two manager mutexes */ struct worker *manager; /* L: purely informational */ struct mutex attach_mutex; /* attach/detach exclusion */ struct list_head workers; /* A: attached workers */ //worker_pool管理的worker添加到本链表中 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 */ //用于添加到unbound_pool_hash中 ... } ____cacheline_aligned_in_smp;
pool_workqueue
充当纽带的做用,用于将workqueue
和worker_pool
关联起来;关键字段描述以下:
struct pool_workqueue { struct worker_pool *pool; /* I: the associated pool */ //指向worker_pool struct workqueue_struct *wq; /* I: the owning workqueue */ //指向所属的workqueue int nr_active; /* L: nr of active works */ //活跃的work数量 int max_active; /* L: max active works */ //活跃的最大work数量 struct list_head delayed_works; /* L: delayed works */ //延迟执行的work挂入本链表 struct list_head pwqs_node; /* WR: node on wq->pwqs */ //用于添加到workqueue链表中 struct list_head mayday_node; /* MD: node on wq->maydays */ //用于添加到workqueue链表中 ... } __aligned(1 << WORK_STRUCT_FLAG_BITS);
再来张图,首尾呼应一下:
workqueue
子系统的初始化分红两步来完成的:workqueue_init_early
和workqueue_init
。workqueue
子系统早期初始化函数完成的主要工做包括:
pool_workqueue
的SLAB缓存,用于动态分配struct pool_workqueue
结构;worker_pool
,其中的nice值分别为0和HIGHPRI_NICE_LEVEL
,而且为每一个worker_pool
从worker_pool_idr
中分配一个ID号;struct workqueue_attrs
属性,主要描述内核线程的nice值,以及cpumask值,分别针对优先级以及容许在哪些CPU上执行;从图中能够看出建立工做队列的接口为:alloc_workqueue
,以下图:
alloc_workqueue
完成的主要工做包括:
struct workqueue_struct
的数据结构,而且对该结构中的字段进行初始化操做;workqueue
最终须要和worker_pool
关联起来,而这个纽带就是pool_workqueue
,alloc_and_link_pwqs
函数就是完成这个功能:1)若是工做队列是绑定到CPU上的,则为每一个CPU都分配pool_workqueue
而且初始化,经过link_pwq
将工做队列与pool_workqueue
创建链接;2)若是工做队列不绑定到CPU上,则按内存节点(NUMA,参考以前内存管理的文章)来分配pool_workqueue
,调用get_unbound_pool
来实现,它会根据wq属性先去查找,若是没有找到相同的就建立一个新的pool_workqueue
,而且添加到unbound_pool_hash
哈希表中,最后也会调用link_pwq
来创建链接;WQ_MEM_RECLAIM
标志,则会新建rescuer worker
,对应rescuer_thread
内核线程。当内存紧张时,新建立worker
可能会失败,这时候由rescuer
来处理这种状况;workqueues
中;workqueue
子系统第二阶段的初始化:
worker_pool
,添加一个初始的worker
;create_worker
函数中,建立的内核线程名字为kworker/XX:YY
或者kworker/uXX:YY
,其中XX
表示worker_pool
的编号,YY
表示worker
的编号,u
表示unbound
;workqueue
子系统初始化完成后,基本就已经将数据结构的关联创建好了,当有work
来进行调度的时候,就能够进行处理了。
以schedule_work
接口为例进行分析:
schedule_work
默认是将work
添加到系统的system_work
工做队列中;
queue_work_on
接口中的操做判断要添加work
的标志位,若是已经置位了WORK_STRUCT_PENDING_BIT
,代表已经添加到了队列中等待执行了,不然,须要调用__queue_work
来进行添加。注意了,这个操做是在关中断的状况下进行的,由于工做队列使用WORK_STRUCT_PENDING_BIT
位来同步work
的插入和删除操做,设置了这个比特后,而后才能执行work
,这个过程可能被中断或抢占打断;
workqueue
的标志位设置了__WQ_DRAINING
,代表工做队列正在销毁,全部的work
都要处理完,此时不容许再将work
添加到队列中,有一种特殊状况:销毁过程当中,执行work
时又触发了新的work
,也就是所谓的chained work
;
判断workqueue
的类型,若是是bound
类型,根据CPU来获取pool_workqueue
,若是是unbound
类型,经过node号来获取pool_workqueue
;
get_work_pool
获取上一次执行work
的worker_pool
,若是本次执行的worker_pool
与上次执行的worker_pool
不一致,且经过find_worker_executing_work
判断work
正在某个worker_pool
中的worker
中执行,考虑到缓存热度,放到该worker
执行是更合理的选择,进而根据该worker
获取到pool_workqueue
;
判断pool_workqueue
活跃的work
数量,少于最大限值则将work
加入到pool->worklist
中,不然加入到pwq->delayed_works
链表中,若是__need_more_worker
判断没有worker
在执行,则唤醒worker
内核线程执行;
总结:
schedule_work
完成的工做是将work
添加到对应的链表中,而在添加的过程当中,首先是须要肯定pool_workqueue
;pool_workqueue
对应一个worker_pool
,所以肯定了pool_workqueue
也就肯定了worker_pool
,进而能够将work
添加到工做链表中;pool_workqueue
的肯定分为三种状况:1)bound
类型的工做队列,直接根据CPU号获取;2)unbound
类型的工做队列,根据node号获取,针对unbound
类型工做队列,pool_workqueue
的释放是异步执行的,须要判断refcnt
的计数值,所以在获取pool_workqueue
时可能要屡次retry
;3)根据缓存热度,优先选择正在被执行的worker_pool
;work
添加到工做队列后,最终的执行在worker_thread
函数中:
在建立worker
时,建立内核线程,执行函数为worker_thread
;
worker_thread
在开始执行时,设置标志位PF_WQ_WORKER
,调度器在进行调度处理时会对task进行判断,针对workerqueue worker
有特殊处理;
worker
对应的内核线程,在没有处理work
的时候是睡眠状态,当被唤醒的时候,跳转到woke_up
开始执行;
woke_up
以后,若是此时worker
是须要销毁的,那就进行清理工做并返回。不然,离开IDLE
状态,并进入recheck
模块执行;
recheck
部分,首先判断是否须要更多的worker
来处理,若是没有任务处理,跳转到sleep
地方进行睡眠。有任务须要处理时,会判断是否有空闲内核线程以及是否须要动态建立,再清除掉worker
的标志位,而后遍历工做链表,对链表中的每一个节点调用process_one_worker
来处理;
sleep
部分比较好理解,没有任务处理时,worker
进入空闲状态,并将当前的内核线程设置成睡眠状态,让出CPU;
总结:
worker_pool
的内核线程池时,若是有PENDING
状态的work
,而且发现没有正在运行的工做线程(worker_pool->nr_running == 0
),唤醒空闲状态的内核线程,或者动态建立内核线程;work
已经在同一个worker_pool
的其余worker
中执行,再也不对该work
进行处理;work
的执行函数为process_one_worker
:
work
可能在同一个CPU上不一样的worker
中运行,直接退出;worker->current_func()
,完成最终work
的回调函数执行;worker_pool
经过nr_running
字段来在不一样的状态机之间进行切换;worker_pool
中有work
须要处理时,须要至少保证有一个运行状态的worker
,当nr_running
大于1时,将多余的worker
进入IDLE状态,没有work
须要处理时,全部的worker
都会进入IDLE状态;work
时,若是回调函数阻塞运行,那么会让worker
进入睡眠状态,此时调度器会进行判断是否须要唤醒另外一个worker
;worker
都存放在idle_list
链表中,若是空闲时间超过了300秒,则会将其进行销毁;Running->Suspend
worker
进入睡眠状态时,若是该worker_pool
没有其余的worker
处于运行状态,那么是须要唤醒一个空闲的worker
来维持并发处理的能力;Suspend->Running
wake_up_worker
来进行唤醒处理,最终判断若是该worker
不在运行状态,则增长worker_pool
的nr_running
值;worker_pool
初始化时,注册了timer的回调函数,用于定时对空闲链表上的worker
进行处理,若是worker
太多,且空闲时间太长,超过了5分钟,那么就直接进行销毁处理了;worker_thread
函数时,若是没有空闲的worker
,会调用manage_workers
接口来建立更多的worker
来处理工做;Documentation/core-api/workqueue.rst
http://kernel.meizu.com/linux-workqueue.html
洗洗睡了,收工!
欢迎关注公众号,不按期分享Linux内核机制文章