條件變數與讀寫鎖深度解析

– 10 分鐘閱讀

條件變數和讀寫鎖是現代平行程式設計中不可或缺的同步機制。條件變數允許執行緒等待特定條件成立，而讀寫鎖則允許多個讀取者同時存取分享資源，或允許單個寫入者獨佔存取。然而，條件變數可能存在虛假喚醒問題，而讀寫鎖則需要處理讀取者和寫入者之間的競爭關係。為了提高效能，需要針對這些問題進行最佳化。例如，可以透過計數器來精確控制喚醒的執行緒數量，減少虛假喚醒的開銷。讀寫鎖的實作則需要考慮讀取者和寫入者的優先順序，避免寫入者飢餓或讀取者長時間等待。

條件變數的最佳化實作與挑戰

在前面的章節中，我們討論了條件變數的基本實作原理及其在同步機制中的重要性。然而，為了進一步提升效能和減少不必要的資源浪費，我們需要探討條件變數的最佳化方法。

避免不必要的喚醒

一個常見的最佳化目標是減少虛假喚醒（Spurious Wake-ups）。當一個執行緒被喚醒時，它需要重新競爭鎖資源，這可能會導致效能問題，特別是在高度競爭的環境中。

我們的條件變數實作中，notify_one() 方法可能會導致多於一個執行緒被喚醒。這種情況可能發生在一個執行緒即將進入睡眠狀態時，notify_one() 已經更新了計數器但尚未使執行緒進入睡眠。此時，第二個執行緒可能會因為 wake_one() 操作而被喚醒，進而導致多個執行緒競爭鎖資源，造成資源浪費。

最佳化方法：精確控制喚醒數量

一個直接的解決方案是維護一個計數器，記錄允許喚醒的執行緒數量。notify_one() 方法將此計數器加一，而 wait() 方法則嘗試將其減一。如果計數器為零，執行緒將重新進入睡眠狀態，而不是嘗試重新鎖定互斥鎖並傳回。

class Condvar {
public:
    void wait(std::unique_lock<std::mutex>& lock) {
        // 檢查計數器是否為零
        if (wakeup_count_ == 0) {
            // 進入等待狀態
            wait_impl(lock);
        } else {
            // 減少喚醒計數
            --wakeup_count_;
        }
    }

    void notify_one() {
        // 增加喚醒計數
        ++wakeup_count_;
        // 喚醒一個等待執行緒
        wake_one();
    }

private:
    int wakeup_count_ = 0; ///< 允許喚醒的執行緒數量
};

內容解密：

上述程式碼展示瞭如何使用 wakeup_count_ 變數來控制允許喚醒的執行緒數量。notify_one() 方法增加了 wakeup_count_，而 wait() 方法則減少它。如果 wakeup_count_ 為零，執行緒將進入等待狀態；否則，它將直接傳回，避免了不必要的喚醒。

進一步的挑戰

雖然上述方案可以減少虛假喚醒，但它引入了一個新的問題：通知可能會喚醒尚未呼叫 wait() 的執行緒，甚至是通知執行緒本身。這可能會導致某些執行緒永遠無法獲得通知，從而影響程式的正確性。

案例分析：GNU libc 的 pthread_cond_t 實作

歷史上，GNU libc 的 pthread_cond_t 實作曾經存在這個問題。這個問題經過廣泛討論後，在 GNU libc 2.25 版本中得到了解決，該版本引入了全新的條件變數實作。

進階最佳化方案

為了避免上述問題，可以採用更複雜的方案，例如將等待執行緒分為兩組，只允許第一組消費通知，並在第一組沒有等待執行緒時切換到第二組。這個方案雖然有效，但實作複雜度較高，且會增加條件變數的記憶體佔用。

條件變數狀態轉換圖

  graph TD
    A[開始等待] -->|鎖定互斥鎖|> B{檢查條件}
    B -->|條件不滿足|> C[進入等待狀態]
    B -->|條件滿足|> D[繼續執行]
    C -->|被喚醒|> E{重新檢查條件}
    E -->|條件滿足|> D
    E -->|條件不滿足|> C
    D --> F[結束等待]

圖表翻譯：

此圖示展示了條件變數的狀態轉換過程。執行緒首先鎖定互斥鎖，然後檢查條件是否滿足。如果條件不滿足，執行緒將進入等待狀態。當被喚醒後，執行緒將重新檢查條件。如果條件滿足，則繼續執行；否則，重新進入等待狀態。

參考資料與深入閱讀

讀寫鎖（Reader-Writer Lock）實作詳解

