Concurrency Managed Workqueue(三)建立workqueue代碼分析

1、前言

本文主要以__alloc_workqueue_key函數爲主線，描述CMWQ中的建立一個workqueue實例的代碼過程。

 

2、WQ_POWER_EFFICIENT的處理

__alloc_workqueue_key函數的一開始有以下的代碼：

if ((flags & WQ_POWER_EFFICIENT) && wq_power_efficient)
        flags |= WQ_UNBOUND;

在kernel中，有兩種線程池，一種是線程池是per cpu的，也就是說，系統中有多少個cpu，就會建立多少個線程池，cpu x上的線程池建立的worker線程也只會運行在cpu x上。另一種是unbound thread pool，該線程池建立的worker線程能夠調度到任意的cpu上去。因爲cache locality的緣由，per cpu的線程池的性能會好一些，可是對power saving有一些影響。設計每每如此，workqueue須要在performance和power saving之間平衡，想要更好的性能，那麼最好讓一個cpu上的worker thread來處理work，這樣的話，cache命中率會比較高，性能會更好。可是，從電源管理的角度來看，最好的策略是讓idle狀態的cpu儘量的保持idle，而不是反覆idle，working，idle again。

咱們來一個例子輔助理解上面的內容。在t1時刻，work被調度到CPU A上執行，t2時刻work執行完畢，CPU A進入idle，t3時刻有一個新的work須要處理，這時候調度work到那個CPU會好些呢？是處於working狀態的CPU B仍是處於idle狀態的CPU A呢？若是調度到CPU A上運行，那麼，因爲以前處理過work，其cache內容新鮮熱辣，處理起work固然是駕輕就熟，速度很快，可是，這須要將CPU A從idle狀態中喚醒。選擇CPU B呢就不存在將CPU 從idle狀態喚醒，從而獲取power saving方面的好處。

瞭解了上面的基礎內容以後，咱們再來檢視per cpu thread pool和unbound thread pool。當workqueue收到一個要處理的work，若是該workqueue是unbound類型的話，那麼該work由unbound thread pool處理並把調度該work去哪個CPU執行這樣的策略交給系統的調度器模塊來完成，對於scheduler而言，它會考慮CPU core的idle狀態，從而儘量的讓CPU保持在idle狀態，從而節省了功耗。所以，若是一個workqueue有WQ_UNBOUND這樣的flag，則說明該workqueue上掛入的work處理是考慮到power saving的。若是workqueue沒有WQ_UNBOUND flag，則說明該workqueue是per cpu的，這時候，調度哪個CPU core運行worker thread來處理work已經不是scheduler能夠控制的了，這樣，也就間接影響了功耗。

有兩個參數能夠控制workqueue在performance和power saving之間的平衡：

一、各個workqueue須要經過WQ_POWER_EFFICIENT來標記本身在功耗方面的屬性

二、系統級別的內核參數workqueue.power_efficient。

使用workqueue的用戶知道本身在電源管理方面的特色，若是該workqueue在unbound的時候會極大的下降功耗，那麼就須要加上WQ_POWER_EFFICIENT的標記。這時候，若是沒有標記WQ_UNBOUND，那麼缺省workqueue會建立per cpu thread pool來處理work。不過，也能夠經過workqueue.power_efficient這個內核參數來修改workqueue的行爲：

#ifdef CONFIG_WQ_POWER_EFFICIENT_DEFAULT
static bool wq_power_efficient = true;
#else
static bool wq_power_efficient;
#endif

module_param_named(power_efficient, wq_power_efficient, bool, 0444);

若是wq_power_efficient設定爲true，那麼WQ_POWER_EFFICIENT的標記的workqueue就會強制按照unbound workqueue來處理，即便沒有標記WQ_UNBOUND。

 

3、分配workqueue的內存

if (flags & WQ_UNBOUND)
    tbl_size = nr_node_ids * sizeof(wq->numa_pwq_tbl[0]); －－－only for unbound workqueue

wq = kzalloc(sizeof(*wq) + tbl_size, GFP_KERNEL);

if (flags & WQ_UNBOUND) {
        wq->unbound_attrs = alloc_workqueue_attrs(GFP_KERNEL); －－only for unbound workqueue
    }

代碼很簡單，與其要解釋代碼，不如來解釋一些基本概念。

一、workqueue和pool workqueue的關係

咱們先給出一個簡化版本的workqueue_struct定義，以下：

struct workqueue_struct {
    struct list_head    pwqs; 
    struct list_head    list;

    struct pool_workqueue __percpu *cpu_pwqs;  －－－－－指向per cpu的pool workqueue
    struct pool_workqueue __rcu *numa_pwq_tbl[]; －－－－指向per node的pool workqueue
};

