Как подождать обработчик asio?

Question

У меня есть объект, который работает вокруг boost::asio::io_service, который имеет некоторые свойства. Нечто подобное:

class Foo 
{ 
    private: 

    // Not an int in my real code, but it doesn't really matter. 
    int m_bar; 
    boost::asio::io_service& m_io_service; 
    boost::asio::strand m_bar_strand; 
}; 

m_bar должен использоваться только из обработчика, который вызывается через нити m_bar_strand. Это позволяет мне не блокироваться внутри этих обработчиков.

Чтобы установить m_bar свойство из вне потока, который проходит io_service::run() я написал asynchronous_setter, например так:

class Foo 
{ 
    public: 
    void async_get_bar(function<void (int)> handler) 
    { 
     m_bar_strand.post(bind(&Foo::do_get_bar, this, handler)); 
    } 

    void async_set_bar(int value, function<void()> handler) 
    { 
     m_bar_strand.post(bind(&Foo::do_set_bar, this, value, handler)); 
    } 

    private: 

    void do_get_bar(function<void (int)> handler) 
    { 
     // This is only called from within the m_bar_strand, so we are safe. 

     // Run the handler to notify the caller. 
     handler(m_bar); 
    } 

    void do_set_bar(int value, function<void()> handler) 
    { 
     // This is only called from within the m_bar_strand, so we are safe. 
     m_bar = value; 

     // Run the handler to notify the caller. 
     handler(); 
    } 

    int m_bar; 
    boost::asio::io_service& m_io_service; 
    boost::asio::strand m_bar_strand; 
}; 

Это прекрасно работает, но теперь я хотел бы написать синхронную версию set_bar, что устанавливает значение и возвращает только тогда, когда набор был эффективным. Он должен по-прежнему гарантировать, что эффективный набор будет установлен в пределах m_bar_strand. В идеале, что-то повторное.

Я могу представить решения с семафорами, которые были бы изменены изнутри обработчика, но все, что я придумал, кажется хакерским и действительно не изящным. Есть ли что-то в Boost/Boost Asio, которое позволяет такое?

Как вы примените этот метод?

Answer 1

Если вам нужно синхронно ждать на значение, которое будет установлено, то Boost.Thread-х futures может обеспечить элегантное решение:

Библиотека фьючерсными предоставляет средства обработок синхронных будущих значений, то ли те значения генерируются другим потоком или одним потоком в ответ на внешние стимулы или по требованию.

Короче говоря, создается boost::promise и позволяет устанавливать на нем значение. Затем значение может быть получено с помощью связанного с ним boost::future.Вот простой пример:

boost::promise<int> promise; 
boost::unique_future<int> future = promise.get_future(); 

// start asynchronous operation that will invoke future.set_value(42) 
... 

assert(future.get() == 42); // blocks until future has been set. 

Два других заметных преимуществ такого подхода:

• future является частью C++ 11.
• Исключения могут быть переданы только future через promise::set_exception(), поддерживая элегантный способ предоставления исключений или ошибок вызывающему абоненту.

---

Вот полный пример, основанный на исходном коде:

#include <boost/asio.hpp> 
#include <boost/bind.hpp> 
#include <boost/thread.hpp> 

class Foo 
{ 
public: 

    Foo(boost::asio::io_service& io_service) 
    : m_io_service(io_service), 
     m_bar_strand(io_service) 
    {} 

public: 

    void async_get_bar(boost::function<void(int)> handler) 
    { 
    m_bar_strand.post(bind(&Foo::do_get_bar, this, handler)); 
    } 

    void async_set_bar(int value, boost::function<void()> handler) 
    { 
    m_bar_strand.post(bind(&Foo::do_set_bar, this, value, handler)); 
    } 

    int bar() 
    { 
    typedef boost::promise<int> promise_type; 
    promise_type promise; 

    // Pass the handler to async operation that will set the promise. 
    void (promise_type::*setter)(const int&) = &promise_type::set_value; 
    async_get_bar(boost::bind(setter, &promise, _1)); 

    // Synchronously wait for promise to be fulfilled. 
    return promise.get_future().get(); 
    } 

    void bar(int value) 
    { 
    typedef boost::promise<void> promise_type; 
    promise_type promise; 

    // Pass the handler to async operation that will set the promise. 
    async_set_bar(value, boost::bind(&promise_type::set_value, &promise)); 

    // Synchronously wait for the future to finish. 
    promise.get_future().wait(); 
    } 

private: 

    void do_get_bar(boost::function<void(int)> handler) 
    { 
    // This is only called from within the m_bar_strand, so we are safe. 

    // Run the handler to notify the caller. 
    handler(m_bar); 
    } 

    void do_set_bar(int value, boost::function<void()> handler) 
    { 
    // This is only called from within the m_bar_strand, so we are safe. 
    m_bar = value; 

    // Run the handler to notify the caller. 
    handler(); 
    } 

