2021q3 Homework2 (hp)

contributed by < u1f383 >

1. 解釋上述程式碼運作原理

首先介紹 macro 以及資料結構:

#define HP_MAX_THREADS 128 /* hp 能 handle 的最大 thread 數量 */
#define HP_MAX_HPS 5 /* 最多可以維護多少個 hp */
/**
 * 為了避免 false sharing 的問題，必須將 struct padding 至 cacheline 的倍數 (64 * n)
 * CLPAD 應該是 cacheline padding 的縮寫
 * > 看了人家的紀錄，intel 一次 cache 的資料量為 128 bytes
 */
#define CLPAD (128 / sizeof(uintptr_t)) // 16
#define HP_THRESHOLD_R 0 /* This is named 'R' in the HP paper */

/* 每個 thread 擁有的最大 retired objects 數量 */
#define HP_MAX_RETIRED (HP_MAX_THREADS * HP_MAX_HPS)
#define TID_UNKNOWN -1 /* 每個 thread 會配發一個 tid，而在還沒配發之前都是 -1 */

typedef struct {
    int size;
    uintptr_t *list;
} retirelist_t;

typedef struct list_hp list_hp_t;
typedef void(list_hp_deletefunc_t)(void *);

/**
 * hazard pointer 的 list 結構
 * @max_hps: 該 list 中最多能有多少 hp
 * @hp: 儲存指向 hp, array 的 index 代表對應的 thread
 * @rl: 指向 retire_list, array 的 index 代表對應的 thread
 * @deletefunc: 指向可以用來刪除 hp node 的 function
 */
struct list_hp {
    int max_hps;
    /* atomic_uintptr_t 可以想像成就是 unsigned long */
    alignas(128) atomic_uintptr_t *hp[HP_MAX_THREADS];
    /* 每個 element 間隔 128 bytes */
    alignas(128) retirelist_t *rl[HP_MAX_THREADS * CLPAD];
    list_hp_deletefunc_t *deletefunc;
};

/**
 * 設定 tid_v 為 per-thread data，一開始的初始值即為 -1，
 * 用來代表每個 thread (類似 thread id)
 */
static thread_local int tid_v = TID_UNKNOWN;
/* 此 function 用來初始化 atomic type variable */
static atomic_int_fast32_t tid_v_base = ATOMIC_VAR_INIT(0);

新建一個大小為 HP_MAX_HPS 的 hp array，並且考慮到 cacheline 作了許多優化，可謂說以空間 (memory space) 換取時間 (no false sharing):

/* Create a new hazard pointer array of size 'max_hps' (or a reasonable
 * default value if 'max_hps' is 0). The function 'deletefunc' will be
 * used to delete objects protected by hazard pointers when it becomes
 * safe to retire them.
 */
list_hp_t *list_hp_new(size_t max_hps, list_hp_deletefunc_t *deletefunc)
{
    list_hp_t *hp = aligned_alloc(128, sizeof(*hp));
    assert(hp);

    /* 如果沒有給定 max_hps，預設值就是 HP_MAX_HPS (5) */
    if (max_hps == 0)
        max_hps = HP_MAX_HPS;

    /**
     * 初始化 list_hp 部分的 member，仍有 hp 以及 rl 尚未初始化
     * 而這邊 deletefunc 會是 __list_node_delete
     */
    
    *hp = (list_hp_t){.max_hps = max_hps, .deletefunc = deletefunc};

    for (int i = 0; i < HP_MAX_THREADS; i++) {
        /**
         * 猜測 CLPAD * 2 的原因是 hp->hp[i][0] 為 uint (4 bytes)，
         * 原本是以 sizeof(uintptr_t) 作為最小單位，現在必須改成 4 bytes，
         * 因此 padding 的 size 必須要 * 2
         * 
         * hp 一共有 [HP_MAX_THREADS][128]
         **/
        hp->hp[i] = calloc(CLPAD * 2, sizeof(hp->hp[i][0]));
        /**
         * rl 每個 CLPAD 為一個單位去分配記憶體，而資料型態又是 uintptr_t，
         * 因此每個 rl element 間隔 128 bytes
         */
        hp->rl[i * CLPAD] = calloc(1, sizeof(*hp->rl[0]));
        for (int j = 0; j < hp->max_hps; j++)
            /* 將初始值都設為 0 */
            atomic_init(&hp->hp[i][j], 0);
        /* 每個 thread 分配 (最大 thread 數 * 最多 HP 數) 的記憶體空間 */
        hp->rl[i * CLPAD]->list = calloc(HP_MAX_RETIRED, sizeof(uintptr_t));
    }

    return hp;
}

