Linux 核心專題: 針對鏈結串列的資料排序改進

執行人: fennecJ
解說錄影

Reviewed by Wufangni

• 問題一

  為了使合併過程更加均衡， Timsort 還提出了 minrun （每個 run 的最小長度），若 run 的長度太短，就用 insertion sort 將 run 後面的元素插入到前面的 run 中，直到每個 run 的長度都不小於 minrun 。顯然，minrun 的選擇對 Timsort 算法效率有相當程度的影響。

  想問「後面的元素」是指後一個 run 嗎，且在合併過程中是不是需要重新做排序呢?

是指當前 run 後面的元素，不一定會將後面的 run 完全合併進來，只要合併到當前 run 長度足夠即可。
在合併過程中會透過 insertion sort 重新進行排序，確保合併後的 run 仍符合單調遞增 / 遞減。

• 問題二
  從同學印出的 ext4_fs_map 資料分布觀察下來似乎呈現比較隨機的分布情況，且根據同學在 連結 的實驗結果，timsort 在資料量較少的情況會優於 listsort，可資料量逐漸增加後執行效率明顯低於後者，這是不是能說明 ext4_fs_map 在長久情況下產生的資料量隨機性會較高?

我認為無法僅因為少數幾筆長度較長資料就下這樣的結論，另外一點是，我在影片中有提到，根據我目前的實驗結果，不同於 XFS ， ext4_fs_map 取得的資料長度是在格式化時進行 partition 分割後便會固定下來，不會因為使用時間是否長久產生變更，而是根據 partition 分割的容量大小決定資料長度。

不過我們還是能透過檢視現有蒐集到的資料嘗試歸納出規律：
要回答這個問題有一個困難點：即我該用何種方式說明一筆資料相較另一筆資料「分佈更為隨機」。起先我嘗試透過 隨機性檢定 說明這件事，但我留意到對 timsort 何其衍生算法來說，其不擅長處理是未排序部份佔比很高的資料，和隨機性仍略有不同。
因此我想到一個作法是統計出資料集的 runs 總數，顯然，若一資料越接近排序完成的情況，則 runs 總數會越小。
但是針對越多資料的資料集而言，其 runs 總數本身就會變多，因此無法單純將不同長度資料集的 runs 數量進行比較。
除了 runs 數量外，也可以比較另一項指標，即 runs 的最大長度。而為了讓不同長度的資料集也能進行比較，可以改比較 runs 的最大長度與資料集長度的比值。
此外，也能比較 runs 的長度平均。我修改了實驗程式，將 ext4_fs_map 每筆資料的：

• runs 長度平均
• runs 的最大長度與資料集長度的比值
  統計出來並作圖：

由左到右 partition 的容量漸增，可以看到隨著容量增大，取得的資料集變多，無論是 runs 長度平均或是 run 最大長度與資料集長度的比值均呈現下降的趨勢。

• 問題三

  Timsort 在處理隨機資料時，表現略遜 mergesort 一籌，但對於部分已排序的資料則有出色的表現。

  從實驗部分目前沒有看到有特別針對隨機分佈或部分排序的資料分佈，若是想驗證上述提及 timsort 及 listsort 兩者在兩種分布資料下的效能，或許能再多採用差異更大的資料集進行測試。

針對這個問題，主要原因是因為我目前拿來進行實驗的資料來源皆是從 kernel 中實際的應用場景取得資料分佈並進行實驗。而我目前尚未找到能夠取得隨機分佈的資料分佈來源。不過我先前的確進行過相關實驗，具體可參考我的 HW2 連結 ，在該份筆記中，我實做了 cpython 這個測試資料集中的各種資料分佈並進行實驗。唯該項統計結果是在 user space 進行，測試結果可能受 interrupt/preemption 影響導致不準確。
此外，我留意到 randyuncle 同學進行的 期末專題 - 針對鏈結串列的資料排序改進 有針對隨機資料以及部份排序資料進行實驗，且撰寫為 kernel module 並關閉 interrupt/preemption 使實驗結果更為精確，也可以關注他的後續實驗情形檢視timsort 及 listsort 兩者在兩種分布資料下的效能。

Reviewed by weihsinyeh

問題一 ： PowerSort 發生合併的情況在真實情況會很頻繁嗎？還是合併後提昇的效能會低於計算出「目前要排序的 2 個 run 的中點連線後包含在此區間內的最小節點深度為何」的成本？

PowerSort 發生合併的情況在真實情況是否很頻繁，取決於使用 PowerSort 的資料集本身特性。但由於其合併確保 stack 中 power 越來越高的方式，會讓合併過程趨於平衡，相比原本的 list_sort 若資料符合部份排序的特性則能降低合併的總體次數。
至於這種合併演算法帶來的效益是否會低於計算出「目前要排序的 2 個 run 的中點連線後包含在此區間內的最小節點深度為何」的成本？ 取決於資料本身部份排序的完成度高不高。假設共有 k 個 runs ， PowerSort 總是會計算 k 次「目前要排序的 2 個 run 的中點連線後包含在此區間內的最小節點深度為何」 。若資料本身部份排序的完成度高，則 k 會較小（因為每個 run 長度較長），所需的計算成本也隨之降低。此時這種合併的方法帶來的效能提昇會高於計算的成本。反之，若資料本身部份排序的完成度低，則 k 會較大，帶來的效能提昇可能就會被計算的成本攤平，甚至低於計算所需的成本。

Reviewed by randyuncle

問題：從你的 github repository 中的 datasets 來看，有不少的分佈都是 3 位數的資料量，甚至有 2 位數的資料量。那是不是需要額外特別測小資料量中的排序效率及穩定性？

我不確定這裡同學指的穩定性是指 list 中相同鍵值的元素排序前後相對順序應保持不變還是指同個資料集分類中蒐集到的不同資料分佈下效能表現落差是否穩定，我假設這裡的穩定性是指前者。
針對穩定性的議題，我有在 data_1D.h 及 data_2D.h 中實做 check_list 這個函式進行檢查

static bool check_list(struct list_head *head, size_t statistic)
{
    ...
    if (entry->val == safe->val && entry->seq > safe->seq)
        unstable++;
}

至於對小資料量中的排序效率相關實驗，在我放上 repo 中的資料集有一系列資料叫做 extent_list ，其實這個資料來源有部份資料長度僅為不足 10 個（通常是 5 ～ 7 個），針對這些資料我也有進行實驗，首先，由於 check_list 這個函式的存在，因此我有檢查過穩定性沒有問題。
至於排序效率，可能是 extent_list 本身的資料特性所至，大多時候 timsort 和其衍生算法比較次數和排序時間皆小於 list_sort 。

任務描述

以第一週測驗二給定的 Timsort 為基礎，針對 Linux 核心風格的鏈結串列，提出資料排序的改進方案。

TODO: 探討 Timsort 原理並提出改進方案

Timsort 可視為 merge sort 的改進，其建立在一重要觀察之上： 在現實世界中，許多資料往往是由一些已經 部分排序 的序列組合而成，於是 Timsort 的策略是首先識別出資料中已排序的子序列（這些子序列稱為 run ），然後透過不斷合併這些 run 來達成全資料範圍的排序。對於隨機資料，最佳的排序效率上限被約束在 O(nlogn)，但即便是在 O(nlogn) 等級的排序演算法中，其效能與理論上的極限值 O(nlogn) 。
之間仍存在差距。Timsort 在處理隨機資料時，表現略遜 mergesort 一籌，但對於部分已排序的資料則有出色的表現。

Timsort 將一組資料分為多組單調遞增/遞減的資料序列，並把每個單調遞增/遞減的資料稱呼為 run ，同時在尋找 run 時若遇到單調遞減的 run 則依需求將其改為單調遞增的 run （假設目標是將全局資料排序為單調遞增，反之則是把遇到的單調遞增 run 改為單調遞減）。由於每次合併時會合併兩個 run ，因此若 run 的數量等於或略小於 2 的冪，合併效率最好；若略大於 2 的冪，效率就會特別低。
為了使合併過程更加均衡， Timsort 還提出了 minrun （每個 run 的最小長度），若 run 的長度太短，就用 insertion sort 將 run 後面的元素插入到前面的 run 中，直到每個 run 的長度都不小於 minrun 。顯然，minrun 的選擇對 Timsort 算法效率有相當程度的影響。

通常來說，若欲排序的資料存於陣列中， minrun 會被設定為 64 ，由於長度 64 以內的資料可放入一條 cacheline 中，因此進行 insertion sort 時可以取得相當好的效能。但是對於 linked-list 而言，其資料並非連續存於記憶體中，因此該如何選擇其他合適的minrun ，甚或是究竟是否需要設定 minrun 也是一個重要的議題。