    int m_bar; 
    boost::asio::io_service& m_io_service; 
    boost::asio::strand m_bar_strand; 
}; 

int main() 
{ 
    boost::asio::io_service io_service; 
    boost::asio::io_service::work work(io_service); 
    boost::thread t(
     boost::bind(&boost::asio::io_service::run, boost::ref(io_service))); 

    Foo foo(io_service); 
    foo.bar(21); 
    std::cout << "foo.bar is " << foo.bar() << std::endl; 
    foo.bar(2 * foo.bar()); 
    std::cout << "foo.bar is " << foo.bar() << std::endl; 

    io_service.stop(); 
    t.join(); 
} 

, который обеспечивает следующий вывод:

foo.bar is 21 
foo.bar is 42 

Answer 2

Вы можете использовать трубку для уведомления синхронного метода, когда значение установлено в async_set_bar(). Предупреждение, приведенный ниже код является мозг скомпилированные и, вероятно, есть ошибки, но он должен получить через точку

#include <boost/asio.hpp> 

#include <iostream> 
#include <thread>                              

class Foo                                
{ 
public:                                 
    Foo(boost::asio::io_service& io_service) :                       
     _bar(0), 
     _io_service(io_service),                          
     _strand(_io_service),                           
     _readPipe(_io_service), 
     _writePipe(_io_service) 
    { 
     boost::asio::local::connect_pair(_readPipe, _writePipe); 
    } 

    void set_async(int v) { 
     _strand.post([=] 
      { 
       _bar = v; 
       std::cout << "sending " << _bar << std::endl; 
       _writePipe.send(boost::asio::buffer(&_bar, sizeof(_bar))); 
      } 
      ); 
    } 

    void set_sync(int v) { 
     this->set_async(v); 
     int value; 
     _readPipe.receive(boost::asio::buffer(&value, sizeof(value))); 
     std::cout << "set value to " << value << std::endl; 
    } 


private: 
    int _bar; 
    boost::asio::io_service& _io_service; 
    boost::asio::io_service::strand _strand; 
    boost::asio::local::stream_protocol::socket _readPipe; 
    boost::asio::local::stream_protocol::socket _writePipe; 
}; 

int 
main() 
{ 
    boost::asio::io_service io_service; 
    boost::asio::io_service::work w(io_service); 
    std::thread t([&]{ io_service.run(); }); 
    Foo f(io_service); 
    f.set_sync(20); 
    io_service.stop(); 
    t.join(); 
} 

, если вы не можете использовать C++ 11 лямбды, замените их boost::bind и еще несколько методов обработчика завершения.

Answer 3

Это то, что я придумал:

class synchronizer_base 
{ 
    protected: 
     synchronizer_base() : 
      m_has_result(false), 
      m_lock(m_mutex) 
     { 
     } 

     void wait() 
     { 
      while (!m_has_result) 
      { 
       m_condition.wait(m_lock); 
      } 
     } 

     void notify_result() 
     { 
      m_has_result = true; 
      m_condition.notify_all(); 
     } 

    private: 

     boost::atomic<bool> m_has_result; 
     boost::mutex m_mutex; 
     boost::unique_lock<boost::mutex> m_lock; 
     boost::condition_variable m_condition; 
}; 

template <typename ResultType = void> 
class synchronizer : public synchronizer_base 
{ 
    public: 

     void operator()(const ResultType& result) 
     { 
      m_result = result; 

      notify_result(); 
     } 

     ResultType wait_result() 
     { 
      wait(); 

      return m_result; 
     } 

    private: 

     ResultType m_result; 
}; 

template <> 
class synchronizer<void> : public synchronizer_base 
{ 
    public: 

     void operator()() 
     { 
      notify_result(); 
     } 

     void wait_result() 
     { 
      wait(); 
     } 
}; 

И я могу использовать его, таким образом:

class Foo 
{ 
    public: 
    void async_get_bar(function<void (int)> handler) 
    { 
     m_bar_strand.post(bind(&Foo::do_get_bar, this, value, handler)); 
    } 

    void async_set_bar(int value, function<void()> handler) 
    { 
     m_bar_strand.post(bind(&Foo::do_set_bar, this, value, handler)); 
    } 

    int get_bar() 
    { 
     synchronizer<int> sync; 

     async_get_bar(boost::ref(sync)); 

     return sync.wait_result(); 
    } 

    void set_bar(int value) 
    { 
     synchronizer<void> sync; 

     async_set_bar(value, boost::ref(sync)); 

     sync.wait_result(); 
    } 
}; 

boost::ref необходимо, потому что случаи synchronizer не являются копируемыми. Этого можно избежать, обернув synchronizer в какой-нибудь другой контейнерный класс, но я в порядке с этим решением, как есть.

Примечание: Do NOT вызывают такие «синхронизированные» функции изнутри обработчика, или это может быть просто тупик!