task_runner.hpp

#pragma once
 
#include "fly/fly.hpp"
#include "fly/task/types.hpp"
 
#include <chrono>
#include <cstdint>
#include <functional>
#include <memory>
#include <mutex>
#include <queue>
#include <type_traits>
 
#define FROM_HERE                                                                                  \
    fly::task::TaskLocation({__FILE__, __FUNCTION__, static_cast<std::uint32_t>(__LINE__)})
 
namespace fly::task {
 
class TaskManager;
 
class TaskRunner : public std::enable_shared_from_this<TaskRunner>
{
    friend class TaskManager;
 
public:
    virtual ~TaskRunner() = default;
 
    template <typename TaskType>
    bool post_task(TaskLocation &&location, TaskType &&task);
 
    template <typename OwnerType, typename TaskType>
    bool post_task(TaskLocation &&location, std::weak_ptr<OwnerType> weak_owner, TaskType &&task);
 
    template <typename TaskType, typename ReplyType>
    bool post_task_with_reply(TaskLocation &&location, TaskType &&task, ReplyType &&reply);
 
    template <typename OwnerType, typename TaskType, typename ReplyType>
    bool post_task_with_reply(
        TaskLocation &&location,
        std::weak_ptr<OwnerType> weak_owner,
        TaskType &&task,
        ReplyType &&reply);
 
    template <typename TaskType>
    bool
    post_task_with_delay(TaskLocation &&location, std::chrono::milliseconds delay, TaskType &&task);
 
    template <typename OwnerType, typename TaskType>
    bool post_task_with_delay(
        TaskLocation &&location,
        std::weak_ptr<OwnerType> weak_owner,
        std::chrono::milliseconds delay,
        TaskType &&task);
 
    template <typename TaskType, typename ReplyType>
    bool post_task_with_delay_and_reply(
        TaskLocation &&location,
        std::chrono::milliseconds delay,
        TaskType &&task,
        ReplyType &&reply);
 
    template <typename OwnerType, typename TaskType, typename ReplyType>
    bool post_task_with_delay_and_reply(
        TaskLocation &&location,
        std::weak_ptr<OwnerType> weak_owner,
        std::chrono::milliseconds delay,
        TaskType &&task,
        ReplyType &&reply);
 
protected:
    TaskRunner(std::shared_ptr<TaskManager> task_manager) noexcept;
 
    virtual bool post_task_internal(TaskLocation &&location, Task &&task) = 0;
 
    virtual void task_complete(TaskLocation &&location) = 0;
 
    bool post_task_to_task_manager(TaskLocation &&location, Task &&task);
 
    bool post_task_to_task_manager_with_delay(
        TaskLocation &&location,
        std::chrono::milliseconds delay,
        Task &&task);
 
private:
    template <typename TaskType>
    struct TaskHolder
    {
        TaskType m_task;
    };
 
    template <typename TaskType>
    Task wrap_task(TaskType &&task);
 
    template <typename OwnerType, typename TaskType>
    Task wrap_task(std::weak_ptr<OwnerType> weak_owner, TaskType &&task);
 
    template <typename TaskType, typename ReplyType>
    Task wrap_task(TaskType &&task, ReplyType &&reply);
 
    template <typename OwnerType, typename TaskType, typename ReplyType>
    Task wrap_task(std::weak_ptr<OwnerType> weak_owner, TaskType &&task, ReplyType &&reply);
 
    void execute(TaskLocation &&location, Task &&task);
 
    std::weak_ptr<TaskManager> m_weak_task_manager;
};
 
class ParallelTaskRunner : public TaskRunner
{
public:
    static std::shared_ptr<ParallelTaskRunner> create(std::shared_ptr<TaskManager> task_manager);
 
protected:
    explicit ParallelTaskRunner(std::shared_ptr<TaskManager> task_manager) noexcept;
 
    bool post_task_internal(TaskLocation &&location, Task &&task) override;
 
    void task_complete(TaskLocation &&location) override;
};
 
class SequencedTaskRunner : public TaskRunner
{
public:
    static std::shared_ptr<SequencedTaskRunner> create(std::shared_ptr<TaskManager> task_manager);
 
protected:
    explicit SequencedTaskRunner(std::shared_ptr<TaskManager> task_manager) noexcept;
 
