只要擺脫鎖，實(shí)現(xiàn)支持安全并發(fā)訪問(wèn)的數(shù)據(jù)結(jié)構(gòu)，就有可能解決大粒度鎖影響并發(fā)程度以及錯(cuò)誤的加鎖方式導(dǎo)致死鎖的問(wèn)題。這種數(shù)據(jù)結(jié)構(gòu)稱為無(wú)鎖數(shù)據(jù)結(jié)構(gòu)。

在了解本文時(shí)，務(wù)必讀懂內(nèi)存次序章節(jié)。

在設(shè)計(jì)無(wú)鎖數(shù)據(jù)結(jié)構(gòu)時(shí)，需要極為小心謹(jǐn)慎，因?yàn)樗鼈兊恼_實(shí)現(xiàn)相當(dāng)不容易。導(dǎo)致代碼出錯(cuò)的情形難以復(fù)現(xiàn)。

1 定義和推論

算法和數(shù)據(jù)結(jié)構(gòu)中，只要采用了互斥，條件變量或future進(jìn)行同步操作，就稱之為阻塞型算法和阻塞型數(shù)據(jù)結(jié)構(gòu)。

如果應(yīng)用程序調(diào)用某些庫(kù)函數(shù)，發(fā)起調(diào)用的線程便會(huì)暫停運(yùn)行，即在函數(shù)的調(diào)用點(diǎn)阻塞，等到另一線程完成某項(xiàng)相關(guān)操作，阻塞才會(huì)解除，前者才會(huì)繼續(xù)運(yùn)行。這種庫(kù)函數(shù)的調(diào)用被命名為阻塞型調(diào)用。

1.1 非阻塞型數(shù)據(jù)結(jié)構(gòu)

在實(shí)踐中，我們需要參考下列定義，根據(jù)適用的條款，分辨該型別/函數(shù)屬于哪一類：

1 無(wú)阻礙：假定其他線程全部暫停，則目標(biāo)線程將在有限步驟內(nèi)完成自己的操作。

2 無(wú)鎖：如果多個(gè)線程共同操作一份數(shù)據(jù)，那么在有限步驟內(nèi)，其中某一線程能夠完成自己的操作。

3 免等：在某份數(shù)據(jù)上，每個(gè)線程經(jīng)過(guò)有限步驟就能完成自己的操作，即使該份數(shù)據(jù)同時(shí)被其他多個(gè)線程所操作。

絕大多數(shù)時(shí)候無(wú)障礙算法并不切實(shí)有用，因?yàn)槠渌€程全部暫停這對(duì)于一個(gè)項(xiàng)目來(lái)說(shuō)是一個(gè)非常匪夷所思的場(chǎng)景。

1.2 無(wú)鎖數(shù)據(jù)結(jié)構(gòu)

免等和無(wú)鎖數(shù)據(jù)結(jié)構(gòu)能夠避免線程受餓問(wèn)題，也就是說(shuō)，兩個(gè)并發(fā)執(zhí)行的線程，其中一個(gè)按部就班的執(zhí)行操作，另一個(gè)總是在錯(cuò)誤的時(shí)機(jī)開(kāi)始執(zhí)行操作，導(dǎo)致被迫中止，反復(fù)開(kāi)始，試圖完成操作。

1.3 免等的數(shù)據(jù)結(jié)構(gòu)

是具備額外功能的無(wú)鎖數(shù)據(jù)結(jié)構(gòu)，如果它被多個(gè)線程訪問(wèn)，不論其他線程上發(fā)生了什么，每個(gè)線程都能在有限的步驟內(nèi)完成自己的操作。若多個(gè)線程之間存在沖突，導(dǎo)致某算法無(wú)限制地反復(fù)嘗試執(zhí)行操作，那它就是免等算法（比如使用while循環(huán)進(jìn)行的一些操作，在里面執(zhí)行比較-交換操作）。

1.4 無(wú)鎖數(shù)據(jù)結(jié)構(gòu)的優(yōu)點(diǎn)和缺點(diǎn)

1.4.1 優(yōu)點(diǎn)

本質(zhì)上，使用無(wú)鎖數(shù)據(jù)結(jié)構(gòu)的首要原因是：最大限度地實(shí)現(xiàn)并發(fā)。

