coro::when_any (#298)

Adds a new construct that will return the first task's result upon its
completion. All other task results will be discarded/detached/orphaned.

There is two ways to currently invoke when_any, one with a
std::stop_token that will signal to the other tasks that a task has
already completed, this requires the user to check for that stop token
requesting a stop, it is not automatic.

The other way is fire and forget, all tasks will be required to complete
but only the first tasks result will be used. This method isn't
particularly recommended but the API is available in the case where a
stop token isn't required.

EMSCRIPTEN does not support std::stop_source|token so this new feature
is currently disabled on that platform, I do not want to shim it in.

Closes #279

Author: jbaldwin

.githooks/pre-commit

@@ -82,4 +82,8 @@ template_contents=$(cat 'README.md')
 example_contents=$(cat 'examples/coro_when_all.cpp')
 echo "${template_contents/\$\{EXAMPLE_CORO_WHEN_ALL\}/$example_contents}" > README.md
 
+template_contents=$(cat 'README.md')
+example_contents=$(cat 'examples/coro_when_any.cpp')
+echo "${template_contents/\$\{EXAMPLE_CORO_WHEN_ANY\}/$example_contents}" > README.md
+
 git add README.md

.githooks/readme-template.md

@@ -16,6 +16,7 @@
 * Higher level coroutine constructs
     - [coro::sync_wait(awaitable)](#sync_wait)
     - [coro::when_all(awaitable...) -> awaitable](#when_all)
+    - [coro::when_any(awaitable...) -> awaitable](#when_any)
     - [coro::task<T>](#task)
     - [coro::generator<T>](#generator)
     - [coro::event](#event)
@@ -70,7 +71,7 @@ Offload Result = 20
 ```
 
 ### when_all
-The `when_all` construct can be used within coroutines to await a set of tasks, or it can be used outside coroutinne context in conjunction with `sync_wait` to await multiple tasks. Each task passed into `when_all` will initially be executed serially by the calling thread so it is recommended to offload the tasks onto a scheduler like `coro::thread_pool` or `coro::io_scheduler` so they can execute in parallel.
+The `when_all` construct can be used within coroutines to await a set of tasks, or it can be used outside coroutine context in conjunction with `sync_wait` to await multiple tasks. Each task passed into `when_all` will initially be executed serially by the calling thread so it is recommended to offload the tasks onto an executor like `coro::thread_pool` or `coro::io_scheduler` so they can execute in parallel.
 
 ```C++
 ${EXAMPLE_CORO_WHEN_ALL}
@@ -87,6 +88,21 @@ $ ./examples/coro_when_all
 first: 1.21 second: 20
 ```
 
+### when_any
+The `when_any` construct can be used within coroutines to await a set of tasks and only return the result of the first task that completes. This can also be used outside of a coroutine context in conjunction with `sync_wait` to await the first result. Each task passed into `when_any` will initially be executed serially by the calling thread so it is recommended to offload the tasks onto an executor like `coro::thread_pool` or `coro::io_scheduler` so they can execute in parallel.
+
+```C++
+${EXAMPLE_CORO_WHEN_ANY}
+```
+
+Expected output:
+```bash
+$ ./examples/coro_when_any
+result = 1
+result = -1
+```
+
+
 ### task
 The `coro::task<T>` is the main coroutine building block within `libcoro`.  Use task to create your coroutines and `co_await` or `co_yield` tasks within tasks to perform asynchronous operations, lazily evaluation or even spreading work out across a `coro::thread_pool`.  Tasks are lightweight and only begin execution upon awaiting them.
 

CMakeLists.txt

@@ -93,6 +93,7 @@ set(LIBCORO_SOURCE_FILES
     include/coro/thread_pool.hpp src/thread_pool.cpp
     include/coro/time.hpp
     include/coro/when_all.hpp
+    include/coro/when_any.hpp
 )
 
 if(LIBCORO_FEATURE_NETWORKING)

README.md

@@ -16,6 +16,7 @@
 * Higher level coroutine constructs
     - [coro::sync_wait(awaitable)](#sync_wait)
     - [coro::when_all(awaitable...) -> awaitable](#when_all)
+    - [coro::when_any(awaitable...) -> awaitable](#when_any)
     - [coro::task<T>](#task)
     - [coro::generator<T>](#generator)
     - [coro::event](#event)
@@ -101,7 +102,7 @@ Offload Result = 20
 ```
 
 ### when_all
-The `when_all` construct can be used within coroutines to await a set of tasks, or it can be used outside coroutinne context in conjunction with `sync_wait` to await multiple tasks. Each task passed into `when_all` will initially be executed serially by the calling thread so it is recommended to offload the tasks onto a scheduler like `coro::thread_pool` or `coro::io_scheduler` so they can execute in parallel.
+The `when_all` construct can be used within coroutines to await a set of tasks, or it can be used outside coroutine context in conjunction with `sync_wait` to await multiple tasks. Each task passed into `when_all` will initially be executed serially by the calling thread so it is recommended to offload the tasks onto an executor like `coro::thread_pool` or `coro::io_scheduler` so they can execute in parallel.
 
 ```C++
 #include <coro/coro.hpp>
@@ -169,6 +170,68 @@ $ ./examples/coro_when_all
 first: 1.21 second: 20
 ```
 
+### when_any
+The `when_any` construct can be used within coroutines to await a set of tasks and only return the result of the first task that completes. This can also be used outside of a coroutine context in conjunction with `sync_wait` to await the first result. Each task passed into `when_any` will initially be executed serially by the calling thread so it is recommended to offload the tasks onto an executor like `coro::thread_pool` or `coro::io_scheduler` so they can execute in parallel.
+
+```C++
+#include <coro/coro.hpp>
+#include <iostream>
+
+int main()
+{
+    // Create a scheduler to execute all tasks in parallel and also so we can
+    // suspend a task to act like a timeout event.
+    auto scheduler = coro::io_scheduler::make_shared();
+
+    // This task will behave like a long running task and will produce a valid result.
+    auto make_long_running_task = [](std::shared_ptr<coro::io_scheduler> scheduler,
+                                     std::chrono::milliseconds           execution_time) -> coro::task<int64_t>
+    {
+        // Schedule the task to execute in parallel.
+        co_await scheduler->schedule();
+        // Fake doing some work...
+        co_await scheduler->yield_for(execution_time);
+        // Return the result.
+        co_return 1;
+    };
+
+    auto make_timeout_task = [](std::shared_ptr<coro::io_scheduler> scheduler) -> coro::task<int64_t>
+    {
+        // Schedule a timer to be fired so we know the task timed out.
+        co_await scheduler->schedule_after(std::chrono::milliseconds{100});
+        co_return -1;
+    };
+
+    // Example showing the long running task completing first.
+    {
+        std::vector<coro::task<int64_t>> tasks{};
+        tasks.emplace_back(make_long_running_task(scheduler, std::chrono::milliseconds{50}));
+        tasks.emplace_back(make_timeout_task(scheduler));
+
+        auto result = coro::sync_wait(coro::when_any(std::move(tasks)));
+        std::cout << "result = " << result << "\n";
+    }
+
+    // Example showing the long running task timing out.
+    {
+        std::vector<coro::task<int64_t>> tasks{};
+        tasks.emplace_back(make_long_running_task(scheduler, std::chrono::milliseconds{500}));
+        tasks.emplace_back(make_timeout_task(scheduler));
+
+        auto result = coro::sync_wait(coro::when_any(std::move(tasks)));
+        std::cout << "result = " << result << "\n";
+    }
+}
+```
+
+Expected output:
+```bash
+$ ./examples/coro_when_any
+result = 1
+result = -1
+```
+
+
 ### task
 The `coro::task<T>` is the main coroutine building block within `libcoro`.  Use task to create your coroutines and `co_await` or `co_yield` tasks within tasks to perform asynchronous operations, lazily evaluation or even spreading work out across a `coro::thread_pool`.  Tasks are lightweight and only begin execution upon awaiting them.
 

examples/CMakeLists.txt

@@ -69,4 +69,10 @@ if(LIBCORO_FEATURE_NETWORKING)
     add_executable(coro_http_200_ok_server coro_http_200_ok_server.cpp)
     target_link_libraries(coro_http_200_ok_server PUBLIC libcoro)
     target_compile_options(coro_http_200_ok_server PUBLIC ${LIBCORO_EXAMPLE_OPTIONS})
+
+    if(NOT EMSCRIPTEN)
+        add_executable(coro_when_any coro_when_any.cpp)
+        target_link_libraries(coro_when_any PUBLIC libcoro)
+        target_compile_options(coro_when_any PUBLIC ${LIBCORO_EXAMPLE_OPTIONS})
+    endif()
 endif()

examples/coro_when_any.cpp

@@ -0,0 +1,48 @@
+#include <coro/coro.hpp>
+#include <iostream>
+
+int main()
+{
+    // Create a scheduler to execute all tasks in parallel and also so we can
+    // suspend a task to act like a timeout event.
+    auto scheduler = coro::io_scheduler::make_shared();
+
+    // This task will behave like a long running task and will produce a valid result.
+    auto make_long_running_task = [](std::shared_ptr<coro::io_scheduler> scheduler,
+                                     std::chrono::milliseconds           execution_time) -> coro::task<int64_t>
+    {
+        // Schedule the task to execute in parallel.
+        co_await scheduler->schedule();
+        // Fake doing some work...
+        co_await scheduler->yield_for(execution_time);
+        // Return the result.
+        co_return 1;
+    };
+
+    auto make_timeout_task = [](std::shared_ptr<coro::io_scheduler> scheduler) -> coro::task<int64_t>
+    {
+        // Schedule a timer to be fired so we know the task timed out.
+        co_await scheduler->schedule_after(std::chrono::milliseconds{100});
+        co_return -1;
+    };
+
+    // Example showing the long running task completing first.
+    {
+        std::vector<coro::task<int64_t>> tasks{};
+        tasks.emplace_back(make_long_running_task(scheduler, std::chrono::milliseconds{50}));
+        tasks.emplace_back(make_timeout_task(scheduler));
+
+        auto result = coro::sync_wait(coro::when_any(std::move(tasks)));
+        std::cout << "result = " << result << "\n";
+    }
+
+    // Example showing the long running task timing out.
+    {
+        std::vector<coro::task<int64_t>> tasks{};
+        tasks.emplace_back(make_long_running_task(scheduler, std::chrono::milliseconds{500}));
+        tasks.emplace_back(make_timeout_task(scheduler));
+
+        auto result = coro::sync_wait(coro::when_any(std::move(tasks)));
+        std::cout << "result = " << result << "\n";
+    }
+}

include/coro/coro.hpp

@@ -41,3 +41,4 @@
 #include "coro/thread_pool.hpp"
 #include "coro/time.hpp"
 #include "coro/when_all.hpp"
+#include "coro/when_any.hpp"

include/coro/when_all.hpp

@@ -180,12 +180,12 @@ class when_all_ready_awaitable
     when_all_ready_awaitable(when_all_ready_awaitable&& other) noexcept(
         std::is_nothrow_move_constructible_v<task_container_type>)
         : m_latch(std::move(other.m_latch)),
-          m_tasks(std::move(m_tasks))
+          m_tasks(std::move(other.m_tasks))
     {
     }
 
     auto operator=(const when_all_ready_awaitable&) -> when_all_ready_awaitable& = delete;
-    auto operator=(when_all_ready_awaitable&) -> when_all_ready_awaitable&       = delete;
+    auto operator=(when_all_ready_awaitable&&) -> when_all_ready_awaitable&       = delete;
 
     auto operator co_await() & noexcept
     {

include/coro/when_any.hpp

@@ -0,0 +1,122 @@
+#pragma once
+
+// EMSCRIPTEN does not currently support std::jthread or std::stop_source|token.
+#ifndef EMSCRIPTEN
+
+    #include "coro/concepts/awaitable.hpp"
+    #include "coro/detail/task_self_deleting.hpp"
+    #include "coro/event.hpp"
+    #include "coro/mutex.hpp"
+    #include "coro/task.hpp"
+
+    #include <atomic>
+    #include <cassert>
+    #include <coroutine>
+    #include <stop_token>
+    #include <utility>
+    #include <vector>
+
+namespace coro
+{
+
+namespace detail
+{
+
+template<concepts::awaitable awaitable, typename return_type>
+static auto make_when_any_task(
+    awaitable                   a,
+    coro::mutex&                m,
+    std::atomic<bool>&          return_value_set,
+    coro::event&                notify,
+    std::optional<return_type>& return_value) -> coro::task<void>
+{
+    auto result = co_await static_cast<awaitable&&>(a);
+    co_await m.lock();
+    // Its important to only touch return_value and notify once since their lifetimes will be destroyed
+    // after being set ane notified the first time.
+    if (return_value_set.load(std::memory_order::acquire) == false)
+    {
+        return_value_set.store(true, std::memory_order::release);
+        return_value = std::move(result);
+        notify.set();
+    }
+
+    co_return;
+}
+
+template<
+    std::ranges::range  range_type,
+    concepts::awaitable awaitable_type = std::ranges::range_value_t<range_type>,
+    typename return_type               = typename concepts::awaitable_traits<awaitable_type>::awaiter_return_type,
+    typename return_type_base          = std::remove_reference_t<return_type>>
+static auto make_when_any_controller_task(
+    range_type awaitables, coro::event& notify, std::optional<return_type_base>& return_value)
+    -> coro::detail::task_self_deleting
+{
+    // These must live for as long as the longest running when_any task since each task tries to see
+    // if it was the first to complete. Only the very first task to complete will set the return_value
+    // and notify.
+    coro::mutex       m{};
+    std::atomic<bool> return_value_set{false};
+
+    // This detatched task will maintain the lifetime of all the when_any tasks.
+    std::vector<coro::task<void>> tasks{};
+
+    if constexpr (std::ranges::sized_range<range_type>)
+    {
+        tasks.reserve(std::size(awaitables));
+    }
+
+    for (auto&& a : awaitables)
+    {
+        tasks.emplace_back(make_when_any_task<awaitable_type, return_type_base>(
+            std::move(a), m, return_value_set, notify, return_value));
+    }
+
+    co_await coro::when_all(std::move(tasks));
+    co_return;
+}
+
+} // namespace detail
+
+template<
+    std::ranges::range  range_type,
+    concepts::awaitable awaitable_type = std::ranges::range_value_t<range_type>,
+    typename return_type               = typename concepts::awaitable_traits<awaitable_type>::awaiter_return_type,
+    typename return_type_base          = std::remove_reference_t<return_type>>
+[[nodiscard]] auto when_any(std::stop_source stop_source, range_type awaitables) -> coro::task<return_type_base>
+{
+    // Using an std::optional to prevent the need to default construct the type on the stack.
+    std::optional<return_type_base> return_value{std::nullopt};
+    coro::event                     notify{};
+
+    auto controller_task =
+        detail::make_when_any_controller_task(std::forward<range_type>(awaitables), notify, return_value);
+    controller_task.handle().resume();
+
+    co_await notify;
+    stop_source.request_stop();
+    co_return std::move(return_value.value());
+}
+
+template<
+    std::ranges::range  range_type,
+    concepts::awaitable awaitable_type = std::ranges::range_value_t<range_type>,
+    typename return_type               = typename concepts::awaitable_traits<awaitable_type>::awaiter_return_type,
+    typename return_type_base          = std::remove_reference_t<return_type>>
+[[nodiscard]] auto when_any(range_type awaitables) -> coro::task<return_type_base>
+{
+    std::optional<return_type_base> return_value{std::nullopt};
+    coro::event                     notify{};
+
+    auto controller_task =
+        detail::make_when_any_controller_task(std::forward<range_type>(awaitables), notify, return_value);
+    controller_task.handle().resume();
+
+    co_await notify;
+    co_return std::move(return_value.value());
+}
+
+} // namespace coro
+
+#endif // EMSCRIPTEN

test/CMakeLists.txt

@@ -17,6 +17,12 @@ set(LIBCORO_TEST_SOURCE_FILES
     catch_amalgamated.cpp
 )
 
+if(NOT EMSCRIPTEN)
+    list(APPEND LIBCORO_TEST_SOURCE_FILES
+        test_when_any.cpp
+    )
+endif()
+
 if(LIBCORO_FEATURE_NETWORKING)
     list(APPEND LIBCORO_TEST_SOURCE_FILES
         net/test_ip_address.cpp