在不同的線程上發佈工作時如何將信號連接到boost :: asio :: io_service？

Question

我正在嘗試使用boost :: lockfree隊列來管理任務。這些任務檢索數據並在工作線程上處理。一旦檢索到數據，就應該將數據發送到主線程。工作線程在應用程序的開始時產生，並且只是保持輪詢隊列。我是Boost :: Asio的新手，但從我的研究來看，它似乎是在線程之間發送信號的最佳機制。

我看幾個例子，尤其是：

• Confused when boost::asio::io_service run method blocks/unblocks
• boost asio post not working , io_service::run exits right after post

這裏是我的代碼：

#include "stdafx.h" 
#include <thread> 

#include <boost/asio.hpp> 
#include <boost/bind.hpp> 
#include <boost/lockfree/spsc_queue.hpp> 
#include <boost/optional.hpp> 
#include <boost/thread.hpp> 
#include <boost/signals2.hpp> 

typedef boost::signals2::signal<void(int)> signal_type; 

class Task 
{ 
public: 
    Task(int handle) : _handle(handle) {}; 
    ~Task() {}; 

    virtual void Execute() 
    { 
     int result = _handle * 2; 
    } 

private: 
    int _handle; 
}; 


class Manager 
{ 
public: 
    Manager() 
    { 
     _mainService = std::make_shared<boost::asio::io_service>(); 
     _workerService = std::make_shared<boost::asio::io_service>(); 
     _work = std::make_shared<boost::asio::io_service::work>(*_workerService); 

     _threadStarted = false; 
     Start(); 
    }; 

    ~Manager() {}; 

    void WorkerMain() 
    { 
     _workerService->poll(); 
    } 

    void Start() 
    { 
     if (_threadStarted) return; 

     _workerThread = std::thread(&Manager::WorkerMain, this); 
     _threadStarted = true; 
    } 

    void Stop() 
    { 
     if (_threadStarted == false) return; 

     _mainService->stop(); 
     _workerThread.join(); 
     _mainService.reset(); 
    } 

    void OnSignalFetchCompleted(int value) 
    { 
     int asdf = 0; //do stuff with data on main thread 
    } 

    void ProcessData(signal_type& signal) 
    { 
     int i = 0; 

     do 
     { 
     _queue.consume_one([&](std::shared_ptr<Task> task) 
     { 
      task->Execute(); 
      //get data from task; send out signal with data 
     }); 

     i++; 
     } while (i < 3); 
    } 

    void QueueData(int handle) 
    { 
     _signalFetchCompleted.connect(boost::bind(&Manager::OnSignalFetchCompleted, this, _1)); 
     _workerService->post(boost::bind(&Manager::ProcessData, boost::ref(_signalFetchCompleted))); //!!does not compile 

     std::shared_ptr<Task> task = std::make_shared<Task>(handle); 
     _queue.push(task); 
    } 

private: 
    boost::lockfree::spsc_queue<std::shared_ptr<Task>, boost::lockfree::capacity<1024>> _queue; 
    std::thread _workerThread; 
    bool _threadStarted; 

    std::shared_ptr<boost::asio::io_service> _mainService; 
    std::shared_ptr<boost::asio::io_service> _workerService; 
    std::shared_ptr<boost::asio::io_service::work> _work; 

    signal_type _signalFetchCompleted; 
}; 


int _tmain(int argc, _TCHAR* argv[]) 
{ 
    std::shared_ptr<Manager> mgr = std::make_shared<Manager>(); 
    mgr->QueueData(5); 
    mgr->QueueData(10); 

    mgr->Stop(); 

    return 0; 
} 

我得到一個編譯錯誤的_workerService->郵政行，我一直無法reso lve：

1>C:\Boost\boost/bind/mem_fn.hpp(333): error C2784: 'T *boost::get_pointer(const boost::scoped_ptr<T> &)' : could not deduce template argument for 'const boost::scoped_ptr<T> &' from 'const signal_type' 
1>   C:\Boost\boost/smart_ptr/scoped_ptr.hpp(150) : see declaration of 'boost::get_pointer' 
1>   C:\Boost\boost/bind/mem_fn.hpp(352) : see reference to function template instantiation 'R (__cdecl &boost::_mfi::dm<R,Manager>::call<const U>(U &,const void *) const)' being compiled 
1>   with 
1>   [ 
1>    R=void (signal_type &) 
1> ,   U=signal_type 
1>   ] 
1>   C:\Boost\boost/bind/mem_fn.hpp(352) : see reference to function template instantiation 'R (__cdecl &boost::_mfi::dm<R,Manager>::call<const U>(U &,const void *) const)' being compiled 
1>   with 
1>   [ 
1>    R=void (signal_type &) 
1> ,   U=signal_type 
1>   ] 
1>   C:\Boost\boost/bind/bind.hpp(243) : see reference to function template instantiation 'R (__cdecl &boost::_mfi::dm<R,Manager>::operator()<T>(const U &) const)' being compiled 
1>   with 
1>   [ 
1>    R=void (signal_type &) 
1> ,   T=signal_type 
1> ,   U=signal_type 
1>   ] 

任何幫助解決這個編譯錯誤或一般意見在這種方法將不勝感激。謝謝。

Answer 1

鑑於最新信息，問題出在您的boost::bind。您正試圖調用一個沒有對象的成員函數來調用它：您正試圖調用ProcessData，但是您尚未告知綁定了您希望調用它的對象。你需要給它一個Manager調用它：

_workerService->post(boost::bind(&Manager::ProcessData, this, boost::ref(_signalFetchCompleted)));

這將調用ProcessData上this和參考傳遞給_signalFetchCompleted

Answer 2

編譯器錯誤似乎是談論你構建一個boost::asio::io_service::work對象和要傳遞它不正確的參數：

error C2664: 'boost::asio::io_service::work::work(const boost::asio::io_service::work &)' : cannot convert argument 1 from 'std::shared_ptr<boost::asio::io_service>' to 'boost::asio::io_service &' 

boost::asio::io_service::work有一個構造函數，需要一個boost::asio::io_service&和拷貝構造函數;然而，你傳遞一個std::shared_ptr<boost::asio::io_service>：

_work = std::make_shared<boost::asio::io_service::work>(_workerService); 

這裏，_workerService是std::shared_ptr<boost::asio::io_service>，但你需要一個boost::asio::io_service&。嘗試改爲：

_work = std::make_shared<boost::asio::io_service::work>(*_workerService); 

Answer 3

我認爲boost :: asio不是您的任務的最佳解決方案。你讀過關於條件變量嗎？他們更簡單，可以用來實現你的目標。