# 2024q1 Homework6 (integration)
contributed by < [jeremy90307](https://github.com/jeremy90307) >
## 閱讀 [The Linux Kernel Module Programming Guide](https://sysprog21.github.io/lkmpg/#headers)
花了些許時間閱讀，把相關問題及重點紀錄在這 ----> [The Linux Kernel Module Programming Guide 筆記](https://hackmd.io/@jeremytsai/HW6_note)
### 解釋 [simrupt](https://github.com/sysprog21/simrupt) 程式碼裡頭的 mutex lock 的使用方式，並探討能否改寫為 [lock-free](https://hackmd.io/@sysprog/concurrency-lockfree)
> 參考 : [2021 年的筆記](https://hackmd.io/@linD026/simrupt-vwifi)
### kfifo
**解釋**
> 參考 [kfifo（linux kernel 无锁队列）](https://zhuanlan.zhihu.com/p/500070147)
>
kfifo 使用 in 和 out 兩個變量作為進入和離開的索引，如圖
- 進入 n 筆資料時， in 變量就 `+ n`
- 離開 n 筆資料時， out 變量就 `- n`
- out 不允許大於 in (out = in 表示 fifo 為空)
- in 不允許比 out 大(即超出 fifo 的總大小，會有液出)
在 simrupt.c 的註釋中解釋為何無須再使用額外的鎖
```
/* NOTE: the usage of kfifo is safe (no need for extra locking), until there is
* only one concurrent reader and one concurrent writer. Writes are serialized
* from the interrupt context, readers are serialized using this mutex.
*/
```
使用到的 kfifo API
`DECLARE_KFIFO_PTR` — macro to declare a fifo pointer object
`kfifo_in` - put data into the fifo
`kfifo_to_user` — copies data from the fifo into user space
`kfifo_len` — returns the number of used elements in the fifo
`kfifo_alloc` — dynamically allocates a new fifo buffer
`kfifo_free` — frees the fifo
> 更多 FIFO 的 API 參考 -> [kfifo](https://archive.kernel.org/oldlinux/htmldocs/kernel-api/kfifo.html)
在 [/include/linux/kfifo.h](https://elixir.bootlin.com/linux/latest/source/include/linux/kfifo.h) 可以看到 **kfifo 的結構體**
```c
struct __kfifo {
unsigned int in; // 進入索引
unsigned int out; // 離開索引
unsigned int mask; // 為佇列總大小 - 1
unsigned int esize; // 單個元素大小
void *data; // 指向佇列記憶體位置
};
#define __STRUCT_KFIFO_COMMON(datatype, recsize, ptrtype) \
union { \
struct __kfifo kfifo; \
datatype *type; \
const datatype *const_type; \
char (*rectype)[recsize]; \
ptrtype *ptr; \
ptrtype const *ptr_const; \
}
```
- `kfifo_in`
```c
unsigned int __kfifo_in(struct __kfifo *fifo,
const void *buf, unsigned int len)
{
unsigned int l;
l = kfifo_unused(fifo);
if (len > l)
len = l;
kfifo_copy_in(fifo, buf, len, fifo->in);
fifo->in += len;
return len;
}
```
`kfifo_unused(fifo)` 用作查詢剩餘大小，若進入的大小大於剩餘大小，則以剩餘的大小為主 `len = l` ，最後 `kfifo_copy_in` 複製資料至記憶體。
```c
/*
* internal helper to calculate the unused elements in a fifo
*/
static inline unsigned int kfifo_unused(struct __kfifo *fifo)
{
return (fifo->mask + 1) - (fifo->in - fifo->out);
}
```
`fifo->mask + 1` 為整個佇列的大小減去已用大小 `fifo->in - fifo->out`
```c
static void kfifo_copy_in(struct __kfifo *fifo, const void *src,
unsigned int len, unsigned int off)
{
unsigned int size = fifo->mask + 1;
unsigned int esize = fifo->esize;
unsigned int l;
off &= fifo->mask;
if (esize != 1) {
off *= esize;
size *= esize;
len *= esize;
}
l = min(len, size - off);
memcpy(fifo->data + off, src, l);
memcpy(fifo->data, src + l, len - l);
/*
* make sure that the data in the fifo is up to date before
* incrementing the fifo->in index counter
*/
smp_wmb();
}
```
:::info
為何使用兩次 `memcpy` ?
:::
- `kfifo_alloc`
申請記憶體空間並初始化 kfifo
```c
int __kfifo_alloc(struct __kfifo *fifo, unsigned int size,
size_t esize, gfp_t gfp_mask)
{
/*
* round up to the next power of 2, since our 'let the indices
* wrap' technique works only in this case.
*/
size = roundup_pow_of_two(size);
fifo->in = 0;
fifo->out = 0;
fifo->esize = esize;
if (size < 2) {
fifo->data = NULL;
fifo->mask = 0;
return -EINVAL;
}
fifo->data = kmalloc_array(esize, size, gfp_mask);
if (!fifo->data) {
fifo->mask = 0;
return -ENOMEM;
}
fifo->mask = size - 1;
return 0;
}
```
當中 `roundup_pow_of_two` 可以求出 size 向上最接近的 2 的冪次
程式碼取自 [/include/linux/log2.h](https://elixir.bootlin.com/linux/latest/source/include/linux/log2.h#L174)
```c
/**
* roundup_pow_of_two - round the given value up to nearest power of two
* @n: parameter
*
* round the given value up to the nearest power of two
* - the result is undefined when n == 0
* - this can be used to initialise global variables from constant data
*/
#define roundup_pow_of_two(n) \
( \
__builtin_constant_p(n) ? ( \
((n) == 1) ? 1 : \
(1UL << (ilog2((n) - 1) + 1)) \
) : \
__roundup_pow_of_two(n) \
)
```
當中 `__builtin_constant_p(n)` 函式中的 n 若為常數回傳 1 ，反之為 0 。
> 參考 [Other Built-in Functions Provided by GCC](https://gcc.gnu.org/onlinedocs/gcc/Other-Builtins.html)
### circular buffer
> 參考
> -> [並行程式設計: Ring buffer](https://hackmd.io/@sysprog/concurrency/%2F%40sysprog%2Fconcurrency-ringbuffer#Ring-Buffer)
> -> [circular buffer](https://zh.wikipedia.org/zh-tw/%E7%92%B0%E5%BD%A2%E7%B7%A9%E8%A1%9D%E5%8D%80)
先看註釋描述，看似要做出一個更快速的 circular buffer ，並結合 kfifo 。
:::info
更快速?
:::
```
/* We use an additional "faster" circular buffer to quickly store data from
* interrupt context, before adding them to the kfifo.
*/
```
查找 [/include/linux/circ_buf.h](https://elixir.bootlin.com/linux/latest/source/include/linux/circ_buf.h) 來了解 simrupt 中 circular buffer 使用到的 API
**結構體**
- `buf` : real pointer that contains the array
- `head`,`tail` : head and tail indices
```c
struct circ_buf {
char *buf;
int head;
int tail;
};
```
`CIRC_CNT` 回傳 buffer 被佔用的大小
> 緩衝區中總是有一個儲存單元保持未使用狀態。因此緩衝區最多存入 `size - 1` 。
```c
/* Return count in buffer. */
#define CIRC_CNT(head,tail,size) (((head) - (tail)) & ((size)-1))
```
`CIRC_SPACE` 回傳緩衝區剩餘大小
```c
/* Return space available, 0..size-1. We always leave one free char
as a completely full buffer has head == tail, which is the same as
empty. */
#define CIRC_SPACE(head,tail,size) CIRC_CNT((tail),((head)+1),(size))
```
- `fast_buf_clear` head = tail = 0 表示均指向同個位置，因此 buffer 為空。
```c
/* Clear all data from the circular buffer fast_buf */
static void fast_buf_clear(void)
{
fast_buf.head = fast_buf.tail = 0;
}
```
- `fast_buf_get`
- `fast_buf_put`
### Bottom half
> 中斷概念參考 [Linux 核心設計: 中斷處理和現代架構考量](https://hackmd.io/@sysprog/linux-interrupt#Linux-%E6%A0%B8%E5%BF%83%E8%A8%AD%E8%A8%88-%E4%B8%AD%E6%96%B7%E8%99%95%E7%90%86%E5%92%8C%E7%8F%BE%E4%BB%A3%E6%9E%B6%E6%A7%8B%E8%80%83%E9%87%8F)
為防止 interrupt 產生打架的情況，分成三種類型：
- Critical : Top-half (不允許其他中斷，立即處理)
ex : acknowledge interrupt
- Non-critical : Top-half (允許其他中斷)
ex : read key scan code, add to buffer
- Non-cirtical deferrable : Bottom half, do it **later** (允許其他中斷)
可以被延遲中斷，意味著可以重新被排程
ex : copy keyboard buffer to terminal handler process
在 Linux 作業系統中，使用了三種機制去實現 Bottom Half
- softirq
- tasklet
- workqueue
softirqs 和 tasklet 執行於 interrupt context ， workqueue 執行於 process context 。
> 相關問題：
> [Difference between interrupt context and process context?](https://stackoverflow.com/questions/57987140/difference-between-interrupt-context-and-process-context)
#### softirq
- 是 reentrant function，同一種 IRQ 可以在多個 CPU 上並行執行。
- 使用靜態建立
- kernel thread
#### tasklet
- 建構在 softirq 之上，tasklet 相較 softirq 允許動態建立。
- 同一時間只執行在一個 CPU 上，且不為 reentrant function 。
- kernel thread
在 simrupt 中
定義 tasklet
```c
#define DECLARE_TASKLET_OLD(arg1, arg2) DECLARE_TASKLET(arg1, arg2, 0L)
```
加入 tasklet
```c
tasklet_schedule(&simrupt_tasklet);
```
刪除 tasklet
```
tasklet_kill(&simrupt_tasklet);
```
#### Workqueue
- 概念為對中斷的處理建立每個 CPU 的 kernel thread
在 simrupt 中建立 workqueue
```c
static struct workqueue_struct *simrupt_workqueue;
```
當中 `WQ_UNBOUND` 表示不綁定任何 CPU
> 更多 Workqueue flags 參考 [/include/linux/workqueue.h](https://elixir.bootlin.com/linux/latest/source/include/linux/workqueue.h#L351)
```c
/* Create the workqueue */
simrupt_workqueue = alloc_workqueue("simruptd", WQ_UNBOUND, WQ_MAX_ACTIVE);
if (!simrupt_workqueue) {
vfree(fast_buf.buf);
device_destroy(simrupt_class, dev_id);
class_destroy(simrupt_class);
ret = -ENOMEM;
goto error_cdev;
}
```
執行指定的 workqueue 任務，當中 `&work` 指向具體任務的指針。
```c
queue_work(simrupt_workqueue, &work);
```
釋放 workqueue
```c
destroy_workqueue(simrupt_workqueue);
```
當 producer 將數據儲存至 kfifo buffer 及 consumer 將數據從 circular buffer 取出時使用互斥鎖。
### Timer
使用 `timer_handler` 函式來模擬中斷的發生，在函式中 `local_irq_disable` 用於停用中斷，來避免中斷時又遇到中斷的發生，接著將延遲中斷的部份使用 `process_data` 將字元放入 circular buffer ，並啟用 tasklet 來處理中斷，最後使用 `local_irq_enable` 函式再次啟用中斷。
### mutex lock
在 simrupt 中定義的鎖
```c
/* NOTE: the usage of kfifo is safe (no need for extra locking), until there is
* only one concurrent reader and one concurrent writer. Writes are serialized
* from the interrupt context, readers are serialized using this mutex.
*/
static DEFINE_MUTEX(read_lock);
```
註釋指出，在特定條件下，即只有一個並行 reader 和一個並行 writer 時，對 kfifo 的使用是安全的，無需額外的鎖定機制。
考慮到並行，因寫入操作是由 interrupt context 進行(不可被搶佔、sleep)，因此只有一個 writer 可以存取 kfifo，但讀取操作就必須使用 `DEFINE_MUTEX(read_lock)` 來限制只能有一位讀者存取 kfifo ，總結最後確保 kfifo 的並行是安全的，在同一時間只有一個 writer 及 reader 可以存取 kfifo 。
```c
/* Mutex to serialize kfifo writers within the workqueue handler */
static DEFINE_MUTEX(producer_lock);
/* Mutex to serialize fast_buf consumers: we can use a mutex because consumers
* run in workqueue handler (kernel thread context).
*/
static DEFINE_MUTEX(consumer_lock);
```
藉由 kernel thread 取得 circular buffer 的資料，使用 mutex 來避免執行緒之間在取得資料時發生問題
```c
/* Consume data from the circular buffer */
mutex_lock(&consumer_lock);
val = fast_buf_get();
mutex_unlock(&consumer_lock);
```
:::info
**疑問**
前面的註釋提到寫入 kfifo 的操作在 interrupt context 已確保同時只能有一個寫著存取，為何這裡還要在上鎖？
:::
```c
/* Store data to the kfifo buffer */
mutex_lock(&producer_lock);
produce_data(val);
mutex_unlock(&producer_lock);
```
當 circular buffer 為空時 `if (val < 0)` 喚醒等待中的佇列
```c
wake_up_interruptible(&rx_wait);
```
```c
/* Wait queue to implement blocking I/O from userspace */
static DECLARE_WAIT_QUEUE_HEAD(rx_wait);
```
### 流程圖
> 參考[江冠霆同學](https://hackmd.io/@Henry0922/linux2024-homework6#simrupt)的作業時發現他整理了一個流程圖，可以加速認識 simrupt 的運作原理，後續回頭複習時能更快了解。
### lock-free
[Linux 核心設計：淺談同步機制](https://www.youtube.com/watch?v=lNt-ewfygy8)
- coarse-grained vs fine-grained
用粗粒(coarse-grained)細粒(fine-grained)來舉 lock 的優劣
- Atomic instruction
`x++`,`x--` -> 確保不會被 CPU 交錯執行
- Compare and swap (CAS)
* 語意: `if (x == y) x = z;`
```C
int compareAndSwap(int *loc, int expectedValue, int valueToStore) {
ATOMIC {
int val = *loc;
if (val == expected) {
*loc = valueToStore;
}
return val;
}
}
```
若傳入為符合預期的值進行 swap ，簡單的程式碼，重點在於 `ATOMIC`
**並非無鎖的操作就是 `lock-free`**
- **一個 lock-free 的程式需要確保符合這個特徵：執行該程式的所有執行緒至少有一個能夠不被阻擋 (blocked) 而繼續執行**
* lock-Free 中的 "Lock" 沒有直接涉及 mutex 和 semaphore 一類 mutual exclusion，而是描述了應用程式受限於特定原因被鎖定的可能，例如可能因為 dead lock, live lock，或者 thread scheduling 致使的 priority inversion 等等
* 下方的程式碼沒有任何 mutex，但卻不符合 lock-free 的屬性：若有兩個執行緒同時執行程式碼，考慮到特定的排程方式，非常可能兩個執行緒同時陷入無窮迴圈，也就是阻擋彼此
* Lock-Free 程式設計的挑戰不僅來自於其任務本身的複雜性，還要始終著眼於對事物本質的洞察
```c
while (X == 0 )
{
X = 1 - X;
}
```
**討論能否改成 lock-free**
下方程式碼使用 mutex 確保從 circular buffer 取得的值是一致的
```c
mutex_lock(&consumer_lock);
val = fast_buf_get();
mutex_unlock(&consumer_lock);
```
在 `fast_buf_get` 中或許可以使用 atomic 的 CAS 操作來改寫，但具體怎麼改寫還在思考。
## 閱讀 [Linux 核心模組運作原理](https://hackmd.io/@sysprog/linux-kernel-module)
- [ ] 閱讀〈Linux 核心模組運作原理〉並對照 Linux 核心原始程式碼 (v6.1+)，解釋 insmod 後，Linux 核心模組的符號 (symbol) 如何被 Linux 核心找到 (使用 List API)、MODULE_LICENSE 巨集指定的授權條款又對核心有什麼影響 (GPL 與否對於可用的符號列表有關)，以及藉由 strace 追蹤 Linux 核心的掛載，涉及哪些系統呼叫和子系統？
因目前設備使用的 kernel 核心版本為 v6.5.0，為了後續實作查看 Linux kernel v6.5.0 的源程式碼
使用 strace 追蹤執行 insmod hello.ko 的過程有哪些系統呼叫被執行:
```
$sudo strace insmod hello.ko
```
當中的 `finit_module` 在 [kernel/module/main.c](https://elixir.bootlin.com/linux/v6.5/source/kernel/module/main.c) 中可以找到，接著呼叫 `idempotent_init_module` -> `init_module_from_file` -> `load_module`，最後呼叫的 `load_module` 為 Linux 核心配置記憶體及載入模組。
接著查看到 `load_module` 的程式碼，-> `simplify_symbols` -> `resolve_symbol_wait ` -> `resolve_symbol` -> `find_symbol` -> `list_for_each_entry_rcu`
`list_for_each_entry_rcu` 是一個 RCU 的 List API
```c
/**
* list_for_each_entry_srcu - iterate over rcu list of given type
* @pos: the type * to use as a loop cursor.
* @head: the head for your list.
* @member: the name of the list_head within the struct.
* @cond: lockdep expression for the lock required to traverse the list.
*
* This list-traversal primitive may safely run concurrently with
* the _rcu list-mutation primitives such as list_add_rcu()
* as long as the traversal is guarded by srcu_read_lock().
* The lockdep expression srcu_read_lock_held() can be passed as the
* cond argument from read side.
*/
#define list_for_each_entry_srcu(pos, head, member, cond) \
for (__list_check_srcu(cond), \
pos = list_entry_rcu((head)->next, typeof(*pos), member); \
&pos->member != (head); \
pos = list_entry_rcu(pos->member.next, typeof(*pos), member))
```
Sign in with Wallet
Connect another wallet