這裏涉及2個數據結構：workqueue_struct和pool_workqueue，爲什麼如此處理呢？咱們知道，在CMWQ中，workqueue和thread pool沒有嚴格的一一對應關係了，所以，系統中的workqueue們共享一組thread pool，所以，workqueue中的成員包括兩個類別：global類型和per thread pool類型的，咱們把那些per thread pool類型的數據集合起來就造成了pool_workqueue的定義。

掛入workqueue的work終究須要worker pool中的某個worker thread來處理，也就是說，workqueue要和系統中那些共享的worker thread pool進行鏈接，這是經過pool_workqueue（該數據結構會包含一個指向worker pool的指針）的數據結構來管理的。和這個workqueue相關的pool_workqueue被掛入一個鏈表，鏈表頭就是workqueue_struct中的pwqs成員。

和舊的workqueue機制同樣，系統維護了一個全部workqueue的list，list head定義以下：

static LIST_HEAD(workqueues);

workqueue_struct中的list成員就是掛入這個鏈表的節點。

workqueue有兩種：unbound workqueue和per cpu workqueue。對於per cpu類型，cpu_pwqs指向了一組per cpu的pool_workqueue數據結構，用來維護workqueue和per cpu thread pool之間的關係。每一個cpu都有兩個thread pool，normal和高優先級的線程池，到底cpu_pwqs指向哪個pool_workqueue（worker thread）是和workqueue的flag相關，若是標有WQ_HIGHPRI，那麼cpu_pwqs指向高優先級的線程池。unbound workqueue對應的pool_workqueue和workqueue屬性相關，咱們在下一節描述。

二、workqueue attribute

掛入workqueue的work終究是須要worker線程來處理，針對worker線程有下面幾個考量點（咱們稱之attribute）：

（1）該worker線程的優先級

（2）該worker線程運行在哪個CPU上

（3）若是worker線程能夠運行在多個CPU上，且這些CPU屬於不一樣的NUMA node，那麼是否在全部的NUMA node中均可以獲取良好的性能。

對於per-CPU的workqueue，2和3不存在問題，哪一個cpu上queue的work就在哪一個cpu上執行，因爲只能在一個肯定的cpu上執行，所以起NUMA的node也是肯定的（一個CPU不可能屬於兩個NUMA node）。置於優先級，per-CPU的workqueue使用WQ_HIGHPRI來標記。綜上所述，per-CPU的workqueue不須要單獨定義一個workqueue attribute，這也是爲什麼在workqueue_struct中只有unbound_attrs這個成員來記錄unbound workqueue的屬性。

unbound workqueue因爲不綁定在具體的cpu上，能夠運行在系統中的任何一個cpu，直覺上彷佛系統中有一個unbound thread pool就OK了，不過讓一個thread pool建立多種屬性的worker線程是一個好的設計嗎？本質上，thread pool應該建立屬性同樣的worker thread。所以，咱們經過workqueue屬性來對unbound workqueue進行分類，workqueue屬性定義以下：

struct workqueue_attrs {
    int            nice;        /* nice level */
    cpumask_var_t        cpumask;    /* allowed CPUs */
    bool            no_numa;    /* disable NUMA affinity */
};

nice是一個和thread優先級相關的屬性，nice越低則優先級越高。cpumask是該workqueue掛入的work容許在哪些cpu上運行。no_numa是一個和NUMA affinity相關的設定。

三、unbound workqueue和NUMA之間的聯繫

UMA系統中，全部的processor看到的內存都是同樣的，訪問速度也是同樣，無所謂local or remote，所以，內核線程若是要分配內存，那麼也是無所謂，統一安排便可。在NUMA系統中，不一樣的一個或者一組cpu看到的memory是不同的，咱們假設node 0中有CPU A和B，node 1中有CPU C和D，若是運行在CPU A上內核線程如今要遷移到CPU C上的時候，悲劇發生了：該線程在A CPU建立並運行的時候，分配的內存是node 0中的memory，這些memory是local的訪問速度很快，當遷移到CPU C上的時候，原來local memory變成remote，性能大大下降。所以，unbound workqueue須要引入NUMA的考量點。

NUMA是內存管理的範疇，本文不會深刻描述，咱們暫且放開NUMA，先思考這樣的一個問題：一個肯定屬性的unbound workqueue須要幾個線程池？看起來一個就夠了，畢竟workqueue的屬性已經肯定了，一個線程池建立相同屬性的worker thread就好了。可是咱們來看一個例子：假設workqueue的work是能夠在node 0中的CPU A和B，以及node 1中CPU C和D上處理，若是隻有一個thread pool，那麼就會存在worker thread在不一樣node之間的遷移問題。爲了解決這個問題，實際上unbound workqueue其實是建立了per node的pool_workqueue（thread pool）