當 hp array 使用完畢，在結束程式前必須將其刪除以釋放空間:

/* Destroy a hazard pointer array and clean up all objects protected
 * by hazard pointers.
 */
void list_hp_destroy(list_hp_t *hp)
{
    /**
     * 這邊就是清空每個 hp array element 使用到的資源，
     * 不過到這仍不確定每個 data struct 的使用方式 / 行為
     */
    for (int i = 0; i < HP_MAX_THREADS; i++) {
        free(hp->hp[i]);
        retirelist_t *rl = hp->rl[i * CLPAD];
        for (int j = 0; j < rl->size; j++) {
            void *data = (void *) rl->list[j];
            hp->deletefunc(data);
        }
        free(rl->list);
        free(rl);
    }
    free(hp);
}

當某個 thread 在 hp 使用完畢後，必須要清空 hp 而非刪除，只需要把 hp list 對到的 hp 設為 0 即可:

/* Clear all hazard pointers in the array for the current thread.
 * Progress condition: wait-free bounded (by max_hps)
 */
void list_hp_clear(list_hp_t *hp)
{
    for (int i = 0; i < hp->max_hps; i++)
        atomic_store_explicit(&hp->hp[tid()][i], 0, memory_order_release);
}

當某 thread 正在使用 ptr 時，可以將 ptr 加進 hp list 當中保護，其他 thread 在 release node 時會先檢查是否在其他 thread 的 hp 內，如果沒有的話才會 release:

/* This returns the same value that is passed as ptr.
 * Progress condition: wait-free population oblivious.
 */
uintptr_t list_hp_protect_ptr(list_hp_t *hp, int ihp, uintptr_t ptr)
{
    /* ihp 為 hp index */
    atomic_store(&hp->hp[tid()][ihp], ptr);
    return ptr;
}

操作完畢後，若 ptr 沒有要受到保護時就可以執行 protect_release，從 hp list 當中移除:

/* Same as list_hp_protect_ptr(), but explicitly uses memory_order_release.
 * Progress condition: wait-free population oblivious.
 */
uintptr_t list_hp_protect_release(list_hp_t *hp, int ihp, uintptr_t ptr)
{
    atomic_store_explicit(&hp->hp[tid()][ihp], ptr, memory_order_release);
    return ptr;
}

把一個不再需要用到的 ptr 給釋放，但是釋放前必須再三確定沒人在使用:

/* Retire an object that is no longer in use by any thread, calling
 * the delete function that was specified in list_hp_new().
 *
 * Progress condition: wait-free bounded (by the number of threads squared)
 */
