Rust 中的無鎖編程技術（三）

在前面兩篇博客中，我們從兩方面介紹了 “鎖”。第一個方面就是我們常說的“鎖” 其實是一個協議，這個協議從外部給出了鎖的行爲規範。另一個方面是 “鎖” 和數據的綁定。但是我們一直沒有提及究竟如何實現 “原始鎖”（請參考前文，“原始鎖” 即是上面提到的協議），這個就是本博客要解釋的。

Atomic 以及 Memory Ordering

在介紹如何實現原始鎖前，我們要對最基本的概念有一定的瞭解。對於原子類，我會做簡單的介紹，因爲原子類的實現牽扯到 CPU 的架構，牽扯到 ISA，我儘量從一個軟件工程師的角度來解釋這個基礎類。而 Memory Ordering 我將會在後面的博客花大篇幅來介紹。

Atomic

Atomic 類是多線程編程最基礎的一個類型。Rust 裏面提供了 AtomicBool，AtomicUsize 等基礎的 atomic 類型。Atomic 類和普通的類型基本一樣，即 Usize 和 AtomicUsize，AtomicBool 和 Bool 基本上沒有太多區別，只是一個可以在多線程環境下使用的，另一個更加適用於單線程，我認爲掌握 Atomic 類型的幾個特有的屬性就可以。

首先，無論在任何情況下，Atomic 類型讀寫一定是完整的。

舉個例子，比如 AtomicUsize，這個類型要看具體的 CPU 的架構，如果是 32 位的機器，那麼我們可以確保的是在內存中或者是在寄存器裏面，我們讀取的時候一定是 32 位一起讀取，賦值也是如此。一定不會出現只讀 AtomicUsize 高 16 位，再去讀別的數據的低 16 位這樣的情況。在多線程編程中，如果對 Atomic 類賦值，那麼結果一定要麼成功，要麼失敗。同理如果讀取 Atomic 的值，那麼結果一定是一個完整的值。

Atomic 類有自己的一套讀寫數據的方法，load， set，以及類似 compare_and_set 的函數。對一個原子類進行讀寫，這個類常用的函數有 load，set，而 compare_and_set 是我們在用原子類實現鎖或者無鎖數據結構常用到的方法。如其名，這個函數其中一個參數是舊的值，另一個參數是新的值，如果當前原子類裏面的值和所提供的舊值一致，那麼將會把這個原子類裏面的值換成新的值。

舉個例子，假設有兩個線程同時對一個 atomic 進行 compare_and_set 的操作，兩個線程在這個函數里面也提供了相同的舊值以及和舊值不相符的新值，那麼只有一個線程會成功，另一個會失敗。compare_and_set 返回值要看具體的接口定義，比如 crossbeam 所定義的 compare_and_set 和 Rust 所提供的原子類的 compare_and_swap 語義類似，但是返回值是不一樣的。

AtomicUsize 經常用作 Atomic<T>。首先正如上面提到 usize 是隨着 CPU 位數變化而變化的，usize 可以用來存地址，所以當我們有一個在 heap 上線程共享的數據，我們可以將其地址存到 AtomicUsize 裏面。所以 AtomicUsize 也可以理解爲原子指針, Rust 中可以這樣使用:

let data = unsafe { &(*(atomic_usize as *const _ as *const Atomic<T>)) };

總的來說，我們可以這樣理解原子類，就好像地鐵入口的旋轉欄杆一樣，對於原子類的任何操作，load，set，compare_and_set，都是獨立且完整的，任意一個時刻僅有一個操作在進行。如果還有不清楚的地方，後面的例子或許能更容易理解。

Memory Ordering

這裏我暫時不會提及特別多關於內存模型的知識點，因爲這個後面會花大篇幅來詳細聊聊到底什麼是內存模型。這裏我會用最簡單的例子來解釋一下這個概念。

假設現在有兩個線程，一個線程進行讀操作，另一個線程進行寫操作，而操作的數據是 AtomicBool 類型。這個數據的初始數據是 false，進行寫操作的線程想把這個值改成 true。現在寫操作已經執完了，那麼進行讀操作的線程是否一定讀到的是 true 呢？

