# 2021q3 Homework2 (hp)
contributed by < `conc95xh` >

:::warning
注意共筆書寫規範: 中英文間用一個半形空白字元區隔
:notes: jserv
:::

## 相關閱讀作業

本次作業主要以 [Lock-Free Data Structures with Hazard Pointers](https://erdani.org/publications/cuj-2004-12.pdf) 這篇論文為基底，該論文只有闡述如何實現在 lock-free 的情況下更新且釋放一個可能正在被讀取的指標，本次作業則去實作一個 lock-free 的 linked list，包含

* 插入新節點
* 刪除節點
* 給定一個 key 搜尋相同key的節點 並傳回適當位置
* 若 linked list 中有完全相等的 key ，則傳回 curr 為指向該節點， prev 指向前一個節點的next field， next 則指向 curr 的下一個節點
* 若 linked list 中沒有完全相等的 key，則傳回的 curr 指向的節點為key 比傳入 key 還大的節點，且 prev 指向前一個節點的 next field，且其 key 比傳入的 key 還小

對於插入新節點的操作，我們預期若 linked-list 裡沒有跟新節點一樣的 key，我們可以先把新節點的 next 指向 curr ，再把 *prev 改寫成新節點的位置(atomic operation)

對於刪除一個節點，我們預期若 linked-list find 找到一個key完全相同的節點，我們可以直接把 *prev 改寫成指向 next，即完成刪除(atomic operation)

不管插入或刪除，都是在對 linked-list 的操作為一個 atomic operation 的前提之下，因此必須使用 CAS 操作(check if the field's value is the same as the expected value, then replace the field's value with the desired value)

儘管以上前提都符合了，仍然有一個問題：

若被刪除的節點仍然有執行緒在讀取使用，則記憶體何時該釋放？

在論文中，提出一個方法:即 Hazard pointer，任意一個執行緒在取用一個節點時，必須將其加入該 thread 的 Hazard pointer list 裡，以向其他執行緒宣告此節點被讀取中。

若有任何刪除節點的行為，不需要鎖定防止其他執行緒干擾，只要刪除的操作命令為 atomic operation 即可從 linked-list 移除，但必項保證該刪除節點的記憶體區塊不會被釋放，直到該節點不在任何 thread 的 hazard pointer list中(即retire)

Hazard pointer 的靈魂即為讀取的執行緒要讀取一個節點的內容時，先
* 把取得節點的位置放入 Hazard pointer list 裡
* 再確認一下取得節點的狀態是不是在放入 Hazard pointer 前的狀態一樣
* 若為真，則該節點可安心讀取不必當心被釋放
* 若為假，則重新操作或放棄(如第10行的 try_again)

因此可以發現，在以下程式碼第13行我們將取得的 curr 放入 hazard pointer 時，隨即檢查 `*prev == unmarked curr` 我們在12行時藉由prev取得 curr，因此 prev 不是`&list->head` 就是上一個一樣放入 hazard pointer 的節點的 next field，因此可以放心讀取不怕存取到被釋放的記憶體

static bool __list_find(list_t *list,
                        list_key_t *key,
                        atomic_uintptr_t **par_prev,
                        list_node_t **par_curr,
                        list_node_t **par_next)
{
    atomic_uintptr_t *prev = NULL;
    list_node_t *curr = NULL, *next = NULL;

try_again:
    prev = &list->head; //prev為指向指向curr的指標，因此為**
    curr = (list_node_t *) atomic_load(prev);
    (void) list_hp_protect_ptr(list->hp, HP_CURR, (uintptr_t) curr); 
    if (atomic_load(prev) != get_unmarked(curr))
        goto try_again; //將curr放入hp時，仍須再檢查一次*prev
                        //指向curr，有可能在放入前被插斷，且被更
                        //改，若結果為真，則代表放入hp成功
    while (true) {
        if (!get_unmarked_node(curr))
            return false;
        next = (list_node_t *) atomic_load(&get_unmarked_node(curr)->next);
        (void) list_hp_protect_ptr(list->hp, HP_NEXT, get_unmarked(next));
        if (atomic_load(&get_unmarked_node(curr)->next) != (uintptr_t) next)
            break; //21和23行這兩段break無法理解，
                   //break後，prev跟curr中間不一定是最適當
                   //插入的位置，因為list有可能再
                   //也不是ordered list？
        if (get_unmarked(next) == atomic_load((atomic_uintptr_t *) &list->tail))
            break; 
        if (atomic_load(prev) != get_unmarked(curr))
            goto try_again;
        if (get_unmarked_node(next) == next) {
            if (!(get_unmarked_node(curr)->key < *key)) {
                *par_curr = curr;
                *par_prev = prev;
                *par_next = next;
                return *key == get_unmarked_node(curr)->key; //Q:FFF
            }
            prev = &get_unmarked_node(curr)->next;
            (void) list_hp_protect_release(list->hp, HP_PREV,
                                           get_unmarked(curr));
        } else { //若curr已經被mark成將要刪除了，則將這個節點
                 // 刪除，刪除成功則繼續往下找
                 // 失敗則重來，因此代表 curr 的 next 不一定有意義
            uintptr_t tmp = get_unmarked(curr);
            if (!atomic_compare_exchange_strong(prev, &tmp, get_unmarked(next)))
                goto try_again;
            list_hp_retire(list->hp, get_unmarked(curr));
        }
        curr = next;
        (void) list_hp_protect_release(list->hp, HP_CURR, get_unmarked(next));
    }
    *par_curr = curr;
    *par_prev = prev;
    *par_next = next;

    return false;
}

## 插入和刪除

不管插入和刪除，利用上述 `__list_find` 得到 prev, curr, next (即節點的相鄰資訊) 後，我們仍然需要利用 CAS 去確認一些事情，並且確認和執行為不可分割的操作

譬如插入節點，我們先確認 *prev==curr，若為真我們才可以把新的node插在兩者中間，若為假，則傳回的 prev,curr,next 都沒有意義，所有動作得重新再來，因此我們可以發現，所有動作都在一個 while (true) 的迴圈中

bool list_insert(list_t *list, list_key_t key)
{
    list_node_t *curr = NULL, *next = NULL;
    atomic_uintptr_t *prev = NULL;

    list_node_t *node = list_node_new(key);

    while (true) {
        if (__list_find(list, &key, &prev, &curr, &next)) {
            list_node_destroy(node);
            list_hp_clear(list->hp);
            return false;
        }
        atomic_store_explicit(&node->next, (uintptr_t) curr,
                              memory_order_relaxed);
        uintptr_t tmp = get_unmarked(curr);
        if (atomic_compare_exchange_strong(prev, &tmp, (uintptr_t) node)) {
            list_hp_clear(list->hp);
            return true;
        }
    }
}

刪除節點則為確認 curr 的 next field 沒有被mark，如果沒有則 mark 成delete，因為是 atomic operation，所以沒有其他人可以同時把它 mark 成delete

若成功則保證只有我們將 curr 標記為 deleted，

若下一個節點同時被 mark 成 deleted 且被移除了怎麼辦？
這樣我們不就把前一個節點的 next 指到一個即將被釋放的node?

next 的 next field 也有可能被mark，但它在會在第18行檢查 *prev 是否為 unmarked curr 時，失敗，直接回傳true，下次再由 __list_find() 的第39行執行真正的移除動作，因此我們不必擔心我們的 prev 會指向一個也已被移除的節點

bool list_delete(list_t *list, list_key_t key)
{
    list_node_t *curr, *next;
    atomic_uintptr_t *prev;
    while (true) {
        if (!__list_find(list, &key, &prev, &curr, &next)) {
            list_hp_clear(list->hp);
            return false;
        }

        uintptr_t tmp = get_unmarked(next);

        if (!atomic_compare_exchange_strong(&curr->next, &tmp,
                                            get_marked(next)))
            continue;

        tmp = get_unmarked(curr);
        if (atomic_compare_exchange_strong(prev, &tmp, get_unmarked(next))) {
            list_hp_clear(list->hp);
            list_hp_retire(list->hp, tmp); //Q:DDD
        } else {
            list_hp_clear(list->hp);
        }
        return true;
    }
}

## 釋放記憶體

list_hp_retire 即為釋放記憶體的主角，傳入一個 ptr，它會先將 PTR 加入retire list，且先檢查每一個在 retire list的ptr是否也在 Hazard pointer 列表中，若無則執行清除釋放動作，因此不只清除自己傳入的 ptr，也會順手把之前積存的 retired pointer 清除

void list_hp_retire(list_hp_t *hp, uintptr_t ptr)
{
    retirelist_t *rl = hp->rl[tid() * CLPAD];
    rl->list[rl->size++] = ptr;
    assert(rl->size < HP_MAX_RETIRED);

    if (rl->size < HP_THRESHOLD_R)
        return;

    for (size_t iret = 0; iret < rl->size; iret++) {
        uintptr_t obj = rl->list[iret];
        bool can_delete = true;
        for (int itid = 0; itid < HP_MAX_THREADS && can_delete; itid++) {
            for (int ihp = hp->max_hps - 1; ihp >= 0; ihp--) {
                if (atomic_load(&hp->hp[itid][ihp]) == obj) {
                    can_delete = false;
                    break;
                }
            }
        }

        if (can_delete) {
            size_t bytes = (rl->size - iret) * sizeof(rl->list[0]);
            memmove(&rl->list[iret], &rl->list[iret + 1], bytes);
            rl->size--;
            hp->deletefunc((void *)obj);  ///Q:RRR
        }
    }
}

## 改善

移除 __list_find 第21及第23行兩個compound statements，因為若在該點break，可能會發生ordered list無法順利排序(？)

嘗試將此程式修改成如下，執行結果相同

diff --git a/list/list.c b/list/list.c
index 5bb75c0..b3347d7 100644
--- a/list/list.c
+++ b/list/list.c
@@ -231,13 +231,19 @@ try_again:
         goto try_again;
     while (true) {
         if (!get_unmarked_node(curr))
-            return false;
+            break;
         next = (list_node_t *) atomic_load(&get_unmarked_node(curr)->next);
         (void) list_hp_protect_ptr(list->hp, HP_NEXT, get_unmarked(next));
-        if (atomic_load(&get_unmarked_node(curr)->next) != (uintptr_t) next)
-            break;
-        if (get_unmarked(next) == atomic_load((atomic_uintptr_t *) &list->tail))
-            break;
+#if 0
+        if (atomic_load(&get_unmarked_node(curr)->next) != (uintptr_t) next) {
+               printf("%s:%s:%d\n", __file__, __fun__, __LINE__);
+               break;
+       }
+        if (get_unmarked(next) == atomic_load((atomic_uintptr_t *) &list->tail)) {
+               printf("%s:%s:%d\n", __file__, __fun__, __LINE__);
+               break;
+       }
+#endif
         if (atomic_load(prev) != get_unmarked(curr))
             goto try_again;
         if (get_unmarked_node(next) == next) {
@@ -259,10 +265,12 @@ try_again:
         curr = next;
         (void) list_hp_protect_release(list->hp, HP_CURR, get_unmarked(next));
     }
+#if 0
     *par_curr = curr;
     *par_prev = prev;
     *par_next = next;
 
+#endif 
     return false;
 }

TODO:
- [x] 執行結果是否一樣？
- [ ] 執行速度跟原先程式比較?

## 對比 rcu_list，解釋同為 lock-free 演算法，跟上述 Hazard pointer 手法有何異同？能否指出 rcu_list 實作缺陷並改進？

在老師提供的範例模板中，針對所有的 rcu_read_lock ，會將現在所有 read threads 串接到 zombie 串列中，並在 rcu_read_unlock 時，檢查自己是否為最後一個 reader ，若為最後一個，則執行清除 zombie list 及 free zombie(dead node) 的動作，若否，則留待 (defer)，直到下一個 reader 退出 reader section 且發現自己是 last reader，幫忙處理 free 的動作。

因此獲得一個好處，即所有的 reader 在標記 reader critical section 時，是 lock-free 的，不用等待其他任何 thread 釋放 lock (包含 writer thread )

競爭的部分只有在將自己串接到 zombie list 的開頭時，需要用 atomic compare & swap 去確認剛才拿到的 old head 是否還有效，如下程式碼，只有在這裡才會付出等待的代價

void rcu_read_lock(read_handle_t *handle)
{
    zombie_node_t *z_node = make_zombie_node();

    z_node->owner = handle;
    handle->zombie = z_node;

    rcu_list_t *list = handle->list;

    zombie_node_t *old_head;
    __atomic_load(&list->zombie_head, &old_head, __ATOMIC_SEQ_CST);

    do {
        __atomic_store(&z_node->next, &old_head, __ATOMIC_SEQ_CST);

    } while (!__atomic_compare_exchange(&list->zombie_head, &old_head, &z_node,
                                        true, __ATOMIC_SEQ_CST,
                                        __ATOMIC_SEQ_CST));
}

在 rcu_read_unlock 時，執行緒會檢查自己後面的其他 readers 是否已完成，若已完成，我們可以代為執行 free的動作。

如下 free(dead_node) 即為將之前已移除但還有 reader 在使用的 node 釋放掉，因為若 last 為 true ，則代表在我們之前的所有 readers 已跳出 reader critical section (注意：我們後面可能還有 readers，但我們不在意，我們只關注我們之前有沒有 readers )

    if (last) {
        while (n) {
            list_node_t *dead_node = n->zombie;
            free(dead_node);

            zombie_node_t *old_node = n;
            __atomic_load(&n->next, &n, __ATOMIC_SEQ_CST);
            free(old_node);
        }

        __atomic_store(&z_node->next, &n, __ATOMIC_SEQ_CST);
    }

    void *null = NULL;
    __atomic_store(&z_node->owner, &null, __ATOMIC_SEQ_CST);

(條件a):「我們之前所有的 readers 已跳出 read-side critical section」

修件a 這句話若為真有什麼意義呢？ 我想若在此時把我們自己也塞入 zombie-list ，是不是代表我們對 rcu-list 某個節點做的移除更動，在條件a成立後，沒有人再需要被移除的節點，因此我們可以放心釋放該節點的資源。

因此，以下新增了一個 list_del_node 函式，基本想法是：在呼叫 list_del_node 時我們雖然是 writer thread ，但我們在利用 atomic operation 做完節點的移除動作後，該節點已無法從 list head 追蹤到，只剩下移除前拿到的人可以繼續使用(~~信賴保護原則~~？)，若將我們自己也標記為 reader section (via rcu_read_lock()) ，然後把 zombie node 的 zombie 設定成要刪除的節點，再直接呼叫 rcu_read_unlock() ，這樣等於將 writer thread 也加入 zombie list。

之後若有人退出 reader section 且為最後一個 reader，則會順手執行 free 的動作。

bool list_del_node(rcu_list_t *list, void *data, finder_func_t finder,                                                                                                       
                     write_handle_t *handle)                                                                                                                                 
{                                                                                                                                                                            
                                                                                                                                                                             
        /* find node and delete it, but defer the free                                                                                                                       
         * until our thread is the last reader thread */                                                                                                                     
                                                                                                                                                                             
        lock_for_write(list);                                                                                                                                                
        iterator_t iter = list_find(list, data, finder, handle);                                                                                                             
                                                                                                                                                                             
        if (!iter.entry)  {                                                                                                                                                  
                unlock_for_write(list);                                                                                                                                      
                return false;                                                                                                                                                
        }                                                                                                                                                                    
                                                                                                                                                                             
        /* remove the node */                                                                                                                                                
        __atomic_store(&iter.entry->prev->next, &iter.entry->next, __ATOMIC_SEQ_CST);                                                                                        
        __atomic_store(&iter.entry->next->prev, &iter.entry->prev, __ATOMIC_SEQ_CST);                                                                                        
                                                                                                                                                                             
        unlock_for_write(list);                                                                                                                                              
                                                                                                                                                                             
        /* add ourself as a reader thread to mark the boundary */                                                                                                            
        rcu_read_lock(handle);                                                                                                                                               
        handle->zombie->zombie = iter.entry;                                                                                                                                 
        rcu_read_unlock(handle);                                  
        return true;                                                                                                                                                             
}        

相關程式碼修改：執行過程中 Thread Sanitizer 沒有吐出警告訊息
為方便 delete node 的實作，將 list head及list tail 改成預設即是dummy node，且在初始化時，即互連。

https://github.com/conc95xh/linux2021-summer-quiz/blob/main/list/rcu_list.c

## 其他相關閱讀

https://developers.redhat.com/blog/2016/01/14/toward-a-better-use-of-c11-atomics-part-1#
http://www2.rdrop.com/~paulmck/RCU/