Как подключить сигнал к boost :: asio :: io_service при публикации работы в разных потоках?

Question

Я пытаюсь использовать boost :: lockfree queue для управления задачами. Эти задачи извлекают данные и обрабатываются в рабочем потоке. После получения данных сигнал должен быть отправлен в основной поток с данными. Рабочий поток создается в начале приложения и просто продолжает опрос очереди. Я новичок в 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-> Post, которую я не смог Решение:

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 - не лучшее решение для вашей задачи. Вы читали об условных переменных? Они намного проще и могут быть использованы для достижения вашей цели.