Timsort 的演算法大致如下:

• 將欲排序資料掃過一次，在掃的同時
  • 把連續遞增/遞減資料視為一組排序好的 run，並依需求將遞增 / 遞減的 run 改為遞減 / 遞增的 run 。
  • 若 run 的長度小於預先設定的數值 minrun ，則將 run 和其他 run 兩兩合併為更大的 run
  • 最後將各個 run 依照順序合併在一起即完成整組資料的排序

針對 linked-list 實做的 Timsort

在 linux 核心中，提供了一個方便的資料結構 struct list_head 和一系列方便的 API 讓使用者可以迅速方便的針對不同的資料結構建立一個 doubly circular linked-list ，詳細介紹可見 你所不知道的 C 語言: linked list 和非連續記憶體

這裡以 linux 核心實作 2024q1 中針對 linux 提供的 doubly circular linked-list 實做的 Timsort 演算法 為例，並討論不同的改善策略。

針對 Timsort 提出的改善

Adaptive ShiversSort

commit 19897f

研讀 Timsort 的相關資料時，發現了 yanjiew1 的實作，裡面提到了 Timsort 的額外兩個變體： Adaptive Shivers Sort 以及 powersort

Adaptive Shivers Sort 改良了 Timsort 的合併策略：假設 X, Y, Z 為目前堆疊上的 Runs ，Z 為堆疊頂部 ，若

$\lfloor {\log }_{2}X\rfloor \le \lfloor {\log }_2(max(Y,Z))\rfloor$ ，則將 X, Y 合併。

我們可以透過 __built_in_clz() 快速計算 log2:

static inline size_t ilog2(size_t x)
{
    return sizeof(unsigned long) * 8 - __builtin_clzl(x) - 1; 
}

修改原本的 merge_collapse 並藉由 ilog2 實作出 adaptive ShiversSort 的合併策略：