固然，是否使用per node的pool workqueue用戶是能夠經過下面的參數進行設定的：

（1）workqueue attribute中的no_numa成員

（2）經過workqueue.disable_numa這個參數，disable全部workqueue的numa affinity的支持。

static bool wq_disable_numa;
module_param_named(disable_numa, wq_disable_numa, bool, 0444);

 

4、初始化workqueue的成員

va_start(args, lock_name);
vsnprintf(wq->name, sizeof(wq->name), fmt, args);－－－－－set workqueue name
va_end(args);

max_active = max_active ?: WQ_DFL_ACTIVE;
max_active = wq_clamp_max_active(max_active, flags, wq->name);
wq->flags = flags;
wq->saved_max_active = max_active;
mutex_init(&wq->mutex);
atomic_set(&wq->nr_pwqs_to_flush, 0);
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, 0);
INIT_LIST_HEAD(&wq->list);

除了max active，沒有什麼要說的，代碼都簡單並且直觀。若是用戶沒有設定max active（或者說max active等於0），那麼系統會給出一個缺省的設定。系統定義了兩個最大值WQ_MAX_ACTIVE（512）和WQ_UNBOUND_MAX_ACTIVE（和cpu數目有關，最大值是cpu數目乘以4，固然也不能大於WQ_MAX_ACTIVE），分別限定per cpu workqueue和unbound workqueue的最大能夠建立的worker thread的數目。wq_clamp_max_active能夠將max active限制在一個肯定的範圍內。

 

5、分配pool workqueue的內存並創建workqueue和pool workqueue的關係

這部分的代碼主要涉及alloc_and_link_pwqs函數，以下：

static int alloc_and_link_pwqs(struct workqueue_struct *wq)
{
    bool highpri = wq->flags & WQ_HIGHPRI;－－－－normal or high priority？
    int cpu, ret;

    if (!(wq->flags & WQ_UNBOUND)) {－－－－－per cpu workqueue的處理
        wq->cpu_pwqs = alloc_percpu(struct pool_workqueue);

        for_each_possible_cpu(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]); 
            link_pwq(pwq);－－－－上面兩行代碼用來創建workqueue、pool wq和thread pool之間的關係
        }
        return 0;
    } else if (wq->flags & __WQ_ORDERED) {－－－－－ordered unbound workqueue的處理
        ret = apply_workqueue_attrs(wq, ordered_wq_attrs[highpri]);
        return ret;
    } else {－－－－－unbound workqueue的處理
        return apply_workqueue_attrs(wq, unbound_std_wq_attrs[highpri]);
    }
}

經過alloc_percpu能夠爲每個cpu分配一個pool_workqueue的memory。每一個pool_workqueue都有一個對應的worker thread pool，對於per-CPU workqueue，它是靜態定義的，以下：

static DEFINE_PER_CPU_SHARED_ALIGNED(struct worker_pool [NR_STD_WORKER_POOLS],
                     cpu_worker_pools);

init_pwq函數初始化pool_workqueue，最重要的是設定其對應的workqueue和worker pool。link_pwq主要是將pool_workqueue掛入它所屬的workqueue的鏈表中。對於unbound workqueue，apply_workqueue_attrs完成分配pool workqueue並創建workqueue和pool workqueue的關係。

 

6、應用新的attribute到workqueue中

unbound workqueue有兩種，一種是normal type，另一種是ordered type，這種workqueue上的work是嚴格按照順序執行的，不存在併發問題。ordered unbound workqueue的行爲相似過去的single thread workqueue。可是，不管那種類型的unbound workqueue都使用apply_workqueue_attrs來創建workqueue、pool wq和thread pool之間的關係。

一、健康檢查。

if (WARN_ON(!(wq->flags & WQ_UNBOUND)))
    return -EINVAL;

if (WARN_ON((wq->flags & __WQ_ORDERED) && !list_empty(&wq->pwqs)))
    return -EINVAL;

只有unbound類型的workqueue纔有attribute，才能夠apply attributes。對於ordered類型的unbound workqueue，屬於它的pool workqueue（worker thread pool）只能有一個，不然沒法限制work是按照順序執行。

二、分配內存並初始化

pwq_tbl = kzalloc(nr_node_ids * sizeof(pwq_tbl[0]), GFP_KERNEL);
new_attrs = alloc_workqueue_attrs(GFP_KERNEL);
tmp_attrs = alloc_workqueue_attrs(GFP_KERNEL);
copy_workqueue_attrs(new_attrs, attrs);
cpumask_and(new_attrs->cpumask, new_attrs->cpumask, cpu_possible_mask);
copy_workqueue_attrs(tmp_attrs, new_attrs);