void list_hp_retire(list_hp_t *hp, uintptr_t ptr)
{
    /* 取得該 thread 的 retire list */
    retirelist_t *rl = hp->rl[tid() * CLPAD];
    /* 新增一個要被釋放的 object */
    rl->list[rl->size++] = ptr;
    assert(rl->size < HP_MAX_RETIRED);

    /**
     * 當數量小於 HP_THRESHOLD_R，就不先 retire object
     * 而是等到 threshold 到達時在一次釋放
     */
    if (rl->size < HP_THRESHOLD_R)
        return;

    for (size_t iret = 0 /* iret == retire index (?) */; iret < rl->size; iret++) {
        uintptr_t obj = rl->list[iret];
        bool can_delete = true;
        /* 遍歷每個 thread */
        for (int itid = 0; itid < HP_MAX_THREADS && can_delete; itid++) {
            /* 遍歷每個 thread 的 hp list*/
            for (int ihp = hp->max_hps - 1; ihp >= 0; ihp--) {
                /* reverse order 檢查有沒有在 hp 當中，如果有的話就不能 delete */
                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]);
            /**
             * 從 &rl->list[iret + 1] copy "bytes" 數量的資料到 &rl->list[iret]，
             * 意思就是把之後的資料整個往前挪
             */
            memmove(&rl->list[iret], &rl->list[iret + 1], bytes);
            rl->size--;
            // RRR;
            hp->deletefunc((void *) obj); iret--;
        }
    }
}

而之後的部分屬於 linked list 的實作，一樣先看使用到哪些 macro、function prototype 以及 data structure:

#include <pthread.h>

#define N_ELEMENTS 128 /* list 最多的 element 數量 */
#define N_THREADS (64) /* 使用多少 thread */
/**
 * NNN = 64 的原因在於，輸出結果為 4098，而其中兩個 node 為頭尾，
 * 每個 thread 則會 insert 128 個，因此 (4098 - 2) / 128 = 32，
 * 但是實際上 delete 也占了一半，因此總共 thread 的數量為 64
 */
#define MAX_THREADS 128

/* 用於紀錄刪除以及插入的次數 */
static atomic_uint_fast32_t deletes = 0, inserts = 0;

/**
 * hp 的 enum，感覺應該要放在上面 (?)，也有可能是因為現在這個部分
 * 比較偏向 user 自己定義
 */
enum { HP_NEXT = 0, HP_CURR = 1, HP_PREV };

#define is_marked(p) (bool) ((uintptr_t)(p) &0x01)
#define get_marked(p) ((uintptr_t)(p) | (0x01))
#define get_unmarked(p) ((uintptr_t)(p) & (~0x01))

#define get_marked_node(p) ((list_node_t *) get_marked(p))
#define get_unmarked_node(p) ((list_node_t *) get_unmarked(p))

typedef uintptr_t list_key_t;

typedef struct list_node {
    alignas(128) uint32_t magic; /* 放 magic number (0xdeadbeef) */
    alignas(128) atomic_uintptr_t next;
    list_key_t key;
} list_node_t;

/* Per list variables */

typedef struct list {
    atomic_uintptr_t head, tail;
    /* 每個 linked list 都會 maintain 一個 hp array */
    list_hp_t *hp;
} list_t;

#define LIST_MAGIC (0xDEADBEAF) /* list element 的 magic number */

接下來是一些常見的 linked list function，像是 delete / insert / find / new 等等。

刪除 node 使用的是 list_node_destroy():

void list_node_destroy(list_node_t *node)
{
    if (!node) /* invalid */
        return;
    /* 檢查 node 是否有被意外改寫 */
    assert(node->magic == LIST_MAGIC);
    free(node);
    /* delete 成功時 delete counter + 1 */
    (void) atomic_fetch_add(&deletes, 1);
}

而 __list_node_delete() 為 list_node_destroy() 的 wrapper function，同時也是 hp 的 deletefunc:

static void __list_node_delete(void *arg)
{
    list_node_t *node = (list_node_t *) arg;
    list_node_destroy(node);
}

