扒开 Linux 中断的底裤之 workqueue

ゝ一世哀愁。 2022-09-02 11:43 179阅读 0赞

workqueue 是除了 softirq 和 tasklet 以外最常用的下半部机制之一。workqueue 的本质是把 work 交给一个内核线程，在进程上下文调度的时候执行。因为这个特点，所以 workqueue 允许重新调度和睡眠，这种异步执行的进程上下文，能解决因为 softirq 和 tasklet 执行时间长而导致的系统实时性下降等问题。

当驱动程序在进程上下文中有异步执行的工作任务时，可以用 work 来描述工作任务。把 work 添加到一个链表 worklist 中，然后由一个内核线程 worker 遍历链表，串行地执行挂入 worklist 中的所有 work。如果 worklist 中没有 work，那么内核线程 worker 就会变成 IDLE 状态；如果有 work，则执行 work 中的回调函数。

## workqueue 相关的数据结构 ##

关于 workqueue 中几个概念都是 work 相关的数据结构非常容易混淆，大概可以这样来理解：

### work\_struct ： ###

工作。初始化一个 work 并添加到工作队列后，将会将其传递到合适的内核线程来进行处理，它是用于调度的最小单位。

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：低比特存放状态位，高比特存放 worker\_pool 的ID或者 pool\_workqueue 的指针
 *  entry：用于添加到其他队列上
 *  func：工作任务的处理函数，在内核线程中回调

### workqueue\_struct ： ###

工作的集合。workqueue 和 work 是一对多的关系。内核中工作队列分为两种：

1.  bound：绑定处理器的工作队列，每个 worker 创建的内核线程绑定到特定的 CPU 上运行。
2.  unbound：不绑定处理器的工作队列，创建的时候需要指定 WQ\_UNBOUND 标志，内核线程可以在处理器间迁移。

struct workqueue_struct {
     struct list_head pwqs;  /* WR: all pwqs of this wq */   
     struct list_head list;  /* PR: list of all workqueues */  
    
     struct list_head maydays; /* MD: pwqs requesting rescue */    
     struct worker  *rescuer; /* I: rescue worker */  
    
     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 */    
     struct pool_workqueue __rcu *numa_pwq_tbl[]; /* PWR: unbound pwqs indexed by node */    //Per-Node创建pool_workqueue
        ...
    };

*  pwqs：所有的 pool\_workqueue 都添加到本链表中
 *  list：用于将工作队列添加到全局链表 workqueues 中
 *  maydays：rescue状态下的 pool\_workqueue 添加到本链表中
 *  rescuer：rescuer 内核线程，用于处理内存紧张时创建工作线程失败的情况
 *  cpu\_pwqs：Per-CPU 创建 pool\_workqueue
 *  numa\_pwq\_tbl\[\]：Per-Node 创建 pool\_workqueue

### pool\_workqueue： ###

中间人 / 中介，负责建立起 workqueue 和 worker\_pool 之间的关系。workqueue 和 pool\_workqueue 是一对多的关系。

struct pool_workqueue {
     struct worker_pool *pool;  /* I: the associated pool */    
     struct workqueue_struct *wq;  /* I: the owning workqueue */   
    
     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 */   //用于添加到workqueue链表中
        ...
    } __aligned(1 << WORK_STRUCT_FLAG_BITS);

*  pool：指向 worker\_pool
 *  wq：指向所属的 workqueue
 *  nr\_active：活跃的 work 数量
 *  max\_active：活跃的最大 work 数量
 *  delayed\_works：延迟执行的 work 挂入本链表
 *  pwqs\_node：用于添加到 workqueue 链表中
 *  mayday\_node：用于添加到 workqueue 链表中

### worker\_pool： ###

工人的集合。pool\_workqueue 和 worker\_pool 是一对一的关系，worker\_pool 和 worker 是一对多的关系。