pwq_tbl數組用來保存unbound workqueue各個node的pool workqueue的指針，new_attrs和tmp_attrs都是一些計算workqueue attribute的中間變量，開始的時候設定爲用戶傳入的workqueue的attribute。

三、如何爲unbound workqueue的pool workqueue尋找對應的線程池？

具體的代碼在get_unbound_pool函數中。本節不描述具體的代碼，只說明基本原理，你們能夠自行閱讀代碼。

per cpu的workqueue的pool workqueue對應的線程池也是per cpu的，每一個cpu有兩個線程池（normal和high priority），所以將pool workqueue和thread pool對應起來是很是簡單的事情。對於unbound workqueue，對應關係沒有那麼直接，若是屬性相同，多個unbound workqueue的pool workqueue可能對應一個thread pool。

系統使用哈希表來保存全部的unbound worker thread pool，定義以下：

static DEFINE_HASHTABLE(unbound_pool_hash, UNBOUND_POOL_HASH_ORDER);

在建立unbound workqueue的時候，pool workqueue對應的worker thread pool須要在這個哈希表中搜索，若是有相同屬性的worker thread pool的話，那麼就不須要建立新的線程池，代碼以下：

hash_for_each_possible(unbound_pool_hash, pool, hash_node, hash) {
    if (wqattrs_equal(pool->attrs, attrs)) { －－－－檢查屬性是否相同
        pool->refcnt++;
        return pool; －－－－－－－在哈希表找到適合的unbound線程池
    }
}

若是沒有相同屬性的thread pool，那麼須要建立一個並掛入哈希表。

四、給各個node分配pool workqueue並初始化

在進入代碼以前，先了解一些基礎知識。缺省狀況下，掛入unbound workqueue的works最好是考慮NUMA Affinity，這樣能夠獲取更好的性能。固然，實際上用戶能夠經過workqueue.disable_numa這個內核參數來關閉這個特性，這時候，系統須要一個default pool workqueue（workqueue_struct的dfl_pwq成員），全部的per node的pool workqueue指針都是執行default pool workqueue。

workqueue.disable_numa是enable的狀況下是否不須要default pool workqueue了呢？也不是，咱們舉一個簡單的例子，一個系統的構成是這樣的：node 0中有CPU A和B，node 1中有CPU C和D，node 2中有CPU E和F，假設workqueue的attribute規定work只能在CPU A 和C上運行，那麼在node 0和node 1中建立本身的pool workqueue是ok的，畢竟node 0中有CPU A，node 1中有CPU C，該node建立的worker thread能夠在A或者C上運行。可是對於node 2節點，沒有任何的CPU容許處理該workqueue的work，在這種狀況下，沒有必要爲node 2創建本身的pool workqueue，而是使用default pool workqueue。

OK，咱們來看代碼：

dfl_pwq = alloc_unbound_pwq(wq, new_attrs); －－－－－分配default pool workqueue

for_each_node(node) { －－－－遍歷node
    if (wq_calc_node_cpumask(attrs, node, -1, tmp_attrs->cpumask)) { －－－是否使用default pool wq
        pwq_tbl[node] = alloc_unbound_pwq(wq, tmp_attrs); －－－該node使用本身的pool wq
    } else {
        dfl_pwq->refcnt++;
        pwq_tbl[node] = dfl_pwq; －－－－該node使用default pool wq
    }
}

值得一提的是wq_calc_node_cpumask這個函數，這個函數會根據該node的cpu狀況以及workqueue attribute中的cpumask成員來更新tmp_attrs->cpumask，所以，在pwq_tbl[node] = alloc_unbound_pwq(wq, tmp_attrs); 這行代碼中，爲該node分配pool workqueue對應的線程池的時候，去掉了本node中不存在的cpu。例如node 0中有CPU A和B，workqueue的attribute規定work只能在CPU A 和C上運行，那麼建立node 0上的pool workqueue以及對應的worker thread pool的時候，須要刪除CPU C，也就是說，node 0上的線程池的屬性中的cpumask僅僅支持CPU A了。

五、安裝

全部的node的pool workqueue及其worker thread pool已經ready，須要安裝到workqueue中了：

for_each_node(node)
        pwq_tbl[node] = numa_pwq_tbl_install(wq, node, pwq_tbl[node]); 
    link_pwq(dfl_pwq);
    swap(wq->dfl_pwq, dfl_pwq);

代碼很是簡單，這裏就不細述了。