Смущает, когда boost:: asio:: io_service запускает метод блокировки/разблокировки

Будучи полным новичком в Boost.Asio, меня путают с io_service::run(). Я был бы признателен, если бы кто-нибудь мог объяснить мне, когда этот метод блокирует/разблокирует. В документах указано:

Функциональные блоки run() блокируются до тех пор, пока все работы не будут завершены, и больше обработчиков не будет отправлено или пока не будет остановлен io_service.

Несколько потоков могут вызывать функцию run() для настройки пула потоков, из которых io_service может выполнять обработчики. Все потоки, ожидающие в пуле, эквивалентны, а io_service может выбрать любой из них для вызова обработчика.

Нормальный выход из функции run() означает, что объект io_service остановлен (функция stopped() возвращает true). Последующие вызовы run(), run_one(), poll() или poll_one() будут немедленно возвращены, если не был предыдущий вызов reset().

Что означает следующее выражение?

[...] больше обработчиков не будет отправлено [...]

При попытке понять поведение io_service::run() я столкнулся с этим примером (пример 3a). Внутри него я замечаю, что io_service->run() блокирует и ждет рабочих заказов.

// WorkerThread invines io_service->run()
void WorkerThread(boost::shared_ptr<boost::asio::io_service> io_service);
void CalculateFib(size_t);

boost::shared_ptr<boost::asio::io_service> io_service(
    new boost::asio::io_service);
boost::shared_ptr<boost::asio::io_service::work> work(
   new boost::asio::io_service::work(*io_service));

// ...

boost::thread_group worker_threads;
for(int x = 0; x < 2; ++x)
{
  worker_threads.create_thread(boost::bind(&WorkerThread, io_service));
}

io_service->post( boost::bind(CalculateFib, 3));
io_service->post( boost::bind(CalculateFib, 4));
io_service->post( boost::bind(CalculateFib, 5));

work.reset();
worker_threads.join_all();

Однако в следующем коде, над которым я работал, клиент соединяется с использованием TCP/IP и блоков метода выполнения, пока данные не будут асинхронно приняты.

typedef boost::asio::ip::tcp tcp;
boost::shared_ptr<boost::asio::io_service> io_service(
    new boost::asio::io_service);
boost::shared_ptr<tcp::socket> socket(new tcp::socket(*io_service));

// Connect to 127.0.0.1:9100.
tcp::resolver resolver(*io_service);
tcp::resolver::query query("127.0.0.1", 
                           boost::lexical_cast< std::string >(9100));
tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
socket->connect(endpoint_iterator->endpoint());

// Just blocks here until a message is received.
socket->async_receive(boost::asio::buffer(buf_client, 3000), 0,
                      ClientReceiveEvent);
io_service->run();

// Write response.
boost::system::error_code ignored_error;
std::cout << "Sending message \n";
boost::asio::write(*socket, boost::asio::buffer("some data"), ignored_error);

Любое объяснение run(), описывающее его поведение в двух приведенных ниже примерах, будет оценено.

Ответы

Ответ 1

Фонд

Давайте начнем с упрощенного примера и рассмотрим соответствующие фрагменты Boost.Asio:

void handle_async_receive(...) { ... }
void print() { ... }

...  

boost::asio::io_service io_service;
boost::asio::ip::tcp::socket socket(io_service);

...

io_service.post(&print);                             // 1
socket.connect(endpoint);                            // 2
socket.async_receive(buffer, &handle_async_receive); // 3
io_service.post(&print);                             // 4
io_service.run();                                    // 5

Что такое обработчик?

Обработчик - это не что иное, как обратный вызов. В примере кода имеется 3 обработчика:

  • Обработчик print (1).
  • Обработчик handle_async_receive (3).
  • Обработчик print (4).

Несмотря на то, что одна и та же функция print() используется дважды, считается, что каждое использование создает свой собственный однозначно идентифицируемый обработчик. Обработчики могут иметь множество форм и размеров, начиная от базовых функций, таких как выше, до более сложных конструкций, таких как функторы, созданные из boost::bind() и lambdas. Независимо от сложности, обработчик по-прежнему остается не чем иным, как обратным вызовом.

Что такое работа?

