task.hpp

// Copyright Take Vos 2022.
// Distributed under the Boost Software License, Version 1.0.
// (See accompanying file LICENSE_1_0.txt or copy at https://www.boost.org/LICENSE_1_0.txt)
 
#pragma once
 
#include "notifier.hpp"
#include "awaitable.hpp"
#include "../utility/utility.hpp"
#include "../concurrency/unfair_mutex.hpp" // XXX #616
#include "../concurrency/thread.hpp" // XXX #616
#include "../macros.hpp"
#include <coroutine>
#include <type_traits>
#include <memory>
#include <exception>
#include <optional>
 
hi_export_module(hikogui.dispatch : task);
 
hi_export namespace hi::inline v1 {
 
template<typename T = void, bool DestroyFrame = false>
class task {
public:
    using value_type = T;
    using notifier_type = notifier<void(value_type)>;
    using callback_type = notifier_type::callback_type;
 
    struct promise_type {
        notifier_type notifier;
        std::optional<value_type> value = {};
        std::exception_ptr exception = nullptr;
 
        template<std::convertible_to<value_type> NewValue>
        void return_value(NewValue&& new_value) noexcept
        {
            value = std::forward<NewValue>(new_value);
        }
 
        void unhandled_exception() noexcept
        {
            exception = std::current_exception();
        }
 
        std::suspend_always final_suspend() noexcept
        {
            if (value) {
                // Trigger the notifier with the co_return value.
                notifier(*value);
 
            } else {
                // Notify also in case of exception.
                hi_assert_not_null(exception);
                if constexpr (std::is_default_constructible_v<value_type>) {
                    notifier(value_type{});
                }
            }
 
            return {};
        }
 
        task get_return_object() noexcept
        {
            return task{handle_type::from_promise(*this)};
        }
 
        std::suspend_never initial_suspend() noexcept
        {
            return {};
        }
 
        template<convertible_to_awaitable RHS>
        decltype(auto) await_transform(RHS&& rhs)
        {
            return awaitable_cast<std::remove_cvref_t<RHS>>{}(std::forward<RHS>(rhs));
        }
    };
 
    using handle_type = std::coroutine_handle<promise_type>;
 
    task(handle_type coroutine) noexcept : _coroutine(coroutine) {}
 
    ~task()
    {
        if (DestroyFrame and _coroutine) {
            _coroutine.destroy();
        }
    }
 
    task() = default;
 
    // task can not be copied because it tracks if the co-routine must be destroyed by the
    // shared_ptr to the value shared between task and the promise.
    task(task const&) = delete;
    task& operator=(task const&) = delete;
 
    task(task&& other) noexcept
    {
        _coroutine = std::exchange(other._coroutine, {});
    }
 
    task& operator=(task&& other) noexcept
    {
        _coroutine = std::exchange(other._coroutine, {});
        return *this;
    }
 
    [[nodiscard]] bool started() const noexcept
    {
        return _coroutine;
    }
 
    [[nodiscard]] bool running() const noexcept
    {
        return _coroutine and not _coroutine.done();
    }
 
    [[nodiscard]] bool done() const noexcept
    {
        return _coroutine and _coroutine.done();
    }
 
    [[nodiscard]] value_type const& value() const
    {
        hi_axiom(done());
 
        auto const& promise = _coroutine.promise();
        if (promise.value) {
            return *promise.value;
 
        } else {
            hi_axiom(promise.exception);
            std::rethrow_exception(promise.exception);
        }
    }
 
    [[nodiscard]] value_type const& operator*() const
    {
        return value();
    }
 
    template<forward_of<void(value_type)> Func>
    [[nodiscard]] callback<void(value_type)> subscribe(Func &&func, callback_flags flags = callback_flags::synchronous) noexcept
    {
        return _coroutine.promise().notifier.subscribe(std::forward<Func>(func), flags);
    }
 
    auto operator co_await() const noexcept
    {
        return _coroutine.promise().notifier.operator co_await();
    }
 
private:
    // Optional value type
    handle_type _coroutine;
};
 
template<bool DestroyFrame>
class task<void, DestroyFrame> {
public:
    using value_type = void;
    using notifier_type = notifier<void()>;
    using callback_type = notifier_type::callback_type;
 
    struct promise_type {
        notifier_type notifier;
        std::exception_ptr exception = nullptr;
 
        void return_void() noexcept {}
 
        void unhandled_exception() noexcept
        {
            exception = std::current_exception();
        }
 
        std::suspend_always final_suspend() noexcept
        {
            notifier();
            return {};
        }
 
        task get_return_object() noexcept
        {
            return task{handle_type::from_promise(*this)};
        }
 
        std::suspend_never initial_suspend() noexcept
        {
            return {};
        }
 
        template<convertible_to_awaitable RHS>
        decltype(auto) await_transform(RHS&& rhs)
        {
            return awaitable_cast<std::remove_cvref_t<RHS>>{}(std::forward<RHS>(rhs));
        }
    };
 
    using handle_type = std::coroutine_handle<promise_type>;
 
    task(handle_type coroutine) noexcept : _coroutine(coroutine) {}
 
    ~task()
    {
        if (DestroyFrame and _coroutine) {
            _coroutine.destroy();
        }
    }
 
    task() = default;
    task(task const&) = delete;
    task& operator=(task const&) = delete;
 
    task(task&& other) noexcept
    {
        _coroutine = std::exchange(other._coroutine, {});
    }
 
    task& operator=(task&& other) noexcept
    {
        _coroutine = std::exchange(other._coroutine, {});
        return *this;
    }
 
    [[nodiscard]] bool started() const noexcept
    {
        return _coroutine;
    }
 
    [[nodiscard]] bool running() const noexcept
    {
        return _coroutine and not _coroutine.done();
    }
 
    [[nodiscard]] bool done() const noexcept
    {
        return _coroutine and _coroutine.done();
    }
 
    void value() const
    {
        hi_axiom(done());
 
        auto const& promise = _coroutine.promise();
        if (promise.exception) {
            std::rethrow_exception(promise.exception);
        }
    }
 
    template<forward_of<void()> Func>
    [[nodiscard]] callback<void()> subscribe(Func &&func, callback_flags flags = callback_flags::synchronous) noexcept
    {
        return _coroutine.promise().notifier.subscribe(std::forward<Func>(func), flags);
    }
 
    auto operator co_await() const noexcept
    {
        return _coroutine.promise().notifier.operator co_await();
    }
 
private:
    // Optional value type
    handle_type _coroutine;
};
 
template<typename T = void>
using scoped_task = task<T, true>;
 
template<typename T>
struct is_task : std::false_type {};
 
template<typename ResultType, bool DestroyFrame>
struct is_task<hi::task<ResultType, DestroyFrame>> : std::true_type {};
 
template<typename T>
constexpr bool is_task_v = is_task<T>::value;
 
template<typename T>
struct task_value_type;
 
template<typename ResultType, bool DestroyFrame>
struct task_value_type<hi::task<ResultType, DestroyFrame>> {
    using type = ResultType;
};
 
template<typename T>
using task_value_type_t = task_value_type<T>::type;
 
template<typename Func, typename... ArgTypes>
struct is_invocable_task {
    constexpr static bool value = is_task_v<std::invoke_result_t<Func, ArgTypes...>>;
};
 
template<typename Func, typename... ArgTypes>
constexpr bool is_invocable_task_v = is_invocable_task<Func, ArgTypes...>::value;
 
template<typename Func, typename... ArgTypes>
struct invoke_task_result {
    using type = task_value_type_t<std::invoke_result_t<Func, ArgTypes...>>;
};
 
template<typename Func, typename... ArgTypes>
using invoke_task_result_t = invoke_task_result<Func, ArgTypes...>::type;
 
} // namespace hi::inline v1