-         if ((n >= 3 &&
-              run_size(tp->prev->prev) <= run_size(tp->prev) + run_size(tp)) ||
-             (n >= 4 && run_size(tp->prev->prev->prev) <=
-                            run_size(tp->prev->prev) + run_size(tp->prev))) {
-             if (run_size(tp->prev->prev) < run_size(tp)) {
-                 tp->prev = merge_at(priv, cmp, tp->prev);
-             } else {
-                 tp = merge_at(priv, cmp, tp);
-             }

+         if (n >= 3 && 
+             ilog2(run_size(tp->prev->prev)) <= ilog2(run_size_max(tp->prev, tp))) {
+             tp->prev = merge_at(priv, cmp, tp->prev);
          } else if (run_size(tp->prev) <= run_size(tp)) {
              size_t max_run = run_size(tp);
              if(priv && ((stat_t*)priv)->max_run < max_run)
                  ((stat_t*)priv)->max_run = max_run;
              tp = merge_at(priv, cmp, tp);
          }

PowerSort

Implement of powerSort

PowerSort 是另一種 Timsort 的改良，其主要概念為想像一個平衡的 binary tree 每個節點構成的 subtree 含有多少數量的節點，之後計算出目前要排序的 2 個 run 的中點連線後包含在此區間內的最小節點深度為何，在論文中將 2 個 run 對應的節點深度稱為 power，並根據後者決定是否要合併 。

留意以下說明中起點 index 設為 0，但我後來檢視其實作將起點設為 1 較合理：論文中的實作有 - (left << 1)，若 left 為 1 的話恰好和虛擬碼計算除法時上下同乘 2 吻合（詳見文後的 虛擬碼和參考實作 ）

假設我們現在有長度為 28 的 list 需要排序，我們先想像其對應的 binary tree 為何：

$$\lfloor lo{g}_{2}28\rfloor =4$$

因此我們可以想像一個深度為 4 的 binary tree ，分別有 power 1, 2, 3, 4

假設我們有 6 個 run {a, b, c, d, e, f}，每個 run_size 分別為：

{5, 3, 3, 14, 1, 2}
{a, b, c,  d, e, f}

則我們可以開始想像他們合併後的 power 為何，首先看前兩個 run {a, b} 合併後的 power。
第一個 run {a} 的 index 為 0, 1, 2, 3, 4
第二個 run {b} 的 index 為 5, 6, 7

計算中點的公式為：

$$begin+\lfloor {\displaystyle \frac{end-begin+1}{2}}\rfloor$$

這兩個 run 的中點分別為：

$0+\lfloor {\displaystyle \frac{4-0+1}{2}}\rfloor =2$

$5+\lfloor {\displaystyle \frac{7-5+1}{2}}\rfloor =6$

而這兩點的連線則為 (2, 6] 的區間，留意左側是開區間，不包含 2 ；而右側是閉區間，包含 6 。我們現在依序由淺到深檢視。
不同深度的 node 可表示為 （ k 為深度）：

$$nodes\mathrm{\_}of\mathrm{\_}depth(k)=\{\begin{array}{ll}({\displaystyle \frac{n}{2}}),& k=1\\ nodes\mathrm{\_}of\mathrm{\_}depth(k-1)\pm ({\displaystyle \frac{n}{2^k}}),& k>1\end{array}$$

其中 n 為 list 總長度，在這裡 n 為 28

• 深度 k 為 1
  

  ${\displaystyle \frac{28}{2}}=14$ ，index 14 並未包含在 (2, 6] 的區間。

• 深度 k 為 2
  在 binary tree 中，有兩個深度為二的節點：
  

  ${\displaystyle \frac{28}{2}}\pm {\displaystyle \frac{28}{2^2}}$
  分別是 index 7 和 index 21
  index 7 和 21 皆並未包含在 (2, 6] 的區間。

• 深度 k 為 3
  

  ${\displaystyle \frac{28}{2^3}}=3.5$ ，深度為三的 index 有：
  7 - 3.5 = 3.5
  7 + 3.5 = 10.5
  21 - 3.5 = 17.5
  21 + 3.5 = 24.5
  其中 index 3.5 被包含在 (2, 6] 的區間。
  因此 run {a, b} 之間的 power 就是 3 。

接下來看 run {b, c} 之間的 power 。
c 的座標為 8~10
根據前述計算， run {b} 的中點為 6 ， run{c} 的中點為：

$8+\lfloor {\displaystyle \frac{10-8+1}{2}}\rfloor =9$
而這兩點的連線則為 (6, 9] 的區間。包含深度為 2 的 index 7 ，可知其 power 為 2 。
之後依序計算出每個 run 的中點：

$mid(ru{n}_d)=18$

$mid(ru{n}_e)=25$

$mid(ru{n}_f)=27$

因此

$power(ru{n}_c,ru{n}_d)=1$, ( 9, 18] 涵蓋 14

$power(ru{n}_d,ru{n}_e)=3$, (18, 25] 涵蓋 24.5

$power(ru{n}_e,ru{n}_f)=4$, (25, 27] 不在深度 1 ~ 3 的範圍內，因此為最大深度 4 。

影片 COMP526 (Fall 2023) 3-6 §3.6 Python's sorting method & Powersort 的後半部分以視覺化的方式呈現前述運算。

虛擬碼和參考實作

現在我們檢視 PowerSort 對應的虛擬碼:

procedure Powersort(A[0..n)){
    i := 0; 
    runs := new Stack();
    j := ExtendRunRight(A, i);
    runs.push((i, j), 0); 
    i := j;
    while(i < n){
        j :=  ExtendRunRight(A, i);
        p := power(runs.top(), (i, j), n);
        while(p <= runs.top().power){
            merge topmost 2 runs
        }
        runs.push((i, j), p);
        i := j;
    }
    while (runs.size() >= 2)
        merge topmost 2 runs    
}

procedure power((i1, j1), (i2, j2), n){
    n1 := j1 - i1
    n2 := j2 - i2
    // interval of (a, b]):
    (float) a := (i1 + n1/2 - 1) / n;
    (float) b := (i2 + n2/2 - 1) / n;
    p := 0
    while(1){
        (int) c := round(a * pow(2, p))
        (int) d := round(b * pow(2, p))
        if(c != d)
            return p;
        p := p + 1
    }
}

〈Nearly-Optimal Mergesorts: Fast, Practical Sorting Methods That Optimally Adapt to Existing Runs〉提出能透過 bitwise 運算用 O(1) 時間計算出 power 的作法：

static int nodePower(int left, int right, int startA , int startB , int endB) {
    int twoN = (right - left + 1) << 1; // 2*n
    long l = startA + startB - (left << 1);
    long r = startB + endB + 1 - (left << 1);
    int a = (int) ((l << 31) / twoN);
    int b = (int) ((r << 31) / twoN);
    return Integer.numberOfLeadingZeros(a ^ b);
}

其中 left 和 right 表示要排序的起點和終點，用來計算這次排序的長度為何用以推估 binary tree 的深度 。而我們的排序固定是整個 list ，因此可以將 left, right 用一個變數 len 取代，到時候只要傳入 list 的長度即可。

以下解釋這個程式碼為何和虛擬碼等效，檢視前述虛擬碼，可以看到 a, b 的計算公式為：

a := (i1 + n1/2 - 1) / n;
b := (i2 + n2/2 - 1) / n;

其中 i1, i2 為 list1, list2 的起點，也就是 start_A 和 start_B ； n1, n2 為 list1, list2 的長度； n 為整個要排序的 list 的長度。
我們將計算 a, b 的公式上下同乘 2 ：

a := (2 * i1 + n1 - 2) / 2n;
b := (2 * i2 + n2 - 2) / 2n;

而 a, b 可對應論文實作中的 l, r
注意到 start_B = start_A + run_size(h1) = start_A + n1
則 l = start_A + start_B - (left << 1) 可寫作 l = 2 * start_A + n1 - (left << 1)

留意前面計算中點時假設起點 index 為 0 ，但我後來檢視其實作似乎應該將起點設為 1 較符合其 pseudo code

若 list 起點 index 為 1，則傳入 function 的 left 會是 1 ，所以 l 的計算會變成 l = 2 * start_A + n1 - 2 ，若將其除上 (2*n) 則和虛擬碼中的 a 相等 (倒數第三行) 。
同理， r 的計算結果也和 b 相等，唯注意計算 r 的時候會多一個 +1 ：

 r = startB + endB + 1 - (left << 1);
                   ^^^

之所以會多這個 +1 是因為我們在檢視兩中點連線時其區間為

$(l,r]$ ，右側為閉區間， r 要被包含在區間內，而原本的虛擬碼中的 a, b 是浮點數，但 r 是整數，因此這裡要 +1 確保　ｒ　有被包含在區間內。
而原本虛擬碼的最後面其運算為：

$while(\lfloor a\ast 2^p\rfloor ==\lfloor b\ast 2^p\rfloor )$

$p=p+1$

原本的 a 的計算為 (2 * i1 + n1 - 2) / 2n ，我們現在用 a' 表達 (2 * i1 + n1 - 2) ， b' 表達 (2 * i2 + n2 - 2) 。 則我們要計算的目標為

$min(p)s.t.$

$(a^{\prime }\ast 2^p\ge 2n)or(b^{\prime }\ast 2^p\ge 2n)$

或者可以說

$min(p)s.t.$

${\displaystyle \frac{a^{\prime }\ast 2^p}{2n}}\ge 1or{\displaystyle \frac{b^{\prime }\ast 2^p}{2n}}\ge 1$

而在計算

${\displaystyle \frac{a^{\prime }\ast 2^p}{2n}}$ 時可以視為

${\displaystyle \frac{a^{\prime }\ast 2^p}{2n}}\ast {\displaystyle \frac{2^{31}}{2^{31}}}=a^{\prime }\ast 2^p\ast {\displaystyle \frac{2^{31}}{2n}}÷2^{31}$

而我們的目標是找到最小的 p 值使

$2^p\ast {\displaystyle \frac{a^{\prime }\ast 2^{31}}{2n}}\ge 2^{31}$ ，即使 ((a' << 31) / 2n) & 0x8000_0000 = true 而 p 就可以透過 __built_in_clz((a' << 31) / 2n) 計算得出。 其中 (a' << 31) / 2n 可對照論文實作的 int a = (int) ((l << 31) / twoN)
而之所以最後在 clz 前要讓 (a ^ b)，是因為這樣得到可以找到最小的 p 。

參考論文實作出計算 nodePower 的函式：

static inline size_t nodePower(struct list_head *h1, struct list_head *h2, 
                                size_t start_A, size_t len)
{
    if(!h1 || !h2){
        return 0;
    }
    size_t len_2 = len << 1;
    size_t start_B = start_A + run_size(h1);
    size_t end_B = start_B + run_size(h2);
    size_t l = start_A + start_B;
    size_t r = start_B + end_B + 1;
    int a = (int)((l << 31) / len_2);
    int b = (int)((r << 31) / len_2);
    return __builtin_clz(a ^ b);
}

其中 start_A 表示在該 run 之前 stack 中的 run 總共有多長，也就是該 run 的起始 index 為何。

power sort 使用一個 stack 紀錄待合併的 run 以及這些 run 之間的 power 。 (i.e. stack[top].power = nodePower(stack[top].run, stack[top - 1].run)

當要進入 stack 的 run r_new 和原本 stack top 中的 run_top 計算出來的 power_new 小於目前 stack[top].power 則將 stack 中最上面兩個 run 合併，直到 stack 中的 power 小於 power_new ，將 r_new 和 power_new 推入 stack 中：

power_new = nodePower(run_top, r_new)
while (power < stack[top].power)
    merge(stack[top].run, stack[top-1].run)
stack[top].run = r_new, stack[top].power = power_new;

之所以要確保新進來的 power 小於原本紀錄的 power ，是因為這樣能盡可能的讓串列的長度平衡。
以前述的 run a~f 為例，其 power 依序為

$power(ru{n}_a,ru{n}_b)=3$

$power(ru{n}_b,ru{n}_c)=2$

$power(ru{n}_c,ru{n}_d)=1$

$power(ru{n}_d,ru{n}_e)=3$

$power(ru{n}_e,ru{n}_f)=4$

接著依序將 run 放入 stack 中：
將

$ru{n}_a$ 放入 stack 中， power 設為 0 ，top 更新為

$ru{n}_a$ 。

嘗試放入

$ru{n}_b$ ，power 為 3 > 0 ，將

$ru{n}_b$ 放入 stack ，top 更新為

$ru{n}_b$ 。

嘗試放入

$ru{n}_c$ ，發現

$power(ru{n}_b,ru{n}_c)=2<3$ ， power 變小表示待處理 run 的數量足以構成一個 sub tree ，有機會使 list 更加平衡，將其合併。

合併 stack 最靠近 top 的兩個 run ：

$merge(ru{n}_a,ru{n}_b)$ ，更新 top 指向 power = 0 。

$ru{n}_{ab}\mathrm{長}\mathrm{度}\mathrm{為}8$

目前 stack 中 power = 0 < 2 ，放入

$ru{n}_c$ ，更新 top 的 power 為 2 。

嘗試放入

$ru{n}_d$ ，其 power 為 1 < 2

合併

$(ru{n}_{ab},ru{n}_c)$ ， 更新 top ，指向 power = 0 。

$ru{n}_{abc}\mathrm{長}\mathrm{度}\mathrm{為}11$

目前 stack[top].power = 0 < 1，將

$ru{n}_d$ 放入 stack ，更新 top 的 power 為 1

嘗試放入

$ru{n}_e$，power = 3 > 1 ，直接放入並更新 power

嘗試放入

$ru{n}_f$ ， power = 4 > 3 ，直接放入並更新 power

已無其他待放入 stack 的 run ，從 top 開始往前合併：
合併

$ru{n}_e,ru{n}_f$ ，

$ru{n}_{ef}$ 長度為

$1+2=3$

合併

$ru{n}_{ef},ru{n}_d$ ，

$ru{n}_{def}$ 長度為

$14+3=17$

合併

$ru{n}_{def},ru{n}_{abc}$ ，

$ru{n}_{abcdef}$ 長度為

$11+17=28$

至此便完成了 PowerSort 的排序。
以上合併過程的示意圖如下：

可以看到合併的過程接近一顆 binary tree，如此確保了 balance。

為了實作出 PowerSort，我指定新的 stack 用來紀錄不同 run 間的 power 以及對應的 start_A（紀錄 run 開頭的 index ）：

static size_t pow_stack[100][2] = {0};
// pow_stack[][0] store power
// pow_stack[][1] store start_A

同時對 Timsort 做了以下修改：

     stk_size = 0;
+    size_t start_A = 0;
     struct list_head *list = head->next, *tp = NULL;
     if (head == head->prev)
         return;
 
     /* Convert to a null-terminated singly-linked list. */
     head->prev->next = NULL;

    do {
+        start_A += run_size(tp);
         /* Find next run */
         struct pair result = find_run(priv, list, cmp);
         result.head->prev = tp;
         tp = result.head;
         list = result.next;
-        tp = merge_collapse(priv, cmp, tp);
+        pow_stack[stk_size][1] = start_A;
+        if(stk_size) {
+            size_t tp_power = nodePower(tp->prev,
+            tp, pow_stack[stk_size-1][1], SAMPLES);
+            tp = merge_collapse_power(priv, cmp, tp, tp_power);
+         }
         stk_size++;

而 merge_collapse_power 的實作如下：

static struct list_head *merge_collapse_power(void *priv,
                                        list_cmp_func_t cmp,
                                        struct list_head *tp,
                                        size_t tp_power)
{
    while (pow_stack[stk_size-1][0] > tp_power) {
        tp->prev = merge_at(priv, cmp, tp->prev);
    }
    pow_stack[stk_size][0] = tp_power;
    return tp;
}

TODO: 理解「目前」核心的資料分布

依據資料分布，提出排序演算法的策略
(挑選) 實作，要在 LKM (Linux kernel module) 展現，要關閉 preemption 和 interrupt

以下所有程式除特別提出外一律指 linux kernel v6.5 的程式碼

要改善 list_sort 之前，我們需要先知道他平常會被哪些應用呼叫，如此才能針對那些應用進行分析尋求可能的改進。
在 linux kernel 中簡單搜尋以後，會用到 list_sort 這個 function 的程式大致可分為三類：

• device driver
  • 如 drivers/iommu/dma-iommu.c, drivers/pci/controller/cadence/pcie-cadence-host.c 等等
• file system
  • 如 fs/btrfs/block-group.c, fs/ext4/fsmap.c 等等
• perf tool
  • 如 tools/perf/util/metricgroup.c, tools/perf/util/parse-events.c 等等

檢視 tools/perf/util/metricgroup.c 中的程式碼可發現如下程式片段：

...
    /* Sort metrics from largest to smallest. */
    list_sort(NULL, &metric_list, metric_list_cmp);
...

看起來當使用者使用 perf stat -M 的命令時便能觸發到 list_sort 。由於該命令能可以直接在 user space 使用，因此我打算將其作為第一個分析的應用。

第一步要做的就是先測試是否能透過 perf stat -M 命令觸發到 list_sort ，這裡我使用了 virtme-ng 這個方便的虛擬環境進行實驗。

virtme-ng 是一個方便的工具，它可以讓使用者輕鬆快速地從源代碼開始重新編譯和測試 Linux kernel
由於要測試是否能用 perf stat -M 的命令觸發 list_sort ，基本上要做兩件事情：

1. 修改 list_sort ，讓他在被呼叫時留下紀錄
2. 針對修改過 list_sort 的 kernel 版本編譯出對應的 perf tools

首先將 virtime-ng 安裝到電腦上。這裡是 build from src ，但它也提供透過 apt install 的方式。

 $ sudo apt install python3-pip python3-argcomplete flake8 pylint \
   cargo rustc qemu-system-x86
 $ git clone --recurse-submodules https://github.com/arighi/virtme-ng.git
 $ cd virtme-ng
 $ BUILD_VIRTME_NG_INIT=1 pip3 install --verbose -r requirements.txt .

再來，clone linux from src

git clone git://git.kernel.org/pub/scm/linux/kernel/git/stable/linux-stable.git

之後 checkout 到 v6.5

git checkout v6.5

接著我透過修改 lib/list_sort.c 由於 lib 下的程式不能直接使用 stdio 的函式，因此這裡我用 printk() 的方式讓它印出特定訊息到 kernel ring buffer 中，可透過 dmesg 查看最近被寫入的資料：

void list_sort(void *priv, struct list_head *head, list_cmp_func_t cmp)
{
	struct list_head *list = head->next, *pending = NULL;
	size_t count = 0;	/* Count of pending */

	printk(KERN_INFO "Hello from list_sort\n");
...

再來要將修改過 list_sort 的 kernel 透過 virtme-ng 啟動：

 $ vng --build
 $ vng

預期可看到如下畫面：

          _      _
   __   _(_)_ __| |_ _ __ ___   ___       _ __   __ _
   \ \ / / |  __| __|  _   _ \ / _ \_____|  _ \ / _  |
    \ V /| | |  | |_| | | | | |  __/_____| | | | (_| |
     \_/ |_|_|   \__|_| |_| |_|\___|     |_| |_|\__  |
                                                |___/
   kernel version: 6.5.0-virtme x86_64
   (CTRL+d to exit)

之後我透過撰寫一個非常簡單的 kernel module 直接呼叫 list_sort 並檢視 dmesg命令的輸出，以下 module 的部份程式碼，完整程式碼可見 gist ：

static int __init my_module_init(void) {
    struct my_data *data;
    int i;

    INIT_LIST_HEAD(&my_list);

    for (i = 10; i > 0; i--) {
        data = kmalloc(sizeof(*data), GFP_KERNEL);
        if (!data)
            return -ENOMEM;

        data->value = 10 - i;
        list_add(&data->list, &my_list);
    }

    printk(KERN_INFO "List before sorting:\n");
    list_for_each_entry(data, &my_list, list) {
        printk(KERN_INFO "Value: %d\n", data->value);
    }

    list_sort(NULL, &my_list, cmp);

    printk(KERN_INFO "List after sorting:\n");
    list_for_each_entry(data, &my_list, list) {
        printk(KERN_INFO "Value: %d\n", data->value);
    }

    return 0;
}

準備好對應的 Makefile 之後透過命令

make

編譯出 list_sort_module.ko ，然後透過命令

sudo insmode list_sort_module.ko

載入到虛擬機核心中。
使用 dmesg 命令確認經過前述的改動後，呼叫 list_sort 的確會留下訊息：

$ sudo dmesg | tail -n 25

輸出畫面如下：

[    0.672210] virtme-ng-init: initialization done
[   30.108569] list_sort_module: loading out-of-tree module taints kernel.
[   30.108752] List before sorting:
[   30.108754] Value: 9
[   30.108755] Value: 8
[   30.108755] Value: 7
[   30.108756] Value: 6
[   30.108757] Value: 5
[   30.108757] Value: 4
[   30.108758] Value: 3
[   30.108758] Value: 2
[   30.108759] Value: 1
[   30.108759] Value: 0
[   30.108760] Hello from list_sort
[   30.108761] List after sorting:
[   30.108762] Value: 0
[   30.108762] Value: 1
[   30.108763] Value: 2
[   30.108763] Value: 3
[   30.108764] Value: 4
[   30.108764] Value: 5
[   30.108765] Value: 6
[   30.108766] Value: 7
[   30.108766] Value: 8
[   30.108767] Value: 9

可以看到修改後的 list_sort 在被呼叫時的確能印出訊息。

透過 perf 呼叫 list_sort

接下來嘗試透過命令 perf stat -M '...' 看看能否呼叫到 metricgroup.c 中的 list_sort ，同時我也在 metricgroup.c 中呼叫 list_sort 前加上一行 printf("Called in metricgroup.c!!\n"); 。
之後一樣使用 vng 啟用虛擬環境，並重新編譯 perf ：

自己編譯 perf

要自行編譯 perf 需要用到 libtraceevent 這個 lib ，進入 vng 後，透過以下命令為系統安裝 libtraceevent:

git clone https://git.kernel.org/pub/scm/libs/libtrace/libtraceevent.git
cd libtraceevent
make
sudo make install

開啟 perf_event_paranoid

為了能夠盡可能透過 perf 觀察到系統資訊，我們需要將 perf_event_paranoid 的權限值設為最高：

/proc/sys/kernel/perf_event_paranoid
      The perf_event_paranoid file can be set to restrict
      access to the performance counters.

      2      allow only user-space measurements (default
             since Linux 4.6).
      1      allow both kernel and user measurements
             (default before Linux 4.6).
      0      allow access to CPU-specific data but not raw
             tracepoint samples.
      -1     no restrictions.

透過以下命令將其設為 -1

sudo sysctl kernel.perf_event_paranoid=-1

接下來到 linux/tools/perf 下使用命令

make

編譯過程中會顯示出目前支援的 system features 有哪些：

編譯完成後便能在 tools/perf 目錄下使用命令 ./perf list 查看可用的 metric group ：

metric group

接下來我嘗試透過命令 ./perf stat -M 'ic_fetch_miss_ratio,macro_ops_retired' sleep 1 測試能否呼叫到 list_sort ，然而卻發現 terminal 上雖然有印出

Called in metricgroup.c!!

但是 sudo dmesg 卻沒有印出 Hello from list_sort

研究一段時間後才發現 perf 會呼叫到的 list_sort 並非是 linux/lib/list_sort.c 中實做的 list_sort ，而是 linux/tools/lib/list_sort.c 中實做的 list_sort 。兩者實做上完全相同，linux/tools/lib/list_sort.c 相當於是的副本。然而，兩者有一個很大的差別在於後者可以使用 stdio ，因此我直接在 linux/tools/lib/list_sort.c 加上 printf("Hello from list_sort in tools\n"); 。

之後再次執行命令 ./perf stat -M 'ic_fetch_miss_ratio,macro_ops_retired' sleep 1 ，可以看到如下輸出：

Called in metrics!!
Hello from list_sort in tools
Hello from list_sort in tools
Hello from list_sort in tools

 Performance counter stats for 'sleep 1':

           17,3349      ic_tag_hit_miss.instruction_cache_miss #     11.0 %  ic_fetch_miss_ratio    
          156,9287      ic_tag_hit_miss.all_instruction_cache_accesses                                      
          318,7990      ex_ret_ops                       # 3187990.00 macro_ops_retired       

       1.048086081 seconds time elapsed

       0.000000000 seconds user
       0.040197000 seconds sys

顯然，在使用該命令時呼叫了三次 list_sort ，而在 perf 中，有用到 list_sort 的

分別為（照呼叫順序）：
tools/perf/util/metricgroup.c 中的

static int parse_groups(struct evlist *perf_evlist,
			const char *pmu, const char *str,
			bool metric_no_group,
			bool metric_no_merge,
			bool metric_no_threshold,
			const char *user_requested_cpu_list,
			bool system_wide,
			struct perf_pmu *fake_pmu,
			struct rblist *metric_events_list,
			const struct pmu_metrics_table *table){
    ...
    /* Sort metrics from largest to smallest. */
	list_sort(NULL, &metric_list, metric_list_cmp);
}

以及
tools/perf/util/parse-events.c 中的

static int parse_events__sort_events_and_fix_groups(struct list_head *list)
{
    ...
	/* Sort events. */
	list_sort(&force_grouped_idx, list, evlist__cmp);
}

其中後者被呼叫了兩次。

先看到 tools/perf/util/metricgroup.c 中 parse_groups 呼叫 list_sort() 的場景，可以看到它呼叫 list_sort 時如下：

static int parse_groups(...){
    ...
    list_sort(NULL, &metric_list, metric_list_cmp);
}

顯然，可以透過 metric_list_cmp 知道如何將 metric_list 中要排序的資料解析出來，而 metric_list_cmp 的實做如下：

/**
 * metric_list_cmp - list_sort comparator that sorts metrics with more events to
 *                   the front. tool events are excluded from the count.
 */
static int metric_list_cmp(void *priv __maybe_unused, const struct list_head *l,
			   const struct list_head *r)
{
	const struct metric *left = container_of(l, struct metric, nd);
	const struct metric *right = container_of(r, struct metric, nd);
	struct expr_id_data *data;
	int i, left_count, right_count;

	left_count = hashmap__size(left->pctx->ids);
	perf_tool_event__for_each_event(i) {
		if (!expr__get_id(left->pctx, perf_tool_event__to_str(i), &data))
			left_count--;
	}

	right_count = hashmap__size(right->pctx->ids);
	perf_tool_event__for_each_event(i) {
		if (!expr__get_id(right->pctx, perf_tool_event__to_str(i), &data))
			right_count--;
	}

	return right_count - left_count;
}

將其作為指引撰寫出如下函式：

static void dump_metric_list_ele(const struct list_head *head){
	struct metric *metric_ele;
	struct expr_id_data *data;
	// = container_of(l, struct metric, nd);
	list_for_each_entry(metric_ele, head, nd){
		int cnt = hashmap__size(metric_ele->pctx->ids);
		int i;
		perf_tool_event__for_each_event(i) {
			if (!expr__get_id(metric_ele->pctx, perf_tool_event__to_str(i), &data))
				cnt--;
		}
		printf("%d\n" ,cnt);
	}
}

之後將其至於 parse_groups 呼叫 list_sort() 的前後：

static int parse_groups(...){
    ...
	printf("Metric_list Before sort:\n");
	dump_metric_list_ele(&metric_list);
	/* Sort metrics from largest to smallest. */
	list_sort(NULL, &metric_list, metric_list_cmp);

	printf("Metric_list After sort:\n");
	dump_metric_list_ele(&metric_list);
}

再次依前面的步驟重啟 vng ，編譯好 perf 和相關所需的 lib ，之後到 tools/perf 目錄下執行以下命令：
./perf stat -M 'ic_fetch_miss_ratio,macro_ops_retired' sleep 1
此時可看到以下輸出

Called in metrics!!
Metric_list Before sort:
1
2
Hello from list_sort in tools
Metric_list After sort:
2
1

如此我們便能簡易的將 parse_groups 要排序的資料 dump 出來以利後續分析。

除了 parse_groups 外，前面也有提到在 tools/perf/util/parse-events.c 中的 parse_events__sort_events_and_fix_groups 也有呼叫 list_sort :

static int parse_events__sort_events_and_fix_groups(struct list_head *list)
{
    ...
	/* Sort events. */
	list_sort(&force_grouped_idx, list, evlist__cmp);
    ...
}

參考 evlist_cmp 實做出對應的 dump_event_list_ele ：

static void dump_event_list_ele(void *_fg_idx, const struct list_head *head){
	struct list_head *cur;
	list_for_each(cur, head){
		const struct perf_evsel *h_core = container_of(cur, struct perf_evsel, node);
		const struct evsel *hev = container_of(h_core, struct evsel, core);
		int *force_grouped_idx = _fg_idx;
		int h_sort_idx;
		const char *hname;
		int last_cmp;

		// First compare index
		if(h_core->leader != h_core || h_core->nr_members > 1){
			h_sort_idx = h_core->leader->idx;
		} else
		{
			h_sort_idx = *force_grouped_idx != -1 && arch_evsel__must_be_in_group(hev)
				? *force_grouped_idx
				: h_core->idx;
		}
		// 2nd compare PMU's name
		hname = hev->group_pmu_name;

		// 3rd compare arch specific order
		last_cmp = hev->core.idx;
		printf("%d %s %d\n", h_sort_idx, hname, last_cmp);
	}
}

並將其置於 parse_events__sort_events_and_fix_groups 中：

static int parse_events__sort_events_and_fix_groups(...){
    ...
	/* Sort events. */
	printf("Called in parse-events!!\n");
	printf("Before list_sort:\n");
	dump_event_list_ele(&force_grouped_idx, list);
	list_sort(&force_grouped_idx, list, evlist__cmp);
	printf("After list_sort:\n");
	dump_event_list_ele(&force_grouped_idx, list);
    ...
}

再次重啟 vng ，編譯好 perf 和相關所需的 lib ，之後到 tools/perf 目錄下執行以下命令：
./perf stat -M 'ic_fetch_miss_ratio,macro_ops_retired' sleep 1

可見到如下輸出：

Called in metrics!!
Metric_list Before sort:
1
2
Hello from list_sort in tools
Metric_list After sort:
2
1
Called in parse-events!!
Before list_sort:
0 cpu 0
0 cpu 1
Hello from list_sort in tools
After list_sort:
0 cpu 0
0 cpu 1
Called in parse-events!!
Before list_sort:
0 cpu 0
Hello from list_sort in tools
After list_sort:
0 cpu 0

不知為何 parse-events 被呼叫了兩次，以及為何第二次要排序的 event 數量會變少。

將先前的 dump_metric_list_ele 改為不同 ele 間印的是空格而非空行並重新編譯 perf ，
現在我們試試看把所有可以監測的 metrics 都打開 ：
( 完整命令可見 gist )

./perf stat -M 'ic_fetch_miss_ratio,macro_ops_retired,...' sleep 1

可以看到如下的輸出畫面：

Metric_list Before sort:
1 1 2 1 1 3 3 2 1 1 1 1 3 1 4 16 4 8 2 5 4 2 3 2 1 1 1 2 1 1 3 3 2 1 2 1 1 1 1 1 3 1 1 3 2 4 1 1 1 1 1 1 1 1 1 1 1 2 3 1 4 4 16 16 4 4 16 8 16 12 12 12 12 2 5 5 4 4 3 3 2 3 5 5 4 4 2 2 2 3 2 1 1 2 1 2 

Hello from list_sort in tools
Metric_list After sort:
16 16 16 16 16 12 12 12 12 8 8 5 5 5 5 5 4 4 4 4 4 4 4 4 4 4 4 4 3 3 3 3 3 3 3 3 3 3 3 3 3 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 

Called in parse-events!!
Before list_sort:
Hello from list_sort in tools
After list_sort:
event syntax error: '{remote_socket_inf1_outbound_data_beats_ccm4/metric-id=remote_socket_inf1_outbound..'
                     \___ Cannot find PMU `remote_socket_inf1_outbound_data_beats_ccm4'. Missing kernel support?
...

顯然， stat 中部份的 metrics 無法被 event-parse 解析，但這不妨礙我們 dump 出要被排序的 stats 資料分佈為何。

另一方面，透過 ./perf list 可以看到一系列可用的 events ，嘗試透過 perf 追蹤特定的 events

./perf stat -e 'branches, branch-misses' sleep 1

可以見到如下輸出

Called in parse-events!!
Before list_sort:
0 cpu 0
1 cpu 1
Hello from list_sort in tools
After list_sort:
0 cpu 0
1 cpu 1

從輸出看來， event 也有機會作為觀察 kernel 資料分佈的重要來源。

接下來介紹 stress-ng ，這是一個壓力測試軟體，可以用來簡單製造如 disk 滿載， cpu 滿載的情況，例如以下命令：

stress-ng --cpu 12 --timeout 180

該命令會產生 workload 將 12 thread 的佔用率佔滿

也可以指定每個 thread 的佔用為何：

stress-ng --cpu 12 --cpu-load 60 --timeout 100

如此，便可將 perf 搭配 stress-ng 製造不同的 workload 來並取得不同 workload 下的 dataset 。

然而試了幾個不同的 workload 後，我發現資料分佈的情形始終如一，經過一系列實驗後，最終確認了：
perf 中透過 ./perf stat -M 'ic_fetch_miss_ratio,macro_ops_retired,…' 呼叫到 list_sort 時，該排序僅按照使用者輸入的 metric list （即 'ic_fetch_miss_ratio,macro_ops_retired,…' ）順序進行排序，和最終測得的數值無關，也就是無論是何種 workload ，只要使用者輸入的 metric list 相同（順序及種類都需相同），那最後 dump 出來的資料分佈就會相同。
顯然，僅透過 perf 無法取得足夠多的資料分佈。既然如此，我需要尋找其他的資料分佈來源，而先前有提到， kernel 中使用到 list_sort 的地方除了 perf 外還有 device_driver 以及 filesystem ，其中 fs/ext4/fsmap.c 吸引了我的目光，我手邊正好有一些檔案系統為 ext4 的隨身碟，若能找到方法觸發到裡面用到 list_sort 的函式，想必能作為一個資料分佈的來源。

透過 ext4_fsmap 取得資料分佈

檢視 fs/ext4/fsmap.c ，發現裡面用到 list_sort 的函式叫做 ext4_getfsmap_find_fixed_metadata ，其中，它會將取得的 meta_list 傳入 list_sort ，並依 fmr_physical 的大小進行排序。
順著 ext4_getfsmap_find_fixed_metadata 繼續找，最終可以找到如下的函式呼叫順序圖：

經過一系列的資料搜尋，我最終發現可以透過以下 函式：
int ioctl(int fd, FS_IOC_GETFSMAP, struct fsmap_head * arg);
搭配適當的 arg 去呼叫到 ext4_getfsmap 。

參考 e2freefrag.c ，我撰寫了一個可以讓使用者輸入 ext4 裝置路徑並回傳 fsmap 內容的程式

使用命令 gcc ext4_fs_map.c -o ext4_fs_map 編譯。
再來將手邊檔案系統為 ext4 的隨身碟插入電腦，使用命令查看目前電腦上的裝置：

ls /dev/ | grep sd

預期可見到如下輸出：

sda

將其 mount 到電腦上：

sudo mkdir /media/tmp_ext4
sudo mount -t ext4 /dev/sda /media/tmp_ext4

之後使用剛剛編譯好的 ext4_fs_map 查看該目錄

./ext4_fs_map /media/tmp_ext4

可見到如下輸出：

Open fd = /media/tmp_ext4
Physical layout of /media/tmp_ext4:
Physical Block   Length           Offset           Flags           
0                4096             0                10              
4096             4096             0                10              
8192             4190208          0                10              
4198400          65536            0                10              
4263936          65536            0                10              
4329472          33488896         0                10              
37818368         96399360         0                10              
134217728        4096             0                10              
134221824        4096             0                10              
134225920        4190208          0                10              
138416128        37842944         0                10              
176259072        226394112        0                10              
402653184        4096             0                10              
...
12893405184      444481536        0                30 

如此我們便透過 ioctl 順利呼叫到 ext4_getfsmap ，並最終呼叫到 list_sort 。但是我們想要取得資料分佈的話還要修改 kernel 才行：
修改 fs/ext4/fsmap.c ， 參考 ext4_getfsmap_compare 新增函式 show_ext4_frPhySize ：

static int ext4_getfsmap_compare(void *priv,
				 const struct list_head *a,
				 const struct list_head *b)
{
	struct ext4_fsmap *fa;
	struct ext4_fsmap *fb;

	fa = container_of(a, struct ext4_fsmap, fmr_list);
	fb = container_of(b, struct ext4_fsmap, fmr_list);
	if (fa->fmr_physical < fb->fmr_physical)
		return -1;
	else if (fa->fmr_physical > fb->fmr_physical)
		return 1;
	return 0;
}

void show_ext4_frPhySize(const struct list_head *meta_list){
	struct ext4_fsmap *entry;
	list_for_each_entry(entry, meta_list, fmr_list){
		pr_info("%llu ", entry->fmr_physical);
	}
}

並在函式 ext4_getfsmap_find_fixed_metadata 中使用剛剛新增的函式將資料分佈輸出到 dmesg ：

        /* Sort the list */
+       pr_info("ext4 list before sort:\n");
+       show_ext4_frPhySize(meta_list);
        list_sort(NULL, meta_list, ext4_getfsmap_compare);
+       pr_info("ext4 list After sort:\n");
+       show_ext4_frPhySize(meta_list);

修改後，我原本嘗試透過 vng 虛擬環境進行測試，但我發現在 vng 虛擬環境中的檔案系統都會變為 tmpfs ，無法順利透過先前編譯的 ext4_fs_map 讀取隨身碟中的資訊。因此我打算編譯完整的 kernel 並直接透過硬體執行。

編譯完整的 Linux kernel

要編譯完整的 linux kernel ，需要先安裝所需套件：

sudo apt-get install git fakeroot build-essential ncurses-dev xz-utils libssl-dev bc flex libelf-dev bison

再來，為了避免手動選擇電腦驅動的麻煩，這裡直接將原本電腦上執行的 ubuntu 系統之 config 拿來用：

cd linux/
cp -v /boot/config-$(uname -r) .config

之後使用命令

make -j$(nproc)

進行編譯，編譯過程中遇到如下錯誤

No rule to make target 'debian/canonical-certs.pem

使用命令

scripts/config --disable SYSTEM_TRUSTED_KEYS
scripts/config --disable SYSTEM_REVOCATION_KEYS

之後再次使用 make 命令編譯即可。

編譯後使用如下命令安裝所需模組

sudo make -j$(nproc) modules_install

最後透過命令

sudo make install

更新 bootloader ，至此便完成 kernel 的編譯與安裝。接下來只要重啟電腦並在 grub 選單中選擇剛剛編譯好的 kernel 即可。

重啟之後，我透過先前編譯好的 ext4_fs_map 將我手邊為 ext4 檔案系統的儲存裝置都掃了一遍，並透過命令 sudo dmesg 檢視結果：

...
[  280.235055] Called in ext4_getfsmap!!
[  280.235064] ext4 list before sort:
[  280.235064] 0 
[  280.235065] 1 
[  280.235066] 2 
[  280.235066] 246 
[  280.235067] 262 
[  280.235068] 278 
[  280.235068] 32768 
[  280.235069] 32769 
[  280.235069] 32770 
[  280.235070] 247 
[  280.235071] 263 
[  280.235071] 767 
...

如此便順利取得了一系列透過 ext4_getfsmap 得到的資料分佈。
另外，我也一併修改了 driver/iommu/dma-iommu.c 以及 fs/btrfs/tree-log.c 藉此取得更多資料分佈。

dma-iommu

一開始我嘗試透過 list_for_each_entry 將裡面的資料 dump 出來：

static void dump_mmio_data(struct list_head *head){
	struct resource_entry *r_ent;
	list_for_each_entry(r_ent, head, node){
		pr_info("%llu ", r_ent->res->start);
	}
}

然而卻出現如下畫面

...
[    0.315310] pci 0000:0d:00.4: Adding to iommu group 28
[    0.315323] pci 0000:0e:00.0: Adding to iommu group 29
[    0.315364] Called in iommu.c!
[    0.315366] Data before sort:
[    0.315368] Hello from list_sort
[    0.315369] Data after sort:
[    0.315425] Called in iommu.c!
[    0.315428] Data before sort:
[    0.315429] Hello from list_sort
[    0.315430] Data after sort:
...
...

起初我以為是我印出東西的方式有錯，不過為了保險起見我加了一個 cnt 變數讓他能夠印出 list 中的 element 數量：

static void dump_mmio_data(struct list_head *head){
	struct resource_entry *r_ent;
+	int cnt = 0;
	list_for_each_entry(r_ent, head, node){
		pr_info("%llu ", r_ent->res->start);
+		cnt++;
	}
+	pr_info("%d items in mmio list", cnt);
}

之後重新編譯 kernel 並重啟電腦，使用 sudo dmesg 命令查看：

...
[    0.315338] pci 0000:0d:00.4: Adding to iommu group 28
[    0.315351] pci 0000:0e:00.0: Adding to iommu group 29
[    0.315392] Called in iommu.c!
[    0.315394] Data before sort:
[    0.315395] 0 items in mmio list
[    0.315396] Hello from list_sort
[    0.315399] Data after sort:
[    0.315400] 0 items in mmio list
...

至此便確定 iommu 的 list_sort 雖會被呼叫多次，但由於 list 中沒有 element ，因此無法取得有效的資料分佈。

tree-log

一樣透過 list_for_each_entry 將呼叫 list_sort 前的資料 dump 出來：

static void dump_extents(const struct list_head *head){
	struct extent_map *em_entry;
	int cnt = 0;
	list_for_each_entry(em_entry, head, list){
		pr_info("%llu ", em_entry->start);
		cnt++;
	}
	pr_info("%d items in brtfs_extents list", cnt);
}

在呼叫 list_sort 的前後呼叫剛剛新增的函式：

static int btrfs_log_changed_extents(...){
...
    +pr_info("btrfs_log_changed_extents");
    +pr_info("extents list before sort");
    +dump_extents(&extents);
    list_sort(NULL, &extents, extent_cmp);
    +pr_info("extents list after sort");
    +dump_extents(&extents);
...
}

之後透過 dmesg 查看資料分佈：

[    8.330219] btrfs_log_changed_extents
[    8.330220] extents list before sort
[    8.330617] 0 items in brtfs_extents list
[    8.330715] systemd[1]: Started Journal Service.
[    8.331000] Hello from list_sort
[    8.331000] extents list after sort
[    8.331001] 0 items in brtfs_extents list
[    8.336609] systemd-journald[535]: Received client request to flush runtime journal.
...
[    8.370147] extents list before sort
[    8.370148] 13471744 
[    8.370148] 13393920 
[    8.370149] 13344768 
[    8.371087] 13271040 
[    8.372007] 12902400 
[    8.372007] 12570624 
[    8.372008] 12419072 
[    8.372008] 12267520 
[    8.372009] 8581120 
[    8.372009] 8564736 
[    8.373902] 8396800 
[    8.373902] 11 items in brtfs_extents list
...
[   89.596783] btrfs_log_changed_extents
[   89.596785] extents list before sort
[   89.596786] 0 items in brtfs_extents list
[   89.596787] Hello from list_sort
[   89.596788] extents list after sort
...

可以看到雖然從 btrfs_log_changed_extents 呼叫 list_sort 時有時 list 會是空的，但也有能取得資料的時候。不過他的 list 長度普遍不長，目前我能蒐集到最長的資料分佈長度僅為 13 。

透過 XFS 取得資料分佈

有了前述透過 ext4 fs, brtfs 取得資料分佈的經驗後，我留意到 XFS 中有大量地方也使用到 list_sort ，經過一系列的實驗後，我最終順利取得

xfs_extent_busy_sort, xlog_cil_push_work, xfs_trans_run_precommits 以及 xfs_buf_delwri_submit_buffers 呼叫 list_sort 前的資料分佈，其中這些 function 都可藉由反覆寫入/刪除大量檔案到 xfs 的檔案系統中手動觸發，因此能方便的取得很多筆資料以便進行實驗。

我仔細檢視了剩下 kernel 中呼叫到 list_sort 的地方，能透過我手邊裝置取得資料分佈的來源就是前述這幾個地方。

TODO: 整合 Timsort 和其改進到 Linux 核心

參照 Refining Data Structure Implementations in the Linux Kernel for Improved Performance

取得 kernel 的資料分布後，下一步就是將 timsort 和其衍生的改良演算法實作到 kernel 中進行實驗，用以和原本 kernel 中實作的 list_sort 進行比較其優劣。

要將 timsort 和其衍生的改良演算法實作到 kernel 中，可以透過撰寫 kernel module 的方式達成。為了方便測試，我將一系列 list_sort algs 實做為一個 kernel module 加載到 kernel 中，並透過 EXPORT_SYMBOL 這個方便的 MACRO 將實做好的一系列 algs function 導出，以便在其他 module 中方便使用。

EXPORT_SYMBOL 的用法如下：
假設有兩個 module A, B
module_A 中實做了 function foo()

foo(){
    ...
}
EXPORT_EYMBOL(foo);

之後在 module_B 中便可使用 foo() 來呼叫它

extern foo();
...

如此，只要確保 module_A 事先運作於核心中，便能透過 module_B 使用到 module_A export 出來的 foo() function 。
另外在編譯 module_B 時的 Makefile 也要加上 path/to/module_A/Module.symvers ，否則會編譯錯誤。

我撰寫了一個 module list_impls ，將其編譯好後透過 insmod 命令使其運作於 kernel 中，並透過 EXPORT_SYMBOL 將 timsort, adaptive_ShiversSort, 以及 power_sort export 出來，之後我們便能透過另外撰寫的 kernel module 呼叫到這些不同的演算法進行比較。其中 power_sort 較為特別，由於其運作上需要事先知道整個 list 的長度，因此這裡額外 export 一個 function EXPORT_SYMBOL(list_impl_set_data_len) 用其來設定整個 list 的長度。

透過另外撰寫的 list_sort.c 便能方便的進行實驗。由於前面從 kernel 中取得的資料分佈每一筆最大也不超過 1KB ，因此這裡是用取巧的方式將蒐集的資料以 global int array 的方式直接包含在程式中。此外，由於蒐集到的資料長度不長，因此實驗進行方式是紀錄對同一筆資料集排序數次（暫定為 4000 次）後的時間加總，透過 ktime_get 取得經過的時間。
同時，為了避免 interrupt 以及 preemption 對實驗造成影響，因此可以透過。

local_irq_disable(); // disable local interrupt
get_cpu();           // disable preemption
// Experiments
local_irq_enable();  // enable local interrupt
put_cpu();           // enable preemption

的方式將 interrupt 以及 preemption 關閉。

經過上述的準備後，我們便能開始進行實驗了。

實驗環境

$ lscpu
Architecture:            x86_64
  CPU op-mode(s):        32-bit, 64-bit
  Address sizes:         48 bits physical, 48 bits virtual
  Byte Order:            Little Endian
CPU(s):                  12
  On-line CPU(s) list:   0-11
Vendor ID:               AuthenticAMD
  Model name:            AMD Ryzen 5 7600X 6-Core Processor
    CPU family:          25
    Model:               97
    Thread(s) per core:  2
    Core(s) per socket:  6
    Socket(s):           1
    Stepping:            2
    CPU max MHz:         5453.0000
    CPU min MHz:         400.0000
    BogoMIPS:            9382.19

$ uname -r
6.5.0-41-generic

透過先前撰寫的 kernel module 進行實驗

$ git clone [e-mail]:fennecJ/list_sort_kernel_module.git
$ cd list_sort_kernel_module
$ cd list_impls
$ make
$ cd ..
$ make
$ sudo insmod list_impls/list_impls.ko
$ sudo insmod list_sort.ko
$ sudo rmmod list_sort

這裡要注意一定要先 list_impls.ko 載入到 kernel 中，再載入 list_sort.ko ，否則會出現以下 error

insmod: ERROR: could not insert module list_sort.ko: Unknown symbol in module

依照前述命令將 list_sort.ko 載入到 kernel 並移除後，預期可透過 sudo dmesg 看到如下結果

[157624.767944] Start test on test_file = datasets/ext4_27G.h
[157624.936833] List done sort with timsort, iter = 4000, data_len = 654:
[157624.936836] Comparison cnt: 17380000
[157624.936836] elapsed time: 161458190 ns
[157625.087793] List done sort with adaptive_ShiversSort, iter = 4000, data_len = 654:
[157625.087796] Comparison cnt: 16696000
[157625.087796] elapsed time: 144081547 ns
[157625.243996] List done sort with powerSort, iter = 4000, data_len = 654:
[157625.243999] Comparison cnt: 16676000
[157625.244000] elapsed time: 149375939 ns
[157625.403735] List done sort with listSort, iter = 4000, data_len = 654:
[157625.403738] Comparison cnt: 16236000
[157625.403739] elapsed time: 152875339 ns
[157628.870676] Module exit, memory freed.

以上實驗均在關閉 preemption 以及 interrupt 的前提下進行，實驗主要紀錄了 資料集為何 ， 排序時呼叫 cmp() 的次數 ， 排序經過的時間 這三項指標。

實驗結果：

由於 dataset 眾多不便呈現，因此我將剩下的實驗結果整理成 google 試算表，依照資料集分類於不同頁面，並將 連結 放置於此。

這裡討論我嘗試的資料，用來進行實驗的資料可大致分為幾個分類：

• ext4 檔案系統的 ext4_getfsmap
• btrfs 檔案系統的 extent_list
• 透過 perf stat -M 照順序輸入所有 metric 的 perf_metric_list
• xfs 檔案系統的 xfs_buf_delwri_submit_buffers, xlog_cil_push_work, xfs_trans_run_precommits 以及 xfs_extent_busy_sort

以下圖片中除特別提，否則：
圖中藍色柱為排序所需時間相比 list_sort 所佔的百分比變化量之加權 (依排序資料量加權） 平均，負值 越大越好（表節省越多時間）
而粉色柱為排序所需比較次數相比 list_sort 所佔的百分比變化量之加權 (依排序資料量加權） 平均，負值 越大越好（表節省越多次比較）

先來看 ext4 檔案系統的實驗結果：

可以看到除了 powersort 的總體加權時間較原本的 list_sort 有改善外，其餘指標針對 ext4_getfsmap 的整體表現不如 list_sort 。

接著來看 brtfs 檔案系統的實驗結果：

可以看到 timsort 和其衍生算法表現皆很不錯，其中 powerSort 表現最佳，相比 list_sort 可節省 53.18% 的時間。然而 powerSort 本身需要事先知道總排序串列長度才可正常運作，考量到這點且其相比 adaptive_ShiverSort 表現差距不大，因此 adaptive_ShiverSort 會是相對好的選擇。

再來看透過 perf stat -M 照順序輸入所有 metric 後得到的資料分佈排序的結果：

這裡有個很神奇的現象， timsort 和其衍生算法相對 list_sort 的比較次數皆較多，可是執行時間皆較少。目前尚未找到造成此現象的原因。原先我以為造成此現象的原因是因為在 list_sort 中如下的程式碼片段所造成：

static void merge_final(void *priv, list_cmp_func_t cmp, struct list_head *head,
			struct list_head *a, struct list_head *b)
{
	...
	u8 count = 0;
    ...
	do {
		/*
		 * If the merge is highly unbalanced (e.g. the input is
		 * already sorted), this loop may run many iterations.
		 * Continue callbacks to the client even though no
		 * element comparison is needed, so the client's cmp()
		 * routine can invoke cond_resched() periodically.
		 */
		if (unlikely(!++count))
			cmp(priv, b, b);
		b->prev = tail;
		tail = b;
		b = b->next;
	} while (b);
    ...

在 list_sort 中的 merge_final ，為了增強系統的即時性，每 256 次合併便會透過一次無意義的 cmp 觸發 cond_resched() 使 cpu 根據情況排程。
但我嘗試在 timsort 和其衍生算法中也引入相同的機制，最後測得的數據仍有前述排序時比較次數較多可是所花時間較少的現象。
目前尚未知道具體原因。

user space 的 list_sort 實做考量

順帶一題，前面提到 linux/tools/lib/list_sort.c 可視為 linux/lib/list_sort.c 的副本，其內容和 linux/lib/list_sort.c 相同，當然也有每 256 次呼叫一次無意義 cmp 的行為。然而，邱冠維同學指出 linux/tools/lib/list_sort.c 只會被 user space 下的 perf 程式所使用，而在 user space 下不需要每 256 次呼叫一次無意義 cmp 的行為。

詳細討論可見 [PATCH] tools/lib/list_sort: Remove redundant code for cond_resched handling

最後來看一系列 XFS 檔案系統相關的資料分佈實驗結果

xfs_buf_delwri_submit_buffers

可以看到對於 xfs_buf_delwri_submit_buffers 會排序的資料， timsort 和其衍生算法總體表現不如 list_sort 。

xlog_cil_push_work

xlog_cli 相關的資料集是我收集最多的資料集，同時它也是 xfs 系列中最容易被觸發的資料收集來源，使用頻率非常高，只要寫入或刪除一定大小的檔案到 xfs 檔案系統的目錄中就會用到。這裡收集的資料分佈之 list 中所含資料量涵蓋範圍短從 11 ，長至 4755 ，應有一定的代表性能反應 kernel 中實際的資料分佈情形。

• 加權平均：
  

  
  可以看到 timsort 和其衍生算法非常擅長處理 xlog_cli 相關的資料集，無論是哪種算法相對 list_sort 皆能節省可觀的時間和排序比較次數。其中由 adaptive_Shiver sort 表現最好。

• 最長的資料集：
  

  
  目前收集到的最長資料集長度為 4755 ，可以看到 timsort 和其衍生算法帶來的效能提昇非常顯著。

xfs_trans_run_precommits

透過 xfs_trans_run_precommits 取得的資料分佈長度甚短，且含有大量無法排序之資料，原本我預期 timsort 和其衍生算法無法佔到什麼便宜。
至於無法排序之資料，原 kernel 中的 cmp 程式是將不可排序之資料都放到同一側，若比較的兩邊皆為不可排序之資料則不改動相對順序。這裡為了方便在遇到不可排序之資料時自動視為整數最大值，因為這樣也可以達到「不可排序之資料都放到同一側，若比較的兩邊皆為不可排序之資料則不改動相對順序」的效果。

雖然我預期 timsort 和其衍生算法無法佔到什麼便宜，但卻可以看到 timsort 和其衍生算法表現不俗。探究其原因，可以直接拿資料集來看：

//data_set 1
int datas[] = {
2147483647,
2147483647,
2147483647,
33875550,
33875539,
2147483647,
};
//data_set 2
int datas[] = {
33875539,
33875550,
2147483647,
2147483647,
2147483647,
2147483647
};

可以看到，由於存在大量不可排序之資料，且這裡將其設為整數最大值，最終取得的資料集已經是十分接近排序完成的型態了。因而使用 timsort 和其衍生算法時能有不錯的效能表現。

xfs_extent_busy_sort

這個資料集是最特別的一個資料集，原因在於其 cmp function 會比較兩個鍵值，當第一個鍵值不同時，直接以第一個鍵值進行比較，反之則以第二個鍵值進行比較。以下為參考虛擬碼：

int cmp (void *priv, const struct list_head *l, const struct list_head *r){
    if(l->key1 == r->key1)
        return l->key2 - r->key2; 
    return l->key1 - r->key1;
}

因此它 dump 出來的資料是二維的資料，我也為此特別寫了一系列 2D 的 function 。

可以看到 timsort 和其衍生算法時都能有不錯的效能表現，其中 adaptive_ShiverSort 表現最佳。

結論

在不同的資料分佈中， XFS 中有部份應用非常適合由 timsort 和其衍生算法處理，我認為可以將 adaptive_ShiverSort 直接實作到 list_sort.c 中，讓使用者依據情況選擇要使用原本的 list_sort 或是 adaptive_ShiverSort 。
而 powerSort 雖有部份應用表現最佳，然而其須事先知道 list 總長的代價較高，不如 adaptive_ShiverSort 容易實作。

Reference

linux 核心實做 2024q1
Timsort 研究與對 Linux 核心貢獻嘗試
Adaptive Shivers Sort: An Alternative Sorting Algorithm
COMP526 (Fall 2023) 3-6 §3.6 Python's sorting method & Powersort
Nearly-Optimal Mergesorts: Fast, Practical Sorting Methods That Optimally Adapt to Existing Runs
vfs/xfs/ext4: GETFSMAP support
ioctl_getfsmap(2) — Linux manual page
Proper Locking Under a Preemptible Kernel: Keeping Kernel Code Preempt-Safe
Linux 核心設計: Timer 及其管理機制
I/O access and Interrupts
ktime accessors
How to Build Linux Kernel From Scratch {Step-By-Step Guide}