基于鎖的實(shí)現(xiàn)往往導(dǎo)致線程需要阻塞，在無(wú)鎖數(shù)據(jù)結(jié)構(gòu)上，總是存在某個(gè)線程能執(zhí)行下一步操作。免等數(shù)據(jù)結(jié)構(gòu)則完全無(wú)需等待，但是難以實(shí)現(xiàn)，很容易寫(xiě)成自旋鎖。

還有一點(diǎn)是，代碼健壯性。

假設(shè)數(shù)據(jù)結(jié)構(gòu)的寫(xiě)操作受鎖保護(hù)，如果線程在持鎖期間終止，那么該數(shù)據(jù)結(jié)構(gòu)僅完成了部分改動(dòng)，且此后無(wú)從修補(bǔ)。但是，若某線程操作無(wú)鎖數(shù)據(jù)結(jié)構(gòu)時(shí)意外終結(jié)，則丟失的數(shù)據(jù)僅限于它持有的部分，其他數(shù)據(jù)依然完好，能被別的線程進(jìn)行處理（不會(huì)阻塞等待鎖）

1.4.2 缺點(diǎn)。需要注意的地方

1.4.2.1 不變量相關(guān)

力求保持不變量成立，或選取別的可以一直成立的不變量作為替代。

1.4.2.2 留心內(nèi)存次序約束

1.4.2.3 數(shù)據(jù)修改使用原子操作

1.4.2.4 就其他線程所見(jiàn)，各項(xiàng)修改步驟次序正確

1.4.2.5 避免活鎖

假設(shè)兩個(gè)線程同時(shí)更改同一份數(shù)據(jù)結(jié)構(gòu)，若它們所做的改動(dòng)都導(dǎo)致對(duì)方從頭開(kāi)始操作，那雙方就會(huì)反復(fù)循環(huán)，不斷重試，這種現(xiàn)象稱為活鎖。它的出現(xiàn)與否完全取決于現(xiàn)成的調(diào)度次序，故往往只會(huì)短暫存在。因此它們僅降低程序性能，不會(huì)導(dǎo)致嚴(yán)重問(wèn)題。也因此可能會(huì)導(dǎo)致提高了操作同一個(gè)數(shù)據(jù)結(jié)構(gòu)的并發(fā)程度，縮短了單個(gè)線程因等待消耗的時(shí)間，卻降低了整體的性能。

1.4.2.6 緩存乒乓

并且，如果多個(gè)線程訪問(wèn)相同的原子變量，硬件必須在線程之間同步數(shù)據(jù)，這還會(huì)造成緩存乒乓現(xiàn)象，導(dǎo)致嚴(yán)重的性能損耗。

2 無(wú)鎖數(shù)據(jù)結(jié)構(gòu)范例

無(wú)鎖數(shù)據(jù)結(jié)構(gòu)依賴原子操作和內(nèi)存次序約束（作用是令其他線程按正確的內(nèi)存次序見(jiàn)到數(shù)據(jù)操作的過(guò)程），默認(rèn)內(nèi)存次序std::memory_order_seq_cst最易于分析和推理（全部該次序的操作形成確定且唯一的總序列）

2.1 實(shí)現(xiàn)線程安全的無(wú)鎖棧

需要保證：

1 一旦某線程將一項(xiàng)數(shù)據(jù)加入棧容器，就能立即安全地被另一線程取出

2 只有唯一一個(gè)線程能獲取該項(xiàng)數(shù)據(jù)

#include <atomic>
#include <memory>template<typename T>
class lock_free_stack {
private:struct node {std::shared_ptr<T> data;node* next;node(T const& data_) : data(std::make_shared<T>(data_)) {}};std::atomic<node*> head;
public:void push(T const& data) {node* const new_node = new node(data);new_node->next = head.load();while (!head.compare_exchange_weak(new_node->next, new_node));}std::shared_ptr<T> pop() {node* old_head = head.load();while(old_head && !head.compare_exchange_weak(old_head, old_head->next));return old_head ? old_head->data : std::shared_ptr<T>();}
};

