扒開 Linux 中斷的底褲之 workqueue

作者：佚名 2021-08-10 12:05:19

workqueue 是除了 softirq 和 tasklet 以外最常用的下半部機制之一。workqueue 的本質是把 work 交給一個內核線程，在進程上下文調度的時候執行。

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 是一對多的關系。內核中工作隊列分為兩種：

bound：綁定處理器的工作隊列，每個 worker 創建的內核線程綁定到特定的 CPU 上運行。
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 是一對多的關系。

bound 類型的工作隊列：worker_pool 是 Per-CPU 創建，每個 CPU 都有兩個 worker_pool，對應不同的優先級，nice 值分別為 0 和 -20。
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 鏈表中

可以用下圖來總結：

workqueue 的初始化

內核在啟動的時候會對 workqueue 做初始化，workqueue 的初始化包含兩個階段，分別是 workqueue_init_early 和 workqueue_init。

workqueue_init_early

分配 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。

分配 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 調度

將 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。

設置標志位 PF_WQ_WORKER，調度器在進行調度處理時會對 task 進行判斷，針對 workerqueue worker 有特殊的處理。
worker 被喚醒的時候，跳轉到 woke_up 執行。
woke_up 中，如果此時 worker 是需要銷毀的，那就進行清理工作并返回。否則，離開 IDLE 狀態，并進入 recheck 模塊執行。
recheck 中，判斷是否需要更多的 worker 來處理，如果沒有任務處理，跳轉到 sleep 地方進行睡眠。如果有任務需要處理時，遍歷工作鏈表，對鏈表中的每個節點調用 process_one_work 來執行 work 的回調函數，即 INIT_WORK 里的回調函數。

sleep 中，沒有任務處理時，worker 進入空閑狀態，并將當前的內核線程設置成睡眠狀態，讓出 CPU。

總結

責任編輯：龐桂玉來源： Linux學習