答案是否定的，因爲寫操作剛執行完的時候，最新的數據還存在於 CPU 的寄存器當中，還沒有正式寫到內存中（當然這個得基於編譯器的實現以及 CPU 的具體優化策略），因此，非常有可能當寫線程執行完了，也就是其對應的彙編已經執行完了，但是最新的數據卻還僅僅存在於寄存器當中（針對的是物理多核的 CPU）。

那麼，我們怎麼保證當最新的數據只要出現在寄存器中，就立馬同步到內存中呢，這就是 Memory Ordering 的作用。我們現在只要記住，當我們用原子類的 set 方法，提供的內存順序是 Release，在 load 的時候，提供的內存順序是 Acquire 就可以保證進行讀的線程，讀到的一定是寄存器裏面最新的值。換句話說，我們可以暫時這樣理解，Release 是將寄存器裏面的值和內存的值同步，Acquire 是忘記自己當前寄存器的值，而直接去內存裏面讀取最新的值。如果對這一段解釋還是感到困惑，就先暫時記住：對於原子類的操作，如果是寫，那麼內存順序提供 Release，如果是讀，那麼內存順序提供 Acquire。

自旋鎖

我們以自旋鎖爲例子來看看到底是怎麼實現原始鎖的:

pub struct SpinLock {
    inner: AtomicBool,
}
impl Default for SpinLock {
    fn default() -> Self {
        Self {
            inner: AtomicBool::new(false),
        }
    }
}
impl RawLock for SpinLock {
    type Token = ();
    fn lock(&self) {
  ...
        while self.inner.compare_and_swap(false, true, Ordering::Acquire) {
  ...
        }
    }
    unsafe fn unlock(&self, _token: ()) {
        self.inner.store(false, Ordering::Release);
    }
}

自旋鎖其實就是個 AtomicBool，初始值爲 false，在 lock 的時候，我們用了原子類的特性，如果 compare_and_swap 失敗了，搶鎖的線程就會卡在 while 循環處。unlock 也很清晰，就是將 AtomicBool 的值設置成 false。我們來試想一下自旋鎖的工作機制。

一開始有很多線程同時在調用 lock 函數，僅有一個線程，成功地將 AtomicBool 設置成了 true，其餘的線程因爲在 compare_and_swap 提供的舊值都是 false，與當前在 AtomicBool 裏面的值不相符，因此其餘線程會卡在 while 處並且一直循環。當搶到了鎖的線程對共享資源的操作結束後，這個線程會調用 unlock 函數，在 unlock 函數里面，我們強行將 AtomicBool 設置成 false，並且我們強行將這個在寄存器裏面的新值，也就是 false，和內存同步 (Release)，其中一條卡在 while 處的線程將會從內存中讀取最新的值 (Acquire)，它將會是下一個可以對共享資源進行操作的線程。

其實自旋鎖就是內存中一塊線程共享的標識符，當其爲 false 的時候，表示當前共享資源沒有在被使用，當其爲 true 的時候，表示當前共享資源在被某個線程佔用着。

#[repr(C)]
pub struct Lock<L: RawLock, T> {
    lock: L,
    data: UnsafeCell<T>,
}
unsafe impl<L: RawLock, T: Send> Send for Lock<L, T> {}
unsafe impl<L: RawLock, T: Send> Sync for Lock<L, T> {}

最後，我們就可以將這個自旋鎖放到之前的代碼裏面，也就是上述代碼的 lock: L，我們就得到了一個完整的鎖的實現了。

我們能做得更好嗎

我們現在對於鎖有一個相對完整的認識了，自旋鎖只是一個例子，方便大家理解鎖的本質。但實際上自旋鎖是相當粗糙的。

我們仔細想想自旋鎖的工作原理就會發現，線程與線程之間是毫無公平性可言的。比如線程 A，B，C，D 是四個正在搶鎖的線程，這時候新加入了一個線程 E，對於自旋鎖是完完全全有可能讓線程 E 搶到了鎖，而 ABCD 均處在餓死的狀態。

細心的讀者應該已經發現了，自旋鎖的 token 也就是證明其實就是個 ()。自旋鎖的實現並沒有好好利用這個 token，我們是可以利用 token 來克服上面提到的自旋鎖的缺陷的，這個就是我下個博客會提到的關於其他鎖的實現方式。之後在介紹完鎖的相關知識後，我會帶大家重新認識 Memory Ordering。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/1sX87pm0Uwj2WdzTwqC2wA