比較交換操作：如果head指針與第一個(gè)參數(shù)new_node->next所存儲(chǔ)值相同，將head改為指向第二個(gè)參數(shù)new_node，compare_exchange_weak返回true。如果head指針與第一個(gè)參數(shù)new_node->next所存儲(chǔ)值不同，表示head指針被其他線程修改過(guò)，第一個(gè)參數(shù)new_node->next就被更新成head指針的當(dāng)前值，并且compare_exchange_weak返回false，讓循環(huán)繼續(xù)。

上述代碼雖然是無(wú)鎖實(shí)現(xiàn)，但是卻是非免等的，如果compare_exchange_weak總是false，理論上push和pop中的while循環(huán)要持續(xù)進(jìn)行。

2.2 制止內(nèi)存泄漏：在無(wú)鎖數(shù)據(jù)結(jié)構(gòu)中管理內(nèi)存

本質(zhì)問(wèn)題是：若要?jiǎng)h除某節(jié)點(diǎn)，必須先行確認(rèn)，其他線程并未持有指向該節(jié)點(diǎn)的指針。

對(duì)于上述實(shí)現(xiàn)，若有多個(gè)線程同時(shí)調(diào)用pop，需要采取措施判斷何時(shí)刪除節(jié)點(diǎn)。

可以維護(hù)一個(gè)等待刪除鏈表。

#include <atomic>
#include <memory>template<typename T>
class lock_free_stack {
private:struct node {std::shared_ptr<T> data;node* next;node(T const& data_) : data(std::make_shared<T>(data_)) {}};std::atomic<node*> head;std::atomic<unsigned> threads_in_pop;std::atomic<node *> to_be_deleted;static void delete_nodes(node* nodes) {while (nodes) {node* next = nodes->next;delete nodes;nodes = next;}}void try_reclaim(node* old_head) {if (threads_in_pop == 1) {node* nodes_to_delete = to_be_deleted.exchange(nullptr);if (!--threads_in_pop) {delete_nodes(nodes_to_delete);} else if (nodes_to_delete) {chain_pending_nodes(nodes_to_delete);}delete old_head;} else {chain_pending_node(old_head);--threads_in_pop;}}void chain_pending_nodes(node* nodes) {node* last = nodes;while (node* const next = last->next) {last = next;}chain_pending_nodes(nodes, last);}void chain_pending_nodes(node* first, node* last) {last->next = to_be_deleted;while (!to_be_deleted.compare_exchange_weak(last->next, first));}void chain_pending_node(node* n) {chain_pending_nodes(n, n);}  
public:void push(T const& data) {node* const new_node = new node(data);new_node->next = head.load();while (!head.compare_exchange_weak(new_node->next, new_node));}std::shared_ptr<T> pop() {++threads_in_pop;node* old_head = head.load();while(old_head && !head.compare_exchange_weak(old_head, old_head->next));std::shared_ptr<T> res;if (old_head) {res.swap(old_head->data);}try_reclaim(old_head);return res;}
};

2.3 運(yùn)用風(fēng)險(xiǎn)指針檢測(cè)無(wú)法回收的節(jié)點(diǎn)

術(shù)語(yǔ)“風(fēng)險(xiǎn)指針”是一種技法，得名緣由是：若某節(jié)點(diǎn)仍被其他線程指涉，而我們依然刪除它，此舉便成了“冒險(xiǎn)”動(dòng)作。刪除目標(biāo)節(jié)點(diǎn)后，別的線程還持有指向它的引用，還通過(guò)這一引用對(duì)其進(jìn)行訪問(wèn)，便會(huì)導(dǎo)致程序產(chǎn)生未定義行為。

上述機(jī)制產(chǎn)生的基本思想：假設(shè)當(dāng)前線程要訪問(wèn)某對(duì)象，而它卻即將被別的線程刪除，那就讓當(dāng)前線程設(shè)置指涉目標(biāo)對(duì)象的風(fēng)險(xiǎn)指針，以通知其他線程刪除該對(duì)象將產(chǎn)生實(shí)質(zhì)風(fēng)險(xiǎn)。若程序不再需要該對(duì)象，風(fēng)險(xiǎn)指針被清零。

