psc/Aethera
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

加上全局调度器

Browse Source
This commit is contained in:
种太阳的热带美短
2026-08-26 11:29:46 +08:00
parent 5ecb12b8f7
commit 513f838874
13 changed files with 737 additions and 450 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web_server/src/Plot.cpp
+80 -67
View File
@@ -1,6 +1,8 @@
#include "Plot.hpp"
#include "Renderable_Adapter.hpp"
#include "Taskflow_Trace_Json.hpp"
#include <Frame_Pacing_Policy.hpp>
#include <Frame_Scheduler.hpp>
#include <nlohmann/json.hpp>
#include <magic_enum/magic_enum.hpp>
#include <render_2D/plottable/Plottables.hpp>
@@ -47,30 +49,22 @@ std::string exception_description(const std::exception_ptr& failure) {
return "empty Plot failure";
}
enum struct Frame_Pacing_Mode : std::uint8_t {
manual,
fixed_rate,
maximum_rate
};
struct Frame_Pacing_Properties {
bool render_enabled{true};
bool video_enabled{true};
Frame_Pacing_Mode mode{Frame_Pacing_Mode::fixed_rate}; /* 只决定共享时钟到达时是否调用 Scene::render。 */
double fixed_rate_fps{100.0}; /* 页面级时钟当前上限为每秒一百次。 */
};
struct Frame_Policy final {
public:
[[nodiscard]] Frame_Pacing_Properties read() const {
return *pacing.load(std::memory_order_acquire);
}
[[nodiscard]] Frame_Pacing_Properties read() const { return pacing.read(); }
[[nodiscard]] nlohmann::json schema() const;
[[nodiscard]] nlohmann::json write_prop(std::string_view key,
const nlohmann::json& value);
[[nodiscard]] bool accept_periodic_tick(double time_milliseconds) {
return pacing.accept_periodic_tick(time_milliseconds);
}
[[nodiscard]] bool request_immediate() { return pacing.request_immediate(); }
[[nodiscard]] double scheduled_rate_fps() const { return pacing.scheduled_rate_fps(); }
void frame_submitted() { pacing.frame_submitted(); }
[[nodiscard]] bool frame_completed() { return pacing.frame_completed(); }
void frame_rejected() { pacing.frame_rejected(); }
private:
std::atomic<std::shared_ptr<const Frame_Pacing_Properties>> pacing{
std::make_shared<const Frame_Pacing_Properties>()};
Frame_Pacing_Policy pacing{};
};
std::string_view pacing_mode_name(Frame_Pacing_Mode mode) {
@@ -103,26 +97,26 @@ nlohmann::json Frame_Policy::schema() const {
{"id", "frame-analysis"}, {"label", "渲染与媒体流水线"}, {"kind", "analysis"},
{"fields", nlohmann::json::array({
{{"key", "render_enabled"}, {"label", "持续渲染与采样"}, {"editor", "boolean"},
{"editable", true}, {"description", "控制共享帧时钟是否继续调用当前 Scene；画面隐藏不会修改此项。"},
{"technical_description", "Authoritative server-side render and sampling switch."},
{"editable", true}, {"description", "控制当前 Scene 的周期刷新；画面隐藏不会修改此项。"},
{"technical_description", "Authoritative per-scene periodic render switch."},
{"value", current.render_enabled}},
{{"key", "video_enabled"}, {"label", "图集视频传输"}, {"editor", "boolean"},
{"editable", true}, {"description", "控制完成帧是否进入页面级 RGBA 图集；默认开启，用于完整链路压测。"},
{"technical_description", "Authoritative tile publication switch for the shared gallery video."},
{"value", current.video_enabled}},
{{"key", "pacing_mode"}, {"label", "服务端帧策略"}, {"editor", "select"},
{"editable", true}, {"description", "只控制 Scene::render 的调用节奏；完成回调只负责归还帧并发布结果。"},
{"technical_description", "Render policy driven by the common 100 Hz gallery clock."},
{"editable", true}, {"description", "只控制 Scene::render(Frame*) 的调用节奏；Scene 的 Frame 所有权与接口保持不变。"},
{"technical_description", "Per-scene frame pacing policy backed by the Kernel scheduler."},
{"value", pacing_mode_name(current.mode)},
{"options", nlohmann::json::array({
{{"value", "manual"}, {"label", "手动渲染"}},
{{"value", "fixed_rate"}, {"label", "固定频率"}},
{{"value", "maximum_rate"}, {"label", "共享时钟满速"}}
{{"value", "maximum_rate"}, {"label", "最大频率"}}
})}},
{{"key", "fixed_rate_fps"}, {"label", "目标帧率"}, {"editor", "number"},
{"editable", true}, {"minimum", 0.1}, {"maximum", 100.0}, {"step", 0.1},
{"description", "固定频率模式下从页面级 100 Hz 时钟采样当前 Scene 的次数。"},
{"technical_description", "Per-plot rate selected from the common gallery render timeline."},
{"description", "当前 Scene 独立目标帧率；周期策略保存在 Kernel Frame_Pacing_Policy。"},
{"technical_description", "Independent per-scene target frame rate."},
{"value", current.fixed_rate_fps}}
})}
};
@@ -130,26 +124,12 @@ nlohmann::json Frame_Policy::schema() const {
nlohmann::json Frame_Policy::write_prop(std::string_view key,
const nlohmann::json& value) {
const auto update = [this](auto&& edit) {
auto current = pacing.load(std::memory_order_acquire);
for (;;) {
auto next = std::make_shared<Frame_Pacing_Properties>(*current);
edit(*next);
std::shared_ptr<const Frame_Pacing_Properties> desired = next;
if (pacing.compare_exchange_weak(
current, desired, std::memory_order_release,
std::memory_order_acquire))
return desired;
}
};
if (key == "render_enabled" || key == "video_enabled") {
if (!value.is_boolean())
return {{"success", false}, {"error", "frame policy switch requires a boolean"}};
const bool target = value.get<bool>();
update([&](Frame_Pacing_Properties& next) {
(key == "render_enabled" ? next.render_enabled : next.video_enabled) =
target;
});
if (key == "render_enabled") pacing.set_render_enabled(target);
else pacing.set_video_enabled(target);
return {{"success", true}, {"component", "frame-analysis"}, {"key", key},
{"value", target}};
}
@@ -159,7 +139,7 @@ nlohmann::json Frame_Policy::write_prop(std::string_view key,
const auto parsed = parse_pacing_mode(value.get_ref<const std::string&>());
if (!parsed)
return {{"success", false}, {"error", "unknown frame pacing mode"}};
update([&](Frame_Pacing_Properties& next) { next.mode = *parsed; });
pacing.set_mode(*parsed);
return {{"success", true}, {"component", "frame-analysis"}, {"key", key},
{"value", pacing_mode_name(*parsed)}};
}
@@ -169,9 +149,7 @@ nlohmann::json Frame_Policy::write_prop(std::string_view key,
const double next = value.get<double>();
if (!std::isfinite(next) || next < 0.1 || next > 100.0)
return {{"success", false}, {"error", "fixed_rate_fps must be between 0.1 and 100"}};
update([&](Frame_Pacing_Properties& properties) {
properties.fixed_rate_fps = next;
});
pacing.set_fixed_rate(next);
return {{"success", true}, {"component", "frame-analysis"}, {"key", key},
{"value", next}};
}
@@ -363,19 +341,20 @@ struct Plot::Private {
std::unique_ptr<Scene_View> view;
std::once_flag start_once;
std::weak_ptr<Plot> lifetime{}; /* 仅用于 completion 后重新投递 Taskflow，避免在 Scene callback 内重入 render。 */
std::atomic<std::shared_ptr<const Consumer_Map>> consumers{
std::make_shared<const Consumer_Map>()}; /* 低频订阅修改发布不可变版本。 */
std::atomic_uint64_t next_stream_id{1};
std::atomic<std::shared_ptr<const std::string>> terminal_failure{}; /* 首次 Plot Unknown Failure 的唯一终止状态。 */
std::uint64_t next_frame_sequence{1};
Frame_Policy frame_policy{};
Frame_Scheduler::Timer frame_timer{}; /* 每 Plot/Scene 只有轻量时间轮节点，不持有线程。 */
static constexpr std::size_t scene_frame_capacity{3};
std::array<Managed_Frame, scene_frame_capacity> frame_slots{}; /* Scene 借用的稳定三缓冲物理帧。 */
std::atomic_size_t callback_retired_slot{scene_frame_capacity};
Scene scene; /* 析构顺序保证 Scene 先停止，再释放物理帧。 */
std::atomic<std::shared_ptr<const Plot_Render_Tick>> pending_tick{};
std::atomic_bool tick_task_scheduled{}; /* 唯一短任务准入；不占用 Worker 等待。 */
double last_clock_render_time_ms{-std::numeric_limits<double>::infinity()};
std::chrono::steady_clock::time_point clock_origin{std::chrono::steady_clock::now()};
std::atomic_uint64_t received_tick_count{}; /* 页面时钟交付给本 Plot 的 tick 总数。 */
std::atomic_uint64_t coalesced_tick_count{}; /* 尚未消费时被更新 tick 替换的旧 tick 总数。 */
@@ -419,6 +398,7 @@ struct Plot::Private {
[[nodiscard]] Stream_Snapshot stream_snapshot() const;
void publish(std::shared_ptr<const Plot_Stream_Frame> frame) noexcept;
void consume_tick(std::weak_ptr<Plot> lifetime);
void refresh_schedule();
void clock_tick(const Plot_Render_Tick& tick);
void render_frame(Plot_Render_Tick tick);
void publish_completed_frame();
@@ -501,6 +481,14 @@ void Plot::Private::publish(
catch (...) {}
}
void Plot::Private::refresh_schedule() {
if (!frame_timer.valid()) return;
const auto current_consumers = consumers.load(std::memory_order_acquire);
const double fps = frame_policy.scheduled_rate_fps();
if (fps <= 0.0 || current_consumers->empty()) frame_timer.cancel();
else frame_timer.start_periodic(fps);
}
void Plot::Private::consume_tick(std::weak_ptr<Plot> lifetime) {
const auto tick = pending_tick.exchange({}, std::memory_order_acq_rel);
if (tick) clock_tick(*tick);
@@ -519,21 +507,10 @@ void Plot::Private::consume_tick(std::weak_ptr<Plot> lifetime) {
void Plot::Private::clock_tick(const Plot_Render_Tick& tick) {
if (terminal_failure.load(std::memory_order_acquire)) return;
const auto pacing = frame_policy.read();
if (!pacing.render_enabled || pacing.mode == Frame_Pacing_Mode::manual) {
if (!frame_policy.accept_periodic_tick(tick.time_milliseconds)) {
policy_skip_count.fetch_add(1, std::memory_order_relaxed);
return;
}
if (pacing.mode == Frame_Pacing_Mode::fixed_rate) {
if (tick.time_milliseconds < last_clock_render_time_ms)
last_clock_render_time_ms = -std::numeric_limits<double>::infinity();
const double interval = 1'000.0 / pacing.fixed_rate_fps;
if (tick.time_milliseconds - last_clock_render_time_ms + 0.01 < interval) {
policy_skip_count.fetch_add(1, std::memory_order_relaxed);
return;
}
}
last_clock_render_time_ms = tick.time_milliseconds;
render_frame(tick);
}
@@ -638,8 +615,10 @@ void Plot::Private::render_frame(Plot_Render_Tick tick) {
restore_taskflow_trace_claim();
} else {
taskflow_trace_claimed = false;
frame_policy.frame_submitted();
submitted_frame_count.fetch_add(1, std::memory_order_relaxed);
}
if (!result) frame_policy.frame_rejected();
return;
}
auto& output = *std::get<std::unique_ptr<Frame_3D>>(managed->frame);
@@ -653,9 +632,11 @@ void Plot::Private::render_frame(Plot_Render_Tick tick) {
const auto result = scene_3d->render(&output);
if (result == Render_Scene_3D::Render_Result::submitted) {
taskflow_trace_claimed = false;
frame_policy.frame_submitted();
submitted_frame_count.fetch_add(1, std::memory_order_relaxed);
return;
}
frame_policy.frame_rejected();
scene_rejection_count.fetch_add(1, std::memory_order_relaxed);
rollback_unsubmitted();
restore_taskflow_trace_claim();
@@ -794,6 +775,23 @@ void Plot::Private::retire_completed_frame(Render_Frame* frame) {
break;
}
}
if (frame_policy.frame_completed()) {
const auto weak = lifetime;
aethera::schedule_task("web.plot.render.deferred", [weak] {
const auto owner = weak.lock();
if (!owner) return;
try {
const auto now = std::chrono::steady_clock::now();
const auto elapsed = now - owner->d->clock_origin;
owner->d->render_frame(Plot_Render_Tick{
now, 0,
std::chrono::duration<double, std::milli>(elapsed).count()});
}
catch (...) {
owner->d->fail(std::current_exception());
}
});
}
}
void Plot::Private::attach_completion(
@@ -828,14 +826,24 @@ void Plot::attach_scene_completion(
void Plot::ensure_started() {
std::call_once(d->start_once, [this] {
const auto weak = weak_from_this();
d->lifetime = weak;
d->frame_timer = Frame_Scheduler::instance().make_timer(
[weak](Frame_Scheduler::Tick tick) {
if (const auto owner = weak.lock()) {
owner->schedule_render(Plot_Render_Tick{
tick.issued_at, tick.sequence, tick.time_milliseconds});
}
});
d->refresh_schedule();
/*
* 页面级帧时钟只触发 render；Scene 完成回调直接归还帧并发布结果。
* 回调不请求下一帧，也不重新投递任务；Scene 在回调返回时释放准入，
* 提交和页面级媒体图集能够按各自的资源域自然流水。
* Kernel Frame_Scheduler 只产生每个 Scene 独立的周期 tick；
* Scene::render(Frame*) 与外部 Frame 所有权保持原样。完成回调先归还
* 当前帧；若期间出现 immediate 请求，则回调返回后重新投递 Taskflow。
*/
if (auto* scene = std::get_if<std::unique_ptr<Scene_2D>>(&d->scene)) {
(*scene)->set_frame_callback(
[weak = weak_from_this()](Frame_2D* frame) {
[weak](Frame_2D* frame) {
if (auto owner = weak.lock()) {
try { owner->d->retire_completed_frame(frame); }
catch (...) { owner->d->fail(std::current_exception()); }
@@ -843,7 +851,7 @@ void Plot::ensure_started() {
});
} else {
std::get<std::unique_ptr<Scene_3D>>(d->scene)->set_frame_callback(
[weak = weak_from_this()](Frame_3D* frame) {
[weak](Frame_3D* frame) {
if (auto owner = weak.lock()) {
try { owner->d->retire_completed_frame(frame); }
catch (...) { owner->d->fail(std::current_exception()); }
@@ -869,6 +877,7 @@ Plot::Stream_Id Plot::subscribe(Stream_Handler handler) {
std::memory_order_acquire))
break;
}
d->refresh_schedule();
if (d->terminal_failure.load(std::memory_order_acquire)) {
const auto failure = d->terminal_failure.load(std::memory_order_acquire);
try {
@@ -896,8 +905,9 @@ void Plot::unsubscribe(Stream_Id stream) {
if (d->consumers.compare_exchange_weak(
current, desired, std::memory_order_release,
std::memory_order_acquire))
return;
break;
}
d->refresh_schedule();
}
void Plot::configure_stream(Stream_Id stream, std::uint32_t width,
@@ -941,6 +951,7 @@ void Plot::schedule_render(Plot_Render_Tick tick) {
void Plot::render_once() {
ensure_started();
if (!d->frame_policy.request_immediate()) return;
const auto weak = weak_from_this();
aethera::schedule_task("web.plot.render.once", [weak] {
const auto owner = weak.lock();
@@ -987,9 +998,11 @@ nlohmann::json Plot::write_prop(std::string_view component,
std::string_view key,
const nlohmann::json& value) {
ensure_started();
return component == "frame-analysis"
? d->frame_policy.write_prop(key, value)
: d->view->write_prop(component, key, value);
if (component != "frame-analysis")
return d->view->write_prop(component, key, value);
auto result = d->frame_policy.write_prop(key, value);
if (result.value("success", false)) d->refresh_schedule();
return result;
}
nlohmann::json Plot::generate_data(const nlohmann::json& input) {