Работа - это некоторая обработка, которую Boost.Asio запрашивал от имени кода приложения. Иногда Boost.Asio может начать часть работы, как только об этом расскажут, и в других случаях она может подождать, чтобы выполнить работу в более поздний момент времени. После того, как он завершит работу, Boost.Asio сообщит об этом приложению, вызвав предоставленный обработчик.

Boost.Asio гарантирует, что обработчики будут работать только в потоке, который в настоящее время вызывает run(), run_one(), poll() или poll_one(). Это потоки, которые будут работать и обработчики вызовов. Поэтому в приведенном выше примере print() не вызывается, когда он отправляется в io_service (1). Вместо этого он добавляется в io_service и будет вызываться в более поздний момент времени. В этом случае он находится в пределах io_service.run() (5).

Что такое асинхронные операции?

А <а href= "http://www.boost.org/doc/libs/1_56_0/doc/html/boost_asio/reference/asynchronous_operations.html" rel= "noreferrer" > асинхронная операция создает работу и Boost.Asio будет вызывать обработчик, чтобы сообщить приложение, когда работа завершена. Асинхронные операции создаются путем вызова функции с именем с префиксом async_. Эти функции также известны как инициирующие функции.

Асинхронные операции и могут быть разложены на три уникальных шага:

  • Инициирование или информирование связанного с ним io_service, которое должно быть выполнено. Операция async_receive (3) сообщает io_service, что ей потребуется асинхронно считывать данные из сокета, а затем async_receive немедленно возвращается.
  • Выполнение фактической работы. В этом случае, когда socket получает данные, байты будут считаны и скопированы в buffer. Фактическая работа будет выполнена либо:
    • Инициирующая функция (3), если Boost.Asio может определить, что она не будет блокироваться.
    • Когда приложение явно запускает io_service (5).
  • Вызов handle_async_receive ReadHandler. Еще раз, обработчики вызываются только в потоках, выполняющих io_service. Таким образом, независимо от того, когда работа выполнена (3 или 5), гарантируется, что handle_async_receive() будет вызван только в io_service.run() (5).

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

Что делает io_service.run() Do?

Когда поток вызывает io_service.run(), из этого потока будут вызываться работа и обработчики. В приведенном выше примере io_service.run() (5) будет заблокирован до:

  • Он вызывается и возвращается из обоих обработчиков print, операция получения завершается с успехом или неудачей, и его обработчик handle_async_receive был вызван и возвращен.
  • io_service явно остановлен через io_service::stop().
  • Исключение выбрано из обработчика.

Один потенциальный псевдо-иш-поток можно описать следующим образом:

create io_service
create socket
add print handler to io_service (1)
wait for socket to connect (2)
add an asynchronous read work request to the io_service (3)
add print handler to io_service (4)
run the io_service (5)
  is there work or handlers?
    yes, there is 1 work and 2 handlers
      does socket have data? no, do nothing
      run print handler (1)
  is there work or handlers?
    yes, there is 1 work and 1 handler
      does socket have data? no, do nothing
      run print handler (4)
  is there work or handlers?
    yes, there is 1 work
      does socket have data? no, continue waiting
  -- socket receives data --
      socket has data, read it into buffer
      add handle_async_receive handler to io_service
  is there work or handlers?
    yes, there is 1 handler
      run handle_async_receive handler (3)
  is there work or handlers?
    no, set io_service as stopped and return

Обратите внимание, что когда прочитанный закончен, он добавляет другой обработчик к io_service. Эта тонкая деталь является важной особенностью асинхронного программирования. Это позволяет привязывать обработчиков вместе. Например, если handle_async_receive не получил все ожидаемые данные, тогда его реализация могла бы отправить другую асинхронную операцию чтения, в результате получив io_service больше работы и, таким образом, не возвращаясь из io_service.run().

Обратите внимание, что при завершении работы io_service приложение должно reset() io_service перед повторным запуском.

Пример и пример кода 3a

Теперь рассмотрим две части кода, на которые ссылается вопрос.

Код вопроса

socket->async_receive добавляет работу к io_service. Таким образом, io_service->run() будет блокироваться до тех пор, пока операция чтения не завершится с успехом или ошибкой, а ClientReceiveEvent либо закончит выполнение, либо исключит исключение.