    bool post_task_internal(TaskLocation &&location, Task &&task) override;
 
    void task_complete(TaskLocation &&location) override;
 
private:
    struct PendingTask
    {
        TaskLocation m_location;
        Task m_task;
    };
 
    bool maybe_post_task(TaskLocation &&location, Task &&task);
 
    std::mutex m_pending_tasks_mutex;
    std::queue<PendingTask> m_pending_tasks;
    bool m_has_running_task {false};
};
 
//==================================================================================================
template <typename TaskType>
bool TaskRunner::post_task(TaskLocation &&location, TaskType &&task)
{
    return post_task_internal(std::move(location), wrap_task(std::forward<TaskType>(task)));
}
 
//==================================================================================================
template <typename OwnerType, typename TaskType>
bool TaskRunner::post_task(
    TaskLocation &&location,
    std::weak_ptr<OwnerType> weak_owner,
    TaskType &&task)
{
    return post_task_internal(
        std::move(location),
        wrap_task(std::move(weak_owner), std::forward<TaskType>(task)));
}
 
//==================================================================================================
template <typename TaskType, typename ReplyType>
bool TaskRunner::post_task_with_reply(TaskLocation &&location, TaskType &&task, ReplyType &&reply)
{
    return post_task_internal(
        std::move(location),
        wrap_task(std::forward<TaskType>(task), std::forward<ReplyType>(reply)));
}
 
//==================================================================================================
template <typename OwnerType, typename TaskType, typename ReplyType>
bool TaskRunner::post_task_with_reply(
    TaskLocation &&location,
    std::weak_ptr<OwnerType> weak_owner,
    TaskType &&task,
    ReplyType &&reply)
{
    return post_task_internal(
        std::move(location),
        wrap_task(
            std::move(weak_owner),
            std::forward<TaskType>(task),
            std::forward<ReplyType>(reply)));
}
 
//==================================================================================================
template <typename TaskType>
bool TaskRunner::post_task_with_delay(
    TaskLocation &&location,
    std::chrono::milliseconds delay,
    TaskType &&task)
{
    return post_task_to_task_manager_with_delay(
        std::move(location),
        std::move(delay),
        wrap_task(std::forward<TaskType>(task)));
}
 
//==================================================================================================
template <typename OwnerType, typename TaskType>
bool TaskRunner::post_task_with_delay(
    TaskLocation &&location,
    std::weak_ptr<OwnerType> weak_owner,
    std::chrono::milliseconds delay,
    TaskType &&task)
{
    return post_task_to_task_manager_with_delay(
        std::move(location),
        std::move(delay),
        wrap_task(std::move(weak_owner), std::forward<TaskType>(task)));
}
 
//==================================================================================================
template <typename TaskType, typename ReplyType>
bool TaskRunner::post_task_with_delay_and_reply(
    TaskLocation &&location,
    std::chrono::milliseconds delay,
    TaskType &&task,
    ReplyType &&reply)
{
    return post_task_to_task_manager_with_delay(
        std::move(location),
        std::move(delay),
        wrap_task(std::forward<TaskType>(task), std::forward<ReplyType>(reply)));
}
 
//==================================================================================================
template <typename OwnerType, typename TaskType, typename ReplyType>
bool TaskRunner::post_task_with_delay_and_reply(
    TaskLocation &&location,
    std::weak_ptr<OwnerType> weak_owner,
    std::chrono::milliseconds delay,
    TaskType &&task,
    ReplyType &&reply)
{
    return post_task_to_task_manager_with_delay(
        std::move(location),
        std::move(delay),
        wrap_task(std::move(weak_owner), std::forward<TaskType>(task), std::move(reply)));
}
 
//==================================================================================================
template <typename TaskType>
Task TaskRunner::wrap_task(TaskType &&task)
{
    static_assert(std::is_invocable_v<TaskType>, "Task must be invocable without any arguments");
 
    TaskHolder<TaskType> holder {std::forward<TaskType>(task)};
 
    return [holder = std::move(holder)](TaskRunner *, TaskLocation) mutable {
        FLY_UNUSED(std::invoke(std::move(holder.m_task)));
    };
}
 
//==================================================================================================
template <typename OwnerType, typename TaskType>
Task TaskRunner::wrap_task(std::weak_ptr<OwnerType> weak_owner, TaskType &&task)
{
    using StrongOwnerType = std::shared_ptr<OwnerType>;
 
    static_assert(
        std::is_invocable_v<TaskType, StrongOwnerType>,
        "Task must be invocable with only a strong pointer to its owner");
 
    TaskHolder<TaskType> holder {std::forward<TaskType>(task)};
 
    return [weak_owner = std::move(weak_owner),
            holder = std::move(holder)](TaskRunner *, TaskLocation) mutable {
        if (StrongOwnerType owner = weak_owner.lock(); owner)
        {
            FLY_UNUSED(std::invoke(std::move(holder.m_task), std::move(owner)));
        }
    };
}
 
//==================================================================================================
template <typename TaskType, typename ReplyType>
Task TaskRunner::wrap_task(TaskType &&task, ReplyType &&reply)
{
    static_assert(std::is_invocable_v<TaskType>, "Task must be invocable without any arguments");
 
    using ResultType = std::invoke_result_t<TaskType>;
    static constexpr bool s_result_is_void = std::is_void_v<ResultType>;
 
    static_assert(
        (s_result_is_void && std::is_invocable_v<ReplyType>) ||
            (!s_result_is_void && std::is_invocable_v<ReplyType, ResultType>),
        "Either the task must return a non-void type and the reply must be invocable with only "
        "that type, or the task must return void and the reply must be invocable without any "
        "arguments");
 
    TaskHolder<TaskType> task_holder {std::forward<TaskType>(task)};
    TaskHolder<ReplyType> reply_holder {std::forward<ReplyType>(reply)};
 
    return [task_holder = std::move(task_holder),
            reply_holder =
                std::move(reply_holder)](TaskRunner *runner, TaskLocation location) mutable {
        if constexpr (s_result_is_void)
        {
            std::invoke(std::move(task_holder.m_task));
            runner->post_task(std::move(location), std::move(reply_holder.m_task));
        }
        else
        {
            auto result = std::invoke(std::move(task_holder.m_task));
 
            runner->post_task(
                std::move(location),
                std::bind(std::move(reply_holder.m_task), std::move(result)));
        }
    };
}
 
//==================================================================================================
template <typename OwnerType, typename TaskType, typename ReplyType>
Task TaskRunner::wrap_task(std::weak_ptr<OwnerType> weak_owner, TaskType &&task, ReplyType &&reply)
{
    using StrongOwnerType = std::shared_ptr<OwnerType>;
 
    static_assert(
        std::is_invocable_v<TaskType, StrongOwnerType>,
        "Task must be invocable with only a strong pointer to its owner");
 
    using ResultType = std::invoke_result_t<TaskType, StrongOwnerType>;
    static constexpr bool s_result_is_void = std::is_void_v<ResultType>;
 
    static_assert(
        (s_result_is_void && std::is_invocable_v<ReplyType, StrongOwnerType>) ||
            (!s_result_is_void && std::is_invocable_v<ReplyType, ResultType, StrongOwnerType>),
        "Either the task must return a non-void type and the reply must be invocable with that "
        "type and a strong pointer to its owner, or the task must return void and the reply must "
        "be invocable with only a strong pointer to its owner");
 
    TaskHolder<TaskType> task_holder {std::forward<TaskType>(task)};
    TaskHolder<ReplyType> reply_holder {std::forward<ReplyType>(reply)};
 
    return [weak_owner = std::move(weak_owner),
            task_holder = std::move(task_holder),
            reply_holder =
                std::move(reply_holder)](TaskRunner *runner, TaskLocation location) mutable {
        if (StrongOwnerType owner = weak_owner.lock(); owner)
        {
            if constexpr (s_result_is_void)
            {
                std::invoke(std::move(task_holder.m_task), std::move(owner));
 
                runner->post_task(
                    std::move(location),
                    std::move(weak_owner),
                    std::move(reply_holder.m_task));
            }
            else
            {
                auto result = std::invoke(std::move(task_holder.m_task), std::move(owner));
 
                runner->post_task(
                    std::move(location),
                    std::move(weak_owner),
                    std::bind(
                        std::move(reply_holder.m_task),
                        std::move(result),
                        std::placeholders::_1));
            }
        }
    };
}
 
} // namespace fly::task