gr的多线程同步控制boost::barrier同步点(主线程控制所有子线...

东方耀

gr的多线程同步控制boost::barrier同步点(主线程控制所有子线程的同步)




CMakeLists.txt的配置:

1. find_package(Boost)
   
2. #  Found Boost: /usr/local/include (found version "1.69.0")
   
3. target_link_libraries(main -lboost_system -lboost_thread -lpthread)

复制代码

1. #include <iostream>
   
2. #include <list>
   
3. #include <string>
   
4. #include <unistd.h>
   
5. 
   
6. #include <boost/shared_ptr.hpp>
   
7. #include <boost/make_shared.hpp>
   
8. #include <boost/function.hpp>
   
9. #include <boost/utility.hpp>
   
10. 
    
11. #include <boost/thread/barrier.hpp>
    
12. #include <boost/thread/condition_variable.hpp>
    
13. #include <boost/thread/locks.hpp>
    
14. #include <boost/thread/mutex.hpp>
    
15. #include <boost/thread/thread.hpp>
    
16. #include <boost/thread/shared_mutex.hpp>
    
17. 
    
18. typedef boost::thread thread;
    
19. typedef boost::mutex mutex;
    
20. typedef boost::unique_lock<boost::mutex> scoped_lock;
    
21. typedef boost::condition_variable condition_variable;
    
22. typedef boost::barrier barrier;
    
23. // boost::barrier是多线程的同步点，构造函数会要求传递一个数字，表示需要同步多少个线程
    
24. typedef boost::shared_ptr<barrier> barrier_sptr;
    
25. 
    
26. class thread_group : public boost::noncopyable
    
27. {
    
28. public:
    
29.     thread_group(){};
    
30.     ~thread_group()
    
31.     {
    
32.         for (std::list<thread *>::iterator it = m_threads.begin();
    
33.              it != m_threads.end();
    
34.              ++it)
    
35.         {
    
36.             delete (*it);
    
37.         }
    
38.     };
    
39. 
    
40.     thread *create_thread(const boost::function0<void> &threadfunc)
    
41.     {
    
42.         // 是不是这里只能用new的 堆空间的地址
    
43.         std::unique_ptr<thread> thrd(new thread(threadfunc));
    
44.         // thrd是智能指针, thrd.get()是裸指针,
    
45.         add_thread(thrd.get());
    
46.         // 没明白为什么要 release()
    
47.         // 如果使用智能指针而没有调用Release () ，它会自动释放，不会资源泄漏
    
48.         // release只是切断了联系, release返回的指针通常用来初始化另一个智能指针，或给智能指针赋值
    
49.         return thrd.release();
    
50.     };
    
51.     void add_thread(thread *thrd)
    
52.     {
    
53.         // 为什么要锁: 同时来创建线程? add_thread
    
54.         boost::lock_guard<boost::shared_mutex> guard(m_mutex);
    
55. 
    
56.         // For now we'll simply ignore requests to add a thread object
    
57.         // multiple times. Should we consider this an error and either
    
58.         // throw or return an error value?
    
59.         std::list<thread *>::iterator it =
    
60.             std::find(m_threads.begin(), m_threads.end(), thrd);
    
61.         BOOST_ASSERT(it == m_threads.end());
    
62.         if (it == m_threads.end())
    
63.             m_threads.push_back(thrd);
    
64.     };
    
65.     void remove_thread(thread *thrd)
    
66.     {
    
67.         boost::lock_guard<boost::shared_mutex> guard(m_mutex);
    
68. 
    
69.         // For now we'll simply ignore requests to remove a thread
    
70.         // object that's not in the group. Should we consider this an
    
71.         // error and either throw or return an error value?
    
72.         std::list<thread *>::iterator it =
    
73.             std::find(m_threads.begin(), m_threads.end(), thrd);
    
74.         BOOST_ASSERT(it != m_threads.end());
    
75.         if (it != m_threads.end())
    
76.             m_threads.erase(it);
    
77.     };
    
78.     void join_all()
    
79.     {
    
80.         boost::shared_lock<boost::shared_mutex> guard(m_mutex);
    
81.         for (std::list<thread *>::iterator it = m_threads.begin();
    
82.              it != m_threads.end();
    
83.              ++it)
    
84.         {
    
85.             (*it)->join();
    
86.         }
    
87.     };
    
88.     void interrupt_all()
    
89.     {
    
90.         boost::shared_lock<boost::shared_mutex> guard(m_mutex);
    
91.         for (std::list<thread *>::iterator it = m_threads.begin(),
    
92.                                                   end = m_threads.end();
    
93.              it != end;
    
94.              ++it)
    
95.         {
    
96.             // interrupt 中断
    
97.             (*it)->interrupt();
    
98.         }
    
99.     };
    
100.     size_t size() const
     
101.     {
     
102.         boost::shared_lock<boost::shared_mutex> guard(m_mutex);
     
103.         return m_threads.size();
     
104.     };
     
105. 
     
106. private:
     
107.     std::list<thread *> m_threads;
     
108.     mutable boost::shared_mutex m_mutex;
     
109. };
     
110. 
     
111. 
     
112. 
     
113. template <class F>
     
114. class thread_body_wrapper
     
115. {
     
116. private:
     
117.     F d_f;
     
118.     std::string d_name;
     
119. 
     
120. public:
     
121.     explicit thread_body_wrapper(F f, const std::string& name = "") : d_f(f), d_name(name)
     
122.     {
     
123.         printf("第二步:thread_body_wrapper的构造, name=%s\n", d_name.c_str());
     
124.     }
     
125. 
     
126.     void operator()()
     
127.     {
     
128. 
     
129.         // d_f();
     
130. 
     
131.         printf("第三步:thread_body_wrapper的()操作符运算, name=%s\n", d_name.c_str());
     
132. 
     
133.         try {
     
134.             d_f();
     
135.         } catch (boost::thread_interrupted const&) {
     
136.             std::cerr << "thread[" << d_name << "]: " << "中断了:interrupted" << std::endl;
     
137.         } catch (std::exception const& e) {
     
138.             std::cerr << "thread[" << d_name << "]: " << e.what() << std::endl;
     
139.         } catch (...) {
     
140.             std::cerr << "thread[" << d_name << "]: "
     
141.                       << "caught unrecognized exception\n";
     
142.         }
     
143.     }
     
144. };
     
145. 
     
146. 
     
147. 
     
148. class tpb_thread_body
     
149. {
     
150. 
     
151. public:
     
152.     tpb_thread_body(std::string name,
     
153.                     barrier_sptr start_sync){
     
154. 
     
155.                         printf("第五步:tpb_thread_body的构造(里面有死循环,子线程开始等待wait()了),name=%s\n", name.c_str());
     
156.                         start_sync->wait();  // 所有的线程都到这 才一起走 同步的核心
     
157. 
     
158.                         while(1){
     
159.                             // std::cout << "运行:" + name << std::endl;
     
160.                             printf("线程体运行:%s\n", name.c_str());
     
161.                             sleep(1);
     
162.                             // std::this_thread::sleep_for(chrono::milliseconds(1000));
     
163.                         }
     
164.                     };
     
165.     ~tpb_thread_body(){};
     
166. };
     
167. 
     
168. 
     
169. 
     
170. 
     
171. class tpb_container
     
172. {
     
173.     std::string d_name;
     
174.     barrier_sptr d_start_sync;
     
175. 
     
176. public:
     
177.     tpb_container(std::string name,
     
178.                   barrier_sptr start_sync)
     
179.         : d_name(name), d_start_sync(start_sync)
     
180.     {
     
181.         printf("第一步:tpb_container的构造,name=%s\n", d_name.c_str());
     
182.     }
     
183. 
     
184.     void operator()()
     
185.     {
     
186.         printf("第四步:tpb_container的()操作符运算,name=%s\n", d_name.c_str());
     
187.         tpb_thread_body body(d_name, d_start_sync);
     
188.     }
     
189. };
     
190. 
     
191. 
     
192. 
     
193. int main()
     
194. {
     
195.     // gr的多线程同步控制boost::barrier同步点(主线程控制所有子线程的同步)
     
196.     int thread_num = 3;
     
197.     std::string thread_name;
     
198. 
     
199.     // 为什么要 子线程数+1 是主线程吗? 是的
     
200.     barrier_sptr start_sync =
     
201.         boost::make_shared<barrier>(thread_num + 1);
     
202. 
     
203.     thread_group d_threads;  // 线程池
     
204. 
     
205.     for (int i = 0; i < thread_num; ++i)
     
206.     {
     
207.         thread_name = "thread_name=" + std::to_string(i);
     
208.         d_threads.create_thread(thread_body_wrapper<tpb_container>(
     
209.             tpb_container("tpb_container:"+thread_name, start_sync), "thread_body_wrapper:"+thread_name));
     
210.     }
     
211.     printf("主线程开始等待了wait()\n");
     
212.     start_sync->wait();   // 主线程的 只有到这 所有子线程才会都开始跑起来
     
213. 
     
214.     // d_threads.join_all();  // 所有子线程 阻塞主线程
     
215. 
     
216.     sleep(3);
     
217. 
     
218.     d_threads.interrupt_all();
     
219.     printf("主线程指令: 所有的子线程中断了, 没起作用?\n");
     
220. 
     
221.     sleep(5);
     
222. 
     
223.     return 0;
     
224. }

复制代码

东方耀

boost::barrier这个类，非常简单的实现，简单说就是：计数等待