find node function 可以說是 hp 的關鍵，用來找 key 對應到的 node 是否存在於 list 當中，並且過程中並無使用 lock:

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;
    curr = (list_node_t *) atomic_load(prev);
    /* 告訴大家正在使用 curr 此 ptr */
    (void) list_hp_protect_ptr(list->hp, HP_CURR, (uintptr_t) curr);
    /**
     * 在過程中原本的 curr (old list->head) 被更新，必須重來，
     * 因為有可能此時 curr (old list->head) 已經被 delete 了
     */
    if (atomic_load(prev) != get_unmarked(curr))
        goto try_again;
    /* 確保目前 curr 不會被動到 */
    while (true) {
        /* curr 是 NULL ptr，代表整個 list 已經空了 */
        if (!get_unmarked_node(curr))
            return false;
        /* 取得 curr->next */
        next = (list_node_t *) atomic_load(&get_unmarked_node(curr)->next);
        /* 一樣加到 hp array 當中保護 */
        (void) list_hp_protect_ptr(list->hp, HP_NEXT, get_unmarked(next));
        /**
         * 結果在加到 hp 的過程中 next 又被更動，代表已經被 delete 掉，
         * 因為只有 list_delete() 會動到 curr->next
         * ...應該吧 @_@
         */
        if (atomic_load(&get_unmarked_node(curr)->next) != (uintptr_t) next)
            break;
        /**
         * next 為 curr->next，也就是 list->head->next，如果接到 tail，
         * 代表 list 是空的
         */
        if (get_unmarked(next) == atomic_load((atomic_uintptr_t *) &list->tail))
            break;
        /* 在過程中 list->head 又被更動，可能已經不一樣了，所以必須重新來過 */
        /* 因為我們用 hp 保護 curr 以及 next，因此不需要考慮到他們被 delete 的狀況 */
        if (atomic_load(prev) != get_unmarked(curr))
            goto try_again;
        /* 當 next 被 marked 時 condition 為 false */
        if (get_unmarked_node(next) == next) {
            /* 當前 node 的 key >= 要找的 key */
            if (!(get_unmarked_node(curr)->key < *key)) {
                /* par 是 parallel 的意思嗎 ? */
                *par_curr = curr;
                *par_prev = prev;
                *par_next = next;
                /* 看是不是要找的 node */
                // return FFF;
                return get_unmarked_node(curr)->key == *key;
            }
            /* prev 更新成下一個 node @_@? */
            prev = &get_unmarked_node(curr)->next;
            /* 將 curr 設為 hp array 中的 PREV 作為保護 */
            (void) list_hp_protect_release(list->hp, HP_PREV,
                                           get_unmarked(curr));
        } else {
            
            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，代表要繼續 traverse */
        curr = next;
        /**
         * 更新 curr 並用 hp array 作保護，
         * 而 release 代表用 memory_order_release
         */
        (void) list_hp_protect_release(list->hp, HP_CURR, get_unmarked(next));
    }
    /* 沒有很懂這三個是什麼 @_@ */
    *par_curr = curr;
    *par_prev = prev;
    *par_next = next;

    return false;
}

insert node 時呼叫 list_insert():

bool list_insert(list_t *list, list_key_t key)
{
    list_node_t *curr = NULL, *next = NULL;
    atomic_uintptr_t *prev = NULL;
    /* 新增一個 node */
    list_node_t *node = list_node_new(key);

    while (true) {
        if (__list_find(list, &key, &prev, &curr, &next)) {
            /* 如果 node 重複，就不新增，釋放 node 以及清空 hp */
            list_node_destroy(node);
            list_hp_clear(list->hp);
            return false;
        }
        /* 如果沒找到對應 key 的 node，就將 node 加在 list 的一開始 */
        atomic_store_explicit(&node->next, (uintptr_t) curr,
                              memory_order_relaxed);
        uintptr_t tmp = get_unmarked(curr);
        /**
         * 如果 prev == curr 則 prev = node，不過 prev 也沒用到，
         * 為什麼不直接 compare 就好 ?
         */
        if (atomic_compare_exchange_strong(prev, &tmp, (uintptr_t) node)) {
            /* 確實新增成功，清空 hp array 後 return */
            list_hp_clear(list->hp);
            return true;
        }
    }
}

使用 list_delete() 刪除 node:

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)) {
            /* 如果找不到對應的 node 也不需要 delete 了 */
            list_hp_clear(list->hp);
            return false;
        }

        uintptr_t tmp = get_unmarked(next);

        /* 將 next mark 起來，代表上一個點有被動到 */
        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);
            // DDD;
            /* 將當前的點 retire，表示刪除 */
            list_hp_retire(list->hp, get_unmarked(curr));
        } else {
            list_hp_clear(list->hp);
        }
        return true;
    }
}