Пример 3a Код

В надежде на то, что это будет легче понять, вот небольшой аннотированный пример 3a:

void CalculateFib(std::size_t n);

int main()
{
  boost::asio::io_service io_service;
  boost::optional<boost::asio::io_service::work> work =       // '. 1
      boost::in_place(boost::ref(io_service));                // .'

  boost::thread_group worker_threads;                         // -.
  for(int x = 0; x < 2; ++x)                                  //   :
  {                                                           //   '.
    worker_threads.create_thread(                             //     :- 2
      boost::bind(&boost::asio::io_service::run, &io_service) //   .'
    );                                                        //   :
  }                                                           // -'

  io_service.post(boost::bind(CalculateFib, 3));              // '.
  io_service.post(boost::bind(CalculateFib, 4));              //   :- 3
  io_service.post(boost::bind(CalculateFib, 5));              // .'

  work = boost::none;                                         // 4
  worker_threads.join_all();                                  // 5
}

На высоком уровне программа создаст 2 потока, которые будут обрабатывать цикл событий io_service (2). Это приводит к простому пулу потоков, который будет вычислять числа Фибоначчи (3).

Единственное важное различие между кодом вопроса и этим кодом состоит в том, что этот код вызывает io_service::run() (2) до того, как фактическая работа и обработчики добавлены в io_service (3). Чтобы предотвратить немедленное возвращение io_service::run(), создается объект io_service::work (1). Этот объект предотвращает работу io_service; поэтому io_service::run() не будет возвращаться в результате отсутствия работы.

Общий поток выглядит следующим образом:

  • Создайте и добавьте объект io_service::work, добавленный в io_service.
  • Создан пул потоков, который вызывает io_service::run(). Эти рабочие потоки не возвращаются из io_service из-за объекта io_service::work.
  • Добавьте 3 обработчиков, которые вычисляют числа Фибоначчи в io_service и немедленно возвращаются. Рабочие потоки, а не основной поток, могут немедленно запустить эти обработчики.
  • Удалить объект io_service::work.
  • Дождитесь завершения рабочих потоков. Это произойдет только после завершения всех трех обработчиков, так как io_service не имеет обработчиков и не работает.

Код может быть написан по-разному, таким же образом, как и исходный код, где обработчики добавляются в io_service, а затем обрабатывается цикл событий io_service. Это устраняет необходимость использования io_service::work и приводит к следующему коду:

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

  io_service.post(boost::bind(CalculateFib, 3));              // '.
  io_service.post(boost::bind(CalculateFib, 4));              //   :- 3
  io_service.post(boost::bind(CalculateFib, 5));              // .'

  boost::thread_group worker_threads;                         // -.
  for(int x = 0; x < 2; ++x)                                  //   :
  {                                                           //   '.
    worker_threads.create_thread(                             //     :- 2
      boost::bind(&boost::asio::io_service::run, &io_service) //   .'
    );                                                        //   :
  }                                                           // -'
  worker_threads.join_all();                                  // 5
}

Синхронный и асинхронный

Хотя в этом вопросе используется асинхронная операция, он эффективно работает синхронно, поскольку он ожидает завершения асинхронной операции:

socket.async_receive(buffer, handler)
io_service.run();

эквивалентно:

boost::asio::error_code error;
std::size_t bytes_transferred = socket.receive(buffer, 0, error);
handler(error, bytes_transferred);

Как правило, старайтесь избегать смешивания синхронных и асинхронных операций. Часто это может превратить сложную систему в сложную систему. Этот ответ показывает преимущества асинхронного программирования, некоторые из которых также описаны в документации Boost.Asio .

Ответ 2

Чтобы упростить, как это делает run, подумайте об этом как о сотруднике, который должен обработать кучу бумаги; он берет один лист, делает то, что говорит лист, отбрасывает лист и берет следующий; когда у него заканчиваются листы, он покидает офис. На каждом листе может быть любая инструкция, даже добавив новый лист в кучу. Вернемся к asio: вы можете дать работе io_service двумя способами, по существу: используя post на нем, как в примере, который вы связали, или используя другие объекты, которые внутренне вызывают post на io_service, как методы socket и async_*.