1.  bound 类型的工作队列：worker\_pool 是 Per-CPU 创建，每个 CPU 都有两个 worker\_pool，对应不同的优先级，nice 值分别为 0 和 -20。
2.  unbound 类型的工作队列：worker\_pool 创建后会添加到 unbound\_pool\_hash 哈希表中。

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 */
    
     unsigned long  watchdog_ts; /* L: watchdog timestamp */
    
     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 worker  *manager; /* L: purely informational */
     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 */    
        ...
    } ____cacheline_aligned_in_smp;

*  cpu：绑定到 CPU 的 workqueue，代表 CPU ID
 *  node：非绑定类型的 workqueue，代表内存 Node ID
 *  worklist：pending 状态的 work 添加到本链表
 *  nr\_workers：worker 的数量
 *  idle\_list：处于 IDLE 状态的 worker 添加到本链表
 *  busy\_hash：工作状态的 worker 添加到本哈希表中
 *  workers：worker\_pool 管理的 worker 添加到本链表中
 *  hash\_node：用于添加到 unbound\_pool\_hash 中

### worker ： ###

工人。在代码中 worker 对应一个 work\_thread() 内核线程。

struct worker {
     /* on idle list while idle, on busy hash table while busy */
     union {
      struct list_head entry; /* L: while idle */     
      struct hlist_node hentry; /* L: while busy */ 
     };
    
     struct work_struct *current_work; /* L: work being processed */  
     work_func_t  current_func; /* L: current_work's fn */                
     struct pool_workqueue *current_pwq; /* L: current_work's pwq */   
    
     struct list_head scheduled; /* L: scheduled works */   
      
     /* 64 bytes boundary on 64bit, 32 on 32bit */
    
     struct task_struct *task;  /* I: worker task */   
     struct worker_pool *pool;  /* I: the associated pool */   
          /* L: for rescuers */
     struct list_head node;  /* A: anchored at pool->workers */  //添加到worker_pool->workers链表中
          /* A: runs through worker->node */
        ...
    };

*  entry：用于添加到 worker\_pool 的空闲链表中
 *  hentry：用于添加到 worker\_pool 的忙碌列表中
 *  current\_work：当前正在处理的 work
 *  current\_func：当前正在执行的 work 回调函数
 *  current\_pwq：指向当前 work 所属的 pool\_workqueue
 *  scheduled：所有被调度执行的 work 都将添加到该链表中
 *  task：指向内核线程
 *  pool：该 worker 所属的 worker\_pool
 *  node：添加到 worker\_pool->workers 链表中

可以用下图来总结：

![eb0f398450587a774a7d7c2fcb88754b.png][] ![cb9b9743b797d2b57c2fd9ea8ca48a8a.png][]

## workqueue 的初始化 ##

内核在启动的时候会对 workqueue 做初始化，workqueue 的初始化包含两个阶段，分别是 workqueue\_init\_early 和 workqueue\_init。

### workqueue\_init\_early ###

![7eca3a802a2d4c817636c91ec3c9f294.png][]

*  分配 worker\_pool，并且对该结构中的字段进行初始化操作
 *  分配 workqueue\_struct，并且对该结构中的字段进行初始化操作
 *  alloc\_and\_link\_pwqs：分配 pool\_workqueue，将 workqueue\_struct 和 worker\_pool 关联起来

### workqueue\_init ###

这里主要完成的工作是给之前创建好的 worker\_pool，添加一个初始的 worker，然后利用函数 create\_worker，创建名字为 kworker/XX:YY 或者 kworker/uXX:YY 的内核线程。其中 XX 表示 worker\_pool 的编号，YY 表示 worker 的编号，u 表示unbound。

![744c76b5524eaae20ff898321ff13e73.png][]

*  分配 worker，并且对该结构中的字段进行初始化操作
 *  为 worker 创建内核线程 worker\_thread
 *  将 worker 添加到 worker\_pool 中
 *  worker 进入 IDLE 状态

经过上面两个阶段的初始化，workqueue 子系统基本就已经将数据结构的关联建立好了，当有 work 来进行调度的时候，就可以进行处理了。

## 使用 workqueue ##

