---
# System prepended metadata

title: 'Linux 核心設計: Concurrency Managed Workqueue'
tags: [Linux Kernel Internals]

---

# Linux 核心設計: Concurrency Managed Workqueue(CMWQ)

## Why CMWQ? 
> [Concurrency Managed Workqueue (cmwq)](https://www.kernel.org/doc/html/latest/core-api/workqueue.html)

在 Linux 中提供 workqueue 介面以實現非同步的流程執行。使用者以一個 "work item" 來描述需被執行的 context，將其加入到所建立 queue 之中。則 kernel 會選出合適的 "worker" thread 來完成之。

在 CMWQ 之前的 workqueue 實作中，worker thread 的數量是與被建立的 workqueue 數量直接掛勾的。意味著使用者每建立一個新的 workqueue，系統上的 worker thread 總量便會增加。實作中包含了兩種 worker thread 的建立方式：

* multi threaded (MT) wq: 在每個 CPU 上會有一個 worker thread，work item 可以分配不同的 CPU 上執行
* single threaded (ST) wq: 僅以單一的 worker thread 來完成系統上所有的 work item

在 MT wq 的部份，這代表系統需要建立與 CPU 數量乘上 workqueue 總量等價的 worker thread，這是很嚴重的的問題。畢竟隨著 Linux 的演進，一方面 workqueue 的使用者逐漸增加，而現代電腦的 CPU 核心數量同時也不斷增加。這代表一些系統在剛啟動(booting)時，其 32k 的 PID 空間就會被 worker thread 的 process 給佔滿。

資源限制和 concurrency level 之間存在取捨關係，ST wq 是為整個系統提供唯一的 context，因此所有的 work 都是一一排隊執行，造成 concurrency level 不佳，雖然對資源的使用量比較低。而 MT wq 為每個 CPU 提供一個 context，代價是消耗了大量資源。但即便如此它提供的 concurrency 效果卻不盡如人意。一個關鍵的原因是 MT wq 的 worker thread 的數目是與 CPU 數量嚴格綁定的，且 worker thread 的 work item 也不能互相轉移。另一方面，傳統 workqueue 也容易出現 deadlock 問題。

總結來說，使用 ST wq 得到的 concurrency level 較低，但資源上較為節省。然而使用 MT wq 固然嚴重的消耗資源，能提高的 concurrency level 卻很有限。這樣不對等的 trade off 讓過去的 workqueue 並不是那麼好用。

CMWQ 就在此背景下誕生，並專注在以下目標:
* 與原始 workqueue API 的相容性
* 捨棄各個 workqueue 獨立對應一組 worker-pools 的作法。取而代之，所有 workqueue 會共享 per-CPU worker pools，並按需提供靈活的 concurrency level，避免大量的資源耗損
* worker pool 和 concurrency level 間的調節由系統內部處理，對使用者來說可以視為一個黑盒子

## CMWQ 的設計

在 CMWQ 中，每一項函式流程的執行被抽象為一個 work item ([`work_struct`](https://elixir.bootlin.com/linux/v6.4/source/include/linux/workqueue.h#L98))。

```cpp
typedef void (*work_func_t)(struct 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
};
```

work itme 中包含一個 `func` 成員，是所要被執行的函式指標。當 Linux driver 或 subsystem 想要非同步式的執行某個函數，它就必須建立 `work_struct` 並設置 `func` 以指向要運行的函數，並將其加入到 workqueue 中。

worker thread 負責從 queue 中一個接一個地取出 work item 並執行其中的 `func`。如果沒有 work item 在排隊，則 worker thread 就會處於 idle 狀態。在 worker thread 之上對其管理的結構稱為 worker-pools。

設計上，CMWQ 中對於每個 CPU 會有兩個 worker-pools。一個針對普通的 work，另一個則針對高優先級的 work。此外還有一些未綁定 CPU 的額外 worker-pools，這些 pools 的數量則是動態的。使用者通過 workqueue API 建立 work item 並進行排隊，並且允許設置 flags 來影響部份 work item 的執行方式。例如 CPU locality、concurrency 的限制、優先級等。

當將 work item 排隊到 workqueue 時，會根據 queue 參數和 workqueue 的屬性決定目標的 worker-pools，比如可以設定 workqueue 中的 work item 僅由指定 CPU 之普通或高優先級的 worker-pools 來執行。

對於限定 CPU 的 worker-pools，其 concurrency 的管理方式是將自己與 scheduler 掛勾。則每當任一 worker 被喚醒或是進入睡眠時，worker-pools 會收到通知，藉此來跟踪當前可運行的 woker 數量。一般而言，work item 不太會佔用 CPU 資源，因此設計上傾向使用最少數量的 worker thread 但避免 bandwidth 的損失。實作上，只要該 CPU 上有一個或多個 runnable worker thread，worker-pools 就暫時不會執行新的 work。一直到最後一個正在運行的 worker thread 進入睡眠狀態時，才立即排程一個新的 worker thread，這樣 CPU 就不會在仍有尚未處理的 work item 時無所事事，但也不至於過度的建立大量 worker thread。

另一方面，除了 kthreads 的內存空間之外，保留 idle 的 worker thread 不會消耗其他資源，因此 CWMQ 會保留一個  idle 的 worker thread 一段時間。這樣如果隨即有 work 要處理則可以直接沿用，不必再重新建立。

如果使用不限定 CPU 的 workqueue，使用者可以藉由對應 API 來為此設定自定義的屬性，則 workqueue 將自動建立與屬性對應的 worker-pools。此種 workqueue 對 concurrency level 的調節將由使用者定義。此外，如果想讓綁定 CPU 的 workqueue 也有此特性，API 也提供相應的 flag 以忽略 concurrency 的管理。

## CMWQ 的初始化

要允許核心內部或者使用者能夠註冊 workqueue，kernel 首先要透過 [`workqueue_init_early`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L6253) 和 [`workqueue_init`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L6331) 兩個階段的初始化來將整個子系統準備好。

### 上半部份: `workqueue_init_early`


作為上半部份的 `workqueue_init_early` 主要目的在於盡早初始化好 kernel 中的相關資料結構，以利於 kernel 的其他部份可以在早期就能建立 workqueue 和註冊 work item。由於 kthread 尚不能建立，這時候 work 還不能被完成，必須等到下半部份的  `workqueue_init` 也完成後才能繼續進行。

```cpp
void __init workqueue_init_early(void)
{
	int std_nice[NR_STD_WORKER_POOLS] = { 0, HIGHPRI_NICE_LEVEL };
	int i, cpu;

	BUILD_BUG_ON(__alignof__(struct pool_workqueue) < __alignof__(long long));

	BUG_ON(!alloc_cpumask_var(&wq_unbound_cpumask, GFP_KERNEL));
	cpumask_copy(wq_unbound_cpumask, housekeeping_cpumask(HK_TYPE_WQ));
	cpumask_and(wq_unbound_cpumask, wq_unbound_cpumask, housekeeping_cpumask(HK_TYPE_DOMAIN));

	pwq_cache = KMEM_CACHE(pool_workqueue, SLAB_PANIC);
```

具體初始化的項目其一是 [`wq_unbound_mask`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L337)，這標示可以拿來服務 unbound workqueue 的 CPU 有哪些。

而 [`pwq_cache`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L308) 是建立可用來配置 `pool_workqueue` 的 memory pool。若預期未來會有頻繁建立和釋放 `pool_workqueue` 空間的需求，使用 `kmem_cache` 相較於直接採通用的 `kmalloc` 配置可以獲得效率上的優勢或是減輕記憶體碎片化的問題。

```cpp
	/* initialize CPU pools */
	for_each_possible_cpu(cpu) {
		struct worker_pool *pool;

		i = 0;
		for_each_cpu_worker_pool(pool, cpu) {
			BUG_ON(init_worker_pool(pool));
			pool->cpu = cpu;
			cpumask_copy(pool->attrs->cpumask, cpumask_of(cpu));
			pool->attrs->nice = std_nice[i++];
			pool->node = cpu_to_node(cpu);

			/* alloc pool ID */
			mutex_lock(&wq_pool_mutex);
			BUG_ON(worker_pool_assign_id(pool));
			mutex_unlock(&wq_pool_mutex);
		}
	}
```

接著會為每個 CPU 建立個別的 `struct worker_pool`。內層 for loop 是因為前面有提到的會有一個針對普通的 work 和另一個針對高優先級的 work

```cpp
	/* create default unbound and ordered wq attrs */
	for (i = 0; i < NR_STD_WORKER_POOLS; i++) {
		struct workqueue_attrs *attrs;

		BUG_ON(!(attrs = alloc_workqueue_attrs()));
		attrs->nice = std_nice[i];
		unbound_std_wq_attrs[i] = attrs;

		/*
		 * An ordered wq should have only one pwq as ordering is
		 * guaranteed by max_active which is enforced by pwqs.
		 * Turn off NUMA so that dfl_pwq is used for all nodes.
		 */
		BUG_ON(!(attrs = alloc_workqueue_attrs()));
		attrs->nice = std_nice[i];
		attrs->no_numa = true;
		ordered_wq_attrs[i] = attrs;
	}
```

unbound 的部份不會建先建立 worker pool，不過會建立預設的 attribute [`unbound_std_wq_attrs`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L363) 和 [`ordered_wq_attrs`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L366)。

```cpp
	system_wq = alloc_workqueue("events", 0, 0);
	system_highpri_wq = alloc_workqueue("events_highpri", WQ_HIGHPRI, 0);
	system_long_wq = alloc_workqueue("events_long", 0, 0);
	system_unbound_wq = alloc_workqueue("events_unbound", WQ_UNBOUND,
					    WQ_UNBOUND_MAX_ACTIVE);
	system_freezable_wq = alloc_workqueue("events_freezable",
					      WQ_FREEZABLE, 0);
	system_power_efficient_wq = alloc_workqueue("events_power_efficient",
					      WQ_POWER_EFFICIENT, 0);
	system_freezable_power_efficient_wq = alloc_workqueue("events_freezable_power_efficient",
					      WQ_FREEZABLE | WQ_POWER_EFFICIENT,
					      0);
	BUG_ON(!system_wq || !system_highpri_wq || !system_long_wq ||
	       !system_unbound_wq || !system_freezable_wq ||
	       !system_power_efficient_wq ||
	       !system_freezable_power_efficient_wq);
}
```

在 worker pool 被創建完成之後，就能利用 [`alloc_workqueue`](#建立-workqueue) 來建立 workqueue `struct workqueue_struct` 了。這裡可以看到 kernel 內部事先建立出了各種不同類型的 workqueue。

### `workqueue_init`

[`workqueue_init`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L6331) 是 workqueue 子系統的下半部分。此階段主要在建立 workqueue 所需的 worker thread，且也允許日後建立更多的 thread。

```cpp
void __init workqueue_init(void)
{
	struct workqueue_struct *wq;
	struct worker_pool *pool;
	int cpu, bkt;

	/*
	 * It'd be simpler to initialize NUMA in workqueue_init_early() but
	 * CPU to node mapping may not be available that early on some
	 * archs such as power and arm64.  As per-cpu pools created
	 * previously could be missing node hint and unbound pools NUMA
	 * affinity, fix them up.
	 *
	 * Also, while iterating workqueues, create rescuers if requested.
	 */
	wq_numa_init();

	mutex_lock(&wq_pool_mutex);

	for_each_possible_cpu(cpu) {
		for_each_cpu_worker_pool(pool, cpu) {
			pool->node = cpu_to_node(cpu);
		}
	}
```
這裡先將各個 pool 對應到正確的 node id。細心觀察可以發現 `workqueue_init_early` 已經做過類似的初始化，這裡還需進行的理由可以參照註解。

```cpp
	list_for_each_entry(wq, &workqueues, list) {
		wq_update_unbound_numa(wq, smp_processor_id(), true);
		WARN(init_rescuer(wq),
		     "workqueue: failed to create early rescuer for %s",
		     wq->name);
	}

	mutex_unlock(&wq_pool_mutex);
```

`list_for_each_entry` 走訪所有 [`workqueues`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L333)(每個用 `alloc_workqueue` 建立的 workqueue 會 pending 到 list 中) list 中的 `workqueue_struct`，並逐一進行以下兩項操作:

* `wq_update_unbound_numa`: CPU 熱插拔相關?
* `init_rescuer`: 假設 workqueue 中的 work 可以釋放記憶體，但資源吃緊情況下，為了順利進行需要釋放記憶體建立更多的 worker thread，這就形成 deadlock 的狀況。為了避免這種問題，CMWQ 針對與記憶體釋放有關的 workqueue 保留了一個 rescuer thread，後者保證了工作的進展。這就是透過 `init_rescuer` 建立的。如果 workqueue 的 flag 設定含 `WQ_MEM_RECLAIM`，會為這類型的 wq 嘗試建立 rescuer。

```cpp
	/* create the initial workers */
	for_each_online_cpu(cpu) {
		for_each_cpu_worker_pool(pool, cpu) {
			pool->flags &= ~POOL_DISASSOCIATED;
			BUG_ON(!create_worker(pool));
		}
	}
```

這裡與 CPU-bounded worker pool 有關。[`create_worker`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L1938) 正式建立 `worker_pool` 底下的 [`worker`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue_internal.h#L24)，`worker` 也就包含了處理 worker item 的 kernel thread。

![](https://hackmd.io/_uploads/SJr0DbqA3.png)


```cpp
	hash_for_each(unbound_pool_hash, bkt, pool, hash_node)
		BUG_ON(!create_worker(pool));
```

初始化也應用在目前存在的 unboued worker pool，也就是一開始在 `workqueue_init_early` 設置的那些，這些是透過 hash table 來管理。

```cpp
	wq_online = true;
	wq_watchdog_init();
}
```

上下兩個階段的 workqueue 初始化完成後，可以將 `wq_online` 設置為 true。主要可以幫助 kernel 當下是否可以直接建立 worker 或 rescuer kthread。




## API 的使用

基於 6.4 版本之 Linux kernel 進行探討

### 建立 workqueue

workqueue 的建立需透過 [`alloc_workqueue`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L4425)，可藉之得到配置的 [`struct workqueue_struct`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L264)

```cpp
struct workqueue_struct *alloc_workqueue(const char *fmt,
					 unsigned int flags,
					 int max_active, ...);
```

`fmt` 和最後的 Variable Length Argument (`...`) 和 workqueue 的命名有關。`max_active` 則是限定 **per-CPU(不是 global!)** 可處理 work item 的最大context 數量。

參數的 `flags` 可用來設置 workqueue 的特性，在 [Concurrency Managed Workqueue (cmwq): flags](https://www.kernel.org/doc/html/latest/core-api/workqueue.html#flags) 一節有更詳細的解釋。

#### `WQ_UNBOUND`
表示被加入到該 queue 中的 work item 是由不指定 CPU 的特殊 worker-pools 所服務的。這種情況下 kernel 不會對該 workqueue 提供 concurrent 管理。worker-pools 會嘗試盡快開始執行 work item。

#### `WQ_FREEZABLE`
Linux 的電源管理機制會讓系統進入 suspend 狀態，此情況下 userspace 的 thread 和部分的 kernel thread 會被凍結。設置此 flag 則 workqueue 也會在 suspend 時暫停: kernel 會處理完當前的所有 work 且不接受新的 work，直到解凍。

#### `WQ_MEM_RECLAIM`

CMWQ 中存在所謂 **rescuer thread** 的機制。一般來說，為了保證每個 workqueue 都能順利有進度，這依賴於 kernel 可以建立更多的 worker thread。然而在記憶體資源吃緊的狀態下，worker thread 並不一定可以成功建立。這可能導致 deadlock 的狀況。

舉例來說，workqueue A 目前正被處理，但其依賴於 workqueue B 釋放記憶體資源。結果 workqueue B 此時卻因為記憶體不足導致建不出 worker thread 來處理，自然沒機會將記憶體釋放出去。

當設置 `WQ_MEM_RECLAIM`，則 kernel 將預先考慮此狀況讓建立的 workqueue 可以無視記憶體的壓力來建立 **rescuer thread** 以避免 deadlock 的發生。

#### `WQ_HIGHPRI`

使用此 flag 的 workqueue 將使用高優先級的 worker-pools 去處理(nice value 較低的 worker thread)。

注意到高優先級與普通的 worker-pools 之間是不能互通的。

#### `WQ_CPU_INTENSIVE`

被設置此 flag 的 workqueue 中的 work 與其是特別消耗 CPU 資源的，則 kernel 將不會將其造成的壓力計算到 concurrent level。具體的影響是，這些 workqueue 中的 work item 將歸 scheduler 管理，相關的 work item 不會阻止同一 worker-pools 中的其他 work item 被執行。

#### `WQ_SYSFS`

使 workqueue 信息在 sysfs 中可以被 user 查看到。

#### `WQ_POWER_EFFICIENT`

per-CPU 的 workqueue 由於 cache locality，往往會表現出更好的性能。這也使 scheduler 無法選擇 CPU 來執行 worker thread。然而這也不是那麼完美的，其中一個不幸的副作用是造成功耗的增加。

在省電的考量上，如果特定 CPU 上沒有任何任務要執行，則 scheduler 會認為該 CPU 是 idle，並嘗試維持 idle 狀態。然而，例如在 idle CPU 上的 interrupt handler 可能在其對應的 workqueue 上添加新的 work item，進而強制 scheduler 打破 idle 狀態。換句話說，系統在省電的選擇就受限於 workqueue 而可能無法選擇最佳選項。

針對此狀況，標有 `WQ_POWER_EFFICIENT` 的 workqueue 預設是綁定 CPU 的，但如果指定了 `workqueue.power_efficient` 這個 kernel 參數，可以讓該 workqueue 變為未綁定。則可以一定程度的應付上述問題。

:::warning
`WQ_SYSFS` 和 `WQ_POWER_EFFICIENT` 的詳細解釋目前在 [Concurrency Managed Workqueue (cmwq)](https://www.kernel.org/doc/html/latest/core-api/workqueue.html) 是沒有的
:::
    
```cpp
__printf(1, 4)
struct workqueue_struct *alloc_workqueue(const char *fmt,
					 unsigned int flags,
					 int max_active, ...)
{
	size_t tbl_size = 0;
	va_list args;
	struct workqueue_struct *wq;
	struct pool_workqueue *pwq;

	/*
	 * Unbound && max_active == 1 used to imply ordered, which is no
	 * longer the case on NUMA machines due to per-node pools.  While
	 * alloc_ordered_workqueue() is the right way to create an ordered
	 * workqueue, keep the previous behavior to avoid subtle breakages
	 * on NUMA.
	 */
	if ((flags & WQ_UNBOUND) && max_active == 1)
		flags |= __WQ_ORDERED;

	/* see the comment above the definition of WQ_POWER_EFFICIENT */
	if ((flags & WQ_POWER_EFFICIENT) && wq_power_efficient)
		flags |= WQ_UNBOUND;

```

首先，根據參數，`alloc_workqueue` 會加上一些額外的 flag。`__WQ_ORDERED` 表示 work item 會嚴格的按照在 queue 中的順序運行。這個 flag 是通過設置 `WQ_UNBOUND` 和 `max_active` 為 1 來隱式添加的。

此外，標有 `WQ_POWER_EFFICIENT` 的 workqueue 如果指定了 `workqueue.power_efficient` 會變成 unbound，這在前面有先說明。


```cpp
	/* allocate wq and format name */
	if (flags & WQ_UNBOUND)
		tbl_size = nr_node_ids * sizeof(wq->numa_pwq_tbl[0]);

	wq = kzalloc(sizeof(*wq) + tbl_size, GFP_KERNEL);
	if (!wq)
		return NULL;

	if (flags & WQ_UNBOUND) {
		wq->unbound_attrs = alloc_workqueue_attrs();
		if (!wq->unbound_attrs)
			goto err_free_wq;
	}
```

配置 `workqueue_struct` 所需要的空間。對於 unbounded wq，worker-pools 需要額外的空間來描述，此外也需要空間來記錄自定義的屬性。

```cpp
	va_start(args, max_active);
	vsnprintf(wq->name, sizeof(wq->name), fmt, args);
	va_end(args);
```

通過參數設置 wq 的名稱的字串。

```cpp
	max_active = max_active ?: WQ_DFL_ACTIVE;
	max_active = wq_clamp_max_active(max_active, flags, wq->name);
```

參數的 `max_active` 如果給 0 的話，則系統會使用預設的 [`WQ_DFL_ACTIVE`](https://elixir.bootlin.com/linux/v6.4/source/include/linux/workqueue.h#L347)。然後再由 `wq_clamp_max_active` 校正 `max_active` 至合理範圍。

```cpp
	if (alloc_and_link_pwqs(wq) < 0)
		goto err_unreg_lockdep;
```

最後部分大多與底下資料結構的初始化相關，暫且先略過大部分。但需要留意 [`alloc_and_link_pwqs`](#`alloc_and_link_pwqs`) 這個地方會建立 `workqueue_struct`、`pool_workqueue`、`worker_pool` 之間的關聯。



### `alloc_and_link_pwqs`

[`alloc_and_link_pwqs`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L4336) 把將在 `alloc_workqueue` 配置的 `workqueue_struct` 與 `pool_workqueue`、`worker_pool` 的關係連結在一起。

```cpp
static int alloc_and_link_pwqs(struct workqueue_struct *wq)
{
	bool highpri = wq->flags & WQ_HIGHPRI;
	int cpu, ret;

	if (!(wq->flags & WQ_UNBOUND)) {
		wq->cpu_pwqs = alloc_percpu(struct pool_workqueue);
		if (!wq->cpu_pwqs)
			return -ENOMEM;

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

			mutex_lock(&wq->mutex);
			link_pwq(pwq);
			mutex_unlock(&wq->mutex);
		}
		return 0;
	}
```

如果是 bounded waitqueue，workqueue 下會配置與 CPU 數量等量的 `pool_workqueue`，維護在 `wq->cpu_pwqs` 中。且每個 `pool_workqueue` 會各自對應到前面在 `workqueue_init_early` 建立的 `worker_pool`，根據 flag 可能是普通或高優先級的 workqueue。

![](https://hackmd.io/_uploads/SkQWw9FRn.png)


```cpp
	cpus_read_lock();
	if (wq->flags & __WQ_ORDERED) {
		ret = apply_workqueue_attrs(wq, ordered_wq_attrs[highpri]);
		/* there should only be single pwq for ordering guarantee */
		WARN(!ret && (wq->pwqs.next != &wq->dfl_pwq->pwqs_node ||
			      wq->pwqs.prev != &wq->dfl_pwq->pwqs_node),
		     "ordering guarantee broken for workqueue %s\n", wq->name);
	} else {
		ret = apply_workqueue_attrs(wq, unbound_std_wq_attrs[highpri]);
	}
	cpus_read_unlock();

	return ret;
}
```

unbouded workqueue 則是根據是否是需要嚴格按照 work item 在 queue 中的順序運行的 `__WQ_ORDERED`。若是，則使用 `ordered_wq_attrs`，否的或會使用 `unbound_std_wq_attrs`。對應的 worker pool 則是透過 attribute 去取得([`get_unbound_pool`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L3763))。

### worker thread

```cpp
/**
 * worker_thread - the worker thread function
 * @__worker: self
 *
 * The worker thread function.  All workers belong to a worker_pool -
 * either a per-cpu one or dynamic unbound one.  These workers process all
 * work items regardless of their specific target workqueue.  The only
 * exception is work items which belong to workqueues with a rescuer which
 * will be explained in rescuer_thread().
 *
 * Return: 0
 */
static int worker_thread(void *__worker)
{
	struct worker *worker = __worker;
	struct worker_pool *pool = worker->pool;

	/* tell the scheduler that this is a workqueue worker */
	set_pf_worker(true);
```

在 CMWQ 中，每個 worker-pool 中各自會建立自己的 [`worker_thread`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L2497)。需要知道 workqueue 只是對使用者端的抽象，對後端的 `worker_thread` 來說，無論是屬於 bounded 還是 unbounded 的 worker-pools，`worker_thread` 並不會區分 workqueue(除了需要 rescuer thread 的 workqueue)，對其來說就只是有個 work list 由它負責處理與完成而已。

[`set_pf_worker`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L2475) 對當前的 task 標注 [`PF_WQ_WORKER`](https://elixir.bootlin.com/linux/v6.4/source/include/linux/sched.h#L1729) 這個 flag，這讓 scheduler 可以和 CMWQ 機制掛勾並為其提供特殊的處理。

```cpp
woke_up:
	raw_spin_lock_irq(&pool->lock);

	/* am I supposed to die? */
	if (unlikely(worker->flags & WORKER_DIE)) {
		raw_spin_unlock_irq(&pool->lock);
		set_pf_worker(false);

		set_task_comm(worker->task, "kworker/dying");
		ida_free(&pool->worker_ida, worker->id);
		worker_detach_from_pool(worker);
		WARN_ON_ONCE(!list_empty(&worker->entry));
		kfree(worker);
		return 0;
	}

	worker_leave_idle(worker);
```

`woke_up` tag 下的流程是 worker thread 重新啟動時需進行的，包含 `WORKER_DIE` 被設置時需要將其從 pool 中移除，並回收此 worker。

若非為了結束該 worker 而喚醒之，那麼 [`worker_leave_idle`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L1837) 將自己的狀態設為非 idle，並把自己從對應 worker_pool 的 idle list 中移除出來。

```cpp
recheck:
	/* no more worker necessary? */
	if (!need_more_worker(pool))
		goto sleep;

	/* do we need to manage? */
	if (unlikely(!may_start_working(pool)) && manage_workers(worker))
		goto recheck;

	/*
	 * ->scheduled list can only be filled while a worker is
	 * preparing to process a work or actually processing it.
	 * Make sure nobody diddled with it while I was sleeping.
	 */
	WARN_ON_ONCE(!list_empty(&worker->scheduled));
```

`recheck` 會進行一系列的檢查以動態調整 worker thread。

[`need_more_worker`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L806) 根據待解的 work 數量和已經 running 的 worker 數量(`nr_running`)，決定此 worker thread 是否也需被啟動，又或者返回 sleep 狀態。


如果沒有 idle worker ([`!may_start_working()`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L812))，會判斷此 worker thread 是否需要作為 manager 以 [`manage_workers`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L2268) 來建立 idle thread。

被喚醒的 worker thread 之 `scheduled` list 不應該有任何的 work，若發生則這裡 `WARN_ON_ONCE` 對此異常做出警告。

```cpp
	/*
	 * Finish PREP stage.  We're guaranteed to have at least one idle
	 * worker or that someone else has already assumed the manager
	 * role.  This is where @worker starts participating in concurrency
	 * management if applicable and concurrency management is restored
	 * after being rebound.  See rebind_workers() for details.
	 */
	worker_clr_flags(worker, WORKER_PREP | WORKER_REBOUND);
```


[`worker_clr_flags`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L1006) 清除 `WORKER_PREP` 和 `WORKER_REBOUND` 兩個 flag。`WORKER_PREP` 表示 worker thread 是準備可以執行 work 的狀態，清除之意外這 worker thread 將開始挑選 work 來處理。`WORKER_REBOUND` 則與 [`rebind_workers`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L5225) 有關。

如果清除這些 flag 使得 worker 不再具有 `WORKER_NOT_RUNNING`，同時也將 pool 的 `nr_running` + 1。

:::info
```cpp
WORKER_NOT_RUNNING = WORKER_PREP | 
    WORKER_CPU_INTENSIVE | 
    WORKER_UNBOUND | 
    WORKER_REBOUND;
```
:::

```cpp
	do {
		struct work_struct *work =
			list_first_entry(&pool->worklist,
					 struct work_struct, entry);

		pool->watchdog_ts = jiffies;

		if (likely(!(*work_data_bits(work) & WORK_STRUCT_LINKED))) {
			/* optimization path, not strictly necessary */
			process_one_work(worker, work);
			if (unlikely(!list_empty(&worker->scheduled)))
				process_scheduled_works(worker);
		} else {
			move_linked_works(work, &worker->scheduled, NULL);
			process_scheduled_works(worker);
		}
	} while (keep_working(pool));

	worker_set_flags(worker, WORKER_PREP);
```

這裡是 worker thread 挑選 work item 以進行的流程。具體所為是從對應的 worker_pool 之 linked list `pool->worklist` 上一個個取得 `work` 來處裡，直到所有 work 都被完成。( [`!keep_working()`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L818)，但每個 pool 會留下至少一個 worker thread)。


對於每個 `work_struct`，如果該其 data bitmask 未設置 `WORK_STRUCT_LINKED`，這表示 work 與 work 之間沒有依賴關係，可以直接以 `process_one_work` 來執行該 work 所想執行的內容即可。反之則表示 work 之間存在依賴關係，[`move_linked_works`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L1089) 會將一系列標註 `WORK_STRUCT_LINKED` 的 work 移轉到另一個 list `worker->scheduled`，並按次序執行再用 `process_one_work` 去完成該 list 上的 work。換言之，work 間沒有依賴關係的情形就用優化的 code path 來進行即可。

返回 sleep 狀態情形下，用 [`worker_set_flags`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L981) 去設置回 `WORKER_PREP`。

```cpp
sleep:
	/*
	 * pool->lock is held and there's no work to process and no need to
	 * manage, sleep.  Workers are woken up only while holding
	 * pool->lock or from local cpu, so setting the current state
	 * before releasing pool->lock is enough to prevent losing any
	 * event.
	 */
	worker_enter_idle(worker);
	__set_current_state(TASK_IDLE);
	raw_spin_unlock_irq(&pool->lock);
	schedule();
	goto woke_up;
}
```

與 [`worker_leave_idle`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L1837)  相對，[`worker_enter_idle`](https://elixir.bootlin.com/linux/v6.4/source/kernel/workqueue.c#L1804) 根據 worker 回到 idle 的情況設置對應的狀態，並安排下次被喚醒的時間。

### 初始化

workqueue_init -> create_worker -> worker_thread

### 加入新任務至 workqueue

`struct workqueue_struct` 

## Reference
* [Concurrency Managed Workqueue (cmwq)](https://www.kernel.org/doc/html/latest/core-api/workqueue.html)
* [Concurrency Managed Workqueue之（二）：CMWQ概述](http://www.wowotech.net/irq_subsystem/cmwq-intro.html)