---
# System prepended metadata

title: 'linux2023: vax-r'
tags: [linux2023]

---

## 測驗 $\alpha$
在s-tree被初始化之後先隨意插入100個數值
接著依照ascending order印出來
打印的函式為`__treeint_dump` 因為它也是BST所以要依照ascending order走訪的話只需要從root開始進行inorder traversal
```c
/* ascending order */
static void __treeint_dump(struct st_node *n, int depth)
{
    if (!n)
        return;

    __treeint_dump(FFFF, depth + 1);

    struct treeint *v = treeint_entry(n);
    printf("%d\n", v->value);

    __treeint_dump(GGGG, depth + 1);
}
```
所以`FFFF`會是`n`的左子樹, `GGGG`是`n`的右子樹
使用原本就定義好的巨集`st_left`和`st_right`來取得`n`的左子樹和右子樹
`FFFF`:`st_left(n)`
`GGGG`:`st_right(n)`

剩下五個問題都在`st_remove`當中
依照註解的敘述分別會是
`AAAA`:`st_replace_right(del, least)`
`BBBB`:`st_update(root, st_first(least))`
`CCCC`:`st_replace_left(del, most)`
`DDDD`:`st_update(root, st_last(most))`
`EEEE`:`st_update(root, parent)`
`FFFF`:`st_left(n)`
`GGGG`:`st_right(n)`
## 測驗 $\alpha - 1$
### Insert
插入tree node主要由兩個函式負責
首先使用`treeint_insert`
接著實際插入節點的動作由`st_insert`進行
**treeint_insert**
```c
struct treeint *treeint_insert(int a)
{
    struct st_node *p = NULL;
    enum st_dir d;

    // basic insertion step
    for (struct st_node *n = st_root(tree); n;) {
        struct treeint *t = container_of(n, struct treeint,
                                         st_n);  // get the address of st_n in n
        if (a == t->value)
            return t;

        p = n;

        if (a < t->value) {
            n = st_left(n);
            d = LEFT;
        } else if (a > t->value) {
            n = st_right(n);
            d = RIGHT;
        }
    }

    // create a new node and insert it into the tree
    struct treeint *i = calloc(sizeof(struct treeint), 1);
    if (st_root(tree))
        st_insert(&st_root(tree), p, &i->st_n, d);
    else
        st_root(tree) = &i->st_n;

    i->value = a;
    return i;
}
```
此函式執行的動作可以分為兩部份
1. 找到新節點要插入的位置
    過程和binary search相同, 透過for loop向子節點尋找要插入的位置
    若找到數值和自己相同節點則回傳該節點
2. 判斷目前此樹是否為空, 若是則將新節點作為root, 不是則呼叫`st_insert`進行插入的動作

**st_insert**
將新節點放入新的位置並將父節點對到child的pointer和自己指向父節點的pointer設定好
之後交由`st_update`來更新此樹, 確保此樹符合規則
### Update
像紅黑樹還有AVL tree一樣
此樹也有update method來確保insertion或deletion之後, 此樹依然保持它要求的規則
例如維持平衡
由`st_update`負責執行此動作

首先利用`st_balance`計算此樹的balance factor
基本上是左子樹高減掉右子樹高
應該注意到每個`st_node`都有一個成員`hint`作為儲存節點在數中高度的數字
不過它並非完全精準的計算 或許後面可以改進此部份計算高度的函式
* 若左傾, 則進行左旋
* 若右傾, 則進行右旋

需要注意的是如果n就是root, 則需要先將root改變再進行旋轉
且如果旋轉完之後n變成leaf node或者跟之前的高度不同
則持續往上對n的parent做update
### Remove
remove節點由`treeint_remove`執行
先找出要刪除的節點位置後進行`st_remove`
之後再free該節點
## 測驗 $\alpha - 2$
我認為這段程式碼所實作出的S-tree有數個可以改進的地方
1. **st_balance跟st_max_hint動作重複很多, 可以在計算st_balance同時更新hint**
    `st_balance`和`st_max_hint`程式碼高度重複
    可以在計算balance factor的時候同時更新`n->hint`
    並且我們知道若插入節點後沒有造成旋轉, 則`hint`不會改變, 所以可以把`st_max_hint`搬入rotate function當中以減少重複運算的次數
2. **沒有rotate就應該繼續往上update**
    原先是否繼續往上update還要經過判斷
    但插入節點後所有父輩節點的高度都會改變都有機會不平衡, 只要沒有進行旋轉都應該繼續update
    一旦旋轉後這個操作必須保證讓s tree回到平衡所以不需要繼續update