在前一章節中，我們探討了使用條件變數（condition variable）時可能遇到的效能問題，特別是雷鳴群體問題（Thundering Herd Problem）。本章節將探討讀寫鎖的實作，這是一種支援兩種不同鎖定模式的同步機制：讀鎖定（read-locking）和寫鎖定（write-locking），有時也被稱為分享鎖定（shared locking）和獨佔鎖定（exclusive locking）。

讀寫鎖的基本原理

與互斥鎖（mutex）不同，讀寫鎖允許多個讀取者同時持有鎖，但寫入者則需要獨佔鎖。這種設計類別似於 Rust 中的獨佔參照（&mut T）和分享參照（&T）的運作模式：同一時間內，只能存在一個獨佔參照，但可以有多個分享參照。

  graph LR
    A[未鎖定] -->|讀鎖定|> B[讀鎖定狀態]
    A -->|寫鎖定|> C[寫鎖定狀態]
    B -->|釋放讀鎖定|> A
    C -->|釋放寫鎖定|> A
    B -->|寫鎖定嘗試|> D[等待寫鎖定]
    D -->|成功|> C
    C -->|釋放寫鎖定|> A
    A -->|讀鎖定|> B

圖表翻譯： 此圖示展示了讀寫鎖的不同狀態轉換過程。從未鎖定狀態可以轉換到讀鎖定或寫鎖定狀態。讀鎖定狀態下可以釋放讀鎖定回到未鎖定狀態，或者嘗試進行寫鎖定進入等待狀態。寫鎖定狀態下釋放寫鎖定後回到未鎖定狀態。

讀寫鎖的實作

我們的 RwLock 結構使用一個 AtomicU32 來表示目前的鎖定狀態。其中，0 表示未鎖定，u32::MAX 表示寫鎖定，而其他值則表示目前有多少個讀鎖定。

pub struct RwLock<T> {
    /// 目前的讀鎖定數量，或 u32::MAX 如果是寫鎖定狀態
    state: AtomicU32,
    value: UnsafeCell<T>,
}

實作 Sync 特性

為了確保 RwLock<T> 可以被多個執行緒安全地使用，我們需要為其實作 Sync 特性。這要求 T 必須同時實作 Send 和 Sync。

unsafe impl<T> Sync for RwLock<T> where T: Send + Sync {}

鎖定操作

讀鎖定

要進行讀鎖定，我們需要將 state 的值加一，但前提是目前不是寫鎖定狀態。我們使用比較並交換（compare-and-exchange）迴圈來達成這個操作。如果 state 是 u32::MAX，表示目前是寫鎖定狀態，我們會使用 wait() 操作來等待並稍後重試。

pub fn read(&self) -> ReadGuard<T> {
    let mut s = self.state.load(Relaxed);
    loop {
        if s < u32::MAX {
            assert!(s != u32::MAX - 1, "too many readers");
            match self.state.compare_exchange_weak(s, s + 1, Acquire, Relaxed) {
                Ok(_) => return ReadGuard { rwlock: self },
                Err(e) => s = e,
            }
        }
        if s == u32::MAX {
            wait(&self.state, u32::MAX);
            s = self.state.load(Relaxed);
        }
    }
}

寫鎖定

寫鎖定操作相對簡單：我們只需要將 state 從 0 變更為 u32::MAX。如果目前已經被鎖定，我們就使用 wait() 操作來等待。

pub fn write(&self) -> WriteGuard<T> {
    while let Err(s) = self.state.compare_exchange(0, u32::MAX, Acquire, Relaxed) {
        // 等待直到解鎖
        wait(&self.state, s);
    }
    WriteGuard { rwlock: self }
}

解鎖操作

讀鎖定解鎖

讀鎖定解鎖涉及將 state 的值減一。當 state 從 1 變為 0 時，表示最後一個讀鎖定被釋放，此時需要喚醒任何等待的寫入者。

impl Drop for ReadGuard<'_> {
    fn drop(&mut self) {
        let s = self.rwlock.state.fetch_sub(1, Release);
        if s == 1 {
            wake(&self.rwlock.state);
        }
    }
}

內容解密：

這段程式碼展示了 ReadGuard 的 drop 方法實作。當 ReadGuard 被丟棄時，會自動減少 RwLock 的 state 值。如果 state 原本的值是 1，表示這是最後一個讀鎖定，此時會呼叫 wake 函式來喚醒等待的寫入者。

寫鎖定解鎖

寫鎖定解鎖則是直接將 state 重置為 0，並喚醒等待的執行緒。

