Пошаговое руководство. Реализация фьючерсов
В данном разделе рассматриваются способы реализации фьючерсов в приложении. В этом разделе показано, как объединить существующие функциональные возможности в среде выполнения с параллелизмом для создания дополнительных функциональных возможностей.
Задача — это вычисление, которое можно разложить на дополнительные, более мелкие вычисления. Фьючерс — это асинхронная задача, которая вычисляет значение для последующего использования.
Для реализации фьючерсов в этом разделе определяется класс async_future. Классом async_future используются следующие компоненты среды выполнения с параллелизмом: класс Concurrency::task_group и класс Concurrency::single_assignment. Класс async_future использует класс task_group для асинхронного вычисления значения и класс single_assignment для хранения результата вычисления. Конструктор класса async_future принимает рабочую функцию, вычисляющую результат, а метод get извлекает результат.
Реализация класса фьючерса
Объявите шаблонный класс с именем async_future, параметризованный по типу результирующего вычисления. Добавьте в этот класс разделы public и private.
template <typename T> class async_future { public: private: };В разделе private класса async_future объявите элементы данных task_group и single_assignment.
// Executes the asynchronous work function. task_group _tasks; // Stores the result of the asynchronous work function. single_assignment<T> _value;Реализуйте конструктор в разделе public класса async_future. Конструктор представляет собой шаблон, параметризованный по рабочей функции, вычисляющей результат. Конструктор асинхронно выполняет рабочую функцию в элементе данных task_group и использует функцию Concurrency::send для записи результата в элемент данных single_assignment.
template <class Functor> explicit async_future(Functor&& fn) { // Execute the work function in a task group and send the result // to the single_assignment object. _tasks.run([fn, this]() { send(_value, fn()); }); }Реализуйте деструктор в разделе public класса async_future. Деструктор ожидает завершения выполнения задачи.
~async_future() { // Wait for the task to finish. _tasks.wait(); }Реализуйте метод get в разделе public класса async_future. Этот метод использует функцию Concurrency::receive для извлечения результата рабочей функции.
// Retrieves the result of the work function. // This method blocks if the async_future object is still // computing the value. T get() { return receive(_value); }
Пример
Описание
В приведенном ниже примере демонстрируется полный класс async_future и пример его использования. Функция wmain создает объект std::vector, содержащий 10000 случайных целых чисел. Затем объекты async_future используются для поиска наименьшего и наибольшего значений, содержащихся в объекте vector.
Код
// futures.cpp
// compile with: /EHsc
#include <ppl.h>
#include <agents.h>
#include <vector>
#include <algorithm>
#include <iostream>
#include <numeric>
#include <random>
using namespace Concurrency;
using namespace std;
template <typename T>
class async_future
{
public:
template <class Functor>
explicit async_future(Functor&& fn)
{
// Execute the work function in a task group and send the result
// to the single_assignment object.
_tasks.run([fn, this]() {
send(_value, fn());
});
}
~async_future()
{
// Wait for the task to finish.
_tasks.wait();
}
// Retrieves the result of the work function.
// This method blocks if the async_future object is still
// computing the value.
T get()
{
return receive(_value);
}
private:
// Executes the asynchronous work function.
task_group _tasks;
// Stores the result of the asynchronous work function.
single_assignment<T> _value;
};
int wmain()
{
// Create a vector of 10000 integers, where each element
// is between 0 and 9999.
mt19937 gen(2);
vector<int> values(10000);
generate(values.begin(), values.end(), [&gen]{ return gen()%10000; });
// Create a async_future object that finds the smallest value in the
// vector.
async_future<int> min_value([&]() -> int {
int smallest = INT_MAX;
for_each(values.begin(), values.end(), [&](int value) {
if (value < smallest)
{
smallest = value;
}
});
return smallest;
});
// Create a async_future object that finds the largest value in the
// vector.
async_future<int> max_value([&]() -> int {
int largest = INT_MIN;
for_each(values.begin(), values.end(), [&](int value) {
if (value > largest)
{
largest = value;
}
});
return largest;
});
// Calculate the average value of the vector while the async_future objects
// work in the background.
int sum = accumulate(values.begin(), values.end(), 0);
int average = sum / values.size();
// Print the smallest, largest, and average values.
wcout << L"smallest: " << min_value.get() << endl
<< L"largest: " << max_value.get() << endl
<< L"average: " << average << endl;
}
Комментарии
В данном примере получается следующий результат:
smallest: 0
largest: 9999
average: 4981
В примере для извлечения результатов вычисления используется метод async_future::get. Если вычисление еще выполняется, метод async_future::get ожидает завершения вычисления.
Надежное программирование
Для расширения класса async_future, чтобы он обрабатывал исключения, вызываемые рабочей функцией, измените метод async_future::get таким образом, чтобы он вызывал метод Concurrency::task_group::wait. Метод task_group::wait создает все исключения, возникшие в рабочей функции.
В следующем примере показан измененный вариант класса async_future: Функция wmain использует блок try-catch, чтобы вывести на печать результат объекта async_future или вывести на печать значение исключения, созданного рабочей функцией.
// futures-with-eh.cpp
// compile with: /EHsc
#include <ppl.h>
#include <agents.h>
#include <vector>
#include <algorithm>
#include <iostream>
using namespace Concurrency;
using namespace std;
template <typename T>
class async_future
{
public:
template <class Functor>
explicit async_future(Functor&& fn)
{
// Execute the work function in a task group and send the result
// to the single_assignment object.
_tasks.run([fn, this]() {
send(_value, fn());
});
}
~async_future()
{
// Wait for the task to finish.
_tasks.wait();
}
// Retrieves the result of the work function.
// This method blocks if the async_future object is still
// computing the value.
T get()
{
// Wait for the task to finish.
// The wait method throws any exceptions that were generated
// by the work function.
_tasks.wait();
// Return the result of the computation.
return receive(_value);
}
private:
// Executes the asynchronous work function.
task_group _tasks;
// Stores the result of the asynchronous work function.
single_assignment<T> _value;
};
int wmain()
{
// For illustration, create a async_future with a work
// function that throws an exception.
async_future<int> f([]() -> int {
throw exception("error");
});
// Try to read from the async_future object.
try
{
int value = f.get();
wcout << L"f contains value: " << value << endl;
}
catch (const exception& e)
{
wcout << L"caught exception: " << e.what() << endl;
}
}
В данном примере получается следующий результат:
caught exception: error
Дополнительные сведения о модели обработки исключений в среде выполнения с параллелизмом см. в разделе Обработка исключений в среде выполнения с параллелизмом.
Компиляция кода
Скопируйте код примера и вставьте его в проект Visual Studio или в файл с именем futures.cpp, затем выполните в окне командной строки Visual Studio 2010 следующую команду.
cl.exe /EHsc futures.cpp
См. также
Ссылки
Основные понятия
Обработка исключений в среде выполнения с параллелизмом
Другие ресурсы
Практические и пошаговые руководства, посвященные асинхронным агентам