---
tags: Linux Kernel
---

# 2022q1 Homework5 (quiz5)
contributed by < [steven1lung](https://github.com/steven1lung) >

## 測驗二

### Hazard pointer
Hazard pointer 是處理 lock-free 資料結構的記憶體管理問題的一種方式，通常這個問題會出現是因為執行環境沒提供[資源回收](https://en.wikipedia.org/wiki/Garbage_collection_(computer_science))的機制。

Lock-free 的資料結構都會使用 `CAS()` 來處理同步問題，例如：

Node* currentNode = this->head;  // assume the load from "this->head" is atomic
Node* nextNode = currentNode->next;  // assume this load is also atomic

假設這兩個指令都是 atomic，但是如果第 1 條指令執行完後被其他執行緒 preempt，可能會更改到 `this->head` 並且將 `this->head` 釋放掉。那這樣第 2 行指令再針對 `this->head` 操作就會操作到已經被釋放的資料。

這時候 hazard pointers 就可以解決這個問題，每個執行緒都會有一個 list 的 hazard pointers，這些 hazard pointers 會儲存執行緒正在存取的節點，被 hazard pointer 指到的節點就不能被其他執行緒更動或是釋放。

當一個執行緒想要移除一個節點，這時後節點不會被馬上釋放掉，因為有可能會被其他執行緒的 hazard pointer 指到，所以會將這個節點放在一個存「等等再釋放」的 list 裡（retire list），等到沒有 hazard pointer 指到此節點時，再進行釋放。

#### CAS
CAS 全名是 compare and swap，這個函式會需要 3 個參數
1. 指標
2. 舊的值
3. 新的值

所做的操作就是會先將傳入的指標取值，去跟舊的值比較：
- 如果兩個值相同，那就會將指標裡的值改成新的值。
- 如果不同，那就不會做值的更動，代表裡面的值被改動過了。

可以透過這樣的方式來避免在執行過程中順序不同或是被打斷的問題，可以看到程式碼最上方有定義 `atomic_cas`：

#define atomic_cas(dst, expected, desired)                                 \
    __atomic_compare_exchange(dst, expected, desired, 0, __ATOMIC_SEQ_CST, \
                              __ATOMIC_SEQ_CST)

而 `__atomic_compare_exchange()` 是 C11 的 built-in，在文件中得知每個 `__atomic` 開頭的 built-in 都不會有 data races，因為操作都是最小單位，不會被打斷。

bool __atomic_compare_exchange_n (type *ptr, type *expected, type desired, bool weak, int success_memorder, int failure_memorder)

函式會將 `*ptr` 與 `*expected` 比對，如果相同會將 `desired` 寫入 `*ptr`。如果不同，此次的操作就會變成將 `*ptr` 的值寫入 `*expected`，並不會更動指標的值。
>This built-in function implements an atomic compare and exchange operation. This compares the contents of *ptr with the contents of *expected. If equal, the operation is a read-modify-write operation that writes desired into *ptr. If they are not equal, the operation is a read and the current contents of *ptr are written into *expected. weak is true for weak compare_exchange, which may fail spuriously, and false for the strong variation, which never fails spuriously. Many targets only offer the strong variation and ignore the parameter. When in doubt, use the strong variation.
>
>If desired is written into *ptr then true is returned and memory is affected according to the memory order specified by success_memorder. There are no restrictions on what memory order can be used here.
>
>Otherwise, false is returned and memory is affected according to failure_memorder. This memory order cannot be __ATOMIC_RELEASE nor __ATOMIC_ACQ_REL. It also cannot be a stronger order than that specified by success_memorder.

注意到 `weak` 如果被設成 `true`，那操作會變成 weak compare_exchange，代表在 `*ptr` 跟 `*expected` 相同時還是會有機會回傳 `false`。這個錯誤被稱為 spurious failure，會出現這個錯誤是因為有些架構 (ARM) 會使用 `LL` (Load Linked) / `SC` (Store Conditional) 指令來實作 `CAS()`。

`LL` 指令會讀取記憶體中的位址，並且將這個位址記起來保存在別處，而 `SC` 指令會在剛剛保存的位址中的值沒有被更動的情況下寫入給定的記憶體位址。但是保存的記憶體位址很有可能在沒更動下被硬體認為被更改過，有以下可能的原因：
1. 位址被寫入相同的值
2. 保存位址附近被寫入，但因為是對 cache line 操作，會讓硬體將 cache line 裡附近的位址都設定成 `dirty`
3. 其他可能會讓 CPU 遺失現在的狀態，例如：context switch，page table switch 等等

所以在不確定要使用 weak 或 strong 時，文件是建議使用 strong variation。

但可能會納悶既然 weak 會出現無法預測的問題，那為什麼還要區分 weak、strong 呢。原因是 spurious failure 發生的機率不大，但是相比可以得到更好的效能提升。因為 strong 每次都要檢查有沒有 spurious failure。 所以會讓 weak 跟 strong 有使用情境上的選擇與 trade off。

### 解釋程式碼

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

這個結構體就是 hazard pointer 的實作，有一個指標會指向正在被使用的節點，一個指標指向下一個 hazrad pointer。

/* 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;
}

這個函式會要求一個新的 `hp_t`，將這個 hazard pointer 的節點指標指向傳入的指標，並將此 hazard pointer 放到佇列裡。

/* 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;
}

這個函式會在 `head` 裡尋找空的節點，並且放入我們要的資料，不然就加入 `head` 中。在迭代每個節點的時候，如果讀取到的節點為 `0`，就會將傳入的指標放入這個節地中。跑過每個節點都沒找到空的節點時，就會自己將傳入的指標加入 `head` 中。

/* 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;
}

這邊的操作跟上面很像，只不過是在找一個特定的指標，而不是找空的指標。

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;
}

函式會迭代過每一個節點，如果其中節點裡的指標等於傳入的指標，就會回傳 `true`。

/* Forces the cleanup of old objects that have not been deallocated yet. Just
 * like `swap`, if `flags` is 0, this function will wait until there are no
 * more references to each object. If `flags` is `DEFER_DEALLOC`, only
 * objects that already have no living references will be deallocated.
 */
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 (EXP3)
                dom->deallocator((void *) ptr);
        } else if (!(flags & DEFER_DEALLOC)) {
            /* Spin until all readers are done, then deallocate */
            while (list_contains(&dom->pointers, ptr))
                ;
            if (EXP4)
                dom->deallocator((void *) ptr);
        }
    }
}

