# 2024q1 Homework6 (integration)
contributed by < `vax-r` >
## 自我檢查清單
- [X] 實體電腦運行 GNU/Linux
- [X] 閱讀〈Linux 核心模組運作原理〉並對照 Linux 核心原始程式碼 (v6.1+)，解釋 insmod 後，Linux 核心模組的符號 (symbol) 如何被 Linux 核心找到 (使用 List API)、MODULE_LICENSE 巨集指定的授權條款又對核心有什麼影響 (GPL 與否對於可用的符號列表有關)，以及藉由 strace 追蹤 Linux 核心的掛載，涉及哪些些系統呼叫和子系統？
Ans : 在 Linux kernel v6.8.5 當中，巨集 `MODULE_LICENSE()` 在 [/include/linux/module.h](https://elixir.bootlin.com/linux/latest/source/include/linux/module.h) 定義如下
```c
#define MODULE_LICENSE(_license) MODULE_FILE MODULE_INFO(license, _license)
```
免費的模組標記是 `GPL, GPL v2, GPL and additional rights, Dual BSD/GPL, Dual MIT/GPL, Dual MPL/GPL` ，非免費的是 `Proprietary` 。 `GPL, GPL v2` 這兩個標記是授權符號都代表模組是註冊在 GPL v2 底下，有這兩個版本是由於歷史因素，主要是要讓 `Proprietary` 標記能使用並避免讓非免費模組使用 `EXPORT_SYMBOL_GPL` 已經使用的 symbol 。這些標記存在的理由有以下數個
1. 使 `modinfo` 可以顯示 licencse info 讓使用者確認他們使用的模組是免費的。
2. 使開發者社群可以忽略模組顯示的錯誤回報。
在編譯完核心模組並利用 `insmod` 命令將核心模組載入的過程中，透過 `strace` 命令觀察呼叫的系統呼叫，注意到會進行 `finit_module` 系統呼叫，並在 `idempotent_init_module` 當中的 `init_module_from_file` 呼叫 `load_module` ，值得注意的是 `load_module` 當中的 `simplify_symbols` 會透過一系列操作與函式呼叫將核心模組的 symbols 給載入，當中呼叫 `resolve_symbol_wait` 再呼叫 `resolve_symbol` 再呼叫 `find_symbol` ，特別注意到在 `find_symbol` 當中有用到 List API `list_for_each_entry_rcu` 來設定 module 當中的為雙向鍵結串列，也可以注意到搜尋 symbol 的函式 `find_exported_symbol_in_section` 利用到一個特殊的搜尋方式 `bsearch()` ，需進一步探討實作，還有尋找 symbol 的過程有用到 mutex lock ，許多 lock 機制的使用方式或許有改進空間。並且發現 `insmod` 大量使用了 `mmap` 系統呼叫，重複的進行 `openat(), read(), newfstatat(), mmap(), close()` 一系列系統呼叫。
- [X] 閱讀《The Linux Kernel Module Programming Guide》(LKMPG) 並解釋 simrupt 程式碼裡頭的 mutex lock 的使用方式，並探討能否改寫為 lock-free;
Ans : simrupt 當中在 `simrupt_read` 時會使用到 mutex lock ，當中嘗試獲取 mutex lock 的方式是透過 `mutex_lock_interruptible()` ，依照 [kernel/locking/mutex.c](https://elixir.bootlin.com/linux/latest/source/kernel/locking/mutex.c#L982) 當中的實作說明，若嘗試獲取該 lock 的行程是睡眠狀態且有訊號傳入導致中斷，則該函式會回傳 `-EINTR` 且該 lock 不會被該行程取得。除此之外就和普通的 `mutex_lock()` 一樣，成功就回傳 0。在 Critical section 當中做的事是將 kfifo buffer 的內容傳至 userspace ，若 `rx_fifo` 的內容為空則將該行程放入 wait queue 當中。值得注意的是放入 wait queue 的行程會進入睡眠狀態，要喚醒需要透過 `wake_up_interruptible()` 函式。
:::warning
`mutex_lock()` 和 `mutex_lock_interruptible()` 實際行為差異在哪裡？
:::
- [ ] 探討 Timsort, Pattern Defeating Quicksort (pdqsort) 及 Linux 核心 lib/sort.c 在排序過程中的平均比較次數，並提供對應的數學證明
- [ ] 研讀 CMWQ (Concurrency Managed Workqueue) 文件，對照 simrupt 專案的執行表現，留意到 worker-pools 類型可指定 "Bound" 來分配及限制特定 worker 執行於指定的 CPU，Linux 核心如何做到？CMWQ 關聯的 worker thread 又如何與 CPU 排程器互動？
- [ ] 解釋 xoroshiro128+ 的原理 (對照〈Scrambled Linear Pseudorandom Number Generators〉論文)，並利用 ksort 提供的 xoro 核心模組，比較 Linux 核心內建的 /dev/random 及 /dev/urandom 的速度，說明 xoroshiro128+ 是否有速度的優勢？其弱點又是什麼？
---
## 開發紀錄
### simrupt 解析
simrupt 是一個 character device ，同時建立一個對應的 device class ，關於 `struct class` 的完整定義可以參見 [include/linux/device/class.h](https://elixir.bootlin.com/linux/latest/source/include/linux/device/class.h) ，閱讀註解可以知道 `struct class` 是將裝置的實作進行高階抽象化，使得使用者層級的行程可以根據裝置行為和裝置互動，而非裝置的底層實作與如何連接。
`class_create()` 函式會建立對應的 pointer to `struct class` 之後我們呼叫 `device_create()` 的時候可以用。
此核心模組有利用到 circular buffer ，詳細敘述可以參見 [Circular Buffers](https://www.kernel.org/doc/Documentation/core-api/circular-buffers.rst) ，此 buffer 利用兩個整數 `head, tail` 分別代表頭尾，當 `head == tail` 代表 buffer 是空的 ，如果 `head == tail - 1` 則代表 buffer 是滿的， `head` 是讓 producer 插入元素用的， `tail` 則是讓消費者取得元素用的。特別的是這個 buffer 可以允許放入不同大小的元素，但實作者需要小心 buffer 空間出現 segmentation 的問題。
### `vmalloc()` v.s. `kmalloc()`
`kmalloc()` 的實作函式在 [include/linux/slab.h](https://elixir.bootlin.com/linux/latest/source/include/linux/slab.h) 當中，會接受兩個參數分別是 `size` 和 `flags` ，通常用來 kernel space 當中配置小於 page size 的記憶體空間，特別注意配置的記憶體區段至少會和 `ARCH_KMALLOC_MINALIGN` 個 bytes 對齊，若 `size` 是 2 的冪，則至少和 `size` 對齊。
`vmalloc()` 的實作函式在 [mm/vmalloc.c](https://elixir.bootlin.com/linux/latest/source/mm/vmalloc.c) 在 kernel space 配置大量（比 page size 大）連續虛擬記憶體空間。
### CMWQ
Linux kernel 當中利用 workqueue API 來處理非同步行程的執行狀態， work item 代表要執行的函式，它們會被放入 workqueue 當中並且利用執行緒來處理，每個獨立的執行緒代表一個 asynchronous execution context ， queue 稱為 workqueue 而 thread 稱為 worker 。當 workqueue 當中有 work items 時 worker 會一個接一個地處理 work items ，若 workqueue 為空則 worker 們會變為 idle 。
#### Why CMWQ?
原本的 workqueue 實作當中包含 multi-threaded (MT) wq 和 single-threaded wq 而 MT wq 當中每個 CPU 就有一個 worker ， ST wq 當中則是整個系統只有一個 worker 。雖然 MT wq 使用大量資源，但它提供的並行程度非常有限， MT wq 每個 CPU 只能提供一個 execution context 而 ST wq 則是整個系統只能提供一個 execution context ，造成 work items 大量競爭，甚至在 async 或 fscache 當中使用者得實作自己的 thread pool 。
CMWQ 則提供了以下幾個更好的優點
* 保有和原先 wq API 設計的相容性
* 每個 CPU 的 worker pool 都統一，並被所有 wq 共享，如此一來提供相同並行程度的同時也能減少資源的浪費
* 規範 worker pool 和並行的程度，使得 API 使用者不需要考慮這些細節
#### 設計方法
設計代表 work item 的結構體，當中有一個指標指向要被執行的函式，並把該 work item 放入 wq 當中， work item 可以利用 thread 或 BH (softirq) context 執行。
CMWQ 將 subsystem 、 driver 把 work items 放入 wq 的實作和後端處理 worker-pools 和如何處理 work items 的機制分開來。
threaded workqueues 利用 worker-pools 當中的 workers 呼叫 work items 來執行函式。 worker-pools 分為兩種，一種是給 normal work items 而另一個是給較高 priority 的 work items 。注意到 worker pool 的數量是可變的。 BH workqueue 則更簡單因為只能有一個 concurrent execution context ，所以我們不需要處理並行，每個 CPU 的 BH worker pool 只包含一個 pseudo worker ，代表對應的 BH execution context ， BH workqueue 可以被當成 softirq 的使用介面。
我們透過把 CPU 的 worker-pool 掛載到 scheduler 上來實現並行程度的管理。每個 worker 被 wake up 或 sleep 時都會通知 worker-pool ，使得它可以追蹤當前可以使用的 worker 。另外 work items 不應該佔據 CPU 過久，只要 CPU 當前有多個 workers 正在執行，它就不會開始執行新的 work ，但當某個運行中的 worker 進入睡眠狀態，它馬上分配一個新的 worker ，如此一來 CPU 才不會在還有 pending works 的時候進入 idle 狀態。
由於對於 kernel thread 來說， idle worker 只會佔據記憶體，所以 cmwq 的實作在把 idle worker 砍除掉之前會先保持一段時間。
### Data Generation and Propogation
在 `timer_handler()` 當中會透過 `local_irq_disable()` 將中斷取消，再透過 `process_data()` 產生要傳送給 user space 的資料，其中 `fast_buf_put()` 扮演生產者的角色，每次取得 `update_simrupt_data()` 所產生的數字並將該字元放入 `fast_buf` 這個 circular buffer 當中。之後透過 `tasklet_schedule()` 把 `simrupt_tasklet_func()` 加入排程器當中。
:::warning
**疑問**
* softirq 和 hardirq 是什麼？把 CPU 的中斷取消用意何在？
* `smp_wmb()` 是防止寫入指令重排，對應到的 memory model 是何者？更新 `head` 數值的指令不是也不能被重排嗎？為何擺在 `smp_wmb()` 後面？
:::
simrupt 的 CMWQ 的 work item 是 `simrupt_work_func()` ，其中有一個迴圈透過 `fast_buf_get` 作為消費者讀取 `fast_buf` 當中的資料。此處我認為讀取 `ring->tail` 的指令也應該加上 `READ_ONCE()`
```diff
static int fast_buf_get(void)
{
struct circ_buf *ring = &fast_buf;
/* prevent the compiler from merging or refetching accesses for tail */
- unsigned long head = READ_ONCE(ring->head), tail = ring->tail;
+ unsigned long head = READ_ONCE(ring->head), tail = READ_ONCE(ring->tail);
```
`simrupt_tasklet_func()` 負責把 work item 放入 CMWQ 當中。
:::warning
tasklet 是什麼？
> 參閱 LKMPG
:::
`produce_data()` 負責把資料放入 kfifo buffer `rx_fifo` 當中，此時若呼叫 `simrupt_read()` 就可以取得 `rx_fifo` 當中的資料。如果把 `val` 改成某個定值，而非從 `fast_buf` 當中取得，核心模組會卡住，但直接把 `fast_buf` 當中數值固定就沒問題，應探討 `fast_buf` 為何是必要的存在。
如果把 `simrupt_read()` 的迴圈條件改成無論如何都只做一次，也就是每次呼叫 `simrupt_read()` 都只會將當前 kfifo buffer 當中的資料傳一遍給使用者層級，則不會出現無限 read 的情況。同時也代表資料產生到 `fast_buf` 再從 `fast_buf` 搬移到 kfifo buffer 的過程是不斷在核心空間中進行的，不管我們是否呼叫 read 。
### timer_list
Linux Kernel 提供兩種類型的計時器，分別是 dynamic timer 和 interval timers ，前者用在 kernel space 當中而後者用在 user space 當中。 `struct timer_list` 則是運用在 kernel space 當中的 dynamic timers ( 參見 [/include/linux/timer_types.h](https://elixir.bootlin.com/linux/latest/source/include/linux/timer_types.h) )。
在 [/kernel/time/timer.c](https://elixir.bootlin.com/linux/latest/source/kernel/time/timer.c) 當中定義以下函式，注意到每個 cpu 都有自己的 timer 。
```c
void __init init_timers(void)
{
init_timer_cpus();
posix_cputimers_init_work();
open_softirq(TIMER_SOFTIRQ, run_timer_softirq);
}
```
我們可以利用 `$ sudo cat /proc/timer_list` 來觀察 CPU timer_list 資訊。
### Interrupt Handling
[Interrupts and Interrupt Handling](https://0xax.gitbooks.io/linux-insides/content/Interrupts/index.html)
在 Linux Kernel 當中的中斷大致上有數種特性，例如以下兩種
* interrupt handler must execute quickly
* sometimes an interrupt handler must do a large amount of work
有些情況是兩種特性都存在的，但它們看起來互相衝突，該如何解決呢？ Linux kernel 採用一種方式叫 deferred interrupts ，將中斷處理延遲，並且將上述兩種特性分別稱為 top half 和 bottom half 。有時中斷處理會需要很大的工作量，但在一般的 interrupt handler 當中，中斷機制會被暫時取消，若在這時候做很大的工作量整個系統會全部都在等待它完成，為了避免這件事發生才把中斷處理拆成 top half 和 bottom half ，在 top half 只處理一點重要的事然後就將 bottom half 交由之後的 context 處理，在處理 bottom half 的時候是不允許中斷發生的。在 Linux kernel 當中有三種 deferred interrupts 的機制
* softirqs
* tasklets
* workqueues
#### softirqs
透過稱為 `ksoftirqd` 的 kernel thread 來達成，每個 CPU 都有一個這樣的 thread ，可以透過以下命令觀察
```shell
$ systemd-cgls -k | grep ksoft
├─ 15 [ksoftirqd/0]
├─ 23 [ksoftirqd/1]
├─ 29 [ksoftirqd/2]
├─ 35 [ksoftirqd/3]
├─ 41 [ksoftirqd/4]
├─ 47 [ksoftirqd/5]
├─ 53 [ksoftirqd/6]
├─ 59 [ksoftirqd/7]
│ │ └─6999 grep --color=auto ksoft
```
我們可以在 [/kernel/softirq.c](https://elixir.bootlin.com/linux/latest/source/kernel/softirq.c) 看到以下定義，分別有 `softirq_vec, ksoftirqd, softirq_to_name` ，每個 CPU 都有自己的 `ksoftirqd` kernel thread ，而這些 kernel thread 也有各自的 `softirq_vec` ，分別對應到 `softirq_to_name` 所對應的種類。
```c
static struct softirq_action softirq_vec[NR_SOFTIRQS] __cacheline_aligned_in_smp;
DEFINE_PER_CPU(struct task_struct *, ksoftirqd);
const char * const softirq_to_name[NR_SOFTIRQS] = {
"HI", "TIMER", "NET_TX", "NET_RX", "BLOCK", "IRQ_POLL",
"TASKLET", "SCHED", "HRTIMER", "RCU"
};
```
利用以下命令可以觀察
```shell
$ sudo cat /proc/softirqs
CPU0 CPU1 CPU2 CPU3 CPU4 CPU5 CPU6 CPU7
HI: 63 0 1 0 0 11 0 5
TIMER: 91569 81926 83934 103131 74011 127987 113794 125569
NET_TX: 4 2 0 0 0 3 0 33
NET_RX: 4400 2811 4047 3683 2501 5472 2422 152954
BLOCK: 351 282 679 225 358 15623 39399 486
IRQ_POLL: 0 0 0 0 0 0 0 0
TASKLET: 173719 0 57 524 20 42 0 1908
SCHED: 240706 199132 152716 184508 189760 380257 337690 202108
HRTIMER: 0 0 0 0 0 0 0 1
RCU: 142200 142248 117051 133957 138481 245684 225400 148630
```
被延遲的 interrupt 會被放到對應的欄位當中，透過 `raise_softirq` 來觸發， `wakeup_softirqd` 則是會觸發當前 CPU 的 `ksoftirqd` kernel thread 。
#### tasklets
tasklets 在 Linux kernel 當中的實作位在 [/include/linux/interrupt.h](https://elixir.bootlin.com/linux/latest/source/include/linux/interrupt.h)
```c
struct tasklet_struct
{
struct tasklet_struct *next;
unsigned long state;
atomic_t count;
bool use_callback;
union {
void (*func)(unsigned long data);
void (*callback)(struct tasklet_struct *t);
};
unsigned long data;
};
```
它是實作在 softirq 上，另一種延遲中斷處理的機制，它依賴以下兩種 softirqs
* `TASKLET_SOFTIRQ`
* `HI_SOFTIRQ`
同一種類型的 tasklets 不能同時在多個處理器上運作，從以上 tasklet_struct 的定義來看，可以剖析它的實作包括
* 指向位於 scheduling queue 當中下一個 tasklet 的指標
* tasklet 的狀態
* 該 tasklet 的 callback 函式
* callback 函式的參數
Linux kernel 利用以下三個函式來標示 tasklet 為 ready to run
* `tasklet_schedule()`
* `tasklet_hi_schedule()`
* `tasklet_hi_schedule_first()`
這三個函式實作相近，差別在優先權，第一個函式所標註的 tasklet 優先權最低。
#### workqueues
workqueue 和 tasklet 的概念類似，但依舊有差別， tasklet 透過 software interrupt context 來執行，而 worqueue 當中的 work items 則是透過 kernel process ，這代表 work item 的執行不像 tasklet 一樣是 atomic 的 (換言之，整個 tasklet 的函式只能執行在最初被分配到的 CPU 上)。
Kernel 會建立稱為 `worker threads` 的 kernel threads 來處理 work items ，我們可以透過以下命令來觀察這些 kernel threads 。
```shell
$ systemd-cgls -k | grep kworker
├─ 7 [kworker/0:0-events]
├─ 8 [kworker/0:0H-events_highpri]
├─ 25 [kworker/1:0H-events_highpri]
├─ 31 [kworker/2:0H-events_highpri]
├─ 37 [kworker/3:0H-events_highpri]
├─ 43 [kworker/4:0H-events_highpri]
├─ 55 [kworker/6:0H-events_highpri]
├─ 60 [kworker/7:0-events]
├─ 61 [kworker/7:0H-kblockd]
├─ 72 [kworker/3:1-events]
├─ 85 [kworker/3:1H-kblockd]
├─ 88 [kworker/6:1-events]
├─ 90 [kworker/1:1-mm_percpu_wq]
├─ 91 [kworker/2:1-events]
├─ 92 [kworker/5:1-mm_percpu_wq]
├─ 113 [kworker/4:1H-kblockd]
├─ 121 [kworker/u17:0-i915_flip]
├─ 128 [kworker/5:1H-kblockd]
├─ 129 [kworker/7:1H-kblockd]
├─ 148 [kworker/2:1H-kblockd]
├─ 152 [kworker/0:1H-kblockd]
├─ 153 [kworker/1:1H-kblockd]
├─ 166 [kworker/6:1H-kblockd]
├─ 181 [kworker/5:2H-kblockd]
├─ 918 [kworker/0:3-cgroup_destroy]
├─1323 [kworker/6:2-events]
├─1673 [kworker/u17:1]
├─4813 [kworker/4:0-events]
├─6021 [kworker/2:0-cgroup_destroy]
├─6031 [kworker/7:1-events]
├─6086 [kworker/4:1-cgroup_destroy]
├─6123 [kworker/3:0]
├─6140 [kworker/5:0]
├─6159 [kworker/u16:1-events_power_efficient]
├─6180 [kworker/u16:3-events_freezable_power_]
├─6923 [kworker/1:2]
├─7102 [kworker/u16:2-ext4-rsv-conversion]
├─7147 [kworker/u16:4-events_unbound]
├─7191 [kworker/2:2-events]
├─7192 [kworker/u16:0]
│ │ └─7195 grep --color=auto kworker
```
`queue_work()` 函式則是可以幫我們把 work item 放置到 workqueue 當中。
## 畫棋盤
由於讀取的資料來源是從 `fast_buf` 當中取得，所以先把棋盤的格式寫入 `fast_buf` ，原本專案 ttt 的棋盤有顏色，但這裡暫時將棋盤劃為以下樣式
```shell
| | |
-------
| | |
-------
| | |
-------
| |O|
-------
```
我認為原本使用的 `fast_buf` 要配置一個 page 的虛擬記憶體空間太浪費資源，一張 4x4 的棋盤不需要這麼多空間，因此我將棋盤寫入 kfifo buffer 的流程從 generate data -> fast_buf -> kfifo -> userspace ，改為， generate data -> draw_buffer -> kfifo -> userspace ，其中 `draw_buffer` 僅配置 `N_GRIDS` 個字元大小的空間，同時每次都利用 `kfifo_in` 一口氣把當中全部字元都傳至 userspace ，但如此更改後系統會 crash ，尚需探討原因。
原來是因為原本 `simrupt_work_func()` 當中利用一個不斷重複的迴圈使得生產者可以把資料一個一個 bytes 移動到 kfifo buffer 當中，我改變後的設計是一次把一整個棋盤的資料移動到 kfifo buffer ，因此每次進入 `simrupt_work_func()` 當中只能進行一次資料搬移，接著就要讓出執行權讓 `process_data()` 當中可以再把下一個棋盤畫好。
> [commit 7c3f05a](https://github.com/vax-r/KMLdrv/commit/7c3f05a4a6a58de8ee463fdb9db7a7d363e20c9b)
## 實作 AI 演算法與對弈
首先將 MCTS 移植進入 kernel space 當中並且在 device driver 處撰寫 main loop 控制整體對弈流程，目前兩個 AI 皆採取 MCTS 演算法， main loop 設計如下
```c
while (1) {
mutex_lock(&producer_lock);
char win = check_win(table);
mutex_unlock(&producer_lock);
if (win == 'D' || win != ' ')
break;
if (turn == ai1) {
mutex_lock(&producer_lock);
int move = mcts(table, ai1);
mutex_unlock(&producer_lock);
if (move != -1) {
WRITE_ONCE(table[move], ai1);
}
} else {
mutex_lock(&producer_lock);
int move = mcts(table, ai2);
mutex_unlock(&producer_lock);
if (move != -1) {
WRITE_ONCE(table[move], ai2);
}
}
turn = turn == 'X' ? 'O' : 'X';
smp_wmb();
mutex_lock(&producer_lock);
draw_board(table);
mutex_unlock(&producer_lock);
pr_info("simrupt: [CPU#%d] scheduling tasklet\n", smp_processor_id());
tasklet_schedule(&simrupt_tasklet);
}
```
當前設計結果會使讀取對弈結果時是每次進行一次完整對弈才讓使用者層級讀取到一次棋盤結果，和預期行為不符，首先每個演算法都應該拆成一個 tasklet 才能使不同算法在不同 CPU 上執行，目前實作方式會使得同一局對弈當中的所有執行指令都在同一個 CPU 上，用 htop 可觀察到被指派的 CPU 用量會達到 100% 。
```shell
$ sudo cat /dev/simrupt
X|X|X|
-------
O| |O|
-------
| | |
-------
| | |
-------
X|X|X|O
-------
| | |
-------
|O| |
-------
| | |
-------
X|O| |
-------
X|O| |
-------
X| | |
-------
| | |
-------
```
可以觀察到每次都是該局對弈完才印出結果。
> [commit 5ebaaa2](https://github.com/vax-r/KMLdrv/commit/5ebaaa2e6d1b91b0fbe4ff1c8f2604e5cd1dd50d)
之後我將 AI.1, AI.2 各自寫成兩個 work function `ai_one_work_func(), ai_two_work_func()` 並定義一個 tasklet 用來將這兩個 work item 放入 workqueue 當中，共享變數除了對弈用的棋盤 `table` 以外還有輪到誰下棋的變數 `turn` ，並將 main loop 函式更改如下
```c
static void ai_game(void)
{
WARN_ON_ONCE(!irqs_disabled());
pr_info("simrupt: [CPU#%d] doing AI game\n", smp_processor_id());
pr_info("simrupt: [CPU#%d] scheduling tasklet\n", smp_processor_id());
tasklet_schedule(&simrupt_tasklet);
tasklet_schedule(&ai_game_tasklet);
}
```
:::danger
asynchronous 是「非同步」，不是「異步」。
參見[資訊科技詞彙翻譯](https://hackmd.io/@sysprog/it-vocabulary)
:::
如此一來可以使 kernel thread 進行對弈，但依舊存在一個問題是做 AI algorithm 的時間和畫棋盤的時間相比太長了，每次 interrupt 發生後畫棋盤的 work item 基本上都能完成，但 MCTS 無法在一次 interrupt 時間內完成，造成 `timer_handler` 被呼叫多次但舊的 MCTS work item 依舊在執行，這也造成畫棋盤的時候多次畫出重複的棋盤。我對於 tasklet 、 interrupt 處理方式都不了解，應當了解後重新設計或許能解決此問題。同時 CMWQ 能夠以<s>異步</s> 方式處理 work item ，而同一個 tasklet 會被分配到同一個 CPU 上，對於保證共享資源的 mutual exclusion 是否用 mutex lock 就足夠？ Critical Section 執行的時候若不能保證 atomic ，像 `mcts()` 這類的函式本來執行時間就長，還一直被打斷讓新的 work item 加入，會造成 CPU 用量急速飆高。 memory barrier 在這樣程式當中的作用是什麼？
> [commit 34fd7cf](https://github.com/vax-r/KMLdrv/commit/34fd7cf8f553fa5fca2cfee30913d7bcdc2e3c14)
> 由於 AI 演算法需要較長時間才能執行完畢，因此你需要週期性的任務置於 workqueue 來監控 AI 演算法，從而判定對弈的順序。 $\to$ [Async execution with workqueues](https://events.static.linuxfound.org/sites/events/files/slides/Async%20execution%20with%20wqs.pdf)
> :notes: jserv
理解 tasklet 特性後，我認為它的 atomic 特性可以滿足 AI 演算法需要較長執行時間的需求，我們並不需要每次都把各種 function 丟到 CMWQ 上，因此我改將 AI 演算法包裝成 tasklet 來進行排程，目前只進行一局對弈即停止更新 timer ，以此觀察變化。
```c
static void ai_one_tasklet_func(unsigned long __data)
{
ktime_t tv_start, tv_end;
s64 nsecs;
WARN_ON_ONCE(!in_interrupt());
WARN_ON_ONCE(!in_softirq());
tv_start = ktime_get();
int move;
WRITE_ONCE(move, mcts(table, 'O'));
if (move != -1)
WRITE_ONCE(table[move], 'O');
tv_end = ktime_get();
nsecs = (s64) ktime_to_ns(ktime_sub(tv_end, tv_start));
pr_info("simrupt: [CPU#%d] %s in_softirq: %llu usec\n", smp_processor_id(),
__func__, (unsigned long long) nsecs >> 10);
}
```
進行以上改動後 kernel thread 可以正常進行對弈，每次畫出來的棋盤也不同，但將棋盤寫到使用者空間的反應時間也變長，以下為某個棋局的輸出結果
```shell
$ sudo cat /dev/simrupt
| | |
-------
| | |
-------
| | |
-------
| | |
-------
| |O|
-------
|X| |
-------
| | |
-------
| | |
-------
O| |O|
-------
|X|X|
-------
| | |
-------
| | |
-------
O|O|O|
-------
|X|X|
-------
| | |
-------
| | |
-------
```
而畫出棋盤的時間變長並非因為畫棋盤所需時間變長，是因為每次都等待 AI 演算法完整推算出一步才畫出來，用 `sudo dmesg` 可以觀察到以下結果
```shell
[ 415.416008] simrupt: [CPU#2] ai_one_tasklet_func in_softirq: 2766783 usec
[ 417.613452] simrupt: [CPU#2] ai_two_tasklet_func in_softirq: 2145960 usec
[ 417.614730] simrupt: [CPU#2] enter timer_handler
[ 417.614741] simrupt: [CPU#2] doing AI game
[ 417.614746] simrupt: [CPU#2] scheduling tasklet
[ 417.614750] simrupt: [CPU#2] timer_handler in_irq: 9 usec
[ 417.614833] simrupt: [CPU#2] simrupt_tasklet_func in_softirq: 0 usec
[ 417.614841] simrupt: [CPU#1] simrupt_work_func
[ 418.346857] simrupt: [CPU#2] ai_one_tasklet_func in_softirq: 714876 usec
[ 418.794527] simrupt: [CPU#2] ai_two_tasklet_func in_softirq: 437181 usec
[ 418.794535] simrupt: [CPU#2] enter timer_handler
[ 418.794537] simrupt: [CPU#2] doing AI game
[ 418.794537] simrupt: [CPU#2] scheduling tasklet
[ 418.794538] simrupt: [CPU#2] timer_handler in_irq: 1 usec
[ 418.794559] simrupt: [CPU#2] simrupt_tasklet_func in_softirq: 7 usec
[ 418.794575] simrupt: [CPU#1] simrupt_work_func
[ 419.011141] simrupt: [CPU#2] ai_one_tasklet_func in_softirq: 211507 usec
[ 419.013170] simrupt: [CPU#2] ai_two_tasklet_func in_softirq: 1979 usec
[ 419.013174] simrupt: [CPU#2] enter timer_handler
[ 419.013175] simrupt: O win!!!
[ 419.013176] simrupt: [CPU#2] timer_handler in_irq: 0 usec
[ 419.013185] simrupt: [CPU#2] simrupt_tasklet_func in_softirq: 0 usec
[ 419.013225] simrupt: [CPU#7] simrupt_work_func
```
仔細觀察可以發現，凡是執行 AI 演算法的時候，該 CPU 就會耗費很長的時間在 softirq 當中，這裡尚有幾個問題要解決
* 每一回合（AI.1 , AI.2 各下一步）才畫一個棋盤，要改成每個演算法下一步棋就畫一次
* 將棋盤重新抹白讓棋局重新開始的方法與時間點
* Linux kernel 正在逐漸將 tasklet 拿掉，能否不使用 tasklet 做到上述的行為
> [commit bd6ea30](https://github.com/vax-r/KMLdrv/commit/bd6ea302e3a54ddb421e849e3c68a9f6f0e8da64)
之後我嘗試將 AI 演算法都包裝成 work item 放入 workqueue 當中執行，但這裡會遇到一個問題，該如何判斷輪到誰執行？單純用 mutex lock 是不足夠的，因為有可能某方連續兩次都搶到該 lock ，使得它可以連續下兩步，我採用一個變數 `turn` 來表示當前輪到誰下棋，若非自己則直接結束該 work item ，等待下一次 tasklet 再把 work item 放入，但這裡又有另一個問題，每個 work item 是被不同 CPU 執行，他們看到的 `turn` 有可能不相同，該如何保證跨越 CPU 之間的資料一致性呢？我認為可以使用 memory ordering 的 acquire / release 操作來保證這件事，因此我將程式碼進行以下更改
把 `turn` 從 `char` 改為 `atomic_t`
```diff
- static char turn;
+ static atomic_t turn;
```
判斷輪到何者的方式也進行更改
```diff
- if (turn != 'O')
+ int expect = 0;
+ atomic_read_acquire(&turn);
+ if (!atomic_try_cmpxchg(&turn, &expect, 1))
return;
```
以下展示一個棋局的對弈過程
```shell
$ sudo cat /dev/simrupt
| | |
-------
| | |
-------
| | |
-------
O| | |
-------
| | |
-------
| | |
-------
| | |
-------
O| | |
-------
X| | |
-------
| | |
-------
| | |
-------
O| | |
-------
X| | |
-------
| | |
-------
| | |
-------
O| | |
-------
X| | |
-------
| | |
-------
| | |
-------
O| |O|
-------
X| | |
-------
| | |
-------
| | |
-------
O| |O|
-------
X| | |
-------
| |X|
-------
| | |
-------
O| |O|
-------
X| | |
-------
| |X|
-------
| | |
-------
O| |O|
-------
X| | |
-------
| |X|
-------
| | |
-------
O|O|O|
-------
X| | |
-------
| |X|
-------
| | |
-------
O|O|O|
-------
X| | |
-------
| |X|
-------
| | |
-------
O|O|O|
-------
```
我們可以發現現在可以正確保證不同 AI 演算法的執行順序，但由於把所有任務 （ AI 演算法、畫棋盤）都包裝成 work item 放入 workqueue 當中，我無法預測 AI 演算法每一次會執行多久，所以可能在一步 AI 演算法進行當中其他 CPU 已經完成數次畫棋盤的任務，此處我認為重複畫不是問題，只要有辦法刷新顯示內容就好，接下來實作下一個 AI 演算法後要嘗試和使用者層級進行互動，讓 userspace 可以控制對弈狀態（開始、暫停、持續、結束）。
> [commit 21a4c9d](https://github.com/vax-r/KMLdrv/commit/21a4c9dc0c437814f84f0d80a8efa2e836ae18da)
:::warning
我在 `sudo dmesg` 發現以下錯誤訊息
```shell
[136195.570341] BUG: workqueue leaked lock or atomic: kworker/u16:0/0x7fffffff/15194
last function: ai_two_work_func [kmldrv]
[136195.570346] CPU: 7 PID: 15194 Comm: kworker/u16:0 Tainted: G W OE 6.5.0-27-generic #28~22.04.1-Ubuntu
[136195.570347] Hardware name: ASUSTeK Computer INC. BM6660(BM6360)/BM6660(BM6360), BIOS 0912 02/06/2012
[136195.570348] Workqueue: simruptd ai_two_work_func [kmldrv]
[136195.570351] Call Trace:
[136195.570352] <TASK>
[136195.570352] dump_stack_lvl+0x48/0x70
[136195.570354] dump_stack+0x10/0x20
[136195.570356] process_one_work+0x3fa/0x450
[136195.570358] worker_thread+0x221/0x3f0
[136195.570360] ? __pfx_worker_thread+0x10/0x10
[136195.570362] kthread+0xf2/0x120
[136195.570364] ? __pfx_kthread+0x10/0x10
[136195.570366] ret_from_fork+0x47/0x70
[136195.570368] ? __pfx_kthread+0x10/0x10
[136195.570371] ret_from_fork_asm+0x1b/0x30
[136195.570373] </TASK>
[136195.570373] BUG: scheduling while atomic: kworker/u16:0/15194/0x00000000
```
特別注意到兩個 bug ，我看不懂此處的問題何在，看來是使用 atomic variable 造成的錯誤。
* `[136195.570341] BUG: workqueue leaked lock or atomic: kworker/u16:0/0x7fffffff/15194
last function: ai_two_work_func [kmldrv]`
* `[136195.570373] BUG: scheduling while atomic: kworker/u16:0/15194/0x00000000`
:::
尚未找出上述錯誤訊息的成因，但我認為我使用 atomic variable 的方式錯誤，重新理解後我發現使用 memory barrier 搭配 `READ_ONCE(), WRITE_ONCE()` 的使用也能夠解決多核 CPU 之間資料一致性的問題，因此我改用 memory barrier 來修正，發現這些錯誤訊息能夠被消除。
此處我發現我利用 tasklet 做為 enqueue 操作的 worker ，我預期的執行順序會是 `tasklet enqueue -> work item 1 -> work item 2 -> work item 3` ，但實際上卻經常發生 `tasklet enqueue -> tasklet enqueue -> work item 1 -> tasklet enqueue` 這樣的執行順序，也就是 tasklet 執行的過於頻繁，再來我不知道使用 `in_softirq()` 等巨集的用意何在，何時應該取消 interrupt 何時不應該？
即使利用 `turn` 變數決定當前是誰該下棋，依舊存在一個問題，當 `turn` 判定當前是 AI.1 可以下棋時， workqueue 當中可能存在多個 AI.1 的 work item ，若此時有數個 kernel thread 將他們取出，則他們會同時進行下棋，有可能使 AI.1 同時可以進行數次選擇（但高機率在同樣的棋盤狀態數個 AI.1 都選擇下在同一格，所以沒看出這個問題）。
為了解決上述問題並確保對弈流程，我進行幾個變更，首先是加入 `finish` 變數判定當前是否有 AI 演算法正在執行，若無才透過 `turn` 變數判定要將哪一個 AI 演算法 enqueue 到 CMWQ 當中。
```diff
+ READ_ONCE(finish);
+ READ_ONCE(turn);
+ smp_rmb();
+
+ if (finish && turn == 'O') {
+ WRITE_ONCE(finish, 0);
+ smp_wmb();
queue_work(simrupt_workqueue, &ai_one_work);
+ } else if (finish && turn == 'X') {
+ WRITE_ONCE(finish, 0);
+ smp_wmb();
queue_work(simrupt_workqueue, &ai_two_work);
+ }
```
由於 CMWQ 是一個 FIFO 資料結構，我可以確定先被 enqueue 的 work item 會被先取出執行，但若稍後有其他 AI 演算法被取出執行，由於我無法預測兩者的執行時間，我不能確認是先被取出的那個 AI 演算法會先完成，因此只有當目前的 AI 演算法被執行完，我才允許下一個 AI 演算法的 work item 被放入 CMWQ 當中，藉此確保執行順序。
並且相較在 work item 的 callback function 當中判定是否能執行，由於我在 enqueue 上的順序就確保了執行順序， callback function 當中就沒有判斷的必要，只要在函式執行結束改寫 `turn` 和 `finish` 的值使下一次被 enqueue 的 work item 是不同的 AI 演算法即可。
以 `ai_one_work_func()` 為例
```diff
static void ai_one_work_func(struct work_struct *w)
{
ktime_t tv_start, tv_end;
s64 nsecs;
int cpu;
WARN_ON_ONCE(in_softirq());
WARN_ON_ONCE(in_interrupt());
- READ_ONCE(turn);
- smp_rmb();
- if (turn == 'X')
- return;
...
WRITE_ONCE(turn, 'X');
+ WRITE_ONCE(finish, 1);
smp_wmb();
mutex_unlock(&producer_lock);
```
在 AI 演算法 work function 的頭尾打印訊息，執行之後印出觀察結果如下
```shell
$ sudo dmesg | grep mark
[59263.105190] simrupt: ai_one_work start mark
[59264.331298] simrupt: ai_one_work finish mark
[59264.352892] simrupt: ai_two_work start mark
[59264.358070] simrupt: ai_two_work finish mark
[59264.456828] simrupt: ai_one_work start mark
[59265.321556] simrupt: ai_one_work finish mark
[59265.392898] simrupt: ai_two_work start mark
[59265.395084] simrupt: ai_two_work finish mark
[59265.496840] simrupt: ai_one_work start mark
[59265.906610] simrupt: ai_one_work finish mark
[59265.912808] simrupt: ai_two_work start mark
[59265.913236] simrupt: ai_two_work finish mark
```
可以發現兩個 AI 演算法的執行順序正如預期的樣子，前一個演算法執行完畢才會開始執行下一個演算法。
> [commit 99ea94e](https://github.com/vax-r/KMLdrv/commit/99ea94e22162ae40a6f7e7dbe6e05c3e3ac6a742)
## 計算 MCTS load average
參考 Linux kernel 計算 CPU load average 的方式，我在此處將當前 MCTS 樹當中存在的節點數量作為計算單位，並且每次將計算 load average 的任務置於 work queue 當中進行計算。
特別注意此處我並沒有嚴謹的對 MCTS 的 `nr_active_nodes` 進行 CPU 之間的同步，避免因此對整體性能產生衝擊，有計算出變化量即可。
在 `mcts.c` 當中的一些更動
```diff
int mcts(char *table, char player)
{
...
struct node *root = new_node(-1, player, NULL);
+ mcts_obj.nr_active_nodes = 1;
...
if (node->children[0] == NULL)
- expand(node, temp_table);
+ mcts_obj.nr_active_nodes += expand(node, temp_table);
...
}
```
在 `simrupt.c` 當中也要新增對應計算函式
```c
static void mcts_calc_load(struct work_struct *w)
{
unsigned long active_nodes;
active_nodes = count_active_nodes() * FIXED_1;
mcts_avennode[0] = calc_load(mcts_avennode[0], EXP_1, active_nodes);
mcts_avennode[1] = calc_load(mcts_avennode[1], EXP_5, active_nodes);
mcts_avennode[2] = calc_load(mcts_avennode[2], EXP_15, active_nodes);
int a = mcts_avennode[0] + (FIXED_1 / 200);
int b = mcts_avennode[1] + (FIXED_1 / 200);
int c = mcts_avennode[2] + (FIXED_1 / 200);
pr_info("kmldrv: [MCTS LoadAvg] %d.%02d %d.%02d %d.%02d\n", LOAD_INT(a),
LOAD_FRAC(a), LOAD_INT(b), LOAD_FRAC(b), LOAD_INT(c), LOAD_FRAC(c));
}
```
並週期性的將對應 work item 置於 workqueue 當中
```diff
+ queue_work(kmldrv_workqueue, &mcts_calc_load_work);
```
> [commit 627046f](https://github.com/vax-r/KMLdrv/commit/627046f4488028259d59b20da9db002b94133a0e)
## 使用者層級互動
撰寫使用者層級程式 `kmldrv-user.c` 作為與 `kmldrv` 的檔案裝置互動的媒介，首先是顯示對弈過程
```diff
+ FILE *device_ptr = fopen(KMLDRV_DEVICE_FILE, "r");
+ char display_buf[DRAWBUFFER_SIZE];
+ while (fgets(display_buf, DRAWBUFFER_SIZE, device_ptr)) {
+ printf("%s", display_buf);
+ }
+
+ fclose(device_ptr);
```
透過 `fopen(), fgets()` 即可讀取核心對弈過程。
> [commit baed53d](https://github.com/vax-r/KMLdrv/commit/baed53dceeb4ae56584926254a38ad4131946820)
接下來要透過使用者層級的輸入來判斷是否重新開始一個棋局或者暫停顯示等等操作，需要從使用者層級傳送資料到核心空間，本來我想採用 ioctl 的機制，但由於 ioctl 要事先知道裝置的 major number ，但我想讓系統動態配置裝置的 major number ，因此用 ioctl 並不合適。
於是我選用 sysfs 機制，透過寫入代表 kernel module 當中變數的檔案和 kernel module 進行溝通，以暫停顯示對弈棋盤為例，首先新增 kernel attribute 如下
```c
struct kmldrv_attr {
char display;
char restart;
char end;
rwlock_t lock;
};
```
同時也要定義對應的 `show(), store()` 函式，最後透過 `DEVICE_ATTR_RW()` 巨集來取得 attribute 。
```c
static ssize_t kmldrv_state_show(struct device *dev, struct device_attribute *attr, char *buf)
{
read_lock(&attr_obj.lock);
int ret = sprintf(buf, "%c %c %c\n", attr_obj.display, attr_obj.restart, attr_obj.end);
read_unlock(&attr_obj.lock);
return ret;
}
static ssize_t kmldrv_state_store(struct device *dev, struct device_attribute *attr, const char *buf, size_t count)
{
write_lock(&attr_obj.lock);
sscanf(buf, "%c %c %c", &(attr_obj.display), &(attr_obj.restart), &(attr_obj.end));
write_unlock(&attr_obj.lock);
return count;
}
static DEVICE_ATTR_RW(kmldrv_state);
```
然後要在 kernel module 初始化的時候建立對應 sysfs 檔案，透過 `device_create_file()`
```c
ret = device_create_file(kmldrv_dev, &dev_attr_kmldrv_state);
if (ret < 0) {
printk(KERN_ERR "failed to create sysfs file kmldrv_state\n");
goto error_cdev;
}
```
在使用者層級的程式也要有對應處理機制，讓終端機的 stdin 轉為 raw mode ，並監聽輸入的字元並進行對應的處理。
```diff
+ if (read(STDIN_FILENO, &input, 1) == 1) {
+ switch(input) {
+ case 16:
+ char buf[20];
+ read(attr_fd, buf, 5);
+ buf[0] = buf[0] - '0' ? '0' : '1';
+ write(attr_fd, buf, 5);
+ break;
+ case 17:
+ break;
+ }
+ }
```
到目前為止可以讓 kernel module 進行無限次重複對弈的棋局，並在使用者層級透過 Ctrl + P 控制是否顯示對弈棋盤。
> [commit 783f424](https://github.com/vax-r/KMLdrv/commit/783f42422bfb5270a7e8c2a4b2d6d081217817b9)
上述做法在第一次利用 Ctrl + P 暫停顯示棋盤後，整個使用者層級行程會卡在 `read()` 處等待資料送入，但不會有資料送進該檔案因為 kernel module 停止傳送資料到使用者層級，於是我利用 I/O multiplexing 機制也就是 `select()` 來達到同時監聽 `STDIN_FILENO` 和裝置檔案是否有輸入，藉此使得 Ctrl + P 可以重複使用，並且 Ctrl + Q 可以使整個核心空間的對弈終止。
啟用 Ctrl + Q 需要先在終端機輸入以下命令
```shell
$ stty start '^-' stop '^-'
```
> [commit 33d8dbd](https://github.com/vax-r/KMLdrv/commit/33d8dbd989a188267a48ecb233d03cba3023de71)
### 自動更新對弈畫面
透過在每次輸出前先利用 ASCII code 的跳脫字元來將終端機畫面刷新，就可以避免一直出現重複畫面的問題
```diff
} else if (read_attr && FD_ISSET(device_fd, &readset)) {
FD_CLR(device_fd, &readset);
+ printf("\033[H\033[J"); /* ASCII escape code to clear the screen */
read(device_fd, display_buf, DRAWBUFFER_SIZE);
```
> [commit c2cf2f6](https://github.com/vax-r/KMLdrv/commit/c2cf2f6d830cebe6fb6a6c2cbb8a338f551b595d)
## 成果影片
[Kernel space tic-tac-toe game](https://youtu.be/Y_xdLrDVGzk)
## PRNG Efficiency
Xorshit 系列亂數產生器是 Linear-feedback shift register 的子集合，因此我們先探討何為 LFSR 。
### Linear-feedback shift register
根據[維基百科](https://en.wikipedia.org/wiki/Linear-feedback_shift_register)說明， LFSR 是一個 shift register ，而其 input bit 是它上一個 state 的 linear function 。例如最常用來作為 single bits linear function 的 exclusive-or (XOR) ，因此 LFSR 最常見就是一個 shift register 而 input bit 是由這些 shift register 當中數值位元做 XOR 得到的。
LFSR 的初始值稱為 seed ，由於接下來每一步 register 操作都是 deterministic ，所以中間過程會產生什麼數值完全由初始值決定，而由於 register 所有可能 states 的集合是有限的，因此過程當中產生的數值會形成一個 repeating cycle ，如果我們的 feedback function 挑選的夠好，就可以產生一個超級長的 cycle 使得產生的結果看似是隨機的。
:::info
概念看起來很類似 Markov Decision Process
:::
### Fibonacci LFSRs
Fibonacci LFSR 架構如上圖，其中作為下個 state 的 input 的位元稱為 taps ，可以看到此架構的 taps 是第 11, 13, 14, 16 個位元。LFSR 最右邊的位元又稱為 output bit ，每次從當前 state 變為下個 state 時，將每個位元向右移動一個位元，並且將 taps 經過 XOR 後的結果寫入最左邊的位元，在這些過程結束後，讀取最右邊的位元。
有 m 個位元空間的 LFSR 可以產生 $2^m - 1$ 種序列，除了全部都是 0 的序列非法以外（這樣的序列會讓整個 LFSR 狀態永遠一樣）。另外值得一提的是在某些初始值為 0 的 flip-flop LFSR 當中，可以利用 XNOR 來取代 XOR ，可以獲得一樣的效果，只是此時是全為 1 的序列非法。
此架構中的 feed back polynomial 可以表示為
$$
x^{16} + x^{14} + x^{13} + x^{11} + 1
$$
常數 1 不代表任何 tap ，而是對應到 first bit 。而 LFSR 滿足 maximal-length iff feedback polynomials 都為 primitive over the [Galois field GF(2)](https://en.wikipedia.org/wiki/Finite_field) ，這代表滿足以下兩個必要但非充分條件
* taps 總數為偶數
* taps 集合為 setwise co-prime ，也就是所有 taps 的共同 divisor 只有 1 。
特別注意到對於一個 LFSR length 來說可以有數個不同的 maximum-length tap sequence ，並且如果一個 maximum-length tap sequence 被找到了，則其他的也會跟著被找到。如果一個 n-bit LFSR 的 tap sequence 為 [n, A, B, C, 0] ，則對應的 mirror sequence [n, n-C, n-B, n-A, 0] 。
### Galois LFSRs
又稱為 modular, internal XORs, 或者 one-to-may LFSR ，整個系統進行 transition 時會鎖住，除了 taps 以外的 bits 會直接往右位移一個位元，至於 taps 則會和 ouput bit 進行 XOR 後再送往下一個位元，新產生的 output bit 會是下一個 input bit ，若 output bit 為 0 ，則所有位元皆往右位移一個位元，且 input bit 變為 0 ，若 output bit 為 1 ，則所有 taps bit 都反轉再往右位移，然後 input bit 變為 1 。
### Problems of LFSR
傳統的 LFSR 提供快速產生 pseudo random number 的機制，並且週期很大，且只需要 xor 和 shift 的操作即可做到，提供給硬體或軟體實作上非常大的便利，但他們的線性結構使得它們無法通過某些統計測試，例如 binary-rank test 和 linear-complexity test 。
我們可以透過 **scrambling linear generator** 來去除它們的線性結構，也就是在原本的 state array 上套用 nonlinear function ，結果才是 actual output 。不過即便如此較低的幾個位元受到 scramble 操作的影響非常有限，因此單看那幾個位元的話依然無法通過 linearity test 。
### Scrambled Linear Pseudorandom Number Generators
此論文提出兩種 linear engine 並結合 scrambler ，並且允許 CPU 將 linear engine 內部操作平行化。提出的兩種 linear engine 分別為 **xoroshiro** (xor/rotate/shift/rotate) 和 **xoshiro** (xor/shift/rotate) ，注意到現代的編譯器可以將 rotation 轉為一個 CPU instruction ，所以其操作成本和 shift 相同，但提供更好的 state diffusion ，更重要的是沒有位元會被捨棄。
文中 word size 為 $w$ bits 。 $S$ 代表將一個 $w \times w$ 矩陣左移一個位元（每個 row 左移）， $R$ 代表將 $w \times w$ 矩陣進行一次 left rotation ， $\rho_r(-)$ 代表對一個 $w$ bit 向量進行 r 次 left rotation 。
**xoroshiro**
對一個 state array 循環地每次更新兩個 words ，可以將每次操作定義為以下 $2w \times 2w$ transition matrix
$$
K_{2w} = \begin{pmatrix}
R^a + S^b + I & R^c\\
S^b + I & R^c
\end{pmatrix}
$$
若擴展為 $kw \times kw$ 會變為以下
$$
K_{kw} = \begin{pmatrix}
0 & 0 & \cdots & 0 & R^a+S^b+I & R^c\\
I & 0 & \cdots & 0 & 0 & 0\\
0 & I & \cdots & 0 & 0 & 0\\
\cdots & \cdots & \cdots & \cdots & \cdots & \cdots \\
0 & 0 & \cdots & I & 0 & 0\\
0 & 0 & \cdots & 0 & S^b+I & R^c
\end{pmatrix}
$$
可以注意到對一個 k words 的 state array 來說，上述操作只會影響第一個和最後一個 word ，剩下的 words 都只會位移一個位置。
假設我們的 word size $w = 64$ 且狀態由 128 bits 表示 ，程式碼實作可以表示為以下
```c
const uint64_t s0 = s[0];
uint64_t s1 = s[1];
const uint64_t result_plus = s0 + s1;
const uint64_t result_plusplus = rotl(s0 + s1, R) + s0;
const uint64_t result_star = s0 * S;
const uint64_t result_starstar = rotl(s0 * S, R) * T;
s1 ^= s0;
s[0] = rotl(s0, A) ^ s1 ^ (s1 << B);
s[1] = rotl(s1, C);
```
若 state array 有 16 個數字，也就是 xoroshift1024 ，則 generator 的狀態轉移可以實作為
```c
const int q = p; // holds a number in the interval [0... 16)
const uint64_t s0 = s[p = (p+1) & 15];
uint64_t s15 = s[q];
const uint64_t result_plus = s0 + s15;
const uint64_t result_plusplus = rotl(s0 + s15, R) + s15;
const uint64_t result_star = s0 * S;
const uint64_t result_starstar = rotl(s0 * S, R) * T;
s15 ^= s0;
s[q] = rotl(s0, A) ^ s15 ^ (s15 << B);
s[p] = rotl(s15, C);
```
有 "result" 前綴的變數為 scamblers 的結果，此處可以特別注意到，在 `s1 ^= s0` 或 `s15 ^= s0` 後，兩個 words 的操作就可以完全獨立， data dependency 就只存在這一行而已。
**xoshiro**
每次迴圈的線性操作只有 shift 和 rotation ，並且在 $2w$ bits 是無法定義的，對於 $16w$ bits 來說太慢，因此我們只探討 $4w$ 和 $8w$ ，首先以 state arry 為 4 個 64 位元的 word 組成的狀態來說，實作 xoshiro256 generator 。
```c
const uint64_t result_plus = s[0] + s[3];
const uint64_t result_plusplus = rotl(s[0] + s[3], R) + s[0];
const uint64_t result_starstar = rotl(s[1] * S, R) * T;
const uint64_t t = s[1] << A;
s[2] ^= s[0];
s[3] ^= s[1];
s[1] ^= s[2];
s[0] ^= s[3];
s[2] ^= t;
s[3] = rotl(s[3], B);
```
對應的轉移矩陣為
$$
L_{4w} = \begin{pmatrix}
I & I & I & 0\\
I & I & S^a & R^b\\
0 & I & I & 0\\
I & 0 & 0 & R^b\\
\end{pmatrix}
$$
xoshiro 的轉移矩陣不像 xoroshiro 轉移矩陣一樣容易看出規律，總結來說它的操作為將第二個 word 進行位移並儲存，之後 state array 當中每個 word 都和另一個不同的 word 進行 xor ，而第二個 word 位移後的結果會和倒數第二個 word 做 xor ，最後一個 word 會進行旋轉。
#### Scrambler
* nonlinear mapping from the state of the linear engine to a $w$-bit value
**Sum**
將 state array 中兩個 words 相加，相加的兩個 word 的選擇非常重要，對於整體 generator 的品質有極大的影響，而且 sum scrambler 的 output 的最低位元剛好是一群位元 xor 的結果，遵循某個 linear recurrence ，具有可預測性，因此 sum scrambler 是 weak scramlber
**Multiplication**
將 state array 中某個 word 乘上一個 constant ，該 constant 稱為 multiplier ，必須是奇數， scrambler 才會是 bijection (一對一且映成) 。並且若 second-lowest bit set 在位置 $b$ ，則最低的 $b$ 個位元都會維持不變，剩下的位元會是 0 和 bit $b$ 做 xor 的結果，所以低位元也存在 linear recurrence ，所以 multiplication scrambler 同樣是 weak scrambler 。通常 multiplier 的選擇會很靠近 $\varphi 2^w$ ， $\varphi$ 是 golden ratio ，目的是要最小化 $b$ 也就是最小化不會變化的位元數量。
**Sum, rotation, and again sum**
選擇 state array 當中兩個 words ，先進行相加後，左移 $r$ 個位置，之後再和第一個 word 進行相加，注意相加的前後順序有差別，將第一個和最後一個 word 進行 ++ scrambler 和最後一個 word 與第一個 word 進行 ++ scrambler 不一樣。同時 $r$ 的選擇也很重要，挑選適當的 $r$ 可以使所有位元不具有 low linear complexity ，因此 ++ scrambler 又稱為 strong scrambler 。
**Multiplication, rotation, and again multiplication**
對 state array 當中某個 word 先進行乘法再位移然後再乘法，有三個參數需要選擇，第一個與第二個 multiplier ，還有位移量 $r$ ，和 ++ scrambler 一樣， $r$ 的選擇不難，可以使所有位元都不具有 low linear complexity ，因此 ** scrambler 也是 strong scrambler ，而 multiplier 通常選擇 $2^s + 1$ 的形式，除了必須是奇數以外，這樣的選擇可以使 multiplier 的計算變快很多。
### 整合 xoroshiro 進入 MCTS
本來我想直接讀取 ksort 專案當中的 xoro kernel module 的檔案裝置 `/dev/xoro` 來取得隨機變數，但在 kernel 空間當中開啟檔案的機制和使用者層級不同，因此我直接將 xoroshift PRNG 重新實作在 kmldrv 當中，並在 MCTS 當中利用 xoroshift 取得隨機變數。
我有做些許變更，讓 state array 改用指標的型態傳入每個函式，使得每個使用 xoroshiro 的使用者都能維護自己的 state array ，另外在 `next()` 函式當中原本的實作 scrambler 採用 sum scrambler ，是一個 weak scrambler ，可預測性高，我改用 ++ scramlber 如下
```diff
- const u64 result = s0 + s1;
+ const u64 result = rotl(s0 + s1, 24) + s0;
```
但 rotate 的量 $r$ 我選用 24 並沒有經過實驗測試，是一個未經考量的選擇，尚需改進。
> [commit 414d7b3](https://github.com/vax-r/KMLdrv/commit/414d7b34d64f94fcc7c649a2a6802a23ecd57982)
接下來和原本使用 `wyhash()` 來取得隨機數的方式進行比較，首先比較 MCTS load average ，觀察不同的 PRNG 對 MCTS 節點數量的影響，首先是原本的 `wyhash()` 在一場對弈當中的 load average (此處我將其他的 `pr_info()` 都先註解掉以利觀察)
```shell
[345158.615903] kmldrv: [MCTS LoadAvg] 3313.64 1970.39 769.10
[345158.719962] kmldrv: [MCTS LoadAvg] 3049.57 1937.94 765.05
[345158.823902] kmldrv: [MCTS LoadAvg] 3038.07 1954.01 776.55
[345158.927845] kmldrv: [MCTS LoadAvg] 3027.49 1969.82 787.99
[345159.031864] kmldrv: [MCTS LoadAvg] 3017.77 1985.36 799.37
[345159.135848] kmldrv: [MCTS LoadAvg] 3008.81 2000.65 810.68
[345159.239847] kmldrv: [MCTS LoadAvg] 3000.58 2015.68 821.94
[345159.343849] kmldrv: [MCTS LoadAvg] 2993.01 2030.46 833.13
[345159.447849] kmldrv: [MCTS LoadAvg] 3036.41 2055.44 847.64
[345159.551853] kmldrv: [MCTS LoadAvg] 3076.34 2080.00 862.08
[345159.655854] kmldrv: [MCTS LoadAvg] 3113.06 2104.15 876.43
[345159.759942] kmldrv: [MCTS LoadAvg] 3146.85 2127.91 890.71
[345159.863952] kmldrv: [MCTS LoadAvg] 2895.98 2092.81 886.00
[345159.971847] kmldrv: [MCTS LoadAvg] 2678.57 2061.08 882.22
[345160.075850] kmldrv: [MCTS LoadAvg] 2478.57 2029.86 878.45
[345160.179852] kmldrv: [MCTS LoadAvg] 2294.58 1999.17 874.70
[345160.283853] kmldrv: [MCTS LoadAvg] 2130.62 1970.08 871.33
[345160.384022] kmldrv: [MCTS LoadAvg] 1979.78 1941.47 867.98
[345160.487957] kmldrv: [MCTS LoadAvg] 1822.20 1909.44 863.38
[345160.591890] kmldrv: [MCTS LoadAvg] 1686.77 1879.92 859.45
[345160.695857] kmldrv: [MCTS LoadAvg] 1562.19 1850.88 855.54
[345160.799858] kmldrv: [MCTS LoadAvg] 1448.30 1822.48 851.69
[345160.904048] kmldrv: [MCTS LoadAvg] 1343.54 1794.55 847.87
[345161.007890] kmldrv: X win!!!
```
接著換成 `xoroshiro()` ，同樣取一場對弈的結果觀察
```shell
[345371.736481] kmldrv: [MCTS LoadAvg] 3148.61 3650.60 1722.41
[345371.840295] kmldrv: [MCTS LoadAvg] 4675.81 3958.88 1832.50
[345371.944252] kmldrv: [MCTS LoadAvg] 6080.72 4262.04 1942.01
[345372.048255] kmldrv: [MCTS LoadAvg] 7373.12 4560.17 2050.92
[345372.152269] kmldrv: [MCTS LoadAvg] 8562.03 4853.35 2159.25
[345372.256268] kmldrv: [MCTS LoadAvg] 9655.74 5141.67 2267.00
[345372.360267] kmldrv: [MCTS LoadAvg] 10661.86 5425.20 2374.17
[345372.464267] kmldrv: [MCTS LoadAvg] 11587.41 5704.02 2480.76
[345372.568268] kmldrv: [MCTS LoadAvg] 12469.60 5984.58 2588.85
[345372.672268] kmldrv: [MCTS LoadAvg] 13381.72 6281.34 2703.10
[345372.776260] kmldrv: [MCTS LoadAvg] 14220.81 6573.18 2816.73
[345372.880268] kmldrv: [MCTS LoadAvg] 14992.69 6860.16 2929.76
[345372.984402] kmldrv: [MCTS LoadAvg] 15702.77 7142.39 3042.18
[345373.088330] kmldrv: [MCTS LoadAvg] 14446.52 7024.06 3025.92
[345373.192265] kmldrv: [MCTS LoadAvg] 13447.51 6940.17 3020.25
[345373.296282] kmldrv: [MCTS LoadAvg] 12528.49 6857.68 3014.62
[345373.400299] kmldrv: [MCTS LoadAvg] 11683.07 6776.55 3009.01
[345373.504262] kmldrv: [MCTS LoadAvg] 10905.34 6696.77 3003.43
[345373.608268] kmldrv: [MCTS LoadAvg] 10189.90 6618.32 2997.89
[345373.712266] kmldrv: [MCTS LoadAvg] 9638.73 6563.34 2999.55
[345373.816267] kmldrv: [MCTS LoadAvg] 9131.69 6509.28 3001.20
[345373.920258] kmldrv: [MCTS LoadAvg] 8665.26 6456.12 3002.84
[345374.024419] kmldrv: [MCTS LoadAvg] 8236.18 6403.84 3004.48
[345374.128384] kmldrv: [MCTS LoadAvg] 7577.69 6297.74 2988.41
[345374.232302] kmldrv: [MCTS LoadAvg] 6984.89 6196.09 2973.30
[345374.336276] kmldrv: [MCTS LoadAvg] 6439.57 6096.13 2958.27
[345374.440261] kmldrv: [MCTS LoadAvg] 5943.52 5998.99 2943.69
[345374.544392] kmldrv: [MCTS LoadAvg] 5487.19 5903.47 2929.20
[345374.648370] kmldrv: [MCTS LoadAvg] 5048.67 5805.64 2913.52
[345374.752254] kmldrv: [MCTS LoadAvg] 4652.47 5710.94 2898.42
[345374.856258] kmldrv: [MCTS LoadAvg] 4287.99 5617.80 2883.39
[345374.960258] kmldrv: [MCTS LoadAvg] 3955.27 5526.75 2868.62
[345375.064418] kmldrv: [MCTS LoadAvg] 3649.19 5437.20 2853.93
[345375.168284] kmldrv: X win!!!
```
觀察數字分佈可以很明顯看出利用 `xoroshiro()` 計算隨機數比起 `wyhash()` 計算隨機數，會讓 MCTS 節點的 load average 更大，也就是每一步模擬都會產生更多節點，挖到更深的可能性。
這是否影響到勝率呢？我們採用兩種不同的 PRNG 和 negamax 演算法進行 100 場對弈觀看結果。先在 `timer_handler()` 進行以下變更來計算各自勝場數
```diff
+ #define GAME_LIMIT 100
+ static int game_count = 0;
+ static int ai_one_win = 0;
+ static int ai_two_win = 0;
static void timer_handler(struct timer_list *__timer)
...
+ game_count++;
- if (attr_obj.end == '0') {
+ if (attr_obj.end == '0' && game_count < GAME_LIMIT) {
...
+ if (win == 'O')
+ ai_one_win++;
+ if (win == 'X')
+ ai_two_win++;
```
要等待一段時間讓 kernel thread 進行對弈。
**wyhash**
```shell
[346435.214741] kmldrv: AI_one win 21 games
[346435.214743] kmldrv: AI_two win 79 games
```
**xoroshiro**
```shell
[346849.198520] kmldrv: AI_one win 20 games
[346849.198524] kmldrv: AI_two win 80 games
```
結果看來對勝負率影響幾乎沒有差異，但 xoroshiro 要使用的資源卻遠大於 wyhash 。
## 補充
[Concurrency 學習筆記](https://hackmd.io/@vax-r/concurrency)
Sign in with Wallet
Connect another wallet