# Box64 紅黑樹效能分析

[rbtree_behavior](https://github.com/devarajabc/box64/commits/rbtree_behavior) (未完待續..)

目標：在 box64 執行時紀錄其紅黑樹的記憶體使用情形（"op_name, op_time, memory usage"）。在 box64 中安插「生產者」，將記錄的數據儲存在共享記憶體後由「消費者」來讀取，如此一來便可以獲得 box64 執行時的即時數據

## 分析 [ringbuf-shm](https://github.com/sysprog21/concurrent-programs/tree/master/ringbuf-shm) 的運作模式

測試：
執行 10000 次迴圈，利用隨機休眠的方式讓生產者和消費者執行的順序穿插且執行時間長短隨機，過程中生產者配置隨機大小的記憶體。
ringbuffer 本身配置的大小為 8192 bytes, 最多可以儲存 1024 個 `uint64_t`
而配置的記憶體大小為若干個 `uint64_t` ，記憶體內儲存的內容為迴圈的次數 `cnt`

---
#### 在樹莓派上測試 [ringbuf-shm](https://github.com/sysprog21/concurrent-programs/tree/master/ringbuf-shm)

NOTE: 測試前記得清空之前建立的共享記憶體空間

修改：為了要記錄 box64 紅黑樹的記憶體使用情況，使用以下結構來紀錄：

typedef struct record_item{
    char op_name[16];
    uint64_t Memory_usage;
    uint64_t op_time;
}record_item;

#### 修改生產者
在 ringbuffer 中紀錄`record_item` 在共享陣列中的 index：

uint64_t  *ptr;
if ((ptr = ringbuf_write_request_max(ringbuf, written, &maximum))) {
    assert(maximum >= written);
    *ptr = cnt%ARRAY_LENGTH;
    ringbuf_write_advance(ringbuf, written);
    } 

#### 建立共享陣列：
> 參考 [解析 Linux 共享記憶體機制](https://hackmd.io/@sysprog/linux-shared-memory)、CSAPP:3E ch9

步驟:
1. 由 ringbuffer 獲得欲存入的空間的 index (相當於在 `shared_array` 中預約一個空位) 或許能利用 tail 來獲得 index (即 `*ptr`)
2. 根據獲得的 index 對 `shared_array` 進行資料寫入

shared_array[*ptr].op_name = options[rand()%4];
shared_array[*ptr].op_time = get_time();
shared_array[*ptr].Memory_usage = rand()%500;

3. 消費者時根據 ringbuffer 內儲存的 index 來讀取 `shared_array` 的內容。
#### 驗證程式
需要確保資料的存入是按照時間順序的：

if(shared_array[*ptr].op_time < prev){
    printf("Saving error");
    munmap(shared_array, sizeof(record_item) * ARRAY_LENGTH);
    close(shm_fd);
    shm_unlink("/shm_array");
    exit(1);
}
prev = shared_array[*ptr].op_time;

----
#### 計算所需的 buffer 大小：
表格統計紅黑樹操作在三個 win7 程式中的數量：
|Name|Real Time|Add node|Add tree |Delete node|Delete tree|Total |Operations/sec|
|----|----|----|----|----|----|---|---|
|Chess|34.93|51041|19|28355|19|79434|2243.091|
|Solitaire|16.32|44972|18|22689|18|67697|4148.100|
|Minesweeper|14.69|45509|18|22718|18|68263|4646.902|

- 最糟情況：消費者完全不作為，即 ringbuffer 必須儲存完整的資訊，避免資訊遺失，此處選最大值 79434 * 1.2 (假設初始化階段的用量是全部的九成，即乘上 1.11... 無條件進位成 1.2 ) = 95320.8 算 95321 。將共享陣列設定成此長度能避免資訊遺失，但有必要這麼大嗎(約 93Kb)？
- 最好的情況：資料一存入 shared array 就被讀取，此時只需一個 bytes

估算：
估算的結過建立在以下假設：
1. 程式執行時的`Add node`, `Add tree`, `Delete tree`, `Delete node` 成固定比例(`Add_node` 和 `Delete node` 為 $2:1$)
2. 單位時間的最大執行數量與總執行數量成比例

以過去統計的資料而言，`Add node` 執行總次數為 7736 的程式在一秒內執行的最大次數為 1359 次，根據假設1，總執行數量約為 12023，根據假設2，一秒內的最大執行次數為 8979

根據上述的推算，再加上一些保險起見，我們可以將 shared array 的大小設為 9000

----
#### 整合至 box64

目標：能在 box64 運行時同時紀錄紅黑樹的資訊

在 box64 中有以下四個函式需要紀錄：`rbtree_init`, `delete_rbnode`, `rbtree_delete`, `add_range_next_to`

將 `ARRAY_LENGTH` 設為 9000

紀錄的過程如下:
1. 初始化 `ringbuffer` 和 `share_array`
2. 利用 `ringbuf_write_request_max` 獲得指向 64 位元的空間的指標 `ptr`
3. 利用 `ptr` 將 `(ringbuf->head)/16` 存入 `ringbuffer` 並作為 share_array 的索引值
4. 將紅黑樹操作的資訊存入根據索引值存入 shared_array

策略:
- 生產者 :在 box64 的 rbtree.c 中使用 ringbuffer.c 的 API ，即在 box64 中加入 ringbuffer.h （`ringbuf_shm_init`, `ringbuf_shm_deinit`, `ringbuf_write_request_max`, `ringbuf_write_advance`, `ringbuf_read_request`, `ringbuf_read_advance`），並利用 `(ringbuf->head/16)` 作為 shared array 的索引值。

void Saving(ringbuf_t *ringbuf, uint64_t *index, char *name, uint64_t Size, record_item *shared_array){
    size_t written = PAD(sizeof(uint64_t));
    size_t maximum;
    sem_wait(sender_sem);
    uint64_t *ptr;
    if ((ptr = ringbuf_write_request_max(ringbuf, written, &maximum))){
    *ptr = (ringbuf->head/16);
    //save record_item
    memcpy(shared_array[*ptr].op_name, name, strlen(name)+1);
    shared_array[*ptr].op_time = get_time();
    shared_array[*ptr].Memory_usage = Size;
    ringbuf_write_advance(ringbuf, written);
    }
    sem_post(sender_sem);
}

正確的數據應該擁有以下特性：
1. `Add tree` 和 `Delete tree` 的數據需與利用 printf 統計的數據一致
2. shared_array 內的元素必須連續
3. 索引值必需遞增且連續

問題：紀錄錯誤
以 chess 為例，若將紅黑樹相關操作用 printf 印出並統計，會得到

|Name|Add node|Add tree |Delete node|Delete tree|
|----|----|----|----|----|
|Chess|51041|19|28355|19|79434|

若在 `Saving` 中將儲存進 shared_array 的列印出來並統計

void Saving(ringbuf_t *ringbuf, uint64_t *index, char *name, uint64_t Size, record_item *shared_array){
    /*...skip...*/
printf("%lld, %s, %lld, %lld \n", *ptr,shared_array[*ptr].op_name, shared_array[*ptr].op_time, shared_array[*ptr].Memory_usage); 
    /*...skip...*/
}

則會得到：
|Name|Add node|Add tree |Delete node|Delete tree|
|----|----|----|----|----|
|Chess|53772|33|32021|19||

顯然數據有明顯的誤差，若仔細觀察 [input](https://docs.google.com/spreadsheets/d/1hwuo9NI-Do0J28GIdZThHGrOsrmWgSmv1r0st5npJZc/edit?gid=100635798#gid=100635798) 會注意到表格中的索引值並不連續：
0 ~ 113
215 ~ 895
1039 ~ 4553
82861 ~ 83570
84474 ~ 87036
...

可以看到這些索引值片段並不連續且長短不一，要知道索引值是直接使用 ringbuffer 的 head `(ringbuf->head)/16`，在統計數量不超過 ringbuffer 大小的情況下應該要是單調遞增才對（此次統計將 ringbuffer 大小設定為 150000 萬，遠遠大於統計量）。

從第 23165 筆資料開始又再度回歸索引值為 `1` ，但接下來就沒有索引值錯亂的問題，。

若將 input 的索引值對應到 [output](https://docs.google.com/spreadsheets/d/1ZaYwMoguxIL9OkwQjDBadydcTZJsCevnUP8329bX5k0/edit?usp=sharing) （正確的情況下應該要得到一對一的關係），會注意到 output 錯誤重複的數量又更多了(`Add tree` 的數量高達 109)，此處暗示了被污染的共享記憶體不只有 ringbuffer。

問題：為何 Saving 所使用的 semaphore 無法避免上述問題
問題：如何設計好的保護機制來保護 shared_array 和 ringbuffer

在 `main` 就進行初始化，不過 box64 的 main 在執行的過程中被呼叫了四次

直接檢查 shared_array 的內容會發現 `Add tree` 的數量為 35，依然對不上正確的數據，這說明了 shared_array 紀錄了錯誤的資訊，透過 rinbuffer 讀取又造成一次錯誤。

問題:若只是單純遭遇 race condition 的狀況，應該得到「較少」的資料，因為「紀錄資訊」的指令數量是固定的，只有可能「少紀錄到」，沒可能莫名其妙多出來，但此程式的錯誤輸出正是「更多」且不重複的輸出，究竟這些錯誤的記載是從哪裡來的呢？

未了解決上述問題，可將 [ringbuf-shm](https://github.com/sysprog21/concurrent-programs/tree/master/ringbuf-shm) 修改為多個生產者同時運作的情況，模擬 box64 執行時的場景

>僅靠 atomic 指令，無法達到對共用資源的存取控制，即使簡單如 spinlock 或 reference count，看上去正確的程式碼也可能會出錯。為何？因為 memory (re)ordering:

既然 ringbuf-shm 的設計已經考慮到 lock-free ，以此長景只需設計如何保護 shared_array

修改 `shared_array` 的結構：

typedef struct {
    record_item * shared_array;
    _Atomic int index;
    pthread_mutex_t lock;
    bool init_le_ma;
    const char name[32];
    int fd;
}s_array_t;

將 `shared_array` 初始化 ：

int shared_array_init(s_array_t* s_array, const char *name, size_t size){
    strcpy(s_array->name, name);
    if (! s_array->name)
        return -1;
    s_array->init_le_ma = false;
    s_array->fd = shm_open(s_array->name, O_RDWR | O_CREAT | O_EXCL, S_IRUSR | S_IWUSR);
    if (s_array->fd = -1){
        s_array->init_le_ma = true;
        s_array->fd = shm_open(s_array->name, O_RDWR, S_IRUSR | S_IWUSR);
    }
    if (s_array->fd == -1) {
        free(s_array->name);
        return -1;
    }
    if ((ftruncate(s_array->fd, size) == -1) ||
        ((s_array->shared_array = mmap(NULL, size, PROT_READ | PROT_WRITE,
                                      MAP_SHARED, s_array->fd, 0)) == MAP_FAILED)) {
        shm_unlink(s_array->name);
        close(s_array->fd);
        free(s_array->name);
        return -1;
    }
    if (!s_array->init_le_ma){
        s_array->index = 0;
        pthread_mutex_init(&s_array->lock, NULL);
    }
    return 0; 
}

要確認 shared_array 的大小！！
要確認 index 是否正確初始化！！！

接下來就可以修該 `Saving` :

void Saving(ringbuf_t *ringbuf, char *name, uint64_t Size, s_array_t *s_array){
    size_t written = PAD(sizeof(uint64_t));
    size_t maximum;
    pthread_mutex_lock(&s_array->lock);
    uint64_t *ptr;
    if ((ptr = ringbuf_write_request_max(ringbuf, written, &maximum))){
    *ptr = s_array->index;
    //save record_item
    memcpy(s_array->shared_array[*ptr].op_name, name, strlen(name)+1);
    s_array->shared_array[*ptr].op_time = get_time();
    s_array->shared_array[*ptr].Memory_usage = Size;
    ringbuf_write_advance(ringbuf, written);
    s_array->index += 1;
    }
    pthread_mutex_unlock(&s_array->lock);
}

`Reading`:

void Reading(ringbuf_t *ringbuf, s_array_t *s_array){
    while (1){
    pthread_mutex_lock(&s_array->lock);
    const uint64_t *ptr;
    size_t toread;
    if ((ptr = ringbuf_read_request(ringbuf, &toread))) {
        printf("%s, %lld, %lld \n",s_array->shared_array[*ptr].op_name, s_array->shared_array[*ptr].op_time, s_array->shared_array[*ptr].Memory_usage);
        ringbuf_read_advance(ringbuf);  
    }
    pthread_mutex_unlock(&s_array->lock);
    }
}

#### ringbuf-shm 實驗
>[test](https://github.com/devarajabc/concurrent-programs-for-ringbuffer/commits/master/ringbuf-shm)

box64 在執行時會 fork 兩次必建立多個執行緒，因此要修改 ringbuf-shm 以進行對真實情況的模擬。

- 索引值：
和使用共享變數 `index` 作為 `shared_array` 的索引值會產生不一樣的解果。

以 `ARRAY_LENGTH = 128` 為例，使用 `index` 作為索引值時的範圍是 `0 ~ 127` ，使用 `(ringbuf -> head)/16` 的範圍則是 `0 ~ 63` ，同理 `ARRAY_LENGTH = 256` 時索引值的範圍則為 `0 ~ 127`。 why ?

- 檢查是否支援同時執行多個生產者與消費者 (MPSC)
可以，但有些細節會出錯，見 [Commit 70979f9
](https://github.com/devarajabc/concurrent-programs-for-ringbuffer/commit/70979f996e9e46ad8cec176aff06ba087cd2d421)
錯誤：
1.

    ringbuffer: ringbuffer.c:191: ringbuf_write_advance:     Assertion `written <= ringbuf->rsvd' failed.
    Aborted (core dumped)

2. 輸出結果不一致

如何驗證生產者和消費者儲存正確？
1. 多個執行緒上的生產者將 `ringbuf->head/16` 紀錄在 `ringbuffer` 中。
2. 檢查是存入的數值是否照順序 `assert(*(const uint64_t *) src == cnt)`

參考 [mpmc](https://github.com/sysprog21/concurrent-programs/blob/master/mpmc/mpmc.c)

參考

1. [ringbuf-shm](https://github.com/sysprog21/concurrent-programs/tree/master/ringbuf-shm)
2. [AddressSanitizer](https://clang.llvm.org/docs/AddressSanitizer.html)
3. [並行程式設計: 執行順序](https://hackmd.io/@sysprog/concurrency/%2F%40sysprog%2Fconcurrency-ordering)
4. [Mutex-Locks](https://github.com/angrave/SystemProgramming/wiki/Synchronization,-Part-1:-Mutex-Locks)
5. [ThreadSanitizer](https://clang.llvm.org/docs/ThreadSanitizer.html)
6. [你所不知道的 C 語言：記憶體管理、對齊及硬體特性](https://hackmd.io/@sysprog/c-prog/%2Fs%2FBkuMDQ9K7)