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

lost_templar

回复 19# lost_templar


    将近6万行，还是算了

gaojl0728

真费劲啊，还是我来吧， 我也整个gcc4.9试下。

lost_templar

回复 21# gaojl0728


    问题应该是如猫蜀黍所说的，没有等待 queue 中的工作全部做完就退出了，但是不知道怎么改正。
    编译器无所谓了，从 4.8 往上应该都可以

myworkstation

回复 1# lost_templar


    在 while ( ! all_done_flag )循环外把还在队列中的任务执行一下。

gaojl0728

回复 22# lost_templar


gcc4.9编译还挺麻烦的，依赖很多。

又看了一下你的代码， 但看逻辑的确会有这个问题， 虽然容易修正， 但是会有潜在的风险，simple_thread_pool开始析构以后，joiner会等待线程执行完成，这时候work_queue_mutex，work_queue对象必须得还在内存没有被析构才行。
所以你得保证joiner必须在work_queue_mutex，work_queue之前析构才能保证代码完全没问题， 你可以试一下是不是按照这个顺序析构的。
而且你得保证这几个变量定义的顺序在将来不会被修改。一旦修改风险就来了。
这也是个典型的反面教材啊。

while ( ! all_done_flag && ！work_queue.empty())
{
    function_type current_work;

    {
        std::lock_guard<std::mutex> lg{ work_queue_mutex };
        if ( work_queue.empty() )
        {
            current_work = [](){ std::this_thread::yield(); };
        }
        else
        {
            current_work = work_queue.front();
            work_queue.pop();
        }
    }

    current_work();
}

lost_templar

第一句
while ( ! all_done_flag && ！work_queue.empty())
这里后边加上个 !work_queue.empty() 不是很妥当，任务还没有进入队伍，这个线程就可能结束了。

lost_templar

回复 24# gaojl0728


    前面说得对，我准备把这个 thread_join_guard  去了，在后边手动 join

windoze

给你个阻塞队列的代码吧
https://github.com/windoze/Argos ... /concurrent_queue.h
你要高兴就把boost thread换成std thread，代码基本不用改

lost_templar

windoze 发表于 2014-04-30 18:54
给你个阻塞队列的代码吧
https://github.com/windoze/Argos ... on/concurrent_queue ...

正在学习…………

两个问题：
1） 如何能够保证 simple_thread_pool 退出的时候能够保证 work_queue 中的任务没有遗留？

2) 你的 concurrent_queue 作为一个通用容器，为何去掉了 copy-constructor 和 copy-assignment operator?

在我看来，很简单就可以用如下手法实现的：

1. struct concurrent_queue
   
2. {
   
3. //.....
   
4. private:
   
5.     concurrent_queue( concurrent_queue const& other, std::lock_guard<std::mutex> const& lc ) : /*do concrete copy here */ {}
   
6. public:
   
7.     concurrent_queue( concurrent_queue const& other ) : concurrent_queue( other, std::lock_guard<std::mutex>{other.the_mutex_} ) {}
   
8. };

复制代码

lost_templar

临时这么办了，虽然粗制滥造，但是总算解决了：
让主线程负责擦屁股。

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

复制代码