Как составлять асинхронные операции?

Question

Я ищу способ составления асинхронных операций.Конечная цель состоит в том, чтобы выполнить асинхронную операцию и либо запустить ее до завершения, либо вернуть по истечении заданного пользователем тайм-аута.

В качестве примера предположим, что я ищу способ объединить следующеесопрограммы ¹ :

IAsyncOperation<IBuffer> read(IBuffer buffer, uint32_t count)
{
    auto&& result{ co_await socket_.InputStream().ReadAsync(buffer, count, InputStreamOptions::None) };
    co_return result;
}

с socket_, являющимся экземпляром StreamSocket.

И сопрограмма времени ожидания:

IAsyncAction timeout()
{
    co_await 5s;
}

Я ищу способ объединить эти сопрограммы таким образом, чтобы он возвращался как можно скорее, либо после прочтения данных, либо по истечении времени ожидания.

Этоварианты, которые я до сих пор оценивал:

• C ++ 20 сопрограмм: насколько я понимаю P1056R0 , в настоящее время нет библиотеки или языковой функции ", чтобы разрешить созданиеи состав сопрограмм ".
• Windows Runtime предоставляла асинхронные типы задач, в конечном счете полученные из IAsyncInfo : Опять же, я не нашел никаких средств, которые позволили бы мне комбинировать задачикак мне нужно.
• Concurrency Runtime : Это выглядит многообещающе, особенно шаблон функции when_any выглядит именно то, что мне нужно.

С этогоПохоже, мне нужно использовать Concurrency Runtime.Однако мне трудно собрать все части вместе.Меня особенно смущает вопрос о том, как обрабатывать исключения, и требуется ли отмена соответствующей другой параллельной задачи.

Вопрос состоит из двух частей:

• Является ли среда выполнения параллелизмаединственный вариант (приложение UWP)?
• Как будет выглядеть реализация?

---

¹ Методы являются внутренними для приложения.Не требуется, чтобы они возвращали совместимые с Windows Runtime типы.

Answer 1

Как предполагает Ли МакФерсон в другом ответе , Параллельная среда выполнения выглядит жизнеспособным вариантом.Он предоставляет задачу с, которую можно объединять с другими, объединять в цепочку с помощью продолжений, а также легко интегрировать с асинхронной моделью среды выполнения Windows (см. Создание асинхронных операций в C ++ для приложений UWP ).В качестве бонуса, в том числе заголовок <pplawait.h> предоставляет адаптеры для экземпляров шаблона класса concurrency::task, которые будут использоваться в качестве сопрограмм C ++ 20.

Я не смог ответить на все вопросы, но этоэто то, что я в конце концов придумал.Для простоты (и простоты проверки) я использую Sleep вместо фактической операции чтения и возвращаю int вместо IBuffer.

Compositionзадач

ConcRT предоставляет несколько способов объединения задач.С учетом требований concurrency::when_any можно использовать для создания задачи, которая возвращается после завершения любой из предоставленных задач.Когда в качестве входных данных передаются только 2 задачи, доступен также вспомогательный оператор (operator||).

Распространение исключений

Исключения, поднятые излюбая из входных задач не считается успешным завершением.При использовании с задачей when_any создание исключения не будет достаточным условием ожидания.Как следствие, исключения не могут использоваться для выхода из объединенных задач.Чтобы справиться с этим, я решил вернуть std::optional и вызвать соответствующие исключения в продолжении then.

Отмена задания

Это до сих пор для меня загадка.Похоже, что как только задача удовлетворяет условию ожидания задачи when_any, нет необходимости отменять соответствующие другие невыполненные задачи.Как только они завершены (успешно или нет), они молча рассматриваются.

Ниже приведен код, использующий упрощения, упомянутые ранее.Он создает задачу, состоящую из фактической рабочей нагрузки и задачи тайм-аута, которые возвращают std::optional.Продолжение then проверяет возвращаемое значение и выдает исключение в случае, если его нет (т. Е. timeout_task завершено первым).

#include <Windows.h>

#include <cstdint>
#include <iostream>
#include <optional>
#include <ppltasks.h>
#include <stdexcept>

using namespace concurrency;

task<int> read_with_timeout(uint32_t read_duration, uint32_t timeout)
{
    auto&& read_task
    {
        create_task([read_duration]
            {
                ::Sleep(read_duration);
                return std::optional<int>{42};
            })
    };
    auto&& timeout_task
    {
        create_task([timeout]
            {
                ::Sleep(timeout);
                return std::optional<int>{};
            })
    };

    auto&& task
    {
        (read_task || timeout_task)
        .then([](std::optional<int> result)
            {
                if (!result.has_value())
                {
                    throw std::runtime_error("timeout");
                }
                return result.value();
            })
    };
    return task;
}

Следующий тестовый код

int main()
{
    try
    {
        auto res1{ read_with_timeout(3000, 5000).get() };
        std::cout << "Succeeded. Result = " << res1 << std::endl;
        auto res2{ read_with_timeout(5000, 3000).get() };
        std::cout << "Succeeded. Result = " << res2 << std::endl;
    }
    catch( std::runtime_error const& e )
    {
        std::cout << "Failed. Exception = " << e.what() << std::endl;
    }
}

производит этот вывод:

Succeeded. Result = 42
Failed. Exception = timeout

Answer 2

Я думаю, что проще всего было бы использовать библиотеку concurrency.Вам нужно изменить время ожидания, чтобы оно возвращало тот же тип, что и первый метод, даже если он возвращает ноль.

(я понимаю, что это только частичный ответ ...)

Мой C ++ отстой, но я думаю, что это близко ...

array<task<IBuffer>, 2> tasks =
{
concurrency::create_task([]{return read(buffer, count).get();}),
concurrency::create_task([]{return modifiedTimeout.get();})
};

concurrency::when_any(begin(tasks), end(tasks)).then([](IBuffer buffer)
{ 
    //do something 
});