diff --git a/AGENTS.md b/AGENTS.md index a8421e7..b71b642 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,7 +1,5 @@ ## 设计约束 -复杂的对象 都使用 D:\ae\proj\Aethera\kernel\src\kernel\double_buffer\model.hpp 并且使用其提供的线程模型 - `` 除非特别说明不允许写任何同步等待代码。 除非特别说明不允许写任何同步等待代码。 diff --git a/kernel/kernel/include/frame_policy/design.md b/kernel/kernel/include/frame_policy/design.md deleted file mode 100644 index fb9128d..0000000 --- a/kernel/kernel/include/frame_policy/design.md +++ /dev/null @@ -1,388 +0,0 @@ -有了: - -```cpp -schedule_after(...) -schedule_every(...) -reschedule(...) -``` - -你这个异步图表库的帧策略就很好做了。我觉得至少有 5 种值得保留,而且可以统一到同一套 `Frame_Policy` 上。 - -### 1. 固定帧率 + latest-only - -最适合动画、交互拖动、持续变化场景。 - -```cpp -timer = timers.schedule_every(16.666ms, [this] { - request_render(); -}); -``` - -比如 60 FPS: - -```text -Producer: -状态一直更新 -1 2 3 4 5 6 7 8 9 ... - -Timer: -|----16.67----|----16.67----|----16.67----| - -Render: - latest latest latest -``` - -每个 tick: - -```cpp -void on_frame_tick() -{ - if(rendering) { - return; - } - if(!dirty) { - return; - } - start_render(); -} -``` - -特点是: - -```text -帧率稳定 -不会积压 -旧状态直接被覆盖 -render 慢了就掉帧 -不会延迟越来越大 -``` - -这个应该是你的 **默认实时模式**。 - ---- - -### 2. 变化立即渲染 + 最大 FPS 限制 - -这个我觉得特别适合普通图表。 - -例如最大 120 FPS: - -```text -状态变化 - ↓ -如果距离上一帧已经 > 8.33ms - ↓ -立即 render - -否则 - ↓ -schedule_after(剩余时间) -``` - -逻辑大概: - -```cpp -void request_render() -{ - dirty = true; - - if(rendering || frame_pending) { - return; - } - - auto now = Clock::now(); - auto elapsed = now - last_render; - - if(elapsed >= min_frame_interval) { - start_render(); - } else { - frame_pending = true; - frame_timer = timers.schedule_after( - min_frame_interval - elapsed, - [this] { - frame_pending = false; - if(dirty && !rendering) { - start_render(); - } - } - ); - } -} -``` - -比如用户只偶尔: - -```text -set_data() -``` - -不会傻等下一个固定 16.67ms tick,而是: - -```text -空闲很久 - ↓ -set_data - ↓ -立即渲染 -``` - -如果疯狂变化: - -```text -set_data set_data set_data set_data... -``` - -又自动被限制在: - -```text -≤ 120 FPS -``` - -这个兼顾: - -**低延迟 + 限制资源占用。** - -我认为这个非常适合做你的默认 `Auto` 模式。 - ---- - -### 3. Debounce 静止后渲染 - -直接发挥 `reschedule()` 的优势。 - -例如: - -```cpp -void on_change() -{ - dirty = true; - timers.reschedule(render_timer, 20ms); -} -``` - -连续变化: - -```text -change - ↓20ms -change - ↓20ms -change - ↓20ms -change - ↓20ms -最终静止 - ↓20ms -render -``` - -适合: - -```text -窗口 resize -大批量属性连续设置 -布局参数连续修改 -一次 transaction 内多次 set_xxx() -``` - -但不适合鼠标拖动这种要求实时视觉反馈的场景,因为它可能一直不渲染。 - -所以它应该叫类似: - -```cpp -Frame_Mode::Debounce -``` - -而不是普通实时模式。 - ---- - -### 4. 立即首帧 + 后续合并 - -这个对于 UI/图表体验很好。 - -第一次变化: - -```text -change - ↓ -立即 render -``` - -紧接着的大量变化: - -```text -change change change change - ↓ - 合并一段时间 - ↓ - render latest -``` - -例如: - -```cpp -void request_render() -{ - dirty = true; - - if(!cooldown) { - start_render(); - - cooldown = true; - timer = timers.schedule_after(16ms, [this] { - cooldown = false; - - if(dirty) { - request_render(); - } - }); - } -} -``` - -效果: - -```text -事件 X XXXXXXXX XXXXXX -渲染 R--------R--------R -``` - -第一帧延迟非常低,同时限制后面帧率。 - -这种经常比纯: - -```cpp -schedule_every(16ms) -``` - -用户体验更好,因为不会出现“刚好错过 tick,要额外等 16ms”。 - ---- - -### 5. 动态帧率 / Adaptive - -这是最终可以做得很漂亮的一种。 - -比如: - -```text -有动画 / 拖动 → 120 FPS -普通数据变化 → 60 FPS -几秒没变化 → 不跑 timer -后台低优先级刷新 → 10 FPS -``` - -状态机: - -```text -Idle - │ - │ change - ▼ -Interactive - 120 FPS - │ - │ 一段时间无高频输入 - ▼ -Normal - 60 FPS - │ - │ 长时间不变 - ▼ -Idle - timer取消 -``` - -这里 `schedule_every` 和 `reschedule` 正好可以配合。 - -甚至可以根据 render duration 自动降帧: - -```cpp -if(render_time > 16ms) { - target_fps = 30; -} else if(render_time < 6ms) { - target_fps = 60; -} -``` - -但这里要加迟滞,不能: - -```text -60 → 30 → 60 → 30 -``` - -疯狂抖。 - ---- - -我会给你的库最终留这几个策略: - -| 策略 | 行为 | 适合 | -|-------------|--------------------------|--------------------------| -| `Immediate` | 有变化立即 render | 截图、离线、低频更新 | -| `Fixed` | 固定 tick 取 latest | 动画、稳定 FPS | -| `Throttled` | 立即响应,但限制最大 FPS | **普通交互图表,最推荐** | -| `Debounce` | 静止一段时间才 render | resize、批量修改 | -| `Adaptive` | 动态 120/60/30/Idle | 高级实时模式 | - -如果是你的图表库,我会把默认策略设成: - -```text -Throttled + Latest-only -``` - -即: - -```text -第一帧尽快 -+ -最大 FPS 限制 -+ -render 期间的新变化只标 dirty -+ -render 完成直接看 latest -+ -绝不排 render 队列 -``` - -这几个性质放在一起非常重要。 - -最终状态实际上只需要: - -```cpp -bool dirty; -bool rendering; -bool frame_pending; - -Time_Point last_render; -Timer_Id frame_timer; - -Duration min_frame_interval; -``` - -整个核心流程就是: - -```text -任意线程 set_xxx() - ↓ - dirty - ↓ - request_render - ↓ -┌────────────────────┐ -│ 正在 render? → 等 │ -│ 未到帧间隔? → timer│ -│ 否则 → 立即 render │ -└────────────────────┘ - ↓ -render complete - ↓ -dirty? - ↓ ↓ -yes no - ↓ -重新走 request_render -``` - -这套特别适合你现在这种 **没有主线程、任意线程调用、异步渲染**的架构。 diff --git a/kernel/kernel/include/function/Scene.hpp b/kernel/kernel/include/function/Scene.hpp new file mode 100644 index 0000000..2342295 --- /dev/null +++ b/kernel/kernel/include/function/Scene.hpp @@ -0,0 +1,2 @@ +#pragma once +struct Scene {}; diff --git a/kernel/kernel/include/function/frame_policy/detail/Throttled_Latest_Only_State.cpp b/kernel/kernel/include/function/frame_policy/detail/Throttled_Latest_Only_State.cpp new file mode 100644 index 0000000..71e8168 --- /dev/null +++ b/kernel/kernel/include/function/frame_policy/detail/Throttled_Latest_Only_State.cpp @@ -0,0 +1,287 @@ +#include "Throttled_Latest_Only_State.ipp" + +#include +#include + +namespace aethera::detail { +namespace { +template +struct Scope_Exit { + explicit Scope_Exit(Function function) : + function(std::move(function)) {} + ~Scope_Exit() noexcept { + if (active) { + function(); + } + } + Scope_Exit(const Scope_Exit&) = delete; + Scope_Exit& operator=(const Scope_Exit&) = delete; + void release() noexcept { + active = false; + } + + Function function; /* 离开当前作用域时执行的异常安全清理。 */ + bool active{true}; /* release 后不再执行清理。 */ +}; +} + +Throttled_Latest_Only_State::Throttled_Latest_Only_State( + proxy scene, + proxy sink) : + d(std::make_unique()) { + d->scene = std::move(scene); + d->sink = std::move(sink); +} + +Throttled_Latest_Only_State::~Throttled_Latest_Only_State() = default; + +Start_Throttled_Latest_Only_Result +Throttled_Latest_Only_State::prepare_start() { + { + std::lock_guard lock(d->mutex); + switch (d->phase) { + case Private::Phase::running: + return Start_Throttled_Latest_Only_Result::already_running; + case Private::Phase::starting: + return Start_Throttled_Latest_Only_Result::start_in_progress; + case Private::Phase::stopping: + return Start_Throttled_Latest_Only_Result::stop_in_progress; + case Private::Phase::stopped: + break; + } + if (!d->scene || !d->sink) { + return Start_Throttled_Latest_Only_Result::dependency_unavailable; + } + d->phase = Private::Phase::starting; + } + + Scope_Exit rollback([this] { + std::lock_guard lock(d->mutex); + if (d->phase == Private::Phase::starting) { + d->phase = Private::Phase::stopped; + } + }); + auto frame = d->scene->create_frame(); + if (!frame) { + return Start_Throttled_Latest_Only_Result::frame_unavailable; + } + + { + std::lock_guard lock(d->mutex); + if (d->phase != Private::Phase::starting || d->frame) { + std::terminate(); + } + d->frame = std::move(frame); + d->frame_phase = Private::Frame_Phase::idle; + d->timer_cancelled = false; + } + rollback.release(); + return Start_Throttled_Latest_Only_Result::started; +} + +void Throttled_Latest_Only_State::timer_started(Timer_Id id) { + std::lock_guard lock(d->mutex); + if (id == 0 || + d->phase != Private::Phase::starting || + !d->frame || + d->timer_id.has_value()) { + std::terminate(); + } + d->timer_id = id; + d->phase = Private::Phase::running; +} + +void Throttled_Latest_Only_State::start_failed() { + proxy retired_frame; + { + std::lock_guard lock(d->mutex); + if (d->phase != Private::Phase::starting || d->timer_id) { + std::terminate(); + } + retired_frame = std::move(d->frame); + d->frame_phase = Private::Frame_Phase::idle; + d->phase = Private::Phase::stopped; + } + retired_frame.reset(); +} + +Stop_Throttled_Latest_Only_Request +Throttled_Latest_Only_State::request_stop( + Throttled_Latest_Only_Stop_Completion completion) { + if (!completion) { + return { + Stop_Throttled_Latest_Only_Result::completion_missing, + std::nullopt}; + } + + std::lock_guard lock(d->mutex); + switch (d->phase) { + case Private::Phase::stopped: + return { + Stop_Throttled_Latest_Only_Result::already_stopped, + std::nullopt}; + case Private::Phase::starting: + return { + Stop_Throttled_Latest_Only_Result::start_in_progress, + std::nullopt}; + case Private::Phase::stopping: + return { + Stop_Throttled_Latest_Only_Result::already_stopping, + std::nullopt}; + case Private::Phase::running: + break; + } + if (!d->timer_id) { + std::terminate(); + } + d->phase = Private::Phase::stopping; + d->stop_completion = std::move(completion); + d->timer_cancelled = false; + return { + Stop_Throttled_Latest_Only_Result::stopping, + d->timer_id}; +} + +void Throttled_Latest_Only_State::cancel_failed() { + std::lock_guard lock(d->mutex); + if (d->phase != Private::Phase::stopping || + d->timer_cancelled) { + std::terminate(); + } + d->stop_completion = {}; + d->phase = Private::Phase::running; +} + +void Throttled_Latest_Only_State::timer_cancelled() { + { + std::lock_guard lock(d->mutex); + if (d->phase != Private::Phase::stopping || + d->timer_cancelled) { + std::terminate(); + } + d->timer_id.reset(); + d->timer_cancelled = true; + } + finish_stop_if_ready(); +} + +void Throttled_Latest_Only_State::frame_due() { + { + std::lock_guard lock(d->mutex); + if (d->phase != Private::Phase::running || + d->frame_phase != Private::Frame_Phase::idle) { + return; + } + d->frame_phase = Private::Frame_Phase::rendering; + } + + auto self = shared_from_this(); + try { + d->scene->render( + d->frame, + [self = std::move(self)](proxy& frame) { + self->rendered(frame); + }); + } + catch (...) { + abandon_render(); + throw; + } +} + +void Throttled_Latest_Only_State::rendered( + proxy& completed_frame) { + { + std::lock_guard lock(d->mutex); + if (std::addressof(completed_frame) != + std::addressof(d->frame) || + d->frame_phase != Private::Frame_Phase::rendering) { + std::terminate(); + } + d->frame_phase = Private::Frame_Phase::sending; + } + + auto self = shared_from_this(); + try { + d->sink->send( + d->frame, + [self = std::move(self)](proxy& frame) { + self->sent(frame); + }); + } + catch (...) { + abandon_send(); + throw; + } +} + +void Throttled_Latest_Only_State::sent( + proxy& completed_frame) { + { + std::lock_guard lock(d->mutex); + if (std::addressof(completed_frame) != + std::addressof(d->frame) || + d->frame_phase != Private::Frame_Phase::sending) { + std::terminate(); + } + d->frame_phase = Private::Frame_Phase::idle; + } + finish_stop_if_ready(); +} + +void Throttled_Latest_Only_State::abandon_render() { + { + std::lock_guard lock(d->mutex); + if (d->frame_phase != Private::Frame_Phase::rendering) { + return; + } + d->frame_phase = Private::Frame_Phase::idle; + } + finish_stop_if_ready(); +} + +void Throttled_Latest_Only_State::abandon_send() { + { + std::lock_guard lock(d->mutex); + if (d->frame_phase != Private::Frame_Phase::sending) { + return; + } + d->frame_phase = Private::Frame_Phase::idle; + } + finish_stop_if_ready(); +} + +void Throttled_Latest_Only_State::finish_stop_if_ready() { + proxy retired_frame; + Throttled_Latest_Only_Stop_Completion completion; + { + std::lock_guard lock(d->mutex); + if (d->phase != Private::Phase::stopping || + !d->timer_cancelled || + d->frame_phase != Private::Frame_Phase::idle) { + return; + } + retired_frame = std::move(d->frame); + completion = std::move(d->stop_completion); + d->timer_cancelled = false; + d->phase = Private::Phase::stopped; + } + + retired_frame.reset(); + try { + completion(); + } + catch (...) { + std::terminate(); + } +} + +bool Throttled_Latest_Only_State::destructible() const noexcept { + std::lock_guard lock(d->mutex); + return d->phase == Private::Phase::stopped && + d->frame_phase == Private::Frame_Phase::idle && + !d->frame && + !d->timer_id && + !d->stop_completion; +} +} diff --git a/kernel/kernel/include/function/frame_policy/detail/Throttled_Latest_Only_State.hpp b/kernel/kernel/include/function/frame_policy/detail/Throttled_Latest_Only_State.hpp new file mode 100644 index 0000000..58b054b --- /dev/null +++ b/kernel/kernel/include/function/frame_policy/detail/Throttled_Latest_Only_State.hpp @@ -0,0 +1,40 @@ +#pragma once + +#include "function/frame_policy/global.hpp" +#include "time_thread/Timer_Service.hpp" + +#include +#include + +namespace aethera::detail { +struct Stop_Throttled_Latest_Only_Request { + Stop_Throttled_Latest_Only_Result result; /* 本次停止请求的已知结果。 */ + std::optional timer_id; /* stopping 时必须异步取消的计时器。 */ +}; + +struct Throttled_Latest_Only_State : + Non_Copyable, + std::enable_shared_from_this { + Throttled_Latest_Only_State(proxy scene, + proxy sink); + ~Throttled_Latest_Only_State(); + + Start_Throttled_Latest_Only_Result prepare_start(); + void timer_started(Timer_Id id); + void start_failed(); + Stop_Throttled_Latest_Only_Request request_stop( + Throttled_Latest_Only_Stop_Completion completion); + void cancel_failed(); + void timer_cancelled(); + void frame_due(); + [[nodiscard]] bool destructible() const noexcept; +private: + struct Private; + void rendered(proxy& completed_frame); + void sent(proxy& completed_frame); + void abandon_render(); + void abandon_send(); + void finish_stop_if_ready(); + std::unique_ptr d; +}; +} diff --git a/kernel/kernel/include/function/frame_policy/detail/Throttled_Latest_Only_State.ipp b/kernel/kernel/include/function/frame_policy/detail/Throttled_Latest_Only_State.ipp new file mode 100644 index 0000000..b6e7968 --- /dev/null +++ b/kernel/kernel/include/function/frame_policy/detail/Throttled_Latest_Only_State.ipp @@ -0,0 +1,31 @@ +#pragma once + +#include "Throttled_Latest_Only_State.hpp" + +#include + +namespace aethera::detail { +struct Throttled_Latest_Only_State::Private { + enum struct Phase : std::uint8_t { + stopped, + starting, + running, + stopping + }; + enum struct Frame_Phase : std::uint8_t { + idle, + rendering, + sending + }; + + proxy scene; /* 帧类型和渲染借用的业务实现所有权。 */ + proxy sink; /* 当前帧媒体分派的业务实现所有权。 */ + proxy frame; /* 策略创建、复用并在停止完成前销毁的唯一帧。 */ + Throttled_Latest_Only_Stop_Completion stop_completion; /* 本次异步停止的唯一完成回调。 */ + std::optional timer_id; /* running/stopping 阶段的周期计时器标识。 */ + mutable std::mutex mutex; /* 保护生命周期和跨线程帧借用阶段。 */ + Phase phase{Phase::stopped}; /* start/stop 生命周期的唯一权威状态。 */ + Frame_Phase frame_phase{Frame_Phase::idle}; /* 唯一帧当前借用方的权威状态。 */ + bool timer_cancelled{}; /* stopping 阶段的异步取消确认。 */ +}; +} diff --git a/kernel/kernel/include/function/frame_policy/global.cpp b/kernel/kernel/include/function/frame_policy/global.cpp new file mode 100644 index 0000000..7d5c318 --- /dev/null +++ b/kernel/kernel/include/function/frame_policy/global.cpp @@ -0,0 +1,87 @@ +#include "global.ipp" + +#include "time_thread/Timer_Service.hpp" + +#include +#include + +namespace aethera { +Throttled_Latest_only::Private::Private( + proxy scene, + proxy sink) : + state(std::make_shared( + std::move(scene), + std::move(sink))) {} + +Throttled_Latest_only::Throttled_Latest_only( + proxy scene, + proxy sink) : + d(std::make_unique(std::move(scene), std::move(sink))) {} + +Throttled_Latest_only::~Throttled_Latest_only() noexcept { + if (!d->state->destructible()) { + std::terminate(); + } +} + +Start_Throttled_Latest_Only_Result +Throttled_Latest_only::start(double frames_per_second) { + const auto interval = + detail::timer_interval_from_frames_per_second(frames_per_second); + if (!interval) { + return interval.error() == + Schedule_Every_Fps_Result::invalid_frames_per_second + ? Start_Throttled_Latest_Only_Result::invalid_frames_per_second + : Start_Throttled_Latest_Only_Result::interval_out_of_range; + } + + const auto prepared = d->state->prepare_start(); + if (prepared != Start_Throttled_Latest_Only_Result::started) { + return prepared; + } + + const auto state = d->state; + Timer_Id timer_id{}; + try { + timer_id = + Timer_Service::instance().schedule_every( + *interval, + [state] { + state->frame_due(); + }); + } + catch (...) { + state->start_failed(); + throw; + } + state->timer_started(timer_id); + return Start_Throttled_Latest_Only_Result::started; +} + +Stop_Throttled_Latest_Only_Result +Throttled_Latest_only::stop( + Throttled_Latest_Only_Stop_Completion completion) { + auto request = d->state->request_stop(std::move(completion)); + if (request.result != + Stop_Throttled_Latest_Only_Result::stopping) { + return request.result; + } + if (!request.timer_id) { + std::terminate(); + } + + try { + const auto state = d->state; + Timer_Service::instance().cancel( + *request.timer_id, + [state] { + state->timer_cancelled(); + }); + } + catch (...) { + d->state->cancel_failed(); + throw; + } + return Stop_Throttled_Latest_Only_Result::stopping; +} +} diff --git a/kernel/kernel/include/function/frame_policy/global.hpp b/kernel/kernel/include/function/frame_policy/global.hpp new file mode 100644 index 0000000..fd2011d --- /dev/null +++ b/kernel/kernel/include/function/frame_policy/global.hpp @@ -0,0 +1,60 @@ +#pragma once +#include "../../global.hpp" +#include +#include +#include +#include +namespace aethera { +using Time_Type = std::uint64_t; +PRO_DEF_MEM_DISPATCH(FP_Frame_use_time, use_time); +struct FP_Frame : facade_builder +::add_convention +::build {}; +using FP_Frame_Completion = +std::function&)>; +PRO_DEF_MEM_DISPATCH(FP_create_frame, create_frame); +PRO_DEF_MEM_DISPATCH(FP_render, render); +struct FP_Scene : facade_builder +::add_convention()> +::add_convention&, FP_Frame_Completion)> +::build {}; +PRO_DEF_MEM_DISPATCH(FP_send, send); +struct FP_Sink : facade_builder +::add_convention&, FP_Frame_Completion)> +::build {}; +enum struct Start_Throttled_Latest_Only_Result : std::uint8_t { + started, + already_running, + start_in_progress, + stop_in_progress, + dependency_unavailable, + frame_unavailable, + invalid_frames_per_second, + interval_out_of_range +}; +enum struct Stop_Throttled_Latest_Only_Result : std::uint8_t { + stopping, + already_stopped, + already_stopping, + start_in_progress, + completion_missing +}; +using Throttled_Latest_Only_Stop_Completion = +std::function; +/* + * 固定帧率、最多一帧在途的策略。Scene 创建具体帧;策略持有并重复使用, + * stop completion 执行前会排空 Scene/Sink 借用并销毁该帧。 + */ +struct Throttled_Latest_only : Immovable { + Throttled_Latest_only(proxy scene, + proxy sink); + ~Throttled_Latest_only() noexcept; + Start_Throttled_Latest_Only_Result start( + double frames_per_second); + Stop_Throttled_Latest_Only_Result stop( + Throttled_Latest_Only_Stop_Completion completion); +private: + struct Private; + std::unique_ptr d; /* 策略外壳实现的唯一所有权。 */ +}; +} diff --git a/kernel/kernel/include/function/frame_policy/global.ipp b/kernel/kernel/include/function/frame_policy/global.ipp new file mode 100644 index 0000000..c4247cc --- /dev/null +++ b/kernel/kernel/include/function/frame_policy/global.ipp @@ -0,0 +1,13 @@ +#pragma once + +#include "global.hpp" +#include "detail/Throttled_Latest_Only_State.hpp" + +namespace aethera { +struct Throttled_Latest_only::Private { + Private(proxy scene, + proxy sink); + + std::shared_ptr state; /* 帧、借用阶段和停止排空的共享所有权。 */ +}; +} diff --git a/kernel/kernel/include/function/renderable.hpp b/kernel/kernel/include/function/renderable.hpp new file mode 100644 index 0000000..6f70f09 --- /dev/null +++ b/kernel/kernel/include/function/renderable.hpp @@ -0,0 +1 @@ +#pragma once diff --git a/kernel/kernel/include/global.hpp b/kernel/kernel/include/global.hpp index d339d29..7b8141a 100644 --- a/kernel/kernel/include/global.hpp +++ b/kernel/kernel/include/global.hpp @@ -7,7 +7,10 @@ #include #include #include +#include "proxy/v4/detail/facade_creation.h" namespace aethera { +using pro::facade_builder; +using pro::proxy; struct Non_Copyable { protected: Non_Copyable() = default; diff --git a/kernel/kernel/include/task_flow/Task_Graph.cpp b/kernel/kernel/include/task_flow/Task_Graph.cpp index d448525..b52dcb4 100644 --- a/kernel/kernel/include/task_flow/Task_Graph.cpp +++ b/kernel/kernel/include/task_flow/Task_Graph.cpp @@ -1,5 +1,5 @@ #include "Task_Graph.hpp" -#include "Task_Graph_Execution.hpp" +#include "detail/Taskflow_Execution.ipp" #include #include #include diff --git a/kernel/kernel/include/task_flow/Task_Graph_Execution.hpp b/kernel/kernel/include/task_flow/Task_Graph_Execution.hpp deleted file mode 100644 index 93dcd10..0000000 --- a/kernel/kernel/include/task_flow/Task_Graph_Execution.hpp +++ /dev/null @@ -1,33 +0,0 @@ -#pragma once -#include "Task_Graph.hpp" -#include "Taskflow_Trace.hpp" -#include -#include -#include -#include -namespace tf { -class Taskflow; -} -namespace aethera::detail { -/* 一次异步执行对根图及全部子图的独占租约。 */ -struct Task_Graph_Execution final { -private: - struct Construction_Key {}; -public: - static std::shared_ptr try_acquire( - Task_Graph& graph); - ~Task_Graph_Execution(); - Task_Graph_Execution(const Task_Graph_Execution&) = delete; - Task_Graph_Execution(Task_Graph_Execution&&) = delete; - Task_Graph_Execution& operator=(const Task_Graph_Execution&) = delete; - Task_Graph_Execution& operator=(Task_Graph_Execution&&) = delete; - explicit Task_Graph_Execution(Construction_Key); - tf::Taskflow& native_taskflow() noexcept; - std::vector nodes() const; - void bind_observation(void* observation) noexcept; - std::exception_ptr take_failure() noexcept; -private: - std::shared_ptr root; - std::vector> graphs; -}; -} diff --git a/kernel/kernel/include/task_flow/Task_Runtime.cpp b/kernel/kernel/include/task_flow/Task_Runtime.cpp index e7e9ee6..076d5bd 100644 --- a/kernel/kernel/include/task_flow/Task_Runtime.cpp +++ b/kernel/kernel/include/task_flow/Task_Runtime.cpp @@ -1,9 +1,9 @@ #include "Task_Runtime.hpp" -#include "Task_Graph_Execution.hpp" -#include "Taskflow_Observation_Execution.hpp" +#include "detail/Taskflow_Execution.ipp" #include #include #include +#include #include #include #include @@ -98,24 +98,31 @@ public: Run_Taskflow_Result run( Task_Graph& graph, Taskflow_Completion completion, - Taskflow_Observation observation) { + std::optional observation) { if (!ready()) return Run_Taskflow_Result::runtime_not_initialized; if (!completion) return Run_Taskflow_Result::completion_missing; auto graph_execution = detail::Task_Graph_Execution::try_acquire(graph); if (!graph_execution) return Run_Taskflow_Result::graph_busy; std::string taskflow_name; std::vector nodes; - if (detail::Taskflow_Observation_Execution::requested(observation)) { + if (observation && + detail::Taskflow_Observation_Execution::requested(*observation)) { auto& taskflow = graph_execution->native_taskflow(); taskflow_name = taskflow.name(); nodes = graph_execution->nodes(); } - auto observation_execution = - detail::Taskflow_Observation_Execution::try_start( - std::move(observation), - executor->num_workers(), - std::move(taskflow_name), - std::move(nodes)); + auto observation_execution = [&]() -> std::expected< + detail::Taskflow_Observation_Execution, + detail::Start_Taskflow_Observation_Result> { + if (!observation) { + return detail::Taskflow_Observation_Execution{}; + } + return detail::Taskflow_Observation_Execution::try_start( + std::move(*observation), + executor->num_workers(), + std::move(taskflow_name), + std::move(nodes)); + }(); if (!observation_execution) { return Run_Taskflow_Result::observation_unavailable; } @@ -172,6 +179,14 @@ Initialize_Task_Runtime_Result initialize_task_runtime( std::size_t workers) { return Task_Resource::instance().initialize(workers); } +Run_Taskflow_Result run_taskflow( + Task_Graph& graph, + Taskflow_Completion completion) { + return Task_Resource::instance().run( + graph, + std::move(completion), + std::nullopt); +} Run_Taskflow_Result run_taskflow( Task_Graph& graph, Taskflow_Completion completion, diff --git a/kernel/kernel/include/task_flow/Task_Runtime.hpp b/kernel/kernel/include/task_flow/Task_Runtime.hpp index 48f3334..e75de1f 100644 --- a/kernel/kernel/include/task_flow/Task_Runtime.hpp +++ b/kernel/kernel/include/task_flow/Task_Runtime.hpp @@ -23,9 +23,13 @@ enum struct Run_Taskflow_Result : std::uint8_t { using Taskflow_Completion = std::function; Initialize_Task_Runtime_Result initialize_task_runtime( std::size_t workers = std::thread::hardware_concurrency()); -/* 完成回调接收业务任务传播出的 Unknown Failure;observation 自身决定是否记录。 */ +/* 完成回调接收业务任务传播出的 Unknown Failure;本次执行不记录观察数据。 */ +Run_Taskflow_Result run_taskflow( + Task_Graph& graph, + Taskflow_Completion completion); +/* 完成回调接收业务任务传播出的 Unknown Failure;observation 记录本次执行。 */ Run_Taskflow_Result run_taskflow( Task_Graph& graph, Taskflow_Completion completion, - Taskflow_Observation observation = Taskflow_Observation{}); + Taskflow_Observation observation); } diff --git a/kernel/kernel/include/task_flow/Taskflow_Observation.cpp b/kernel/kernel/include/task_flow/Taskflow_Observation.cpp index 34912fc..800078a 100644 --- a/kernel/kernel/include/task_flow/Taskflow_Observation.cpp +++ b/kernel/kernel/include/task_flow/Taskflow_Observation.cpp @@ -1,5 +1,5 @@ #include "Taskflow_Observation.hpp" -#include "Taskflow_Observation_Execution.hpp" +#include "detail/Taskflow_Execution.ipp" #include #include #include @@ -28,7 +28,9 @@ struct Taskflow_Observation::Private { std::vector starts{}; /* 对应 Worker 的嵌套调用栈。 */ std::size_t ignored_depth{}; /* 栈写入失败后用于配平回调的嵌套深度。 */ }; - explicit Private(std::string observation_stage) : stage(std::move(observation_stage)) {} + explicit Private(std::string observation_stage) { + trace.stage = std::move(observation_stage); + } ~Private() { delete failure.exchange(nullptr, std::memory_order_acq_rel); } @@ -54,20 +56,14 @@ struct Taskflow_Observation::Private { failure.exchange(nullptr, std::memory_order_acq_rel)}; return value ? std::move(*value) : std::exception_ptr{}; } - std::string stage{}; /* requested 阶段尚未移入结果的业务阶段。 */ std::atomic phase{Observation_Phase::requested}; /* 本次观察从请求到消费的唯一权威状态。 */ Clock::time_point started_at{}; /* 所有执行相对时间的单调时钟原点。 */ - Taskflow_Execution_Trace trace{}; /* recording/ready 阶段的唯一记录数据。 */ + Taskflow_Execution_Trace trace{}; /* requested 阶段保存 stage,随后保存唯一记录数据。 */ std::vector worker_states{}; /* 仅在 recording 阶段存在的逐 Worker 单写状态。 */ std::atomic failure{}; /* 原子槽暂存可空所有权;取出后立即由 unique_ptr 接管。 */ }; -Taskflow_Observation::Taskflow_Observation( - Taskflow_Observation_Mode mode, - std::string stage) { - if (mode == Taskflow_Observation_Mode::record) { - d = std::make_shared(std::move(stage)); - } -} +Taskflow_Observation::Taskflow_Observation(std::string stage) : + d(std::make_shared(std::move(stage))) {} Taskflow_Observation::~Taskflow_Observation() = default; Taskflow_Observation::Taskflow_Observation(const Taskflow_Observation&) = default; Taskflow_Observation::Taskflow_Observation(Taskflow_Observation&&) noexcept = default; @@ -123,7 +119,8 @@ std::expected nodes) { if (!observation.d) { - return Taskflow_Observation_Execution{std::move(observation)}; + return std::unexpected{ + Start_Taskflow_Observation_Result::unavailable}; } auto phase = observation.d->phase.load(std::memory_order_acquire); if (phase != Observation_Phase::requested || @@ -140,6 +137,7 @@ std::expected( std::chrono::duration_cast(system_now) @@ -151,7 +149,6 @@ std::expected value) noexcept : + observation(std::move(value)) {} +detail::Taskflow_Observation_Execution::Taskflow_Observation_Execution() noexcept = default; detail::Taskflow_Observation_Execution::~Taskflow_Observation_Execution() { cancel(); } detail::Taskflow_Observation_Execution::Taskflow_Observation_Execution( Taskflow_Observation_Execution&&) noexcept = default; void* detail::Taskflow_Observation_Execution::binding() noexcept { - return observation.d.get(); + return observation ? this : nullptr; } void detail::Taskflow_Observation_Execution::observe_entry( std::size_t worker, @@ -179,7 +178,7 @@ void detail::Taskflow_Observation_Execution::observe_entry( std::size_t queue_size, std::size_t queue_capacity, Clock::time_point entered) noexcept { - auto& data = *observation.d; + auto& data = *observation; if (worker >= data.worker_states.size()) std::terminate(); auto& state = data.worker_states[worker]; if (state.ignored_depth != 0) { @@ -204,7 +203,7 @@ void detail::Taskflow_Observation_Execution::observe_exit( std::size_t worker, std::uint64_t native_id, Clock::time_point finished) noexcept { - auto& data = *observation.d; + auto& data = *observation; if (worker >= data.worker_states.size()) std::terminate(); auto& state = data.worker_states[worker]; if (state.ignored_depth != 0) { @@ -234,23 +233,24 @@ void detail::Taskflow_Observation_Execution::observe_exit( } void detail::Taskflow_Observation_Execution::finish( Clock::time_point executor_finished) noexcept { - if (!observation.d) return; - auto& data = *observation.d; + if (!observation) return; + auto& data = *observation; data.trace.executor_finished_ms = data.elapsed_ms(executor_finished); data.trace.observation_finished_ms = data.elapsed_ms(Clock::now()); data.phase.store(Observation_Phase::ready, std::memory_order_release); } void detail::Taskflow_Observation_Execution::cancel() noexcept { - if (!observation.d || - observation.d->phase.load(std::memory_order_acquire) != + if (!observation || + observation->phase.load(std::memory_order_acquire) != Observation_Phase::recording) { return; } - auto& data = *observation.d; + auto& data = *observation; delete data.failure.exchange(nullptr, std::memory_order_acq_rel); data.worker_states.clear(); - data.stage = std::move(data.trace.stage); + auto stage = std::move(data.trace.stage); data.trace = {}; + data.trace.stage = std::move(stage); data.phase.store(Observation_Phase::requested, std::memory_order_release); } } diff --git a/kernel/kernel/include/task_flow/Taskflow_Observation.hpp b/kernel/kernel/include/task_flow/Taskflow_Observation.hpp index 645abd1..cbb980d 100644 --- a/kernel/kernel/include/task_flow/Taskflow_Observation.hpp +++ b/kernel/kernel/include/task_flow/Taskflow_Observation.hpp @@ -4,29 +4,22 @@ #include #include #include -#include "../global.hpp" namespace aethera { namespace detail { struct Taskflow_Observation_Execution; } -enum struct Taskflow_Observation_Mode : std::uint8_t { - disabled, - record -}; enum struct Take_Taskflow_Observation_Result : std::uint8_t { not_recorded, recording, already_taken }; /* - * 一次 Taskflow 执行的可选观察数据。Mode 是是否分配 Worker 记录的唯一来源。 + * 一次 Taskflow 执行的观察数据。只有把该对象传给 run_taskflow 才会记录。 * 复制句柄共享同一次观察;执行完成后任一句柄均可取走唯一结果。 */ struct Taskflow_Observation { public: - explicit Taskflow_Observation( - Taskflow_Observation_Mode mode = Taskflow_Observation_Mode::disabled, - std::string stage = {}); + explicit Taskflow_Observation(std::string stage); ~Taskflow_Observation(); Taskflow_Observation(const Taskflow_Observation&); Taskflow_Observation(Taskflow_Observation&&) noexcept; diff --git a/kernel/kernel/include/task_flow/design.md b/kernel/kernel/include/task_flow/design.md index c711555..929ac2b 100644 --- a/kernel/kernel/include/task_flow/design.md +++ b/kernel/kernel/include/task_flow/design.md @@ -9,14 +9,14 @@ | `Task_Graph` | 构图、组合子图、保存业务节点元数据 | | `Task_Graph_Execution` | 独占本次执行涉及的根图和子图,并管理运行期借用 | | `Task_Runtime` | 初始化 Executor、提交执行、传递完成与异常 | -| `Taskflow_Observation` | 决定本次执行是否记录,并独占记录状态 | +| `Taskflow_Observation` | 表达一次显式记录请求,并独占记录状态 | | `Taskflow_Observation_Execution` | 持有本次可选记录;提交失败时自动回滚 | | `Taskflow_Execution_Trace` | 执行完成后移出的原始观察结果 | ## 执行流程 1. `run_taskflow` 创建 `Task_Graph_Execution`,独占根图及全部子图;任一图已占用则返回 `graph_busy`。 -2. Observation 为 `record` 时创建逐 Worker 状态,并把其非拥有指针写入所有节点;禁用时不分配记录数据。 +2. 调用方传入 Observation 时创建逐 Worker 状态,并把执行对象的非拥有指针写入所有节点;未传入时不分配记录数据。 3. 内部执行对象持有 Graph 租约、Observation 执行对象和完成回调,然后提交给 Taskflow。 4. Observer 从节点读取 Observation 指针;为空立即返回,否则只写 `worker_tasks[worker_id]`。 5. topology 完成后发布 Observation、清除节点指针、释放 Graph,最后调用业务完成回调。 @@ -24,9 +24,9 @@ ## 状态与生命周期 - 每张 Graph 的 `active` 是其占用状态的唯一来源;`Task_Graph_Execution` 析构时统一清除借用并释放根图和子图。 -- 禁用 Observation 没有 Private;启用后 `requested → recording → ready → consumed` 是唯一状态链。 +- 未传入 Observation 时没有记录状态;传入后 `requested → recording → ready → consumed` 是唯一状态链。 - `Taskflow_Execution` 是异步生命周期所有者。调用方可在提交后销毁或移动原 `Task_Graph`/Observation 句柄。 -- 节点中的 `void*` 是可空、非拥有借用;Observation 的共享所有权保证它在最后一次 Observer 回调前有效。 +- 节点中的 `void*` 是指向 Observation 执行对象的可空、非拥有借用;执行所有权保证它在最后一次 Observer 回调前有效。 - Trace 只保存原始数据。Worker ID 等于 `worker_tasks` 外层下标;后继关系由节点的 `predecessors` 计算。 ## 线程契约 @@ -47,7 +47,6 @@ ```cpp Taskflow_Observation observation{ - Taskflow_Observation_Mode::record, "render_2d.frame"}; const auto result = run_taskflow( diff --git a/kernel/kernel/include/task_flow/Taskflow_Observation_Execution.hpp b/kernel/kernel/include/task_flow/detail/Taskflow_Execution.ipp similarity index 53% rename from kernel/kernel/include/task_flow/Taskflow_Observation_Execution.hpp rename to kernel/kernel/include/task_flow/detail/Taskflow_Execution.ipp index 70ab6ea..55b3601 100644 --- a/kernel/kernel/include/task_flow/Taskflow_Observation_Execution.hpp +++ b/kernel/kernel/include/task_flow/detail/Taskflow_Execution.ipp @@ -1,18 +1,48 @@ #pragma once -#include "Taskflow_Observation.hpp" +#include "../Task_Graph.hpp" +#include "../Taskflow_Observation.hpp" +#include "../Taskflow_Trace.hpp" #include #include #include +#include #include +#include +#include #include #include +namespace tf { +class Taskflow; +} namespace aethera::detail { +/* 一次异步执行对根图及全部子图的独占租约。 */ +struct Task_Graph_Execution final { +private: + struct Construction_Key {}; +public: + static std::shared_ptr try_acquire( + Task_Graph& graph); + ~Task_Graph_Execution(); + Task_Graph_Execution(const Task_Graph_Execution&) = delete; + Task_Graph_Execution(Task_Graph_Execution&&) = delete; + Task_Graph_Execution& operator=(const Task_Graph_Execution&) = delete; + Task_Graph_Execution& operator=(Task_Graph_Execution&&) = delete; + explicit Task_Graph_Execution(Construction_Key); + tf::Taskflow& native_taskflow() noexcept; + std::vector nodes() const; + void bind_observation(void* observation) noexcept; + std::exception_ptr take_failure() noexcept; +private: + std::shared_ptr root; /* 根 DAG 的执行期共享所有权。 */ + std::vector> graphs; /* 本次独占的全部 DAG 非拥有借用。 */ +}; enum struct Start_Taskflow_Observation_Result : std::uint8_t { unavailable }; /* 一次异步执行对 Observation 记录状态的所有权。 */ struct Taskflow_Observation_Execution final { public: + Taskflow_Observation_Execution() noexcept; static bool requested( const Taskflow_Observation& observation) noexcept; static std::expected< @@ -46,8 +76,8 @@ public: std::chrono::steady_clock::time_point executor_finished) noexcept; private: explicit Taskflow_Observation_Execution( - Taskflow_Observation observation) noexcept; + std::shared_ptr observation) noexcept; void cancel() noexcept; - Taskflow_Observation observation; + std::shared_ptr observation; /* 可空记录所有权;空表示本次执行未请求观察。 */ }; } diff --git a/kernel/kernel/include/time_thread/Timer_Service.cpp b/kernel/kernel/include/time_thread/Timer_Service.cpp index 2f75b9d..8d4e06e 100644 --- a/kernel/kernel/include/time_thread/Timer_Service.cpp +++ b/kernel/kernel/include/time_thread/Timer_Service.cpp @@ -1,19 +1,25 @@ #include "Timer_Service.hpp" -#include "timer-wheel.h" -#include +#include "Timer_Time_Source.hpp" +#include "detail/Timer_Scheduler.hpp" + #include #include #include -#include +#include #include #include -#include #include -#include namespace aethera { +namespace { +struct Steady_Timer_Time_Source final : Timer_Time_Source { + Time_Point now() const noexcept override { + return std::chrono::steady_clock::now(); + } +}; +} + struct Timer_Service::Impl { - static constexpr std::chrono::microseconds Tick_Duration{10}; - static constexpr Tick Max_Wait_Ticks = 100000; + static constexpr std::chrono::seconds Maximum_Wait{1}; enum class Command_Type { Add, Cancel, @@ -23,38 +29,17 @@ struct Timer_Service::Impl { struct Command { Command_Type type; Timer_Id id{}; - Tick delay{}; - Tick interval{}; + std::chrono::nanoseconds delay{}; + std::chrono::nanoseconds interval{}; Callback callback; }; - struct Event : TimerEventInterface { - Impl* owner; - Timer_Id id; - Tick interval; - Callback callback; - Event(Impl* owner, Timer_Id id, Tick interval, Callback callback) : owner(owner), id(id), interval(interval), callback(std::move(callback)) {} - void execute() override { - try { - callback(); - } - catch (...) {} - if (interval == 0) { - owner->completed.emplace_back(id); - return; - } - owner->repeating.emplace_back(id); - } - }; - std::unique_ptr wheel{std::make_unique()}; - std::unordered_map> events; - std::vector completed; - std::vector repeating; + Steady_Timer_Time_Source time_source; + detail::Timer_Scheduler scheduler{time_source}; std::atomic next_id{1}; std::mutex mutex; std::condition_variable cv; std::deque commands; std::thread worker; - std::chrono::steady_clock::time_point last_update{std::chrono::steady_clock::now()}; Impl() : worker([this] { run(); }) {} @@ -62,17 +47,9 @@ struct Timer_Service::Impl { push(Command{Command_Type::Stop}); worker.join(); } - static Tick to_ticks(std::chrono::nanoseconds duration) { - if (duration <= std::chrono::nanoseconds::zero()) { - return 1; - } - auto ns = duration.count(); - auto tick_ns = std::chrono::duration_cast(Tick_Duration).count(); - return static_cast((ns + tick_ns - 1) / tick_ns); - } Timer_Id schedule(std::chrono::nanoseconds delay, std::chrono::nanoseconds interval, Callback callback) { Timer_Id id = next_id.fetch_add(1, std::memory_order_relaxed); - push(Command{Command_Type::Add, id, to_ticks(delay), interval == std::chrono::nanoseconds::zero() ? 0 : to_ticks(interval), std::move(callback)}); + push(Command{Command_Type::Add, id, delay, interval, std::move(callback)}); return id; } void push(Command command) { @@ -82,36 +59,30 @@ struct Timer_Service::Impl { } cv.notify_one(); } - void reset_wheel() { - wheel = std::make_unique(); - last_update = std::chrono::steady_clock::now(); - } void apply(Command&& command, bool& stop) { switch (command.type) { case Command_Type::Add: { - if (events.empty()) { - reset_wheel(); - } - auto event = std::make_unique(this, command.id, command.interval, std::move(command.callback)); - Event* ptr = event.get(); - events.emplace(command.id, std::move(event)); - wheel->schedule(ptr, command.delay); + scheduler.schedule( + command.id, + command.delay, + command.interval, + std::move(command.callback)); break; } case Command_Type::Cancel: { - auto it = events.find(command.id); - if (it != events.end()) { - it->second->cancel(); - events.erase(it); + scheduler.cancel(command.id); + if (command.callback) { + try { + command.callback(); + } + catch (...) { + std::terminate(); + } } break; } case Command_Type::Reschedule: { - auto it = events.find(command.id); - if (it != events.end()) { - it->second->cancel(); - wheel->schedule(it->second.get(), command.delay); - } + scheduler.reschedule(command.id, command.delay); break; } case Command_Type::Stop: stop = true; @@ -128,56 +99,22 @@ struct Timer_Service::Impl { apply(std::move(command), stop); } } - void finish_events() { - for (Timer_Id id : completed) { - events.erase(id); - } - completed.clear(); - Tick current = wheel->now(); - for (Timer_Id id : repeating) { - auto it = events.find(id); - if (it == events.end()) { - continue; - } - Event& event = *it->second; - Tick previous = event.scheduled_at(); - Tick periods = current >= previous ? (current - previous) / event.interval + 1 : 1; - Tick next = previous + periods * event.interval; - wheel->schedule(&event, std::max(1, next - current)); - } - repeating.clear(); - } - void advance_to_now() { - if (events.empty()) { - return; - } - auto now = std::chrono::steady_clock::now(); - auto elapsed = now - last_update; - Tick ticks = static_cast(std::chrono::duration_cast(elapsed).count() / std::chrono::duration_cast(Tick_Duration).count()); - if (ticks == 0) { - return; - } - wheel->advance(ticks); - last_update += Tick_Duration * ticks; - finish_events(); - } void run() { bool stop = false; while (!stop) { - advance_to_now(); + scheduler.advance_to_current_time(); drain(stop); if (stop) { break; } - if (events.empty()) { + if (scheduler.empty()) { std::unique_lock lock(mutex); cv.wait(lock, [this] { return !commands.empty(); }); continue; } - Tick wait_ticks = std::min(wheel->ticks_to_next_event(), Max_Wait_Ticks); - auto deadline = last_update + Tick_Duration * wait_ticks; + const auto deadline = scheduler.next_deadline(Maximum_Wait); std::unique_lock lock(mutex); if (!commands.empty()) { continue; @@ -186,10 +123,6 @@ struct Timer_Service::Impl { return !commands.empty(); }); } - for (auto& [id, event] : events) { - event->cancel(); - } - events.clear(); } }; Timer_Service& Timer_Service::instance() { @@ -203,14 +136,29 @@ Timer_Id Timer_Service::schedule_after(std::chrono::nanoseconds delay, Callback } Timer_Id Timer_Service::schedule_every(std::chrono::nanoseconds interval, Callback callback) { if (interval <= std::chrono::nanoseconds::zero()) { - interval = Impl::Tick_Duration; + interval = std::chrono::nanoseconds{1}; } return impl->schedule(interval, interval, std::move(callback)); } +std::expected Timer_Service::schedule_every_fps( + double frames_per_second, + Callback callback) { + const auto interval = + detail::timer_interval_from_frames_per_second(frames_per_second); + if (!interval.has_value()) { + return std::unexpected(interval.error()); + } + return schedule_every(*interval, std::move(callback)); +} void Timer_Service::reschedule(Timer_Id id, std::chrono::nanoseconds delay) { - impl->push(Impl::Command{Impl::Command_Type::Reschedule, id, Impl::to_ticks(delay)}); + impl->push(Impl::Command{Impl::Command_Type::Reschedule, id, delay}); } -void Timer_Service::cancel(Timer_Id id) { - impl->push(Impl::Command{Impl::Command_Type::Cancel, id}); +void Timer_Service::cancel(Timer_Id id, Callback completion) { + impl->push(Impl::Command{ + Impl::Command_Type::Cancel, + id, + {}, + {}, + std::move(completion)}); } } diff --git a/kernel/kernel/include/time_thread/Timer_Service.hpp b/kernel/kernel/include/time_thread/Timer_Service.hpp index 72460bb..f76232d 100644 --- a/kernel/kernel/include/time_thread/Timer_Service.hpp +++ b/kernel/kernel/include/time_thread/Timer_Service.hpp @@ -1,18 +1,31 @@ #pragma once #include #include +#include #include #include #include "global.hpp" namespace aethera { using Timer_Id = std::uint64_t; +enum struct Schedule_Every_Fps_Result : std::uint8_t { + invalid_frames_per_second, + interval_out_of_range +}; struct Timer_Service : Non_Copyable { using Callback = std::function; static Timer_Service& instance(); + // 多少秒后执行一次 Timer_Id schedule_after(std::chrono::nanoseconds delay, Callback callback); + template Timer_Id schedule_after(std::chrono::duration delay, Callback callback); + // 循环执行 Timer_Id schedule_every(std::chrono::nanoseconds interval, Callback callback); + template Timer_Id schedule_every(std::chrono::duration interval, Callback callback); + std::expected schedule_every_fps(double frames_per_second, Callback callback); + // 推后下一次执行 void reschedule(Timer_Id id, std::chrono::nanoseconds delay); - void cancel(Timer_Id id); + template void reschedule(Timer_Id id, std::chrono::duration delay); + // 异步取消;completion 在该计时器不存在在途或后续回调时执行。 + void cancel(Timer_Id id, Callback completion); private: struct Impl; Timer_Service(); @@ -20,3 +33,4 @@ private: std::unique_ptr impl; }; } +#include "Timer_Service.ipp" diff --git a/kernel/kernel/include/time_thread/Timer_Service.ipp b/kernel/kernel/include/time_thread/Timer_Service.ipp new file mode 100644 index 0000000..16761e7 --- /dev/null +++ b/kernel/kernel/include/time_thread/Timer_Service.ipp @@ -0,0 +1,78 @@ +#pragma once + +#include +#include +#include +#include + +namespace aethera::detail { +template +std::chrono::nanoseconds timer_duration_to_nanoseconds( + std::chrono::duration duration) noexcept { + using Floating_Nanoseconds = + std::chrono::duration; + constexpr auto Maximum = + static_cast( + std::numeric_limits::max()); + + const auto nanoseconds = + std::chrono::duration_cast(duration).count(); + if (!(nanoseconds > 0.0L)) { + return std::chrono::nanoseconds::zero(); + } + if (!std::isfinite(nanoseconds) || nanoseconds >= Maximum) { + return std::chrono::nanoseconds::max(); + } + return std::chrono::nanoseconds{ + static_cast( + std::ceil(nanoseconds))}; +} + +inline std::expected +timer_interval_from_frames_per_second(double frames_per_second) noexcept { + if (!std::isfinite(frames_per_second) || frames_per_second <= 0.0) { + return std::unexpected( + Schedule_Every_Fps_Result::invalid_frames_per_second); + } + constexpr long double Nanoseconds_Per_Second = 1'000'000'000.0L; + const long double interval = + Nanoseconds_Per_Second / static_cast(frames_per_second); + constexpr auto Maximum_Interval = + static_cast( + std::numeric_limits::max()); + if (!std::isfinite(interval) || interval > Maximum_Interval) { + return std::unexpected( + Schedule_Every_Fps_Result::interval_out_of_range); + } + return std::chrono::nanoseconds{ + static_cast( + std::max(1.0L, std::ceil(interval)))}; +} +} + +namespace aethera { +template +Timer_Id Timer_Service::schedule_after( + std::chrono::duration delay, + Callback callback) { + return schedule_after( + detail::timer_duration_to_nanoseconds(delay), + std::move(callback)); +} + +template +Timer_Id Timer_Service::schedule_every( + std::chrono::duration interval, + Callback callback) { + return schedule_every( + detail::timer_duration_to_nanoseconds(interval), + std::move(callback)); +} + +template +void Timer_Service::reschedule( + Timer_Id id, + std::chrono::duration delay) { + reschedule(id, detail::timer_duration_to_nanoseconds(delay)); +} +} diff --git a/kernel/kernel/include/time_thread/Timer_Time_Source.hpp b/kernel/kernel/include/time_thread/Timer_Time_Source.hpp new file mode 100644 index 0000000..077f2a4 --- /dev/null +++ b/kernel/kernel/include/time_thread/Timer_Time_Source.hpp @@ -0,0 +1,12 @@ +#pragma once + +#include + +namespace aethera { +struct Timer_Time_Source { + using Time_Point = std::chrono::steady_clock::time_point; + + virtual ~Timer_Time_Source() = default; + virtual Time_Point now() const noexcept = 0; +}; +} diff --git a/kernel/kernel/include/time_thread/detail/Timer_Scheduler.cpp b/kernel/kernel/include/time_thread/detail/Timer_Scheduler.cpp new file mode 100644 index 0000000..7e1acd9 --- /dev/null +++ b/kernel/kernel/include/time_thread/detail/Timer_Scheduler.cpp @@ -0,0 +1,203 @@ +#include "Timer_Scheduler.hpp" + +#include "timer-wheel.h" + +#include +#include +#include +#include + +namespace aethera::detail { +struct Timer_Scheduler::Impl { + static constexpr std::chrono::microseconds Tick_Duration{10}; + + struct Event final : TimerEventInterface { + Impl& owner; + Timer_Id id; + Tick interval; + Timer_Service::Callback callback; + + Event(Impl& owner, + Timer_Id id, + Tick interval, + Timer_Service::Callback callback) : + owner(owner), + id(id), + interval(interval), + callback(std::move(callback)) {} + + void execute() override { + try { + callback(); + } + catch (...) {} + if (interval == 0) { + owner.completed.emplace_back(id); + return; + } + owner.repeating.emplace_back(id); + } + }; + + explicit Impl(Timer_Time_Source& time_source) : + time_source(time_source) {} + + static Tick to_ticks(std::chrono::nanoseconds duration) noexcept { + if (duration <= std::chrono::nanoseconds::zero()) { + return 1; + } + const auto nanoseconds = duration.count(); + constexpr auto Tick_Nanoseconds = + std::chrono::duration_cast( + Tick_Duration).count(); + return static_cast( + nanoseconds / Tick_Nanoseconds + + (nanoseconds % Tick_Nanoseconds == 0 ? 0 : 1)); + } + + void reset_wheel() { + wheel = std::make_unique(); + origin_time = time_source.now(); + } + + Timer_Time_Source::Time_Point current_tick_time() const noexcept { + return origin_time + + Tick_Duration * static_cast(wheel->now()); + } + + void schedule(Timer_Id id, + std::chrono::nanoseconds delay, + std::chrono::nanoseconds interval, + Timer_Service::Callback callback) { + if (events.empty()) { + reset_wheel(); + } + const Tick interval_ticks = + interval == std::chrono::nanoseconds::zero() + ? 0 + : to_ticks(interval); + auto event = std::make_unique( + *this, + id, + interval_ticks, + std::move(callback)); + wheel->schedule(event.get(), to_ticks(delay)); + events.emplace(id, std::move(event)); + } + + void reschedule(Timer_Id id, std::chrono::nanoseconds delay) { + const auto event = events.find(id); + if (event == events.end()) { + return; + } + event->second->cancel(); + wheel->schedule(event->second.get(), to_ticks(delay)); + } + + void cancel(Timer_Id id) { + const auto event = events.find(id); + if (event == events.end()) { + return; + } + event->second->cancel(); + events.erase(event); + } + + void finish_events() { + for (const Timer_Id id : completed) { + events.erase(id); + } + completed.clear(); + + const Tick current = wheel->now(); + for (const Timer_Id id : repeating) { + const auto entry = events.find(id); + if (entry == events.end()) { + continue; + } + Event& event = *entry->second; + const Tick previous = event.scheduled_at(); + const Tick periods = current >= previous + ? (current - previous) / event.interval + 1 + : 1; + const Tick next = previous + periods * event.interval; + wheel->schedule( + std::addressof(event), + std::max(1, next - current)); + } + repeating.clear(); + } + + void advance_to_current_time() { + if (events.empty()) { + return; + } + const auto current_time = time_source.now(); + const auto elapsed = current_time - current_tick_time(); + if (elapsed < Tick_Duration) { + return; + } + const auto elapsed_nanoseconds = + std::chrono::duration_cast(elapsed); + constexpr auto Tick_Nanoseconds = + std::chrono::duration_cast( + Tick_Duration).count(); + const Tick ticks = static_cast( + elapsed_nanoseconds.count() / Tick_Nanoseconds); + wheel->advance(ticks); + finish_events(); + } + + Timer_Time_Source::Time_Point next_deadline( + std::chrono::nanoseconds maximum_wait) const { + const Tick wait_ticks = std::min( + wheel->ticks_to_next_event(), + to_ticks(maximum_wait)); + return current_tick_time() + + Tick_Duration * static_cast(wait_ticks); + } + + Timer_Time_Source& time_source; /* Required non-owning time authority. */ + std::unique_ptr wheel{std::make_unique()}; + std::unordered_map> events; + std::vector completed; + std::vector repeating; + Timer_Time_Source::Time_Point origin_time{}; +}; + +Timer_Scheduler::Timer_Scheduler(Timer_Time_Source& time_source) : + impl(std::make_unique(time_source)) {} + +Timer_Scheduler::~Timer_Scheduler() = default; + +void Timer_Scheduler::schedule( + Timer_Id id, + std::chrono::nanoseconds delay, + std::chrono::nanoseconds interval, + Timer_Service::Callback callback) { + impl->schedule(id, delay, interval, std::move(callback)); +} + +void Timer_Scheduler::reschedule( + Timer_Id id, + std::chrono::nanoseconds delay) { + impl->reschedule(id, delay); +} + +void Timer_Scheduler::cancel(Timer_Id id) { + impl->cancel(id); +} + +void Timer_Scheduler::advance_to_current_time() { + impl->advance_to_current_time(); +} + +bool Timer_Scheduler::empty() const noexcept { + return impl->events.empty(); +} + +Timer_Time_Source::Time_Point Timer_Scheduler::next_deadline( + std::chrono::nanoseconds maximum_wait) const { + return impl->next_deadline(maximum_wait); +} +} diff --git a/kernel/kernel/include/time_thread/detail/Timer_Scheduler.hpp b/kernel/kernel/include/time_thread/detail/Timer_Scheduler.hpp new file mode 100644 index 0000000..8894922 --- /dev/null +++ b/kernel/kernel/include/time_thread/detail/Timer_Scheduler.hpp @@ -0,0 +1,28 @@ +#pragma once + +#include "time_thread/Timer_Service.hpp" +#include "time_thread/Timer_Time_Source.hpp" + +#include +#include + +namespace aethera::detail { +struct Timer_Scheduler : Non_Copyable { + explicit Timer_Scheduler(Timer_Time_Source& time_source); + ~Timer_Scheduler(); + + void schedule(Timer_Id id, + std::chrono::nanoseconds delay, + std::chrono::nanoseconds interval, + Timer_Service::Callback callback); + void reschedule(Timer_Id id, std::chrono::nanoseconds delay); + void cancel(Timer_Id id); + void advance_to_current_time(); + [[nodiscard]] bool empty() const noexcept; + [[nodiscard]] Timer_Time_Source::Time_Point next_deadline( + std::chrono::nanoseconds maximum_wait) const; +private: + struct Impl; + std::unique_ptr impl; +}; +} diff --git a/kernel/kernel/test/function/frame_policy/Throttled_Latest_Only_Tests.cpp b/kernel/kernel/test/function/frame_policy/Throttled_Latest_Only_Tests.cpp new file mode 100644 index 0000000..754832f --- /dev/null +++ b/kernel/kernel/test/function/frame_policy/Throttled_Latest_Only_Tests.cpp @@ -0,0 +1,248 @@ +#include "function/frame_policy/detail/Throttled_Latest_Only_State.hpp" + +#include + +#include +#include +#include +#include +#include + +namespace { +struct Frame_Policy_Test_Control { + void complete_render() { + ASSERT_TRUE(rendering_frame.has_value()); + auto frame = *rendering_frame; + auto completion = std::move(render_completion); + rendering_frame.reset(); + completion(frame.get()); + } + + void complete_send() { + ASSERT_TRUE(sending_frame.has_value()); + auto frame = *sending_frame; + auto completion = std::move(send_completion); + sending_frame.reset(); + completion(frame.get()); + } + + std::optional>> + rendering_frame; /* Scene 当前借用的策略帧。 */ + std::optional>> + sending_frame; /* Sink 当前借用的策略帧。 */ + aethera::FP_Frame_Completion render_completion; /* Scene 归还帧的完成回调。 */ + aethera::FP_Frame_Completion send_completion; /* Sink 归还帧的完成回调。 */ + aethera::proxy* first_frame{}; /* 可空、非拥有的首次帧地址。 */ + std::size_t frames_created{}; /* Scene 创建的物理帧数量。 */ + std::size_t frames_alive{}; /* 尚未析构的物理帧数量。 */ + std::size_t render_calls{}; /* 已接受的 Scene 借用次数。 */ + std::size_t send_calls{}; /* 已接受的 Sink 借用次数。 */ + bool reused_same_frame{true}; /* 全部借用是否指向同一策略帧。 */ + bool complete_render_synchronously{}; /* Scene 是否在 render 调用内立即归还帧。 */ + bool complete_send_synchronously{}; /* Sink 是否在 send 调用内立即归还帧。 */ + bool throw_from_send{}; /* Sink 是否以 Unknown Failure 退出本次分派。 */ +}; + +struct Test_Frame { + explicit Test_Frame( + std::shared_ptr control) : + control(std::move(control)) { + ++this->control->frames_alive; + } + + Test_Frame(Test_Frame&& other) noexcept : + control(std::move(other.control)) {} + Test_Frame& operator=(Test_Frame&&) = delete; + Test_Frame(const Test_Frame&) = delete; + Test_Frame& operator=(const Test_Frame&) = delete; + + ~Test_Frame() { + if (control) { + --control->frames_alive; + } + } + + aethera::Time_Type use_time() { + return 0; + } + + std::shared_ptr control; /* 测试帧生命周期记录的共享所有权。 */ +}; + +struct Test_Scene { + aethera::proxy create_frame() { + ++control->frames_created; + return pro::make_proxy(control); + } + + void render( + aethera::proxy& frame, + aethera::FP_Frame_Completion completion) { + if (control->first_frame == nullptr) { + control->first_frame = std::addressof(frame); + } + control->reused_same_frame = + control->reused_same_frame && + control->first_frame == std::addressof(frame); + ++control->render_calls; + if (control->complete_render_synchronously) { + completion(frame); + return; + } + control->rendering_frame = frame; + control->render_completion = std::move(completion); + } + + std::shared_ptr control; /* Scene 测试行为的共享控制数据。 */ +}; + +struct Test_Sink { + void send( + aethera::proxy& frame, + aethera::FP_Frame_Completion completion) { + control->reused_same_frame = + control->reused_same_frame && + control->first_frame == std::addressof(frame); + ++control->send_calls; + if (control->throw_from_send) { + throw std::runtime_error("test sink failure"); + } + if (control->complete_send_synchronously) { + completion(frame); + return; + } + control->sending_frame = frame; + control->send_completion = std::move(completion); + } + + std::shared_ptr control; /* Sink 测试行为的共享控制数据。 */ +}; + +std::shared_ptr +make_state(const std::shared_ptr& control) { + auto scene = + pro::make_proxy(control); + auto sink = + pro::make_proxy(control); + return std::make_shared< + aethera::detail::Throttled_Latest_Only_State>( + std::move(scene), + std::move(sink)); +} + +TEST(Throttled_Latest_Only, Reuses_One_Scene_Created_Frame_And_Drops_Busy_Ticks) { + auto control = std::make_shared(); + auto state = make_state(control); + ASSERT_EQ( + state->prepare_start(), + aethera::Start_Throttled_Latest_Only_Result::started); + state->timer_started(7); + EXPECT_EQ(control->frames_created, 1); + EXPECT_EQ(control->frames_alive, 1); + + state->frame_due(); + state->frame_due(); + EXPECT_EQ(control->render_calls, 1); + EXPECT_EQ(control->send_calls, 0); + + control->complete_render(); + state->frame_due(); + EXPECT_EQ(control->render_calls, 1); + EXPECT_EQ(control->send_calls, 1); + control->complete_send(); + + state->frame_due(); + EXPECT_EQ(control->render_calls, 2); + control->complete_render(); + control->complete_send(); + EXPECT_TRUE(control->reused_same_frame); + + bool stopped = false; + const auto stop = state->request_stop([&stopped] { + stopped = true; + }); + ASSERT_EQ( + stop.result, + aethera::Stop_Throttled_Latest_Only_Result::stopping); + ASSERT_EQ(stop.timer_id, 7); + state->timer_cancelled(); + + EXPECT_TRUE(stopped); + EXPECT_EQ(control->frames_alive, 0); + EXPECT_TRUE(state->destructible()); +} + +TEST(Throttled_Latest_Only, Stop_Waits_For_Render_And_Send_Borrows_To_Return) { + auto control = std::make_shared(); + auto state = make_state(control); + ASSERT_EQ( + state->prepare_start(), + aethera::Start_Throttled_Latest_Only_Result::started); + state->timer_started(11); + state->frame_due(); + + bool stopped = false; + const auto stop = state->request_stop([&stopped] { + stopped = true; + }); + ASSERT_EQ( + stop.result, + aethera::Stop_Throttled_Latest_Only_Result::stopping); + state->timer_cancelled(); + EXPECT_FALSE(stopped); + EXPECT_EQ(control->frames_alive, 1); + + control->complete_render(); + EXPECT_FALSE(stopped); + EXPECT_EQ(control->send_calls, 1); + EXPECT_EQ(control->frames_alive, 1); + + control->complete_send(); + EXPECT_TRUE(stopped); + EXPECT_EQ(control->frames_alive, 0); + EXPECT_TRUE(state->destructible()); +} + +TEST(Throttled_Latest_Only, Reports_Missing_Dependencies_And_Stop_Completion) { + auto missing_dependencies = std::make_shared< + aethera::detail::Throttled_Latest_Only_State>( + aethera::proxy{}, + aethera::proxy{}); + + EXPECT_EQ( + missing_dependencies->prepare_start(), + aethera::Start_Throttled_Latest_Only_Result::dependency_unavailable); + EXPECT_EQ( + missing_dependencies->request_stop({}).result, + aethera::Stop_Throttled_Latest_Only_Result::completion_missing); + EXPECT_TRUE(missing_dependencies->destructible()); +} + +TEST(Throttled_Latest_Only, Synchronous_Completions_And_Failure_Restore_The_Frame) { + auto control = std::make_shared(); + control->complete_render_synchronously = true; + control->throw_from_send = true; + auto state = make_state(control); + ASSERT_EQ( + state->prepare_start(), + aethera::Start_Throttled_Latest_Only_Result::started); + state->timer_started(13); + + EXPECT_THROW(state->frame_due(), std::runtime_error); + control->throw_from_send = false; + control->complete_send_synchronously = true; + EXPECT_NO_THROW(state->frame_due()); + EXPECT_EQ(control->render_calls, 2); + EXPECT_EQ(control->send_calls, 2); + + bool stopped = false; + ASSERT_EQ( + state->request_stop([&stopped] { + stopped = true; + }).result, + aethera::Stop_Throttled_Latest_Only_Result::stopping); + state->timer_cancelled(); + EXPECT_TRUE(stopped); + EXPECT_TRUE(state->destructible()); +} +} diff --git a/kernel/kernel/test/task_flow/Taskflow_Observation_Tests.cpp b/kernel/kernel/test/task_flow/Taskflow_Observation_Tests.cpp new file mode 100644 index 0000000..2b03e60 --- /dev/null +++ b/kernel/kernel/test/task_flow/Taskflow_Observation_Tests.cpp @@ -0,0 +1,153 @@ +#include "task_flow/Task_Runtime.hpp" +#include "task_flow/detail/Taskflow_Execution.ipp" + +#include + +#include +#include +#include +#include +#include +#include +#include + +namespace { + +using Run_Without_Observation = aethera::Run_Taskflow_Result(*)( + aethera::Task_Graph&, + aethera::Taskflow_Completion); +using Run_With_Observation = aethera::Run_Taskflow_Result(*)( + aethera::Task_Graph&, + aethera::Taskflow_Completion, + aethera::Taskflow_Observation); + +static_assert(!std::default_initializable); +static_assert(std::constructible_from< + aethera::Taskflow_Observation, + std::string>); +static_assert(std::same_as< + decltype(static_cast( + &aethera::run_taskflow)), + Run_Without_Observation>); +static_assert(std::same_as< + decltype(static_cast( + &aethera::run_taskflow)), + Run_With_Observation>); + +struct Taskflow_Observation_Test : testing::Test { + Taskflow_Observation_Test() { + aethera::Taskflow_Node_Trace node{}; + node.native_id = native_id; + node.node_id = "test.graph/task"; + node.name = "task"; + node.type = "static"; + nodes.push_back(std::move(node)); + } + + static constexpr std::uint64_t native_id{17}; + aethera::Taskflow_Observation observation{"test.stage"}; /* 本测试共享的显式记录请求。 */ + std::vector nodes{}; /* 本次执行使用的静态节点定义。 */ +}; + +TEST_F(Taskflow_Observation_Test, Explicit_Request_Records_And_Consumes_One_Trace) { + auto execution = + aethera::detail::Taskflow_Observation_Execution::try_start( + observation, + 1, + "test.graph", + nodes); + ASSERT_TRUE(execution.has_value()); + EXPECT_EQ( + execution->binding(), + static_cast(std::addressof(*execution))); + + const auto entered = std::chrono::steady_clock::now(); + execution->observe_entry(0, native_id, 3, 8, entered); + execution->observe_exit( + 0, + native_id, + std::chrono::steady_clock::now()); + execution->finish(std::chrono::steady_clock::now()); + + auto trace = observation.take(); + ASSERT_TRUE(trace.has_value()); + EXPECT_EQ(trace->stage, "test.stage"); + EXPECT_EQ(trace->taskflow_name, "test.graph"); + ASSERT_EQ(trace->nodes.size(), 1); + EXPECT_EQ(trace->nodes.front().native_id, native_id); + ASSERT_EQ(trace->worker_tasks.size(), 1); + ASSERT_EQ(trace->worker_tasks.front().size(), 1); + const auto& task = trace->worker_tasks.front().front(); + EXPECT_EQ(task.native_id, native_id); + EXPECT_EQ(task.worker_queue_size, 3); + EXPECT_EQ(task.worker_queue_capacity, 8); + EXPECT_GE(task.completed_ms, task.finished_ms); +} + +TEST_F(Taskflow_Observation_Test, Cancelled_Start_Preserves_The_Only_Stage_Value) { + { + auto cancelled = + aethera::detail::Taskflow_Observation_Execution::try_start( + observation, + 1, + "cancelled.graph", + nodes); + ASSERT_TRUE(cancelled.has_value()); + } + + auto restarted = + aethera::detail::Taskflow_Observation_Execution::try_start( + observation, + 1, + "completed.graph", + nodes); + ASSERT_TRUE(restarted.has_value()); + restarted->finish(std::chrono::steady_clock::now()); + + auto trace = observation.take(); + ASSERT_TRUE(trace.has_value()); + EXPECT_EQ(trace->stage, "test.stage"); + EXPECT_EQ(trace->taskflow_name, "completed.graph"); +} + +TEST_F(Taskflow_Observation_Test, Copied_Handles_Consume_One_Authoritative_Result) { + auto copy = observation; + auto execution = + aethera::detail::Taskflow_Observation_Execution::try_start( + observation, + 1, + "test.graph", + nodes); + ASSERT_TRUE(execution.has_value()); + execution->finish(std::chrono::steady_clock::now()); + + ASSERT_TRUE(copy.take().has_value()); + const auto second = observation.take(); + ASSERT_FALSE(second.has_value()); + EXPECT_EQ( + second.error(), + aethera::Take_Taskflow_Observation_Result::already_taken); +} + +TEST(Taskflow_Observation_Execution, Missing_Request_Has_No_Observer_Binding) { + aethera::detail::Taskflow_Observation_Execution execution; + EXPECT_EQ(execution.binding(), nullptr); +} + +TEST_F(Taskflow_Observation_Test, Moved_From_Request_Is_Unavailable_Not_Disabled) { + auto owner = std::move(observation); + const auto execution = + aethera::detail::Taskflow_Observation_Execution::try_start( + observation, + 1, + "test.graph", + nodes); + ASSERT_FALSE(execution.has_value()); + EXPECT_EQ( + execution.error(), + aethera::detail::Start_Taskflow_Observation_Result::unavailable); + EXPECT_EQ(owner.take().error(), + aethera::Take_Taskflow_Observation_Result::not_recorded); +} + +} diff --git a/kernel/kernel/test/time_thread/Timer_Service_Tests.cpp b/kernel/kernel/test/time_thread/Timer_Service_Tests.cpp new file mode 100644 index 0000000..6076b56 --- /dev/null +++ b/kernel/kernel/test/time_thread/Timer_Service_Tests.cpp @@ -0,0 +1,182 @@ +#include "time_thread/Timer_Service.hpp" +#include "time_thread/Timer_Time_Source.hpp" +#include "time_thread/detail/Timer_Scheduler.hpp" + +#include +#include + +#include +#include +#include + +namespace { + +using namespace std::chrono_literals; + +template +concept Timer_Duration_Api = requires( + aethera::Timer_Service& service, + aethera::Timer_Id id, + Duration duration, + aethera::Timer_Service::Callback callback) { + { service.schedule_after(duration, callback) } -> + std::same_as; + { service.schedule_every(duration, callback) } -> + std::same_as; + { service.reschedule(id, duration) } -> std::same_as; +}; + +static_assert(Timer_Duration_Api); +static_assert(Timer_Duration_Api>); +static_assert(Timer_Duration_Api< + std::chrono::duration>>); + +struct Mock_Timer_Time_Source : aethera::Timer_Time_Source { + MOCK_METHOD(Time_Point, now, (), (const, noexcept, override)); +}; + +struct Timer_Scheduler_Test : testing::Test { + Timer_Scheduler_Test() { + EXPECT_CALL(time_source, now()) + .Times(testing::AnyNumber()) + .WillRepeatedly([this] { + return current_time; + }); + } + + void advance_by(std::chrono::nanoseconds duration) { + current_time += duration; + scheduler.advance_to_current_time(); + } + + aethera::Timer_Time_Source::Time_Point current_time{}; + testing::StrictMock time_source; + aethera::detail::Timer_Scheduler scheduler{time_source}; +}; + +TEST(Timer_Service_Duration, Converts_Integral_Floating_And_Subnanosecond_Units) { + EXPECT_EQ( + aethera::detail::timer_duration_to_nanoseconds(2ms), + 2'000'000ns); + EXPECT_EQ( + aethera::detail::timer_duration_to_nanoseconds( + std::chrono::duration{1.5}), + 1'500'000ns); + EXPECT_EQ( + aethera::detail::timer_duration_to_nanoseconds( + std::chrono::duration{0.25}), + 1ns); + EXPECT_EQ( + aethera::detail::timer_duration_to_nanoseconds(-1s), + 0ns); + EXPECT_EQ( + aethera::detail::timer_duration_to_nanoseconds( + std::chrono::duration{ + std::numeric_limits::infinity()}), + std::chrono::nanoseconds::max()); +} + +TEST(Timer_Service_Fps, Converts_Fractional_Frame_Rates_And_Rejects_Invalid_Values) { + const auto fifty_fps = + aethera::detail::timer_interval_from_frames_per_second(50.0); + const auto fractional_fps = + aethera::detail::timer_interval_from_frames_per_second(29.97); + ASSERT_TRUE(fifty_fps.has_value()); + EXPECT_EQ(*fifty_fps, 20ms); + ASSERT_TRUE(fractional_fps.has_value()); + EXPECT_NEAR( + static_cast(fractional_fps->count()), + 33'366'700.0, + 1.0); + + const auto zero = + aethera::detail::timer_interval_from_frames_per_second(0.0); + const auto negative = + aethera::detail::timer_interval_from_frames_per_second(-60.0); + const auto infinite = + aethera::detail::timer_interval_from_frames_per_second( + std::numeric_limits::infinity()); + const auto not_a_number = + aethera::detail::timer_interval_from_frames_per_second( + std::numeric_limits::quiet_NaN()); + const auto too_slow = + aethera::detail::timer_interval_from_frames_per_second( + std::numeric_limits::denorm_min()); + + ASSERT_FALSE(zero.has_value()); + EXPECT_EQ( + zero.error(), + aethera::Schedule_Every_Fps_Result::invalid_frames_per_second); + ASSERT_FALSE(negative.has_value()); + EXPECT_EQ( + negative.error(), + aethera::Schedule_Every_Fps_Result::invalid_frames_per_second); + ASSERT_FALSE(infinite.has_value()); + EXPECT_EQ( + infinite.error(), + aethera::Schedule_Every_Fps_Result::invalid_frames_per_second); + ASSERT_FALSE(not_a_number.has_value()); + EXPECT_EQ( + not_a_number.error(), + aethera::Schedule_Every_Fps_Result::invalid_frames_per_second); + ASSERT_FALSE(too_slow.has_value()); + EXPECT_EQ( + too_slow.error(), + aethera::Schedule_Every_Fps_Result::interval_out_of_range); +} + +TEST_F(Timer_Scheduler_Test, After_Fires_Only_When_Mock_Time_Reaches_Deadline) { + testing::StrictMock> callback; + EXPECT_CALL(callback, Call()).Times(0); + scheduler.schedule(1, 25us, 0ns, callback.AsStdFunction()); + + advance_by(29us); + testing::Mock::VerifyAndClearExpectations(&callback); + + EXPECT_CALL(callback, Call()).Times(1); + advance_by(1us); + EXPECT_TRUE(scheduler.empty()); +} + +TEST_F(Timer_Scheduler_Test, Every_Uses_Mock_Time_Without_Drifting_Early) { + testing::StrictMock> callback; + scheduler.schedule(2, 20us, 20us, callback.AsStdFunction()); + + EXPECT_CALL(callback, Call()).Times(0); + advance_by(19us); + testing::Mock::VerifyAndClearExpectations(&callback); + + EXPECT_CALL(callback, Call()).Times(1); + advance_by(1us); + testing::Mock::VerifyAndClearExpectations(&callback); + + EXPECT_CALL(callback, Call()).Times(0); + advance_by(19us); + testing::Mock::VerifyAndClearExpectations(&callback); + + EXPECT_CALL(callback, Call()).Times(1); + advance_by(1us); + scheduler.cancel(2); +} + +TEST_F(Timer_Scheduler_Test, Reschedule_And_Cancel_Use_The_Same_Time_Authority) { + testing::StrictMock> callback; + scheduler.schedule(3, 100us, 0ns, callback.AsStdFunction()); + scheduler.reschedule(3, 25us); + + EXPECT_CALL(callback, Call()).Times(0); + advance_by(29us); + testing::Mock::VerifyAndClearExpectations(&callback); + + EXPECT_CALL(callback, Call()).Times(1); + advance_by(1us); + testing::Mock::VerifyAndClearExpectations(&callback); + + EXPECT_CALL(callback, Call()).Times(0); + scheduler.schedule(4, 20us, 0ns, callback.AsStdFunction()); + scheduler.cancel(4); + advance_by(200us); + EXPECT_TRUE(scheduler.empty()); +} + +} diff --git a/kernel/main.cmake b/kernel/main.cmake index f655bd7..7fd523a 100644 --- a/kernel/main.cmake +++ b/kernel/main.cmake @@ -58,7 +58,10 @@ if (Aethera_BUILD_TESTS) append_glob_source(Aethera_Kernel_module_test_sources "${Aethera_Kernel_test_dir}/concurrent" - "${Aethera_Kernel_test_dir}/model") + "${Aethera_Kernel_test_dir}/model" + "${Aethera_Kernel_test_dir}/task_flow" + "${Aethera_Kernel_test_dir}/time_thread" + "${Aethera_Kernel_test_dir}/function/frame_policy") foreach (Aethera_Kernel_module_test_source IN LISTS Aethera_Kernel_module_test_sources) if (NOT Aethera_Kernel_module_test_source MATCHES "\\.(c|cc|cpp|cxx)$") continue() @@ -71,7 +74,8 @@ if (Aethera_BUILD_TESTS) "${Aethera_Kernel_module_test_source}") target_link_libraries("${Aethera_Kernel_module_test_target}" PRIVATE Aethera_Kernel - GTest::gtest_main) + GTest::gtest_main + GTest::gmock) add_test(NAME "${Aethera_Kernel_module_test_target}" COMMAND "${Aethera_Kernel_cdb}" -c "g;q" "$")