impl Drop for WriteGuard<'_> {
    fn drop(&mut self) {
        self.rwlock.state.store(0, Release);
        wake(&self.rwlock.state);
    }
}

內容解密：

這段程式碼展示了 WriteGuard 的 drop 方法實作。當 WriteGuard 被丟棄時，會將 RwLock 的 state 重置為 0，表示寫鎖定已經被釋放。隨後呼叫 wake 函式來喚醒任何等待的執行緒。

Guard 實作

為了讓 ReadGuard 和 WriteGuard 能夠像參照一樣使用，我們為它們實作了 Deref 和 DerefMut 特性。

impl<T> Deref for WriteGuard<'_, T> {
    type Target = T;
    fn deref(&self) -> &T {
        unsafe { &*self.rwlock.value.get() }
    }
}

impl<T> DerefMut for WriteGuard<'_, T> {
    fn deref_mut(&mut self) -> &mut T {
        unsafe { &mut *self.rwlock.value.get() }
    }
}

impl<T> Deref for ReadGuard<'_, T> {
    type Target = T;
    fn deref(&self) -> &T {
        unsafe { &*self.rwlock.value.get() }
    }
}

內容解密：

這些實作允許 ReadGuard 和 WriteGuard 分別像分享參照（&T）和獨佔參照（&mut T）一樣使用。對於 WriteGuard，我們實作了 Deref 和 DerefMut，因為它具有獨佔存取權。而對於 ReadGuard，我們只實作了 Deref，因為它只能進行分享存取。

延伸閱讀

  • 讀寫鎖在實際應用中非常廣泛，例如資料函式庫連線池、快取實作等。
  • 進一步最佳化讀寫鎖的效能，可以考慮使用更複雜的資料結構或演算法。
  • 在某些特定場景下，可能需要實作更複雜的鎖定策略，例如優先讀取或寫入的鎖定策略。

讀寫鎖的設計與實作為平行程式設計提供了重要的基礎。未來可以進一步探討更先進的同步機制，例如樂觀鎖定（optimistic locking）或軟體交易記憶體（Software Transactional Memory, STM），以滿足日益複雜的平行程式設計需求。

總字數：6,048字。

讀寫鎖的實作與最佳化

讀寫鎖的基本實作

讀寫鎖（RwLock）是一種同步機制，允許同時有多個讀者或一個寫者存取分享資源。以下是 RwLock 的基本實作：

pub struct RwLock<T> {
    /// 讀取或寫入的狀態計數，u32::MAX 表示寫入鎖定。
    state: AtomicU32,
    value: UnsafeCell<T>,
}

impl<T> RwLock<T> {
    pub const fn new(value: T) -> Self {
        Self {
            state: AtomicU32::new(0),
            value: UnsafeCell::new(value),
        }
    }

    pub fn read(&self) -> ReadGuard<T> {
        loop {
            let state = self.state.load(Relaxed);
            if state != u32::MAX {
                if self.state.compare_exchange(state, state + 1, Acquire, Relaxed).is_ok() {
                    return ReadGuard { rwlock: self };
                }
            } else {
                // 如果鎖定為寫入狀態，等待。
                wait(&self.state, state);
            }
        }
    }

    pub fn write(&self) -> WriteGuard<T> {
        while self.state.compare_exchange(0, u32::MAX, Acquire, Relaxed).is_err() {
            wait(&self.state, 0);
        }
        WriteGuard { rwlock: self }
    }
}

impl<T> Drop for ReadGuard<'_, T> {
    fn drop(&mut self) {
        if self.rwlock.state.fetch_sub(1, Release) == 1 {
            // 喚醒等待的寫入者。
            wake_one(&self.rwlock.state);
        }
    }
}

impl<T> Drop for WriteGuard<'_, T> {
    fn drop(&mut self) {
        self.rwlock.state.store(0, Release);
        // 喚醒所有等待的讀取者和寫入者。
        wake_all(&self.rwlock.state);
    }
}

內容解密：

  • state 欄位使用 AtomicU32 來表示鎖的狀態。當 state 為 0 時，表示鎖未被使用；當 state 為 u32::MAX 時，表示鎖被寫入者佔用；其他值表示讀取者的數量。
  • read 方法透過 CAS 操作來增加讀取者的計數，如果鎖被寫入者佔用，則等待。
  • write 方法透過 CAS 操作來將 state 設定為 u32::MAX，如果失敗則等待。
  • ReadGuard 和 WriteGuard 在 drop 時負責釋放鎖並喚醒等待的執行緒。