新增新的 list 時呼叫 list_new():

list_t *list_new(void)
{
    /* 分配一個 list 大小的空間 */
    list_t *list = calloc(1, sizeof(*list));
    assert(list);
    /* 新增頭尾兩個點 */
    list_node_t *head = list_node_new(0),
                *tail = list_node_new(UINTPTR_MAX /* 0xffffffff */);
    assert(head), assert(tail);
    /* 新增一個 hp */
    list_hp_t *hp = list_hp_new(3, __list_node_delete);

    /**
     * list->head -> head ---next--
     *                             |
     * list->tail -> tail <--------|
     */
    atomic_init(&head->next, (uintptr_t) tail);
    *list = (list_t){.hp = hp}; /* assign hp */
    atomic_init(&list->head, (uintptr_t) head);
    atomic_init(&list->tail, (uintptr_t) tail);

    return list;
}

而 insert_thread() 以及 delete_thread() 為 thread 執行的兩個 function，主要就是持續新增以及刪除 node:

static uintptr_t elements[MAX_THREADS + 1][N_ELEMENTS];

static void *insert_thread(void *arg)
{
    list_t *list = (list_t *) arg;

    for (size_t i = 0; i < N_ELEMENTS; i++)
        (void) list_insert(list, (uintptr_t) &elements[tid()][i]);
    return NULL;
}

static void *delete_thread(void *arg)
{
    list_t *list = (list_t *) arg;

    for (size_t i = 0; i < N_ELEMENTS; i++)
        (void) list_delete(list, (uintptr_t) &elements[tid()][i]);
    return NULL;
}

main():

int main(void)
{
    list_t *list = list_new(); /* 建立 + 初始化 list */

    pthread_t thr[N_THREADS];

    /* delete_thread 以及 insert_thread 各半 */
    for (size_t i = 0; i < N_THREADS; i++)
        pthread_create(&thr[i], NULL, (i & 1) ? delete_thread : insert_thread,
                       list);

    for (size_t i = 0; i < N_THREADS; i++)
        pthread_join(thr[i], NULL);

    /* 確實將每個 node delete 乾淨 */
    for (size_t i = 0; i < N_ELEMENTS; i++) {
        for (size_t j = 0; j < tid_v_base; j++)
            list_delete(list, (uintptr_t) &elements[j][i]);
    }
    
    /* 釋放 list 的資源 */
    list_destroy(list);

    fprintf(stderr, "inserts = %zu, deletes = %zu\n", atomic_load(&inserts),
            atomic_load(&deletes));

    return 0;
}

2. 指出改進空間並著手實作

執行list_hp_retire() 時，會將即將被刪除的 node 之後的資料往前移動，不過 iret 在下一個 iteration 時還是會執行 iret++，可是 rl->list[iret] 還沒處理過，因此加上 iret-- 使下次 iteration 時能走訪到還沒被處理的資料

hp->deletefunc((void *) obj); iret--;

P.S. __list_find() 的實作看不是很懂，所以也沒辦法做更深入的優化

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

程式碼中定義 rcu 結構的部分:

/* A concurrent linked list utilizing the simplified RCU algorithm */

#include <stdbool.h>

typedef struct rcu_list rcu_list_t;

/* 用來遍歷整個 list 的 iterator type */
typedef struct {
    struct list_node *entry;
} iterator_t;

typedef struct {
    struct rcu_list *list;
    struct zombie_node *zombie;
} rcu_handle_t;

typedef rcu_handle_t read_handle_t;
typedef rcu_handle_t write_handle_t;

typedef void (*deleter_func_t)(void *);
typedef bool (*finder_func_t)(void *, void *);

#define _GNU_SOURCE
#include <pthread.h>
#include <stdbool.h>
#include <stdlib.h>

