• C++ 并发编程实战 第六章 设计基于锁的并发数据结构

    目录

    6.1 并发设计的内涵

    6.2 基于锁的并发数据结构

    6.2.1 采用锁实现线程安全的栈容器

    6.2.2 采用锁和条件变量实现线程安全的队列容器

    6.2.3 采用精细粒度的锁和条件变量实现线程安全的队列容器

    6.3 设计更复杂的基于锁的并发数据结构

    6.3.1 采用锁编写线程安全的查找表

    6.3.2 采用多种锁编写线程安全的链表、

    参考：https://github.com/xiaoweiChen/CPP-Concurrency-In-Action-2ed-2019/blob/master/content/chapter5/5.0-chinese.md

    6.1 并发设计的内涵

    设计并发数据结构是为了让多线程并发访问，并且线程可对数据结构做相同或不同的操作。多线程环境下，无数据丢失和损毁，所有的数据需要维持原样，且无条件竞争的数据结构，称之为“线程安全”的数据结构。通常情况下，多个线程对数据结构进行并发操作是安全的，但不同操作需要单线程独立访问数据结构。当线程执行不同的操作时，对同一数据结构的并发操作是安全的，而多线程执行同样的操作时，可能会出现问题。

    实际的设计意义并不止上面提到的那样，而是要为线程提供并发访问数据结构的机会。本质上，在互斥量的保护下同一时间内只有一个线程可以获取锁。互斥量为了保护数据，会显式阻止线程对数据结构的并发访问。

    串行化(serialzation)则是线程轮流访问数据，对数据进行串行访问。因此，需要对数据结构仔细斟酌，确保能进行真正的并发。虽然，有些数据结构比其他结构的并发访问范围更大，但思路都是一样的：减少保护区域，减少序列化操作，提升并发访问的能力。

    6.2 基于锁的并发数据结构

    基于锁的并发数据结构需要确保访问线程持有锁的时间最短，对于只有一个互斥量的数据结构来说十分困难。需要锁之外的操作不能访问数据，保证不会产生条件竞争。使用多个互斥量保护数据结构不同的区域时，问题会更加明显。当操作需要获取多个互斥锁时，可能会产生死锁，所以使用多个互斥量时要格外小心。

    6.2.1 采用锁实现线程安全的栈容器

    代码6.1 线程安全栈的类定义

    1. #include 
    2.  
    3. struct empty_stack: std::exception
    4. {
    5.   const char* what() const throw();
    6. };
    7.  
    8. template<typename T>
    9. class threadsafe_stack
    10. {
    11. private:
    12.   std::stack data;
    13.   mutable std::mutex m;
    14. public:
    15.   threadsafe_stack(){}
    16.   threadsafe_stack(const threadsafe_stack& other)
    17.   {
    18.     std::lock_guard lock(other.m);
    19.     data=other.data;
    20.   }
    21.  
    22.   threadsafe_stack& operator=(const threadsafe_stack&) = delete;
    23.  
    24.   void push(T new_value)
    25.   {
    26.     std::lock_guard lock(m);
    27.     data.push(std::move(new_value));  // 1
    28.   }
    29.   std::shared_ptr pop()
    30.   {
    31.     std::lock_guard lock(m);
    32.     if(data.empty()) throw empty_stack();  // 2
    33.     std::shared_ptr const res(
    34.       std::make_shared(std::move(data.top())));  // 3
    35.     data.pop();  // 4
    36.     return res;
    37.   }
    38.   void pop(T& value)
    39.   {
    40.     std::lock_guard lock(m);
    41.     if(data.empty()) throw empty_stack();
    42.     value=std::move(data.top());  // 5
    43.     data.pop();  // 6
    44.   }
    45.   bool empty() const
    46.   {
    47.     std::lock_guard lock(m);
    48.     return data.empty();
    49.   }
    50. };

    6.2.2 采用锁和条件变量实现线程安全的队列容器

    代码6.2 使用条件变量实现的线程安全队列

    1. template<typename T>
    2. class threadsafe_queue
    3. {
    4. private:
    5.   mutable std::mutex mut;
    6.   std::queue data_queue;
    7.   std::condition_variable data_cond;
    8.  
    9. public:
    10.   threadsafe_queue()
    11.   {}
    12.  
    13.   void push(T data)
    14.   {
    15.     std::lock_guard lk(mut);
    16.     data_queue.push(std::move(data));
    17.     data_cond.notify_one();  // 1
    18.   }
    19.  
    20.   void wait_and_pop(T& value)  // 2
    21.   {
    22.     std::unique_lock lk(mut);
    23.     data_cond.wait(lk,[this]{return !data_queue.empty();});
    24.     value=std::move(data_queue.front());
    25.     data_queue.pop();
    26.   }
    27.  
    28.   std::shared_ptr wait_and_pop()  // 3
    29.   {
    30.     std::unique_lock lk(mut);
    31.     data_cond.wait(lk,[this]{return !data_queue.empty();});  // 4
    32.     std::shared_ptr res(
    33.       std::make_shared(std::move(data_queue.front())));
    34.     data_queue.pop();
    35.     return res;
    36.   }
    37.  
    38.   bool try_pop(T& value)
    39.   {
    40.     std::lock_guard lk(mut);
    41.     if(data_queue.empty())
    42.       return false;
    43.     value=std::move(data_queue.front());
    44.     data_queue.pop();
    45.     return true;
    46.   }
    47.  
    48.   std::shared_ptr try_pop()
    49.   {
    50.     std::lock_guard lk(mut);
    51.     if(data_queue.empty())
    52.       return std::shared_ptr();  // 5
    53.     std::shared_ptr res(
    54.       std::make_shared(std::move(data_queue.front())));
    55.     data_queue.pop();
    56.     return res;
    57.   }
    58.  
    59.   bool empty() const
    60.   {
    61.     std::lock_guard lk(mut);
    62.     return data_queue.empty();
    63.   }
    64. };

    异常安全会有一些变化，不止一个线程等待对队列进行推送操作时，只会有一个线程因data_cond.notify_one()而继续工作。但是，如果工作线程在wait_and_pop()中抛出一个异常，例如：构造新的std::shared_ptr<>对象④时抛出异常，那么其他线程则会永世长眠。这种情况不可以，所以调用函数需要改成data_cond.notify_all()，这个函数将唤醒所有的工作线程，不过当大多线程发现队列依旧是空时，又会耗费资源让线程重新进入睡眠。第二种替代方案，有异常抛出时，让wait_and_pop()函数调用notify_one()，从而让个另一个线程去索引存储的值。第三种替代方案，将std::shared_ptr<>的初始化过程移到push()中，并且存储std::shared_ptr<>实例，而不是直接使用数据值，将std::shared_ptr<>拷贝到内部std::queue<>中就不会抛出异常了，这样wait_and_pop()又是安全的了。下面的代码，就是根据第三种方案修改的。

    代码6.3 持有std::shared_ptr<>实例的线程安全队列

    1. template<typename T>
    2. class threadsafe_queue
    3. {
    4. private:
    5.   mutable std::mutex mut;
    6.   std::queue > data_queue;
    7.   std::condition_variable data_cond;
    8. public:
    9.   threadsafe_queue()
    10.   {}
    11.  
    12.   void wait_and_pop(T& value)
    13.   {
    14.     std::unique_lock lk(mut);
    15.     data_cond.wait(lk,[this]{return !data_queue.empty();});
    16.     value=std::move(*data_queue.front());  // 1
    17.     data_queue.pop();
    18.   }
    19.  
    20.   bool try_pop(T& value)
    21.   {
    22.     std::lock_guard lk(mut);
    23.     if(data_queue.empty())
    24.       return false;
    25.     value=std::move(*data_queue.front());  // 2
    26.     data_queue.pop();
    27.     return true;
    28.   }
    29.  
    30.   std::shared_ptr wait_and_pop()
    31.   {
    32.     std::unique_lock lk(mut);
    33.     data_cond.wait(lk,[this]{return !data_queue.empty();});
    34.     std::shared_ptr res=data_queue.front();  // 3
    35.     data_queue.pop();
    36.     return res;
    37.   }
    38.  
    39.   std::shared_ptr try_pop()
    40.   {
    41.     std::lock_guard lk(mut);
    42.     if(data_queue.empty())
    43.       return std::shared_ptr();
    44.     std::shared_ptr res=data_queue.front();  // 4
    45.     data_queue.pop();
    46.     return res;
    47.   }
    48.  
    49.   void push(T new_value)
    50.   {
    51.     std::shared_ptr data(
    52.     std::make_shared(std::move(new_value)));  // 5
    53.     std::lock_guard lk(mut);
    54.     data_queue.push(data);
    55.     data_cond.notify_one();
    56.   }
    57.  
    58.   bool empty() const
    59.   {
    60.     std::lock_guard lk(mut);
    61.     return data_queue.empty();
    62.   }
    63. };

    6.2.3 采用精细粒度的锁和条件变量实现线程安全的队列容器

    代码6.6 线程安全队列——细粒度锁版

    1. template<typename T>
    2. class threadsafe_queue
    3. {
    4. private:
    5.   struct node
    6.   {
    7.     std::shared_ptr data;
    8.     std::unique_ptr next;
    9.   };
    10.   std::mutex head_mutex;
    11.   std::unique_ptr head;
    12.   std::mutex tail_mutex;
    13.   node* tail;
    14.  
    15.   node* get_tail()
    16.   {
    17.     std::lock_guard tail_lock(tail_mutex);
    18.     return tail;
    19.   }
    20.  
    21.   std::unique_ptr pop_head()
    22.   {
    23.     std::lock_guard head_lock(head_mutex);
    24.     if(head.get()==get_tail())
    25.     {
    26.       return nullptr;
    27.     }
    28.     std::unique_ptr old_head=std::move(head);
    29.     head=std::move(old_head->next);
    30.     return old_head;
    31.   }
    32. public:
    33.   threadsafe_queue():
    34.   head(new node),tail(head.get())
    35.   {}
    36.   threadsafe_queue(const threadsafe_queue& other)=delete;
    37.   threadsafe_queue& operator=(const threadsafe_queue& other)=delete;
    38.  
    39.   std::shared_ptr try_pop()
    40.   {
    41.      std::unique_ptr old_head=pop_head();
    42.      return old_head?old_head->data:std::shared_ptr();
    43.   }
    44.  
    45.   void push(T new_value)
    46.   {
    47.     std::shared_ptr new_data(
    48.       std::make_shared(std::move(new_value)));
    49.     std::unique_ptr p(new node);
    50.     node* const new_tail=p.get();
    51.     std::lock_guard tail_lock(tail_mutex);
    52.     tail->data=new_data;
    53.     tail->next=std::move(p);
    54.     tail=new_tail;
    55.   }
    56. };

    6.3 设计更复杂的基于锁的并发数据结构

    6.3.1 采用锁编写线程安全的查找表

    代码6.11 线程安全的查询表

    1. template<typename Key,typename Value,typename Hash=std::hash >
    2. class threadsafe_lookup_table
    3. {
    4. private:
    5.   class bucket_type
    6.   {
    7.   private:
    8.     typedef std::pair bucket_value;
    9.     typedef std::list bucket_data;
    10.     typedef typename bucket_data::iterator bucket_iterator;
    11.  
    12.     bucket_data data;
    13.     mutable std::shared_mutex mutex;  // 1
    14.  
    15.     bucket_iterator find_entry_for(Key const& key) const  // 2
    16.     {
    17.       return std::find_if(data.begin(),data.end(),
    18.       [&](bucket_value const& item)
    19.       {return item.first==key;});
    20.     }
    21.   public:
    22.     Value value_for(Key const& key,Value const& default_value) const
    23.     {
    24.       std::shared_lock lock(mutex);  // 3
    25.       bucket_iterator const found_entry=find_entry_for(key);
    26.       return (found_entry==data.end())?
    27.         default_value:found_entry->second;
    28.     }
    29.  
    30.     void add_or_update_mapping(Key const& key,Value const& value)
    31.     {
    32.       std::unique_lock lock(mutex);  // 4
    33.       bucket_iterator const found_entry=find_entry_for(key);
    34.       if(found_entry==data.end())
    35.       {
    36.         data.push_back(bucket_value(key,value));
    37.       }
    38.       else
    39.       {
    40.         found_entry->second=value;
    41.       }
    42.     }
    43.  
    44.     void remove_mapping(Key const& key)
    45.     {
    46.       std::unique_lock lock(mutex);  // 5
    47.       bucket_iterator const found_entry=find_entry_for(key);
    48.       if(found_entry!=data.end())
    49.       {
    50.         data.erase(found_entry);
    51.       }
    52.     }
    53.   };
    54.  
    55.   std::vector > buckets;  // 6
    56.   Hash hasher;
    57.  
    58.   bucket_type& get_bucket(Key const& key) const  // 7
    59.   {
    60.     std::size_t const bucket_index=hasher(key)%buckets.size();
    61.     return *buckets[bucket_index];
    62.   }
    63.  
    64. public:
    65.   typedef Key key_type;
    66.   typedef Value mapped_type;
    67.  
    68.   typedef Hash hash_type;
    69.   threadsafe_lookup_table(
    70.     unsigned num_buckets=19,Hash const& hasher_=Hash()):
    71.     buckets(num_buckets),hasher(hasher_)
    72.   {
    73.     for(unsigned i=0;i
    74.     {
    75.       buckets[i].reset(new bucket_type);
    76.     }
    77.   }
    78.  
    79.   threadsafe_lookup_table(threadsafe_lookup_table const& other)=delete;
    80.   threadsafe_lookup_table& operator=(
    81.     threadsafe_lookup_table const& other)=delete;
    82.  
    83.   Value value_for(Key const& key,
    84.                   Value const& default_value=Value()) const
    85.   {
    86.     return get_bucket(key).value_for(key,default_value);  // 8
    87.   }
    88.  
    89.   void add_or_update_mapping(Key const& key,Value const& value)
    90.   {
    91.     get_bucket(key).add_or_update_mapping(key,value);  // 9
    92.   }
    93.  
    94.   void remove_mapping(Key const& key)
    95.   {
    96.     get_bucket(key).remove_mapping(key);  // 10
    97.   }
    98. };

    6.3.2 采用多种锁编写线程安全的链表、

    提供了这些操作，链表才能为通用容器，这将帮助我们添加更多功能，比如：指定位置上插入元素，不过这对于查询表来说就没有必要了，所以算是给读者们留的一个作业吧。

    代码6.13 线程安全链表——支持迭代器（包含我写的指定位置插入---丐版）

    1. #ifndef THREADSAFE_LIST_H_
    2. #define THREADSAFE_LIST_H
    3. #include
    4. #include
    5. template<typename T>
    6. class threadsafe_list
    7. {
    8.   struct node  // 1
    9.   {
    10.     std::mutex m;
    11.     std::shared_ptr data;
    12.     std::unique_ptr next;
    13.     node():  // 2
    14.       next()
    15.     {}
    16.  
    17.     node(T const& value):  // 3
    18.       data(std::make_shared(value))
    19.     {}
    20.   };
    21.  
    22.   node head;
    23.  
    24. public:
    25.   threadsafe_list()
    26.   {}
    27.  
    28.   ~threadsafe_list()
    29.   {
    30.     remove_if([](node const&){return true;});
    31.   }
    32.  
    33.   threadsafe_list(threadsafe_list const& other)=delete;
    34.   threadsafe_list& operator=(threadsafe_list const& other)=delete;
    35.  
    36.   void push_front(T const& value)
    37.   {
    38.     std::unique_ptr new_node(new node(value));  // 4
    39.     std::lock_guard lk(head.m);
    40.     new_node->next=std::move(head.next);  // 5
    41.     head.next=std::move(new_node);  // 6
    42.   }
    43.  
    44.   template<typename Function>
    45.   void for_each(Function f)  // 7
    46.   {
    47.     node* current=&head;
    48.     std::unique_lock lk(head.m);  // 8
    49.     while(node* const next=current->next.get())  // 9
    50.     {
    51.       std::unique_lock next_lk(next->m);  // 10
    52.       lk.unlock();  // 11
    53.       f(*next->data);  // 12
    54.       current=next;
    55.       lk=std::move(next_lk);  // 13
    56.     }
    57.   }
    58.  
    59.   template<typename Predicate>
    60.   std::shared_ptr find_first_if(Predicate p)  // 14
    61.   {
    62.     node* current=&head;
    63.     std::unique_lock lk(head.m);
    64.     while(node* const next=current->next.get())
    65.     {
    66.       std::unique_lock next_lk(next->m);
    67.       lk.unlock();
    68.       if(p(*next->data))  // 15
    69.       {
    70.          return next->data;  // 16
    71.       }
    72.       current=next;
    73.       lk=std::move(next_lk);
    74.     }
    75.     return std::shared_ptr();
    76.   }
    77.  
    78.   template<typename Predicate>
    79.   void remove_if(Predicate p)  // 17
    80.   {
    81.     node* current=&head;
    82.     std::unique_lock lk(head.m);
    83.     while(node* const next=current->next.get())
    84.     {
    85.       std::unique_lock next_lk(next->m);
    86.       if(p(*next->data))  // 18
    87.       {
    88.         std::unique_ptr old_next=std::move(current->next);
    89.         current->next=std::move(next->next);
    90.         next_lk.unlock();
    91.       }  // 20
    92.       else
    93.       {
    94.         lk.unlock();  // 21
    95.         current=next;
    96.         lk=std::move(next_lk);
    97.       }
    98.     }
    99.   }
    100.  
    101.   void insert_in_position(T const& value,const int& position){
    102.   	std::unique_ptr new_node(new node(value));
    103. 	node* current=&head;
    104. 	node* next = current->next.get();
    105.  
    106. 	std::unique_lock lk(head.m);
    107. 	for(int i = 0; i
    108. 		std::unique_lock next_lk(next->m);
    109. 		lk.unlock();
    110. 		current = next;
    111. 		lk = std::move(next_lk);
    112. 		next = current->next.get();
    113. 	}
    114. 	lk.unlock();
    115. 	
    116. 	std::unique_lock cur_lk(current->m);
    117. 	new_node->next = std::move(current->next);
    118. 	current->next = std::move(new_node);
    119.   }
    120. };
    121.  
    122. #endif

  • 相关阅读:
    PyTorch之张量的相关操作大全 -＞（个人学习记录笔记）
    38.CSS文本动画效果
    微信小程序组件化
    k8s运维问题整理
    @ConditionalOnBean系列注解使用误区
    python中protobuf和json互相转换应用
    spring
    【信号与系统】信号频谱和测量之汉明窗
    预测杭州五一黄金周的旅游出行人数
    策略引擎Kyverno
  • 原文地址：https://blog.csdn.net/qq_52758467/article/details/133383048