並行程式設計: Hazard pointer

貢獻者: sternacht, jserv

在並行程式設計中，當我們在存取共用的記憶體物件時，需要考慮到其他執行緒是否有可能也正在存取同一個物件，若要釋放該記憶體物件時，不考慮這個問題，會引發嚴重的後果，例如 dangling pointer。

使用 mutex 是最簡單且直觀的方法：存取共用記憶體時，acquire lock 即可保證沒有其他執行緒正在存取同一物件，也就可安全地釋放記憶體。但若我們正在存取的是一種 lock-free 資料結構，當然就不能恣意地使用 lock，因為會違反 lock-free 特性，即無論任何執行緒失敗，其他執行緒都能可繼續執行。於是乎，我們需要某種同為 lock-free 的記憶體物件回收機制。

對於 C 這樣缺乏內建 concurrent GC 機制的程式語言來說，若要實作 lock-free 演算法，就要自行處理記憶體釋放的議題。Hazard pointer 是其中一種解決方案，其原理是讀取端執行緒對指標進行識別，指標 (特別是指向的記憶體區塊) 若要釋放時，會事先保存，延遲到確認沒有讀取端執行緒，才進行真正的釋放。Linux 核心的 RCU 同步機制是另一種 lock-free 程式設計演算法和記憶體回收機制。

"hazard" 一詞多用來指「危險物、危害物」，與 "danger" 的主要區別:

1. hazard 是可數名詞
2. "hazard" 不是立刻會發生的危險，而是可能會帶來危險的「隱患」，例如 a health hazard (一個健康隱患) 和 a safety hazard (一個[行為或功能的]安全隱患)

〈Lock-Free Data Structures with Hazard Pointers〉寫道:

Each reader thread owns a single-writer/multi-reader shared pointer called Hazard pointer. When a reader thread assigns the address of a map to its hazard pointer, it is basically announcing to other threads (writers), "I am reading this map. You can replace it if you want, but don’t change its contents and certainly keep your deleteing hands off it."

Hazard pointer 可簡稱為 "HP"，其關鍵的結構有:

• Hazard pointer
• retire list (也稱為 thread free list)

hazard pointer 是一種可以解決 lock-free data 在動態記憶體配置的問題之方法。其基本的概念是允許 hazard pointer 可以被一個 thread 寫而多個 thread 讀，當存在 reader thread 正在操作該 hazard pointer 時，hazard pointer 會被標註，於是 writer thread 可得知此訊息並延遲對其釋放。

在 hazard pointer 架構中，每個 thread 首先需要各自維護的兩個關鍵集合是:

• hazard pointer: 儲存此 thread 正在存取的指標，因此該指標不能直接被釋放
• retire list: 被這個 thread 要求釋放的指標，但實際尚未釋放

因此要安全的釋放記憶體，其基本想法就是:

• 每個執行緒放在 hazard pointer 中的指標尚不能被釋放
• 每個執行緒要求釋放的指標先放在 retire list 中
• 掃描 retire list 可以得知所有執行緒皆不使用的指標，則可將其真正的釋放給作業系統

〈Lock-Free Data Structures with Hazard Pointers〉提到，lock-free programing 為了保證持續進展的特性，每個執行緒都有機會在任意時間去操作一個 object。當 thread A 在釋放一個物件時，需確保沒有另一個 thread B 仍取得該物件的 reference 且正要存取。如果先行釋放此 object，另一個 thread 的存取就會出錯。以下是經典的解決方式:

1. Reference counting: 基本想法是每當一個 thread 取得 reference 就將一個計數(其地址不同於 pointer)加一，解除 reference 時則減一。但是實際上其正確性要確保取得 pointer 和 reference count 加一是一個 atomic operation，需要使用特殊的 DCAS (Double Compare And-Swap) 指令，但其效率不彰
2. Wait and delete: 就是 「等一段時間」 再刪除，但等待時間未知
3. Keep a reference count next to the pointer: 使用比起 DCAS 較合理且效率比較好的 CAS2，可以 atomic 的交換 memory 中的兩個相鄰的 words，不過 64 位元的處理器上一般不存在這種指令(但可以透過一些操作指標的技巧達到同等效果)