/* list node 的資料結構，為雙向的 linked list */
typedef struct list_node {
    bool deleted;
    struct list_node *next, *prev;
    void *data;
} list_node_t;

/* 為單向的 linked list */
typedef struct zombie_node {
    struct zombie_node *next;
    struct list_node *zombie;
    rcu_handle_t *owner;
} zombie_node_t;

/**
 * rcu list 的資料結構
 * @write_lock: 用來限制只能有一個 writer 的 mutex lock
 * @head, tail: 指向 list 頭尾的 pointer
 * @zombie_head: 指向 zombie list
 * @deleter: 為指向 delete function 的 function pointer
 */
struct rcu_list {
    pthread_mutex_t write_lock; /* exclusive lock acquired by writers */
    list_node_t *head, *tail;   /* head and tail of the "live" list */
    zombie_node_t *zombie_head; /* head of the zombie list */
    deleter_func_t deleter;
};

static list_node_t *make_node(void *data);
static zombie_node_t *make_zombie_node(void);
static void lock_for_write(rcu_list_t *list);
static void unlock_for_write(rcu_list_t *list);

新增一個初始化 delete function pointer 的 list:

static rcu_list_t *list_new_with_deleter(deleter_func_t deleter)
{
    if (!deleter)
        return NULL;

    rcu_list_t *list = malloc(sizeof(rcu_list_t));
    if (!list)
        return NULL;

    /* 初始化 mutex lock */
    if (pthread_mutex_init(&list->write_lock, NULL) != 0) {
        free(list);
        return NULL;
    }

    list->head = list->tail = NULL;
    list->zombie_head = NULL;
    /* 初始化 delete func ptr 指向 deleter */
    list->deleter = deleter;

    return list;
}

釋放 rcu list:

static void list_free(void *arg)
{
    rcu_list_t *list = arg;
    /* iterator 遍歷整格 list，一個個刪除 rcu 上的 node */
    for (list_node_t *iter = list->head; iter;) {
        list_node_t *tmp = iter->next;
        free(iter->data);
        free(iter);
        iter = tmp;
    }
    free(list);
}

新增一個以 list_free() 作為 delete function 的 list，因此在這個例子中 delete == free:

rcu_list_t *list_new(void)
{
    return list_new_with_deleter(list_free);
}

list delete wrapper:

void list_delete(rcu_list_t *list)
{
    /* 檢查 list 以及 list->deleter 存在 */
    if (!list || !list->deleter)
        return;
    list->deleter(list);
}

新增一筆資料到 list 中，而過程中會先為資料新增一個 node，並以 node 的形式加到 list 當中:

void list_push_front(rcu_list_t *list, void *data, write_handle_t *handle)
{
    if (!list)
        return;

    /* 為 data 新增一個 node */
    list_node_t *node = make_node(data);
    if (!node)
        return;

    /* 為 list 上鎖，確保一次只有一個 writer (updater) */
    /* 都用 lock 了，為什麼還需要 atomic ? */
    lock_for_write(list);

    list_node_t *old_head;
    __atomic_load(&list->head, &old_head, __ATOMIC_RELAXED);

    if (!old_head) {
        /* list is currently empty */
        /* empty list，頭尾直接指向新的 node */
        __atomic_store(&list->head, &node, __ATOMIC_SEQ_CST);
        __atomic_store(&list->tail, &node, __ATOMIC_SEQ_CST);
    } else {
        /* general case */
        /* 將 node 加到 list 的開頭 */
        __atomic_store(&node->next, &old_head, __ATOMIC_SEQ_CST);
        __atomic_store(&old_head->prev, &node, __ATOMIC_SEQ_CST);
        __atomic_store(&list->head, &node, __ATOMIC_SEQ_CST);
    }

    unlock_for_write(list);
}

list 找是否有某個 node maintain 特定的 data:

