Compiler Explorer

Source code

// Std library
#include <iostream>
#include <chrono>
#include <thread>
#include <future>
#include <vector>

// Application
#include "ResultType.hpp"
#include "JoinTasks.hpp"

// Exception handler
auto exceptionHandler = [](std::exception_ptr eptr) 
{
    try 
    {
        if (eptr) std::rethrow_exception(eptr);
    } catch (const std::exception& ex) 
    {
        std::cerr << "Caught exception: " << ex.what() << '\n';
    }
};

// Tests

template <class Executor>
auto test_join_tasks(Executor&& executor) -> utils::coro::ResultType<void>
{
    
    using namespace std::chrono_literals;

auto task1 = [] { std::this_thread::sleep_for(10ms); std::cout << "tid=" << std::this_thread::get_id() << ": Task 1\n";};
    [[maybe_unused]] auto task2 = [] { std:: cout << "tid=" << std::this_thread::get_id(); throw std::runtime_error("Error in Task 2"); };
    auto task3 = [] { std::cout << "tid=" << std::this_thread::get_id() << ": Task 3\n"; };

// Wait on the tasks completion
    co_await utils::coro::join_tasks(exceptionHandler, std::forward<Executor>(executor), task1, task3);
    //co_await utils::coro::join_tasks(exceptionHandler, std::forward<Executor>(executor), task1, task2, task3); // for handling exception
}

auto test_join_tasks_on_thread() -> utils::coro::ResultType<void>
{
    std::cout << __func__ << ", tid: " << std::this_thread::get_id() << '\n';

// Simple thread-based executor
    auto executor = [](auto&& task) 
    {
        std::thread(std::move(task)).detach();  
    };

// return test_join_tasks(nullptr); // for joining the tasks on the same thread as caller
    return test_join_tasks(std::move(executor)); // for joining the tasks on a different threads
}

auto test_join_tasks_on_async() -> utils::coro::ResultType<int>
{
    using namespace std::chrono_literals;

std::cout << __func__ << ", tid: " << std::this_thread::get_id() << '\n';

auto task1 = []()->int { std::cout << "tid=" << std::this_thread::get_id() << ": Task 1\n"; return 100;};
    auto task2 = []()->int { std::cout << "tid=" << std::this_thread::get_id() << ": Task 2\n"; return 200;};

std::vector<std::future<int>> results;
    results.reserve(2);
    
    // Executor - use std::async
    auto executor = [&](auto&& task) mutable
    {
         results.push_back(std::async(std::launch::async, std::move(task)));
    };

co_await utils::coro::join_tasks(exceptionHandler, std::move(executor), task1, task2);

// Sum the values, after all tasks are completed
    int sum = 0; for (auto& result : results) {sum += result.get();}

co_return sum;
}

int main()
{
    // Test on the std::async as executor
    const auto result = test_join_tasks_on_async(); 
    std::cout << "All tasks completed\n";
    std::cout << "sum= " << result.get() << '\n';

// Test on detached thread
    auto r = test_join_tasks_on_thread(); 
    std::cout << "All tasks completed";

return 0;
}

Source code

//
// @author: Damir Ljubic
// @email: damirlj@yahoo.com
// All rights reserved!
//

#pragma once

#include <coroutine>
#include <concepts>
#include <tuple>
#include <latch>
#include <functional>
#include <exception>
#include <utility>
#include <type_traits>
#include <thread>

