async_read_some для эмуляции получения синхронного тайм-аута

Моя программа всегда использовала зависящий от платформы синхронный прием, который блокировал выполнение до истечения времени ожидания или получения события, например:

recv(buf, size, timeout);

Теперь я хочу заменить этот код на boost, чтобы сделать его кроссплатформенным. Я нашел решение, но я думаю, что оно довольно уродливо (по сравнению с вызовом одной функции). Я написал это:

void IPV4_HOST::recv_timer_handler(const boost::system::error_code & e)
{
    if (e.value() == boost::system::errc::success) {
        f_recv_timeout = true;
        asio_socket.cancel();
    }
}

void IPV4_HOST::recv_handler(const boost::system::error_code & , size_t bytes)
{
    recv_timer.cancel();
    bytes_received = bytes;
}

int IPV4_HOST::receive(void * buf, size_t buf_size, TIME::INTERVAL timeout)
{
    f_recv_timeout = false;
    bytes_received = 0;

    recv_timer.expires_from_now(timeout.get_boost_milli());
    recv_timer.async_wait(boost::bind(&IPV4_HOST::recv_timer_handler, this, boost::asio::placeholders::error));

    asio_socket.async_read_some(boost::asio::buffer(buf, buf_size),
                                boost::bind(&IPV4_HOST::recv_handler, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));

    io_service.run();

    if (f_recv_timeout)
        throw IO::TimeoutReceiveException();

    return bytes_received;
}

Не подскажете, правильно я делаю или нет? Есть ли более простой способ сделать это?


person Aleksei Petrenko    schedule 01.03.2013    source источник
comment
См. следующий ответ автора библиотеки: boost.2283326.n4.nabble.com/   -  person Igor R.    schedule 01.03.2013
comment
Это почти то же самое, что и я, но все равно спасибо   -  person Aleksei Petrenko    schedule 01.03.2013


Ответы (1)


Это в правильном направлении, но есть несколько тонких моментов, которые следует учитывать:

  • Если какая-либо другая работа будет опубликована в io_service, то IPV4_HOST::receive() начнет обрабатывать эту работу.
  • Когда io_service::run() возвращается в нормальных условиях, подразумевается, что io_service был остановлен. Последующие вызовы run(), run_one(), poll() или poll_one() немедленно возвращаются, если io_service не reset().
  • Обе асинхронные операции могут выполняться одновременно, что делает оба обработчика завершения готовыми к успешному выполнению. Это поведение подчеркивается в разделе примечаний deadline_timer::cancel(); однако все асинхронные операции демонстрируют такое поведение. В существующем коде это может привести к тому, что IO::TimeoutReceiveException будет выброшено, когда bytes_received больше нуля.

Одно решение для обработки деталей io_service, а также недетерминированного порядка выполнения с обработчиками завершения может выглядеть примерно так:

void IPV4_HOST::recv_timer_handler(const boost::system::error_code & e)
{
  timer_handled = true;
  if (!e) {
    f_recv_timeout = true;
    asio_socket.cancel();
  }
}

void IPV4_HOST::recv_handler(const boost::system::error_code &,
                             size_t bytes)
{
  recv_handled = true;
  recv_timer.cancel();
  bytes_received = bytes;
}

int IPV4_HOST::receive(void * buf, size_t buf_size, TIME::INTERVAL timeout)
{
  timer_handled  = false;
  recv_handled   = false;
  f_recv_timeout = false;
  bytes_received = 0;

  recv_timer.expires_from_now(timeout.get_boost_milli());
  recv_timer.async_wait(
    boost::bind(&IPV4_HOST::recv_timer_handler, this,
                boost::asio::placeholders::error));

  asio_socket.async_read_some(
    boost::asio::buffer(buf, buf_size),
    boost::bind(&IPV4_HOST::recv_handler, this, 
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred));

  // If a handler has not ran, then keep processing work on the io_service.
  // We need to consume both handlers so that old handlers are not in the
  // io_service the next time receive is called.
  while (!timer_handled || !recv_handled)
  {
    io_service.run_one();
  }

  // If the io_service has stopped (due to running out of work), then reset
  // it so that it can be run on next call to receive.
  if (io_service.stopped())
    io_service.reset();

  // If no bytes were received and the timeout occurred, then throw.  This
  // handles the case where both a timeout and receive occurred at the 
  // same time.
  if (!bytes_received && f_recv_timeout)
    throw IO::TimeoutReceiveException();

  return bytes_received;
}

Кроме того, поскольку вы пытаетесь добиться межплатформенного поведения, прочитайте примечания basic_stream_socket::cancel(). Есть некоторые особенности поведения платформы, о которых следует знать.

person Tanner Sansbury    schedule 01.03.2013
comment
Спасибо за этот ответ, я проверил код с вашими исправлениями, и теперь он работает очень стабильно. Единственное, чего я не понял, так это почему мы должны проверять io_service.stopped() перед сбросом? Почему мы не можем просто вызывать reset каждый раз? Это на тот случай, если мы используем io_service где-то еще для других обработчиков? - person Aleksei Petrenko; 01.03.2013
comment
Верный. reset() нельзя безопасно вызывать, если есть незавершенные вызовы функций run(), run_one(), poll() или poll_one(). Я не был уверен, используется ли io_service для других обработчиков, поэтому выбрал более безопасное решение. - person Tanner Sansbury; 01.03.2013