Параллельное программирование на С++ в действии. Практика разработки многопоточных программ - [43]

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

Шаблон std::promise дает возможность задать значение (типа T), которое впоследствии можно будет прочитать с помощью ассоциированного объекта std::future. Пара std::promise/std::future реализует один из возможных механизмов такого рода; ожидающий поток приостанавливается в ожидании будущего результата, тогда как поток, поставляющий данные, может с помощью promise установить ассоциированное значение и сделать будущий результат готовым.

Чтобы получить объект std::future, ассоциированный с данным обещанием std::promise, мы должны вызвать функцию-член get_future() — так же, как в случае std::packaged_task. После установки значения обещания (с помощью функции-члена set_value()) будущий результат становится готовым, и его можно использовать для получения установленного значения. Если уничтожить объект std::promise, не установив значение, то в будущем результате будет сохранено исключение. О передаче исключений между потоками см. раздел 4.2.4.

В листинге 4.10 приведен код потока обработки соединений, написанный по только что изложенной схеме. В данном случае для уведомления об успешной передаче блока исходящих данных применяется пара std::promise/std::future; ассоциированное с будущим результатом значение — это просто булевский флаг успех/неудача. Для входящих пакетов в качестве ассоциированных данных могла бы выступать полезная нагрузка пакета.

Листинг 4.10. Обработка нескольких соединений в одном потоке с помощью объектов-обещаний

#include

void process_connections(connection_set& connections) {

 while(!done(connections)) {             ←(1)

  for (connection_iterator               ←(2)

   connection = connections.begin(), end = connections.end();
   connection != end;
   ++connection) {

   if (connection->has_incoming_data()) {←(3)

    data_packet data = connection->incoming();
    std::promise& p =

     connection->get_promise(data.id);   ←(4)

    p.set_value(data.payload);
   }

   if (connection->has_outgoing_data()) {←(5)

    outgoing_packet data =
     connection->top_of_outgoing_queue();
    connection->send(data.payload);

    data.promise.set_value(true);        ←(6)

   }
  }
 }
}

Функция process_connections() повторяет цикл, пока done() возвращает true(1). На каждой итерации поочередно проверяется каждое соединение (2); если есть входящие данные, они читаются (3), а если в очереди имеются исходящие данные, они отсылаются (5). При этом предполагается, что в каждом входящем пакете хранится некоторый идентификатор и полезная нагрузка, содержащая собственно данные. Идентификатору сопоставляется объект std::promise (возможно, путем поиска в ассоциативном контейнере) (4), значением которого является полезная нагрузка пакета. Исходящие пакеты просто извлекаются из очереди отправки и передаются но соединению. После завершения передачи в обещание, ассоциированное с исходящими данными, записывается значение true, обозначающее успех (6). Насколько хорошо эта схема ложится на фактический сетевой протокол, зависит от самого протокола; в конкретном случае схема обещание/будущий результат может и не подойти, хотя структурно она аналогична поддержке асинхронного ввода/вывода в некоторых операционных системах.

В коде выше мы полностью проигнорировали возможные исключения. Хотя мир, в котором всё всегда работает правильно, был бы прекрасен, действительность не так радужна. Переполняются диски, не находятся искомые данные, отказывает сеть, «падает» база данных — всякое бывает. Если бы операция выполнялась в том потоке, которому нужен результат, программа могла бы просто сообщить об ошибке с помощью исключения. Но было бы неоправданным ограничением требовать, чтобы всё работало правильно только потому, что мы захотели воспользоваться классами std::packaged_task или std::promise.

Поэтому в стандартной библиотеке С++ имеется корректный способ учесть возникновение исключений в таком контексте и сохранить их как часть ассоциированного результата.

4.2.4. Сохранение исключения в будущем результате

Рассмотрим следующий коротенький фрагмент. Если передать функции square_root() значение -1, то она возбудит исключение, которое увидит вызывающая программа:

double square_root(double x) {
 if (x<0) {
  throw std::out_of_range("x<0");
 }
 return sqrt(x);
}

А теперь предположим, что вместо вызова square_root() в текущем потоке

double y = square_root(-1);

мы вызываем ее асинхронно:

std::future f = std::async(square_root,-1);
double y = f.get();

В идеале хотелось бы получить точно такое же поведение: чтобы поток, вызывающий f.get(), мог увидеть не только нормальное значение y, но и исключение — как в однопоточной программе.

Что ж, именно так на самом деле и происходит: если функция, вызванная через