第三個方案看似可行，但實際上仍有缺陷，因為 CAS 語意要求 object 和 reference count 要同時一致才能替換上新的數值，write 操作需要等待到沒有 read 操作 (reference count) 才能進行。換句話說，在一個 read 操作結束前如果又有下一個 read 進來，write 就只能空等待。

為克服上述方法的問題，hazard pointer 是一個更佳方案 (尤其針對 Write-Rarely-Read-Many 的情境)。內文中以 lock-free 的 map 為案例:

template <class K, class V>

class WRRMMap {
   Map<K, V> * pMap_;
   ...
};

當 WRRMMap 需要被更新時，writer thread 會去取得一個 *pMap_ 指向 map 的副本並更改，然後再將 *pMap_ 指向新的副本，並且回收舊的。麻煩的問題是這個舊的 map 可能有其他 thread 正嘗試讀取。

而 hazard pointer 的概念就是讓 reader 將正在存取的 map 加入到自己的單寫多讀(single-writer-multiple-reader)的 hazard pointer 中，其目的等同向其他 writer 宣告禁止回收該 map 。當 writer 把舊的 map 替換下來，map 先被放在一個 thread 獨立的 list 而不先釋放 ，直到 list 中的舊 map 達到一個上限，再去掃描每個 reader 的 hazard pointer 中是否有相匹配。如果某個舊 map 不與任何 reader 的 hazard pointer 匹配，那麼釋放該 map 就是安全的，否則就繼續將其留在 list 中，直到下次掃描。

// Hazard pointer record
class HPRecType {
   HPRecType *pNext_;
   int active_;
   static HPRecType *pHead_;
   static int listLen_;

public:
   void *pHazard_;
   static HPRecType *Head() {
      return pHead_;
   }

hazard pointer 的基本結構由 linked list 構成:

• *pNext 指向下個節點
• active_ 表示該節點被使用與否
• pHead_ 是 linked list 的首個節點
• listLen_ 則是 list 中的節點數量
• pHazard_ 是物件的指標本體(例如 map)

    // Acquires one hazard pointer
   static HPRecType * Acquire() {
      // Try to reuse a retired HP record
      HPRecType * p = pHead_;
      for (; p; p = p->pNext_) {
         if (p->active_ ||
            !CAS(&p->active_, 0, 1))
           continue;
         return p;
      }

      // Increment the list length
      int oldLen;
      do {
         oldLen = listLen_;
      } while (!CAS(&listLen_, oldLen, oldLen + 1));

      // Allocate a new one
      HPRecType * p = new HPRecType;
      p->active_ = 1;
      p->pHazard_ = 0;

      // Push it to the front
      do {
         old = pHead_;
         p->pNext_ = old;
      } while (!CAS(&pHead_, old, p));

      return p;
   }
   ...

要將一個指標加入 hazard pointer 需要 acquire 一個可用的節點，首先先看看有沒有已經被建立的空節點(沒有對應要保護的 pHazard 實體)可以直接使用。如果沒有，則需要建立新的 node 插入到 hazard pointer linked list 的開頭，listLen_ 也要相應的做加一

   // Releases a hazard pointer
   static void Release(HPRecType* p) {
      p->pHazard_ = 0;
      p->active_ = 0;
   }
};

要解除對指標的保護則透過 release，單純的將該節點的 pHazard_ 重置且 active_ 允許重用即可。

// Per-thread private variable
__per_thread__ vector<Map<K, V> *> rlist;

每個 thread 還有一個獨立存取的 retire list，存放該 thread 不再需要，若其它thread 也不再需要就可以被釋放 pointer。 retire list 因為只有每個 thread 自己可以存取的，因此不需要被同步。

template <class K, class V>
class WRRMMap {
   Map<K, V> * pMap_;
   ...
private:

   static void Retire(Map<K, V> * pOld) {
   // put it in the retired list
   rlist.push_back(pOld);
   if (rlist.size() >= R) {
      Scan(HPRecType::Head())
      }
   }
};

Retire 操作將要原本應該要釋放的 pointer 加入 vector 中，先儲存起來延遲釋放。當 vector 所放的 pointer 超過一個大小就呼叫 Scan 去掃描是否有可以真正釋放的 pointer。

