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

提升观测系统

Browse Source 
隔离taskflow
This commit is contained in:
种太阳的热带美短
2026-08-25 10:15:59 +08:00
parent c0d5a13350
commit 2c3105cc07
37 changed files with 1538 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web_server/src/Plot.cpp
+188 -9
View File
@@ -202,6 +202,54 @@ void append_event_statistics_json(nlohmann::json& output,
}
}
nlohmann::json taskflow_trace_json(const Taskflow_Frame_Trace& trace) {
nlohmann::json graphs = nlohmann::json::array();
std::unordered_map<std::uint64_t, std::string> node_ids;
for (const auto& graph : trace.graphs) {
nlohmann::json nodes = nlohmann::json::array();
for (const auto& node : graph.nodes) {
node_ids.emplace(node.native_id, node.node_id);
nlohmann::json predecessors = nlohmann::json::array();
for (const auto native_id : node.predecessors)
predecessors.push_back(std::to_string(native_id));
nlohmann::json successors = nlohmann::json::array();
for (const auto native_id : node.successors)
successors.push_back(std::to_string(native_id));
nodes.push_back({
{"native_id", std::to_string(node.native_id)}, {"id", node.node_id},
{"parent_id", node.parent_node_id}, {"name", node.name},
{"type", node.type}, {"predecessors", std::move(predecessors)},
{"successors", std::move(successors)}});
}
graphs.push_back({
{"stage", graph.stage}, {"name", graph.taskflow_name},
{"submitted_ms", graph.submitted_ms},
{"finished_ms", graph.finished_ms},
{"completed", graph.completed}, {"nodes", std::move(nodes)}});
}
nlohmann::json executions = nlohmann::json::array();
for (const auto& task : trace.tasks) {
const auto found = node_ids.find(task.native_id);
executions.push_back({
{"native_id", std::to_string(task.native_id)},
{"node_id", found == node_ids.end() ? std::string{} : found->second},
{"worker_id", task.worker_id},
{"worker_queue_size", task.worker_queue_size},
{"worker_queue_capacity", task.worker_queue_capacity},
{"ready_ms", task.ready_ms}, {"started_ms", task.started_ms},
{"finished_ms", task.finished_ms},
{"duration_ms", task.duration_ms},
{"queue_wait_ms", task.queue_wait_ms}});
}
return {
{"sequence", trace.identity.sequence},
{"correlation_id", trace.identity.correlation_id},
{"created_time_unix_ns", trace.created_time_unix_ns},
{"worker_count", trace.worker_count},
{"graphs", std::move(graphs)},
{"executions", std::move(executions)}};
}
template <typename Scene_Object>
void dispatch_plot_input(Scene_Object& scene, const Plot_Input_Event& input) {
@@ -299,6 +347,17 @@ struct Plot::Private {
std::mutex render_preparation_mutex{}; /* 仅串行本图 CPU prepare；Scene 管理帧在途。 */
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 总数。 */
std::atomic_uint64_t policy_skip_count{}; /* 帧策略拒绝的 tick 总数。 */
std::atomic_uint64_t preparation_busy_count{}; /* 本 Plot Prepare 已在执行而跳过的提交次数。 */
std::atomic_uint64_t frame_slot_busy_count{}; /* 三个物理帧槽均被占用的提交次数。 */
std::atomic_uint64_t scene_rejection_count{}; /* Scene 单帧准入拒绝的提交次数。 */
std::atomic_uint64_t submitted_frame_count{}; /* 成功提交给 Scene 的帧总数。 */
std::atomic_size_t taskflow_trace_remaining{}; /* 尚待标记的实际渲染帧数。 */
mutable std::mutex taskflow_trace_mutex{}; /* 只保护低频请求结果的交换。 */
std::size_t taskflow_trace_requested_count{}; /* 当前批次请求总帧数。 */
std::vector<Taskflow_Frame_Trace> taskflow_traces{}; /* 已完成帧直接发布的 DAG 与 Observer 结果。 */
template <typename Scene_Object>
Private(std::unique_ptr<Scene_Object> value_scene,
@@ -319,6 +378,7 @@ struct Plot::Private {
void clock_tick(const Plot_Render_Tick& tick);
void render_frame(Plot_Render_Tick tick);
void queue_completed_frame(Render_Frame* frame);
[[nodiscard]] bool mark_taskflow_trace(Render_Frame& frame);
void fail(std::exception_ptr failure) noexcept;
};
@@ -403,7 +463,7 @@ void Plot::Private::consume_tick(std::weak_ptr<Plot> lifetime) {
if (!schedule_again) tick_task_scheduled = false;
}
if (schedule_again) {
aethera::schedule_task([lifetime] {
aethera::schedule_task("web.plot.tick.consume", [lifetime] {
const auto plot = lifetime.lock();
if (!plot) return;
try { plot->d->consume_tick(lifetime); }
@@ -415,20 +475,42 @@ 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.snapshot();
if (!pacing.render_enabled || pacing.mode == Frame_Pacing_Mode::manual) return;
if (!pacing.render_enabled || pacing.mode == Frame_Pacing_Mode::manual) {
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) return;
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);
}
bool Plot::Private::mark_taskflow_trace(Render_Frame& frame) {
auto remaining = taskflow_trace_remaining.load(std::memory_order_acquire);
while (remaining != 0) {
if (taskflow_trace_remaining.compare_exchange_weak(
remaining, remaining - 1, std::memory_order_acq_rel,
std::memory_order_acquire)) {
frame.request_taskflow_trace();
return true;
}
}
return false;
}
void Plot::Private::render_frame(Plot_Render_Tick tick) {
std::unique_lock preparation_lock(render_preparation_mutex, std::try_to_lock);
if (!preparation_lock.owns_lock()) return;
if (!preparation_lock.owns_lock()) {
preparation_busy_count.fetch_add(1, std::memory_order_relaxed);
return;
}
if (terminal_failure.load(std::memory_order_acquire)) return;
const auto streams = stream_snapshot();
const auto pacing = frame_policy.snapshot();
@@ -437,7 +519,10 @@ void Plot::Private::render_frame(Plot_Render_Tick tick) {
std::lock_guard lock(frame_mutex);
if (std::ranges::none_of(frame_slots, [](const Managed_Frame& slot) {
return slot.state == Frame_State::available;
})) return;
})) {
frame_slot_busy_count.fetch_add(1, std::memory_order_relaxed);
return;
}
}
tick.width = streams.width;
tick.height = streams.height;
@@ -446,7 +531,22 @@ void Plot::Private::render_frame(Plot_Render_Tick tick) {
* 进入 Scene::render 后只剩已经准备好的 Visual 批次与轻量提交；
* 共享 Render Domain 不承担业务数据生成。
*/
const auto update_started = std::chrono::steady_clock::now();
view->update(tick);
const auto update_elapsed = std::chrono::steady_clock::now() - update_started;
const auto tick_queue_elapsed = tick.issued_at.time_since_epoch().count() == 0
? std::chrono::steady_clock::duration::zero()
: update_started - tick.issued_at;
const auto record_plot_measurements = [&](Render_Frame& frame) {
const auto nanoseconds = [](std::chrono::steady_clock::duration duration) {
return static_cast<std::uint64_t>(std::max<std::int64_t>(0,
std::chrono::duration_cast<std::chrono::nanoseconds>(duration).count()));
};
frame.record(Frame_Trace_Measurement::plot_tick_queue_ns,
nanoseconds(tick_queue_elapsed));
frame.record(Frame_Trace_Measurement::plot_update_ns,
nanoseconds(update_elapsed));
};
const std::uint64_t sequence = next_frame_sequence++;
const Frame_Identity identity{sequence, tick.sequence == 0 ? sequence : tick.sequence};
@@ -474,6 +574,11 @@ void Plot::Private::render_frame(Plot_Render_Tick tick) {
if (slot.state == Frame_State::in_flight)
slot.state = Frame_State::available;
};
bool taskflow_trace_claimed{};
const auto restore_taskflow_trace_claim = [this, &taskflow_trace_claimed] {
if (!std::exchange(taskflow_trace_claimed, false)) return;
taskflow_trace_remaining.fetch_add(1, std::memory_order_release);
};
try {
if (auto* scene_2d = std::get_if<std::unique_ptr<Scene_2D>>(&scene)) {
@@ -481,26 +586,44 @@ void Plot::Private::render_frame(Plot_Render_Tick tick) {
output.begin(identity, pacing.video_enabled
? render_2d::Pixel_Format::rgba8
: Frame_2D::native_pixel_format);
taskflow_trace_claimed = mark_taskflow_trace(output);
record_plot_measurements(output);
(*scene_2d)->set<&Render_Scene_2D::Prop::viewport>(
Size{static_cast<int>(tick.width), static_cast<int>(tick.height)});
const auto result = (*scene_2d)->render(&output);
if (!result) rollback_unsubmitted();
if (!result) {
scene_rejection_count.fetch_add(1, std::memory_order_relaxed);
rollback_unsubmitted();
restore_taskflow_trace_claim();
} else {
taskflow_trace_claimed = false;
submitted_frame_count.fetch_add(1, std::memory_order_relaxed);
}
return;
}
auto& output = *std::get<std::unique_ptr<Frame_3D>>(managed->frame);
output.begin(identity, pacing.video_enabled ? Frame_3D_Output::pixels
: Frame_3D_Output::diagnostics,
Frame_3D::native_pixel_format);
taskflow_trace_claimed = mark_taskflow_trace(output);
record_plot_measurements(output);
auto& scene_3d = std::get<std::unique_ptr<Scene_3D>>(scene);
scene_3d->set<&Render_Scene_3D::Prop::viewport>(Extent{tick.width, tick.height});
const auto result = scene_3d->render(&output);
if (result == Render_Scene_3D::Render_Result::submitted) return;
if (result == Render_Scene_3D::Render_Result::submitted) {
taskflow_trace_claimed = false;
submitted_frame_count.fetch_add(1, std::memory_order_relaxed);
return;
}
scene_rejection_count.fetch_add(1, std::memory_order_relaxed);
rollback_unsubmitted();
restore_taskflow_trace_claim();
if (result == Render_Scene_3D::Render_Result::backend_unavailable)
throw std::runtime_error("3D render backend became unavailable before submission");
}
catch (...) {
rollback_unsubmitted();
restore_taskflow_trace_claim();
throw;
}
}
@@ -550,6 +673,11 @@ void Plot::Private::queue_completed_frame(Render_Frame* frame) {
};
try {
if (frame->taskflow_trace_requested()) {
auto trace = frame->taskflow_trace();
std::lock_guard lock(taskflow_trace_mutex);
taskflow_traces.push_back(std::move(trace));
}
const auto pacing = frame_policy.snapshot();
const auto identity = frame->identity();
Frame_Identity rendered_identity = identity;
@@ -589,7 +717,12 @@ void Plot::Private::queue_completed_frame(Render_Frame* frame) {
const auto published = std::make_shared<const Plot_Stream_Frame>(
Plot_Stream_Frame{{}, std::move(pixels)});
const auto publish_started = std::chrono::steady_clock::now();
publish(std::move(published));
frame->record(Frame_Trace_Measurement::plot_publish_ns,
static_cast<std::uint64_t>(std::max<std::int64_t>(0,
std::chrono::duration_cast<std::chrono::nanoseconds>(
std::chrono::steady_clock::now() - publish_started).count())));
retire();
}
catch (...) {
@@ -680,9 +813,12 @@ void Plot::configure_stream(Stream_Id stream, std::uint32_t width,
void Plot::schedule_render(Plot_Render_Tick tick) {
ensure_started();
if (d->terminal_failure.load(std::memory_order_acquire)) return;
d->received_tick_count.fetch_add(1, std::memory_order_relaxed);
bool schedule{};
{
std::lock_guard lock(d->tick_mutex);
if (d->pending_tick)
d->coalesced_tick_count.fetch_add(1, std::memory_order_relaxed);
d->pending_tick = tick;
if (!d->tick_task_scheduled) {
d->tick_task_scheduled = true;
@@ -691,7 +827,7 @@ void Plot::schedule_render(Plot_Render_Tick tick) {
}
if (!schedule) return;
const auto weak = weak_from_this();
aethera::schedule_task([weak] {
aethera::schedule_task("web.plot.tick.consume", [weak] {
const auto owner = weak.lock();
if (!owner) return;
try {
@@ -706,13 +842,14 @@ void Plot::schedule_render(Plot_Render_Tick tick) {
void Plot::render_once() {
ensure_started();
const auto weak = weak_from_this();
aethera::schedule_task([weak] {
aethera::schedule_task("web.plot.render.once", [weak] {
const auto owner = weak.lock();
if (!owner) return;
try {
const auto elapsed = std::chrono::steady_clock::now() -
owner->d->clock_origin;
owner->d->render_frame(Plot_Render_Tick{
std::chrono::steady_clock::now(),
0, std::chrono::duration<double, std::milli>(elapsed).count()});
}
catch (...) {
@@ -832,6 +969,14 @@ nlohmann::json Plot::diagnostics() const {
{"fixed_rate_fps", pacing.fixed_rate_fps},
{"render_enabled", pacing.render_enabled},
{"video_enabled", pacing.video_enabled}}},
{"plot_scheduler", {
{"received_ticks", d->received_tick_count.load(std::memory_order_relaxed)},
{"coalesced_ticks", d->coalesced_tick_count.load(std::memory_order_relaxed)},
{"policy_skips", d->policy_skip_count.load(std::memory_order_relaxed)},
{"preparation_busy", d->preparation_busy_count.load(std::memory_order_relaxed)},
{"frame_slot_busy", d->frame_slot_busy_count.load(std::memory_order_relaxed)},
{"scene_rejections", d->scene_rejection_count.load(std::memory_order_relaxed)},
{"submitted_frames", d->submitted_frame_count.load(std::memory_order_relaxed)}}},
{"frame_statistics", std::move(frame_statistics)},
{"input_statistics", std::move(input_statistics)}};
if (is_3d) {
@@ -865,6 +1010,40 @@ nlohmann::json Plot::diagnostics() const {
return output;
}
void Plot::request_taskflow_trace(std::size_t frame_count) {
if (frame_count == 0 || frame_count > 120)
throw std::invalid_argument("Taskflow trace frame_count must be between 1 and 120");
ensure_started();
{
std::lock_guard lock(d->taskflow_trace_mutex);
if (d->taskflow_trace_requested_count != d->taskflow_traces.size())
throw std::logic_error("A Taskflow frame trace request is already active");
d->taskflow_traces.clear();
d->taskflow_traces.reserve(frame_count);
d->taskflow_trace_requested_count = frame_count;
}
d->taskflow_trace_remaining.store(frame_count, std::memory_order_release);
}
nlohmann::json Plot::taskflow_trace() const {
nlohmann::json frames = nlohmann::json::array();
std::size_t requested{};
{
std::lock_guard lock(d->taskflow_trace_mutex);
requested = d->taskflow_trace_requested_count;
for (const auto& trace : d->taskflow_traces)
frames.push_back(taskflow_trace_json(trace));
}
const auto remaining = d->taskflow_trace_remaining.load(
std::memory_order_acquire);
return {
{"protocol", "aethera.taskflow.frames"}, {"version", 1},
{"requested", requested}, {"remaining", remaining},
{"captured", frames.size()},
{"complete", requested != 0 && frames.size() == requested},
{"frames", std::move(frames)}};
}
void Plot::reset_diagnostics() {
std::visit([](auto& scene) { scene->reset_frame_statistics(); }, d->scene);
}