一、管程
1.1 简介
管程可以视为一个线程安全的数据结构，其内部提供了互斥与同步操作，向外提供访问共享数据的专用接口(接口被称为管程的过程)，通过管程提供的接口即可达成共享数据的保护与线程间同步。
使用管程，可以简化线程间互斥、同步的编码复杂度(否则需自己控制互斥、同步机制，并保证正确)，可以集中分散的互斥、同步操作代码，更容易验证、查错，也更可读(反之，信号量的PV操作可能分散到各个地方，验证、阅读相对麻烦)。
1.2 管程的特点
- 共享数据仅能被管程的过程访问
- 线程通过调用管程的过程进入管程
- 任何时候仅能有一个线程在管程中执行，其他阻塞直到可用
- 管程内共享数据不可用时
- 需要共享数据的线程将阻塞并释放管程
- 其他线程可进入管程构造数据可用条件，并通知阻塞线程
- 管程内共享数据可用时， 被阻塞线程将在合适时间重新进入管程
1.3 管程分类
管程中仅能有一个线程在其中执行，根据发起通知时，被唤醒线程(T1)执行，还是唤醒线程(T2)继续执行，可将管程分为三种：
- Mesa管程(Lampson/Redell)：T2继续执行直到退出，T1进入入口队列，后面与其他线程公平竞争(由于不是立刻执行，所以当T1执行时，条件可能已经不满足，因此需循环检测条件是否仍旧达成)。
- Hoare管程：T2立即阻塞并释放管程，T1马上执行，退出后T2恢复执行(多两次线程切换)。
- Brinch Hanson管程：T2的通知操作仅允许在退出时发送通知，通知发送后T2结束，T1开始执行。
1.4 管程的实现
根据前述管程特点，管程应该是一个对象，内部封装了资源，该对象实现了互斥及与阻塞、唤醒机制
使用锁，搭配条件变量来实现互斥及与阻塞、唤醒机制
下面简单介绍条件变量，后面的管程代码有一个小优化，涉及对条件变量实现的理解
1.4.1 条件变量
条件变量是同步原语，其内部有一个队列，用于存放被wait阻塞的线程，当另一个线程发起通知时，如果队列不为空，队头线程将被唤醒，否则，什么也不做。条件变量也可以一次性唤醒队列中的全部阻塞线程。
1.4.1.1 条件变量的一种实现
条件变量里有一个存储阻塞线程的队列。由于阻塞线程和唤醒通知线程都需要访问这同一队列，所以还有一个用于保护队列的锁，锁粒度不大，并且不需要线程切换，应为自旋锁。
wait函数会将当前线程入队，并原子的进行锁释放与当前线程阻塞，阻塞直到另一线程通知才解除，解除后重新获取锁。锁释放与当前线程阻塞必须是原子的，否则，别的线程发出的唤醒通知，可能发生在当前线程阻塞之前，这会造成唤醒丢失。
signal函数出队一个阻塞线程并唤醒他，当队列为空时，不做任何事情。
broadcast函数将逐个唤醒队列中的所有阻塞线程。
1.4.2 管程实现代码
下面实现了一个管程，同时符合mesa与hanson管程的定义，并且做了一些优化，取消掉了不必要的通知操作。
1 #ifndef __MONITOR_H_
2 #define __MONITOR_H_
3
4 #include <list>
5 #include <mutex>
6 #include <utility>
7 #include <condition_variable>
8
9 template<typename T>
10 class Monitor{
11 public:
12 Monitor(): m_iMaxCount(100), m_bStop(false){
13 }
14 Monitor(int iMaxCount) : m_iMaxCount(iMaxCount), m_bStop(false){
15 }
16 ~Monitor() = default;
17
18 void Enqueue(const T& data){
19 Append(data);
20 }
21
22 void Enqueue(T&& data){
23 Append(std::forward<T>(data)); //转发data的原属性，此处转发data的右值引用
24 }
25
26 void Dequeue(T& data){
27 std::unique_lock<std::mutex> lk(m_mMutex);
28 m_cvNotEmpty.wait(lk, [this](){return m_bStop || !IsEmpty(); });
29 if(m_bStop){
30 return;
31 }
32 bool bNeedNotify = IsFull();
33 data = m_listData.front();
34 m_listData.pop_front();
35 lk.unlock();
36 if(bNeedNotify){
37 m_cvNotFull.notify_one();
38 }
39 }
40
41 void Stop(){
42 {
43 std::lock_guard<std::mutex> lk(m_mMutex);
44 m_bStop = true;
45 }
46 m_cvNotEmpty.notify_all();
47 m_cvNotFull.notify_all();
48 }
49
50 private:
51 template<typename U>
52 void Append(U&& data){ //实现通用引用
53 std::unique_lock<std::mutex> lk(m_mMutex);
54 m_cvNotFull.wait(lk, [this](){return m_bStop || !IsFull(); });
55 if(m_bStop){
56 return;
57 }
58 bool bNeedNotify = IsEmpty();
59 m_listData.emplace_back(std::forward<U>(data)); //再次转发
60 lk.unlock();
61 if(bNeedNotify){
62 m_cvNotEmpty.notify_one();
63 }
64 }
65
66 bool IsFull(){
67 return static_cast<int>(m_listData.size()) == m_iMaxCount;
68 }
69
70 bool IsEmpty(){
71 return 0 == static_cast<int>(m_listData.size());
72 }
73
74 Monitor(const Monitor& rhs) = delete;
75 Monitor(Monitor&& rhs) = delete;
76 Monitor& operator=(const Monitor& rhs) = delete;
77 Monitor& operator=(Monitor&& rhs) = delete;
78
79 private:
80 int m_iMaxCount;
81 bool m_bStop;
82 std::mutex m_mMutex;
83 std::list<T> m_listData;
84 std::condition_variable m_cvNotEmpty;
85 std::condition_variable m_cvNotFull;
86 };
87
88 #endif //!__MONITOR_H_
- 代码中条件变量使用了带可调用对象作为参数的wait，它的实现是这样的
1 template<typename _Predicate> 2 void 3 wait(unique_lock<mutex>& __lock, _Predicate __p) 4 { 5 while (!__p()) 6 wait(__lock); 7 }
可调用对象返回true则什么都不做，否则，重复wait到满足条件，这可以避免虚假唤醒，同时也是Mesa管程的行为
- 进行队列不满、不空通知前，检查队列原状态 ，通过状态判断是否进行notify
- 在由满到不满，空到不空这两种状态变化下，才进行通知
- 根据上面的条件变量的signal函数实现可知，通知时会使用到锁，虽然锁粒度很小，并且是自旋锁，但是也存在较大的性能消耗，通过一个简单的判断，避免了通知，也就避免了锁
- 模板方法Append实现了通用引用，使得Append同时支持引用左值与右值，配合std::forward可将参数原属性(左值引用或右值引用)转发给另一个函数当参数
- 实现通用引用必须含有&&，并且需要发生类型推导，所以Append必须是不同的模板参数U，而不能和类模板Monitor一样为T(T的类型在模板实例化时固定，将不能发生类型推导)
- 一切变量都是左值(变量本身分配有内存空间，可以对其取地址)，所以在调用实现了通用引用函数时，应该std::forward一下，否则通用引用绑定的是变量本身(左值)，即得到左值引用
- std::forward靠引用折叠实现对原属性的转发：含左值引用折叠后为左值引用，否则为右值引用
1 template<typename _Tp>
2 constexpr _Tp&&
3 forward(typename std::remove_reference<_Tp>::type& __t) noexcept
4 { return static_cast<_Tp&&>(__t); }
5
6 template<typename _Tp>
7 constexpr _Tp&&
8 forward(typename std::remove_reference<_Tp>::type&& __t) noexcept
9 {
10 static_assert(!std::is_lvalue_reference<_Tp>::value, "template argument"
11 " substituting _Tp is an lvalue reference type");
12 return static_cast<_Tp&&>(__t);
13 }
- 参数_Tp若为type &(左值引用)
则static_cast<_Tp&&>为static_cast<type& &&>，折叠后为static_cast<type&>， 所以转发返回值仍为为左值引用。
- 参数_Tp若为type &&(右值引用)
则static_cast<_Tp&&>为static_cast<type&& &&>，折叠后为static_cast<type&&>， 所以转发返回仍为右值引用 。
- Enqueue最好不要以通用引用方式实现，除非调用者明确知道使用std::forward(不用的代价为丢失右值语义)
二、同步队列
2.1 简介
管程其实也是一种同步队列，现在
- 在管程的基础上添加Try类操作，Try类操作操作时，若条件无法满足，则立刻返回false，不阻塞
- 添加获取元素个数的方法
得到同步队列如下
2.2 代码
1 #ifndef __SyncQueue_H_ 2 #define __SyncQueue_H_ 3 4 #include <list> 5 #include <mutex> 6 #include <utility> 7 #include <condition_variable> 8 9 template<typename T> 10 class SyncQueue{ 11 public: 12 SyncQueue() : m_iMaxCount(100), m_bStop(false){ 13 } 14 SyncQueue(int iMaxCount) : m_iMaxCount(iMaxCount), m_bStop(false){ 15 } 16 ~SyncQueue() = default; 17 18 void Enqueue(const T& data){ 19 Append(data); 20 } 21 22 void Enqueue(T&& data){ 23 Append(std::forward<T>(data)); //转发data的原属性，此处转发data的右值引用 24 } 25 26 bool TryEnqueue(const T& data){ 27 return TryAppend(data); 28 } 29 30 bool TryEnqueue(T&& data){ 31 return TryAppend(std::forward<T>(data)); 32 } 33 34 void Dequeue(T& data){ 35 std::unique_lock<std::mutex> lk(m_mMutex); 36 m_cvNotEmpty.wait(lk, [this](){return m_bStop || !IsEmpty(); }); 37 if(m_bStop){ 38 return; 39 } 40 bool bNeedNotify = IsFull(); 41 data = m_listData.front(); 42 m_listData.pop_front(); 43 lk.unlock(); 44 if(bNeedNotify){ 45 m_cvNotFull.notify_one(); 46 } 47 } 48 49 bool TryDequeue(T& data){ 50 std::unique_lock<std::mutex> lk(m_mMutex); 51 if(m_bStop || IsEmpty()){ 52 return false; 53 } 54 bool bNeedNotify = IsFull(); 55 data = m_listData.front(); 56 m_listData.pop_front(); 57 lk.unlock(); 58 if(bNeedNotify){ 59 m_cvNotFull.notify_one(); 60 } 61 return true; 62 } 63 64 void Stop(){ 65 { 66 std::lock_guard<std::mutex> lk(m_mMutex); 67 m_bStop = true; 68 } 69 m_cvNotEmpty.notify_all(); 70 m_cvNotFull.notify_all(); 71 } 72 73 int Size(){ 74 std::lock_guard<std::mutex> lk(m_mMutex); 75 return static_cast<int>(m_listData.size()); 76 } 77 78 private: 79 template<typename U> 80 bool TryAppend(U&& data){ //实现通用引用 81 std::unique_lock<std::mutex> lk(m_mMutex); 82 if(m_bStop || IsFull()){ 83 return false; 84 } 85 bool bNeedNotify = IsEmpty(); 86 m_listData.emplace_back(std::forward<U>(data)); //再次转发 87 lk.unlock(); 88 if(bNeedNotify){ 89 m_cvNotEmpty.notify_one(); 90 } 91 return true; 92 } 93 94 template<typename U> 95 void Append(U&& data){ //实现通用引用 96 std::unique_lock<std::mutex> lk(m_mMutex); 97 m_cvNotFull.wait(lk, [this](){return m_bStop || !IsFull(); }); 98 if(m_bStop){ 99 return; 100 } 101 bool bNeedNotify = IsEmpty(); 102 m_listData.emplace_back(std::forward<U>(data)); //再次转发 103 lk.unlock(); 104 if(bNeedNotify){ 105 m_cvNotEmpty.notify_one(); 106 } 107 } 108 109 bool IsFull(){ 110 return static_cast<int>(m_listData.size()) == m_iMaxCount; 111 } 112 113 bool IsEmpty(){ 114 return 0 == static_cast<int>(m_listData.size()); 115 } 116 117 SyncQueue(const SyncQueue& rhs) = delete; 118 SyncQueue(SyncQueue&& rhs) = delete; 119 SyncQueue& operator=(const SyncQueue& rhs) = delete; 120 SyncQueue& operator=(SyncQueue&& rhs) = delete; 121 122 private: 123 int m_iMaxCount; 124 bool m_bStop; 125 std::mutex m_mMutex; 126 std::list<T> m_listData; 127 std::condition_variable m_cvNotEmpty; 128 std::condition_variable m_cvNotFull; 129 }; 130 131 #endif //!__SyncQueue_H_
三、转载于
https://www.cnblogs.com/Keeping-Fit/p/15064039.html#wiz-toc-6-203609834