使用Boost線程和io_service創建線程池

Question

我已經看過周圍堆棧溢出，並且在這方面有幾個非常好的答案（我的代碼實際上是基於this answer here），但由於某種原因我收到了奇怪的行爲 - 該thread_func應該被調用ls1次，但它只在線程退出前的0到2次之間運行。看起來ioService.stop（）在完成之前切斷排隊的作業，但從我理解的情況來看，這不應該發生。這裏是相關的代碼片段：

boost::asio::io_service ioService; 
boost::asio::io_service::work work(ioService); 

boost::thread_group threadpool; 

for (unsigned t = 0; t < num_threads; t++) 
{ 
    threadpool.create_thread(boost::bind(&boost::asio::io_service::run, &ioService)); 
} 

//Iterate over the dimensions of the matrices 
for (unsigned i = 0; i < ls1; i++) 
{ 
    ioService.post(boost::bind(&thread_func,i, rs1, rs2, ls2, arr, left_arr, &result)); 
} 

ioService.stop(); 
threadpool.join_all(); 

任何幫助將不勝感激，謝謝！

Answer 1

io_service::stop()引起run()或run_one()所有調用，以儘快恢復。它不會刪除任何已排入io_service的未完成處理程序。當調用io_service::stop()時，threadpool中的線程將盡快返回，導致每個執行線程都完成。

由於io_service::post()將請求該io_service調用處理程序後立即返回，它具有不確定性至於多少張貼處理程序將通過線程threadpool停止io_service之前被調用。

如果您希望thread_func被調用ls1次，那麼一個簡單的辦法是使工作在threadpool服務之前加入到io_service它，然後應用程序讓io_service運行到完成重組的代碼。

boost::asio::io_service ioService; 

// Add work to ioService. 
for (unsigned i = 0; i < ls1; i++) 
{ 
    ioService.post(boost::bind(
     &thread_func,i, rs1, rs2, ls2, arr, left_arr, &result)); 
} 

// Now that the ioService has work, use a pool of threads to service it. 
boost::thread_group threadpool;  
for (unsigned t = 0; t < num_threads; t++) 
{ 
    threadpool.create_thread(boost::bind(
     &boost::asio::io_service::run, &ioService)); 
} 

// Once all work has been completed (thread_func invoked ls1 times), the 
// threads in the threadpool will be completed and can be joined. 
threadpool.join_all(); 

Answer 2

如果您預計thread_func被稱爲ls1次，那麼您應該等到實際調用ls1次後再停止io_service。正如所寫，stop()可能會在任何線程有機會甚至被調度之前調用。

有很多方法可以等待這種情況。例如，你可以使用一個條件變量：

#include <boost/asio.hpp> 
#include <boost/thread.hpp> 
unsigned num_threads = 10, ls1=11; 
int result = 0; 
boost::mutex m; 
boost::condition_variable cv; 
void thread_func(unsigned , int* result) { 
    /* do stuff */ 
    { 
     boost::lock_guard<boost::mutex> lk(m); 
     ++*result; 
    } 
    cv.notify_one(); 
} 
int main() 
{ 
    boost::asio::io_service ioService; 
    boost::asio::io_service::work work(ioService); 
    boost::thread_group threadpool; 
    for (unsigned t = 0; t < num_threads; t++) 
     threadpool.create_thread(boost::bind(&boost::asio::io_service::run, 
              &ioService)); 
    for (unsigned i = 0; i < ls1; i++) 
     ioService.post(boost::bind(&thread_func,i,&result)); 

    { 
     boost::unique_lock<boost::mutex> lk(m); 
     cv.wait(lk, []{return result == ls1; }); 
    } 
    ioService.stop(); 
    threadpool.join_all(); 
    std::cout << "result = " << result << '\n'; 
}