3. **此樹並沒有做到真的平衡**
    如果依照順序插入10, 5, 20, 2, 7, 25
    此時s tree是平衡的沒問題
    但如果這時候插入23, 結果會發現它只有執行一次right rotate, 樹還是不平衡的
    甚至接著繼續插入19, 它依舊沒發現自己不平衡
    s tree判斷平衡的方式和AVL tree很類似
    但AVL tree還有將不平衡的類型分為四種而進行對應不同的旋轉方式
    s tree在RL, LR的情況旋轉之後依然不平衡
    只要依序插入20, 25, 23即可發現
4. **replace裡面的if太多**
    `st_replace_right, st_replace_left`當中的if判斷式太多
    可讀性很差


**改進**
1. **進行st_balance同時計算n->hint**
    如此一來只有旋轉後才需要進行`st_max_hint`來更新節點高度
```diff
static inline int st_balance(struct st_node *n)
{
    int l = 0, r = 0;

    if (st_left(n))
        l = st_left(n)->hint + 1;

    if (st_right(n))
        r = st_right(n)->hint + 1;
+    n->hint = l > r ? l : r;
    return l - r;
}
```
2. **改變向上執行st_update的條件**
    只要插入後沒有旋轉都應該繼續往上update
    而只有b被偵測出不平衡才會旋轉
    所以b <= 1 or b >= -1都應該進行update
    如此一來還能減少一個if的判斷
```diff
static inline void st_update(struct st_node **root, struct st_node *n)
{
    if (!n)
        return;

    int b = st_balance(n);
    int prev_hint = n->hint;
    struct st_node *p = st_parent(n);
    if (b < -1) {
        /* leaning to the right */
        if (n == *root)
            *root = st_right(n);
        st_rotate_right(n);

    }

    else if (b > 1) {
        /* leaning to the left */
        if (n == *root)
            *root = st_left(n);
        st_rotate_left(n);
    
+    } else {
+        st_update(root, p);
+    }
-    n->hint = st_max_hint(n);
-    if (n->hint == 0 || n->hint != prev_hint)
-        st_update(root, p);
}
```
3. **處理RL, LR旋轉後依然不平衡的情況**
    相較AVL tree在這兩種情況會進行兩次旋轉的處理方式
    我只進行一次, 不過做法和一般旋轉不同
    如此一來可以保證經過旋轉後s tree一定平衡
    *實作有錯誤的地方 delete的時候會造成segmentation fault 仍須更正*
```c
static inline void st_rl_rotate(struct st_node *n)
{
    struct st_node *r = st_right(n), *p = st_parent(n);
    struct st_node *rl = st_left(r);

    st_parent(rl) = st_parent(n);
    st_right(n) = st_left(rl);
    st_left(r) = st_right(rl);
    st_parent(n) = rl;
    st_parent(r) = rl;
    st_left(rl) = n;
    st_right(rl) = r;

    if (p && st_left(p) == n)
        st_left(p) = rl;
    else if (p)
        st_right(p) = rl;
    
    if (st_right(n))
        st_rparent(n) = n;
    if (st_left(r))
        st_lparent(r) = r;

    n->hint = st_max_hint(n);
}

static inline void st_update(struct st_node **root, struct st_node *n)
{
    if (!n)
        return;

    int b = st_balance(n);
    int prev_hint = n->hint;
    struct st_node *p = st_parent(n);
    if (b < -1) {
        /* leaning to the right */
        if (n == *root)
            *root = st_right(n);
        if (st_right(st_right(n)))
            st_rotate_right(n);
        else
            st_rl_rotate(n);
    }

    else if (b > 1) {
        /* leaning to the left */
        if (n == *root)
            *root = st_left(n);
        if (st_left(st_left(n)))
            st_rotate_left(n);
        else
            st_lr_rotate(n);
    } else {
        st_update(root, p);
    }
}
```

### 測驗 $\alpha - 3$
TODO

---