namespace utils::coro
{

struct empty_type {};
    using exception_handler = std::function<void(std::exception_ptr)>; // std::function non-copyable

/**
     * Join operator.
     * For waiting on the arbitrary number of tasks to complete.
     * 
     *
     * @param handler The exception handler
     * @param executor The dedicated executor context, for hosting the tasks in parallel, asynchronous
     * manner. If none is provided (nullptr), the tasks will be executed sequentially, within the same
     * context as the suspended caller (coroutine).
     *
     * @param tasks The list of tasks to wait for.
     * The expectation is that all tasks are of the same signature (parameterless), may - or may not
     * return the value
     *
     * @return The Awaitable to synchronize with on co_await
     */
    template <typename Executor, typename... Tasks>
    requires (sizeof...(Tasks) > 0)
    auto join_tasks(exception_handler handler, Executor&& executor, Tasks&&... tasks)
    {
        
        struct Awaitable final : public std::suspend_always
        {
            // C-tor
            explicit Awaitable(exception_handler handler, Executor&& executor, Tasks&&... tasks) noexcept
                : exception_handler_(handler)
                , executor_(std::forward<Executor>(executor))
                , tasks_(std::make_tuple(std::forward<Tasks>(tasks)...))
            {
                if constexpr (not std::is_null_pointer_v<Executor>)
                    static_assert((std::invocable<Executor, Tasks> && ...), "<Awaitable> Invalid Executor type!");
            }

auto await_suspend(std::coroutine_handle<> other)
            {
                /*
                 * Equip each task additionally with the std::latch - count down mechanism (decorator pattern)
                 */
                auto signaling_task = [this](auto& task) mutable
                {
                    std::cout << __func__ << ", tid=" << std::this_thread::get_id() << '\n';

auto job = [&, this]() mutable
                    {
                        using result_type = std::invoke_result_t<decltype(task)>;
                        constexpr bool has_return = not std::is_void_v<result_type>;

std::conditional_t<has_return, result_type, empty_type> result {};

try
                        {
                            if constexpr (has_return) result = std::invoke(task);
                            else std::invoke(task);
                        }
                        catch (...)
                        {
                            const auto exp = std::current_exception();
                            if (exception_handler_) std::invoke(exception_handler_, exp);
                        }

signal_.count_down();

if constexpr (has_return) return result;
                    };

// Execute job
                    if constexpr (std::is_null_pointer_v<Executor>) std::invoke(std::move(job));  // on the same thread as caller
                    else std::invoke(executor_, std::move(job));  // on a dedicated execution context

}; // singaling_task

/*
                 * Depends on the executor, the tasks can be spawned on the different execution context,
                 * or be executed sequentially (std::nullptr_t): on the 
                 * same thread as the caller
                 */
                auto joiner = [&, this]<std::size_t... Is>(std::index_sequence<Is...>) mutable
                { 
                    (signaling_task(std::get<Is>(tasks_)), ...); 
                    
                    // Wait on all tasks being completed
                    signal_.wait();
                    // Resume the caller afterwards
                    other.resume();
                };

/*
                 * @note: this is still synchronous - waiting join (!), since we rely on the 
                 * internal sync mechanism (std::latch) - and the values are passed by reference: we can't detach
                 */

// joiner(std::make_index_sequence<sizeof...(Tasks)>{});
                std::thread([&, this]{joiner(std::make_index_sequence<sizeof...(Tasks)>{});}).join();
            
            }

private:
            std::latch signal_{sizeof...(Tasks)}; /**Signal the tasks completion*/

exception_handler exception_handler_; /**Exception handler*/
            using executor_type = std::conditional_t<not std::is_null_pointer_v<Executor>, std::remove_cvref_t<Executor>, std::nullptr_t>;
            executor_type executor_; /** Execution context for the tasks, or std::nullptr_t*/
            std::tuple<std::remove_cvref_t<Tasks>...> tasks_; /**list of the tasks*/
        };  // Awaitable

return Awaitable{handler, std::forward<Executor>(executor), std::forward<Tasks>(tasks)...};
    }
}  // namespace utils::coro

Source code

#pragma once

#include <coroutine>
#include <exception>
#include <utility>

namespace utils::coro
{
    
    template <typename T, bool lazy> struct promise_type_impl; //forward declaration

template <typename T, bool lazy = false>
    class [[nodiscard]] ResultType final
    {
        
      public:
       
        using value_type = std::remove_cvref_t<T>;
        using promise_type = promise_type_impl<value_type, lazy>;
        friend promise_type;

using handle_type = std::coroutine_handle<promise_type>;
        
        ~ResultType() { if (handle_) { handle_.destroy(); } } // sealed class

// Make the result type move-only, due to ownership over the handle
        ResultType(const ResultType&) = delete;
        ResultType& operator=(const ResultType&) = delete;

// Move semantic
        ResultType(ResultType&& other) noexcept
            : handle_(std::exchange(other.handle_, nullptr))
        {}

ResultType& operator=(ResultType&& other) noexcept
        {
            if (this != std::addressof(other))
            {
                using namespace std;
                ResultType tmp{std::move(other)};
                swap(*this, tmp);
            }   
            return *this;    
        }

void resume()
        {
            if (not handle_.done()) { handle_.resume(); }
        }

auto get() const 
        {
            if constexpr (std::is_void_v<value_type>) return; 
            return handle_.promise().get(); 
        }

explicit operator std::coroutine_handle<>() const noexcept { return handle_; }

private:
        // C-tor
        explicit ResultType(promise_type* promise) noexcept
            : handle_(handle_type::from_promise(*promise))
        {}
      private:
        handle_type handle_{nullptr};
    };

template <bool lazy>
    struct promise_type_base 
    {
        auto initial_suspend() noexcept 
        { 
            if constexpr (lazy) return std::suspend_always{}; 
            else return std::suspend_never{}; 
        }
        std::suspend_always final_suspend() noexcept { return {}; }
        void unhandled_exception() { std::rethrow_exception(std::current_exception()); }
    };

template <typename T, bool lazy>
    struct promise_type_impl : promise_type_base<lazy>
    {
        using value_type = T;
        
        auto get_return_object() { return ResultType<value_type, lazy>{this}; }

template <typename U>
        requires std::is_nothrow_constructible_v<T, U&&>
        void return_value(U&& value) noexcept { value_ = std::forward<U>(value); }
        
        decltype(auto) get() const & { return value_; }
        auto get() const && { return std::move(value_);}

private:
          // use std::expected<value_type, std::exception_ptr> in C++23 instead, for preventing exception rethrow
          value_type value_;

};

template <bool lazy>
    struct promise_type_impl<void, lazy> : promise_type_base<lazy>
    {
        auto get_return_object() { return ResultType<void, lazy>{this}; }
        void return_void() {}
    };

template <typename T>
    using ResultTypeLazy = ResultType<T, true>;
    
}  // namespace utils::coro