Boost:: asio-сервер для асинхронного проектирования

Ответ 1

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

• Включить отслеживание обработчика для Boost.Asio 1.47+. Просто определите BOOST_ASIO_ENABLE_HANDLER_TRACKING и Boost.Asio будет записывать вывод отладки, включая временные метки, в стандартный поток ошибок. Эти временные метки могут использоваться, чтобы помочь отфильтровать задержки, введенные кодом приложения (parseHeader(), parsePacket() и т.д.).
• Убедитесь, что порядок байтов обрабатывается правильно. Например, если протокол определяет поле заголовка size как два байта в сетевом порядке, а сервер обрабатывает это поле как необработанное, то после получения сообщения с размером тела 10:
  • Больничный компьютер выберет async_read, чтобы прочитать 10 байты. Операция чтения должна завершиться быстро, поскольку в гнезде уже имеется тело байта 10, доступное для чтения.
  • Маленькая машина вызовет async_read чтение 2560 байтов. Операция чтения, вероятно, останется невыполненной, так как больше байт пытается читать, чем предполагалось.
• Используйте инструменты трассировки, такие как strace, ltrace и т.д.
• Измените Boost.Asio, добавив отметки времени во время вызова. Boost.Asio поставляется в виде библиотеки только для заголовков. Таким образом, пользователи могут изменять его, чтобы обеспечить столько же многословий, сколько требуется. Хотя это не самый простой или простой подход, добавление инструкции печати с метками времени во время вызова может помочь обеспечить видимость времени.
• Попробуйте дублировать поведение в кратком, простом, самодостаточном примере. Начните с простейших примеров, чтобы определить, является ли задержка систамтической. Затем итеративно расширьте пример, чтобы он стал ближе к реальному коду с каждой итерацией.

Вот простой пример, с которого я начал:

#include <iostream>

#include <boost/array.hpp>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/make_shared.hpp>
#include <boost/shared_ptr.hpp>

class tcp_server
  : public boost::enable_shared_from_this< tcp_server >
{
private:

  enum 
  {
     header_size = 4,
     data_size   = 10,
     buffer_size = 1024,
     max_stamp   = 50
  };

  typedef boost::asio::ip::tcp tcp;

public:

  typedef boost::array< boost::posix_time::ptime, max_stamp > time_stamps;

public:

  tcp_server( boost::asio::io_service& service,
              unsigned short port )
    : strand_( service ),
      acceptor_( service, tcp::endpoint( tcp::v4(), port ) ),
      socket_( service ),
      index_( 0 )
  {}

  /// @brief Returns collection of timestamps.
  time_stamps& stamps()
  {
    return stamps_;
  }

  /// @brief Start the server.
  void start()
  {
    acceptor_.async_accept( 
      socket_,
      boost::bind( &tcp_server::handle_accept, this,
                   boost::asio::placeholders::error ) );
  }

private:

  /// @brief Accept connection.
  void handle_accept( const boost::system::error_code& error ) 
  {
    if ( error )
    {  
      std::cout << error.message() << std::endl;
      return;
    }

    read_header();
  }

  /// @brief Read header.
  void read_header()
  {
    boost::asio::async_read(
      socket_,
      boost::asio::buffer( buffer_, header_size ),
      boost::bind( &tcp_server::handle_read_header, this,
                   boost::asio::placeholders::error,
                   boost::asio::placeholders::bytes_transferred ) );
  }

  /// @brief Handle reading header.
  void
  handle_read_header( const boost::system::error_code& error,
                      std::size_t bytes_transferred )
  {
    if ( error )
    {  
      std::cout << error.message() << std::endl;
      return;
    }

    // If no more stamps can be recorded, then stop the async-chain so
    // that io_service::run can return.
    if ( !record_stamp() ) return;

    // Read data.
    boost::asio::async_read(
      socket_,
      boost::asio::buffer( buffer_, data_size ),
      boost::bind( &tcp_server::handle_read_data, this,
                   boost::asio::placeholders::error,
                   boost::asio::placeholders::bytes_transferred ) );

  }

  /// @brief Handle reading data.
  void handle_read_data( const boost::system::error_code& error,
                         std::size_t bytes_transferred )
  {
    if ( error )
    {  
      std::cout << error.message() << std::endl;
      return;
    }

    // If no more stamps can be recorded, then stop the async-chain so
    // that io_service::run can return.
    if ( !record_stamp() ) return;

    // Start reading header again.
    read_header();
  }

  /// @brief Record time stamp.
  bool record_stamp()
  {
    stamps_[ index_++ ] = boost::posix_time::microsec_clock::local_time();

    return index_ < max_stamp;
  }

private:
  boost::asio::io_service::strand strand_;
  tcp::acceptor acceptor_;
  tcp::socket socket_;
  boost::array< char, buffer_size > buffer_;
  time_stamps stamps_;
  unsigned int index_;
};


