1866 lines
83 KiB
C++
1866 lines
83 KiB
C++
#include "Plot.hpp"
|
||
#include "Renderable_Adapter.hpp"
|
||
#include "Taskflow_Trace_Json.hpp"
|
||
#include <Frame_Policy/Frame_Policy.hpp>
|
||
#include <Frame_Policy/Frame_Policy_3D.hpp>
|
||
#include <Frame_Policy/Frame_Scheduler.hpp>
|
||
#include <concurrentqueue-1.0.5/concurrentqueue.h>
|
||
#include <nlohmann/json.hpp>
|
||
#include <magic_enum/magic_enum.hpp>
|
||
#include <render_2D/plottable/Plottables.hpp>
|
||
#include <render_3D/Render_3D.hpp>
|
||
#include <render_3D/Gpu_Completion_State.hpp>
|
||
#include <render_common.hpp>
|
||
#include <algorithm>
|
||
#include <array>
|
||
#include <atomic>
|
||
#include <chrono>
|
||
#include <cmath>
|
||
#include <concepts>
|
||
#include <exception>
|
||
#include <initializer_list>
|
||
#include <limits>
|
||
#include <memory>
|
||
#include <optional>
|
||
#include <span>
|
||
#include <stdexcept>
|
||
#include <unordered_map>
|
||
#include <unordered_set>
|
||
#include <utility>
|
||
#include <variant>
|
||
#include <vector>
|
||
|
||
namespace aethera::web {
|
||
namespace {
|
||
using namespace render_2d;
|
||
using namespace render_3d;
|
||
using Scene_2D = Render_Scene_2D;
|
||
using Scene_3D = Render_Scene_3D;
|
||
constexpr std::uint16_t plot_stream_protocol_version{9};
|
||
constexpr std::size_t diagnostic_window_capacity{600};
|
||
|
||
|
||
std::string exception_description(const std::exception_ptr& failure) {
|
||
try {
|
||
if (failure) std::rethrow_exception(failure);
|
||
}
|
||
catch (const std::exception& error) {
|
||
return error.what();
|
||
}
|
||
catch (...) {
|
||
return "non-standard Plot failure";
|
||
}
|
||
return "empty Plot failure";
|
||
}
|
||
|
||
std::string_view pacing_mode_name(Frame_Pacing_Mode mode) {
|
||
const auto name = magic_enum::enum_name(mode);
|
||
if (name.empty()) throw std::logic_error("unknown frame pacing mode");
|
||
return name;
|
||
}
|
||
|
||
std::optional<Frame_Pacing_Mode> parse_pacing_mode(std::string_view value) {
|
||
return magic_enum::enum_cast<Frame_Pacing_Mode>(value);
|
||
}
|
||
|
||
std::string_view pixel_format_name(render_2d::Pixel_Format format) {
|
||
const auto name = magic_enum::enum_name(format);
|
||
if (name.empty()) throw std::logic_error("unknown 2D pixel format");
|
||
return name;
|
||
}
|
||
|
||
std::string_view pixel_format_name(render_3d::Pixel_Format format) {
|
||
switch (format) {
|
||
case render_3d::Pixel_Format::rgba8_unorm:
|
||
return "rgba8";
|
||
}
|
||
throw std::logic_error("unknown 3D pixel format");
|
||
}
|
||
|
||
const Frame_Policy::State& frame_policy_common_state(
|
||
const Frame_Policy& policy) noexcept {
|
||
return policy.read_state<Frame_Policy::Base_Tag>();
|
||
}
|
||
|
||
const Frame_Policy::State& frame_policy_common_state(
|
||
const Frame_Policy_3D& policy) noexcept {
|
||
return policy.read_state<Frame_Policy_3D::Base_Tag>().common;
|
||
}
|
||
|
||
template <typename Policy>
|
||
nlohmann::json frame_policy_schema(const Policy& policy) {
|
||
const auto& current = frame_policy_common_state(policy);
|
||
return {
|
||
{"id", "frame-analysis"}, {"label", "渲染与媒体流水线"}, {"kind", "analysis"},
|
||
{"fields", nlohmann::json::array({
|
||
{{"key", "render_enabled"}, {"label", "持续渲染与采样"}, {"editor", "boolean"},
|
||
{"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", "控制完成帧是否进入页面级采样器；2D BGRA 与 3D 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(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", "最大吞吐"}}
|
||
})}},
|
||
{{"key", "fixed_rate_fps"}, {"label", "目标帧率"}, {"editor", "number"},
|
||
{"editable", true}, {"minimum", 0.1}, {"maximum", 100.0}, {"step", 0.1},
|
||
{"description", "仅 fixed_rate 使用；maximum_rate 在完成准入释放后异步自驱下一帧。"},
|
||
{"technical_description", "Independent per-scene target frame rate."},
|
||
{"value", current.fixed_rate_fps}}
|
||
})}
|
||
};
|
||
}
|
||
|
||
template <typename Policy>
|
||
nlohmann::json write_frame_policy_prop(Policy& pacing,
|
||
std::string_view key,
|
||
const nlohmann::json& value) {
|
||
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>();
|
||
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}};
|
||
}
|
||
if (key == "pacing_mode") {
|
||
if (!value.is_string())
|
||
return {{"success", false}, {"error", "pacing_mode requires a string"}};
|
||
const auto parsed = parse_pacing_mode(value.get_ref<const std::string&>());
|
||
if (!parsed)
|
||
return {{"success", false}, {"error", "unknown frame pacing mode"}};
|
||
pacing.set_mode(*parsed);
|
||
return {{"success", true}, {"component", "frame-analysis"}, {"key", key},
|
||
{"value", pacing_mode_name(*parsed)}};
|
||
}
|
||
if (key == "fixed_rate_fps") {
|
||
if (!value.is_number())
|
||
return {{"success", false}, {"error", "fixed_rate_fps requires a number"}};
|
||
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"}};
|
||
pacing.set_fixed_rate(next);
|
||
return {{"success", true}, {"component", "frame-analysis"}, {"key", key},
|
||
{"value", next}};
|
||
}
|
||
return {{"success", false}, {"error", "unknown frame runtime property"}};
|
||
}
|
||
|
||
nlohmann::json frame_policy_state_json(const Frame_Policy::State& state) {
|
||
const auto milliseconds = [](std::uint64_t nanoseconds) {
|
||
return static_cast<double>(nanoseconds) / 1'000'000.0;
|
||
};
|
||
const auto ratio = [](std::uint64_t numerator, std::uint64_t denominator) {
|
||
return denominator == 0 ? 0.0 :
|
||
static_cast<double>(numerator) / static_cast<double>(denominator);
|
||
};
|
||
const auto observed_ns = state.observed_until_ns > state.observation_started_ns
|
||
? state.observed_until_ns - state.observation_started_ns : 0U;
|
||
const auto observed_seconds = static_cast<double>(observed_ns) / 1'000'000'000.0;
|
||
const auto rate = [observed_seconds](std::uint64_t count) {
|
||
return observed_seconds > 0.0
|
||
? static_cast<double>(count) / observed_seconds : 0.0;
|
||
};
|
||
const auto completion_span = state.last_completion_ns > state.first_completion_ns
|
||
? state.last_completion_ns - state.first_completion_ns : 0U;
|
||
const auto effective_fps = completion_span != 0 && state.completed_frame_count > 1
|
||
? static_cast<double>(state.completed_frame_count - 1U) *
|
||
1'000'000'000.0 / static_cast<double>(completion_span)
|
||
: 0.0;
|
||
const auto interval_mean = state.completion_interval_count == 0 ? 0.0 :
|
||
static_cast<double>(state.completion_interval_total_ns) /
|
||
static_cast<double>(state.completion_interval_count);
|
||
const auto interval_variance = state.completion_interval_count == 0 ? 0.0 :
|
||
std::max(0.0,
|
||
state.completion_interval_squared_total_ns2 /
|
||
static_cast<double>(state.completion_interval_count) -
|
||
interval_mean * interval_mean);
|
||
const auto target_achievement =
|
||
state.mode == Frame_Pacing_Mode::fixed_rate && state.fixed_rate_fps > 0.0
|
||
? effective_fps / state.fixed_rate_fps : 0.0;
|
||
return {
|
||
{"generation", state.generation},
|
||
{"configuration", {
|
||
{"mode", pacing_mode_name(state.mode)},
|
||
{"render_enabled", state.render_enabled},
|
||
{"video_enabled", state.video_enabled},
|
||
{"fixed_rate_fps", state.fixed_rate_fps}}},
|
||
{"observation", {
|
||
{"duration_ms", milliseconds(observed_ns)},
|
||
{"request_count", state.request_count},
|
||
{"submitted_frame_count", state.submitted_frame_count},
|
||
{"completed_frame_count", state.completed_frame_count},
|
||
{"active_frame_count", state.active_frame_count}}},
|
||
{"throughput", {
|
||
{"request_rate_fps", rate(state.request_count)},
|
||
{"submission_rate_fps", rate(state.submitted_frame_count)},
|
||
{"completion_rate_fps", effective_fps},
|
||
{"target_achievement_ratio", target_achievement},
|
||
{"latest_frame_interval_ms",
|
||
milliseconds(state.latest_completion_interval_ns)},
|
||
{"average_frame_interval_ms", milliseconds(
|
||
state.completion_interval_count == 0 ? 0U :
|
||
state.completion_interval_total_ns /
|
||
state.completion_interval_count)},
|
||
{"frame_interval_jitter_ms",
|
||
std::sqrt(interval_variance) / 1'000'000.0}}},
|
||
{"requests", {
|
||
{"periodic", state.periodic_request_count},
|
||
{"immediate", state.immediate_request_count},
|
||
{"maximum_rate", state.maximum_rate_request_count},
|
||
{"accepted", state.accepted_request_count},
|
||
{"coalesced", state.coalesced_request_count},
|
||
{"policy_rejected", state.policy_rejection_count},
|
||
{"frame_slot_backpressure", state.frame_slot_backpressure_count},
|
||
{"scene_rejected", state.scene_rejection_count},
|
||
{"acceptance_ratio", ratio(
|
||
state.accepted_request_count, state.request_count)},
|
||
{"coalescing_ratio", ratio(
|
||
state.coalesced_request_count, state.request_count)},
|
||
{"backpressure_ratio", ratio(
|
||
state.frame_slot_backpressure_count,
|
||
state.accepted_request_count)}}},
|
||
{"latency", {
|
||
{"latest_tick_queue_ms", milliseconds(state.latest_tick_queue_ns)},
|
||
{"average_tick_queue_ms", milliseconds(
|
||
state.submitted_frame_count == 0 ? 0U :
|
||
state.tick_queue_total_ns / state.submitted_frame_count)},
|
||
{"maximum_tick_queue_ms", milliseconds(state.maximum_tick_queue_ns)},
|
||
{"latest_completion_ms", milliseconds(
|
||
state.latest_completion_latency_ns)},
|
||
{"average_completion_ms", milliseconds(
|
||
state.completed_frame_count == 0 ? 0U :
|
||
state.completion_latency_total_ns /
|
||
state.completed_frame_count)},
|
||
{"maximum_completion_ms", milliseconds(
|
||
state.maximum_completion_latency_ns)}}},
|
||
{"last_frame", {
|
||
{"sequence", state.last_frame_sequence},
|
||
{"request_source", magic_enum::enum_name(
|
||
state.last_request_source)}}}
|
||
};
|
||
}
|
||
|
||
void append_statistic_json(nlohmann::json& output,
|
||
const Frame_Statistics_State& state) {
|
||
for (const auto statistic : magic_enum::enum_values<Frame_Statistic>()) {
|
||
if (statistic == Frame_Statistic::count) continue;
|
||
const auto& value =
|
||
state.values[static_cast<std::size_t>(statistic)];
|
||
if (value.count == 0) continue;
|
||
output[magic_enum::enum_name(statistic)] = {
|
||
{"count", value.count}, {"latest", value.latest},
|
||
{"minimum", value.minimum}, {"maximum", value.maximum},
|
||
{"average", value.average},
|
||
{"trimmed_average", value.trimmed_average},
|
||
{"variability", value.variability}, {"p50", value.p50},
|
||
{"p95", value.p95}, {"p99", value.p99}};
|
||
}
|
||
}
|
||
|
||
void append_event_statistics_json(nlohmann::json& output,
|
||
const Event_Statistics_State& state) {
|
||
for (const auto type : magic_enum::enum_values<Event_Type>()) {
|
||
auto& event = output[magic_enum::enum_name(type)];
|
||
const auto& values = state.values[static_cast<std::size_t>(type)];
|
||
for (const auto statistic : magic_enum::enum_values<Event_Statistic>()) {
|
||
if (statistic == Event_Statistic::count) continue;
|
||
const auto& value = values[static_cast<std::size_t>(statistic)];
|
||
if (value.count == 0) continue;
|
||
event[magic_enum::enum_name(statistic)] = {
|
||
{"count", value.count}, {"latest", value.latest},
|
||
{"minimum", value.minimum}, {"maximum", value.maximum},
|
||
{"average", value.average},
|
||
{"trimmed_average", value.trimmed_average},
|
||
{"variability", value.variability}, {"p50", value.p50},
|
||
{"p95", value.p95}, {"p99", value.p99}};
|
||
}
|
||
if (event.empty()) output.erase(std::string{magic_enum::enum_name(type)});
|
||
}
|
||
}
|
||
|
||
nlohmann::json datoviz_observation_json(
|
||
const Datoviz_Frame_Observation& value) {
|
||
const auto milliseconds = [](std::uint64_t nanoseconds) {
|
||
return static_cast<double>(nanoseconds) / 1'000'000.0;
|
||
};
|
||
nlohmann::json result{
|
||
{"render_sequence", value.render_sequence},
|
||
{"path", magic_enum::enum_name(value.path)},
|
||
{"gpu_timing_requested", value.gpu_timing_requested},
|
||
{"readback_requested", value.readback_requested},
|
||
{"timings_ms", {
|
||
{"render_domain_queue_wait",
|
||
milliseconds(value.render_domain_queue_wait_ns)},
|
||
{"apply", milliseconds(value.apply_ns)},
|
||
{"emit", milliseconds(value.emit_ns)},
|
||
{"execute", milliseconds(value.execute_ns)},
|
||
{"submit", milliseconds(value.submit_ns)},
|
||
{"gpu_fence_wait", milliseconds(value.gpu_fence_wait_ns)},
|
||
{"readback", milliseconds(value.readback_ns)}}},
|
||
{"traffic", {
|
||
{"uploaded_bytes", value.uploaded_bytes},
|
||
{"readback_bytes", value.readback_bytes}}},
|
||
{"frame_plan", {
|
||
{"resource_version", value.artifact_resource_version},
|
||
{"frame_index", value.artifact_frame_index},
|
||
{"status", value.artifact_status}}},
|
||
{"validation", {
|
||
{"performed", value.validation_performed},
|
||
{"ok", value.validation_ok},
|
||
{"code", value.validation_code},
|
||
{"command_index", value.validation_command_index}}}
|
||
};
|
||
if (value.gpu) {
|
||
result["gpu_ms"] = {
|
||
{"render", milliseconds(value.gpu->render_ns)},
|
||
{"transition", milliseconds(value.gpu->transition_ns)},
|
||
{"copy", milliseconds(value.gpu->copy_ns)},
|
||
{"total", milliseconds(value.gpu->total_ns)}};
|
||
}
|
||
if (!value.artifact_json.empty())
|
||
result["frame_plan"]["artifact_json"] = value.artifact_json;
|
||
return result;
|
||
}
|
||
|
||
}
|
||
|
||
nlohmann::json taskflow_trace_json(
|
||
const Taskflow_Frame_Trace& trace,
|
||
const nlohmann::json& captured_components,
|
||
const nlohmann::json& captured_backend) {
|
||
nlohmann::json markers = nlohmann::json::object();
|
||
for (const auto& marker : trace.markers)
|
||
markers[magic_enum::enum_name(marker.marker)] =
|
||
static_cast<double>(marker.elapsed_ns) / 1'000'000.0;
|
||
nlohmann::json measurements = nlohmann::json::object();
|
||
for (const auto& measurement : trace.measurements)
|
||
measurements[magic_enum::enum_name(measurement.measurement)] =
|
||
static_cast<double>(measurement.value_ns) / 1'000'000.0;
|
||
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));
|
||
nlohmann::json attributes = nlohmann::json::object();
|
||
for (const auto& [key, value] : node.attributes)
|
||
attributes[key] = value;
|
||
nlohmann::json encoded{
|
||
{"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)},
|
||
{"attributes", std::move(attributes)}};
|
||
const auto owner = encoded["attributes"].value(
|
||
"owner_component", std::string{});
|
||
if (!owner.empty() && captured_components.contains(owner)) {
|
||
const auto& captured = captured_components.at(owner);
|
||
encoded["owner"] = {
|
||
{"component", owner},
|
||
{"label", captured.value("label", owner)},
|
||
{"kind", captured.value("kind", std::string{})}};
|
||
encoded["prop"] = captured.value("prop", nlohmann::json::object());
|
||
encoded["state"] = captured.value("state", nlohmann::json::object());
|
||
}
|
||
nodes.push_back(std::move(encoded));
|
||
}
|
||
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}, {"entered_ms", task.entered_ms},
|
||
{"started_ms", task.started_ms}, {"finished_ms", task.finished_ms},
|
||
{"completed_ms", task.completed_ms},
|
||
{"duration_ms", task.duration_ms},
|
||
{"cooperative_wait_ms", task.cooperative_wait_ms},
|
||
{"observer_entry_ms", task.observer_entry_ms},
|
||
{"observer_exit_ms", task.observer_exit_ms},
|
||
{"queue_wait_ms", task.queue_wait_ms}});
|
||
}
|
||
nlohmann::json result{
|
||
{"sequence", trace.identity.sequence},
|
||
{"correlation_id", trace.identity.correlation_id},
|
||
{"request_source", magic_enum::enum_name(trace.request_source)},
|
||
{"frame_policy", frame_policy_state_json(trace.frame_policy)},
|
||
{"created_time_unix_ns", trace.created_time_unix_ns},
|
||
{"worker_count", trace.worker_count},
|
||
{"markers", std::move(markers)},
|
||
{"measurements", std::move(measurements)},
|
||
{"graphs", std::move(graphs)},
|
||
{"executions", std::move(executions)}};
|
||
if (!captured_backend.empty()) result["datoviz"] = captured_backend;
|
||
return result;
|
||
}
|
||
|
||
namespace {
|
||
|
||
template <typename Scene_Object>
|
||
void dispatch_plot_input(Scene_Object& scene, const Plot_Input_Event& input) {
|
||
const auto dispatch = [&](auto event) {
|
||
scene.template submit_stream<aethera::Scene_Event_Stream_Tag>(std::move(event));
|
||
};
|
||
const auto apply_pointer = [&](auto& event) {
|
||
event.position = input.position;
|
||
event.global_position = input.global_position;
|
||
event.button = input.button;
|
||
event.buttons = input.buttons;
|
||
event.modifiers = input.modifiers;
|
||
};
|
||
switch (input.type) {
|
||
case Event_Type::pointer_move:
|
||
case Event_Type::pointer_press:
|
||
case Event_Type::pointer_release: {
|
||
auto event = scene.template make_event<Basic_Pointer_Event<Point_F>>(
|
||
input.type);
|
||
apply_pointer(*event);
|
||
dispatch(std::move(event));
|
||
break;
|
||
}
|
||
case Event_Type::wheel: {
|
||
auto event = scene.template make_event<Basic_Wheel_Event<Point_F>>();
|
||
apply_pointer(*event);
|
||
event->pixel_delta_x = input.pixel_delta_x;
|
||
event->pixel_delta_y = input.pixel_delta_y;
|
||
event->angle_delta_x = input.angle_delta_x;
|
||
event->angle_delta_y = input.angle_delta_y;
|
||
dispatch(std::move(event));
|
||
break;
|
||
}
|
||
case Event_Type::key_press:
|
||
case Event_Type::key_release: {
|
||
auto event = scene.template make_event<Key_Event>(input.type);
|
||
event->key = input.key;
|
||
event->native_key = input.native_key;
|
||
event->modifiers = input.modifiers;
|
||
event->auto_repeat = input.auto_repeat;
|
||
dispatch(std::move(event));
|
||
break;
|
||
}
|
||
default:
|
||
dispatch(scene.template make_event<Event>(input.type));
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
|
||
struct Plot::Private {
|
||
using Scene = std::variant<std::unique_ptr<Scene_2D>, std::unique_ptr<Scene_3D>>;
|
||
using Frame = std::variant<std::unique_ptr<Frame_2D>, std::unique_ptr<Frame_3D>>;
|
||
|
||
enum struct Frame_State : std::uint8_t {
|
||
available,
|
||
rendering, /* Scene::advance -> Plot pixel publish，不可重入。 */
|
||
consuming /* 外接 Taskflow 正在消费已发布帧；允许下一帧渲染。 */
|
||
};
|
||
|
||
enum struct Render_Admission_State : std::uint8_t {
|
||
ready,
|
||
rendering,
|
||
frame_slots_exhausted
|
||
};
|
||
|
||
struct Managed_Frame {
|
||
std::chrono::microseconds presentation_time{}; /* 共享页面时钟产生的媒体时间戳。 */
|
||
std::chrono::steady_clock::time_point tick_issued_at{}; /* 本逻辑帧请求进入 Plot 的时刻。 */
|
||
std::chrono::steady_clock::time_point submitted_at{}; /* Scene 接受本逻辑帧的时刻。 */
|
||
Frame frame{}; /* 三缓冲物理槽拥有且反复承载逻辑帧。 */
|
||
std::atomic<Frame_State> state{Frame_State::available}; /* 本槽唯一生命周期状态。 */
|
||
std::atomic_size_t diagnostic_readers{}; /* 无锁诊断读取认领；非零时该槽不可复用。 */
|
||
std::uint64_t statistics_generation{}; /* 与本槽中完成帧统计共同发布。 */
|
||
Frame_Statistics_State statistics{}; /* 统计结果归属当前完成帧，不在 Plot 复制。 */
|
||
std::atomic<Managed_Frame*> retired_next{}; /* 无锁退役队列的槽内侵入链接。 */
|
||
};
|
||
|
||
struct Consumer {
|
||
Stream_Handler handler;
|
||
std::uint32_t width{};
|
||
std::uint32_t height{};
|
||
};
|
||
using Consumer_Map = std::unordered_map<Stream_Id, Consumer>;
|
||
|
||
struct Stream_Snapshot {
|
||
std::shared_ptr<const Consumer_Map> consumers;
|
||
std::uint32_t width{};
|
||
std::uint32_t height{};
|
||
};
|
||
|
||
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};
|
||
std::unique_ptr<Frame_Policy> frame_policy_2d{};
|
||
std::unique_ptr<Frame_Policy_3D> frame_policy_3d{};
|
||
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<Managed_Frame*> latest_statistics_frame{}; /* 只定位权威帧槽，不保存统计副本。 */
|
||
std::atomic<Managed_Frame*> retired_frames{}; /* 多回调生产、唯一 Taskflow 任务消费。 */
|
||
std::atomic_bool retired_frame_task_scheduled{};
|
||
Scene scene; /* 析构顺序保证 Scene 先停止，再释放物理帧。 */
|
||
moodycamel::ConcurrentQueue<Plot_Render_Tick> tick_requests{}; /* 多生产者提交、唯一短任务消费的帧请求流。 */
|
||
std::optional<Plot_Render_Tick> deferred_tick{}; /* 仅 tick consumer 任务访问的 latest 延后请求。 */
|
||
std::atomic_uint64_t tick_request_generation{}; /* 请求入队后推进，用于关闭 consumer 尾部唤醒竞争窗口。 */
|
||
std::atomic_bool tick_task_scheduled{}; /* 唯一短任务准入；不占用 Worker 等待。 */
|
||
std::atomic<Render_Admission_State> render_admission{Render_Admission_State::ready}; /* Plot 渲染准入及物理槽背压的唯一状态源。 */
|
||
std::chrono::steady_clock::time_point clock_origin{std::chrono::steady_clock::now()};
|
||
std::atomic_size_t taskflow_trace_remaining{}; /* 尚待标记的实际渲染帧数。 */
|
||
static constexpr std::size_t maximum_taskflow_trace_frames{120};
|
||
/* 高 32 位 requested，低 32 位 captured。每槽只发布一次不可变 Trace，
|
||
* GET 直接读取已发布槽位，不复制或重排整个历史容器。 */
|
||
std::atomic_uint64_t taskflow_trace_control{};
|
||
std::array<std::atomic<std::shared_ptr<const nlohmann::json>>,
|
||
maximum_taskflow_trace_frames> taskflow_trace_slots{};
|
||
std::atomic_size_t post_publish_trace_remaining{}; /* 仅捕获 publish 后外接 DAG 的剩余样本。 */
|
||
std::atomic_uint64_t post_publish_trace_control{}; /* 高 32 位 requested，低 32 位 captured。 */
|
||
std::array<std::atomic<std::shared_ptr<const nlohmann::json>>,
|
||
maximum_taskflow_trace_frames> post_publish_trace_slots{};
|
||
Task_Node completion_tail{}; /* Scene 图内固定停在 plot.frame.publish。 */
|
||
Task_Graph post_publish_graph{"plot.post_publish"}; /* publish 后外接 DAG；不再占用 Scene render admission。 */
|
||
Task_Node post_publish_tail{};
|
||
bool has_post_publish_tail{};
|
||
std::atomic_bool post_publish_busy{};
|
||
std::vector<std::unique_ptr<Task_Graph>> completion_extensions{}; /* 生命周期覆盖 post-publish module 借用。 */
|
||
Frame_Statistics_Accumulator completed_frame_statistics{diagnostic_window_capacity};
|
||
std::uint64_t applied_statistics_generation{}; /* 仅完成帧退役任务读写。 */
|
||
std::atomic_uint64_t statistics_generation{}; /* reset 只推进代次，不触碰单写者累加器。 */
|
||
|
||
template <typename Scene_Object>
|
||
Private(std::unique_ptr<Scene_Object> value_scene,
|
||
std::unique_ptr<Scene_View> value_view)
|
||
: view(std::move(value_view)), scene(std::move(value_scene)) {
|
||
if constexpr (std::same_as<Scene_Object, Scene_2D>) {
|
||
auto built_policy = Frame_Policy::Builder<Frame_Policy>{}.build();
|
||
if (!built_policy)
|
||
throw std::logic_error("2D Frame Policy dependency graph is invalid");
|
||
frame_policy_2d = std::move(*built_policy);
|
||
}
|
||
else {
|
||
auto built_policy =
|
||
Frame_Policy_3D::Builder<Frame_Policy_3D>{}.build();
|
||
if (!built_policy)
|
||
throw std::logic_error("3D Frame Policy dependency graph is invalid");
|
||
frame_policy_3d = std::move(*built_policy);
|
||
frame_policy_3d->set_mode(Frame_Pacing_Mode::maximum_rate);
|
||
static_cast<void>(frame_policy_3d->consume_events());
|
||
}
|
||
for (auto& slot : frame_slots) {
|
||
if constexpr (std::same_as<Scene_Object, Scene_2D>)
|
||
slot.frame = std::make_unique<Frame_2D>(Frame_Identity{});
|
||
else
|
||
slot.frame = std::make_unique<Frame_3D>(Frame_Identity{});
|
||
}
|
||
if constexpr (std::same_as<Scene_Object, Scene_2D>) {
|
||
auto& completion = std::get<std::unique_ptr<Scene_2D>>(scene)
|
||
->completion_taskflow();
|
||
completion_tail = completion.add("plot.frame.publish", [this] {
|
||
publish_completed_frame();
|
||
});
|
||
completion_tail.describe("owner", "plot")
|
||
.describe("stage", "completed pixels publish");
|
||
}
|
||
}
|
||
|
||
template <typename Callback>
|
||
decltype(auto) with_frame_policy(Callback&& callback) {
|
||
if (frame_policy_2d)
|
||
return std::forward<Callback>(callback)(*frame_policy_2d);
|
||
return std::forward<Callback>(callback)(*frame_policy_3d);
|
||
}
|
||
|
||
template <typename Callback>
|
||
decltype(auto) with_frame_policy(Callback&& callback) const {
|
||
if (frame_policy_2d)
|
||
return std::forward<Callback>(callback)(*frame_policy_2d);
|
||
return std::forward<Callback>(callback)(*frame_policy_3d);
|
||
}
|
||
|
||
[[nodiscard]] const Frame_Policy::State& pacing_state() const noexcept {
|
||
if (frame_policy_2d)
|
||
return frame_policy_2d->read_state<Frame_Policy::Base_Tag>();
|
||
return frame_policy_3d->read_state<Frame_Policy_3D::Base_Tag>().common;
|
||
}
|
||
|
||
[[nodiscard]] bool consume_frame_policy_events() {
|
||
const bool configuration_changed = with_frame_policy(
|
||
[](auto& policy) { return policy.consume_events(); });
|
||
if (frame_policy_3d)
|
||
static_cast<void>(frame_policy_3d->consume_3d_events());
|
||
return configuration_changed;
|
||
}
|
||
|
||
[[nodiscard]] nlohmann::json schema() const;
|
||
[[nodiscard]] Stream_Snapshot stream_snapshot() const;
|
||
void publish(std::shared_ptr<const Plot_Stream_Frame> frame) noexcept;
|
||
void submit_tick_request(Plot_Render_Tick tick);
|
||
void keep_latest_tick(const Plot_Render_Tick& tick);
|
||
void arm_tick_consumer(std::weak_ptr<Plot> lifetime);
|
||
void release_render_admission(std::weak_ptr<Plot> lifetime);
|
||
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(Render_Frame* completed = nullptr);
|
||
void consume_completed_frame(Render_Frame* frame);
|
||
void retire_completed_frame(Render_Frame* frame);
|
||
void consume_retired_frames();
|
||
void finalize_retired_frame(Render_Frame* frame);
|
||
void arm_retired_frame_consumer();
|
||
void attach_completion(std::unique_ptr<Task_Graph> completion);
|
||
[[nodiscard]] bool mark_taskflow_trace(Render_Frame& frame);
|
||
[[nodiscard]] bool mark_post_publish_taskflow_trace();
|
||
void store_trace(std::atomic_uint64_t& control,
|
||
std::array<std::atomic<std::shared_ptr<const nlohmann::json>>,
|
||
maximum_taskflow_trace_frames>& slots,
|
||
const Taskflow_Frame_Trace& trace,
|
||
const nlohmann::json& captured_components = {},
|
||
const nlohmann::json& captured_backend = {});
|
||
[[nodiscard]] nlohmann::json trace_response(
|
||
const std::atomic_uint64_t& control,
|
||
const std::atomic_size_t& remaining,
|
||
const std::array<std::atomic<std::shared_ptr<const nlohmann::json>>,
|
||
maximum_taskflow_trace_frames>& slots) const;
|
||
void fail(std::exception_ptr failure) noexcept;
|
||
};
|
||
|
||
void Plot::Private::fail(std::exception_ptr failure) noexcept {
|
||
try {
|
||
auto description = std::make_shared<const std::string>(
|
||
exception_description(failure));
|
||
std::shared_ptr<const std::string> empty;
|
||
if (!terminal_failure.compare_exchange_strong(
|
||
empty, description, std::memory_order_acq_rel,
|
||
std::memory_order_acquire)) return;
|
||
const auto output = std::make_shared<const Plot_Stream_Frame>(
|
||
Plot_Stream_Frame{nlohmann::json{
|
||
{"kind", "plot_error"},
|
||
{"protocol", "aethera.video.frame"},
|
||
{"version", plot_stream_protocol_version},
|
||
{"message", *description}
|
||
}.dump(), {}});
|
||
publish(std::move(output));
|
||
}
|
||
catch (...) {}
|
||
}
|
||
|
||
nlohmann::json Plot::Private::schema() const {
|
||
auto result = view->schema();
|
||
auto analysis = with_frame_policy(
|
||
[](const auto& policy) { return frame_policy_schema(policy); });
|
||
if (frame_policy_3d) {
|
||
analysis["fields"].push_back({
|
||
{"key", "pipeline_capacity"}, {"label", "3D 帧槽容量"},
|
||
{"editor", "integer"}, {"editable", false}, {"value", 3},
|
||
{"description", "3D 独立策略最多允许三帧处于 Prepare、GPU 和退休阶段。"},
|
||
{"technical_description", "3D-only submitted-driven pipeline capacity; 2D policy is unchanged."}
|
||
});
|
||
}
|
||
const auto generator = view->data_generator_schema();
|
||
if (!generator.is_null()) analysis["data_generator"] = generator;
|
||
result["frame_analysis"] = std::move(analysis);
|
||
return result;
|
||
}
|
||
|
||
Plot::Private::Stream_Snapshot Plot::Private::stream_snapshot() const {
|
||
Stream_Snapshot result;
|
||
result.consumers = consumers.load(std::memory_order_acquire);
|
||
for (const auto& [id, consumer] : *result.consumers) {
|
||
static_cast<void>(id);
|
||
if (consumer.width == 0 || consumer.height == 0) continue;
|
||
result.width = std::max(result.width, consumer.width);
|
||
result.height = std::max(result.height, consumer.height);
|
||
}
|
||
result.width = std::clamp(result.width == 0 ? 320U : result.width, 160U, 1920U) & ~1U;
|
||
result.height = std::clamp(result.height == 0 ? 192U : result.height, 120U, 1080U) & ~1U;
|
||
return result;
|
||
}
|
||
|
||
void Plot::Private::publish(
|
||
std::shared_ptr<const Plot_Stream_Frame> frame) noexcept {
|
||
if (!frame) return;
|
||
try {
|
||
const auto snapshot = stream_snapshot();
|
||
std::vector<Stream_Id> failed_consumers;
|
||
for (const auto& [id, consumer] : *snapshot.consumers) {
|
||
if (!consumer.handler) continue;
|
||
try {
|
||
consumer.handler(frame);
|
||
}
|
||
catch (...) {
|
||
failed_consumers.push_back(id);
|
||
}
|
||
}
|
||
if (failed_consumers.empty()) return;
|
||
auto current = consumers.load(std::memory_order_acquire);
|
||
for (;;) {
|
||
auto next = std::make_shared<Consumer_Map>(*current);
|
||
for (const auto id : failed_consumers) next->erase(id);
|
||
std::shared_ptr<const Consumer_Map> desired = next;
|
||
if (consumers.compare_exchange_weak(
|
||
current, desired, std::memory_order_release,
|
||
std::memory_order_acquire))
|
||
break;
|
||
}
|
||
}
|
||
catch (...) {}
|
||
}
|
||
|
||
void Plot::Private::refresh_schedule() {
|
||
if (!frame_timer.valid()) return;
|
||
const auto current_consumers = consumers.load(std::memory_order_acquire);
|
||
const auto& pacing = pacing_state();
|
||
if (!pacing.render_enabled || current_consumers->empty()) {
|
||
frame_timer.cancel();
|
||
return;
|
||
}
|
||
if (pacing.mode == Frame_Pacing_Mode::fixed_rate) {
|
||
frame_timer.start_periodic(pacing.fixed_rate_fps);
|
||
return;
|
||
}
|
||
frame_timer.cancel();
|
||
if (pacing.mode != Frame_Pacing_Mode::maximum_rate) return;
|
||
const auto now = std::chrono::steady_clock::now();
|
||
submit_tick_request(Plot_Render_Tick{
|
||
.issued_at = now,
|
||
.time_milliseconds = std::chrono::duration<double, std::milli>(
|
||
now - clock_origin).count(),
|
||
.source = Frame_Request_Source::maximum_rate});
|
||
arm_tick_consumer(lifetime);
|
||
}
|
||
|
||
void Plot::Private::submit_tick_request(Plot_Render_Tick tick) {
|
||
const auto source = tick.source;
|
||
const auto issued_at = tick.issued_at;
|
||
if (!tick_requests.enqueue(std::move(tick)))
|
||
throw std::bad_alloc{};
|
||
with_frame_policy([&](auto& policy) {
|
||
policy.record_request(source, issued_at);
|
||
});
|
||
tick_request_generation.fetch_add(1, std::memory_order_release);
|
||
}
|
||
|
||
void Plot::Private::keep_latest_tick(const Plot_Render_Tick& tick) {
|
||
const auto priority = [](Frame_Request_Source source) {
|
||
switch (source) {
|
||
case Frame_Request_Source::unspecified: return 0;
|
||
case Frame_Request_Source::periodic: return 0;
|
||
case Frame_Request_Source::maximum_rate: return 1;
|
||
case Frame_Request_Source::immediate: return 2;
|
||
}
|
||
return 0;
|
||
};
|
||
if (deferred_tick) {
|
||
const auto current_priority = priority(deferred_tick->source);
|
||
const auto next_priority = priority(tick.source);
|
||
with_frame_policy([](auto& policy) {
|
||
policy.record_request_coalesced();
|
||
});
|
||
if (current_priority > next_priority ||
|
||
(current_priority == next_priority &&
|
||
deferred_tick->issued_at >= tick.issued_at))
|
||
return;
|
||
}
|
||
deferred_tick = tick;
|
||
}
|
||
|
||
void Plot::Private::arm_tick_consumer(std::weak_ptr<Plot> lifetime) {
|
||
if (terminal_failure.load(std::memory_order_acquire)) return;
|
||
if (tick_task_scheduled.exchange(true, std::memory_order_acq_rel)) return;
|
||
aethera::schedule_task("web.plot.tick.consume", [lifetime] {
|
||
const auto plot = lifetime.lock();
|
||
if (!plot) return;
|
||
try { plot->d->consume_tick(lifetime); }
|
||
catch (...) { plot->d->fail(std::current_exception()); }
|
||
});
|
||
}
|
||
|
||
void Plot::Private::release_render_admission(std::weak_ptr<Plot> lifetime) {
|
||
auto expected = Render_Admission_State::rendering;
|
||
if (!render_admission.compare_exchange_strong(
|
||
expected, Render_Admission_State::ready,
|
||
std::memory_order_acq_rel, std::memory_order_acquire))
|
||
return;
|
||
const auto& pacing = pacing_state();
|
||
if (pacing.render_enabled && pacing.mode == Frame_Pacing_Mode::maximum_rate) {
|
||
const auto current_consumers = consumers.load(std::memory_order_acquire);
|
||
if (!current_consumers->empty()) {
|
||
const auto now = std::chrono::steady_clock::now();
|
||
submit_tick_request(Plot_Render_Tick{
|
||
.issued_at = now,
|
||
.time_milliseconds = std::chrono::duration<double, std::milli>(
|
||
now - clock_origin).count(),
|
||
.source = Frame_Request_Source::maximum_rate});
|
||
}
|
||
}
|
||
arm_tick_consumer(std::move(lifetime));
|
||
}
|
||
|
||
void Plot::Private::consume_tick(std::weak_ptr<Plot> lifetime) {
|
||
if (consume_frame_policy_events()) refresh_schedule();
|
||
const auto observed_generation =
|
||
tick_request_generation.load(std::memory_order_acquire);
|
||
Plot_Render_Tick requested;
|
||
while (tick_requests.try_dequeue(requested))
|
||
keep_latest_tick(requested);
|
||
if (render_admission.load(std::memory_order_acquire) ==
|
||
Render_Admission_State::ready) {
|
||
auto tick = std::exchange(deferred_tick, {});
|
||
if (tick) clock_tick(*tick);
|
||
}
|
||
if (consume_frame_policy_events()) refresh_schedule();
|
||
tick_task_scheduled.store(false, std::memory_order_release);
|
||
if (tick_request_generation.load(std::memory_order_acquire) !=
|
||
observed_generation ||
|
||
(render_admission.load(std::memory_order_acquire) ==
|
||
Render_Admission_State::ready && deferred_tick))
|
||
arm_tick_consumer(std::move(lifetime));
|
||
}
|
||
|
||
void Plot::Private::clock_tick(const Plot_Render_Tick& tick) {
|
||
if (terminal_failure.load(std::memory_order_acquire)) return;
|
||
const auto& pacing = pacing_state();
|
||
const bool accepted = pacing.render_enabled &&
|
||
(tick.source == Frame_Request_Source::immediate ||
|
||
(tick.source == Frame_Request_Source::periodic &&
|
||
pacing.mode == Frame_Pacing_Mode::fixed_rate) ||
|
||
(tick.source == Frame_Request_Source::maximum_rate &&
|
||
pacing.mode == Frame_Pacing_Mode::maximum_rate));
|
||
if (!accepted) {
|
||
with_frame_policy([](auto& policy) {
|
||
policy.record_policy_rejection();
|
||
});
|
||
return;
|
||
}
|
||
with_frame_policy([&](auto& policy) {
|
||
policy.record_request_accepted(tick.source);
|
||
});
|
||
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;
|
||
}
|
||
|
||
bool Plot::Private::mark_post_publish_taskflow_trace() {
|
||
auto remaining = post_publish_trace_remaining.load(std::memory_order_acquire);
|
||
while (remaining != 0) {
|
||
if (post_publish_trace_remaining.compare_exchange_weak(
|
||
remaining, remaining - 1, std::memory_order_acq_rel,
|
||
std::memory_order_acquire))
|
||
return true;
|
||
}
|
||
return false;
|
||
}
|
||
|
||
void Plot::Private::store_trace(
|
||
std::atomic_uint64_t& control,
|
||
std::array<std::atomic<std::shared_ptr<const nlohmann::json>>,
|
||
maximum_taskflow_trace_frames>& slots,
|
||
const Taskflow_Frame_Trace& value,
|
||
const nlohmann::json& captured_components,
|
||
const nlohmann::json& captured_backend) {
|
||
auto trace = std::make_shared<const nlohmann::json>(
|
||
taskflow_trace_json(value, captured_components, captured_backend));
|
||
auto state = control.load(std::memory_order_acquire);
|
||
for (;;) {
|
||
const auto requested = static_cast<std::uint32_t>(state >> 32U);
|
||
const auto captured = static_cast<std::uint32_t>(state);
|
||
if (captured >= requested) return;
|
||
slots[captured].store(trace, std::memory_order_release);
|
||
const auto next = (static_cast<std::uint64_t>(requested) << 32U) |
|
||
static_cast<std::uint64_t>(captured + 1U);
|
||
if (control.compare_exchange_weak(
|
||
state, next, std::memory_order_release,
|
||
std::memory_order_acquire))
|
||
return;
|
||
}
|
||
}
|
||
|
||
nlohmann::json Plot::Private::trace_response(
|
||
const std::atomic_uint64_t& control,
|
||
const std::atomic_size_t& remaining,
|
||
const std::array<std::atomic<std::shared_ptr<const nlohmann::json>>,
|
||
maximum_taskflow_trace_frames>& slots) const {
|
||
nlohmann::json frames = nlohmann::json::array();
|
||
const auto state = control.load(std::memory_order_acquire);
|
||
const auto requested = static_cast<std::uint32_t>(state >> 32U);
|
||
const auto captured = static_cast<std::uint32_t>(state);
|
||
for (std::uint32_t index = 0; index < captured; ++index)
|
||
if (const auto trace = slots[index].load(std::memory_order_acquire))
|
||
frames.push_back(*trace);
|
||
const auto left = remaining.load(std::memory_order_acquire);
|
||
return {
|
||
{"protocol", "aethera.taskflow.frames"}, {"version", 1},
|
||
{"requested", requested}, {"remaining", left},
|
||
{"captured", frames.size()},
|
||
{"complete", requested != 0 && frames.size() == requested},
|
||
{"frames", std::move(frames)}};
|
||
}
|
||
|
||
void Plot::Private::render_frame(Plot_Render_Tick tick) {
|
||
if (terminal_failure.load(std::memory_order_acquire)) return;
|
||
const auto streams = stream_snapshot();
|
||
const auto& pacing = pacing_state();
|
||
if (!pacing.render_enabled || streams.consumers->empty()) return;
|
||
|
||
auto admission_expected = Render_Admission_State::ready;
|
||
if (!render_admission.compare_exchange_strong(
|
||
admission_expected, Render_Admission_State::rendering,
|
||
std::memory_order_acq_rel,
|
||
std::memory_order_acquire)) {
|
||
keep_latest_tick(tick);
|
||
return;
|
||
}
|
||
|
||
std::size_t slot_index{};
|
||
Managed_Frame* managed{};
|
||
/*
|
||
* 只有 rendering 槽受 Scene 不可重入门约束；consuming 槽表示上一帧
|
||
* 已经完成 Plot 像素发布，外接 H264/WebRTC 仍可继续持有该物理帧的
|
||
* 诊断生命周期。只要还有 available 槽，下一帧即可进入。
|
||
*/
|
||
for (std::size_t index = 0; index < frame_slots.size(); ++index) {
|
||
auto* candidate = &frame_slots[index];
|
||
auto* published = candidate;
|
||
const bool was_latest = latest_statistics_frame.compare_exchange_strong(
|
||
published, nullptr, std::memory_order_acq_rel,
|
||
std::memory_order_acquire);
|
||
if (candidate->diagnostic_readers.load(std::memory_order_acquire) != 0) {
|
||
if (was_latest) {
|
||
Managed_Frame* empty{};
|
||
static_cast<void>(latest_statistics_frame.compare_exchange_strong(
|
||
empty, candidate, std::memory_order_release,
|
||
std::memory_order_relaxed));
|
||
}
|
||
continue;
|
||
}
|
||
auto expected = Frame_State::available;
|
||
if (!candidate->state.compare_exchange_strong(
|
||
expected, Frame_State::rendering,
|
||
std::memory_order_acq_rel, std::memory_order_acquire)) {
|
||
if (was_latest) {
|
||
Managed_Frame* empty{};
|
||
static_cast<void>(latest_statistics_frame.compare_exchange_strong(
|
||
empty, candidate, std::memory_order_release,
|
||
std::memory_order_relaxed));
|
||
}
|
||
continue;
|
||
}
|
||
slot_index = index;
|
||
managed = &frame_slots[index];
|
||
break;
|
||
}
|
||
if (!managed) {
|
||
with_frame_policy([](auto& policy) {
|
||
policy.record_frame_slot_backpressure();
|
||
});
|
||
keep_latest_tick(tick);
|
||
/*
|
||
* 三个槽都仍被外接消费者持有时，只保留 latest pending。这里绝不能
|
||
* 立即 arm tick consumer，否则会在没有任何槽可用期间形成
|
||
* consume -> no slot -> consume 的 Taskflow 任务风暴。真正的唤醒点
|
||
* 是 retire_completed_frame：某个 consuming 槽变回 available 后只唤醒一次。
|
||
*/
|
||
render_admission.store(
|
||
Render_Admission_State::frame_slots_exhausted,
|
||
std::memory_order_release);
|
||
return;
|
||
}
|
||
managed->presentation_time =
|
||
std::chrono::duration_cast<std::chrono::microseconds>(
|
||
std::chrono::duration<double, std::milli>(tick.time_milliseconds));
|
||
managed->tick_issued_at = tick.issued_at;
|
||
managed->submitted_at = {};
|
||
const auto rollback_unsubmitted = [this, slot_index] {
|
||
auto& slot = frame_slots[slot_index];
|
||
auto expected = Frame_State::rendering;
|
||
static_cast<void>(slot.state.compare_exchange_strong(
|
||
expected, Frame_State::available, std::memory_order_acq_rel,
|
||
std::memory_order_acquire));
|
||
};
|
||
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 {
|
||
tick.width = streams.width;
|
||
tick.height = streams.height;
|
||
const std::uint64_t sequence = next_frame_sequence++;
|
||
const Frame_Identity identity{
|
||
sequence, tick.sequence == 0 ? sequence : tick.sequence};
|
||
const auto request_source = tick.source;
|
||
const auto& policy_state = pacing_state();
|
||
Render_Frame* logical_frame{};
|
||
if (auto* frame_2d = std::get_if<std::unique_ptr<Frame_2D>>(&managed->frame)) {
|
||
(*frame_2d)->begin(identity, Frame_2D::native_pixel_format,
|
||
request_source, policy_state);
|
||
logical_frame = frame_2d->get();
|
||
} else {
|
||
auto& frame_3d = std::get<std::unique_ptr<Frame_3D>>(managed->frame);
|
||
frame_3d->begin(identity,
|
||
pacing.video_enabled ? Frame_3D_Output::pixels
|
||
: Frame_3D_Output::diagnostics,
|
||
Frame_3D::native_pixel_format, request_source,
|
||
policy_state);
|
||
logical_frame = frame_3d.get();
|
||
}
|
||
taskflow_trace_claimed = mark_taskflow_trace(*logical_frame);
|
||
/*
|
||
* 各图的采样、网格构造和属性读取都在 Plot 自己的准备域完成。
|
||
* 进入 Scene::render 后只剩已经准备好的 Visual 批次与轻量提交；
|
||
* 共享 Render Domain 不承担业务数据生成。
|
||
*/
|
||
logical_frame->mark(Frame_Trace_Marker::plot_update_started);
|
||
const auto update_started = std::chrono::steady_clock::now();
|
||
view->update(tick);
|
||
const auto update_elapsed = std::chrono::steady_clock::now() - update_started;
|
||
logical_frame->mark(Frame_Trace_Marker::plot_update_finished);
|
||
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));
|
||
};
|
||
|
||
if (auto* scene_2d = std::get_if<std::unique_ptr<Scene_2D>>(&scene)) {
|
||
auto& output = *std::get<std::unique_ptr<Frame_2D>>(managed->frame);
|
||
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) {
|
||
with_frame_policy([](auto& policy) {
|
||
policy.record_scene_rejection();
|
||
});
|
||
rollback_unsubmitted();
|
||
restore_taskflow_trace_claim();
|
||
release_render_admission(lifetime);
|
||
} else {
|
||
taskflow_trace_claimed = false;
|
||
managed->submitted_at = std::chrono::steady_clock::now();
|
||
with_frame_policy([&](auto& policy) {
|
||
policy.record_frame_submitted(
|
||
sequence, request_source,
|
||
managed->submitted_at - managed->tick_issued_at);
|
||
});
|
||
}
|
||
return;
|
||
}
|
||
auto& output = *std::get<std::unique_ptr<Frame_3D>>(managed->frame);
|
||
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) {
|
||
taskflow_trace_claimed = false;
|
||
managed->submitted_at = std::chrono::steady_clock::now();
|
||
with_frame_policy([&](auto& policy) {
|
||
policy.record_frame_submitted(
|
||
sequence, request_source,
|
||
managed->submitted_at - managed->tick_issued_at);
|
||
});
|
||
return;
|
||
}
|
||
with_frame_policy([](auto& policy) {
|
||
policy.record_scene_rejection();
|
||
});
|
||
rollback_unsubmitted();
|
||
restore_taskflow_trace_claim();
|
||
release_render_admission(lifetime);
|
||
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();
|
||
release_render_admission(lifetime);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
|
||
void Plot::Private::publish_completed_frame(Render_Frame* completed) {
|
||
Render_Frame* frame{};
|
||
Managed_Frame* managed{};
|
||
for (std::size_t index = 0; index < frame_slots.size(); ++index) {
|
||
if (frame_slots[index].state.load(std::memory_order_acquire) !=
|
||
Frame_State::rendering)
|
||
continue;
|
||
if (managed)
|
||
throw std::logic_error("Plot has multiple frames in Scene rendering");
|
||
frame = std::visit(
|
||
[](const auto& value) -> Render_Frame* { return value.get(); },
|
||
frame_slots[index].frame);
|
||
if (completed && frame != completed) continue;
|
||
managed = &frame_slots[index];
|
||
if (completed) break;
|
||
}
|
||
if (!managed)
|
||
throw std::logic_error("Scene completion graph has no rendering Plot frame");
|
||
|
||
const auto& pacing = pacing_state();
|
||
const auto identity = frame->identity();
|
||
Frame_Identity rendered_identity = identity;
|
||
std::shared_ptr<const std::vector<std::byte>> pixel_storage;
|
||
Plot_Pixel_Layout pixel_layout{Plot_Pixel_Layout::rgba8};
|
||
std::uint32_t width{};
|
||
std::uint32_t height{};
|
||
if (auto *frame_2d =
|
||
std::get_if<std::unique_ptr<Frame_2D>>(&managed->frame)) {
|
||
pixel_layout = Plot_Pixel_Layout::bgra8;
|
||
const auto image = (*frame_2d)->image();
|
||
width = static_cast<std::uint32_t>(image.width);
|
||
height = static_cast<std::uint32_t>(image.height);
|
||
if (pacing.video_enabled) {
|
||
auto output = (*frame_2d)->output_pixels();
|
||
pixel_storage = std::make_shared<const std::vector<std::byte>>(
|
||
std::move(output.bytes));
|
||
width = static_cast<std::uint32_t>(output.width);
|
||
height = static_cast<std::uint32_t>(output.height);
|
||
}
|
||
} else {
|
||
auto &frame_3d = std::get<std::unique_ptr<Frame_3D>>(managed->frame);
|
||
rendered_identity = frame_3d->rendered_identity();
|
||
const auto extent = frame_3d->extent();
|
||
width = extent.width;
|
||
height = extent.height;
|
||
if (pacing.video_enabled && frame_3d->output() == Frame_3D_Output::pixels)
|
||
pixel_storage = frame_3d->share_pixels();
|
||
}
|
||
|
||
auto pixels = std::make_shared<const Plot_Pixel_Frame>(Plot_Pixel_Frame{
|
||
std::move(pixel_storage), pixel_layout, managed->presentation_time,
|
||
identity.sequence, identity.correlation_id, rendered_identity.sequence,
|
||
rendered_identity.correlation_id, width, height});
|
||
|
||
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())));
|
||
auto expected = Frame_State::rendering;
|
||
if (!managed->state.compare_exchange_strong(
|
||
expected, Frame_State::consuming, std::memory_order_acq_rel,
|
||
std::memory_order_acquire))
|
||
throw std::logic_error("Plot frame left rendering before pixel publish");
|
||
}
|
||
|
||
void Plot::Private::consume_completed_frame(Render_Frame* frame) {
|
||
if (!frame)
|
||
throw std::invalid_argument("Plot received a null completed frame");
|
||
Managed_Frame* managed{};
|
||
for (auto& slot : frame_slots) {
|
||
auto* address = std::visit(
|
||
[](const auto& value) -> Render_Frame* { return value.get(); },
|
||
slot.frame);
|
||
if (address != frame) continue;
|
||
if (slot.state.load(std::memory_order_acquire) != Frame_State::consuming)
|
||
throw std::logic_error("completed Plot frame was not published");
|
||
managed = &slot;
|
||
break;
|
||
}
|
||
if (!managed)
|
||
throw std::logic_error("frame callback has no owned Plot frame");
|
||
|
||
/*
|
||
* Scene 已在调用本 callback 前释放自己的 render admission；这里同步
|
||
* 释放 Plot 的 view/update 门，并立刻唤醒 busy 期间保留的 latest tick。
|
||
* 之后外接 DAG 仍在当前 Frame trace 内执行，但不会阻塞下一帧渲染。
|
||
*/
|
||
if (std::holds_alternative<std::unique_ptr<Frame_2D>>(managed->frame))
|
||
release_render_admission(lifetime);
|
||
if (post_publish_graph.empty()) return;
|
||
|
||
bool expected = false;
|
||
if (!post_publish_busy.compare_exchange_strong(
|
||
expected, true, std::memory_order_acq_rel,
|
||
std::memory_order_acquire))
|
||
return;
|
||
|
||
/*
|
||
* 外接图异步提交；上一轮尚未完成时直接合并到 sampler 内的 latest，绝不
|
||
* 在 Taskflow Worker 内等待。诊断使用独立 Render_Frame 保存外接图观察
|
||
* 窗口，因此 Scene 的物理帧可立即退役并被下一次渲染复用。
|
||
*/
|
||
const bool local_post_publish_trace = mark_post_publish_taskflow_trace();
|
||
auto trace_frame = local_post_publish_trace
|
||
? std::make_shared<Render_Frame>(frame->identity())
|
||
: std::shared_ptr<Render_Frame>{};
|
||
bool local_trace_started{};
|
||
if (trace_frame) {
|
||
trace_frame->request_taskflow_trace();
|
||
local_trace_started =
|
||
aethera::detail::begin_taskflow_trace(*trace_frame);
|
||
}
|
||
const auto weak = lifetime;
|
||
auto completion = [weak, trace_frame, local_trace_started] {
|
||
const auto owner = weak.lock();
|
||
if (!owner) return;
|
||
if (local_trace_started) {
|
||
aethera::detail::finish_taskflow_trace(*trace_frame);
|
||
owner->d->store_trace(
|
||
owner->d->post_publish_trace_control,
|
||
owner->d->post_publish_trace_slots,
|
||
trace_frame->take_taskflow_trace());
|
||
}
|
||
owner->d->post_publish_busy.store(false, std::memory_order_release);
|
||
};
|
||
try {
|
||
if (trace_frame)
|
||
aethera::detail::run_taskflow(
|
||
post_publish_graph, *trace_frame, "plot.post_publish",
|
||
std::move(completion));
|
||
else
|
||
aethera::detail::run_taskflow(
|
||
post_publish_graph, std::move(completion));
|
||
}
|
||
catch (...) {
|
||
if (local_trace_started)
|
||
aethera::detail::finish_taskflow_trace(*trace_frame);
|
||
if (local_post_publish_trace)
|
||
post_publish_trace_remaining.fetch_add(1, std::memory_order_release);
|
||
post_publish_busy.store(false, std::memory_order_release);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
void Plot::Private::arm_retired_frame_consumer() {
|
||
bool expected{};
|
||
if (!retired_frame_task_scheduled.compare_exchange_strong(
|
||
expected, true, std::memory_order_acq_rel,
|
||
std::memory_order_acquire)) return;
|
||
const auto weak = lifetime;
|
||
schedule_task("web.plot.frame.retire", [weak] {
|
||
if (const auto owner = weak.lock())
|
||
owner->d->consume_retired_frames();
|
||
});
|
||
}
|
||
|
||
void Plot::Private::consume_retired_frames() {
|
||
for (;;) {
|
||
auto* list = retired_frames.exchange(nullptr, std::memory_order_acq_rel);
|
||
if (!list) break;
|
||
std::vector<Managed_Frame*> frames;
|
||
while (list) {
|
||
auto* next = list->retired_next.exchange(
|
||
nullptr, std::memory_order_relaxed);
|
||
frames.push_back(list);
|
||
list = next;
|
||
}
|
||
std::ranges::sort(frames, {}, [](const Managed_Frame* managed) {
|
||
return std::visit(
|
||
[](const auto& value) { return value->identity().sequence; },
|
||
managed->frame);
|
||
});
|
||
for (auto* managed : frames) {
|
||
auto* frame = std::visit(
|
||
[](const auto& value) -> Render_Frame* { return value.get(); },
|
||
managed->frame);
|
||
finalize_retired_frame(frame);
|
||
}
|
||
}
|
||
retired_frame_task_scheduled.store(false, std::memory_order_release);
|
||
if (retired_frames.load(std::memory_order_acquire))
|
||
arm_retired_frame_consumer();
|
||
}
|
||
|
||
void Plot::Private::retire_completed_frame(Render_Frame* frame) {
|
||
if (!frame)
|
||
throw std::invalid_argument("Plot received a null retired frame");
|
||
Managed_Frame* managed{};
|
||
for (auto& slot : frame_slots) {
|
||
auto* address = std::visit(
|
||
[](const auto& value) -> Render_Frame* { return value.get(); },
|
||
slot.frame);
|
||
if (address != frame) continue;
|
||
managed = &slot;
|
||
break;
|
||
}
|
||
if (!managed)
|
||
throw std::logic_error("retired frame has no owned Plot slot");
|
||
|
||
auto* head = retired_frames.load(std::memory_order_relaxed);
|
||
do {
|
||
managed->retired_next.store(head, std::memory_order_relaxed);
|
||
} while (!retired_frames.compare_exchange_weak(
|
||
head, managed, std::memory_order_release,
|
||
std::memory_order_relaxed));
|
||
arm_retired_frame_consumer();
|
||
}
|
||
|
||
void Plot::Private::finalize_retired_frame(Render_Frame* frame) {
|
||
Managed_Frame* managed{};
|
||
for (auto& slot : frame_slots) {
|
||
auto* address = std::visit(
|
||
[](const auto& value) -> Render_Frame* { return value.get(); },
|
||
slot.frame);
|
||
if (address == frame) {
|
||
managed = &slot;
|
||
break;
|
||
}
|
||
}
|
||
if (!managed)
|
||
throw std::logic_error("retired frame has no owned Plot slot");
|
||
|
||
with_frame_policy([&](auto& policy) {
|
||
policy.record_frame_completed(
|
||
frame->identity().sequence,
|
||
managed->submitted_at.time_since_epoch().count() == 0
|
||
? std::chrono::steady_clock::duration::zero()
|
||
: std::chrono::steady_clock::now() - managed->submitted_at);
|
||
});
|
||
|
||
const auto statistics_generation_value =
|
||
statistics_generation.load(std::memory_order_acquire);
|
||
if (applied_statistics_generation != statistics_generation_value) {
|
||
completed_frame_statistics.reset();
|
||
applied_statistics_generation = statistics_generation_value;
|
||
}
|
||
managed->statistics = completed_frame_statistics.submit(*frame);
|
||
managed->statistics_generation = statistics_generation_value;
|
||
|
||
std::optional<Taskflow_Frame_Trace> captured_trace;
|
||
nlohmann::json captured_components;
|
||
nlohmann::json captured_backend;
|
||
if (frame->taskflow_trace_requested()) {
|
||
captured_trace.emplace(frame->take_taskflow_trace());
|
||
std::unordered_set<std::uint64_t> executed_nodes;
|
||
for (const auto& execution : captured_trace->tasks)
|
||
executed_nodes.insert(execution.native_id);
|
||
std::vector<std::string> executed_components;
|
||
for (const auto& graph : captured_trace->graphs) {
|
||
for (const auto& node : graph.nodes) {
|
||
if (!executed_nodes.contains(node.native_id)) continue;
|
||
const auto owner = std::ranges::find(
|
||
node.attributes, "owner_component",
|
||
&std::pair<std::string, std::string>::first);
|
||
if (owner == node.attributes.end() || owner->second.empty() ||
|
||
std::ranges::find(executed_components, owner->second) !=
|
||
executed_components.end()) continue;
|
||
executed_components.push_back(owner->second);
|
||
}
|
||
}
|
||
captured_components = view->capture_components(executed_components);
|
||
if (auto* frame_3d = dynamic_cast<render_3d::Frame_3D*>(frame)) {
|
||
if (auto observation = frame_3d->take_datoviz_observation())
|
||
captured_backend = datoviz_observation_json(*observation);
|
||
}
|
||
}
|
||
|
||
auto expected = Frame_State::consuming;
|
||
if (!managed->state.compare_exchange_strong(
|
||
expected, Frame_State::available, std::memory_order_acq_rel,
|
||
std::memory_order_acquire))
|
||
throw std::logic_error("retired Plot frame is not consuming");
|
||
latest_statistics_frame.store(managed, std::memory_order_release);
|
||
|
||
/* 物理槽是唯一背压原因；归还任意槽后只解除一次耗尽状态。 */
|
||
auto admission = Render_Admission_State::frame_slots_exhausted;
|
||
static_cast<void>(render_admission.compare_exchange_strong(
|
||
admission, Render_Admission_State::ready,
|
||
std::memory_order_acq_rel, std::memory_order_acquire));
|
||
arm_tick_consumer(lifetime);
|
||
|
||
if (captured_trace) {
|
||
auto owner = lifetime;
|
||
schedule_task("plot.taskflow.serialize", [owner,
|
||
trace = std::move(*captured_trace),
|
||
components = std::move(captured_components),
|
||
backend = std::move(captured_backend)]() mutable {
|
||
if (const auto plot = owner.lock())
|
||
plot->d->store_trace(
|
||
plot->d->taskflow_trace_control,
|
||
plot->d->taskflow_trace_slots, trace, components,
|
||
backend);
|
||
});
|
||
}
|
||
}
|
||
|
||
void Plot::Private::attach_completion(
|
||
std::unique_ptr<Task_Graph> completion) {
|
||
if (!completion || completion->empty())
|
||
throw std::invalid_argument("Plot completion pipeline is empty");
|
||
completion_extensions.push_back(std::move(completion));
|
||
auto extension = post_publish_graph.compose(
|
||
completion_extensions.back()->name(), *completion_extensions.back());
|
||
extension.describe("owner", "plot")
|
||
.describe("stage", "post-publish frame pipeline extension");
|
||
if (has_post_publish_tail) post_publish_tail.precede(extension);
|
||
post_publish_tail = std::move(extension);
|
||
has_post_publish_tail = true;
|
||
}
|
||
|
||
Plot::Plot(std::unique_ptr<Scene_2D> scene,
|
||
std::unique_ptr<Scene_View> view)
|
||
: d(std::make_unique<Private>(std::move(scene), std::move(view))) {}
|
||
Plot::Plot(std::unique_ptr<Scene_3D> scene,
|
||
std::unique_ptr<Scene_View> view)
|
||
: d(std::make_unique<Private>(std::move(scene), std::move(view))) {}
|
||
Plot::~Plot() = default;
|
||
|
||
void Plot::attach_scene_completion(
|
||
std::unique_ptr<Task_Graph> completion) {
|
||
d->attach_completion(std::move(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{
|
||
.issued_at = tick.issued_at,
|
||
.sequence = tick.sequence,
|
||
.time_milliseconds = tick.time_milliseconds,
|
||
.source = Frame_Request_Source::periodic});
|
||
}
|
||
});
|
||
/*
|
||
* 2D 的 callback 在 Scene render admission 已释放后运行：先执行所有
|
||
* post-publish 外接 DAG；Scene 完成 frame_ready 与 trace 收口后，再由
|
||
* retired callback 归还物理槽。这样 H264(N) 可与 Render(N+1) 重叠。
|
||
*/
|
||
if (auto* scene = std::get_if<std::unique_ptr<Scene_2D>>(&d->scene)) {
|
||
(*scene)->set_frame_callback([weak](Frame_2D* frame) {
|
||
if (auto owner = weak.lock()) {
|
||
try { owner->d->consume_completed_frame(frame); }
|
||
catch (...) { owner->d->fail(std::current_exception()); }
|
||
}
|
||
});
|
||
(*scene)->set_frame_retired_callback([weak](Frame_2D* frame) {
|
||
if (auto owner = weak.lock()) {
|
||
try { owner->d->retire_completed_frame(frame); }
|
||
catch (...) { owner->d->fail(std::current_exception()); }
|
||
}
|
||
});
|
||
} else {
|
||
auto& scene_3d = std::get<std::unique_ptr<Scene_3D>>(d->scene);
|
||
scene_3d->set_submitted_frame_callback(
|
||
[weak](Frame_3D* frame, bool overlaps_gpu) {
|
||
if (auto owner = weak.lock()) {
|
||
try {
|
||
owner->d->frame_policy_3d->record_gpu_submitted(
|
||
frame->identity().sequence, overlaps_gpu);
|
||
owner->d->release_render_admission(weak);
|
||
}
|
||
catch (...) { owner->d->fail(std::current_exception()); }
|
||
}
|
||
});
|
||
scene_3d->set_frame_callback(
|
||
[weak](Frame_3D* frame) {
|
||
if (auto owner = weak.lock()) {
|
||
try {
|
||
owner->d->publish_completed_frame(frame);
|
||
owner->d->frame_policy_3d->record_gpu_completed(
|
||
frame->identity().sequence);
|
||
owner->d->consume_completed_frame(frame);
|
||
owner->d->retire_completed_frame(frame);
|
||
}
|
||
catch (...) { owner->d->fail(std::current_exception()); }
|
||
}
|
||
});
|
||
}
|
||
/* callback 必须先于周期时钟安装，避免首帧在初始化窗口进入 Scene。 */
|
||
d->refresh_schedule();
|
||
});
|
||
}
|
||
|
||
Plot::Stream_Id Plot::subscribe(Stream_Handler handler) {
|
||
if (!handler)
|
||
throw std::invalid_argument("Plot subscription requires a handler");
|
||
ensure_started();
|
||
const auto id = d->next_stream_id.fetch_add(1, std::memory_order_relaxed);
|
||
const auto notification = handler;
|
||
auto current = d->consumers.load(std::memory_order_acquire);
|
||
for (;;) {
|
||
auto next = std::make_shared<Private::Consumer_Map>(*current);
|
||
next->emplace(id, Private::Consumer{handler});
|
||
std::shared_ptr<const Private::Consumer_Map> desired = next;
|
||
if (d->consumers.compare_exchange_weak(
|
||
current, desired, std::memory_order_release,
|
||
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 {
|
||
notification(std::make_shared<const Plot_Stream_Frame>(
|
||
Plot_Stream_Frame{nlohmann::json{
|
||
{"kind", "plot_error"},
|
||
{"protocol", "aethera.video.frame"},
|
||
{"version", plot_stream_protocol_version},
|
||
{"message", failure ? *failure : "Plot unavailable"}
|
||
}.dump(), {}}));
|
||
}
|
||
catch (...) {
|
||
unsubscribe(id);
|
||
}
|
||
}
|
||
return id;
|
||
}
|
||
|
||
void Plot::unsubscribe(Stream_Id stream) {
|
||
auto current = d->consumers.load(std::memory_order_acquire);
|
||
while (current->contains(stream)) {
|
||
auto next = std::make_shared<Private::Consumer_Map>(*current);
|
||
next->erase(stream);
|
||
std::shared_ptr<const Private::Consumer_Map> desired = next;
|
||
if (d->consumers.compare_exchange_weak(
|
||
current, desired, std::memory_order_release,
|
||
std::memory_order_acquire))
|
||
break;
|
||
}
|
||
d->refresh_schedule();
|
||
}
|
||
|
||
void Plot::configure_stream(Stream_Id stream, std::uint32_t width,
|
||
std::uint32_t height) {
|
||
auto current = d->consumers.load(std::memory_order_acquire);
|
||
for (;;) {
|
||
const auto found = current->find(stream);
|
||
if (found == current->end()) return;
|
||
auto next = std::make_shared<Private::Consumer_Map>(*current);
|
||
auto& consumer = next->at(stream);
|
||
consumer.width = width;
|
||
consumer.height = height;
|
||
std::shared_ptr<const Private::Consumer_Map> desired = next;
|
||
if (d->consumers.compare_exchange_weak(
|
||
current, desired, std::memory_order_release,
|
||
std::memory_order_acquire))
|
||
return;
|
||
}
|
||
}
|
||
|
||
void Plot::schedule_render(Plot_Render_Tick tick) {
|
||
ensure_started();
|
||
if (d->terminal_failure.load(std::memory_order_acquire)) return;
|
||
d->submit_tick_request(std::move(tick));
|
||
d->arm_tick_consumer(weak_from_this());
|
||
}
|
||
|
||
void Plot::render_once() {
|
||
ensure_started();
|
||
const auto now = std::chrono::steady_clock::now();
|
||
const auto elapsed = now - d->clock_origin;
|
||
schedule_render(Plot_Render_Tick{
|
||
.issued_at = now,
|
||
.time_milliseconds =
|
||
std::chrono::duration<double, std::milli>(elapsed).count(),
|
||
.source = Frame_Request_Source::immediate});
|
||
}
|
||
|
||
void Plot::submit_input(Plot_Input_Event event) {
|
||
ensure_started();
|
||
if (d->terminal_failure.load(std::memory_order_acquire)) return;
|
||
/*
|
||
* WebSocket 线程只向 Scene 的当前事件缓冲追加一个由 Scene
|
||
* memory_resource 分配的基类指针。Prepare 边界交换完整批次，
|
||
* Scene 在 Renderable 完成消费时按 Event_Type 增量统计，并随自身
|
||
* State 双缓冲发布；Web 层只在低频 diagnostics 请求中读取结果。
|
||
*/
|
||
try {
|
||
if (auto* scene_2d = std::get_if<std::unique_ptr<Scene_2D>>(&d->scene))
|
||
dispatch_plot_input(**scene_2d, event);
|
||
else
|
||
dispatch_plot_input(*std::get<std::unique_ptr<Scene_3D>>(d->scene), event);
|
||
}
|
||
catch (...) {
|
||
d->fail(std::current_exception());
|
||
}
|
||
}
|
||
|
||
nlohmann::json Plot::schema() {
|
||
ensure_started();
|
||
return d->schema();
|
||
}
|
||
|
||
nlohmann::json Plot::write_prop(std::string_view component,
|
||
std::string_view key,
|
||
const nlohmann::json& value) {
|
||
ensure_started();
|
||
if (component != "frame-analysis")
|
||
return d->view->write_prop(component, key, value);
|
||
auto result = d->with_frame_policy([&](auto& policy) {
|
||
return write_frame_policy_prop(policy, key, value);
|
||
});
|
||
if (result.value("success", false))
|
||
d->arm_tick_consumer(weak_from_this());
|
||
return result;
|
||
}
|
||
|
||
nlohmann::json Plot::component_state(std::string_view component) const {
|
||
return d->view->component_state(component);
|
||
}
|
||
|
||
nlohmann::json Plot::generate_data(const nlohmann::json& input) {
|
||
ensure_started();
|
||
return d->view->generate_data(input);
|
||
}
|
||
|
||
nlohmann::json Plot::diagnostics() const {
|
||
nlohmann::json frame_statistics = nlohmann::json::object();
|
||
nlohmann::json input_statistics = nlohmann::json::object();
|
||
Frame_Identity identity{};
|
||
std::uint64_t created_time_unix_ns{};
|
||
std::uint64_t dropped_sequences{};
|
||
double frame_rate{};
|
||
bool is_3d{};
|
||
const auto read_scene_statistics = [&](const auto& state) {
|
||
append_event_statistics_json(input_statistics, state.event_statistics);
|
||
};
|
||
std::visit([&](const auto& scene) {
|
||
using Scene_Pointer = std::remove_cvref_t<decltype(scene)>;
|
||
if constexpr (std::same_as<Scene_Pointer, std::unique_ptr<Scene_2D>>) {
|
||
scene->template access_state<Render_Scene_2D::Base_Tag>(
|
||
read_scene_statistics);
|
||
} else {
|
||
is_3d = true;
|
||
scene->template access_state<Render_Scene_3D::Base_Tag>(
|
||
read_scene_statistics);
|
||
}
|
||
}, d->scene);
|
||
|
||
{
|
||
const auto generation =
|
||
d->statistics_generation.load(std::memory_order_acquire);
|
||
auto* completed_frame =
|
||
d->latest_statistics_frame.load(std::memory_order_acquire);
|
||
Frame_Statistics_State statistics{};
|
||
if (completed_frame) {
|
||
completed_frame->diagnostic_readers.fetch_add(
|
||
1, std::memory_order_acq_rel);
|
||
if (d->latest_statistics_frame.load(std::memory_order_acquire) ==
|
||
completed_frame &&
|
||
completed_frame->state.load(std::memory_order_acquire) ==
|
||
Private::Frame_State::available &&
|
||
completed_frame->statistics_generation == generation)
|
||
statistics = completed_frame->statistics;
|
||
completed_frame->diagnostic_readers.fetch_sub(
|
||
1, std::memory_order_release);
|
||
}
|
||
append_statistic_json(frame_statistics, statistics);
|
||
identity = statistics.identity;
|
||
created_time_unix_ns = statistics.created_time_unix_ns;
|
||
dropped_sequences = statistics.dropped_sequences;
|
||
const auto& interval = statistics.values[
|
||
static_cast<std::size_t>(Frame_Statistic::frame_interval_ms)];
|
||
frame_rate = interval.trimmed_average > 0.0
|
||
? 1'000.0 / interval.trimmed_average : 0.0;
|
||
}
|
||
|
||
const auto& pacing = d->pacing_state();
|
||
const auto stream = d->stream_snapshot();
|
||
const auto admission = d->render_admission.load(std::memory_order_acquire);
|
||
const auto admission_name = [&] {
|
||
switch (admission) {
|
||
case Private::Render_Admission_State::ready: return "ready";
|
||
case Private::Render_Admission_State::rendering: return "rendering";
|
||
case Private::Render_Admission_State::frame_slots_exhausted:
|
||
return "frame_slots_exhausted";
|
||
}
|
||
return "unknown";
|
||
}();
|
||
nlohmann::json supported_formats = nlohmann::json::array();
|
||
if (is_3d) {
|
||
for (const auto format : Frame_3D::supported_pixel_formats)
|
||
supported_formats.push_back(pixel_format_name(format));
|
||
} else {
|
||
for (const auto format : Frame_2D::supported_pixel_formats)
|
||
supported_formats.push_back(pixel_format_name(format));
|
||
}
|
||
const auto format = is_3d
|
||
? pixel_format_name(Frame_3D::native_pixel_format)
|
||
: pixel_format_name(Frame_2D::native_pixel_format);
|
||
const auto native_format = is_3d
|
||
? pixel_format_name(Frame_3D::native_pixel_format)
|
||
: pixel_format_name(Frame_2D::native_pixel_format);
|
||
const std::size_t byte_length = pacing.video_enabled
|
||
? static_cast<std::size_t>(stream.width) * stream.height * 4U : 0U;
|
||
nlohmann::json output{
|
||
{"protocol", "aethera.plot.diagnostics"}, {"version", 4},
|
||
{"dimension", is_3d ? "3D" : "2D"},
|
||
{"sequence", identity.sequence},
|
||
{"correlation_id", identity.correlation_id},
|
||
{"rendered_sequence", identity.sequence},
|
||
{"rendered_correlation_id", identity.correlation_id},
|
||
{"generated_time_unix_ms",
|
||
static_cast<double>(created_time_unix_ns) / 1'000'000.0},
|
||
{"delivery", pacing.video_enabled ? "gallery-video" : "diagnostics"},
|
||
{"frame_rate_fps", frame_rate},
|
||
{"dropped_sequence_count", dropped_sequences},
|
||
{"window_capacity", diagnostic_window_capacity},
|
||
{"pixel", {{"width", stream.width}, {"height", stream.height},
|
||
{"format", format}, {"native_format", native_format},
|
||
{"supported_formats", std::move(supported_formats)},
|
||
{"byte_length", byte_length}}},
|
||
{"frame_policy", frame_policy_state_json(pacing)},
|
||
{"render_admission", admission_name},
|
||
{"frame_statistics", std::move(frame_statistics)},
|
||
{"input_statistics", std::move(input_statistics)}};
|
||
if (is_3d) {
|
||
const auto& pipeline = d->frame_policy_3d->read_state<
|
||
Frame_Policy_3D::Base_Tag>();
|
||
output["frame_policy"]["three_dimensional_pipeline"] = {
|
||
{"capacity", pipeline.pipeline_capacity},
|
||
{"gpu_submission_count", pipeline.gpu_submission_count},
|
||
{"overlapped_release_count", pipeline.overlapped_release_count},
|
||
{"completion_gated_release_count",
|
||
pipeline.completion_gated_release_count},
|
||
{"gpu_completion_count", pipeline.gpu_completion_count},
|
||
{"gpu_in_flight", pipeline.gpu_in_flight},
|
||
{"peak_gpu_in_flight", pipeline.peak_gpu_in_flight},
|
||
{"last_submitted_sequence", pipeline.last_submitted_sequence},
|
||
{"last_completed_sequence", pipeline.last_completed_sequence}
|
||
};
|
||
const auto gpu = gpu_completion_state();
|
||
const auto milliseconds = [](std::uint64_t nanoseconds) {
|
||
return static_cast<double>(nanoseconds) / 1'000'000.0;
|
||
};
|
||
output["gpu_completion_domain"] = {
|
||
{"capacity", gpu.capacity}, {"in_flight", gpu.in_flight},
|
||
{"peak_in_flight", gpu.peak_in_flight}, {"watched", gpu.watched},
|
||
{"peak_watched", gpu.peak_watched},
|
||
{"active_fences", gpu.active_fences},
|
||
{"pending_fences", gpu.pending_fences},
|
||
{"reservation_count", gpu.reservation_count},
|
||
{"completion_count", gpu.completion_count},
|
||
{"cancellation_count", gpu.cancellation_count},
|
||
{"fence_probe_count", gpu.fence_probe_count},
|
||
{"fence_wait_count", gpu.fence_wait_count},
|
||
{"fence_wait_timeout_count", gpu.fence_wait_timeout_count},
|
||
{"fence_wait_total_ms", milliseconds(gpu.fence_wait_total_ns)},
|
||
{"fence_wait_max_ms", milliseconds(gpu.fence_wait_max_ns)},
|
||
{"callback_total_ms", milliseconds(gpu.callback_total_ns)},
|
||
{"callback_max_ms", milliseconds(gpu.callback_max_ns)},
|
||
{"callback_failure_count", gpu.callback_failure_count},
|
||
{"backpressure_count", gpu.backpressure_count},
|
||
{"fault_count", gpu.fault_count},
|
||
{"abandoned_count", gpu.abandoned_count}};
|
||
}
|
||
if (const auto failure = d->terminal_failure.load(std::memory_order_acquire))
|
||
output["terminal_failure"] = *failure;
|
||
return output;
|
||
}
|
||
|
||
void Plot::request_taskflow_trace(std::size_t frame_count) {
|
||
if (frame_count == 0 ||
|
||
frame_count > Private::maximum_taskflow_trace_frames)
|
||
throw std::invalid_argument("Taskflow trace frame_count must be between 1 and 120");
|
||
ensure_started();
|
||
auto control = d->taskflow_trace_control.load(std::memory_order_acquire);
|
||
for (;;) {
|
||
const auto requested = static_cast<std::uint32_t>(control >> 32U);
|
||
const auto captured = static_cast<std::uint32_t>(control);
|
||
if (requested != captured)
|
||
throw std::logic_error("A Taskflow frame trace request is already active");
|
||
const auto next = static_cast<std::uint64_t>(frame_count) << 32U;
|
||
if (d->taskflow_trace_control.compare_exchange_weak(
|
||
control, next, std::memory_order_release,
|
||
std::memory_order_acquire))
|
||
break;
|
||
}
|
||
for (auto& slot : d->taskflow_trace_slots)
|
||
slot.store({}, std::memory_order_release);
|
||
d->taskflow_trace_remaining.store(frame_count, std::memory_order_release);
|
||
}
|
||
|
||
nlohmann::json Plot::taskflow_trace() const {
|
||
return d->trace_response(d->taskflow_trace_control,
|
||
d->taskflow_trace_remaining,
|
||
d->taskflow_trace_slots);
|
||
}
|
||
|
||
void Plot::request_post_publish_taskflow_trace(std::size_t frame_count) {
|
||
if (frame_count == 0 ||
|
||
frame_count > Private::maximum_taskflow_trace_frames)
|
||
throw std::invalid_argument(
|
||
"Taskflow post-publish trace frame_count must be between 1 and 120");
|
||
ensure_started();
|
||
auto control = d->post_publish_trace_control.load(std::memory_order_acquire);
|
||
for (;;) {
|
||
const auto requested = static_cast<std::uint32_t>(control >> 32U);
|
||
const auto captured = static_cast<std::uint32_t>(control);
|
||
if (requested != captured)
|
||
throw std::logic_error(
|
||
"A post-publish Taskflow trace request is already active");
|
||
const auto next = static_cast<std::uint64_t>(frame_count) << 32U;
|
||
if (d->post_publish_trace_control.compare_exchange_weak(
|
||
control, next, std::memory_order_release,
|
||
std::memory_order_acquire))
|
||
break;
|
||
}
|
||
for (auto& slot : d->post_publish_trace_slots)
|
||
slot.store({}, std::memory_order_release);
|
||
d->post_publish_trace_remaining.store(frame_count, std::memory_order_release);
|
||
}
|
||
|
||
nlohmann::json Plot::post_publish_taskflow_trace() const {
|
||
return d->trace_response(d->post_publish_trace_control,
|
||
d->post_publish_trace_remaining,
|
||
d->post_publish_trace_slots);
|
||
}
|
||
|
||
void Plot::reset_diagnostics() {
|
||
std::visit([](auto& scene) { scene->reset_diagnostics(); }, d->scene);
|
||
d->statistics_generation.fetch_add(1, std::memory_order_acq_rel);
|
||
d->with_frame_policy([](auto& policy) { policy.reset_statistics(); });
|
||
if (d->frame_policy_3d) d->frame_policy_3d->reset_3d_statistics();
|
||
d->arm_tick_consumer(weak_from_this());
|
||
}
|
||
}
|