void Scan(HPRecType * head) {

   // Stage 1: Scan hazard pointers list
   // collecting all non-null ptrs
   vector<void*> hp;
   while (head) {
      void * p = head->pHazard_;
      if (p) 
          hp.push_back(p);
      head = head->pNext_;
   }

   // Stage 2: sort the hazard pointers
   sort(hp.begin(), hp.end(), less<void*>());
      
   // Stage 3: Search for'em!
   vector<Map<K, V>*>::iterator i = rlist.begin();
   while (i != rlist.end()) {
      if (!binary_search(hp.begin(), hp.end(), *i) {
         // Aha!
         delete *i;
         if (&*i != &rlist.back()) {
            *i = rlist.back();
         }
          rlist.pop_back();
      } else {
         ++i;
      }
   }
}

Scan 的任務是找出 retire list 集合和所有 thread 的 hazard pointer 集合的差集，那些 pointer 就是可以被釋放而且不會導致錯誤的。

list2.c 是 Hazrd pointer 的簡化 C11 實作，搭配 GNU extension。

當執行緒進行讀取的操作時，hazard pointer 指向讀取的位址，此時若有其他的執行緒要對同一個位址進行釋放的動作，就需要先走訪 hazard pointers (hazard pointers 為鏈結串列結構)，確認是否有相同的位址存在，如果存在就不能釋放。retire list 則對應到該執行緒預定要釋放的指標空間，同樣以鏈結串列做串接，一定條件之後會對鏈結串列裡的所有節點進行嘗試釋放的動作，也就是上述提到的走訪 hazard pointer，滿足條件才會將其釋放，此 retire list 不需對其他執行緒同步，只有持有的執行緒自身才看得到。

typedef struct __hp {
    uintptr_t ptr;
    struct __hp *next;
} hp_t;

typedef struct {
    hp_t *pointers;
    hp_t *retired;
    void (*deallocator)(void *);
} domain_t;

hazard pointer 及 retired list 的基本結構中， __hp 是最小的單元，也就是一個節點，存有一個指向目標的指標以及另一個用來串接 linked list 的指標。 domain_t，則是一個 thread 中各持有一個的結構，包含 hazard pointer 以及 retired list ， deallocator 照其字面上的意思就是用來釋放空間的函式。

針對 list 的操作

/* Allocate a new node with specified value and append to list */
static hp_t *list_append(hp_t **head, uintptr_t ptr)
{
    hp_t *new = calloc(1, sizeof(hp_t));
    if (!new)
        return NULL;

    new->ptr = ptr;
    hp_t *old = atomic_load(head);

    do {
        new->next = old;
    } while (!atomic_cas(head, &old, &new));

    return new;
}

/* Attempt to find an empty node to store value, otherwise append a new node.
 * Returns the node containing the newly added value.
 */
hp_t *list_insert_or_append(hp_t **head, uintptr_t ptr)
{
    hp_t *node;
    bool need_alloc = true;

    LIST_ITER (head, node) {
        uintptr_t expected = atomic_load(&node->ptr);
        if (expected == 0 && atomic_cas(&node->ptr, &expected, &ptr)) {
            need_alloc = false;
            break;
        }
    }

    if (need_alloc)
        node = list_append(head, ptr);

    return node;
}

/* Remove a node from the list with the specified value */
bool list_remove(hp_t **head, uintptr_t ptr)
{
    hp_t *node;
    const uintptr_t nullptr = 0;

    LIST_ITER (head, node) {
        uintptr_t expected = atomic_load(&node->ptr);
        if (expected == ptr && atomic_cas(&node->ptr, &expected, &nullptr))
            return true;
    }

    return false;
}

/* Returns 1 if the list currently contains an node with the specified value */
bool list_contains(hp_t **head, uintptr_t ptr)
{
    hp_t *node;

    LIST_ITER (head, node) {
        if (atomic_load(&node->ptr) == ptr)
            return true;
    }

    return false;
}

/* Frees all the nodes in a list - NOT THREAD SAFE */
void list_free(hp_t **head)
{
    hp_t *cur = *head;
    while (cur) {
        hp_t *old = cur;
        cur = cur->next;
        free(old);
    }
}

針對 list 的操作共有四個，分別是 insert_and_append, remove, contains, free。

• insert_and_append 操作會有兩種不同的結果，一開始先遍尋 list，若當中有一個指標是空的，則該函式會作 CAS 操作 ，此為 insert，若沒有空指標的話則是作 append，向系統索取一塊新的空間來存放，返回值為新加入的節點。
• remove 逐一比對目標是否存在於 list 並將其移走，成功回傳 true，失敗回傳 false。
• contains 逐一比對目標是否存在 list 中，是則回傳 true，否則回傳 false。
• free 將給定的 list 整個釋放掉。

針對 domain_t 的操作

uintptr_t load(domain_t *dom, const uintptr_t *prot_ptr)
{
    const uintptr_t nullptr = 0;

    while (1) {
        uintptr_t val = atomic_load(prot_ptr);
        hp_t *node = list_insert_or_append(&dom->pointers, val);
        if (!node)
            return 0;

        /* Hazard pointer inserted successfully */
        if (atomic_load(prot_ptr) == val)
            return val;

        /*
         * This pointer is being retired by another thread - remove this hazard
         * pointer and try again. We first try to remove the hazard pointer we
         * just used. If someone else used it to drop the same pointer, we walk
         * the list.
         */
        uintptr_t tmp = val;
        if (!atomic_cas(&node->ptr, &tmp, &nullptr))
            list_remove(&dom->pointers, val);
    }
}

load 對應到 reader 執行緒讀取某一個指標內容的操作，操作時的順序為

• 將目標地址的值讀到 val 變數中，並嘗試將目標地址之指標放入 hazard pointer 中，失敗則回傳 0
• 再一次確定目標地址中的值與 val 中的值相同，以確保過程中沒有其他執行緒對該地址的值進行寫入操作
• 若地址中的值被改變，則嘗試將 hazard pointer 中的值改寫回空指標，並把 val 從 hazard pointer 中刪除。

void swap(domain_t *dom, uintptr_t *prot_ptr, uintptr_t new_val, int flags)
{
    const uintptr_t old_obj = atomic_exchange(prot_ptr, new_val);
    cleanup_ptr(dom, old_obj, flags);
}

swap 對應到 writer thread 寫入的動作，傳入的 new_val 會取代 old_val ，接著再將 old_val 空間釋放掉。

static void cleanup_ptr(domain_t *dom, uintptr_t ptr, int flags)
{
    if (!list_contains(&dom->pointers, ptr)) { /* deallocate straight away */
        dom->deallocator((void *) ptr);
    } else if (flags & DEFER_DEALLOC) { /* Defer deallocation for later */
        list_insert_or_append(&dom->retired, ptr);
    } else { /* Spin until all readers are done, then deallocate */
        while (list_contains(&dom->pointers, ptr))
            ;
        dom->deallocator((void *) ptr);
    }
}

cleanup_ptr 嘗試將一個指標從 hazard pointer 中移除，並釋放其空間，或是先將該指標移到 retired list 中存放，待稍後再將其移除。若是 flags 沒有標注要放到 retired list，則函式會不斷嘗試將指標釋放直到成功為止。

void cleanup(domain_t *dom, int flags)
{
    hp_t *node;

    LIST_ITER (&dom->retired, node) {
        uintptr_t ptr = node->ptr;
        if (!ptr)
            continue;

        if (!list_contains(&dom->pointers, ptr)) {
            /* We can deallocate straight away */
            if (list_remove(&wconfig->retired, ptr))
                dom->deallocator((void *) ptr);
        } else if (!(flags & DEFER_DEALLOC)) {
            /* Spin until all readers are done, then deallocate */
            while (list_contains(&dom->pointers, ptr))
                ;
            if (list_remove(&wconfig->retired, ptr))
                dom->deallocator((void *) ptr);
        }
    }
}

cleanup 則是試圖將整個 retired list 刪除，首先要先確認要釋放的指標是否在 hazard pointer 內，若沒有則將該指標從 retired list 中刪除，並釋放空間。如果還有其他 hazard pointer 存有該指標，且 flags 沒有標注要暫緩釋放的動作，則函式會不斷等到可以釋放為止。

void drop(domain_t *dom, uintptr_t safe_val)
{
    if (!list_remove(&dom->pointers, safe_val))
        __builtin_unreachable();
}

drop 的用途很簡單，就是將一個指標從 hazard pointer 中刪除。
比較特別的是底下的 __builtin_unreachable()，其用途是告訴編譯器，在這之後的程式碼是不會執行到的( 執行此函式為未定義行為 )，其中一種比較常見的用法是接在結束程式的函式後頭，或是程式要跳躍到其他地方，而不會回來的時候，加上這個函式則可以避免觸發編譯器的警告，也節省編譯後的組合語言行數，底下是一個例子，在條件結束的函式後面分別加與不加 __builtin_unreachable() ，看看編譯後的結果為何。

What optimizations does __builtin_unreachable facilitate?

void exit_if_true(bool cond)
{
    if (cond)
        exit(0);
}

void foo(bool x)
{
    if (x) {
        exit_if_true(1);
        __builtin_unreachable();
        printf("shall not come here");
    } else 
        return;
}

void foo2(bool x)
{
    if (x) {
        exit_if_true(1);
        // __builtin_unreachable();
        printf("shall not come here");
    } else 
        return;
}

關閉最佳化並輸出組合語言 gcc -S -O0，因 exit_if_true 傳入值必為真，因此開啟最佳化會自行刪除後面的程式碼

foo:
        pushq   %rbp
        movq    %rsp, %rbp
        subq    $16, %rsp
        movl    %edi, %eax
        movb    %al, -4(%rbp)
        cmpb    $0, -4(%rbp)
        je      .L7
        movl    $1, %edi
        call    exit_if_true
.L7:
        nop
        leave
        ret
.LC0:
        .string "shall not come here"
foo2:
        pushq   %rbp
        movq    %rsp, %rbp
        subq    $16, %rsp
        movl    %edi, %eax
        movb    %al, -4(%rbp)
        cmpb    $0, -4(%rbp)
        je      .L11
        movl    $1, %edi
        call    exit_if_true
        movl    $.LC0, %edi
        movl    $0, %eax
        call    printf
        jmp     .L8
.L11:
        nop
.L8:
        leave
        ret

組合語言的前半部分是相同的，但在 foo 裡面，一旦執行到 call exit_if_true 之後程式就判斷是結束了，即使後面還有一個 printf() ，這就是 __builin_unreachable() 的效果，反觀 foo2 ，即使執行到 call exit_if_true 之後程式仍沒有結束，並嘗試執行 printf() 輸出字串。

實作的改進

仔細看程式碼會發現 cleanup 從頭到尾都沒有被使用到，而在 swap 函式中，flag 參數也是設為 0 ，表示目前的實作並沒有用到 retired list ，若 reader 數量很多， writer 將會花費相當大量的時間在 cleanup_ptr 的自旋上。

透過 valgrind --tool=massif 命令產生的檔案可以觀察到程式執行的指令數量 (預設)，或是執行的時間，在有 100 個 reader ，並迭代(N_ITER) 100,000 次的情況下，使用 retired list 的版本僅花費 2 分鐘就完成，而原本的版本則在執行 10 分鐘之後仍未完成。

在一開始的範例中， writer 只有一個，retired list 則是 public 的，因此處理 retired list 的步驟可以等到 deinit 再去執行，而不是在 writer thread 結束的時候。反之若是只在 writer thread 結束之前做處理，則在 writer 比任何 reader 更早完成的情況下，會產生 memory leak 的問題。

但是這麼做會衍伸另一個問題，當我們把迭代次數 (N_ITER) 設的很高，且有一定數量的 reader 時(共享空間指標容易在 hazard pointers 裡)，會有較多的指標被放到 retired list 中等待在程式最後釋放，因此空間就被占據。

在論文〈Lock-Free Data Structures with Hazard Pointers〉提到 retired list 應該在甚麼時機做 cleanup 比較好，即 retired list 中的各數達到一個上限時，而這個上限的設定與 reader threads 的數量呈線性關係並略大於該數量，例如當 retired list 中的各數達到 reader 的 1.25 倍時就會觸發一次 cleanup，為此，我們需要一個額外的計數器來記錄 retired list 中目前有多少個待釋放的節點，並在 cleanup 一次之後重新計算剩餘的節點數，改動的部分如下

/* Compute the size of list */
uint32_t list_size(hp_t *head)
{
    if (!head)
        return 0;
    uint32_t c = 0;
    hp_t *node = head;
    while(node) {
         if(node->ptr)
            c++;
         node = node->next;
    }
    return c;
}

typedef struct {
    hp_t *pointers;
    hp_t *retired;
    uint32_t r_count;
    void (*deallocator)(void *);
} domain_t;

static void cleanup_ptr(domain_t *dom, uintptr_t ptr, int flags)
{
    if (!list_contains(&dom->pointers, ptr)) { /* deallocate straight away */
        dom->deallocator((void *) ptr);
    } else if (flags & DEFER_DEALLOC) { /* Defer deallocation for later */
        list_insert_or_append(&dom->retired, ptr);
        dom->r_count += 1;
    } else { /* Spin until all readers are done, then deallocate */
        while (list_contains(&dom->pointers, ptr))
            ;
        dom->deallocator((void *) ptr);
    }
}

... 
        // in 'writer thread' function
        swap(config_dom, (uintptr_t *) &shared_config, (uintptr_t) new_config,
             1);
        print_config("----updated config ", new_config);
        if (config_dom->r_count > r_limit){
            cleanup(config_dom, 1);
            config_dom->r_count = list_size(config_dom->retired);
        }
...

接著同樣用 valgrind 來測試執行所需的時間以及記憶體使用量，以下分別是三種不同方式的結果

• 在每次迭代時都做一次 cleanup

--------------------------------------------------------------------------------
  n       time(ms)         total(B)   useful-heap(B) extra-heap(B)    stacks(B)
--------------------------------------------------------------------------------
 55        113,229           28,488           27,600           888            0
 56        114,315           28,488           27,600           888            0
 57        115,402           28,488           27,600           888            0
 58        116,487           28,488           27,600           888            0
 59        117,580           28,184           27,316           868            0

• 只在最後做 cleanup

--------------------------------------------------------------------------------
  n       time(ms)         total(B)   useful-heap(B) extra-heap(B)    stacks(B)
--------------------------------------------------------------------------------
 80        112,131           33,600           30,580         3,020            0
 81        113,058           33,600           30,580         3,020            0
 82        113,985           33,600           30,580         3,020            0
 83        114,912           33,600           30,580         3,020            0
 84        116,460           30,728           29,012         1,716            0

• retired list 達一定數量後做 cleanup

--------------------------------------------------------------------------------
  n       time(ms)         total(B)   useful-heap(B) extra-heap(B)    stacks(B)
--------------------------------------------------------------------------------
 49        101,150           30,008           28,540         1,468            0

測試參數
N_READER = 100
N_WRITER = 1
N_ITER = 100,000
r_limit = 125 (N_READER * 1.25)

在記憶體使用上如預期的，第一種方式最少，第二種最多，而第三種則介於兩者之間，而在執行時間方面，第三種方法相較前面兩者得到約 14% 的進步。

lock-free hazard pointer

上方 hazard pointer 實作實際屬於 lock-free 的機制，而非 wait-free 的原因是在 reader thread 的 load。

uintptr_t load(domain_t *dom, const uintptr_t *prot_ptr)
{
    const uintptr_t nullptr = 0;

    while (1) {
        
        ...
            
        if (!atomic_cas(&node->ptr, &tmp, &nullptr))
            list_remove(&dom->pointers, val);
    }
}

load 在完成讀取並設定好 hazard pointer 之後，會再一次的對原本讀取的指標 p 內的值做一次比較，目的是避免在 讀取 到 設定 hazard pointer 之間被搶佔，並且 p 被釋放，導致 reader 後續使用時發生錯誤。而這樣的保護機制並不保證能夠在 constant time 裡完成，假設有一個 writer 一直頻繁的寫入新的值並釋放舊指標，那 load 即很有可能會長時間的困在 while 迴圈內。

void swap(domain_t *dom, uintptr_t *prot_ptr, uintptr_t new_val, int flags)
{
    // In multiple-writer situation, CAS replace exchange here
    const uintptr_t old_obj = atomic_exchange(prot_ptr, new_val);
    cleanup_ptr(dom, old_obj, flags);
}

同理在 multi-writer 的情況中， writer 也可能會在寫入的過程中與其他 writer 發生衝突，也就是在同一時間對同一個共享物件進行寫入，先完成的 writer 會成功，後完成的則會失敗。在部分實作中不會嘗試對失敗的寫入進行重新寫入，但如果有，則 writer 也並非 wait-free。

測驗 2 原本的程式碼適用於 single-writer-multiple-reader 的環境，因此不會有 writers 衝突的情況發生，用 atomic-exchange 是可行的

trap 機制

假設 writer 不會嘗試寫入直到成功，或是在 single-writer 的環境中，那 writer 就算是 wait-free 了，至於 reader 該如何成為 wait-free，〈Practical Lock-Free and Wait-Free LL/SC/VL Implementations Using 64-Bit CAS〉給出了一種途徑，也就是 trap 機制。

首先簡述 trap 的使用方式

• reader
  reader 會在嘗試設定 hazard pointer 失敗一次之後設置一個 trap，並重新設定一次 hazard pointer， seq (sequence number) 是物件內一個變數。
  接著檢查剛剛設置的 trap 中是否有符合條件的指標，若沒有，則表示該物件在重新設定的過程中沒有被變更，因此可以安全的讀取物件內的值，反之則表示內容被變更，但我們可以從 trap 內取得一個符合條件且尚未被釋放的指標來讀取其中的值，並把舊的指標設為 NULL 。
  最後，返回之前再把 trap 給釋放掉，以防其他指標落入 trap 中。
• writer
  writer 在寫入時，先取一塊新的區塊 b 來存放要寫入的值，接著讀取變數 seq，該變數用於計數該物件有幾次成功的寫入，當前為
  $se{q}_{th}$ ；同時將 b 的 seq 加一，表示當 b 成功寫入時，是為
  $seq+1_{th}$ 更新。

用在 trap 上，就只會捕捉大於等於自身 seq 的物件，表示只有在更新過 seq 次之後的物件才對 trap 是有效的，目的是防止讀取到舊的值。
接著 writer 利用 CAS 嘗試將新的值寫入，一旦失敗，整個寫入就算是結束了，而成功的時後則會對目前所有的 trap 做遍尋，確定所更動的物件是否有相對應的 trap。

圖片截取自論文第 8 頁

接著講解各項參數與實作細節。

一個 trap 由 SetTrap 函式建立，並設定其各項參數。

$Var$ 推測是指向一個物件的指標，表示這個 trap 只對該物件生效；

$Seq$ 的用途在前面已經提過；

$Captured$ 用來標記目前 trap 的狀態，以及捕捉到的物件之指標；

$traph{p}_p$ 本質上是 hazard pointer ，用途上也相同，保證直到 hazard pointer 被釋放為止，其所指向的空間不會被釋放；

$Active$ 用來標記 trap 是否被啟用，在這個函式中被寫為 true。

$ta{g}_p$ 被用在

$Captured$ 及

$traph{p}_p$ 的初始化之中，其值為 non-pointer value ，對 64-bits 電腦來說，一次取值的大小為 64-bits，也就是 8 bytes ，在位址上要做到 alignment ，就必須為 8 的倍數。因此只要知道位址的最後三位，就可以得知是否為 non-pointer value ， (詳細關於 data alignment，可參考 你所不知道的 C 語言：記憶體管理、對齊及硬體特性) 。當

$Captured$ 內的值為 non-pointer value 時，表示 trap 尚未捕捉到任何物件，且尚未被釋放。

trap 透過 ReleaseTrap 清除掉舊的資訊，但並不一定是釋放空間，而是等待下一次的使用，過程中先將

$Active$ 寫為 false ，此一條件也是後續 ScanTraps 判斷的第一個依據；接著將

$Captured$ 及

$traph{p}_p$ 寫入 NULL，表示 trap 已被釋放。

GetCapturedBlock 取出

$Captured$ 中的值，若是 non-pointer value ，表示 trap 未捕捉到任何物件，此時回傳 NULL 給 reader。若是一個 pointer value，則直接回傳。

ScanTrap 在流程上會逐一對每個 trap 檢查

$Active$ 是否為 true 以及

$Captured$ 中的值，確認 trap 的狀態是否還在運作，並且尚未捕捉到任何物件。接著讀取 trap 中的參數，並在 list 中尋找是否有符合條件的物件，若有則嘗試將找到的物件之指標寫入

$Captured$ 及

$traph{p}_p$ ，寫入失敗或是沒找到時，就換下一個 trap，最後結束前呼叫 RetireNode(list) 。
ScanTrap 在前述應用範例中，是每一次 writer 寫入成功就做一次，但這樣做的缺點是時間花費與寫入的成功次數呈線性關係，意味著寫入越多就越浪費時間。因此在論文中採取另一種方式，先累積一定數量因寫入成功被交換下來的舊物件，再一口氣做 ScanTrap。

圖片截取自論文第 11 頁