Aethera/kernel/kernel/include/time_thread/Timer_Service.cpp

#include "Timer_Service.hpp"

#include "detail/Timer_Scheduler.hpp"
#include "function/frame_policy/rely_facade.h"
#include "model/Model.hpp"

#include <atomic>
#include <condition_variable>
#include <deque>
#include <exception>
#include <memory>
#include <mutex>
#include <thread>
#include <utility>

namespace aethera {
namespace {
struct Steady_Timer_Time_Source : Def<Steady_Timer_Time_Source, Root> {
    struct Private;
    Steady_Timer_Time_Source() = default;
};
struct Steady_Timer_Time_Source::Private : Prev_Private {
    Timer_Time_Point now() const noexcept { return std::chrono::steady_clock::now(); }
};

struct Thread_Timer_Service_State : Immovable {
    struct Impl {
        static constexpr std::chrono::seconds Maximum_Wait{1};

        enum class Command_Type {
            add,
            cancel,
            reschedule,
            stop
        };

        struct Command {
            Command_Type type;                   /* 计时线程消费的操作类型。 */
            Timer_Id id{};                       /* add 时新建、其他操作时定位计时器的标识。 */
            std::chrono::nanoseconds delay{};    /* add/reschedule 相对当前时间的延迟。 */
            std::chrono::nanoseconds interval{}; /* add 周期计时器的周期；零表示单次执行。 */
            Timer_Callback callback{};           /* add 的计时回调或 cancel 的完成回调。 */
        };

        explicit Impl(proxy<Timer_Time_Source> time_source) :
            scheduler(std::move(time_source)) {}

        Timer_Id schedule(
            std::chrono::nanoseconds delay,
            std::chrono::nanoseconds interval,
            Timer_Callback callback) {
            const Timer_Id id =
                next_id.fetch_add(1, std::memory_order_relaxed);
            push(Command{
                Command_Type::add,
                id,
                delay,
                interval,
                std::move(callback)});
            return id;
        }

        void push(Command command) {
            {
                std::lock_guard lock(mutex);
                commands.emplace_back(std::move(command));
            }
            cv.notify_one();
        }

        void apply(Command&& command, bool& stop_requested) {
            switch (command.type) {
            case Command_Type::add:
                scheduler.schedule(
                    command.id,
                    command.delay,
                    command.interval,
                    std::move(command.callback));
                break;
            case Command_Type::cancel:
                scheduler.cancel(command.id);
                if (command.callback) {
                    try {
                        command.callback();
                    }
                    catch (...) {
                        std::terminate();
                    }
                }
                break;
            case Command_Type::reschedule:
                scheduler.reschedule(command.id, command.delay);
                break;
            case Command_Type::stop:
                stop_requested = true;
                break;
            }
        }

        void drain(bool& stop_requested) {
            std::deque<Command> pending;
            {
                std::lock_guard lock(mutex);
                pending.swap(commands);
            }
            for (auto& command : pending) {
                apply(std::move(command), stop_requested);
            }
        }

        /* Timer Service 的专用计时线程允许等待命令或期限；调用方接口只入队，绝不在该等待上同步阻塞。 */
        void run() {
            bool stop_requested = false;
            while (!stop_requested) {
                scheduler.advance_to_current_time();
                drain(stop_requested);
                if (stop_requested) {
                    break;
                }
                if (scheduler.empty()) {
                    std::unique_lock lock(mutex);
                    cv.wait(lock, [this] {
                        return !commands.empty();
                    });
                    continue;
                }
                const auto deadline = scheduler.next_deadline(Maximum_Wait);
                std::unique_lock lock(mutex);
                if (!commands.empty()) {
                    continue;
                }
                cv.wait_until(lock, deadline, [this] {
                    return !commands.empty();
                });
            }
        }

        detail::Timer_Scheduler scheduler; /* 仅由计时线程访问的计时权威状态。 */
        std::atomic<Timer_Id> next_id{1};  /* 跨提交线程分配且不重复的计时器标识。 */
        std::mutex mutex;                  /* 只保护待处理命令队列。 */
        std::condition_variable cv;        /* 命令到达或期限到达时唤醒计时线程。 */
        std::deque<Command> commands;      /* 提交线程向唯一计时线程转移的命令所有权。 */
    };

    explicit Thread_Timer_Service_State(proxy<Timer_Time_Source> time_source) :
        impl(std::make_shared<Impl>(std::move(time_source))) {
        std::thread([state = impl] {
            state->run();
        }).detach();
    }

    ~Thread_Timer_Service_State() {
        impl->push(Impl::Command{Impl::Command_Type::stop});
    }

    Timer_Id schedule_after(
        std::chrono::nanoseconds delay,
        Timer_Callback callback) {
        return impl->schedule(
            delay,
            std::chrono::nanoseconds::zero(),
            std::move(callback));
    }

    Timer_Id schedule_every(
        std::chrono::nanoseconds interval,
        Timer_Callback callback) {
        if (interval <= std::chrono::nanoseconds::zero()) {
            interval = std::chrono::nanoseconds{1};
        }
        return impl->schedule(interval, interval, std::move(callback));
    }

    void reschedule(Timer_Id id, std::chrono::nanoseconds delay) {
        impl->push(Impl::Command{
            Impl::Command_Type::reschedule,
            id,
            delay});
    }

    void cancel(Timer_Id id, Timer_Callback completion) {
        impl->push(Impl::Command{
            Impl::Command_Type::cancel,
            id,
            {},
            {},
            std::move(completion)});
    }

    std::shared_ptr<Impl> impl; /* proxy 与计时线程共享的运行状态所有权。 */
};
struct Thread_Timer_Service : Def<Thread_Timer_Service, Root> {
    struct Private;
    explicit Thread_Timer_Service(proxy<Timer_Time_Source> time_source) : Def(std::move(time_source)) {}
};
struct Thread_Timer_Service::Private : Prev_Private {
    explicit Private(proxy<Timer_Time_Source> time_source) : state(std::move(time_source)) {}
    Timer_Id schedule_after(std::chrono::nanoseconds delay, Timer_Callback callback) { return state.schedule_after(delay, std::move(callback)); }
    Timer_Id schedule_every(std::chrono::nanoseconds interval, Timer_Callback callback) { return state.schedule_every(interval, std::move(callback)); }
    void reschedule(Timer_Id id, std::chrono::nanoseconds delay) { state.reschedule(id, delay); }
    void cancel(Timer_Id id, Timer_Callback completion) { state.cancel(id, std::move(completion)); }
    Thread_Timer_Service_State state;
};
}

proxy<Timer_Time_Source> make_steady_timer_time_source() { return make_model_proxy<Timer_Time_Source, Steady_Timer_Time_Source>(); }

std::expected<proxy<Timer_Service>, Make_Timer_Service_Result>
make_timer_service(proxy<Timer_Time_Source> time_source) {
    if (!time_source) {
        return std::unexpected(
            Make_Timer_Service_Result::time_source_unavailable);
    }
    return make_model_proxy_shared<Timer_Service, Thread_Timer_Service>(std::move(time_source));
}
proxy<frame_policy::Timer_Service> frame_policy::make_timer_service() {
    return make_model_proxy_shared<frame_policy::Timer_Service, Thread_Timer_Service>(make_steady_timer_time_source());
}
}