#include <atomic>
#include <thread>
unsigned const max_hazard_pointers=100;
struct hazard_pointer {std::atomic<std::thread::id> id;std::atomic<void*> pointer;
};hazard_pointer hazard_pointers[max_hazard_pointers];class hp_owner {hazard_pointer* hp;public:hp_owner(hp_owner const&)=delete;hp_owner operator=(hp_owner const&)=delete;hp_owner(): hp(nullptr) {for (unsigned i = 0; i < max_hazard_pointers; ++i) {std::thread::id old_id;if (hazard_pointers[i].id.compare_exchange_strong(old_id, std::this_thread::get_id())) {hp = &hazard_pointers[i];break;}}if (!hp) {throw std::runtime_error("No hazard");}}std::atomic<void*>& get_pointer() {return hp->pointer;}~hp_owner() {hp->pointer.store(nullptr);hp->id.store(std::thread::id());}
};std::atomic<void*>& get_hazard_pointer_for_current_thread() {thread_local static hp_owner hazard;return hazard.get_pointer();
}

2.4 借引用計(jì)數(shù)檢測(cè)正在使用中的節(jié)點(diǎn)

#include <atomic>
#include <memory>template<typename T>
class lock_free_stack {
private:struct node {std::shared_ptr<T> data;node* next;node(T const& data_) : data(std::make_shared<T>(data_)) {}};std::shared_ptr<node> head;
public:void push(T const& data) {std::shared_ptr<node> const new_node = std::make_shared<node>(data);new_node->next = std::atomic_load(&head);while (!std::atomic_compare_exchange_weak(&head, &new_node->next, new_node));}std::shared_ptr<T> pop() {std::shared_ptr<node> old_head = std::atomic_load(&head);while (old_head && std::atomic_compare_exchange_weak(&head, &old_head, std::atomic_load(&old_head->next)));if (old_head) {std::atomic_store(&old_head->next, std::shared_ptr<node>());return old_head->next;}return std::shared_ptr<T>();}~lock_free_stack() {while (pop());}
};

引用計(jì)數(shù)針對(duì)各個(gè)節(jié)點(diǎn)分別維護(hù)一個(gè)計(jì)數(shù)器，隨時(shí)了解訪問(wèn)它的線程數(shù)目。

std::shared_ptr的引用計(jì)數(shù)在這里無(wú)法借鑒，因?yàn)樗脑犹匦圆灰欢ㄍㄟ^(guò)無(wú)鎖機(jī)制實(shí)現(xiàn)，若強(qiáng)行按照無(wú)鎖方式實(shí)現(xiàn)該指針類的原子操作，很可能造成額外開(kāi)銷(xiāo)。

2.4.1?std::experimental::atomic_shared_ptr<T>

std::shared_ptr無(wú)法結(jié)合std::atomic<>使用，原因是std::shared_ptr<T>并不具備平實(shí)拷貝語(yǔ)義。但是std::experimental::atomic_shared_ptr<T>支持，因此可以正確的處理引用計(jì)數(shù)，同時(shí)令操作原子化。

#include <atomic>
#include <memory>template<typename T>
class lock_free_stack {
private:struct node {std::shared_ptr<T> data;std::experimental::atomic_shared_ptr<node> next;node(T const& data_) : data(std::make_shared<T>(data_)) {}};std::experimental::atomic_shared_ptr<node> head;
public:void push(T const& data) {std::shared_ptr<node> const new_node = std::make_shared<node>(data);new_node->next = std::atomic_load(&head);while (!std::atomic_compare_exchange_weak(&head, &new_node->next, new_node));}std::shared_ptr<T> pop() {std::shared_ptr<node> old_head = std::atomic_load(&head);while (old_head && std::atomic_compare_exchange_weak(&head, &old_head, std::atomic_load(&old_head->next)));if (old_head) {std::atomic_store(&old_head->next, std::shared_ptr<node>());return old_head->data;}return std::shared_ptr<T>();}~lock_free_stack() {while (pop());}
};

2.4.2?內(nèi)、外部計(jì)數(shù)器進(jìn)行引用計(jì)數(shù)

一種經(jīng)典的實(shí)現(xiàn)是，使用兩個(gè)計(jì)數(shù)器：內(nèi)、外部計(jì)數(shù)器各一。兩個(gè)計(jì)數(shù)器之和即為節(jié)點(diǎn)的總引用數(shù)目。

