谁来来帮我看看这个简单的线程池(c++11)

lost_templar

1. #include <cstddef>
   
2. #include <functional>
   
3. #include <queue>
   
4. #include <thread>
   
5. #include <vector>
   
6. #include <mutex>
   
7. #include <atomic>
   
8. #include <algorithm>
   
9. 
   
10.     namespace simple_thread_pool_private_sdnjasfdlkj
    
11.     {
    
12.         template< typename Container_Type >
    
13.         struct thread_join_guard
    
14.         {
    
15.             Container_Type & container;
    
16.             explicit thread_join_guard( Container_Type& container_ ) : container( container_ ) {}
    
17.             ~thread_join_guard()
    
18.             {
    
19.                 for ( auto& the_thread : container )
    
20.                     if ( the_thread.joinable() )
    
21.                         the_thread.join();
    
22.             }
    
23.         };
    
24.     }//namespace simple_thread_pool_private_sdnjasfdlkj
    
25. 
    
26.     struct simple_thread_pool
    
27.     {
    
28.         typedef std::function<void()>                                           function_type;
    
29. 
    
30.         std::atomic_bool                                                        all_done_flag;
    
31.         std::queue<function_type>                                               work_queue;
    
32.         std::mutex                                                              work_queue_mutex;
    
33.         std::vector<std::thread>                                                thread_array;
    
34. 
    
35.         simple_thread_pool_private_sdnjasfdlkj::thread_join_guard<std::vector<std::thread>> joiner;
    
36. 
    
37.         simple_thread_pool() : all_done_flag{ false }, joiner{ thread_array }
    
38.         {
    
39.             std::size_t const max_threads = std::thread::hardware_concurrency();
    
40. 
    
41.             for ( std::size_t index = 0; index != max_threads; ++index )
    
42.                 thread_array.push_back( std::thread{ &simple_thread_pool::work_loader, this } );
    
43.     }
    
44. 
    
45.         ~simple_thread_pool()
    
46.         {
    
47.             all_done_flag = true;
    
48.         }
    
49. 
    
50.         template< typename Function >
    
51.         void commit_work( Function f )
    
52.         {
    
53.             std::lock_guard<std::mutex> lk{ work_queue_mutex };
    
54.             work_queue.emplace( function_type{f} );
    
55.         }
    
56. 
    
57.         void work_loader()
    
58.         {
    
59.             while ( ! all_done_flag )
    
60.             {
    
61.                 function_type current_work;
    
62. 
    
63.                 {
    
64.                     std::lock_guard<std::mutex> lg{ work_queue_mutex };
    
65.                     if ( work_queue.empty() )
    
66.                     {
    
67.                         current_work = [](){ std::this_thread::yield(); };
    
68.                     }
    
69.                     else
    
70.                     {
    
71.                         current_work = work_queue.front();
    
72.                         work_queue.pop();
    
73.                     }
    
74.                 }
    
75. 
    
76.                 current_work();
    
77.             }
    
78.         }
    
79. 
    
80.         simple_thread_pool( simple_thread_pool const& ) = delete;
    
81.         simple_thread_pool& operator = ( simple_thread_pool const& ) = delete;
    
82.         simple_thread_pool( simple_thread_pool && ) = delete;
    
83.         simple_thread_pool& operator = ( simple_thread_pool && ) = delete;
    
84. 
    
85.     };//struct simple_thead_pool
    

复制代码

随手写了准备这么用的：

1. 
   
2.         std::vector<result_type> accumulator_array( ... );
   
3.         {
   
4.             simple_thread_pool stp;
   
5.             for ( ... )
   
6.             {               
   
7.                 stp.commit_work( [&accumulator_array,..]{...;} );
   
8.             }        
   
9.         }
   
10. 
    
11.        auto const ans = *std::max_element( accumulator_array.begin(), accumulator_array.end() );

复制代码

发现老是在退出前还没有执行完 commit_work 注册的函数～～

lost_templar

@windoze
@蔡万钊
@cokeboL
@OwnWaterloo
@pprpg

lost_templar

@starwing83

windoze

sdnjasfdlkj，猛一看还以为是乱码……

lost_templar

@AD8018
@walleeee

windoze

昨晚没仔细看代码，今天一看顿觉I服了U啊……

std::queue拿来做同步真得大丈夫？线程在那里忙循环不怕烧CPU？看到“!all_done_flag”就结束你也不管queue里还有没有剩下的东西？
……
太多了，不吐槽了，lz还是先补补课吧

lost_templar

回复 6# windoze

很少写基于 cpu 的并行程序，求指点。


   

gaojl0728

回复 6# windoze


    你耐心点看完他的代码， 跟本不会有你说的这两个问题，这两个问题都考虑到了，只不过CPU忙等待解决的方法不太好罢了。

gaojl0728

回复 1# lost_templar


整体而言程序还可以，有两个问题需要指出。
1. CPU空循环的问题， std::this_thread::yield()虽然可以释放当前的CPU时间， 等并不能阻止内核还会调度这个线程，下次跑进来一样会浪费CPU的时间，更好的解决办法使用条件变量解决。
队列位空的时空等待条件变量， 然后在commit_work里面**条件变量
2. 你的线程永远不会退出, simple_thread_pool析构时会等待所有线程退出， 然后会设置all_done_flag = true， 但是线程只有在all_done_flag为true才会退出，看到了没，这里就锁死了

lost_templar

gaojl0728 发表于 2014-04-30 15:28
回复 1# lost_templar


回复 1# lost_templar


整体而言程序还可以，有两个问题需要指出。
1. CPU空循环的问题， std::this_thread::yield()虽然可以释放当前的CPU时间， 等并不能阻止内核还会调度这个线程，下次跑进来一样会浪费CPU的时间，更好的解决办法使用条件变量解决。
队列位空的时空等待条件变量， 然后在commit_work里面**条件变量
2. 你的线程永远不会退出, simple_thread_pool析构时会等待所有线程退出， 然后会设置all_done_flag = true， 但是线程只有在all_done_flag为true才会退出，看到了没，这里就锁死了

待我想想

对于2，我认为是先把 all_done_flag 设置为 true，然后才析构 thread_join_guard 等待所有线程退出的，难道是我理解错了？我写段代码测试一下。