| 1 | // Copyright (C) Microsoft Corporation. All rights reserved. |
| 2 | |
| 3 | /*++ |
| 4 | |
| 5 | Module Name: |
| 6 | |
| 7 | AsyncExecution.h |
| 8 | |
| 9 | Abstract: |
| 10 | |
| 11 | Provides ForEachAsync, a generic helper for executing a work callback |
| 12 | over a collection concurrently in bounded batches using std::async. |
| 13 | |
| 14 | --*/ |
| 15 | #pragma once |
| 16 | |
| 17 | #include <algorithm> |
| 18 | #include <future> |
| 19 | #include <optional> |
| 20 | #include <utility> |
| 21 | #include <vector> |
| 22 | #include <wil/result_macros.h> |
| 23 | |
| 24 | namespace wsl::windows::wslc { |
| 25 | |
| 26 | // Invokes onWork for each element in items concurrently, in batches of batchSize. |
| 27 | // Results are delivered serially to onSuccess. Errors are delivered serially to onError. |
| 28 | // |
| 29 | // This keeps wall time proportional to ceil(N / batchSize) rather than N for operations |
| 30 | // that have inherent per-item latency (e.g. network or IPC calls). |
| 31 | // |
| 32 | // Note: worker threads have no guaranteed per-thread initialization (e.g. COM). Callers |
| 33 | // whose onWork requires per-thread setup (such as CoInitializeEx) are responsible for |
| 34 | // performing it at the start of the onWork lambda. |
| 35 | // |
| 36 | // TWork : TItem -> TResult (called concurrently) |
| 37 | // TSuccess: TResult -> void (called serially) |
| 38 | // TError : (TItem, wil::ResultException) -> void (called serially) |
| 39 | template <typename TItem, typename TWork, typename TSuccess, typename TError> |
| 40 | void ForEachAsync(const std::vector<TItem>& items, TWork onWork, TSuccess onSuccess, TError onError, size_t batchSize = 10) |
| 41 | { |
| 42 | WI_ASSERT(batchSize > 0); |
| 43 | THROW_HR_IF(E_INVALIDARG, batchSize == 0); |
| 44 | |
| 45 | using TResult = decltype(onWork(std::declval<TItem>())); |
| 46 | |
| 47 | struct BatchResult |
| 48 | { |
| 49 | explicit BatchResult(TItem capturedItem) : item(std::move(capturedItem)) |
| 50 | { |
| 51 | } |
| 52 | |
| 53 | TItem item; |
| 54 | std::optional<TResult> result; |
| 55 | wil::ResultException error{S_OK}; |
| 56 | bool hasError{false}; |
| 57 | }; |
| 58 | |
| 59 | for (size_t batchStart = 0; batchStart < items.size(); batchStart += batchSize) |
| 60 | { |
| 61 | const size_t batchEnd = std::min(batchStart + batchSize, items.size()); |
| 62 | |
| 63 | std::vector<std::future<BatchResult>> futures; |
| 64 | futures.reserve(batchEnd - batchStart); |
| 65 | |
| 66 | for (size_t i = batchStart; i < batchEnd; ++i) |
| 67 | { |
| 68 | const auto& item = items[i]; |
| 69 | futures.push_back(std::async(std::launch::async, [&onWork, item]() -> BatchResult { |
| 70 | BatchResult result{item}; |
| 71 | try |
| 72 | { |
| 73 | result.result = onWork(item); |
| 74 | } |
| 75 | catch (const wil::ResultException& ex) |
| 76 | { |
| 77 | result.hasError = true; |
| 78 | result.error = ex; |
| 79 | } |
| 80 | return result; |
| 81 | })); |
| 82 | } |
| 83 | |
| 84 | for (auto& future : futures) |
| 85 | { |
| 86 | auto batchResult = future.get(); |
| 87 | |
| 88 | if (batchResult.hasError) |
| 89 | { |
| 90 | onError(batchResult.item, batchResult.error); |
| 91 | } |
| 92 | else if (batchResult.result.has_value()) |
| 93 | { |
| 94 | onSuccess(*batchResult.result); |
| 95 | } |
| 96 | } |
| 97 | } |
| 98 | } |
| 99 | |
| 100 | } // namespace wsl::windows::wslc |