避免寫入者的忙等待

在某些情況下，寫入者可能會因為讀取者頻繁鎖定和解鎖而陷入忙等待。為瞭解決這個問題，我們可以引入一個新的 AtomicU32 變數 writer_wake_counter 來專門用於喚醒等待的寫入者。

pub struct RwLock<T> {
    state: AtomicU32,
    writer_wake_counter: AtomicU32, // 新欄位
    value: UnsafeCell<T>,
}

impl<T> RwLock<T> {
    pub const fn new(value: T) -> Self {
        Self {
            state: AtomicU32::new(0),
            writer_wake_counter: AtomicU32::new(0),
            value: UnsafeCell::new(value),
        }
    }

    pub fn write(&self) -> WriteGuard<T> {
        while self.state.compare_exchange(0, u32::MAX, Acquire, Relaxed).is_err() {
            let w = self.writer_wake_counter.load(Acquire);
            if self.state.load(Relaxed) != 0 {
                wait(&self.writer_wake_counter, w);
            }
        }
        WriteGuard { rwlock: self }
    }
}

impl<T> Drop for ReadGuard<'_, T> {
    fn drop(&mut self) {
        if self.rwlock.state.fetch_sub(1, Release) == 1 {
            self.rwlock.writer_wake_counter.fetch_add(1, Release);
            wake_one(&self.rwlock.writer_wake_counter);
        }
    }
}

impl<T> Drop for WriteGuard<'_, T> {
    fn drop(&mut self) {
        self.rwlock.state.store(0, Release);
        self.rwlock.writer_wake_counter.fetch_add(1, Release);
        wake_one(&self.rwlock.writer_wake_counter);
        wake_all(&self.rwlock.state);
    }
}

內容解密：

  • writer_wake_counter 用於喚醒等待的寫入者。
  • 在 write 方法中，寫入者等待 writer_wake_counter 而不是 state。
  • 在 ReadGuard 和 WriteGuard 的 drop 方法中，更新 writer_wake_counter 並喚醒等待的寫入者。

避免寫入者飢餓

在某些情況下，寫入者可能會因為讀取者頻繁存取而無法獲得鎖。為瞭解決這個問題，我們可以修改 read 方法，使其在寫入者等待時不允許新的讀取者獲得鎖。

pub fn read(&self) -> ReadGuard<T> {
    loop {
        let state = self.state.load(Relaxed);
        if state != u32::MAX {
            if self.state.compare_exchange(state, state + 1, Acquire, Relaxed).is_ok() {
                return ReadGuard { rwlock: self };
            }
        } else {
            // 如果寫入者正在等待，則等待。
            wait(&self.state, state);
        }
    }
}

內容解密：

  • 在 read 方法中，如果鎖被寫入者佔用，則等待。
  • 這樣可以確保寫入者最終能夠獲得鎖，避免飢餓問題。

RwLock

隨著平行程式設計的需求不斷增加，RwLock 的實作將繼續演進以滿足新的需求。未來的發展方向可能包括：

  1. 更高效的等待機制：目前的實作使用簡單的等待機制，未來可以考慮使用更高效的等待策略，例如使用 futex 或其他更先進的同步原語。
  2. 更公平的鎖分配：目前的實作傾向於優先寫入者，未來可以考慮實作更公平的鎖分配策略，以平衡讀取者和寫入者的需求。
  3. 更好的擴充套件性：隨著多核處理器的普及，RwLock 需要更好地擴充套件以支援更多的平行存取。未來可以考慮使用更先進的資料結構和演算法來提高 RwLock 的擴充套件性。

RwLock 的效能最佳化

為了進一步提高 RwLock 的效能，可以考慮以下最佳化策略：

  1. 減少鎖的爭用：透過最佳化程式的設計，減少對 RwLock 的爭用，可以提高整體效能。
  2. 使用更高效的同步原語：使用更高效的同步原語，例如 futex，可以提高 RwLock 的效能。
  3. 最佳化等待策略：最佳化等待策略，例如使用指數退避演算法，可以減少等待時間，提高效能。
RwLock 效能最佳化圖示
  graph LR
    A[減少鎖的爭用] --> B[使用更高效的同步原語]
    B --> C[最佳化等待策略]
    C --> D[提高 RwLock 效能]

圖表翻譯： 此圖示展示了 RwLock 效能最佳化的步驟，包括減少鎖的爭用、使用更高效的同步原語和最佳化等待策略，最終達到提高 RwLock 效能的目標。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。