# 2024q1 quiz8
contribute by `SimonLee0316`

## 測驗1
[題目](https://hackmd.io/@sysprog/linux2024-quiz8)

[task](https://gist.github.com/jserv/745dab1d9be33a9a635b193aeb53191e) 是個運用 coroutine 實作的任務排程器，允許搭配 POSIX Thread 使用，使用 C11 Atomics 和 x86-64 組合語言開發。

## test_task_start

void test_task_start(void)
{
    /*...*/
    quick_start(co_root_entry, co_cleanup, &(int){99999999});
    /*...*/
}

執行 co_root_entry 了，在裡面會不斷的新建 child coroutine。

void co_child_entry(void *udata)
{
    int udid = *(int *) udata;
    assert(udid == nudid);
}
void co_root_entry(void *udata)
{
    assert(*(int *) udata == 99999999);
    assert(*(int *) task_udata() == 99999999);
    assert(task_info_running() == 1);
    for (int i = 0; i < NCHILDREN; i++) {
        printf("start child %d\n",i);
        assert(started == 1 + i);
        assert(cleaned == 1 + i - 1);
        nudid = i;
        quick_start(co_child_entry, co_cleanup, &(int){i});
    }
}

運行順序如下。

test_task_start
start co_root_entry
start child 0
start co_child_entry udid: 0
start child 1
start co_child_entry udid: 1
start child 2
...
start child 99
start co_child_entry udid: 99

總結來說，`do_test(test_task_start)`，會建立多個　coroutine 。

## test_task_sleep
透過 quick_start 呼叫 co_sleep 函式執行，會在呼叫 task_sleep。

void co_sleep(void *udata)
{
    assert(*(int *) udata == 99999999);
    task_sleep(1e8); /* 100ms CPU-based sleep */
}
static void task_sleep(int64_t nanosecs)
{
    int64_t start = getnow();
    while (getnow() - start < nanosecs)
        task_yield();
}

task_sleep 中的 task_yeild 功能主要是將當前 coroutine 放入 task_yielders 的串列當中，這可以表示成主動放棄 cpu 的 coroutine ，可以想成在 sleep 的狀態並等待排程，將自己放入 task_yielders 串列的末端後，在呼叫task_switch 呼叫在 task_nrunners 排隊的 coroutine 執行。

這裡會不斷的將 coroutine 加入 task_yielders 直到 100ms 時間到。

## test_task_pause
建立100個 `co_pause_one` coroutine 還有 一個 `co_resume_all` coroutine ，並輪流執行。

void test_task_pause(void)
{
    reset_stats();
    for (int i = 0; i < NCHILDREN; i++) {
        quick_start(co_pause_one, co_cleanup, &i);
    }
    quick_start(co_resume_all, co_cleanup, 0);
    while (task_active()) {
        task_resume(0);
    }
    assert(npaused == 0);
}

* co_pause_one
建立100個的 `co_pause_one` coroutine，每次都會做一次順序暫停一次逆序暫停，再做一次順序再做一次逆序。
順序暫停的實做如下

void co_pause_one(void *udata)
{
    int index = *(int *) udata;
    assert(index == npaused);
    paused[index] = task_id();
    /* pause in order */
    npaused++;
    is_paused[index] = true;
    task_pause();
    //return from pause
    is_paused[index] = false;
    npaused--;
    while (!all_resumed)
        task_yield();
}

將目前被暫停的 coroutine 標記為 true ，並做暫停切換到其他 coroutine (包含 test_task_pause)，從暫停回來之後將標記改回 false ，如果還有人在被暫停的話，那就將自己排到`task_yielders` ，等待所有人都恢復。
程式碼是如何做到逆序暫停？
答： 將 id 較小的 coroutine sleep 久一點 這樣 id 較大的 coroutine 就會比較先被暫停。

void co_pause_one(void *udata)
{
    /*...*/
     /* pause in reverse */
    task_sleep(1e6 * (NCHILDREN - index));
    npaused++;
    is_paused[index] = true;
    task_pause();
    is_paused[index] = false;
    npaused--;
    while (!all_resumed) {
        task_yield();
    }
}

* co_resume_all
有 100 個 `co_pause_one` coroutine 和 1 個 `co_resume_all` 依序交換執行之下，在 `co_pause_one` 裡面每一個逆/順序暫停中，最後都會檢查是否所有人都被恢復執行了才會繼續往下一個逆/順序暫停，而負責恢復執行的 coroutine 就是 `co_resume_all` ，在程式碼中可以看到，在暫停的時候的順序是 1. 順序暫停 2. 逆序暫停 3. 順序暫停 4. 逆序暫停，而恢復執行的順序卻是 1. 順序恢復 2. 順序恢復 3. 順序恢復 4. 逆序恢復 ，這種情況會使就算 coroutine 是早暫停的將會晚恢復執行。

void co_resume_all(void *udata)
{
    (void) udata;

    /* wait for all to be paused */
    while (npaused < NCHILDREN) {
        task_yield();
    }
    all_resumed = false;
    /* resume in order */
    for (int i = 0; i < NCHILDREN; i++) {
        task_resume(paused[i]);
    }
    while (npaused > 0) {
        task_yield();
    }
    all_resumed = true;
    /* resume in order */
    /* resume in reverse order */
    /* resume in reverse order */
    /*...*/
}

注意 `task_yield` 是將自己插入 `task_yielders` 末端，這樣的操作是可以再度被執行的。
但是如果使用 `task_pause` 的話是將自己插入到 `task_paused` 是一個 `task_map` 的結構末端， 也就是直到它被 resume 之前都不會被返回 `task_yielders`。

## test_task_exit
程式碼運作流程：
exitvals[] ： 用來紀錄執行順序。
1. 啟動 `co_one` coroutine
2. `co_one` 紀錄 1，並啟動 `co_two`
3. `co_two` 將自己放入 `task_yielders` ，task_sleep(1e7 * 2)，它會比 `co_three` 晚被恢復執行
4. 回到 `co_one` ，啟動 `co_three`
5. `co_three` 將自己放入 `task_yielders` ，task_sleep(1e7)
6. 回到 `co_one` ，啟動 `co_four`
7. `co_four` 紀錄 4 將自己放入 `task_yielders`
8. `co_one` 呼叫 `task_exit`，回到 `test_task_exit`
9. `test_task_exit` 確認是否還有 coroutine 沒有執行完，如果有就去執行它

    void test_task_exit(void)
    {
        /*...*/
        while (task_active()) {
            task_resume(0);
        }
        /*...*/
    }

10. 恢復執行 `co_three` ，紀錄 3
11. 恢復執行 `co_two` ，紀錄 2
12. test_task_exit 紀錄-2，表示所有任務完成。
至此 exitvals 內的值為 [1,4,-1,3,2,-2]
## test_task_order
測試 coroutine 執行順序。
`quickstart(co_yield)` -> `task_start` -> `coro_start` -> `coro_switch0` -> `coro_switch1` -> `task_entry`

static void task_entry(void *udata)
{
    /* Initialize a new coroutine on the user's stack. */
    /*...*/
    if (task_cur) {
        /* Reschedule the coroutine that started this one */
        task_list_push_back(&task_yielders, co);
        task_list_push_back(&task_yielders, task_cur);
        task_nyielders += 2;
        task_switch(false, false);
    }
    task_cur = co;
    if (task_user_entry) {
        task_user_entry(udata);
    }
    /* This coroutine is finished. Switch to the next coroutine. */
    task_switch(false, true);
}

1. 啟動 co_yield 是第一個 coroutine 所以沒有 task_cur，接著執行 co_yield
task_cur ＝ co_yield
task_nrunners = {} , task_nyielders = {}
a = [A]
quick_start(co_yield1) -> ... ->`task_entry`
2. task_cur 是 co_yield，所以將 co_yeld1 和 co_yeld0 插入 task_nyielders
task_nrunners = {} , task_nyielders = { co_yield1 -> co_yield }
task_cur = co_yield1
呼叫 `task_switch()`
3. `task_switch()` 現在沒有 runners ，所以將 task_nyielders 全部搬到 runners
task_cur = co_yield1
task_nrunners = { co_yield1 -> co_yield } , task_nyielders = {}
並取第一個 runners switch 過去執行，所以是 co_yield1。
4. 啟動 co_yeild1
a = [AB]
調用 `task_yield`，將自己插到 task_nyielders。
task_cur = co_yield1
task_nrunners = { co_yield } , task_nyielders = {co_yield1}
呼叫 `task_switch()`。
5. 從runners 取第一個 task = co_yield
task_nrunners = {} , task_nyielders = {co_yield1}
a = [ABC]
quick_start(co_yield2) -> ... -> `task_entry`
6. task_cur 是 co_yield，所以將 co_yeld2 和 co_yield 插入 task_nyielders
task_nrunners = {} , task_nyielders = { co_yield1 -> co_yield2 -> co_yield }
呼叫 `task_switch()`
7. 沒有 runners 所以將 task_nyielders 全部搬近來
task_nrunners = { co_yield1 -> co_yield2 -> co_yield } , task_nyielders = {}
task_cur = co_yield1
取第一個 runner 執行 = co_yield1
8. co_yield1
a = [ABCD]
執行結束回到 `task_entry` ，並呼叫 task_switch
task_cur = co_yield1
task_nrunners = { co_yield2 -> co_yield } , task_nyielders = {}
並取第一個 runners switch 過去執行，所以是 co_yield2。
task_cur = co_yield2
9. co_yield2
a = [ABCDE]
調用 `task_yield`，將自己插到 task_nyielders。
task_cur = co_yield2
task_nrunners = { co_yield } , task_nyielders = { co_yield2 }
並取第一個 runners switch 過去執行，所以是 co_yield。
task_cur = co_yield
10. co_yield
a = [ABCDEF]
調用 `task_yield`，將自己插到 task_nyielders。
task_cur = co_yield
task_nrunners = {} , task_nyielders = { co_yield2 -> co_yield }
沒有 runners 所以將 task_nyielders 全部搬近來
task_nrunners = { co_yield2 -> co_yield } , task_nyielders = {}
並取第一個 runners switch 過去執行，所以是 co_yield。
task_cur = co_yield2
11. co_yield2
a = [ABCDEFG]
執行結束回到 `task_entry` ，並呼叫 task_switch
task_nrunners = { co_yield } , task_nyielders = {}
並取第一個 runners switch 過去執行，所以是 co_yield。
12. co_yield
a = [ABCDEFGH]
執行結束回到 `task_entry` ，並呼叫 task_switch
task_nrunners = {} , task_nyielders = {}
結束回到 `test_task_order` 。

## test_task_detach
建立兩個 pthread 分別執行 `threda0` 和 `thread1`
執行後透過 `pthread_join` 等待 pthread 完成並回到 main。

`threda0` 程式碼分析：
啟動 100 個 coroutine ，並當被暫停的 coroutine 數量為 100 的時後(意思就是 task_paused 串列數量為 100)，就開始將 task_paused 的所有 coroutine 搬到 task_detached 串列上。

void *thread0(void *arg)
{
    (void) arg;
    reset_stats();
    for (int i = 0; i < NCHILDREN; i++) {
        quick_start(co_thread_one, co_cleanup, &i);
    }
    while (task_active()) {
        if (task_info_paused() == NCHILDREN) {
            /* Detach all paused */
            for (int i = 0; i < NCHILDREN; i++) {
                task_detach(thpaused[i]);
            }
        }
        task_resume(0);
    }
    return NULL;
}

`thread1` 程式碼分析：
如果 所有 coroutine 還沒都被 detached 的會就會卡在迴圈裡面。
離開迴圈後將所有 coroutine 從 task_detached 串列移回到 task_paused 串列，並把他們重新加回去 task_yielders 讓他們可以繼續被執行。

void *thread1(void *arg)
{
    (void) arg;
    reset_stats();
    while (task_info_detached() < NCHILDREN) {
        /* Wait */
    }

    /* Attach all paused */
    for (int i = 0; i < NCHILDREN; i++) {
        task_attach(thpaused[i]);
        task_resume(thpaused[i]);
    }
    while (task_active()) {
        task_resume(0);
    }
    return NULL;
}

`co_thread_one` 程式碼分析：
將自己加到 task_yielders 串列直到 時間到。
在將自己放到 task_paused 串列。

void co_thread_one(void *udata)
{
    int index = *(int *) udata;
    int64_t id = task_id();
    thpaused[index] = id;
    task_sleep(1e6);
    task_pause();
}

`task_detach` 程式碼分析：

給定 要 detach 的 coroutine id ，找到並加到 task_detached 結構。

void task_detach(int64_t id)
{
    struct task *co = task_map_delete(&task_paused, &(struct task){.id = id});
    if (co) {
        task_npaused--;
        task_lock();
        task_map_insert(&task_detached, co);
        task_ndetached++;
        task_unlock();
    }
}

`task_attach` 程式碼分析：
與 detach 操作差不多。

void task_attach(int64_t id)
{
    task_lock();
    struct task *co = task_map_delete(&task_detached, &(struct task){.id = id});
    if (co) {
        task_ndetached--;
    }
    task_unlock();
    if (co) {
        task_map_insert(&task_paused, co);
        task_npaused++;
    }
}

因為是使用 thread 的關係在操作共享變數的時候，必須要確保同步問題，適當的加入 lock ＆ unlock。

`task_lock` 程式碼分析：
透過 `atomic_compare_exchange_weak` 操作嘗試獲取鎖，如果可以成功更改 `task_locker` 變數則代表可以成功獲取鎖，反之則無法獲取，呼叫 `sched_yield0` 讓出 cpu，並循環嘗試獲取鎖。

static void task_lock(void)
{
    bool expected = false;
    while (!atomic_compare_exchange_weak(&task_locker, &expected, true)) {
        expected = false;
        sched_yield0();
    }
}

`task_unlock`程式碼分析：
使用 `atomic_store(&task_locker, false)` ，確保 對 task_locker 值操作具有原子性。