int main()
{
  boost::asio::io_service service;

  // Create and start the server.
  boost::shared_ptr< tcp_server > server =
    boost::make_shared< tcp_server >( boost::ref(service ), 33333 );  
  server->start();

  // Run.  This will exit once enough time stamps have been sampled.
  service.run();

  // Iterate through the stamps.
  tcp_server::time_stamps& stamps = server->stamps();
  typedef tcp_server::time_stamps::iterator stamp_iterator;
  using boost::posix_time::time_duration;
  for ( stamp_iterator iterator = stamps.begin() + 1,
                       end      = stamps.end();
        iterator != end;
        ++iterator )
  {
     // Obtain the delta between the current stamp and the previous.
     time_duration delta = *iterator - *(iterator - 1);
     std::cout << "Delta: " << delta.total_milliseconds() << " ms"
               << std::endl;
  }
  // Calculate the total delta.
  time_duration delta = *stamps.rbegin() - *stamps.begin();
  std::cout <<    "Total" 
            << "\n  Start: " << *stamps.begin()
            << "\n  End:   " << *stamps.rbegin()
            << "\n  Delta: " << delta.total_milliseconds() << " ms"
            << std::endl;
}

Несколько заметок о реализации:

• Существует только один поток (основной) и одна асинхронная цепочка read_header- > handle_read_header- > handle_read_data​​i > . Это должно свести к минимуму время, в течение которого готовый к работе обработчик проводит ожидание доступного потока.
• Чтобы сфокусироваться на boost::asio::async_read, шум сводится к минимуму:
  • Использование предварительно выделенного буфера.
  • Не использовать shared_from_this() или strand::wrap.
  • Запись временных меток и выполнение обработки после сбора.

Я собрал на CentOS 5.4 с использованием gcc 4.4.0 и Boost 1.50. Чтобы управлять данными, я решил отправить 1000 байтов, используя netcat:

$ ./a.out > output &
[1] 18623
$ echo "$(for i in {0..1000}; do echo -n "0"; done)" | nc 127.0.0.1 33333
[1]+  Done                    ./a.out >output
$ tail output
Delta: 0 ms
Delta: 0 ms
Delta: 0 ms
Delta: 0 ms
Delta: 0 ms
Delta: 0 ms
Total
  Start: 2012-Sep-10 21:22:45.585780
  End:   2012-Sep-10 21:22:45.586716
  Delta: 0 ms

Наблюдая за задержкой, я расширил этот пример, изменив вызовы boost::asio::async_read, заменив this на shared_from_this() и завернув ReadHandlers в strand_.wrap(). Я запустил обновленный пример и все еще не заметил никакой задержки. К сожалению, это насколько я могу получить на основе кода, размещенного в вопросе.

Рассмотрим расширение на примере, добавляя кусок из реальной реализации с каждой итерацией. Например:

• Начните с использования типа переменной msg для управления буфером.
• Затем отправьте действительные данные и введите функции parseHeader() и parsePacket.
• Наконец, введите lib::GET_SERVER_TIME() print.

Если примерный код как можно ближе к реальному коду, и с boost::asio::async_read не наблюдается задержки, тогда ReadHandler может быть готов к запуску в реальном коде, но они ждут синхронизация (прядь) или ресурс (поток), что приводит к задержке:

• Если задержка является результатом синхронизации со строкой, рассмотрите предложение Robin, прочитав больший блок данных, чтобы потенциально уменьшить количество прочитанных запросов, сообщение.
• Если задержка является результатом ожидания потока, тогда рассмотрите возможность вызова дополнительного потока io_service::run().

Ответ 2

Одна вещь, которая делает Boost.Asio awesome, использует функцию async в полной мере. Опираясь на определенное количество байтов, прочитанных в одной партии, возможно, отбросив часть того, что уже было прочитано, на самом деле вы не должны делать.

Вместо этого посмотрите на пример для веб-сервера, особенно это: http://www.boost.org/doc/libs/1_51_0/doc/html/boost_asio/example/http/server/connection.cpp

Трибулер повышения используется для: a) завершения запроса, если все данные доступны в одной партии; b) вырезать его, если он доступен, но недействителен; c) просто читать больше, когда io_service выбирает, если запрос был неполным, Объект соединения совместно используется обработчиком через общий указатель.

Почему это превосходит большинство других методов? Вы можете сэкономить время между чтениями, которые уже обрабатывают запрос. Это, к сожалению, не соблюдается в этом примере, но вы идеализируете обработчик, чтобы он мог работать с уже доступными данными, в то время как остальные добавляются в буфер. Единственный раз, когда он блокируется, когда данные неполны.

Надеюсь, что это поможет, не может пролить свет на то, почему существует 3 мс задержка между чтением.