從註解可以得知我們需要考慮兩種情況，`flag` 傳入 `0` 或是 `DEFER_DEALLOC`。`0` 會需要等待直到沒有人在使用節點裡的指標，而 `DEFER_DEALLOC` 可以直接釋放掉沒有在使用的指標。

## Chrome 瀏覽器 70% 安全性錯誤來自記憶體

Chrome 工程師在這篇[文章](https://www.chromium.org/Home/chromium-security/memory-safety/)中指出大部分的安全性問題都是出在記憶體問題，對指標的錯誤使用，並且有接近一半是 use-after-free 的問題。這個問題會出現的原因是對已經釋放的記憶體進行操作，對已經釋放或是已分配給其他應用程式的記憶體空間進行操作會造成資安問題。

Google 有提到一個方法，任何工程師都需要遵守這個規則 [The Rule of 2](https://chromium.googlesource.com/chromium/src/+/master/docs/security/rule-of-2.md)。根據這個規則，當一個新的 feature 要被開發時，工程師的程式碼不能打破超過 2 個原則：
1. The code handles untrustworthy inputs
2. The code runs with no sandbox
3. The code is written in an unsafe programming language (C/C++)

會說 C/C++ 是不安全的 (unsafe) 語言是因為當初這些語言被開發時，安全並非首要考量，而且當時的資訊安全議題相對單純。所以 C/C++ 讓開發者對程式的指標有全部的權限，並且在開發者做出會對記憶體危害的操作時，沒有發出警告或是防止這些操作。

這時候 Mozilla 在 Firefox 瀏覽器的開發過程中，提出全新的 Rust 程式語言，被視為最安全的程式語言之一，並且也被視為取代 C/C++ 良好的方案。

Sandboxing 是將每個行程隔開的機制，Chromium 使用的是多處理器架構的模型，可以將不同的權限跟限制給予瀏覽器的各個部分。例如我們想讓 render (算繪模組) 在有限的資源底下運作，因為 render 通常都會與無法信任的輸入溝通。當 render 需要更高的權限時，會透過 [IPC](https://en.wikipedia.org/wiki/Inter-process_communication) 向更高層級的瀏覽器行程要求權限。
:::warning
參見 [論 Render 翻譯](http://breezymove.blogspot.com/2013/12/render.html)
:notes: jserv
:::

但是 sandbox 有一個很關鍵的問題，行程是隔離的最小單位，但是隔離一個行程成本不低。像是在 Android 中，使用越多行程會影響裝置整體的健康，因為當我們建立越多行程，其他背景或是別的應用程式的行程會在更高的頻率下被 killed。

所以我們常說嘴的「chrome 都把記憶體佔走」 的原因就是因為我們想透過 sandbox 的機制來達到隔絕這些 bug 或是不安全的問題，但是要透過 sandbox 隔絕就必須要將每個網頁都建立一個 process 去分開他們，就造成了 chrome 行程數量暴增的原因，進而導致記憶體被 chrome 佔光。

而其中一個 Google 正在嘗試的方法是在可行的情況下去使用 Rust。目前 chrome 絕大多數的程式是 C/C++，所以如果要使用 Rust，就必須要考慮到相容性問題。Chrome 工程師認為 Rust 可以呼叫 C++ 的程式碼是一個很重要的問題。如果只能讓 C++ 呼叫 Rust，那 Rust 就會變成 leaf nodes，無法讓 Rust 跟整個程式庫互動，那這樣是否該花費成本去增加一個末端的語言就該被重新討論。

工程師認為 Rust 要能呼叫 C++ 函式需要達到以下的條件：

* No need for the "unsafe" keyword unless something is known to be less safe than normal C++
> 對 Rust 來說，所有的 C++ 程式碼都是 "unsafe"。但如果每個 C++ 程式碼都被稱為不安全，那不安全也就沒了意義，所以要將這個關鍵字限制在不安全的 Rust 程式碼或是 C++ 相容性的程式碼裡有多個 ownership 或是其他複雜的情況。這個條件在 [cxx 函式庫](https://github.com/dtolnay/cxx)就被滿足了。
* No overhead in the general case
> Overhead 可以想成是一個執行一個操作所需要的成本，例如我們可以開車從台南車站到成大，但這樣的 overhead 不小，所以我們可能會選擇走路。但是如果我們想從台南開車去台北的話，這個 overhead 我們就可以接受。
>
> LTO (link-time-optimization) 跟 [cross-language inlining](http://blog.llvm.org/2019/09/closing-gap-cross-language-lto-between.html) 已經解決這個問題。例如從 C++ 傳字串到 Rust 會需要進行 UTF Check，這就是一個 overhead。但我們可以在 Rust 裡透過 `&[u8]` 來直接處理這些字串。
* No boilerplate or redeclarations. No C++ annotations. Ideally, no allowlist
> 如果存在 C++ 的 API，那 Rust 就必須有資格去呼叫這個 API。不應該有白名單，或是需要在 Rust 中被重複宣告。Chrome 的程式庫已經非常複雜，使用更多的 annotation 污染程式庫是個小但會被注意到的技術債。
* Broad type support - with safety
> 現在的 [cxx](https://github.com/dtolnay/cxx) 已經可以支援 C++/Rust 安全地資料交換，但是特別的資料型態像是 `struct` 中又包含 `std::strings` 、需要其他指標參數的函式 (out parameters)，都會讓資料傳遞變得更複雜。所以找出一個用程式碼找出這些 out parameters 來讓 Rust 可以合法地使用它們是現在的問題。

### 案例探討: P2PSocketDispatcherHost UaF
這個問題是一個 use-after-free 的問題，沒在指派數值到 `sockets_[connected_socket_id]` 前確認 `connected_socket_id` 有沒有存在。這樣會導致如果 `connected_socket_id` 存在，那會讓原本的指標被移除，讓原本指到的 socket 在 render 死掉後繼續活著，導致 UAF。

會發生問題的程式碼在 content/browser/renderer_host/p2p/socket_dispatcher_host.cc：

void P2PSocketDispatcherHost::OnAcceptIncomingTcpConnection(
    int listen_socket_id, const net::IPEndPoint& remote_address,
    int connected_socket_id) {
  P2PSocketHost* socket = LookupSocket(listen_socket_id);
  if (!socket) {
    LOG(ERROR) << "Received P2PHostMsg_AcceptIncomingTcpConnection "
    return;
  }
  P2PSocketHost* accepted_connection =
      socket->AcceptIncomingTcpConnection(remote_address, connected_socket_id);
  if (accepted_connection) {
    sockets_[connected_socket_id] = accepted_connection;
  }
}

問題出在第 12 行，如果沒有先檢查`connected_socket_id` 就直接覆蓋過去，會遺失對原本連線 socket 的控制，導致在 render 關掉後連線還存活。

在 issue 被提及的 9 個小時後，chrome 工程師就將問題修復了，會在收到連線之前確認 `connected_socket_id` 有沒有等於 `NULL`。

void P2PSocketDispatcherHost::OnAcceptIncomingTcpConnection(
    int listen_socket_id, const net::IPEndPoint& remote_address,
    int connected_socket_id) {
  P2PSocketHost* socket = LookupSocket(listen_socket_id);
  if (!socket) {
    LOG(ERROR) << "Received P2PHostMsg_AcceptIncomingTcpConnection "
        "for invalid listen_socket_id.";
    return;
  }
  if (LookupSocket(connected_socket_id) != NULL) {
    LOG(ERROR) << "Received P2PHostMsg_AcceptIncomingTcpConnection "
        "for duplicated connected_socket_id.";
    return;
  }
  P2PSocketHost* accepted_connection =
      socket->AcceptIncomingTcpConnection(remote_address, connected_socket_id);
  if (accepted_connection) {
    sockets_[connected_socket_id] = accepted_connection;
  }
}

> 修改的 [commit](https://chromium.googlesource.com/chromium/src.git/+/3ad92fa676ff9515194c9bc9ecafff7942129f5a%5E%21/#F0) 。

### 案例探討: Kernel memory corruption that could be used as a sandbox escape from a compromised renderer

這個問題是一個 Linux 核心 `futex()` 的記憶體問題，可以透過 render 來逃出 sandbox。(見 [Issue 377392](https://bugs.chromium.org/p/chromium/issues/detail?id=377392))

先瞭解一下 `futex()`，man page 說明 `futex` 是快速的 user-space lock。其中問題使用到的指令是 `FUTEX_WAIT_REQUEUE_PI()`。

int futex_wait_requeue_pi (	u32 __user * uaddr,
 	unsigned int flags,
 	u32 val,
 	ktime_t * abs_time,
 	u32 bitset,
 	u32 __user * uaddr2);

呼叫者會在 `uaddr` 中等待，並且會被重新排回 queue 中並且是使用 `uaddr2`。這個指令會需要跟 `FUTEX_CMP_REQUEUE_PI` 搭配，讓被 `FUTEX_WAIT_REQUEUE_PI` 所 block 的人使用 `uaddr2` 排回去 queue 中。

詳細程式碼可見 [fubar.c](https://gist.github.com/steven1lung/8d969f75b35ef8bf7b7ea3374d4de49c)，這個問題產生的原因是因為被 `FUTEX_WAIT_REQUEUE_PI` 擋住後，`q.rt_waiter` 是 `NULL` 但是在 stack 中的 `&rt_waiter` 已經被加回等待的 queue 了。這不應該發生，因為 `rt_waiter` 被設為 `NULL` 是 atomic。

後來工程師修復好 (見 [commit 6f7b0a](https://chromium.googlesource.com/chromiumos/third_party/kernel/+/6f7b0a2a5c0fb03be7c25bd1745baa50582348ef%5E%21/#F0)) 的方法是在 `FUTEX_WAIT_REQUEUE_PI` 裡先檢查 `uaddr == uaddr2`。

> If uaddr == uaddr2, then we have broken the rule of only requeueing
from a non-pi futex to a pi futex with this call. If we attempt this,
as the trinity test suite manages to do, we miss early wakeups as
q.key is equal to key2 (because they are the same uaddr). We will then
attempt to dereference the pi_mutex (which would exist had the futex_q
been properly requeued to a pi futex) and trigger a NULL pointer
dereference.

## Rust Memory Safety (Ownership)
Ownership (所有權) 是 Rust 最特別的地方，而且也涉及到整個語言的架構。Ownership 允許 Rust 在沒有垃圾處理的機制下達到 memory safety，Ownership 規定 Rust 程式如何操作記憶體。

在一些語言中，有提供垃圾處理的機制或是使用者需要自己要求跟釋放記憶體空間，但是 Rust 使用第三種方法：ownership 會被一些規則規定如何管理記憶體空間。並且如果有規定被違反，那程式就不會成功編譯。
>Rust uses a third approach: memory is managed through a system of ownership with a set of rules that the compiler checks. If any of the rules are violated, the program won’t compile. None of the features of ownership will slow down your program while it’s running.

### Stack / Heap
許多程式語言並沒有明確要求開發者要理解 stack 跟 heap，但是在 Rust 中，一個值被儲存在 stack 或是 heap 中會影響程式的行為，也會影響你所要做的決定。Stack 跟 heap 都是在程式運行的時候可以使用的記憶體空間，但是這兩者的架構不一樣。

Stack 在存取資料時的順序是 LIFO，而且存放在 stack 的資料都必須是已知且固定的大小。如果資料在編譯時期大小是未知的，或是資料的大小可能會改變，那我們就應該把這筆資料存放在 heap。

Heap 比較沒有那麼整齊：當我們放一筆資料進 heap，意思就是我們向 heap 要求一個特定大小的空間。系統就會從 heap 分配一個剛好可以放入資料大小的空間，將這個空間標記成使用中，並回傳一個指標。

將資料 push 進 stack 中會比從 heap 要求空間快，因為系統不需要去尋找一段空間來存資料，因為要找的空間就在 stack 最上端。

### Ownership Rules
回到剛剛提到的 ownership 規則：
* Each value in Rust has a variable that’s called its owner.
* There can only be one owner at a time.
* When the owner goes out of scope, the value will be dropped.