## 測驗 $\beta$
```c
#include <stdint.h>

static inline uintptr_t align_up(uintptr_t sz, size_t alignment)
{
    uintptr_t mask = alignment - 1;
    if ((alignment & mask) == 0) {  /* power of two? */
        return (sz + mask) & (~mask);      
    }
    return (((sz + mask) / alignment) * alignment);
}
```
### 測驗 $\beta - 1$
若`alignment`在都是2的指數情況下
也就是1000, 0100, 0010等
答案會是`sz`把`alignment`的msb後面的bit都清為0
之後再加上`alignment`
解答會是`((sz / alignment) + 1) * alignment`
不過提示有說不能有`alignment`, 我們又知道`mask = alignment - 1`
所以想辦法把`alignment`用`mask`表示就可以
因為`alignment`剛好是二的指數所以`mask`剛好會是把`alignment`的msb後面的0全部換成1
利用這個特性先將`mask`給反轉之後跟`sz`做AND 就能達到上述第一步
接著加上`mask + 1`就是加上`alignment`
### 測驗 $\beta - 2$
在[sort.c](https://elixir.bootlin.com/linux/latest/source/lib/sort.c)當中有一個函式如下
```clike
/**
 * is_aligned - is this pointer & size okay for word-wide copying?
 * @base: pointer to data
 * @size: size of each element
 * @align: required alignment (typically 4 or 8)
 *
 * Returns true if elements can be copied using word loads and stores.
 * The size must be a multiple of the alignment, and the base address must
 * be if we do not have CONFIG_HAVE_EFFICIENT_UNALIGNED_ACCESS.
 *
 * For some reason, gcc doesn't know to optimize "if (a & mask || b & mask)"
 * to "if ((a | b) & mask)", so we do that by hand.
 */
__attribute_const__ __always_inline
static bool is_aligned(const void *base, size_t size, unsigned char align)
{
	unsigned char lsbits = (unsigned char)size;

	(void)base;
#ifndef CONFIG_HAVE_EFFICIENT_UNALIGNED_ACCESS
	lsbits |= (unsigned char)(uintptr_t)base;
#endif
	return (lsbits & (align - 1)) == 0;
}
```
## 測驗 $\gamma$
`HHHH`:`pthread_cond_wait(&qs->cond_st, &qs->mtx_st)`
`JJJJ`:`pthread_cond_signal(&qs2->cond_st)`
## 測驗 $\gamma - 1$
主要探討`qsort_mt`和其相關內容
實作此程式碼首先探討兩個特別的結構
```clike
struct qsort {
    enum thread_state st;   /* For coordinating work. */
    struct common *common;  /* Common shared elements. */
    void *a;                /* Array base. */
    size_t n;               /* Number of elements. */
    pthread_t id;           /* Thread id. */
    pthread_mutex_t mtx_st; /* For signalling state change. */
    pthread_cond_t cond_st; /* For signalling state change. */
};

struct common {
    int swaptype;           /* Code to use for swapping */
    size_t es;              /* Element size. */
    void *thunk;            /* Thunk for qsort_r */
    cmp_t *cmp;             /* Comparison function */
    int nthreads;           /* Total number of pool threads. */
    int idlethreads;        /* Number of idle threads in pool. */
    int forkelem;           /* Minimum number of elements for a new thread. */
    struct qsort *pool;     /* Fixed pool of threads. */
    pthread_mutex_t mtx_al; /* For allocating threads in the pool. */
};
```
首先有個`struct common c`
初始化它的mutex lock並建立一個pool, 當中有`maxthreads`個thread
之後依序建立這些thread並執行`qsort_thread`(注意此處還沒真正開始sort, `a`沒有被送進去)
注意到所有thread共享同一個`struct common`
如果取得`qs->mtx_st`這個mutex lock 就可以進行quick sort
也就是進行`qsort_algo`
之後thread進入idle狀態
如果全部thread都idle, 則依序把thread都terminate
如果沒有, 就回到again的地方再來一次

**`qsort_algo`**
此函式為主要進行quick sort的部分
如果array size小於7則直接進行熟知的bubble sort
```clike
if (n < 7) {
        for (pm = (char *) a + es; pm < (char *) a + n * es; pm += es)
            for (pl = pm; pl > (char *) a && CMP(thunk, pl - es, pl) > 0;
                 pl -= es)
                swap(pl, pl - es);
        return;
    }
```
如果大於7則要進行quick sort, 而quick sort在升冪或降冪排列的時候時間複雜度會變成$O(n)$
為了避免這個情形, 盡量讓每次sorting都是average case
要進行以下操作以取得中位數然後放到array head的地方
```clike
    pm = (char *) a + (n / 2) * es;
    if (n > 7) {
        pl = (char *) a;
        pn = (char *) a + (n - 1) * es;
        if (n > 40) {
            d = (n / 8) * es;
            pl = med3(pl, pl + d, pl + 2 * d, cmp, thunk);
            pm = med3(pm - d, pm, pm + d, cmp, thunk);
            pn = med3(pn - 2 * d, pn - d, pn, cmp, thunk);
        }
        pm = med3(pl, pm, pn, cmp, thunk);
    }
    swap(a, pm);
```
然後進行真正的quick sort
```clike
    pa = pb = (char *) a + es;

    pc = pd = (char *) a + (n - 1) * es;
    for (;;) {
        while (pb <= pc && (r = CMP(thunk, pb, a)) <= 0) {
            if (r == 0) {
                swap_cnt = 1;
                swap(pa, pb);
                pa += es;
            }
            pb += es;
        }
        while (pb <= pc && (r = CMP(thunk, pc, a)) >= 0) {
            if (r == 0) {
                swap_cnt = 1;
                swap(pc, pd);
                pd -= es;
            }
            pc -= es;
        }
        if (pb > pc)
            break;
        swap(pb, pc);
        swap_cnt = 1;
        pb += es;
        pc -= es;
    }
```
做的事跟普通的quick sort沒太大差別, 不過這邊比較特別是只把跟pivot相同的element搬到前面
排完後前面有一段element都是相同的元素
cmp有可能出錯而造成sorting被打斷
如果發生這情況就從尾巴往前把還沒做完的部分給sort完

一開始只需要在意pb, pc
pb是array second element, pc是array last element, a就是課本的pivot
如果pb所指向的element跟a指向的element大小相同
則把a跟pb交換並且讓`a += es`
重複上述過程直到pb碰到比a大的元素

不斷重複上述的sorting過程直到`pb > pc`
此時`a~pa`, `pb~pd`都是大小和`*a`相同的元素
以下的swapping procedure是為了把這些相同的元素擺到array中間
```clike
    pn = (char *) a + n * es;
    r = min(pa - (char *) a, pb - pa);
    vecswap(a, pb - r, r);
    r = min(pd - pc, pn - pd - es);
    vecswap(pb, pn - r, r);
```
這時候頭尾兩段元素還沒排好
```clike
    nl = (pb - pa) / es;
    nr = (pd - pc) / es;
```


如果沒有swap過(沒有重複元素)
則進行insertion sort
不過如果inseriton sort的時候swap太多次則直接跳出
不然就做到結束然後return

這時候如果尚未排列好的元素的數量需要多創建threads來排列
則喚醒躺在cond裡面的thread
若沒有則在同一個thread裡遞迴執行qsort
## 測驗 $\gamma - 2$
使用thread sanitizer去檢驗程式碼後
發現數個錯誤
第一個在`allocate_thread`中
```clike
static struct qsort *allocate_thread(struct common *c)
{
    verify(pthread_mutex_lock(&c->mtx_al));
    for (int i = 0; i < c->nthreads; i++)
        if (c->pool[i].st == ts_idle) {
            c->idlethreads--;
            c->pool[i].st = ts_work;
            verify(pthread_mutex_lock(&c->pool[i].mtx_st));
            verify(pthread_mutex_unlock(&c->mtx_al));
            return (&c->pool[i]);
        }
    verify(pthread_mutex_unlock(&c->mtx_al));
    return (NULL);
}
```
`verify(pthread_mutex_lock(&c->pool[i].mtx_st))`之後沒有對應的unlock
修正如下
```diff
static struct qsort *allocate_thread(struct common *c)
{
    verify(pthread_mutex_lock(&c->mtx_al));
    for (int i = 0; i < c->nthreads; i++) {
+        verify(pthread_mutex_lock(&c->pool[i].mtx_st));
        if (c->pool[i].st == ts_idle) {
            c->idlethreads--;
            c->pool[i].st = ts_work;
+            verify(pthread_mutex_unlock(&c->pool[i].mtx_st));
            verify(pthread_mutex_unlock(&c->mtx_al));
            return (&c->pool[i]);
        }
+        verify(pthread_mutex_unlock(&c->pool[i].mtx_st));
    }
    verify(pthread_mutex_unlock(&c->mtx_al));
    return (NULL);
}
```
接著修正`qsort_algo`當中配置資源時候沒有做好的mutual exclusion
```diff
/* Initialize qsort arguments. */
+ verify(pthread_mutex_lock(&qs->mtx_st));
c = qs->common;
+ verify(pthread_mutex_lock(&c->mtx_al));
es = c->es;
cmp = c->cmp;
swaptype = c->swaptype;
+ verify(pthread_mutex_unlock(&c->mtx_al));
a = qs->a;
n = qs->n;
+ verify(pthread_mutex_unlock(&qs->mtx_st));
```
```diff
/* Now try to launch subthreads. */
if (nl > c->forkelem && nr > c->forkelem &&
    (qs2 = allocate_thread(c)) != NULL) {
+     verify(pthread_mutex_lock(&qs2->mtx_st));
    qs2->a = a;
    qs2->n = nl;
    verify(pthread_cond_signal(&qs2->cond_st));
    verify(pthread_mutex_unlock(&qs2->mtx_st));
```
## 測驗 $\gamma - 3$
為了嘗試改進此排序的效能
要分析它的算法
先不考慮multi-thread的情況
在`qsort_algo`當中進行的quick sort每做完一次回圈
`a ~ pa-es`都會是相同的數值
若是重複的數字很少, 甚至是沒有相同的數字
此處的動作就毫無意義, 直接進入以下的insertion sort
還不如一開始就進行insertion sort
若是有大量數字重複, 則此演算法表現應該十分突出




