扒开 Linux 中断的底裤之 workqueue

324次阅读  |  发布于3年以前

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
};

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
    ...
};

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

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;

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

可以用下图来总结：

workqueue 的初始化

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

workqueue_init_early

workqueue_init

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

经过上面两个阶段的初始化，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 调度

worker_thread

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

总结

Copyright© 2013-2020

All Rights Reserved 京ICP备2023019179号-8