iterator_t list_find(rcu_list_t *list,
                     void *data,
                     finder_func_t finder,
                     read_handle_t *handle)
{
    iterator_t iter = {.entry = NULL}; /* initialize an invalid iterator */
    if (!list)
        return iter;

    list_node_t *current;
    __atomic_load(&list->head, &current, __ATOMIC_SEQ_CST);

    /* traverse 所有的 list node，並利用 finder 來檢測資料是否相同 */
    while (current) {
        if (finder(current->data, data)) {
            iter.entry = current;
            break;
        }

        __atomic_load(&current->next, &current, __ATOMIC_SEQ_CST);
    }
    return iter;
}

指向 list 開頭的 iterator:

/* handler 根本沒被執行 @_@ */
iterator_t list_begin(rcu_list_t *list, read_handle_t *handle)
{
    iterator_t iter = {.entry = NULL};
    if (!list)
        return iter;

    list_node_t *head;
    __atomic_load(&list->head, &head, __ATOMIC_SEQ_CST);

    iter.entry = head;
    return iter;
}

回傳 iterator 指向的 data:

void *iterator_get(iterator_t *iter)
{
    /* 基本檢測 iter 以及 iter->entry 是否存在 */
    return (iter && iter->entry) ? iter->entry->data : NULL;
}

為 list 註冊 reader / writer handler:

read_handle_t list_register_reader(rcu_list_t *list)
{
    read_handle_t handle = {.list = list, .zombie = NULL};
    return handle;
}

write_handle_t list_register_writer(rcu_list_t *list)
{
    write_handle_t handle = {.list = list, .zombie = NULL};
    return handle;
}

rcu_read_lock() 看起來是在新增一個 zombie node，猜測是用 zombie node 作為 reader 在過渡 update 的時候存取的值:

void rcu_read_lock(read_handle_t *handle)
{
    /* 新增一個 zombie node */
    zombie_node_t *z_node = make_zombie_node();

    /* 標記 zombie node 的主人 */
    z_node->owner = handle;
    /* 更新該 handler 指向的 zombie node */
    handle->zombie = z_node;

    rcu_list_t *list = handle->list;

    zombie_node_t *old_head;
    /* 取得當前的 zombie_head */
    __atomic_load(&list->zombie_head, &old_head, __ATOMIC_SEQ_CST);

    do {
        /**
         * 將新增的 node 的 next 指向當前的 zombie_head，
         * 並且嘗試將自己更新成 zombie head
         */
        __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));
}

unlock 呢，既然 lock 是以 zombie node 作為記號， unlock 應該是將 zombie node 移除:

void rcu_read_unlock(read_handle_t *handle)
{
    /* 取得 handle maintain 的 zombie node */
    zombie_node_t *z_node = handle->zombie;

    zombie_node_t *cached_next;
    __atomic_load(&z_node->next, &cached_next, __ATOMIC_SEQ_CST);

    bool last = true;

    /* walk through the zombie list to determine if this is the last active
     * reader in the list.
     */
    zombie_node_t *n = cached_next;
    /**
     * 走訪 handle->zombie 之後的 node，看是否 handle->zombie
     * 是最後一個 active 的 node
     */
    while (n) {
        list_node_t *owner;
        __atomic_load(&n->owner, &owner, __ATOMIC_SEQ_CST);

        /* 後面還有 zombie 屬於某個 handler，代表不是最後一個 */
        if (owner) {
            last = false; /* this is not the last active reader */
            break;
        }

        __atomic_load(&n->next, &n, __ATOMIC_SEQ_CST);
    }

    n = cached_next;

    if (last) {
        /* 如果是最後一個，就能安全地將後面每個 zombie node delete */
        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);
    }

    /* 如果還有人在 read，則就不需要將後面的 node delete */
    void *null = NULL;
    __atomic_store(&z_node->owner, &null, __ATOMIC_SEQ_CST);
}

write lock / unlock wrapper，但其實還是執行 read lock/unlock:

void rcu_write_lock(write_handle_t *handle)
{
    rcu_read_lock(handle);
}

