并發 -- 無鎖算法與結構

文章目錄

- - 什么是無鎖算法
  - 什么是原子變量
  - 什么是CAS操作
  - Compare-And-Swap Weak在哪些情況下會失敗
  - 舉例說明無鎖結構
  - 無鎖結構的問題

什么是無鎖算法

無鎖算法（Lock-Free Algorithm）是一種并發編程技術，旨在實現多線程環境下的高效數據共享，而無需使用傳統的鎖機制（如互斥鎖）。
關鍵特性或優點：

無阻塞：至少有一個線程能在有限步驟內完成操作，確保系統整體不會因鎖爭用而停滯。
無死鎖：由于不使用鎖，避免了死鎖問題。
高并發性：適用于多核處理器，多個線程可同時操作共享數據，減少等待時間。

在C++ 可以使用原子變量來實現無鎖算法與結構

什么是原子變量

C++的原子變量（Atomic Variable）是通過std::atomic模板類提供的，用于在多線程環境中安全地操作共享數據
模板原形

// Defined in header <atomic>
template< class T >
struct atomic;  // (1)	(since C++11)
template< class U >
struct atomic<U*>; //(2)	(since C++11)
// Defined in header <memory>
template< class U >
struct atomic<std::shared_ptr<U>>;  //(3)	(since C++20)
template< class U >
struct atomic<std::weak_ptr<U>>; // (4)	(since C++20)
Defined in header <stdatomic.h>
#define _Atomic(T) /* see below */  // (5)	(since C++23)

具體參考cppreference atomic

下面是一個使用atomic_flag 實現自旋鎖的例子

class SpinLock {std::atomic_flag flag(ATOMIC_FLAG_INIT);
public:void lock(){while(flag.test_and_set(std::memory_and_acquire)){}}void unlock(){flag.clear(std::memory_and_release);}};

其中std::memory_and_release與 std::memory_and_acquire的含義需自行查找內存模型與順序

使用atomic 操作就是沒有顯示的使用鎖操作，但atomic內部可能是使用鎖機制來實現的也可能是使用cpu的CAS操作來實現的
通過atomic::is_lock_free可以查看atomic內部是不是無鎖機制實現的

#include <atomic>
#include <iostream>
#include <utility>struct A { int a[100]; };
struct B { int x, y; };
struct C { int x, y, z; };int main()
{std::cout << std::boolalpha<< "std::atomic_char is lock free? "<< std::atomic_char{}.is_lock_free() << '\n'<< "std::atomic_uintmax_t is lock free? "<< std::atomic_uintmax_t{}.is_lock_free() << '\n'<< "std::atomic<A> is lock free? "<< std::atomic<A>{}.is_lock_free() << '\n'<< "std::atomic<B> is lock free? "<< std::atomic<B>{}.is_lock_free() << '\n'<< "std::atomic<C> is lock free? "<< std::atomic<C>{}.is_lock_free() << '\n';
}

是與類型的長度有關的

什么是CAS操作

CAS（Compare-And-Swap，比較并交換）是一種原子操作，用于在多線程環境中實現無鎖同步。CAS操作會檢查某個內存位置的值是否與預期值匹配，如果匹配，則將其更新為新值；否則，不進行任何操作。無論是否更新，CAS操作都會返回內存位置的原始值。

CAS 操作包含三個操作數：

內存地址：需要修改的共享變量的地址。
期望值：當前線程認為該變量應該具有的值。
新值：如果變量的當前值等于期望值，則將其更新為新值。

CAS 的工作流程：

讀取內存地址中的當前值。
比較當前值與期望值：
- 如果相等，則將新值寫入內存地址。
- 如果不相等，則不做任何操作。
返回操作是否成功（通常返回當前值或布爾值）。

CAS 的硬件支持：

x86：CMPXCHG 指令（單字 CAS），CMPXCHG16B 指令（雙字 CAS）。
ARM：LDREX 和 STREX 指令（加載-存儲獨占指令）。
其他架構：大多數現代 CPU 都提供了類似的原子指令。

以下是一個使用 CMPXCHG 實現原子操作的 C++ 示例：

