點擊上方“C語言與CPP編程”，選擇“關注/置頂/星標公眾號”
干貨/資訊/熱點/福利，每天準時送達！
最近有小伙伴說沒有收到當天的文章推送，這是因為微信改了推送機制，確實會一部分有小伙伴刷不到當天的文章，一些比較實用的知識和信息可能會錯過。所以建議大家加個星標??，就能第一時間收到推送。
小伙伴們大家好，我是飛宇。
今天繼續更新《Effective C++》和《C++并發編程實戰》的讀書筆記，下面是已經更新過的內容：
《C++并發編程實戰》讀書筆記（1）：并發、線程管控
《C++并發編程實戰》讀書筆記（2）：并發操作的同步
《C++并發編程實戰》讀書筆記（3）：內存模型和原子操作
《Effective C++》讀書筆記（1）：讓自己習慣C++
《Effective C++》讀書筆記（2）：構造/析構/賦值運算
《Effective C++》讀書筆記（3）：資源管理
本文包括第6章設計基于鎖的并發數據結構與第7章設計無鎖數據結構，后者實在有些燒腦了。此外，發現吳天明版的中譯本有太多太離譜的翻譯錯誤了，還得是中英對照才行：）
第6章 設計基于鎖的并發數據結構
????設計支持并發訪問的數據結構時，一方面需要確保訪問安全，通常需要限定其提供的接口，另一方面需要是按真正的并發操作，僅利用互斥保護并發實際上是串行化。
????設計基于鎖的并發數據結構的奧義就是確保先鎖定合適的互斥，再訪問數據，并盡可能縮短持鎖時間。
????可以采用鎖實現線程安全的棧容器。
struct?empty_stack?:?std::exception?{ const char* what() const throw() { return "empty stack"; }};
template <typename T>class threadsafe_stack {private: std::stack data; mutable std::mutex m;
public: threadsafe_stack() {} threadsafe_stack(const threadsafe_stack& other) { std::lock_guard<std::mutex> lock(other.m); data = other.data; } threadsafe_stack& operator=(const threadsafe_stack&) = delete;
void push(T new_value) { std::lock_guard<std::mutex> lock(m); data.push(std::move(new_value)); } std::shared_ptr pop() { std::lock_guard<std::mutex> lock(m); if (data.empty()) throw empty_stack(); std::shared_ptr const res( std::make_shared(std::move(data.top()))); data.pop(); return res;????}};
????可以采用鎖和條件變量實現線程安全的隊列容器。data_queue中存儲shared_ptr而非原始值，是為了把shared_ptr的初始化從wait_and_pop移動到push處，使得wait_and_pop中不會拋出異常。否則假如push操作通知條件變量時有多個消費者線程在等待，被notify_one通知到的消費者線程初始化shared_ptr時拋出異常，那么其他消費者線程不會被喚醒。
#include #include #include #include
template <typename T>class threadsafe_queue {private: mutable std::mutex mut; std::queue<std::shared_ptr > data_queue; std::condition_variable data_cond;
public:????threadsafe_queue()?{}
std::shared_ptr wait_and_pop() { std::unique_lock<std::mutex> lk(mut); data_cond.wait(lk, [this] { return !data_queue.empty(); }); std::shared_ptr res = data_queue.front(); data_queue.pop(); return res;????}
void push(T new_value) { std::shared_ptr data(std::make_shared(std::move(new_value))); std::lock_guard<std::mutex> lk(mut); data_queue.push(data); data_cond.notify_one(); }};
????上面的容器直接保護整個data_queue，只用了一個互斥，可以采取更精細粒度的鎖操作。為此使用基于單向鏈表實現的隊列，單向鏈表包含一個不含數據的頭節點，后續的每個節點存儲指向數據的指針與指向下一個節點的指針。這樣的話，就可以對頭尾節點分別加鎖，減小鎖的粒度。
template <typename T>class threadsafe_queue {private: struct node { std::shared_ptr data; std::unique_ptr next; };
std::mutex head_mutex;????std::mutex?tail_mutex;????std::unique_ptr?head; node* tail; std::condition_variable data_cond;???? std::unique_ptr wait_pop_head() { std::unique_lock<std::mutex> head_lock(wait_for_data()); return pop_head();????}????????std::unique_lock<std::mutex>?wait_for_data()?{ std::unique_lock<std::mutex> head_lock(head_mutex); data_cond.wait(head_lock, [&] { return head != get_tail(); }); return std::move(head_lock); }????????node*?get_tail()?{ std::lock_guard<std::mutex> tail_lock(tail_mutex); return tail; } ????std::unique_ptr?pop_head()?{ std::unique_ptr const old_head = std::move(head); head = std::move(old_head->next); return old_head; }
public: threadsafe_queue() : head(new node), tail(head.get()) {} threadsafe_queue(const threadsafe_queue& other) = delete; threadsafe_queue& operator=(const threadsafe_queue& other) = delete;
????std::shared_ptr wait_and_pop() { std::unique_ptr const old_head = wait_pop_head(); return old_head->data;????}???? void push(T new_value) { std::shared_ptr new_data(std::make_shared(std::move(new_value))); std::unique_ptr p(new node); { std::lock_guard<std::mutex> tail_lock(tail_mutex); tail->data = new_data; node* const new_tail = p.get(); tail->next = std::move(p); tail = new_tail; } data_cond.notify_one();????}};
????與棧和隊列相比，大多數數據結構支持多種多樣的操作，要考慮更多訪問模式。例如對于字典來說，首先只考慮基本操作增刪改查，其次實現上來說散列表比二叉樹和有序數組更支持細粒度的鎖，因此必須放棄std::map支持的多余接口、迭代器操作、默認的底層實現。
template <typename Key, typename Value, typename Hash = std::hash >class threadsafe_lookup_table {private: class bucket_type { private: typedef std::pair bucket_value; typedef std::list bucket_data; typedef typename bucket_data::iterator bucket_iterator;
bucket_data data; mutable std::shared_mutex mutex;
bucket_iterator find_entry_for(Key const& key) const { return std::find_if( data.begin(), data.end(), [&](bucket_value const& item) { return item.first == key; }); }
public: Value value_for(Key const& key, Value const& default_value) const { std::shared_lock<std::shared_mutex> lock(mutex); bucket_iterator const found_entry = find_entry_for(key); return (found_entry == data.end()) ? default_value : found_entry->second; }
void add_or_update_mapping(Key const& key, Value const& value) { std::unique_lock<std::shared_mutex> lock(mutex); bucket_iterator const found_entry = find_entry_for(key); if (found_entry == data.end()) { data.push_back(bucket_value(key, value)); } else { found_entry->second = value; } }
void remove_mapping(Key const& key) { std::unique_lock<std::shared_mutex> lock(mutex); bucket_iterator const found_entry = find_entry_for(key); if (found_entry != data.end()) { data.erase(found_entry); } } };
std::vector<std::unique_ptr > buckets; Hash hasher;
bucket_type& get_bucket(Key const& key) const { std::size_t const bucket_index = hasher(key) % buckets.size(); return *buckets[bucket_index]; }
public: typedef Key key_type; typedef Value mapped_type; typedef Hash hash_type;
threadsafe_lookup_table(unsigned num_buckets = 19, Hash const& hasher_ = Hash()) : buckets(num_buckets), hasher(hasher_) { for (unsigned i = 0; i < num_buckets; ++i) { buckets[i].reset(new bucket_type); } }
threadsafe_lookup_table(threadsafe_lookup_table const& other) = delete; threadsafe_lookup_table& operator=(threadsafe_lookup_table const& other) = delete;
Value value_for(Key const& key, Value const& default_value = Value()) const { return get_bucket(key).value_for(key, default_value); }
void add_or_update_mapping(Key const& key, Value const& value) { get_bucket(key).add_or_update_mapping(key, value); }
void remove_mapping(Key const& key) { get_bucket(key).remove_mapping(key); }};
??
??再例如對于鏈表來說，倘若想讓鏈表支持迭代，而STL風格的迭代器的生命周期完全不受容器控制，可以以成員函數的形式提供迭代功能。為了減小鎖的粒度，干脆讓每個節點都有自己的互斥。
template?<typename?T>class threadsafe_list { struct node { std::mutex m; std::shared_ptr data; std::unique_ptr next;
????????node()?:?next()?{} node(T const& value) : data(std::make_shared(value)) {} };
node head;
public: threadsafe_list() {}
~threadsafe_list() { remove_if([](T const&) { return true; }); }
threadsafe_list(threadsafe_list const& other) = delete; threadsafe_list& operator=(threadsafe_list const& other) = delete;
void push_front(T const& value) { std::unique_ptr new_node(new node(value)); std::lock_guard<std::mutex> lk(head.m); new_node->next = std::move(head.next); head.next = std::move(new_node); }
template <typename Function> void for_each(Function f) { node* current = &head; std::unique_lock<std::mutex> lk(head.m); while (node* const next = current->next.get()) { std::unique_lock<std::mutex> next_lk(next->m); lk.unlock(); f(*next->data); current = next; lk = std::move(next_lk); }????}};
第7章 設計無鎖數據結構
? ? 非阻塞是指沒有使用互斥、條件變量、future進行同步。僅僅非阻塞往往并不足夠，例如第5章使用atomic_flag實現自旋鎖，非阻塞但效率并不高。
????無鎖代表如果多個線程共同操作同一份數據，那么有限步驟內其中某一線程能夠完成自己的操作。采用無鎖結構可以最大限度地實現并發并且提高代碼健壯性，避免鎖阻塞其他線程，或者持鎖線程拋出異常時其他線程無法繼續處理。
????以下是采用引用計數和寬松原子操作的無鎖棧容器的實現。書中這段代碼講解了28頁，有興趣的可以仔細讀讀原文。簡略地講，引用計數的作用是避免pop時重復delete；由于缺乏原子的共享指針，所以將引用計數分為內部與外部兩個，手動管理引用計數，每當指針被讀取外部計數器自增，讀取完成內部計數器自減。
template?<typename?T>class lock_free_stack {private: struct node; struct counted_node_ptr { int external_count; node* ptr; }; struct node { std::shared_ptr data; std::atomic<int> internal_count; counted_node_ptr next; node(T const& data_) : data(std::make_shared(data_)), internal_count(0) {} };???? std::atomic 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: ~lock_free_stack() { while (pop()) ; } 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)) ; } std::shared_ptr 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();????????????} if (head.compare_exchange_strong(old_head, ptr->next, std::memory_order_relaxed)) {???? std::shared_ptr 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_release) == -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; } } }};
? ? 與棧不同的是，對于隊列結構，push與pop訪問不同部分?
template?<typename?T>class lock_free_queue {private: struct node; struct counted_node_ptr { int external_count; node* ptr; }; std::atomic head; std::atomic tail;
struct node_counter { unsigned internal_count : 30; unsigned external_counters : 2; };
struct node { std::atomic data; std::atomic count; std::atomic next;
node() { node_counter new_count; new_count.internal_count = 0; new_count.external_counters = 2; count.store(new_count); next.ptr = nullptr; next.external_count = 0; }????? void release_ref() { node_counter old_counter = count.load(std::memory_order_relaxed); node_counter new_counter; do { new_counter = old_counter; --new_counter.internal_count; } while (!count.compare_exchange_strong(old_counter, new_counter, std::memory_order_acquire, std::memory_order_relaxed)); if (!new_counter.internal_count && !new_counter.external_counters) { delete this; } } }; static void free_external_counter(counted_node_ptr& old_node_ptr) { node* const ptr = old_node_ptr.ptr; int const count_increase = old_node_ptr.external_count - 2; node_counter old_counter = ptr->count.load(std::memory_order_relaxed); node_counter new_counter; do { new_counter = old_counter; --new_counter.external_counters; new_counter.internal_count += count_increase; } while (!ptr->count.compare_exchange_strong( old_counter, new_counter, std::memory_order_acquire, std::memory_order_relaxed)); if (!new_counter.internal_count && !new_counter.external_counters) { delete ptr; } }???? static void increase_external_count(std::atomic& counter, counted_node_ptr& old_counter) { counted_node_ptr new_counter; do { new_counter = old_counter; ++new_counter.external_count; } while (!counter.compare_exchange_strong(old_counter, new_counter, std::memory_order_acquire, std::memory_order_relaxed)); old_counter.external_count = new_counter.external_count; }
void set_new_tail(counted_node_ptr& old_tail, counted_node_ptr const& new_tail) { node* const current_tail_ptr = old_tail.ptr; while (!tail.compare_exchange_weak(old_tail, new_tail) && old_tail.ptr == current_tail_ptr) ; if (old_tail.ptr == current_tail_ptr) free_external_counter(old_tail); else current_tail_ptr->release_ref(); }
public: std::unique_ptr pop() { counted_node_ptr old_head = head.load(std::memory_order_relaxed); for (;;) { increase_external_count(head, old_head); node* const ptr = old_head.ptr; if (ptr == tail.load().ptr) { return std::unique_ptr(); } counted_node_ptr next = ptr->next.load(); if (head.compare_exchange_strong(old_head, next)) { T* const res = ptr->data.exchange(nullptr); free_external_counter(old_head); return std::unique_ptr(res); } ptr->release_ref(); } }
void push(T new_value) { std::unique_ptr new_data(new T(new_value)); counted_node_ptr new_next; new_next.ptr = new node; new_next.external_count = 1; counted_node_ptr old_tail = tail.load(); for (;;) { increase_external_count(tail, old_tail); T* old_data = nullptr; if (old_tail.ptr->data.compare_exchange_strong(old_data, new_data.get())) { counted_node_ptr old_next = {0}; if (!old_tail.ptr->next.compare_exchange_strong(old_next, new_next)) { delete new_next.ptr; new_next = old_next; } set_new_tail(old_tail, new_next); new_data.release(); break; } else {????????? counted_node_ptr old_next = {0};????????? if (old_tail.ptr->next.compare_exchange_strong(old_next, new_next)) { old_next = new_next;????????? new_next.ptr = new node; }????????? set_new_tail(old_tail, old_next); } } }};
????相信讀者已經了解了正確寫出無鎖代碼的困難與繁復。如果想自行設計，請注意以下原則：
????1、在原型設計中使用std::memory_order_seq_cst次序，便于分析和推理；
????2、使用無鎖的內存回收方案，例如上面的引用計數；
????3、防范ABA問題，即兩次讀取變量的值都相同，但其實變量已經被修改過多次，解決辦法是將變量與其計數器綁定；
????4、找出忙等循環，協助其他線程，例如兩線程同時壓入隊列的話某一線程就會忙等循環，可以像上面隊列中的實現一樣，多個線程同時push只有一個能成功，失敗的線程轉而協助成功線程。
你好，我是飛宇，本碩均于某中流985 CS就讀，先后于百度搜索以及字節跳動電商等部門擔任Linux C/C++后端研發工程師。
同時，我也是知乎博主@韓飛宇，日常分享C/C++、計算機學習經驗、工作體會，歡迎點擊此處查看我以前的學習筆記&經驗&分享的資源。
我組建了一些社群一起交流，群里有大牛也有小白，如果你有意可以一起進群交流。
歡迎你添加我的微信，我拉你進技術交流群。此外，我也會經常在微信上分享一些計算機學習經驗以及工作體驗，還有一些內推機會。
加個微信，打開另一扇窗