外部計(jì)數(shù)器與節(jié)點(diǎn)的指針組成結(jié)構(gòu)體，每當(dāng)指針被讀取，外部計(jì)數(shù)器自增。

內(nèi)部計(jì)數(shù)器位于節(jié)點(diǎn)之中，隨著節(jié)點(diǎn)讀取完成自減。

#include <atomic>
#include <memory>template<typename T>
class lock_free_stack {
private:struct node;struct counted_node_ptr {int external_count;node* ptr;};struct node {std::shared_ptr<T> data;std::atomic<int> internal_count;counted_node_ptr next;node(T const& data_) : data(std::make_shared<T>(data_)), internal_count(0) {}};std::atomic<counted_node_ptr> head;void increase_head_count(counted_node_ptr& old_counter) {counted_node_ptr new_counter;do {new_counter = old_counter;++new_counter.external_count;} while (!head.compare_exchange_strong(old_counter, new_counter));old_counter.external_count = new_counter.external_count;}
public:std::shared_ptr<T> pop() {counted_node_ptr old_head = head.load();for (;;) {increase_head_count(old_head);node* const ptr = old_head.ptr;if (!ptr) {return std::shared_ptr<T>();}if (head.compare_exchange_strong(old_head, ptr->next)) {std::shared_ptr<T> res;res.swap(ptr->data);int const count_increase = old_head.external_count - 2;if (ptr->internal_count.fetch_add(count_increase) == -count_increase) {delete ptr;}return res;} else if (ptr->internal_count.fetch_sub(1) == 1) {delete ptr;}}}void push(T const& data) {counted_node_ptr new_node;new_node.ptr = new node(data);new_node.external_count = 1; // head指針本身算作一個(gè)外部引用new_node.ptr->next = head.load();while (!head.compare_exchange_weak(new_node.ptr->next, new_node));}~lock_free_stack() {while (pop());}
};

結(jié)構(gòu)體counted_node_ptr的尺寸足夠小，如果硬件平臺(tái)支持雙字 比較-交換 操作，那么std::atomic<counted_node_ptr>就屬于無(wú)鎖數(shù)據(jù)。若不支持，那么std::atomic<>涉及的結(jié)構(gòu)體的尺寸過(guò)大，無(wú)法直接通過(guò)原子指令操作，便會(huì)采用互斥來(lái)保證操作原子化。使得“無(wú)鎖”數(shù)據(jù)結(jié)構(gòu)和算法成為基于鎖的實(shí)現(xiàn)。

如果想要縮小結(jié)構(gòu)體counted_node_ptr的尺寸，可以采取另一種方法替代：假定在硬件平臺(tái)上，指針型別有空余的位。（例如，硬件尋址空間只有48位，指針型別的大小是64位），他們可以用來(lái)放置計(jì)數(shù)器，借此將counted_node_ptr結(jié)構(gòu)體縮成單個(gè)機(jī)器字長(zhǎng)。

使用分離引用計(jì)數(shù)的原因：我們通過(guò)外部引用計(jì)數(shù)的自增來(lái)保證，在訪問(wèn)目標(biāo)節(jié)點(diǎn)的過(guò)程中，其指針依然安全有效。（先自增，再讀取，被指涉后自增的值保護(hù)了節(jié)點(diǎn)不被刪除）

具體流程詳解見(jiàn)《cpp 并發(fā)實(shí)戰(zhàn)》p236

以上實(shí)例使用的內(nèi)存次序是std::memory_order_seq_cst。同步開(kāi)銷(xiāo)較大，下面對(duì)于內(nèi)存次序進(jìn)行優(yōu)化。

2.5 為無(wú)鎖棧容器施加內(nèi)存模型

需要先確認(rèn)各項(xiàng)操作哪些存在內(nèi)存次序關(guān)系。

1 next指針只是一個(gè)未被原子化的普通對(duì)象，所以為了安全讀取其值，存儲(chǔ)操作必須再載入操作發(fā)生之前，前者由壓入數(shù)據(jù)的線程執(zhí)行，后者由彈出數(shù)據(jù)的線程執(zhí)行。

#include <atomic>
#include <memory>template<typename T>
class lock_free_stack {
private:struct node;struct counted_node_ptr {int external_count;node* ptr;};struct node {std::shared_ptr<T> data;std::atomic<int> internal_count;counted_node_ptr next;node(T const& data_) : data(std::make_shared<T>(data_)), internal_count(0) {}};std::atomic<counted_node_ptr> head;void increase_head_count(counted_node_ptr& old_counter) {counted_node_ptr new_counter;do {new_counter = old_counter;++new_counter.external_count;} while (!head.compare_exchange_strong(old_counter, new_counter, std::memory_order_acquire, std::memory_order_relaxed));old_counter.external_count = new_counter.external_count;}
public:std::shared_ptr<T> pop() {counted_node_ptr old_head = head.load(std::memory_order_relaxed);for (;;) {increase_head_count(old_head);node* const ptr = old_head.ptr;if (!ptr) {return std::shared_ptr<T>();}if (head.compare_exchange_strong(old_head, ptr->next, std::memory_order_relaxed)) {std::shared_ptr<T> res;res.swap(ptr->data);int const count_increase = old_head.external_count - 2;if (ptr->internal_count.fetch_add(count_increase, std::memory_order_relaxed) == -count_increase) { // ?delete ptr;}return res;} else if (ptr->internal_count.fetch_add(-1, std::memory_order_relaxed) == 1) {ptr->internal_count.load(std::memory_order_acquire);delete ptr;}}}void push(T const& data) {counted_node_ptr new_node;new_node.ptr = new node(data);new_node.external_count = 1;new_node.ptr->next = head.load(std::memory_order_relaxed);while (!head.compare_exchange_weak(new_node.ptr->next, new_node, std::memory_order_release, std::memory_order_relaxed));}~lock_free_stack() {while (pop());}
};

這里的push，唯一的原子操作是compare_exchange_weak，如果需要在線程間構(gòu)成先行關(guān)系，則代碼需要一項(xiàng)釋放操作，因此compare_exchange_weak必須采用std::memory_order_release或者更嚴(yán)格的內(nèi)存次序。

若compare_exchange_weak執(zhí)行失敗，則指針head和new_node均無(wú)變化，代碼繼續(xù)執(zhí)行，這種情況下使用memory_order_relaxed即可。

（沒(méi)懂，為什么是這兩個(gè)次序）

這里的pop，訪問(wèn)next指針前進(jìn)行了額外操作。也就是先調(diào)用了increase_head_count()，該操作收到memory_order_acquire或者更嚴(yán)格的內(nèi)存次序約束。因?yàn)檫@里通過(guò)原子操作獲取的head指針舊值訪問(wèn)next指針。對(duì)原子操作失敗的情況則使用寬松次序。

因?yàn)閜ush中的存儲(chǔ)行為是釋放操作，pop，increase_head這里的是獲取操作，因此存儲(chǔ)行為和載入操作同步，構(gòu)成先行關(guān)系。因此，對(duì)于push中的成員指針ptr的存儲(chǔ)操作先行發(fā)生，然后pop才會(huì)在increase_head_count()中訪問(wèn)ptr->next，代碼符合線程安全。

（push中的head.load不影響上述內(nèi)存次序關(guān)系的分析）

剩余的很多沒(méi)看懂，在《cpp并發(fā)實(shí)戰(zhàn)》p240，有空多看看。

2.6 實(shí)現(xiàn)線程安全的無(wú)鎖隊(duì)列

對(duì)于隊(duì)列，其pop和push分別訪問(wèn)不同的部分。

#include <atomic>
#include <memory>
#include <mutex>template<typename T>
class lock_free_queue {
private:struct node {std::shared_ptr<T> data;node* next;node() : next(nullptr) {}};std::atomic<node*> head;std::atomic<node*> tail;std::unique_ptr<node> pop_head() {node* const old_head = head.load();if (head.get() == tail) {return nullptr;}head.store(old_head->next);return old_head;}public:lock_free_queue() : head(new node), tail(head.load()) {}lock_free_queue(const lock_free_queue& other) = delete;lock_free_queue& operator=(const lock_free_queue& other) = delete;~lock_free_queue() {while (node* const old_head = head.load()) {head.store(old_head->next);delete old_head;}}std::shared_ptr<T> pop() {node* old_head = pop_head();if (!old_head) {return std::shared_ptr<T>();}std::shared_ptr<T> const res(old_head->data);delete old_head;return res;}void push(T new_value) {std::shared_ptr<T> new_data(std::make_shared<T>(std::move(new_value)));node* p = new node;node* const old_tail = tail.load();old_tail->data.swap(new_data);old_tail->next = p;tail.store(p);}
};

tail指針的存儲(chǔ)操作store（push的最后一句）和載入操作load（pop_head的if里面的條件判斷）存在同步。

但是若是由多個(gè)線程進(jìn)行操作，上述代碼就不可行。

其余細(xì)節(jié)具體實(shí)現(xiàn)見(jiàn)書(shū)。

3 實(shí)現(xiàn)無(wú)鎖數(shù)據(jù)結(jié)構(gòu)的原則

3.1 在原型設(shè)計(jì)中使用std::memory_order_seq_cst次序

它令全部操作形成一個(gè)確定的總序列，比較好分析。在這種意義上，使用其他內(nèi)存次序就成為了一種優(yōu)化。

3.2 使用無(wú)鎖的內(nèi)存回收方案

無(wú)所代碼中的內(nèi)存管理很難。最基本的要求是，只要目標(biāo)對(duì)象仍然有可能背其他線程指涉，就不刪除。

在這里介紹了三種方法來(lái)滿足及時(shí)刪除無(wú)用對(duì)象：

1 暫緩所有刪除對(duì)象的動(dòng)作，等到五線程訪問(wèn)再刪除（類似gc）

2 風(fēng)險(xiǎn)指針

3 引用計(jì)數(shù)

3.3 防范ABA問(wèn)題

在所有設(shè)計(jì) 比較-交換 的算法中，都要防范ABA問(wèn)題。

該問(wèn)題產(chǎn)生的過(guò)程如下：

步驟1：線程甲讀取原子變量x，得知其值為A。

步驟2：線程甲根據(jù)A執(zhí)行某項(xiàng)操作，比如查找，或如果x是指針，則依據(jù)它提取出相關(guān)值。

步驟3：線程甲因操作系統(tǒng)調(diào)度而發(fā)生阻塞。

步驟4：另一線程對(duì)原子變量x執(zhí)行別的操作，將其值改成B。

步驟5：又有線程改變了與A相關(guān)的數(shù)據(jù)，使得線程甲原本持有的值失效（步驟2中的相關(guān)值）。這種情形也許是A表示某內(nèi)存地址，而改動(dòng)操作則是釋放指針的目標(biāo)內(nèi)存，或變更目標(biāo)數(shù)據(jù)，最后將產(chǎn)生嚴(yán)重后果。

步驟6：原子變量x再次被某線程改動(dòng)，重新變回A。若x屬于指針型別，其指向目標(biāo)可能在步驟5被改換程一個(gè)新對(duì)象。

步驟7：線程甲繼續(xù)運(yùn)行，在原子變量x上執(zhí)行 比較-交換 操作，與A進(jìn)行對(duì)比。因此 比較-交換 操作成功執(zhí)行（因?yàn)閤的值仍然為A），但A的關(guān)聯(lián)數(shù)據(jù)卻不再有效，即原本在步驟2取的相關(guān)值已經(jīng)失效，但是線程甲卻無(wú)從分辨，這將破壞數(shù)據(jù)結(jié)構(gòu)。

該問(wèn)題最常見(jiàn)的解決方法之一是，在原子變量x中引入一個(gè)ABA計(jì)數(shù)器。將變量x和計(jì)數(shù)器組成單一結(jié)構(gòu)，作為一個(gè)整體執(zhí)行 比較-交換 操作。

3.4 找出忙等循環(huán)，協(xié)助其他線程

若兩個(gè)線程同時(shí)執(zhí)行push操作，那么必須等另一個(gè)結(jié)束，才可以繼續(xù)運(yùn)行。這是忙等，浪費(fèi)cpu。在本例中將非原子變量的數(shù)據(jù)成員改為原子變量，并采用 比較-交換 操作設(shè)置其值。

4 小結(jié)

這節(jié)很難