内核推荐驱动开发者使用默认的 workqueue，而不是新建 workqueue。要使用系统默认的 workqueue，首先需要初始化 work，内核提供了相应的宏 INIT\_WORK。

### 初始化 work ###

#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)

初始化 work 后，就可以调用 shedule\_work 函数把 work 挂入系统默认的 workqueue 中。

### work 调度 ###

![4494a701ebdf174ea446532fcbecc8ea.png][]

*  将 work 添加到系统的 system\_wq 工作队列中。
 *  判断 workqueue 的类型，如果是 bound 类型，根据 CPU 来获取 pool\_workqueue。如果是 unbound 类型，通过 node 号来获取 pool\_workqueue。
 *  判断 pool\_workqueue 活跃的 work 数量，少于最大限值则将 work 加入到 pool->worklist 中，否则加入到 pwq->delayed\_works 链表中。
 *  如果 \_\_need\_more\_worker 判断没有 worker 在执行，则通过 wake\_up\_worker 唤醒 worker 内核线程。

### worker\_thread ###

worker 内核线程的执行函数是 worker\_thread。

![b6ff2b761740a2957a0009f7c9540183.png][]

*  设置标志位 PF\_WQ\_WORKER，调度器在进行调度处理时会对 task 进行判断，针对 workerqueue worker 有特殊的处理。
 *  worker 被唤醒的时候，跳转到 woke\_up 执行。
 *  woke\_up 中，如果此时 worker 是需要销毁的，那就进行清理工作并返回。否则，离开 IDLE 状态，并进入 recheck 模块执行。
 *  recheck 中，判断是否需要更多的 worker 来处理，如果没有任务处理，跳转到 sleep 地方进行睡眠。如果有任务需要处理时，遍历工作链表，对链表中的每个节点调用 process\_one\_work 来执行 work 的回调函数，即 INIT\_WORK 里的回调函数。

![2749d83232219b201602cf0942e08490.png][]

*  sleep 中，没有任务处理时，worker 进入空闲状态，并将当前的内核线程设置成睡眠状态，让出 CPU。

## 总结 ##

![2d1bf13fb75db43f5e023ea3cfa71106.png][]

![虚线阴影分割线][9f0ee2e47c13ca8637901f0b6da9b09d.png]

5T技术资源大放送！包括但不限于：C/C++，Arm, Linux，Android，人工智能，单片机，树莓派，等等。在上面的【人人都是极客】公众号内回复「peter」，即可免费获取！！

![91e5a725fbdefc34c6b17fdc9966b0f9.gif][] **记得点击****分享****、****赞****和****在看****，给我充点儿电吧**

[eb0f398450587a774a7d7c2fcb88754b.png]: https://img-blog.csdnimg.cn/img_convert/eb0f398450587a774a7d7c2fcb88754b.png
[cb9b9743b797d2b57c2fd9ea8ca48a8a.png]: /images/20220829/90f2156b5b7347b3bdf66d14401d415a.png
[7eca3a802a2d4c817636c91ec3c9f294.png]: /images/20220829/fdbc80af3a9b4407a8158d0ba382387d.png
[744c76b5524eaae20ff898321ff13e73.png]: /images/20220829/ced31ff774d14fe7802f88d1ef201126.png
[4494a701ebdf174ea446532fcbecc8ea.png]: /images/20220829/4317041f2229413aa55b6cb21d47298f.png
[b6ff2b761740a2957a0009f7c9540183.png]: /images/20220829/358ca79946214ee6935c1e4e3d4c23c0.png
[2749d83232219b201602cf0942e08490.png]: /images/20220829/7418e4e3eeb34c0897cc6916610d2a41.png
[2d1bf13fb75db43f5e023ea3cfa71106.png]: /images/20220829/3e0d2e24c28e48ac9515caedda7f532d.png
[9f0ee2e47c13ca8637901f0b6da9b09d.png]: /images/20220829/8c606e15615244239e7e679eddf3ce57.png
[91e5a725fbdefc34c6b17fdc9966b0f9.gif]: /images/20220829/ba795f354b294d178b09ab803c78399a.png