#include <iostream>
#include <atomic>// 使用 CMPXCHG 實現原子 CAS 操作
bool atomic_compare_exchange(int* dest, int expected, int new_value) {int result;__asm__ volatile ("lock cmpxchgl %3, %1;"  // lock 前綴確保原子性，cmpxchgl 是 32 位 CMPXCHG: "=a" (result), "+m" (*dest)  // 輸出：result = EAX，dest 是內存操作數: "a" (expected), "r" (new_value)  // 輸入：EAX = expected，new_value 是寄存器操作數: "memory"  // 告訴編譯器內存可能被修改);return result == expected;  // 返回是否成功
}int main() {int shared_value = 10;int expected = 10;int new_value = 20;std::cout << "Initial value: " << shared_value << std::endl;// 嘗試原子地將 shared_value 從 expected (10) 更新為 new_value (20)bool success = atomic_compare_exchange(&shared_value, expected, new_value);if (success) {std::cout << "CAS succeeded. New value: " << shared_value << std::endl;} else {std::cout << "CAS failed. Value is still: " << shared_value << std::endl;}return 0;
}

缺點:

ABA 問題：變量的值從 A 變為 B 又變回 A，CAS 無法檢測到中間變化。
忙等待：在高競爭場景下，CAS 可能導致大量重試，浪費 CPU 資源。
復雜性：實現無鎖數據結構需要復雜的邏輯和嚴格的正確性驗證。

CAS操作的變體：

Compare-And-Swap Weak：允許在某些情況下失敗（如虛假失敗），但性能更高。
Compare-And-Swap Strong：確保只有在值不匹配時才失敗，性能較低但更可靠。

Compare-And-Swap Weak在哪些情況下會失敗

當前值與預期值不匹配
內存問題：
- 并發競爭，如果多個線程同時嘗試修改同一個內存地址，CAS Weak 可能會因為競爭而失敗。即使當前值與預期值相等，其他線程的干擾也可能導致操作失敗。
- 內存訪問沖突：如果內存訪問沖突頻繁發生，硬件可能會放棄 CAS Weak 操作，導致失敗。
- 在某些弱內存模型（如 ARM 或 PowerPC）中，指令重排序可能會導致 CAS Weak 失敗。例如，內存屏障未正確設置時，CAS Weak 可能會觀察到不一致的內存狀態。
- 如果系統資源（如緩存、內存帶寬）緊張，CAS Weak 可能會因為資源競爭而失敗。
- 緩存一致性協議的影響：在某些多核處理器架構中，緩存一致性協議（如 MESI 協議）可能導致 CAS Weak 失敗。例如，當多個核心同時競爭同一內存地址時，緩存行的狀態可能會發生變化，從而導致 CAS Weak 失敗

Compare-And-Swap Weak 的偽代碼實現

template<T>
bool cas_weak(T* dest, T& expected,const T& new_value)
{if (*dest == expected) {if (try_update(dest,new_value)){return true;} else {return false;}} else {expected = *dest;return false}
}

舉例說明無鎖結構

無鎖鏈表

#include <atomic>
#include <memory>
#include <iostream>template <typename T>
class LockFreeLinkedList {
private:struct Node {std::shared_ptr<T> data; // 存儲數據std::atomic<Node*> next; // 指向下一個節點的原子指針Node(T const& value) : data(std::make_shared<T>(value)), next(nullptr) {}};std::atomic<Node*> head; // 鏈表頭指針public:LockFreeLinkedList() : head(nullptr) {}~LockFreeLinkedList() {// 析構時釋放所有節點Node* current = head.load();while (current) {Node* next = current->next.load();delete current;current = next;}}// 插入操作void insert(T const& value) {Node* new_node = new Node(value); // 創建新節點new_node->next = head.load();    // 新節點的 next 指向當前頭節點// CAS 操作：將 head 從當前值更新為新節點while (!head.compare_exchange_weak(new_node->next, new_node)) {// 如果 CAS 失敗，說明 head 已經被其他線程修改，重新嘗試}}// 刪除操作bool remove(T const& value) {Node* prev = head.load(); // 前驅節點Node* curr = prev;        // 當前節點while (curr) {Node* next = curr->next.load(); // 下一個節點// 如果當前節點的值匹配if (curr->data && *curr->data == value) {// CAS 操作：將前驅節點的 next 從當前節點更新為下一個節點if (prev == curr) {// 如果當前節點是頭節點if (head.compare_exchange_weak(curr, next)) {delete curr; // 刪除節點return true;}} else {if (prev->next.compare_exchange_weak(curr, next)) {delete curr; // 刪除節點return true;}}}// 移動指針prev = curr;curr = next;}return false; // 未找到匹配的節點}// 打印鏈表內容（用于調試）void print() const {Node* current = head.load();while (current) {if (current->data) {std::cout << *current->data << " -> ";}current = current->next.load();}std::cout << "nullptr" << std::endl;}
};

無鎖結構的問題

ABA 問題：在無鎖數據結構中，ABA 問題是一個常見的挑戰。可以通過使用帶有版本號的指針（如 std::atomic<std::shared_ptr>）或雙字 CAS（Compare-And-Swap）來解決。
內存管理：無鎖數據結構中的內存管理需要特別小心，因為多個線程可能同時訪問和釋放內存。通常使用智能指針（如 std::shared_ptr）來管理內存。
性能測試：無鎖數據結構的性能在高并發環境下可能優于基于鎖的數據結構，但在低并發環境下可能表現不佳。因此，在實際應用中需要進行充分的性能測試。