void rcu_write_unlock(write_handle_t *handle)
{
    rcu_read_unlock(handle);
}

新增一個 list node:

static list_node_t *make_node(void *data)
{
    list_node_t *node = malloc(sizeof(list_node_t));
    if (!node)
        return NULL;

    node->data = data;
    node->next = node->prev = NULL;
    node->deleted = false;

    return node;
}

新增一個 zombie node:

static zombie_node_t *make_zombie_node(void)
{
    zombie_node_t *z_node = malloc(sizeof(zombie_node_t));
    if (!z_node)
        return NULL;

    z_node->zombie = NULL;
    z_node->owner = NULL;
    z_node->next = NULL;

    return z_node;
}

mutex lock wrapper:

static void lock_for_write(rcu_list_t *list)
{
    pthread_mutex_lock(&list->write_lock);
}

static void unlock_for_write(rcu_list_t *list)
{
    pthread_mutex_unlock(&list->write_lock);
}

測試執行的部分，以 dummy struct 作為 data 的單位:

#include <assert.h>
#include <stdio.h>
#include <stdlib.h>

typedef struct dummy {
    int a, b;
} dummy_t;

static dummy_t *make_dummy(int a, int b)
{
    dummy_t *dummy = malloc(sizeof(dummy_t));
    dummy->a = a, dummy->b = b;
    return dummy;
}

static bool finder(void *x, void *y)
{
    /* 比對 dummy 的 a b 值是否相同 */
    dummy_t *dx = x, *dy = y;
    return (dx->a == dy->a) && (dx->b == dy->b);
}

reader thread 要執行的 function:

static void *reader_thread(void *arg)
{
    rcu_list_t *list = arg;
    /* 註冊 reader handle，回傳一個 read_handle */
    read_handle_t reader = list_register_reader(list);

    rcu_read_lock(&reader);

    /* read from list here */
    /* 找有沒有 {1, 1} 的 dummy */
    iterator_t iter = list_find(list, &(dummy_t){1, 1}, finder, &reader);
    void *data = iterator_get(&iter);
    assert(data);

    /* 找有沒有 {2, 2} 的 dummy */
    iter = list_find(list, &(dummy_t){2, 2}, finder, &reader);
    data = iterator_get(&iter);
    assert(data);

    /* 取得第一個 list node */
    iterator_t first = list_begin(list, &reader);
    data = iterator_get(&first);
    assert(data);

    dummy_t *as_d2 = data;
    assert(as_d2->a == 2);
    assert(as_d2->b == 2);

    assert(iter.entry == first.entry);

    rcu_read_unlock(&reader);
    return NULL;
}

writer thread 要執行的 function:

static void *writer_thread(void *arg)
{
    dummy_t *d1 = make_dummy(1, 1);
    dummy_t *d2 = make_dummy(2, 2);

    rcu_list_t *list = arg;
    /* 為 list 註冊 writer handle，回傳一個 write_handle */
    write_handle_t writer = list_register_writer(list);

    rcu_write_lock(&writer);

    /* write to list here */
    /* writer 必須要 mutex，確保一次只會有一個 writer */
    list_push_front(list, d1, &writer);
    list_push_front(list, d2, &writer);

    rcu_write_unlock(&writer);
    return NULL;
}

main function:

#define N_READERS 10

int main(void)
{
    rcu_list_t *list = list_new();
    dummy_t *d0 = make_dummy(0, 0);
    list_push_front(list, d0, NULL);

    pthread_t t0, t_r[N_READERS];
    /* 一個 writer */
    pthread_create(&t0, NULL, writer_thread, list);
    /* N_READERS 個 reader */
    for (int i = 0; i < N_READERS; i++)
        pthread_create(&t_r[i], NULL, reader_thread, list);

    for (int i = 0; i < N_READERS; i++)
        pthread_join(t_r[i], NULL);
    pthread_join(t0, NULL);

    list_delete(list);

    return EXIT_SUCCESS;
}

TODO