diff --git a/.gitignore b/.gitignore index 736ecea..3b68a56 100644 --- a/.gitignore +++ b/.gitignore @@ -23,3 +23,6 @@ /mcp/server - 副本/ /mcp/core - 副本/ /mcp/tests - 副本/ +/web_server/tests - 副本/ +/web_server/src - 副本/ +/webapp_gallery/src - 副本/ diff --git a/kernel/cmake/rely.cmake b/kernel/cmake/taskflow.cmake similarity index 97% rename from kernel/cmake/rely.cmake rename to kernel/cmake/taskflow.cmake index 03aec3c..7c1f858 100644 --- a/kernel/cmake/rely.cmake +++ b/kernel/cmake/taskflow.cmake @@ -1,6 +1,5 @@ include_guard(GLOBAL) block() - rcl_init(aethera_kernel) set(taskflow_option ${base_options}) list(APPEND taskflow_option -DTF_BUILD_EXAMPLES=OFF diff --git a/kernel/cmake/tracy.cmake b/kernel/cmake/tracy.cmake new file mode 100644 index 0000000..d2025a2 --- /dev/null +++ b/kernel/cmake/tracy.cmake @@ -0,0 +1,54 @@ +include_guard(GLOBAL) +block() + set(tracy_option ${base_options}) + list(APPEND tracy_option + -DTRACY_ENABLE=ON + -DTRACY_ON_DEMAND=OFF + -DTRACY_CALLSTACK=OFF + -DTRACY_NO_CALLSTACK=OFF + -DTRACY_NO_CALLSTACK_INLINES=OFF + -DTRACY_ONLY_LOCALHOST=OFF + -DTRACY_NO_BROADCAST=OFF + -DTRACY_NO_CODE_TRANSFER=OFF + -DTRACY_NO_CONTEXT_SWITCH=OFF + -DTRACY_NO_SAMPLING=OFF + -DTRACY_NO_FRAME_IMAGE=OFF + -DTRACY_NO_SYSTEM_TRACING=OFF + -DTRACY_DELAYED_INIT=OFF + -DTRACY_MANUAL_LIFETIME=OFF + -DTRACY_FIBERS=OFF + -DTRACY_STATIC=ON + -DTRACY_Fortran=OFF + -DTRACY_CLIENT_PYTHON=OFF + ) + _register_git_cmake_library(aethera_kernel::tracy + "https://github.com/wolfpld/tracy.git" + "v0.13.1" + ) + rcl_get_effective_install_dir( + aethera_kernel::tracy + TRACY_ROOT + ) + set(Tracy_DIR + "${TRACY_ROOT}/lib/cmake/Tracy" + ) + rcl_cmake_library_set_cmake_options( + aethera_kernel::tracy + ${tracy_option} + ) + rcl_cmake_library_set_other_use_opt( + aethera_kernel::tracy + "-DTracy_DIR=\"${Tracy_DIR}\"" + ) + rcl_cmake_library_set_init_script( + aethera_kernel::tracy + "set(TRACY_ROOT \"${TRACY_ROOT}\")" + "set(Tracy_DIR \"${Tracy_DIR}\")" + ) + rcl_cmake_library_set_clear_script( + aethera_kernel::tracy + "unset(TRACY_ROOT)" + "unset(Tracy_DIR)" + ) + rcl_export_configuration(aethera_kernel) +endblock() \ No newline at end of file diff --git a/kernel/kernel/include/statistics/Sliding_Statistics.cpp b/kernel/kernel/include/statistics/Sliding_Statistics.cpp index 78a0cfc..a7ad919 100644 --- a/kernel/kernel/include/statistics/Sliding_Statistics.cpp +++ b/kernel/kernel/include/statistics/Sliding_Statistics.cpp @@ -11,6 +11,7 @@ Sliding_Statistics::Sliding_Statistics(std::size_t window_size) : window_size(wi samples.reserve(this->window_size); } void Sliding_Statistics::add(double sample) { + if (!std::isfinite(sample)) return; if (samples.size() < window_size) { samples.push_back(sample); } diff --git a/kernel/kernel/module/task_flow/src/Taskflow_Observation.hpp b/kernel/kernel/module/task_flow/export/Taskflow_Observation.hpp similarity index 51% rename from kernel/kernel/module/task_flow/src/Taskflow_Observation.hpp rename to kernel/kernel/module/task_flow/export/Taskflow_Observation.hpp index 729bfdd..441a34e 100644 --- a/kernel/kernel/module/task_flow/src/Taskflow_Observation.hpp +++ b/kernel/kernel/module/task_flow/export/Taskflow_Observation.hpp @@ -1,19 +1,19 @@ #pragma once -#include "../export/export.h" #include "Taskflow_Trace.hpp" #include #include #include #include namespace aethera { +enum struct Take_Taskflow_Observation_Result : std::uint8_t { + not_recorded, + recording, + already_taken +}; namespace detail { struct Taskflow_Observation_Execution; } -/* - * 前台诊断显式请求的单次 Taskflow 执行记录，例如连续请求 N 帧时为每帧分别创建一个对象。 - * 它只给帧附加原始任务执行信息，不属于业务状态、持续事件流或可复用快照；未传入执行接口时不记录。 - * 复制句柄共享同一次诊断请求；执行完成后任一句柄均可取走唯一结果。 - */ +/* 单次显式诊断请求；复制句柄共享同一请求，执行结束后只能取走一次结果。 */ struct Taskflow_Observation { public: explicit Taskflow_Observation(std::string stage); @@ -22,11 +22,10 @@ public: Taskflow_Observation(Taskflow_Observation&&) noexcept; Taskflow_Observation& operator=(const Taskflow_Observation&); Taskflow_Observation& operator=(Taskflow_Observation&&) noexcept; - std::expected take(); + std::expected take(); private: friend struct detail::Taskflow_Observation_Execution; struct Private; - std::shared_ptr d; /* 本次观察请求、Worker 单写记录和完成结果的共享所有权。 */ + std::shared_ptr d; /* 本次请求、Worker 单写记录和完成结果的共享所有权。 */ }; -} +} // namespace aethera diff --git a/kernel/kernel/module/task_flow/src/Taskflow_Trace.hpp b/kernel/kernel/module/task_flow/export/Taskflow_Trace.hpp similarity index 96% rename from kernel/kernel/module/task_flow/src/Taskflow_Trace.hpp rename to kernel/kernel/module/task_flow/export/Taskflow_Trace.hpp index 67eca80..4314982 100644 --- a/kernel/kernel/module/task_flow/src/Taskflow_Trace.hpp +++ b/kernel/kernel/module/task_flow/export/Taskflow_Trace.hpp @@ -30,6 +30,6 @@ struct Taskflow_Execution_Trace { double executor_finished_ms{}; /* Taskflow completion 进入时间。 */ double observation_finished_ms{}; /* Observer 数据完成并可消费的时间。 */ std::vector nodes{}; /* 本次执行的静态 DAG 元信息。 */ - std::vector> worker_tasks{}; /* 外层下标即 Worker ID，每项仅由对应 Worker 串行写入。 */ + std::vector> worker_tasks{}; /* 外层下标即 Worker ID。 */ }; -} +} // namespace aethera diff --git a/kernel/kernel/module/task_flow/export/export.h b/kernel/kernel/module/task_flow/export/export.h index fa2432c..ab96100 100644 --- a/kernel/kernel/module/task_flow/export/export.h +++ b/kernel/kernel/module/task_flow/export/export.h @@ -1,4 +1,5 @@ #pragma once +#include "Taskflow_Observation.hpp" #include "Error_handling_specification/Failure_Policy.hpp" #include "global.hpp" #include @@ -49,15 +50,13 @@ enum struct Initialize_Task_Runtime_Result : std::uint8_t { already_initialized, initialization_in_progress }; -enum struct Take_Taskflow_Observation_Result : std::uint8_t { - not_recorded, - recording, - already_taken -}; using Taskflow_Completion = std::function; using Taskflow_Exception_Boundary = std::function; namespace task_flow { -struct Task_Node : facade_builder::add_skill::build {}; +struct Task_Node : facade_builder +::add_convention<_describe_node, Describe_Task_Graph_Node_Result(std::string, std::string)> +::add_skill +::build {}; struct Task_Graph_Component : facade_builder::add_skill::build {}; struct Task_Graph_Common_Facade : facade_builder ::add_convention<_component, proxy()> @@ -72,6 +71,7 @@ struct Task_Graph_Compose_Facade : facade_builder ::add_convention<_compose, std::expected, Compose_Task_Graph_Result>(std::string, proxy &)> ::add_convention<_precede, Precede_Task_Graph_Result(proxy &, proxy &)> ::add_convention<_run, Run_Taskflow_Result(Taskflow_Completion)> +::add_convention<_run_observed, Run_Taskflow_Result(Taskflow_Completion, Taskflow_Observation)> ::add_convention<_corun_until, void(std::function)> ::build {}; struct Task_Graph_Facade : facade_builder::add_facade_with_substitution::add_facade_with_substitution::build {}; diff --git a/kernel/kernel/module/task_flow/src/Task_Graph_Model.hpp b/kernel/kernel/module/task_flow/src/Task_Graph_Model.hpp index 415d43b..0475591 100644 --- a/kernel/kernel/module/task_flow/src/Task_Graph_Model.hpp +++ b/kernel/kernel/module/task_flow/src/Task_Graph_Model.hpp @@ -19,6 +19,7 @@ struct Task_Graph_Model::Private : Prev_Private { void clear(); bool empty() const noexcept; Run_Taskflow_Result run(Taskflow_Completion completion); + Run_Taskflow_Result run_observed(Taskflow_Completion completion, Taskflow_Observation observation); void corun_until(std::function predicate); private: std::shared_ptr graph; /* 所有消费 facade 共享的唯一权威任务图。 */ diff --git a/kernel/kernel/module/task_flow/src/Task_Runtime.hpp b/kernel/kernel/module/task_flow/src/Task_Runtime.hpp index 7cb76d8..9f16fd7 100644 --- a/kernel/kernel/module/task_flow/src/Task_Runtime.hpp +++ b/kernel/kernel/module/task_flow/src/Task_Runtime.hpp @@ -1,7 +1,7 @@ #pragma once #include "Task_Graph.hpp" #include "../export/export.h" -#include "Taskflow_Observation.hpp" +#include "../export/Taskflow_Observation.hpp" #include #include #include diff --git a/kernel/kernel/module/task_flow/src/Taskflow_Observation.cpp b/kernel/kernel/module/task_flow/src/Taskflow_Observation.cpp index 159c934..0289c43 100644 --- a/kernel/kernel/module/task_flow/src/Taskflow_Observation.cpp +++ b/kernel/kernel/module/task_flow/src/Taskflow_Observation.cpp @@ -1,4 +1,4 @@ -#include "Taskflow_Observation.hpp" +#include "../export/Taskflow_Observation.hpp" #include "detail/Taskflow_Execution.ipp" #include "Error_handling_specification/Failure_Policy.hpp" #include diff --git a/kernel/kernel/module/task_flow/src/detail/Taskflow_Execution.ipp b/kernel/kernel/module/task_flow/src/detail/Taskflow_Execution.ipp index c6a7572..fea7773 100644 --- a/kernel/kernel/module/task_flow/src/detail/Taskflow_Execution.ipp +++ b/kernel/kernel/module/task_flow/src/detail/Taskflow_Execution.ipp @@ -1,8 +1,8 @@ #pragma once #include "../Task_Graph.hpp" #include "../../export/export.h" -#include "../Taskflow_Observation.hpp" -#include "../Taskflow_Trace.hpp" +#include "../../export/Taskflow_Observation.hpp" +#include "../../export/Taskflow_Trace.hpp" #include #include #include diff --git a/kernel/kernel/module/task_flow/src/task_flow_facade.cpp b/kernel/kernel/module/task_flow/src/task_flow_facade.cpp index 59adb90..de5a947 100644 --- a/kernel/kernel/module/task_flow/src/task_flow_facade.cpp +++ b/kernel/kernel/module/task_flow/src/task_flow_facade.cpp @@ -5,6 +5,9 @@ namespace aethera::task_flow { namespace detail { struct Proxied_Task_Node { explicit Proxied_Task_Node(aethera::Task_Node value) : value(std::move(value)) {} + Describe_Task_Graph_Node_Result describe_node(std::string key, std::string description) { + return value.describe(std::move(key), std::move(description)); + } aethera::Task_Node value; }; struct Proxied_Task_Graph_Component { @@ -42,6 +45,9 @@ bool Task_Graph_Model::Private::empty() const noexcept { Run_Taskflow_Result Task_Graph_Model::Private::run(Taskflow_Completion completion) { return run_taskflow(*graph, std::move(completion)); } +Run_Taskflow_Result Task_Graph_Model::Private::run_observed(Taskflow_Completion completion, Taskflow_Observation observation) { + return run_taskflow(*graph, std::move(completion), std::move(observation)); +} void Task_Graph_Model::Private::corun_until(std::function predicate) { aethera::detail::corun_taskflow_until(std::move(predicate)); } diff --git a/kernel/main.cmake b/kernel/main.cmake index f3da7c6..cebdc2e 100644 --- a/kernel/main.cmake +++ b/kernel/main.cmake @@ -1,5 +1,7 @@ -include(${CMAKE_CURRENT_LIST_DIR}/cmake/rely.cmake) -set(Aethera_Kernel_dependencies global::proxy aethera_kernel::taskflow global::magic_enum) +rcl_init(aethera_kernel) +include(${CMAKE_CURRENT_LIST_DIR}/cmake/taskflow.cmake) +include(${CMAKE_CURRENT_LIST_DIR}/cmake/tracy.cmake) +set(Aethera_Kernel_dependencies global::proxy aethera_kernel::tracy aethera_kernel::taskflow global::magic_enum) set(Aethera_BUILD_TESTS TRUE) if (Aethera_BUILD_TESTS) enable_testing() @@ -16,10 +18,12 @@ if (Aethera_Kernel_dependencies_installed) return() endif () rcl_load_dependency_environment(${Aethera_Kernel_dependencies}) +#option(TRACY_ENABLE "" ON) find_package(Threads REQUIRED) find_package(Taskflow CONFIG REQUIRED) find_package(magic_enum CONFIG REQUIRED) find_package(msft_proxy4 4.1 CONFIG REQUIRED) +find_package(Tracy CONFIG REQUIRED) if (Aethera_BUILD_TESTS) find_package(GTest CONFIG REQUIRED) find_package(benchmark CONFIG REQUIRED) @@ -40,6 +44,7 @@ target_compile_features(Aethera_Kernel_Core PUBLIC cxx_std_20) target_link_libraries(Aethera_Kernel_Core PUBLIC magic_enum::magic_enum msft_proxy4::proxy + Tracy::TracyClient "$" ) if (MSVC) @@ -62,7 +67,6 @@ if (Aethera_BUILD_TESTS) target_link_libraries(Aethera_Kernel_exe PRIVATE Aethera_Kernel GTest::gtest) - append_glob_source(Aethera_Kernel_module_test_sources "${Aethera_Kernel_test_dir}/error" "${Aethera_Kernel_test_dir}/model") diff --git a/mcp/core/Control_Service.cpp b/mcp/core/Control_Service.cpp index 7008f31..5a37dc1 100644 --- a/mcp/core/Control_Service.cpp +++ b/mcp/core/Control_Service.cpp @@ -5,6 +5,7 @@ #include "runtime/Datoviz_Observation_Json.hpp" #include "runtime/Gallery_Plots.hpp" #include "runtime/Input_Event.hpp" +#include "runtime/Taskflow_Trace_Json.hpp" #include #include #include @@ -71,6 +72,22 @@ template return std::visit([](auto value) { return std::string{magic_enum::enum_name(value)}; }, error); } +[[nodiscard]] nlohmann::json frame_policy_state_json(const Frame_Policy::State& state) { + return { + {"timer_ticks", state.timer_ticks}, + {"dropped_timer_ticks", state.dropped_timer_ticks}, + {"completed_frames", state.completed_frames}, + {"effective_frames_per_second", state.effective_frames_per_second}, + {"render_capacity_fps", state.render_capacity_fps ? nlohmann::json{*state.render_capacity_fps} : nlohmann::json{nullptr}}, + {"send_capacity_fps", state.send_capacity_fps ? nlohmann::json{*state.send_capacity_fps} : nlohmann::json{nullptr}}, + {"render_time", statistics_json(state.render_time_ns)}, + {"send_time", statistics_json(state.send_time_ns)}, + {"end_to_end_time", statistics_json(state.end_to_end_time_ns)}, + {"configuration_error", state.configuration_error ? nlohmann::json{magic_enum::enum_name(*state.configuration_error)} : nlohmann::json{nullptr}}, + {"render_submission_error", state.render_submission_error ? nlohmann::json{render_error_name(*state.render_submission_error)} : nlohmann::json{nullptr}} + }; +} + [[nodiscard]] nlohmann::json frame_policy_schema(const Throttled_Latest_only& policy) { const auto user_rate = policy.get(&Frame_Policy::Prop::user_frames_per_second); return { @@ -194,8 +211,6 @@ scene::Render_Result map_render_result(render_3d::Render_Scene_3D::Render_Result } // namespace -Control_Service::Private::Gallery_Frame_Slot::Gallery_Frame_Slot(bool use_3d) : native(use_3d ? Native_Frame{std::in_place_type>, std::make_unique(render_3d::Frame_Identity{})} : Native_Frame{std::in_place_type>, std::make_unique(render_2d::Frame_Identity{})}) {} - proxy Control_Service::Private::Policy_Scene::Private::create_frame() { return entry->create_frame(); } @@ -204,11 +219,7 @@ scene::Render_Result Control_Service::Private::Policy_Scene::Private::render(pro return entry->render(frame, std::move(completion)); } -void Control_Service::Private::Policy_Sink::Private::send(proxy& frame, frame_policy::Frame_Completion completion) { - entry->send(frame, std::move(completion)); -} - -Control_Service::Private::Entry::Entry(std::string value_id, web::Gallery_Build build, Gallery_Output value_output) : components(std::move(build.first)), scene(std::move(build.second)), id(std::move(value_id)), output(std::move(value_output)) {} +Control_Service::Private::Entry::Entry(std::string value_id, web::Gallery_Build build, proxy value_sink, std::uint32_t width, std::uint32_t height) : components(std::move(build.first)), scene(std::move(build.second)), sink(std::move(value_sink)), id(std::move(value_id)), output_width(width), output_height(height) {} Control_Service::Private::Entry::~Entry() = default; @@ -230,8 +241,7 @@ std::shared_ptr Control_Service::Private::Entry::current_ void Control_Service::Private::Entry::start() { auto timer = frame_policy::make_timer_service(); auto scene_endpoint = make_model_proxy(this); - auto sink_endpoint = make_model_proxy(this); - Throttled_Latest_only::Builder builder(std::move(timer), std::move(scene_endpoint), std::move(sink_endpoint)); + Throttled_Latest_only::Builder builder(std::move(timer), std::move(scene_endpoint), std::move(sink)); builder .set(&Frame_Policy::Prop::user_frames_per_second, std::optional{100.0}) .set(&Frame_Policy::Prop::render_rate_limit_enabled, false) @@ -260,13 +270,38 @@ void Control_Service::Private::Entry::request_frame(web::Gallery_Frame_Request r requested_frame.store(std::make_shared(std::move(request)), std::memory_order_release); } +bool Control_Service::Private::Entry::reserve_trace() noexcept { + 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)) return true; + return false; +} + +void Control_Service::Private::Entry::restore_trace() noexcept { + taskflow_trace_remaining.fetch_add(1, std::memory_order_release); +} + +void Control_Service::Private::Entry::store_trace(const Taskflow_Execution_Trace& trace, const web::Gallery_Frame_Request& request, const nlohmann::json& backend) { + auto value = web::taskflow_trace_json(trace, request.sequence, request.correlation_id, {}, backend); + if (const auto current = current_policy()) current->get([&](const Frame_Policy::State& state) { value["frame_policy"] = frame_policy_state_json(state); }); + auto encoded = std::make_shared(std::move(value)); + auto control = taskflow_trace_control.load(std::memory_order_acquire); + for (;;) { + const auto requested = static_cast(control >> 32U); + const auto captured = static_cast(control); + if (captured >= requested) return; + taskflow_trace_slots[captured].store(encoded, std::memory_order_release); + const auto next = (static_cast(requested) << 32U) | static_cast(captured + 1U); + if (taskflow_trace_control.compare_exchange_weak(control, next, std::memory_order_release, std::memory_order_acquire)) return; + } +} + proxy Control_Service::Private::Entry::create_frame() { const bool use_3d = std::holds_alternative>(scene); - return pro::make_proxy(use_3d); + return pro::make_proxy(use_3d); } scene::Render_Result Control_Service::Private::Entry::render(proxy& frame, scene::Render_Completion completion) { - auto* slot = frame ? proxy_cast(&*frame) : nullptr; + auto* slot = frame ? proxy_cast(&*frame) : nullptr; if (!slot) return std::unexpected(scene::Render_Error{scene::Render_State_Result::frame_missing}); if (!completion) return std::unexpected(scene::Render_Error{scene::Render_State_Result::completion_missing}); auto request_owner = requested_frame.exchange({}, std::memory_order_acq_rel); @@ -274,95 +309,59 @@ scene::Render_Result Control_Service::Private::Entry::render(proxy if (!request_owner) { request.issued_at = std::chrono::steady_clock::now(); request.time_milliseconds = std::chrono::duration(request.issued_at - started_at).count(); - request.width = output.width; - request.height = output.height; + request.width = output_width; + request.height = output_height; } request.sequence = next_sequence.fetch_add(1, std::memory_order_relaxed) + 1; if (request.correlation_id == 0) request.correlation_id = request.sequence; slot->request = request; ++slot->generation; + slot->taskflow_trace_reserved = reserve_trace(); components->update(request); auto* policy_frame = std::addressof(frame); /* Non-owning pointer to a stable policy slot; Frame Policy keeps it alive until completion returns. */ if (auto* native = std::get_if>(&slot->native)) { auto& scene_2d = *std::get>(scene); scene_2d.set(&render_2d::Render_Scene_2D::Prop::viewport, render_2d::Size{static_cast(request.width), static_cast(request.height)}); (*native)->begin({request.sequence, slot->generation}, render_2d::Frame_2D::native_pixel_format); - return map_render_result(scene_2d.render(native->get(), [policy_frame, completion = std::move(completion)](render_2d::Frame_2D*, std::exception_ptr failure) mutable { completion(*policy_frame, std::move(failure)); })); + if (slot->taskflow_trace_reserved) (*native)->request_taskflow_trace(); + auto result = map_render_result(scene_2d.render(native->get(), [this, slot, policy_frame, completion = std::move(completion)](render_2d::Frame_2D*, std::exception_ptr failure) mutable { + if (std::exchange(slot->taskflow_trace_reserved, false)) { + if (!failure) { + if (auto trace = std::get>(slot->native)->take_taskflow_trace()) store_trace(*trace, slot->request); + else restore_trace(); + } + else restore_trace(); + } + completion(*policy_frame, std::move(failure)); + })); + if (!result && std::exchange(slot->taskflow_trace_reserved, false)) restore_trace(); + return result; } auto& scene_3d = *std::get>(scene); auto& native = std::get>(slot->native); scene_3d.set(&render_3d::Render_Scene_3D::Prop::viewport, render_3d::Extent{request.width, request.height}); native->begin({request.sequence, slot->generation}, render_3d::Frame_3D_Output::pixels, render_3d::Frame_3D::native_pixel_format); - return map_render_result(scene_3d.render(native.get(), [policy_frame, completion = std::move(completion)](render_3d::Frame_3D*, std::exception_ptr failure) mutable { completion(*policy_frame, std::move(failure)); })); -} - -std::optional Control_Service::Private::Entry::take_pending_send(std::uint64_t sequence) noexcept { - std::lock_guard lock(pending_send_mutex); - const auto found = pending_sends.find(sequence); - if (found == pending_sends.end()) return std::nullopt; - auto pending = std::move(found->second); - pending_sends.erase(found); - return pending; -} - -void Control_Service::Private::Entry::send(proxy& frame, frame_policy::Frame_Completion completion) { - auto* slot = frame ? proxy_cast(&*frame) : nullptr; - if (!slot) throw std::logic_error("Gallery Sink received an unknown frame"); - std::shared_ptr> storage; - web::Gallery_Pixel_Layout layout{web::Gallery_Pixel_Layout::rgba8}; - std::uint64_t rendered_sequence = slot->request.sequence; - std::uint64_t rendered_correlation_id = slot->request.correlation_id; - std::uint32_t width{}; - std::uint32_t height{}; - if (auto* native = std::get_if>(&slot->native)) { - layout = web::Gallery_Pixel_Layout::bgra8; - auto pixels = (*native)->output_pixels(); - width = static_cast(pixels.width); - height = static_cast(pixels.height); - storage = std::make_shared>(std::move(pixels.bytes)); - } - else { - auto& native_3d = std::get>(slot->native); - const auto rendered = native_3d->rendered_identity(); - rendered_sequence = rendered.sequence; - rendered_correlation_id = slot->request.correlation_id; - const auto extent = native_3d->extent(); - width = extent.width; - height = extent.height; - storage = native_3d->share_pixels(); - if (auto observation = native_3d->take_datoviz_observation()) datoviz_observation.store(std::make_shared(datoviz_observation_json(*observation)), std::memory_order_release); - } - const auto timestamp = std::chrono::duration_cast(std::chrono::duration(slot->request.time_milliseconds)); - auto pixels = std::make_shared(web::Gallery_Pixel_Frame{std::move(storage), layout, timestamp, slot->request.sequence, slot->request.correlation_id, rendered_sequence, rendered_correlation_id, width, height}); - auto publication = std::make_shared(web::Gallery_Frame{std::move(pixels)}); - if (!output.publish) { - ++ignored_publications; - completion(frame); - return; - } - { - std::lock_guard lock(pending_send_mutex); - const auto [_, inserted] = pending_sends.emplace(slot->request.sequence, Pending_Send{frame, std::move(completion)}); - if (!inserted) throw std::logic_error("Gallery publication sequence is already in flight"); - } - web::Gallery_Frame_Publication result; - try { - result = output.publish(id, std::move(publication)); - } - catch (...) { - (void)take_pending_send(slot->request.sequence); - throw; - } - if (result == web::Gallery_Frame_Publication::asynchronous) return; - if (result == web::Gallery_Frame_Publication::ignored) ++ignored_publications; - if (auto pending = take_pending_send(slot->request.sequence)) pending->completion(pending->frame.get()); -} - -void Control_Service::Private::Entry::publication_feedback(web::Gallery_Publication_Feedback feedback) noexcept { - auto pending = take_pending_send(feedback.sequence); - if (!pending) return; - if (!feedback.succeeded) ++failed_publications; - pending->completion(pending->frame.get()); + if (slot->taskflow_trace_reserved) native->request_taskflow_trace(); + auto result = map_render_result(scene_3d.render(native.get(), [this, slot, policy_frame, completion = std::move(completion)](render_3d::Frame_3D*, std::exception_ptr failure) mutable { + auto& completed = std::get>(slot->native); + nlohmann::json backend; + if (!failure) { + if (auto observation = completed->take_datoviz_observation()) { + backend = datoviz_observation_json(*observation); + datoviz_observation.store(std::make_shared(backend), std::memory_order_release); + } + } + if (std::exchange(slot->taskflow_trace_reserved, false)) { + if (!failure) { + if (auto trace = completed->take_taskflow_trace()) store_trace(*trace, slot->request, backend); + else restore_trace(); + } + else restore_trace(); + } + completion(*policy_frame, std::move(failure)); + })); + if (!result && std::exchange(slot->taskflow_trace_reserved, false)) restore_trace(); + return result; } nlohmann::json Control_Service::Private::Entry::schema() const { @@ -379,11 +378,6 @@ nlohmann::json Control_Service::Private::Entry::schema() const { nlohmann::json Control_Service::Private::Entry::diagnostics() const { const auto current = current_policy(); if (!current) return {{"protocol", "aethera.plot.diagnostics"}, {"version", 6}, {"available", false}, {"plot", id}}; - std::size_t in_flight_publications{}; - { - std::lock_guard lock(pending_send_mutex); - in_flight_publications = pending_sends.size(); - } nlohmann::json result; current->get([&](const Frame_Policy::State& state) { result = { @@ -392,30 +386,42 @@ nlohmann::json Control_Service::Private::Entry::diagnostics() const { {"available", true}, {"plot", id}, {"dimension", std::holds_alternative>(scene) ? "3D" : "2D"}, - {"frame_policy", { - {"timer_ticks", state.timer_ticks}, - {"dropped_timer_ticks", state.dropped_timer_ticks}, - {"completed_frames", state.completed_frames}, - {"effective_frames_per_second", state.effective_frames_per_second}, - {"render_capacity_fps", state.render_capacity_fps ? nlohmann::json{*state.render_capacity_fps} : nlohmann::json{nullptr}}, - {"send_capacity_fps", state.send_capacity_fps ? nlohmann::json{*state.send_capacity_fps} : nlohmann::json{nullptr}}, - {"render_time", statistics_json(state.render_time_ns)}, - {"send_time", statistics_json(state.send_time_ns)}, - {"end_to_end_time", statistics_json(state.end_to_end_time_ns)}, - {"configuration_error", state.configuration_error ? nlohmann::json{magic_enum::enum_name(*state.configuration_error)} : nlohmann::json{nullptr}}, - {"render_submission_error", state.render_submission_error ? nlohmann::json{render_error_name(*state.render_submission_error)} : nlohmann::json{nullptr}} - }}, - {"publication", {{"in_flight", in_flight_publications}, {"ignored", ignored_publications.load(std::memory_order_relaxed)}, {"failed", failed_publications.load(std::memory_order_relaxed)}}} + {"frame_policy", frame_policy_state_json(state)} }; }); if (const auto observation = datoviz_observation.load(std::memory_order_acquire)) result["datoviz"] = *observation; return result; } -Control_Service::Control_Service(Gallery_Output output) : d(std::make_unique()) { - d->output = std::move(output); +void Control_Service::Private::Entry::reset_diagnostics() noexcept { + datoviz_observation.store({}, std::memory_order_release); } +void Control_Service::Private::Entry::request_trace(std::size_t frame_count) { + if (frame_count == 0 || frame_count > maximum_taskflow_trace_frames) throw std::invalid_argument("Plot Taskflow trace frame_count must be between 1 and 120"); + auto control = taskflow_trace_control.load(std::memory_order_acquire); + for (;;) { + const auto requested = static_cast(control >> 32U); + const auto captured = static_cast(control); + if (requested != captured) throw std::logic_error("A Plot Taskflow trace request is already active"); + const auto next = static_cast(frame_count) << 32U; + if (taskflow_trace_control.compare_exchange_weak(control, next, std::memory_order_release, std::memory_order_acquire)) break; + } + for (auto& slot : taskflow_trace_slots) slot.store({}, std::memory_order_release); + taskflow_trace_remaining.store(frame_count, std::memory_order_release); +} + +nlohmann::json Control_Service::Private::Entry::trace() const { + nlohmann::json frames = nlohmann::json::array(); + const auto control = taskflow_trace_control.load(std::memory_order_acquire); + const auto requested = static_cast(control >> 32U); + const auto captured = static_cast(control); + for (std::uint32_t index = 0; index < captured; ++index) if (const auto frame = taskflow_trace_slots[index].load(std::memory_order_acquire)) frames.push_back(*frame); + return {{"protocol", "aethera.taskflow.frames"}, {"version", 1}, {"requested", requested}, {"remaining", taskflow_trace_remaining.load(std::memory_order_acquire)}, {"captured", frames.size()}, {"complete", requested != 0 && frames.size() == requested}, {"frames", std::move(frames)}}; +} + +Control_Service::Control_Service() : d(std::make_unique()) {} + Control_Service::~Control_Service() { if (!d || d->stop_started) return; try { @@ -427,10 +433,13 @@ Control_Service::~Control_Service() { } std::shared_ptr Control_Service::create(Gallery_Output output) { - auto service = std::shared_ptr(new Control_Service(std::move(output))); + auto service = std::shared_ptr(new Control_Service); service->d->plots.reserve(web::gallery_plot_definitions().size()); for (const auto& definition : web::gallery_plot_definitions()) { - auto entry = std::make_shared(std::string{definition.id}, definition.create(), service->d->output); + auto sink = output.make_sink ? output.make_sink(definition.id) : proxy{}; + /* Headless MCP and partial test catalogs intentionally complete frames without media. */ + if (!sink) sink = make_model_proxy(); + auto entry = std::make_shared(std::string{definition.id}, definition.create(), std::move(sink), output.width, output.height); service->d->plots.emplace(entry->id, entry); entry->start(); } @@ -476,11 +485,6 @@ void Control_Service::request_frame(std::string_view id, web::Gallery_Frame_Requ found->second->request_frame(std::move(request)); } -void Control_Service::submit_publication_feedback(std::string_view id, web::Gallery_Publication_Feedback feedback) noexcept { - const auto found = d->plots.find(std::string{id}); - if (found != d->plots.end()) found->second->publication_feedback(std::move(feedback)); -} - nlohmann::json Control_Service::plot_schema(std::string_view id) const { const auto found = d->plots.find(std::string{id}); if (found == d->plots.end()) throw std::invalid_argument("unknown plot"); @@ -540,6 +544,24 @@ nlohmann::json Control_Service::plot_diagnostics(std::string_view id) const { return found->second->diagnostics(); } +void Control_Service::reset_plot_diagnostics(std::string_view id) { + const auto found = d->plots.find(std::string{id}); + if (found == d->plots.end()) throw std::invalid_argument("unknown plot"); + found->second->reset_diagnostics(); +} + +void Control_Service::request_plot_taskflow_trace(std::string_view id, std::size_t frame_count) { + const auto found = d->plots.find(std::string{id}); + if (found == d->plots.end()) throw std::invalid_argument("unknown plot"); + found->second->request_trace(frame_count); +} + +nlohmann::json Control_Service::plot_taskflow_trace(std::string_view id) const { + const auto found = d->plots.find(std::string{id}); + if (found == d->plots.end()) throw std::invalid_argument("unknown plot"); + return found->second->trace(); +} + nlohmann::json Control_Service::plot_catalog() const { nlohmann::json result = nlohmann::json::array(); for (const auto& definition : web::gallery_plot_definitions()) result.push_back({{"id", definition.id}, {"title", definition.title}, {"category", definition.category}, {"description", definition.description}, {"dimension", web::plot_dimension_name(definition.dimension)}}); diff --git a/mcp/core/Control_Service.hpp b/mcp/core/Control_Service.hpp index e3f3f8a..89f2c6b 100644 --- a/mcp/core/Control_Service.hpp +++ b/mcp/core/Control_Service.hpp @@ -1,6 +1,7 @@ #pragma once #include "runtime/Gallery_Runtime.hpp" #include +#include #include #include #include @@ -14,11 +15,10 @@ struct Gallery_Output { std::uint32_t width{720}; std::uint32_t height{420}; /* - * Entry stop is asynchronous. Values borrowed by this callable must remain - * valid until every asynchronous publication has reported feedback and the - * final Entry retirement has released its copy of the callable. + * Creates the concrete Sink proxy owned by each Plot Frame Policy. The proxy + * decides ownership; borrowed captures must cover asynchronous policy stop. */ - web::Gallery_Frame_Publisher publish{}; + std::function(std::string_view)> make_sink{}; }; enum struct Tool_Call_Result : std::uint8_t { @@ -46,7 +46,6 @@ struct Control_Service final { void stop(std::function completion); void submit_input(std::string_view id, std::unique_ptr event); void request_frame(std::string_view id, web::Gallery_Frame_Request request); - void submit_publication_feedback(std::string_view id, web::Gallery_Publication_Feedback feedback) noexcept; [[nodiscard]] nlohmann::json plot_schema(std::string_view id) const; [[nodiscard]] nlohmann::json write_plot_property( std::string_view id, std::string_view component, @@ -56,13 +55,16 @@ struct Control_Service final { [[nodiscard]] nlohmann::json generate_plot_data( std::string_view id, const nlohmann::json& input); [[nodiscard]] nlohmann::json plot_diagnostics(std::string_view id) const; + void reset_plot_diagnostics(std::string_view id); + void request_plot_taskflow_trace(std::string_view id, std::size_t frame_count); + [[nodiscard]] nlohmann::json plot_taskflow_trace(std::string_view id) const; [[nodiscard]] nlohmann::json plot_catalog() const; [[nodiscard]] nlohmann::json tool_catalog() const; [[nodiscard]] Tool_Call_Output call_tool( std::string_view name, const nlohmann::json& arguments); private: - explicit Control_Service(Gallery_Output output); + Control_Service(); struct Private; std::unique_ptr d; }; diff --git a/mcp/core/Control_Service.ipp b/mcp/core/Control_Service.ipp index 4d952b8..0c83fd0 100644 --- a/mcp/core/Control_Service.ipp +++ b/mcp/core/Control_Service.ipp @@ -1,8 +1,8 @@ #pragma once #include "runtime/Gallery_Runtime.hpp" #include +#include #include -#include #include #include @@ -13,57 +13,49 @@ struct Control_Service::Private { struct Entry_Retirement; struct Entry_Stop_Lifetime; - struct Gallery_Frame_Slot { - using Native_Frame = std::variant, std::unique_ptr>; - explicit Gallery_Frame_Slot(bool use_3d); - Native_Frame native; /* Sole owner of the dimension-specific physical frame stored in one Frame Policy slot. */ - web::Gallery_Frame_Request request{}; /* Metadata for the current use of this physical slot. */ - std::uint64_t generation{}; /* Slot-local reuse generation. */ - }; - struct Policy_Scene : Def { struct Private; }; - struct Policy_Sink : Def { + struct Headless_Sink : Def { struct Private; }; struct Entry { - struct Pending_Send { - std::reference_wrapper> frame; /* Non-owning borrow of the policy slot; completion releases the borrow. */ - frame_policy::Frame_Completion completion; /* Sole owner of the asynchronous Sink completion. */ - }; - - Entry(std::string id, web::Gallery_Build build, Gallery_Output output); + Entry(std::string id, web::Gallery_Build build, proxy sink, std::uint32_t width, std::uint32_t height); ~Entry(); Entry(const Entry&) = delete; Entry& operator=(const Entry&) = delete; void start(); void stop(std::shared_ptr retirement) noexcept; void request_frame(web::Gallery_Frame_Request request); - void publication_feedback(web::Gallery_Publication_Feedback feedback) noexcept; proxy create_frame(); scene::Render_Result render(proxy& frame, scene::Render_Completion completion); - void send(proxy& frame, frame_policy::Frame_Completion completion); [[nodiscard]] nlohmann::json schema() const; [[nodiscard]] nlohmann::json diagnostics() const; + void reset_diagnostics() noexcept; + void request_trace(std::size_t frame_count); + [[nodiscard]] nlohmann::json trace() const; [[nodiscard]] std::shared_ptr current_policy() const noexcept; - [[nodiscard]] std::optional take_pending_send(std::uint64_t sequence) noexcept; + [[nodiscard]] bool reserve_trace() noexcept; + void restore_trace() noexcept; + void store_trace(const Taskflow_Execution_Trace& trace, const web::Gallery_Frame_Request& request, const nlohmann::json& backend = {}); std::unique_ptr components; /* Sole owner of plot models, camera and axes; declared before Scene so it outlives every Scene borrow. */ web::Gallery_Scene scene; /* Sole owner of the dimension-specific Scene. */ + proxy sink; /* Sole owner until start transfers the concrete media Sink into Frame Policy. */ std::atomic> policy{}; /* Shared only across asynchronous stop retirement. */ std::string id; /* Stable catalog identifier. */ - Gallery_Output output; /* Owns a publisher copy through asynchronous Entry retirement; its borrowed captures follow Gallery_Output's contract. */ + std::uint32_t output_width{}; /* Default width for timer-driven requests. */ + std::uint32_t output_height{}; /* Default height for timer-driven requests. */ std::chrono::steady_clock::time_point started_at{std::chrono::steady_clock::now()}; /* Runtime animation clock origin. */ std::atomic_uint64_t next_sequence{}; /* Authoritative next rendered-frame sequence source. */ std::atomic> requested_frame{}; /* Latest explicit request consumed by one subsequent policy tick. */ std::atomic> datoviz_observation{}; /* Latest immutable 3D backend diagnostic publication. */ - std::atomic_uint64_t ignored_publications{}; /* Frames for which no publisher accepted ownership. */ - std::atomic_uint64_t failed_publications{}; /* Asynchronous publications reported unsuccessful. */ - mutable std::mutex pending_send_mutex; /* Protects the publisher-feedback to Frame Policy Sink completion boundary. */ - std::unordered_map pending_sends; /* In-flight asynchronous publisher borrows keyed by frame sequence. */ + static constexpr std::size_t maximum_taskflow_trace_frames{120}; + std::atomic_size_t taskflow_trace_remaining{}; /* Number of requested successful Scene executions not yet captured. */ + std::atomic_uint64_t taskflow_trace_control{}; /* High 32 bits requested, low 32 bits captured. */ + std::array>, maximum_taskflow_trace_frames> taskflow_trace_slots{}; /* Immutable captured frame traces. */ }; struct Entry_Retirement { @@ -77,7 +69,6 @@ struct Control_Service::Private { std::shared_ptr policy; }; - Gallery_Output output{}; /* Immutable configuration used when each Entry is constructed. */ std::unordered_map> plots; /* Stable plot registry; entries retire themselves asynchronously. */ bool stop_started{}; /* Sole Control_Service lifecycle authority; public operations are unavailable after the one shutdown begins. */ }; @@ -89,10 +80,8 @@ struct Control_Service::Private::Policy_Scene::Private : Prev_Private { Entry* entry; /* Required non-owning borrow; Entry owns the policy and outlives asynchronous stop completion. */ }; -struct Control_Service::Private::Policy_Sink::Private : Prev_Private { - explicit Private(Entry* entry) : entry(entry) {} - void send(proxy& frame, frame_policy::Frame_Completion completion); - Entry* entry; /* Required non-owning borrow; Entry owns the policy and outlives asynchronous stop completion. */ +struct Control_Service::Private::Headless_Sink::Private : Prev_Private { + void send(proxy& frame, frame_policy::Frame_Completion completion) { completion(frame); } }; } // namespace aethera::mcp diff --git a/mcp/core/runtime/Gallery_Runtime.hpp b/mcp/core/runtime/Gallery_Runtime.hpp index 16ecc90..5e9b8b4 100644 --- a/mcp/core/runtime/Gallery_Runtime.hpp +++ b/mcp/core/runtime/Gallery_Runtime.hpp @@ -2,6 +2,7 @@ #include #include #include +#include #include #include #include @@ -29,8 +30,17 @@ struct Gallery_Frame_Request { std::uint32_t height{420}; /* Requested physical pixel height. */ }; +struct Gallery_Frame_Slot { + using Native_Frame = std::variant, std::unique_ptr>; + explicit Gallery_Frame_Slot(bool use_3d) : native(use_3d ? Native_Frame{std::in_place_type>, std::make_unique(render_3d::Frame_Identity{})} : Native_Frame{std::in_place_type>, std::make_unique(render_2d::Frame_Identity{})}) {} + Native_Frame native; /* Sole owner of the dimension-specific physical frame held by one Frame Policy slot. */ + Gallery_Frame_Request request{}; /* Metadata for the current use of this slot. */ + std::uint64_t generation{}; /* Slot-local reuse generation. */ + bool taskflow_trace_reserved{}; /* This frame owns one requested diagnostic capture until Scene completion consumes it. */ +}; + struct Gallery_Pixel_Frame { - std::shared_ptr> pixels{}; /* Shared immutable bytes; asynchronous publishers retain this owner until feedback. */ + std::shared_ptr> pixels{}; /* Shared immutable bytes retained by the asynchronous FFmpeg Sink taskflow. */ Gallery_Pixel_Layout layout{Gallery_Pixel_Layout::rgba8}; /* Byte channel order. */ std::chrono::microseconds presentation_time{}; /* Plot animation timestamp. */ std::uint64_t sequence{}; /* Runtime-assigned monotonic rendered-frame sequence. */ @@ -41,24 +51,6 @@ struct Gallery_Pixel_Frame { std::uint32_t height{}; /* Pixel height. */ }; -struct Gallery_Frame { - std::shared_ptr pixels{}; /* Shared immutable publication payload. */ -}; - -enum struct Gallery_Frame_Publication : std::uint8_t { - ignored, /* No consumer accepted the frame; the Sink may release it immediately. */ - completed, /* Publication completed before the publisher returned. */ - asynchronous /* Publication retained the frame and will return Gallery_Publication_Feedback. */ -}; - -struct Gallery_Publication_Feedback { - std::chrono::steady_clock::time_point completed_at{}; /* Monotonic completion time reported by the asynchronous publisher. */ - std::uint64_t sequence{}; /* Gallery frame sequence whose Sink borrow is released. */ - bool succeeded{}; /* Whether the asynchronous publisher completed successfully. */ -}; - -using Gallery_Frame_Publisher = std::function)>; - struct Gallery_Component { virtual ~Gallery_Component() = default; [[nodiscard]] virtual nlohmann::json schema() const = 0; diff --git a/mcp/core/runtime/Taskflow_Trace_Json.cpp b/mcp/core/runtime/Taskflow_Trace_Json.cpp new file mode 100644 index 0000000..0c4ad7a --- /dev/null +++ b/mcp/core/runtime/Taskflow_Trace_Json.cpp @@ -0,0 +1,43 @@ +#include "Taskflow_Trace_Json.hpp" +#include +#include +#include +namespace aethera::web { +nlohmann::json taskflow_trace_json(const Taskflow_Execution_Trace& trace, std::uint64_t sequence, std::uint64_t correlation_id, const nlohmann::json& captured_components, const nlohmann::json& captured_backend) { + std::unordered_map node_ids; + std::unordered_map> successors; + for (const auto& node : trace.nodes) { + node_ids.emplace(node.native_id, node.node_id); + for (const auto predecessor : node.predecessors) successors[predecessor].push_back(node.native_id); + } + nlohmann::json nodes = nlohmann::json::array(); + for (const auto& node : trace.nodes) { + nlohmann::json predecessors = nlohmann::json::array(); + for (const auto native_id : node.predecessors) predecessors.push_back(std::to_string(native_id)); + nlohmann::json following = nlohmann::json::array(); + for (const auto native_id : successors[node.native_id]) following.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(following)}, {"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)); + } + nlohmann::json executions = nlohmann::json::array(); + for (std::size_t worker = 0; worker < trace.worker_tasks.size(); ++worker) { + for (const auto& task : trace.worker_tasks[worker]) { + 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", worker}, {"worker_queue_size", task.worker_queue_size}, {"worker_queue_capacity", task.worker_queue_capacity}, {"ready_ms", task.entered_ms}, {"entered_ms", task.entered_ms}, {"started_ms", task.started_ms}, {"finished_ms", task.finished_ms}, {"completed_ms", task.completed_ms}, {"duration_ms", std::max(0.0, task.finished_ms - task.started_ms)}, {"cooperative_wait_ms", 0.0}, {"cooperative_waits", nlohmann::json::array()}, {"observer_entry_ms", task.entered_ms}, {"observer_exit_ms", task.completed_ms}, {"queue_wait_ms", std::max(0.0, task.started_ms - task.entered_ms)}}); + } + } + nlohmann::json graphs = nlohmann::json::array({{{"stage", trace.stage}, {"name", trace.taskflow_name}, {"submitted_ms", 0.0}, {"finished_ms", trace.executor_finished_ms}, {"completed", true}, {"nodes", std::move(nodes)}}}); + nlohmann::json result{{"sequence", sequence}, {"correlation_id", correlation_id}, {"request_source", "diagnostic"}, {"created_time_unix_ns", trace.started_time_unix_ns}, {"worker_count", trace.worker_tasks.size()}, {"markers", nlohmann::json::object()}, {"measurements", nlohmann::json::object()}, {"graphs", std::move(graphs)}, {"executions", std::move(executions)}}; + if (!captured_backend.empty()) result["datoviz"] = captured_backend; + return result; +} +} // namespace aethera::web diff --git a/mcp/core/runtime/Taskflow_Trace_Json.hpp b/mcp/core/runtime/Taskflow_Trace_Json.hpp new file mode 100644 index 0000000..35bbab6 --- /dev/null +++ b/mcp/core/runtime/Taskflow_Trace_Json.hpp @@ -0,0 +1,7 @@ +#pragma once +#include +#include +#include +namespace aethera::web { +[[nodiscard]] nlohmann::json taskflow_trace_json(const Taskflow_Execution_Trace& trace, std::uint64_t sequence = 0, std::uint64_t correlation_id = 0, const nlohmann::json& captured_components = {}, const nlohmann::json& captured_backend = {}); +} diff --git a/mcp/tests/Control_Path_Benchmarks.cpp b/mcp/tests/Control_Path_Benchmarks.cpp index 3bb8c7b..349091a 100644 --- a/mcp/tests/Control_Path_Benchmarks.cpp +++ b/mcp/tests/Control_Path_Benchmarks.cpp @@ -2,6 +2,7 @@ #include #include #include +#include "Gallery_Test_Sink.hpp" #include #include #include @@ -219,19 +220,15 @@ public: shared_service = Control_Service::create(Gallery_Output{ .width = configuration.width, .height = configuration.height, - .publish = [workload = shared_workload]( - std::string_view id, - std::shared_ptr frame) { + .make_sink = [workload = shared_workload](std::string_view id) -> proxy { const auto found = std::ranges::find( configuration.plot_ids, id); - if (found == configuration.plot_ids.end() || !frame || - !frame->pixels) - return web::Gallery_Frame_Publication::ignored; + if (found == configuration.plot_ids.end()) return {}; const auto index = static_cast( found - configuration.plot_ids.begin()); - workload->completed[index]->fetch_add( - 1, std::memory_order_relaxed); - return web::Gallery_Frame_Publication::completed; + return test::make_gallery_test_sink([workload, index](const web::Gallery_Frame_Slot&) { + workload->completed[index]->fetch_add(1, std::memory_order_relaxed); + }); }}); service = shared_service; workload = shared_workload; diff --git a/mcp/tests/Event_Latency_Benchmarks.cpp b/mcp/tests/Event_Latency_Benchmarks.cpp index d62ddab..d6d6218 100644 --- a/mcp/tests/Event_Latency_Benchmarks.cpp +++ b/mcp/tests/Event_Latency_Benchmarks.cpp @@ -2,6 +2,7 @@ #include #include #include +#include "Gallery_Test_Sink.hpp" #include #include #include @@ -262,26 +263,19 @@ void run_event_latency(benchmark::State& state) { service = Control_Service::create(Gallery_Output{ .width = configuration.width, .height = configuration.height, - .publish = [active_owner]( - std::string_view id, - std::shared_ptr frame) { + .make_sink = [active_owner](std::string_view id) -> proxy { auto& active = *active_owner; const auto found = std::ranges::find(active, id, &Active_Plot::id); - if (found == active.end() || !frame || !frame->pixels) - return web::Gallery_Frame_Publication::ignored; - const auto expected = found->probe->expected_correlation.load( - std::memory_order_acquire); - if (expected != 0 && - frame->pixels->correlation_id == expected) { - std::uint64_t incomplete{}; - static_cast(found->probe->completed_steady_ns. - compare_exchange_strong( - incomplete, steady_time_ns(), - std::memory_order_release, - std::memory_order_relaxed)); - } - return web::Gallery_Frame_Publication::completed; + if (found == active.end()) return {}; + const auto probe = found->probe; + return test::make_gallery_test_sink([probe](const web::Gallery_Frame_Slot& frame) { + const auto expected = probe->expected_correlation.load(std::memory_order_acquire); + if (expected != 0 && frame.request.correlation_id == expected) { + std::uint64_t incomplete{}; + static_cast(probe->completed_steady_ns.compare_exchange_strong(incomplete, steady_time_ns(), std::memory_order_release, std::memory_order_relaxed)); + } + }); }}); std::vector batch_timing(active.size()); std::string failure; diff --git a/mcp/tests/Gallery_Test_Sink.hpp b/mcp/tests/Gallery_Test_Sink.hpp new file mode 100644 index 0000000..1131345 --- /dev/null +++ b/mcp/tests/Gallery_Test_Sink.hpp @@ -0,0 +1,29 @@ +#pragma once +#include +#include +#include +#include +#include + +namespace aethera::mcp::test { + +struct Gallery_Test_Sink : Def { + struct Private; +}; + +struct Gallery_Test_Sink::Private : Prev_Private { + explicit Private(std::function value_observe) : observe(std::move(value_observe)) {} + void send(proxy& frame, frame_policy::Frame_Completion completion) { + auto* slot = frame ? proxy_cast(&*frame) : nullptr; + if (!slot) throw std::invalid_argument("Gallery test Sink received an unknown frame"); + if (observe) observe(*slot); + completion(frame); + } + std::function observe; /* Test-owned callback; benchmark scope covers asynchronous policy stop. */ +}; + +inline proxy make_gallery_test_sink(std::function observe = {}) { + return make_model_proxy(std::move(observe)); +} + +} // namespace aethera::mcp::test diff --git a/render_2D/render_2D/module/scene/Render_Scene_2D.cpp b/render_2D/render_2D/module/scene/Render_Scene_2D.cpp index 81e416a..3274aa5 100644 --- a/render_2D/render_2D/module/scene/Render_Scene_2D.cpp +++ b/render_2D/render_2D/module/scene/Render_Scene_2D.cpp @@ -76,9 +76,10 @@ Render_Scene_2D_Result Render_Scene_2D::Private::render(Frame_2D* frame, Complet auto& target = detail::Frame_2D_Access::render_target(frame); const auto built = build_frame(target, current.viewport, current.background); if (!built) return built; - const auto submitted = taskflow()->run([frame, completion = std::move(completion)](std::exception_ptr failure) mutable { + auto completed = [frame, completion = std::move(completion)](std::exception_ptr failure) mutable { completion(frame, std::move(failure)); - }); + }; + const auto submitted = frame->taskflow_trace_requested() ? taskflow()->run_observed(std::move(completed), detail::Frame_2D_Access::taskflow_observation(frame)) : taskflow()->run(std::move(completed)); if (submitted != Run_Taskflow_Result::submitted) return std::unexpected(Render_Scene_2D_Error{submitted}); return {}; } diff --git a/render_2D/render_2D/module/scene/frame/Frame_2D.cpp b/render_2D/render_2D/module/scene/frame/Frame_2D.cpp index e3466cc..d2af5c5 100644 --- a/render_2D/render_2D/module/scene/frame/Frame_2D.cpp +++ b/render_2D/render_2D/module/scene/frame/Frame_2D.cpp @@ -7,6 +7,7 @@ struct Frame_2D::Private { Frame_Identity identity{}; /* 当前逻辑帧标识。 */ Frame_Request_Source source{Frame_Request_Source::unspecified}; /* 当前逻辑帧请求来源。 */ Pixel_Format output_format{native_pixel_format}; /* 调用方选择的最终像素协议。 */ + std::optional taskflow_observation{}; /* 可选的单次 Scene Taskflow 诊断请求。 */ }; Frame_2D::Frame_2D(Frame_Identity identity, Pixel_Format output_format) : d(std::make_unique()) { @@ -22,6 +23,7 @@ void Frame_2D::begin(Frame_Identity identity, Pixel_Format output_format, Frame_ d->identity = identity; d->source = source; d->output_format = output_format; + d->taskflow_observation.reset(); } Frame_Identity Frame_2D::identity() const noexcept { @@ -44,8 +46,28 @@ Pixel_Buffer Frame_2D::output_pixels() const { return d->color.output_pixels(d->output_format); } +void Frame_2D::request_taskflow_trace() { + d->taskflow_observation.emplace("render_2d.scene"); +} + +bool Frame_2D::taskflow_trace_requested() const noexcept { + return d->taskflow_observation.has_value(); +} + +std::optional Frame_2D::take_taskflow_trace() { + if (!d->taskflow_observation) return std::nullopt; + auto result = d->taskflow_observation->take(); + d->taskflow_observation.reset(); + return result ? std::optional{std::move(*result)} : std::nullopt; +} + Blend2D_Cache& detail::Frame_2D_Access::render_target(Frame_2D* frame) { if (!frame) throw std::invalid_argument("Frame_2D render target is null"); return frame->d->color; } + +Taskflow_Observation detail::Frame_2D_Access::taskflow_observation(Frame_2D* frame) { + if (!frame || !frame->d->taskflow_observation) throw std::logic_error("Frame_2D has no Taskflow observation request"); + return *frame->d->taskflow_observation; +} } // namespace aethera::render_2d diff --git a/render_2D/render_2D/module/scene/frame/Frame_2D.hpp b/render_2D/render_2D/module/scene/frame/Frame_2D.hpp index 5f4e7a9..d6ad0f7 100644 --- a/render_2D/render_2D/module/scene/frame/Frame_2D.hpp +++ b/render_2D/render_2D/module/scene/frame/Frame_2D.hpp @@ -1,8 +1,10 @@ #pragma once #include "Blend2D_Cache.hpp" +#include #include #include #include +#include namespace aethera::render_2d { struct Frame_Identity { std::uint64_t sequence{}; /* 生产者分配的单调帧序号。 */ @@ -17,6 +19,7 @@ struct Frame_2D; namespace detail { struct Frame_2D_Access { [[nodiscard]] static Blend2D_Cache& render_target(Frame_2D* frame); + [[nodiscard]] static Taskflow_Observation taskflow_observation(Frame_2D* frame); }; } @@ -36,6 +39,9 @@ struct Frame_2D final { [[nodiscard]] Pixel_Format output_format() const noexcept; [[nodiscard]] Image_View image() const; [[nodiscard]] Pixel_Buffer output_pixels() const; + void request_taskflow_trace(); + [[nodiscard]] bool taskflow_trace_requested() const noexcept; + [[nodiscard]] std::optional take_taskflow_trace(); private: struct Private; std::unique_ptr d; /* 唯一拥有帧标识、输出协议和最终二维颜色层。 */ diff --git a/render_3D/render_3D/base/Frame_3D.cpp b/render_3D/render_3D/base/Frame_3D.cpp index 4b3b608..fad18b7 100644 --- a/render_3D/render_3D/base/Frame_3D.cpp +++ b/render_3D/render_3D/base/Frame_3D.cpp @@ -25,7 +25,7 @@ struct Frame_3D::Private { std::chrono::steady_clock::time_point created_at{}; std::array markers{}; std::array measurements{}; - std::atomic_bool taskflow_trace_requested{}; + std::optional taskflow_observation{}; /* 可选的单次 Scene Taskflow 诊断请求。 */ std::optional datoviz_observation{}; }; @@ -48,7 +48,7 @@ void Frame_3D::begin(Frame_Identity identity, Frame_3D_Output output, Pixel_Form d->created_at = std::chrono::steady_clock::now(); for (auto& marker : d->markers) marker.store(0, std::memory_order_relaxed); for (auto& measurement : d->measurements) measurement.store(0, std::memory_order_relaxed); - d->taskflow_trace_requested.store(false, std::memory_order_relaxed); + d->taskflow_observation.reset(); d->markers[static_cast(Frame_Trace_Marker::created)].store(encode_present_value(0), std::memory_order_relaxed); } @@ -100,12 +100,19 @@ void Frame_3D::record(Frame_Trace_Measurement measurement, std::uint64_t value_n d->measurements[index].compare_exchange_strong(expected, encode_present_value(value_ns), std::memory_order_release, std::memory_order_relaxed); } -void Frame_3D::request_taskflow_trace() noexcept { - d->taskflow_trace_requested.store(true, std::memory_order_release); +void Frame_3D::request_taskflow_trace() { + d->taskflow_observation.emplace("render_3d.scene"); } bool Frame_3D::taskflow_trace_requested() const noexcept { - return d->taskflow_trace_requested.load(std::memory_order_acquire); + return d->taskflow_observation.has_value(); +} + +std::optional Frame_3D::take_taskflow_trace() { + if (!d->taskflow_observation) return std::nullopt; + auto result = d->taskflow_observation->take(); + d->taskflow_observation.reset(); + return result ? std::optional{std::move(*result)} : std::nullopt; } std::optional Frame_3D::take_datoviz_observation() { @@ -133,4 +140,9 @@ void detail::Frame_3D_Access::assign_datoviz_observation(Frame_3D* frame, Datovi if (!frame) throw std::invalid_argument("Frame_3D diagnostic target is null"); frame->d->datoviz_observation.emplace(std::move(observation)); } + +Taskflow_Observation detail::Frame_3D_Access::taskflow_observation(Frame_3D* frame) { + if (!frame || !frame->d->taskflow_observation) throw std::logic_error("Frame_3D has no Taskflow observation request"); + return *frame->d->taskflow_observation; +} } // namespace aethera::render_3d diff --git a/render_3D/render_3D/base/Frame_3D.hpp b/render_3D/render_3D/base/Frame_3D.hpp index 51b13fd..1f5a4ed 100644 --- a/render_3D/render_3D/base/Frame_3D.hpp +++ b/render_3D/render_3D/base/Frame_3D.hpp @@ -1,6 +1,7 @@ #pragma once #include "Datoviz_Frame_Observation.hpp" #include "Types.hpp" +#include #include #include #include @@ -124,6 +125,7 @@ struct Frame_3D_Access { static void assign_pixels(std::span frames, Extent extent, std::vector pixels, Frame_Identity rendered_identity); static void share_pixels(Frame_3D* source, Frame_3D* target); static void assign_datoviz_observation(Frame_3D* frame, Datoviz_Frame_Observation observation); + [[nodiscard]] static Taskflow_Observation taskflow_observation(Frame_3D* frame); }; } // namespace detail @@ -147,8 +149,9 @@ struct Frame_3D final { [[nodiscard]] Pixel_Format output_format() const noexcept; void mark(Frame_Trace_Marker marker) noexcept; void record(Frame_Trace_Measurement measurement, std::uint64_t value_ns) noexcept; - void request_taskflow_trace() noexcept; + void request_taskflow_trace(); [[nodiscard]] bool taskflow_trace_requested() const noexcept; + [[nodiscard]] std::optional take_taskflow_trace(); [[nodiscard]] std::optional take_datoviz_observation(); private: struct Private; diff --git a/render_3D/render_3D/scene/Render_Scene_3D_Model.ipp b/render_3D/render_3D/scene/Render_Scene_3D_Model.ipp index 8cb4b83..e03599e 100644 --- a/render_3D/render_3D/scene/Render_Scene_3D_Model.ipp +++ b/render_3D/render_3D/scene/Render_Scene_3D_Model.ipp @@ -375,12 +375,13 @@ inline Render_Scene_3D::Render_Result Render_Scene_3D::Private::render(Frame_3D* execution->completion = std::move(completion); auto graph = build_frame_graph(execution, parameters); frame->mark(Frame_Trace_Marker::scene_render_requested); - const auto submitted = graph->run([this, execution](std::exception_ptr failure) { + auto completed = [this, execution](std::exception_ptr failure) { execution->cpu_failure = std::move(failure); if (execution->cpu_failure && !execution->backend_started.load(std::memory_order_acquire)) execution->backend_finished.store(true, std::memory_order_release); execution->cpu_finished.store(true, std::memory_order_release); try_complete(execution); - }); + }; + const auto submitted = frame->taskflow_trace_requested() ? graph->run_observed(std::move(completed), detail::Frame_3D_Access::taskflow_observation(frame)) : graph->run(std::move(completed)); if (submitted != Run_Taskflow_Result::submitted) return Render_Result::backend_unavailable; return Render_Result::submitted; } diff --git a/render_3D/render_3D/scene/detail/Datoviz_Scene_Common.ipp b/render_3D/render_3D/scene/detail/Datoviz_Scene_Common.ipp index 01248e1..79b654d 100644 --- a/render_3D/render_3D/scene/detail/Datoviz_Scene_Common.ipp +++ b/render_3D/render_3D/scene/detail/Datoviz_Scene_Common.ipp @@ -287,6 +287,10 @@ struct Datoviz_Render_Context final : std::enable_shared_from_thisadd("drain", [this] { drain_submissions(); }); + if (!drain) throw std::logic_error("failed to create Datoviz submission drain task"); DvzGpuCtxConfig configuration = dvz_gpu_ctx_config(); dvz_gpu_ctx_config_validation(&configuration, validation_enabled); dvz_gpu_ctx_config_gpu(&configuration, gpu_index); @@ -297,21 +301,29 @@ struct Datoviz_Render_Context final : std::enable_shared_from_thisadd("drain", [lifetime] { - lifetime->drain_submissions(); - }); - if (!task) throw std::logic_error("failed to create Datoviz submission drain task"); - const auto submitted = submission_taskflow_->run([lifetime = std::move(lifetime)](std::exception_ptr failure) { + auto lifetime = shared_from_this(); + const auto observed = submission_generation_.load(std::memory_order_acquire); + const auto submitted = submission_taskflow_->run([lifetime = std::move(lifetime), observed](std::exception_ptr failure) { + std::exception_ptr continuation_failure; + lifetime->submission_drain_active_.store(false, std::memory_order_release); + if (lifetime->submission_generation_.load(std::memory_order_acquire) != observed) { + try { + lifetime->arm_submission_drain(); + } + catch (...) { + continuation_failure = std::current_exception(); + } + } + if (!failure) failure = std::move(continuation_failure); if (failure) Failure_Policy::handle_unknown_failure(std::move(failure)); }); - if (submitted != Run_Taskflow_Result::submitted) throw std::runtime_error("Datoviz submission task runtime is unavailable"); + if (submitted == Run_Taskflow_Result::submitted) return; + submission_drain_active_.store(false, std::memory_order_release); + throw std::runtime_error("Datoviz submission task runtime is unavailable"); } void drain_submissions() noexcept { - const auto observed = submission_generation_.load(std::memory_order_acquire); - std::function < void() > command; + std::function command; while (submissions_.try_dequeue(command)) { try { command(); @@ -319,15 +331,13 @@ struct Datoviz_Render_Context final : std::enable_shared_from_this> submissions_{}; std::atomic_uint64_t submission_generation_{}; std::atomic_bool submission_drain_active_{}; - proxy submission_taskflow_{}; /* 拥有当前唯一异步提交队列任务图。 */ + proxy submission_taskflow_{}; /* Render Context 生命周期内唯一且不替换；所有 corun_until 调用共享稳定的 facade 所有权。 */ }; /* diff --git a/web_server/src/Graph_WebSocket.cpp b/web_server/src/Graph_WebSocket.cpp index 8fb7ffc..e1ccd5b 100644 --- a/web_server/src/Graph_WebSocket.cpp +++ b/web_server/src/Graph_WebSocket.cpp @@ -59,9 +59,7 @@ void Graph_WebSocket::receive(std::string_view message) { return; } if (kind == "manual_render") { - d->control->request_frame(d->plot_id, Frame_Request{ - .issued_at = std::chrono::steady_clock::now(), - .source = Frame_Request_Source::immediate}); + d->control->request_frame(d->plot_id, Gallery_Frame_Request{.issued_at = std::chrono::steady_clock::now()}); return; } if (kind != "input") return; diff --git a/web_server/src/Web_Server.cpp b/web_server/src/Web_Server.cpp index fda04df..3b68002 100644 --- a/web_server/src/Web_Server.cpp +++ b/web_server/src/Web_Server.cpp @@ -7,7 +7,6 @@ #include #include #include -#include #include #include #include @@ -40,10 +39,11 @@ bool taskflow_graph_contains_gallery_media(const nlohmann::json& graph) { for (const auto& node : graph["nodes"]) { if (!node.is_object()) continue; const auto name = node.value("name", std::string{}); - if (name == "gallery.sample.capture" || + if (name == "gallery.atlas.compose" || name == "FFmpeg.H264.encode" || name == "gallery.h264.websocket.publish" || - name == "gallery.sample.complete") + name == "frame.policy.sink.complete" || + name == "gallery.media.complete") return true; } return false; @@ -98,8 +98,6 @@ nlohmann::json merge_gallery_media_trace(nlohmann::json plot_trace, } } int run_web_server(std::uint16_t port, const std::filesystem::path& asset_root) { - auto control_slot = std::make_shared>>(); std::vector gallery_2d; std::vector gallery_3d; gallery_2d.reserve(8); @@ -109,13 +107,7 @@ int run_web_server(std::uint16_t port, const std::filesystem::path& asset_root) (definition.dimension == Plot_Dimension::three_d ? gallery_3d : gallery_2d) - .push_back({id, [control_slot, id]( - Frame_Publication_Feedback feedback) { - if (const auto control = control_slot->load( - std::memory_order_acquire)) - control->submit_publication_feedback( - id, std::move(feedback)); - }}); + .push_back({id}); } const auto entry_order = [](const auto& left, const auto& right) { return left.id < right.id; @@ -161,18 +153,14 @@ int run_web_server(std::uint16_t port, const std::filesystem::path& asset_root) add_media_groups("2d", std::move(gallery_2d)); add_media_groups("3d", std::move(gallery_3d)); auto control = mcp::Control_Service::create(mcp::Gallery_Output{ - .width = 720, - .height = 420, - .publish = [gallery_streams, plot_media_group]( - std::string_view id, - std::shared_ptr frame) { + .width = 720, + .height = 420, + .make_sink = [gallery_streams, plot_media_group](std::string_view id) -> proxy { const auto group = plot_media_group->find(std::string{id}); - if (group == plot_media_group->end()) return Gallery_Frame_Publication::ignored; + if (group == plot_media_group->end()) return {}; const auto stream = gallery_streams->find(group->second); - if (stream == gallery_streams->end()) return Gallery_Frame_Publication::ignored; - return stream->second->accept_frame(id, std::move(frame)); + return stream == gallery_streams->end() ? proxy{} : stream->second->frame_sink(id); }}); - control_slot->store(control, std::memory_order_release); auto websocket = std::make_shared(control); auto gallery_websocket = std::make_shared( @@ -227,12 +215,6 @@ int run_web_server(std::uint16_t port, const std::filesystem::path& asset_root) } callback(json_response(control->plot_diagnostics(plot_id))); }, {drogon::Get, drogon::Delete}); - app.registerHandler("/taskflow/diagnostics", [control]( - const drogon::HttpRequestPtr&, - std::function&& callback) { - const auto output = control->call_tool("aethera_task_runtime", nlohmann::json::object()); - callback(json_response(output.content)); - }, {drogon::Get}); app.registerHandler("/plot/{1}/taskflow", [control, plot_media_group, gallery_streams]( const drogon::HttpRequestPtr& request, diff --git a/web_server/src/main.cpp b/web_server/src/main.cpp index 1cbd6f3..f370070 100644 --- a/web_server/src/main.cpp +++ b/web_server/src/main.cpp @@ -1,11 +1,9 @@ #include "Web_Server.hpp" -#include +#include #include -#include #include #include #include -#include #include #include namespace { @@ -18,37 +16,11 @@ std::uint16_t parse_port(int argc, char** argv) { const auto [end, error] = std::from_chars(text.data(), text.data() + text.size(), value); return error == std::errc{} && end == text.data() + text.size() && value > 0 && value <= 65535 ? static_cast(value) : 0; } -aethera::Task_Runtime_Configuration task_runtime_configuration() { - aethera::Task_Runtime_Configuration result{}; - if (const auto* value = std::getenv("AETHERA_TASK_WORKER_LIMIT_MS")) { - const std::string_view text(value); - std::uint64_t milliseconds{}; - const auto [end, error] = std::from_chars( - text.data(), text.data() + text.size(), milliseconds); - if (error != std::errc{} || end != text.data() + text.size() || - milliseconds == 0) - throw std::invalid_argument( - "AETHERA_TASK_WORKER_LIMIT_MS must be a positive integer"); - result.worker_occupation_limit = - std::chrono::milliseconds(milliseconds); - } - if (const auto* value = - std::getenv("AETHERA_TASK_WORKER_OVERRUN_ACTION")) { - const auto action = magic_enum::enum_cast( - std::string_view(value)); - if (!action) - throw std::invalid_argument( - "AETHERA_TASK_WORKER_OVERRUN_ACTION must be warning or fast_fail"); - result.worker_overrun_action = *action; - } - result.worker_overrun_action = aethera::Task_Overrun_Action::warning; - return result; -} } int main(int argc, char** argv) { const auto port = parse_port(argc, argv); if (port == 0) return 2; - aethera::initialize_runtime(task_runtime_configuration()); + if (aethera::initialize_task_runtime() != aethera::Initialize_Task_Runtime_Result::initialized) return 3; const auto executable = std::filesystem::absolute(argv[0]); std::cout << "Aethera Gallery http://127.0.0.1:" << port << std::endl; return aethera::web::run_web_server(port, executable.parent_path() / "webapp_gallery"); diff --git a/web_server/src/media/Gallery_Video_Stream.cpp b/web_server/src/media/Gallery_Video_Stream.cpp index cd596d9..924f8a6 100644 --- a/web_server/src/media/Gallery_Video_Stream.cpp +++ b/web_server/src/media/Gallery_Video_Stream.cpp @@ -1,7 +1,6 @@ #include "Gallery_Video_Stream.hpp" #include -#include -#include +#include #include #include #include @@ -19,14 +18,10 @@ constexpr std::uint32_t tile_height{420}; constexpr std::uint32_t atlas_columns{4}; constexpr double encoder_nominal_frame_rate{30.0}; constexpr auto metric_interval{std::chrono::seconds(1)}; -nlohmann::json statistic_json(const Statistic_State& value) { +nlohmann::json statistic_json(const Statistics_Summary& value) { return { - {"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} + {"count", value.sample_count}, {"average", value.average}, + {"variability", value.standard_deviation}, {"p95", value.p95} }; } std::string exception_description(const std::exception_ptr& failure) { @@ -73,8 +68,7 @@ bool Gallery_Video_Stream::Private::mark_taskflow_trace() { void Gallery_Video_Stream::Private::restore_taskflow_trace() noexcept { taskflow_trace_remaining.fetch_add(1, std::memory_order_release); } -void Gallery_Video_Stream::Private::store_taskflow_trace( - const Taskflow_Frame_Trace& value) { +void Gallery_Video_Stream::Private::store_taskflow_trace(const Taskflow_Execution_Trace& value) { const auto trace = std::make_shared( taskflow_trace_json(value)); auto state = taskflow_trace_control.load(std::memory_order_acquire); @@ -166,32 +160,22 @@ void Gallery_Video_Stream::Private::release_pending_frames() noexcept { Pending_Plot_Frame pending; while (pending_frames.try_dequeue(pending)) { try { - if (pending.slot < sources.size() && pending.frame && - pending.frame->pixels) - if (sources[pending.slot].entry.publish_completed) - sources[pending.slot].entry.publish_completed({ - .completed_at = std::chrono::steady_clock::now(), - .sequence = pending.frame->pixels->sequence, - .succeeded = false}); + if (pending.policy_frame && pending.completion) pending.completion(*pending.policy_frame); } - catch (...) {} + catch (...) { std::terminate(); } pending = {}; } } void Gallery_Video_Stream::Private::fail( std::exception_ptr failure) noexcept { if (failed.exchange(true, std::memory_order_acq_rel)) return; - if (active_frame && !active_frame->publication_feedback_submitted && - active_frame->source_slot < sources.size()) { + if (active_frame && active_frame->policy_frame && active_frame->completion) { try { - if (sources[active_frame->source_slot].entry.publish_completed) - sources[active_frame->source_slot].entry.publish_completed({ - .completed_at = std::chrono::steady_clock::now(), - .sequence = active_frame->plot_frame_sequence, - .succeeded = false}); - active_frame->publication_feedback_submitted = true; + auto completion = std::move(active_frame->completion); + completion(*active_frame->policy_frame); + active_frame->policy_frame = nullptr; } - catch (...) {} + catch (...) { std::terminate(); } } release_pending_frames(); try { @@ -206,27 +190,58 @@ void Gallery_Video_Stream::Private::fail( } catch (...) {} } -Gallery_Frame_Publication Gallery_Video_Stream::Private::accept_frame( - std::size_t slot, std::shared_ptr frame) { - if (stopping.load(std::memory_order_acquire) || - failed.load(std::memory_order_acquire) || !frame || !frame->pixels || - !frame->pixels->pixels || !consumer_accepts()) - return Gallery_Frame_Publication::ignored; +void Gallery_Video_Stream::Private::accept_frame(std::size_t slot, proxy& frame, frame_policy::Frame_Completion completion, std::weak_ptr lifetime) { + if (!completion) throw std::invalid_argument("Gallery media Sink completion is empty"); + auto* frame_slot = frame ? proxy_cast(&*frame) : nullptr; + if (!frame_slot || slot >= sources.size()) throw std::invalid_argument("Gallery media Sink received an unknown frame"); + if (stopping.load(std::memory_order_acquire) || failed.load(std::memory_order_acquire) || !consumer_accepts()) { + completion(frame); + return; + } + std::shared_ptr> storage; + Gallery_Pixel_Layout layout{Gallery_Pixel_Layout::rgba8}; + std::uint64_t rendered_sequence = frame_slot->request.sequence; + std::uint64_t rendered_correlation_id = frame_slot->request.correlation_id; + std::uint32_t width{}; + std::uint32_t height{}; + if (auto* native_2d = std::get_if>(&frame_slot->native)) { + layout = Gallery_Pixel_Layout::bgra8; + auto pixels = (*native_2d)->output_pixels(); + width = static_cast(pixels.width); + height = static_cast(pixels.height); + storage = std::make_shared>(std::move(pixels.bytes)); + } + else { + auto& native_3d = std::get>(frame_slot->native); + rendered_sequence = native_3d->rendered_identity().sequence; + const auto extent = native_3d->extent(); + width = extent.width; + height = extent.height; + storage = native_3d->share_pixels(); + } + const auto timestamp = std::chrono::duration_cast(std::chrono::duration(frame_slot->request.time_milliseconds)); + auto pixels = std::make_shared(Gallery_Pixel_Frame{std::move(storage), layout, timestamp, frame_slot->request.sequence, frame_slot->request.correlation_id, rendered_sequence, rendered_correlation_id, width, height}); + if (!pixels->pixels) { + completion(frame); + return; + } Pending_Plot_Frame pending{ - slot, std::move(frame), + slot, std::move(pixels), std::addressof(frame), std::move(completion), std::chrono::steady_clock::now(), std::chrono::duration_cast( std::chrono::system_clock::now().time_since_epoch())}; - if (!pending_frames.enqueue(std::move(pending))) throw std::bad_alloc{}; + if (!pending_frames.enqueue(pending)) { + pending.completion(*pending.policy_frame); + throw std::bad_alloc{}; + } if (stopping.load(std::memory_order_acquire) || failed.load(std::memory_order_acquire)) { release_pending_frames(); - return Gallery_Frame_Publication::ignored; + return; } received_frame_count.fetch_add(1, std::memory_order_relaxed); media_work_generation.fetch_add(1, std::memory_order_release); - arm_media_task(object->weak_from_this()); - return Gallery_Frame_Publication::asynchronous; + arm_media_task(std::move(lifetime)); } void Gallery_Video_Stream::Private::update_metrics( const Active_Media_Frame& frame, @@ -293,10 +308,10 @@ void Gallery_Video_Stream::Private::update_metrics( metric_completion_starts[slot] = progress.completion_count; metric_rendered_starts[slot] = progress.rendered_frame_count; } - const auto& compose = compose_ms.state(); - const auto& queue_delay = queue_delay_ms.state(); - const auto& encode = encode_ms.state(); - const auto& publish_time = publish_ms.state(); + const auto compose = compose_ms.summary(); + const auto queue_delay = queue_delay_ms.summary(); + const auto encode = encode_ms.summary(); + const auto publish_time = publish_ms.summary(); auto output = nlohmann::json{ {"kind", "gallery_metrics"}, {"protocol", "aethera.gallery.video"}, @@ -350,8 +365,8 @@ void Gallery_Video_Stream::Private::update_metrics( metric_received_start = received_total; metric_processed_start = processed_total; metric_encoded_start = encoded_total; - object->template update_state<&State::diagnostics>(std::move(output)); - object->template publish_state(); + concurrent().internal.use()->diagnostics = std::move(output); + concurrent().internal.advance(); } void Gallery_Video_Stream::Private::begin_media_frame() { active_frame.reset(); @@ -363,15 +378,10 @@ void Gallery_Video_Stream::Private::begin_media_frame() { if (!pending_frames.try_dequeue(pending)) return; ++processed_frame_count; const auto started = std::chrono::steady_clock::now(); - const auto accepted = atlas->accept_frame( - pending.slot, pending.frame->pixels); + const auto accepted = atlas->accept_frame(pending.slot, pending.pixels); if (accepted == detail::Gallery_Frame_Atlas::Accept_Frame_Result::invalid_frame) { - if (sources[pending.slot].entry.publish_completed) - sources[pending.slot].entry.publish_completed({ - .completed_at = std::chrono::steady_clock::now(), - .sequence = pending.frame->pixels->sequence, - .succeeded = false}); + pending.completion(*pending.policy_frame); throw std::logic_error( "Plot published an invalid gallery pixel frame"); } @@ -386,11 +396,9 @@ void Gallery_Video_Stream::Private::begin_media_frame() { pending.source_time_unix, std::chrono::duration( started - pending.enqueued_at).count(), - pending.slot, pending.frame->pixels->sequence, false, false}; - static_cast(compose_ms.submit( - std::chrono::duration( - std::chrono::steady_clock::now() - started).count())); - static_cast(queue_delay_ms.submit(active_frame->queue_delay_ms)); + pending.slot, pending.pixels->sequence, false, pending.policy_frame, std::move(pending.completion)}; + compose_ms.add(std::chrono::duration(std::chrono::steady_clock::now() - started).count()); + queue_delay_ms.add(active_frame->queue_delay_ms); } void Gallery_Video_Stream::Private::encode_media_frame() { if (!active_frame) return; @@ -412,36 +420,24 @@ void Gallery_Video_Stream::Private::encode_media_frame() { active_frame->source_time_unix)) active_video = std::make_shared( std::move(*encoded)); - static_cast(encode_ms.submit( - std::chrono::duration( - std::chrono::steady_clock::now() - started).count())); + encode_ms.add(std::chrono::duration(std::chrono::steady_clock::now() - started).count()); } void Gallery_Video_Stream::Private::publish_video_frame() { if (!active_video) return; ++encoded_frame_count; const auto started = std::chrono::steady_clock::now(); active_frame->publication_succeeded = publish(active_video, {}); - static_cast(publish_ms.submit( - std::chrono::duration( - std::chrono::steady_clock::now() - started).count())); + publish_ms.add(std::chrono::duration(std::chrono::steady_clock::now() - started).count()); } -void Gallery_Video_Stream::Private::submit_publication_feedback() { - if (!active_frame || active_frame->publication_feedback_submitted) return; - if (active_frame->source_slot >= sources.size()) - throw std::logic_error( - "gallery publication feedback source is invalid"); - if (sources[active_frame->source_slot].entry.publish_completed) - sources[active_frame->source_slot].entry.publish_completed({ - .completed_at = std::chrono::steady_clock::now(), - .sequence = active_frame->plot_frame_sequence, - .succeeded = active_frame->publication_succeeded}); - active_frame->publication_feedback_submitted = true; +void Gallery_Video_Stream::Private::complete_policy_frame() { + if (!active_frame || !active_frame->policy_frame || !active_frame->completion) return; + auto completion = std::move(active_frame->completion); + completion(*active_frame->policy_frame); + active_frame->policy_frame = nullptr; } void Gallery_Video_Stream::Private::complete_media_frame() { if (!active_frame) return; - if (!active_frame->publication_feedback_submitted) - throw std::logic_error( - "gallery media completion preceded policy feedback task"); + if (active_frame->policy_frame || active_frame->completion) throw std::logic_error("gallery media completion preceded Frame Policy Sink completion"); const auto encoded_bytes = active_video ? active_video->annex_b.size() : 0U; const auto encoder_backend = active_video ? std::optional{active_video->backend} @@ -452,32 +448,24 @@ void Gallery_Video_Stream::Private::complete_media_frame() { } std::shared_ptr Gallery_Video_Stream::create( std::vector plots) { - auto built = Gallery_Video_Stream::Builder{}.build(); - if (!built) - throw std::logic_error( - "gallery video stream definition validation failed"); - auto result = std::shared_ptr(std::move(*built)); - static_cast(*result->d).initialize(std::move(plots)); + Gallery_Video_Stream::Builder builder; + auto result = std::shared_ptr{builder.build().release()}; + aethera::detail::model_private(*result).initialize(std::move(plots)); result->initialize_pipeline(); return result; } -Gallery_Frame_Publication Gallery_Video_Stream::accept_frame( - std::string_view plot_id, std::shared_ptr frame) { - auto& data = static_cast(*d); +proxy Gallery_Video_Stream::frame_sink(std::string_view plot_id) { + auto& data = aethera::detail::model_private(*this); const auto found = std::ranges::find_if( data.sources, [plot_id](const Private::Source& source) { return source.entry.id == plot_id; }); - if (found == data.sources.end()) return Gallery_Frame_Publication::ignored; - try { - return data.accept_frame( - static_cast(found - data.sources.begin()), - std::move(frame)); - } - catch (...) { - data.fail(std::current_exception()); - return Gallery_Frame_Publication::ignored; - } + if (found == data.sources.end()) return {}; + return make_model_proxy(shared_from_this(), static_cast(found - data.sources.begin())); +} + +void Gallery_Video_Stream::Sink::Private::send(proxy& frame, frame_policy::Frame_Completion completion) { + aethera::detail::model_private(*stream).accept_frame(slot, frame, std::move(completion), stream); } Gallery_Video_Stream::Gallery_Video_Stream() = default; Gallery_Video_Stream::~Gallery_Video_Stream() { @@ -493,62 +481,46 @@ void Gallery_Video_Stream::Private::arm_media_task( idle, true, std::memory_order_acq_rel, std::memory_order_acquire)) return; - const auto graph = media_graph; bool trace_reserved = mark_taskflow_trace(); - auto trace_frame = trace_reserved - ? std::make_shared(Frame_Identity{ - received_frame_count.load(std::memory_order_acquire), 0}) - : std::shared_ptr{}; - bool trace_started{}; - if (trace_frame) { - trace_frame->request_taskflow_trace(); - trace_started = aethera::detail::begin_taskflow_trace(*trace_frame); - if (!trace_started) { - restore_taskflow_trace(); - trace_reserved = false; - trace_frame.reset(); - } - } + std::optional observation; + if (trace_reserved) observation.emplace("gallery.media"); try { - auto completion = [lifetime, graph, trace_frame, trace_started] { - static_cast(graph); - if (trace_started) aethera::detail::finish_taskflow_trace(*trace_frame); + auto completion = [lifetime, observation, trace_reserved](std::exception_ptr failure) mutable { const auto completed = lifetime.lock(); if (!completed) return; - auto& data = static_cast(*completed->d); - if (trace_started) - data.store_taskflow_trace( - trace_frame->take_taskflow_trace()); + auto& data = aethera::detail::model_private(*completed); + if (failure) data.fail(std::move(failure)); + if (trace_reserved) { + auto trace = observation->take(); + if (trace) data.store_taskflow_trace(*trace); + else data.restore_taskflow_trace(); + } data.media_task_scheduled.store(false, - std::memory_order_release); + std::memory_order_release); if (data.received_frame_count.load(std::memory_order_acquire) > data.processed_frame_count.load(std::memory_order_acquire)) data.arm_media_task(lifetime); }; - if (trace_frame) - aethera::detail::run_taskflow( - *graph, *trace_frame, "gallery.media", - std::move(completion)); - else aethera::detail::run_taskflow(*graph, std::move(completion)); + const auto submitted = observation ? media_graph->run_observed(std::move(completion), *observation) : media_graph->run(std::move(completion)); + if (submitted != Run_Taskflow_Result::submitted) throw std::runtime_error("gallery media Taskflow submission failed"); } catch (...) { - if (trace_started) aethera::detail::finish_taskflow_trace(*trace_frame); if (trace_reserved) restore_taskflow_trace(); media_task_scheduled.store(false, std::memory_order_release); fail(std::current_exception()); } } void Gallery_Video_Stream::initialize_pipeline() { - auto& data = static_cast(*d); + auto& data = aethera::detail::model_private(*this); const auto weak = weak_from_this(); if (data.sources.empty()) throw std::invalid_argument("gallery video stream requires a Plot"); /* 每个 Plot 独立帧策略发布的物理像素帧按到达顺序进入本 DAG。 * Gallery 不再拥有采样时钟、准入或丢帧规则；持久 DAG 每次消费一个 * 完成帧，串行保护图集像素和 FFmpeg 上下文，完成后异步续跑下一帧。 */ - auto media = std::make_shared("gallery.video.frame"); + auto media = task_flow::make_task_graph("gallery.video.frame"); auto compose = media->add("gallery.atlas.compose", [weak] { if (const auto owner = weak.lock()) { - auto& owner_data = static_cast(*owner->d); + auto& owner_data = aethera::detail::model_private(*owner); try { owner_data.begin_media_frame(); } @@ -557,12 +529,13 @@ void Gallery_Video_Stream::initialize_pipeline() { } } }); - compose.describe("owner", "gallery") - .describe("stage", "apply one policy-approved Plot frame") - .describe("admission", "owned by source Plot frame policy"); + if (!compose) throw std::logic_error("gallery media compose task creation failed"); + (*compose)->describe_node("owner", "gallery"); + (*compose)->describe_node("stage", "apply one policy-approved Plot frame"); + (*compose)->describe_node("admission", "owned by source Plot frame policy"); auto encode = media->add("FFmpeg.H264.encode", [weak] { if (const auto owner = weak.lock()) { - auto& owner_data = static_cast(*owner->d); + auto& owner_data = aethera::detail::model_private(*owner); try { owner_data.encode_media_frame(); } @@ -571,14 +544,15 @@ void Gallery_Video_Stream::initialize_pipeline() { } } }); - encode.describe("owner", "gallery") - .describe("stage", "encode every composed Plot completion") - .describe("backend", "FFmpeg") - .describe("codec", "H.264") - .describe("execution", "Taskflow worker"); + if (!encode) throw std::logic_error("gallery media encode task creation failed"); + (*encode)->describe_node("owner", "gallery"); + (*encode)->describe_node("stage", "encode every composed Plot completion"); + (*encode)->describe_node("backend", "FFmpeg"); + (*encode)->describe_node("codec", "H.264"); + (*encode)->describe_node("execution", "Taskflow worker"); auto publish_video = media->add("gallery.h264.websocket.publish", [weak] { if (const auto owner = weak.lock()) { - auto& owner_data = static_cast(*owner->d); + auto& owner_data = aethera::detail::model_private(*owner); try { owner_data.publish_video_frame(); } @@ -587,27 +561,29 @@ void Gallery_Video_Stream::initialize_pipeline() { } } }); - publish_video.describe("owner", "gallery") - .describe("transport", "Drogon WebSocket") - .describe("stage", "H.264 access unit enqueue"); - auto publication_feedback = media->add( - "frame.policy.publish.feedback", [weak] { + if (!publish_video) throw std::logic_error("gallery media publish task creation failed"); + (*publish_video)->describe_node("owner", "gallery"); + (*publish_video)->describe_node("transport", "Drogon WebSocket"); + (*publish_video)->describe_node("stage", "H.264 access unit enqueue"); + auto policy_completion = media->add( + "frame.policy.sink.complete", [weak] { if (const auto owner = weak.lock()) { - auto& owner_data = static_cast(*owner->d); + auto& owner_data = aethera::detail::model_private(*owner); try { - owner_data.submit_publication_feedback(); + owner_data.complete_policy_frame(); } catch (...) { owner_data.fail(std::current_exception()); } } }); - publication_feedback.describe("owner", "frame_policy") - .describe("stage", "media publication feedback") - .describe("execution", "Taskflow worker"); + if (!policy_completion) throw std::logic_error("gallery media Sink completion task creation failed"); + (*policy_completion)->describe_node("owner", "frame_policy"); + (*policy_completion)->describe_node("stage", "media Sink completion"); + (*policy_completion)->describe_node("execution", "Taskflow worker"); auto complete = media->add("gallery.media.complete", [weak] { if (const auto owner = weak.lock()) { - auto& owner_data = static_cast(*owner->d); + auto& owner_data = aethera::detail::model_private(*owner); try { owner_data.complete_media_frame(); } @@ -616,19 +592,17 @@ void Gallery_Video_Stream::initialize_pipeline() { } } }); - complete.describe("owner", "gallery") - .describe("stage", "metrics and Gallery stage ownership release") - .describe("policy_lifecycle", "already reported by preceding task"); - compose.precede(encode); - encode.precede(publish_video); - publish_video.precede(publication_feedback); - publication_feedback.precede(complete); - data.media_graph = media; + if (!complete) throw std::logic_error("gallery media completion task creation failed"); + (*complete)->describe_node("owner", "gallery"); + (*complete)->describe_node("stage", "metrics and Gallery stage ownership release"); + (*complete)->describe_node("policy_lifecycle", "already reported by preceding task"); + if (media->precede(*compose, *encode) != Precede_Task_Graph_Result::preceded || media->precede(*encode, *publish_video) != Precede_Task_Graph_Result::preceded || media->precede(*publish_video, *policy_completion) != Precede_Task_Graph_Result::preceded || media->precede(*policy_completion, *complete) != Precede_Task_Graph_Result::preceded) throw std::logic_error("gallery media task dependency creation failed"); + data.media_graph = std::move(media); } Gallery_Video_Stream::Stream_Id Gallery_Video_Stream::subscribe( std::string connection, Stream_Handler handler, Transport_Readiness readiness, Transport_Diagnostics diagnostics) { - auto& data = static_cast(*d); + auto& data = aethera::detail::model_private(*this); if (data.stopping.load(std::memory_order_acquire)) throw std::logic_error("gallery video stream is shutting down"); if (data.failed.load(std::memory_order_acquire)) throw std::logic_error("gallery video stream is unavailable"); if (!handler) throw std::invalid_argument("gallery video subscription requires a handler"); @@ -685,23 +659,23 @@ Gallery_Video_Stream::Stream_Id Gallery_Video_Stream::subscribe( return id; } void Gallery_Video_Stream::unsubscribe(Stream_Id stream) { - auto& data = static_cast(*d); + auto& data = aethera::detail::model_private(*this); static_cast(data.remove_consumer(stream)); } void Gallery_Video_Stream::request_video_key_frame() { - auto& data = static_cast(*d); + auto& data = aethera::detail::model_private(*this); if (!data.stopping.load(std::memory_order_acquire) && !data.failed.load(std::memory_order_acquire)) data.ffmpeg_transport.request_key_frame(); } void Gallery_Video_Stream::shutdown() noexcept { - auto& data = static_cast(*d); + auto& data = aethera::detail::model_private(*this); if (data.stopping.exchange(true, std::memory_order_acq_rel)) return; for (auto& consumer : data.consumers) consumer.store({}, std::memory_order_release); data.release_pending_frames(); } std::string Gallery_Video_Stream::layout_description() const { - const auto& data = static_cast(*d); + const auto& data = aethera::detail::model_private(*this); const auto description = data.atlas->describe(); nlohmann::json plots = nlohmann::json::object(); for (const auto& source : description.sources) plots[source.id] = {{"column", source.column}, {"row", source.row}}; @@ -737,10 +711,10 @@ nlohmann::json Gallery_Video_Stream::diagnostics() const { {"version", 5}, {"sources", nlohmann::json::object()} }; - this->template access_state([&output](const State& state) { + this->template get([&output](const State& state) { if (!state.diagnostics.empty()) output = state.diagnostics; }); - const auto& data = static_cast(*d); + const auto& data = aethera::detail::model_private(*this); output["received_frame_count"] = data.received_frame_count.load( std::memory_order_relaxed); output["processed_frame_count"] = data.processed_frame_count.load( @@ -763,7 +737,7 @@ void Gallery_Video_Stream::request_taskflow_trace(std::size_t frame_count) { frame_count > Private::maximum_taskflow_trace_frames) throw std::invalid_argument( "Gallery Taskflow trace frame_count must be between 1 and 120"); - auto& data = static_cast(*d); + auto& data = aethera::detail::model_private(*this); auto control = data.taskflow_trace_control.load(std::memory_order_acquire); for (;;) { const auto requested = static_cast(control >> 32U); @@ -782,6 +756,6 @@ void Gallery_Video_Stream::request_taskflow_trace(std::size_t frame_count) { std::memory_order_release); } nlohmann::json Gallery_Video_Stream::taskflow_trace() const { - return static_cast(*d).taskflow_trace_response(); + return aethera::detail::model_private(*this).taskflow_trace_response(); } } diff --git a/web_server/src/media/Gallery_Video_Stream.hpp b/web_server/src/media/Gallery_Video_Stream.hpp index 0a29abf..761a94d 100644 --- a/web_server/src/media/Gallery_Video_Stream.hpp +++ b/web_server/src/media/Gallery_Video_Stream.hpp @@ -1,7 +1,9 @@ #pragma once #include #include "Encoded_Video_Frame.hpp" -#include +#include +#include +#include #include #include #include @@ -13,18 +15,19 @@ struct Gallery_Stream_Frame { std::shared_ptr video{}; /* FFmpeg 订阅者共享的 H.264 access unit。 */ std::string notification{}; /* 仅承载终止错误等低频控制通知。 */ }; -struct Gallery_Video_Stream : Def, +struct Gallery_Video_Stream : Def>, std::enable_shared_from_this { struct Plot_Entry { - std::string id; - std::function publish_completed; + std::string id; }; - struct Prop : Prev_Prop {}; - struct State : Prev_State { + struct State : Prev { nlohmann::json diagnostics{}; /* 最近一秒完成的媒体统计；空对象表示尚未形成窗口。 */ bool operator==(const State&) const = default; }; struct Private; + struct Sink : Def { + struct Private; + }; using Stream_Id = std::uint64_t; using Stream_Handler = std::function; using Transport_Readiness = std::function; @@ -33,14 +36,9 @@ struct Gallery_Video_Stream : Def, ~Gallery_Video_Stream(); Gallery_Video_Stream(const Gallery_Video_Stream&) = delete; Gallery_Video_Stream& operator=(const Gallery_Video_Stream&) = delete; - [[nodiscard]] static std::shared_ptr create( - std::vector plots); - [[nodiscard]] Gallery_Frame_Publication accept_frame( - std::string_view plot_id, std::shared_ptr frame); - [[nodiscard]] Stream_Id subscribe(std::string connection, - Stream_Handler handler, - Transport_Readiness readiness, - Transport_Diagnostics diagnostics); + [[nodiscard]] static std::shared_ptr create(std::vector plots); + [[nodiscard]] proxy frame_sink(std::string_view plot_id); + [[nodiscard]] Stream_Id subscribe(std::string connection, Stream_Handler handler, Transport_Readiness readiness, Transport_Diagnostics diagnostics); void unsubscribe(Stream_Id stream); void request_video_key_frame(); void shutdown() noexcept; diff --git a/web_server/src/media/Gallery_Video_Stream.ipp b/web_server/src/media/Gallery_Video_Stream.ipp index b4bfd85..ffb8772 100644 --- a/web_server/src/media/Gallery_Video_Stream.ipp +++ b/web_server/src/media/Gallery_Video_Stream.ipp @@ -2,8 +2,8 @@ #include "detail/FFmpeg_Frame_Transport.hpp" #include "detail/Gallery_Frame_Atlas.hpp" #include -#include -#include +#include +#include #include #include #include @@ -23,7 +23,9 @@ struct Gallery_Video_Stream::Private : Prev_Private { }; struct Pending_Plot_Frame { std::size_t slot{}; /* 该完成帧所属的稳定图集槽位。 */ - std::shared_ptr frame{}; + std::shared_ptr pixels{}; /* Sink 边界提取的不可变像素所有权。 */ + proxy* policy_frame{}; /* 必填、非拥有借用；Frame Policy 在 completion 前保持槽位存活。 */ + frame_policy::Frame_Completion completion{}; /* 异步 Sink completion 的唯一所有者。 */ std::chrono::steady_clock::time_point enqueued_at{}; /* 进入媒体串行队列的单调时刻。 */ std::chrono::nanoseconds source_time_unix{}; /* 进入媒体串行队列的 Unix 时间。 */ }; @@ -36,10 +38,10 @@ struct Gallery_Video_Stream::Private : Prev_Private { std::size_t source_slot{}; /* 发布反馈路由到唯一来源 Plot 的槽位。 */ std::uint64_t plot_frame_sequence{}; /* Frame_Policy 当前管理的来源帧序号。 */ bool publication_succeeded{}; /* 媒体传输是否至少成功接收一次发布。 */ - bool publication_feedback_submitted{}; /* 防止失败清理重复提交策略反馈。 */ + proxy* policy_frame{}; /* 必填、非拥有 Frame Policy 槽位借用。 */ + frame_policy::Frame_Completion completion{}; /* 媒体 DAG 完成节点恰好调用一次。 */ }; static constexpr std::size_t maximum_consumers{32}; - Object* object{}; /* Def 最终对象的可空非拥有借用；bind_private_crtp 后有效。 */ Sliding_Statistics compose_ms{600}; /* 单个 Plot 帧应用到持久图集的耗时。 */ Sliding_Statistics queue_delay_ms{600}; /* Plot 发布到媒体 Taskflow 开始处理的排队时间。 */ Sliding_Statistics encode_ms{600}; /* H.264 编码耗时。 */ @@ -47,7 +49,7 @@ struct Gallery_Video_Stream::Private : Prev_Private { std::vector sources{}; /* 已按业务标识排序的稳定图集来源。 */ std::unique_ptr atlas{}; /* 图集像素和各槽位进度的唯一所有者。 */ moodycamel::ConcurrentQueue pending_frames{}; /* 各 Plot 已完成像素帧的无损 MPMC 入口。 */ - std::shared_ptr media_graph{}; /* 每次只消费一个 Plot 完成帧的持久异步 DAG。 */ + proxy media_graph{}; /* 独占当前持久媒体 DAG 的跨模块代理。 */ std::atomic_bool media_task_scheduled{}; /* 唯一媒体 DAG 准入；不承担丢帧策略。 */ std::atomic_uint64_t media_work_generation{}; /* 新完成帧入队后推进，封闭任务退出竞争窗口。 */ detail::FFmpeg_Frame_Transport ffmpeg_transport; /* 仅由媒体 DAG 串行访问的编码上下文。 */ @@ -78,10 +80,6 @@ struct Gallery_Video_Stream::Private : Prev_Private { maximum_taskflow_trace_frames> taskflow_trace_slots{}; Private(); ~Private(); - template void bind_private_crtp(Attached_Object* attached) { - Prev_Private::bind_private_crtp(attached); - object = not_null{static_cast(attached)}; - } void initialize(std::vector plots); [[nodiscard]] bool consumer_accepts() const noexcept; [[nodiscard]] bool remove_consumer(Stream_Id stream) noexcept; @@ -90,20 +88,26 @@ struct Gallery_Video_Stream::Private : Prev_Private { std::string notification) noexcept; void release_pending_frames() noexcept; void fail(std::exception_ptr failure) noexcept; - [[nodiscard]] Gallery_Frame_Publication accept_frame( - std::size_t slot, std::shared_ptr frame); + void accept_frame(std::size_t slot, proxy& frame, frame_policy::Frame_Completion completion, std::weak_ptr lifetime); void arm_media_task(std::weak_ptr lifetime); void begin_media_frame(); void encode_media_frame(); void publish_video_frame(); - void submit_publication_feedback(); + void complete_policy_frame(); void complete_media_frame(); void update_metrics(const Active_Media_Frame& frame, std::size_t encoded_bytes, std::optional encoder_backend); [[nodiscard]] bool mark_taskflow_trace(); void restore_taskflow_trace() noexcept; - void store_taskflow_trace(const Taskflow_Frame_Trace& trace); + void store_taskflow_trace(const Taskflow_Execution_Trace& trace); [[nodiscard]] nlohmann::json taskflow_trace_response() const; }; + +struct Gallery_Video_Stream::Sink::Private : Prev_Private { + Private(std::shared_ptr stream, std::size_t slot) : stream(std::move(stream)), slot(slot) {} + void send(proxy& frame, frame_policy::Frame_Completion completion); + std::shared_ptr stream; /* Sink proxy 独占的共享所有权；覆盖 Frame Policy 的全部在途 completion。 */ + std::size_t slot{}; /* 不可变的图集来源槽位。 */ +}; } diff --git a/web_server/src/media/detail/H264_Encoder.cpp b/web_server/src/media/detail/H264_Encoder.cpp index e147d34..2685fff 100644 --- a/web_server/src/media/detail/H264_Encoder.cpp +++ b/web_server/src/media/detail/H264_Encoder.cpp @@ -1,5 +1,4 @@ #include "H264_Encoder.hpp" -#include #include #include #include @@ -50,10 +49,10 @@ struct H264_Encoder::Private { std::chrono::nanoseconds source_time_unix{}; }; double frame_rate{}; - owner codec_context{}; - owner frame{}; - owner packet{}; - owner converter{}; + AVCodecContext* codec_context{}; /* 独占所有权；由 avcodec_free_context 释放。 */ + AVFrame* frame{}; /* 独占所有权；由 av_frame_free 释放。 */ + AVPacket* packet{}; /* 独占所有权；由 av_packet_free 释放。 */ + SwsContext* converter{}; /* 独占所有权；由 sws_freeContext 释放。 */ std::deque submitted_frames{}; std::uint32_t width{}; std::uint32_t height{}; @@ -91,8 +90,7 @@ struct H264_Encoder::Private { if (!codec_address) throw std::runtime_error( "FFmpeg libx264 encoder is unavailable"); - const not_null codec{codec_address}; - codec_context = avcodec_alloc_context3(codec.get()); + codec_context = avcodec_alloc_context3(codec_address); if (!codec_context) throw std::bad_alloc{}; codec_context->width = static_cast(next_width); codec_context->height = static_cast(next_height); @@ -130,7 +128,7 @@ struct H264_Encoder::Private { "threads=1", 0), "setting libx264 Taskflow execution policy"); - require_ffmpeg(avcodec_open2(codec_context, codec.get(), nullptr), + require_ffmpeg(avcodec_open2(codec_context, codec_address, nullptr), "opening libx264 encoder"); frame = av_frame_alloc(); if (!frame) throw std::bad_alloc{}; @@ -213,11 +211,11 @@ std::optional H264_Encoder::encode( if (received == AVERROR(EAGAIN)) return {}; require_ffmpeg(received, "receiving a libx264 access unit"); struct Packet_Scope { - not_null packet; + AVPacket* packet; /* 必填、非拥有借用；外层 H264_Encoder::Private 覆盖本作用域。 */ ~Packet_Scope() { - av_packet_unref(packet.get()); + av_packet_unref(packet); } - } packet_scope{not_null{d->packet}}; + } packet_scope{d->packet}; if (d->submitted_frames.empty()) throw std::logic_error( "libx264 produced an uncorrelated access unit"); diff --git a/web_server/tests/Event_Latency_Benchmarks.cpp b/web_server/tests/Event_Latency_Benchmarks.cpp index 255e8f4..88f832a 100644 --- a/web_server/tests/Event_Latency_Benchmarks.cpp +++ b/web_server/tests/Event_Latency_Benchmarks.cpp @@ -3,8 +3,10 @@ #include #include #include +#include #include -#include +#include +#include #include #include #include @@ -16,6 +18,7 @@ #include #include #include +#include #include #include #include @@ -69,9 +72,6 @@ struct Distribution { struct Event_Result { Distribution websocket_receive_ms{}; - Distribution scene_queue_ms{}; - Distribution scene_dispatch_ms{}; - Distribution scene_total_ms{}; Distribution receive_to_pixel_ms{}; Distribution media_pipeline_ms{}; Distribution end_to_end_ms{}; @@ -101,7 +101,6 @@ struct Event_Timing { std::int64_t submitted_unix_ns{}; std::int64_t accepted_unix_ns{}; std::uint64_t encoded_count_before{}; - std::uint64_t statistic_count_before{}; }; Configuration configuration{}; @@ -118,40 +117,6 @@ std::vector published_results{}; ? static_cast(end - begin) / 1'000'000.0 : 0.0; } -[[nodiscard]] const nlohmann::json* find_event_statistic( - const nlohmann::json& diagnostics, Event_Type type, - std::string_view statistic) { - const auto input = diagnostics.find("input_statistics"); - if (input == diagnostics.end() || !input->is_object()) return nullptr; - const auto event = input->find( - std::string{magic_enum::enum_name(type)}); - if (event == input->end() || !event->is_object()) return nullptr; - const auto value = event->find(std::string{statistic}); - return value == event->end() || !value->is_object() - ? nullptr : &*value; -} - -[[nodiscard]] std::uint64_t event_statistic_count( - const mcp::Control_Service& control, std::string_view id, - Event_Type type) { - const auto diagnostics = control.plot_diagnostics(id); - const auto* statistic = find_event_statistic( - diagnostics, type, "total_ms"); - return statistic ? statistic->value("count", 0ULL) : 0ULL; -} - -[[nodiscard]] double event_statistic_latest( - const nlohmann::json& diagnostics, Event_Type type, - std::string_view statistic) { - const auto* value = find_event_statistic(diagnostics, type, statistic); - if (!value) - throw std::runtime_error( - "missing Scene event statistic " + - std::string{magic_enum::enum_name(type)} + "/" + - std::string{statistic}); - return value->at("latest").get(); -} - [[nodiscard]] std::string input_message(Event_Type type, std::size_t plot_index, std::uint64_t sample) { @@ -184,41 +149,33 @@ std::vector published_results{}; } void configure_plot(mcp::Control_Service& control, std::string_view id) { - if (!control.write_plot_property(id, - "frame-analysis", "pacing_mode", "manual").value( - "success", false)) - throw std::runtime_error("failed to select manual Plot policy"); + if (!control.write_plot_property(id, "frame-analysis", "user_frames_per_second", 100.0).value("success", false)) throw std::runtime_error("failed to configure Plot frame rate"); } -[[nodiscard]] bool manual_pixel_policy_applied( - const mcp::Control_Service& control, std::string_view id) { +[[nodiscard]] bool plot_policy_available(const mcp::Control_Service& control, std::string_view id) { const auto diagnostics = control.plot_diagnostics(id); const auto policy = diagnostics.find("frame_policy"); - if (policy == diagnostics.end() || !policy->is_object()) return false; - const auto state = policy->find("configuration"); - if (state == policy->end() || !state->is_object()) return false; - const auto mode = state->find("mode"); - const auto pixels = state->find("pixel_delivery_enabled"); - return mode != state->end() && mode->is_string() && - *mode == "manual" && pixels != state->end() && - pixels->is_boolean() && pixels->get(); + return diagnostics.value("available", false) && policy != diagnostics.end() && policy->is_object(); } -void cooperatively_wait(std::string_view operation, - const std::function& completed, - std::chrono::seconds timeout) { +void cooperatively_wait(std::string_view operation, const std::function& completed, std::chrono::seconds timeout) { std::string failure; - Task_Graph graph{"web.event-latency.await"}; - graph.add("await", [&] { + std::exception_ptr runtime_failure; + std::binary_semaphore finished{0}; + auto graph = task_flow::make_task_graph("web.event-latency.await"); + if (!graph->add("await", [&] { const auto deadline = Steady_Clock::now() + timeout; - Task_Graph::corun_until([&] { + graph->corun_until([&] { if (completed()) return true; if (Steady_Clock::now() < deadline) return false; failure = std::string{operation} + " timed out"; return true; }); - }); - aethera::detail::run_taskflow(graph); + })) throw std::runtime_error("failed to create Web benchmark await task"); + if (graph->run([&](std::exception_ptr current_failure) noexcept { runtime_failure = current_failure; finished.release(); }) != Run_Taskflow_Result::submitted) throw std::runtime_error("failed to submit Web benchmark await task"); + /* Google Benchmark requires a terminal result; only this harness thread blocks. */ + finished.acquire(); + if (runtime_failure) std::rethrow_exception(runtime_failure); if (!failure.empty()) throw std::runtime_error(failure); } @@ -263,8 +220,6 @@ void print_results() { << std::setw(5) << "Dim" << std::setw(18) << "Event" << std::right << std::setw(10) << "WS P95" - << std::setw(11) << "Queue P95" - << std::setw(11) << "Disp P95" << std::setw(11) << "Pixel P95" << std::setw(11) << "Media P95" << std::setw(11) << "E2E P50" @@ -274,8 +229,6 @@ void print_results() { for (std::size_t index = 0; index < event_count; ++index) { const auto& result = plot.events[index]; const auto websocket = result.websocket_receive_ms.summarize(); - const auto queue = result.scene_queue_ms.summarize(); - const auto dispatch = result.scene_dispatch_ms.summarize(); const auto pixel = result.receive_to_pixel_ms.summarize(); const auto media = result.media_pipeline_ms.summarize(); const auto end_to_end = result.end_to_end_ms.summarize(); @@ -284,8 +237,6 @@ void print_results() { << std::setw(18) << magic_enum::enum_name( event_types[index]) << std::right << std::setw(10) << websocket.p95 - << std::setw(11) << queue.p95 - << std::setw(11) << dispatch.p95 << std::setw(11) << pixel.p95 << std::setw(11) << media.p95 << std::setw(11) << end_to_end.p50 @@ -301,8 +252,6 @@ void run_event_latency(benchmark::State& state) { try { const auto definitions = selected_definitions(); - auto control_slot = std::make_shared>>(); auto videos = std::make_shared>>(); active.reserve(definitions.size()); @@ -310,12 +259,7 @@ void run_event_latency(benchmark::State& state) { for (const auto& definition : definitions) { const auto id = std::string{definition.id}; auto probe = std::make_shared(); - auto video = media::Gallery_Video_Stream::create({ - {id, [control_slot, id](Frame_Publication_Feedback feedback) { - if (const auto control = control_slot->load()) - control->submit_publication_feedback( - id, std::move(feedback)); - }}}); + auto video = media::Gallery_Video_Stream::create({{id}}); videos->emplace(id, video); const auto video_subscription = video->subscribe( "web-event-latency-" + std::string{definition.id}, @@ -341,15 +285,10 @@ void run_event_latency(benchmark::State& state) { auto control = mcp::Control_Service::create(mcp::Gallery_Output{ .width = configuration.width, .height = configuration.height, - .publish = [videos]( - std::string_view id, - std::shared_ptr frame) { + .make_sink = [videos](std::string_view id) -> proxy { const auto found = videos->find(std::string{id}); - return found == videos->end() - ? Gallery_Frame_Publication::ignored - : found->second->accept_frame(id, std::move(frame)); + return found == videos->end() ? proxy{} : found->second->frame_sink(id); }}); - control_slot->store(control); for (auto& item : active) { configure_plot(*control, item.id); item.input = std::make_shared(control, item.id); @@ -361,9 +300,9 @@ void run_event_latency(benchmark::State& state) { }.dump()); } - cooperatively_wait("manual Plot policies", [&] { + cooperatively_wait("Plot policies", [&] { return std::ranges::all_of(active, [&control](const Active_Plot& item) { - return manual_pixel_policy_applied(*control, item.id); + return plot_policy_available(*control, item.id); }); }, std::chrono::seconds{10}); @@ -386,8 +325,8 @@ void run_event_latency(benchmark::State& state) { std::vector timings(active.size()); std::string failure; - Task_Graph coordinator{"web.event-latency.measure"}; - coordinator.add("measure", [&] { + auto coordinator = task_flow::make_task_graph("web.event-latency.measure"); + if (!coordinator->add("measure", [&] { for (std::uint64_t sample = 0; sample < configuration.samples; ++sample) { for (std::size_t event_index = 0; @@ -400,8 +339,6 @@ void run_event_latency(benchmark::State& state) { timing.encoded_count_before = item.probe->frame_count.load( std::memory_order_acquire); - timing.statistic_count_before = event_statistic_count( - *control, item.id, type); const auto message = input_message( type, plot_index, sample); timing.submitted_unix_ns = system_time_ns(); @@ -413,17 +350,14 @@ void run_event_latency(benchmark::State& state) { const auto deadline = Steady_Clock::now() + std::chrono::seconds{60}; - Task_Graph::corun_until([&] { + coordinator->corun_until([&] { bool complete{true}; for (std::size_t index = 0; index < active.size(); ++index) { const auto encoded = active[index].probe-> frame_count.load(std::memory_order_acquire) > timings[index].encoded_count_before; - const auto observed = event_statistic_count( - *control, active[index].id, type) > - timings[index].statistic_count_before; - complete = complete && encoded && observed; + complete = complete && encoded; } if (complete) return true; if (Steady_Clock::now() < deadline) return false; @@ -442,9 +376,6 @@ void run_event_latency(benchmark::State& state) { const auto published = active[plot_index].probe-> published_time_unix_ns.load( std::memory_order_acquire); - const auto diagnostics = - control->plot_diagnostics( - active[plot_index].id); auto& result = published_results[plot_index]. events[event_index]; result.websocket_receive_ms.add(milliseconds( @@ -456,19 +387,19 @@ void run_event_latency(benchmark::State& state) { source, published)); result.end_to_end_ms.add(milliseconds( timing.submitted_unix_ns, published)); - result.scene_queue_ms.add(event_statistic_latest( - diagnostics, type, "queue_wait_ms")); - result.scene_dispatch_ms.add(event_statistic_latest( - diagnostics, type, "dispatch_ms")); - result.scene_total_ms.add(event_statistic_latest( - diagnostics, type, "total_ms")); } } } - }); + })) throw std::logic_error("failed to create Web event latency measurement task"); - for ([[maybe_unused]] auto iteration : state) - aethera::detail::run_taskflow(coordinator); + for ([[maybe_unused]] auto iteration : state) { + std::exception_ptr runtime_failure; + std::binary_semaphore finished{0}; + if (coordinator->run([&](std::exception_ptr current_failure) noexcept { runtime_failure = current_failure; finished.release(); }) != Run_Taskflow_Result::submitted) throw std::runtime_error("failed to submit Web event latency measurement task"); + /* Benchmark-only terminal join; all asynchronous engine work remains completion-driven. */ + finished.acquire(); + if (runtime_failure) std::rethrow_exception(runtime_failure); + } if (!failure.empty()) state.SkipWithError(failure); Distribution aggregate; @@ -557,7 +488,7 @@ BENCHMARK(run_event_latency)->Iterations(1)->UseRealTime(); int main(int argc, char** argv) { if (!aethera::web::event_latency_benchmarks::configure(argc, argv)) return 2; - aethera::initialize_runtime({}); + if (aethera::initialize_task_runtime() != aethera::Initialize_Task_Runtime_Result::initialized) return 2; benchmark::Initialize(&argc, argv); if (benchmark::ReportUnrecognizedArguments(argc, argv)) return 2; benchmark::AddCustomContext( diff --git a/web_server/tests/Gallery_Plots_Tests.cpp b/web_server/tests/Gallery_Plots_Tests.cpp index f1e1452..8195b9a 100644 --- a/web_server/tests/Gallery_Plots_Tests.cpp +++ b/web_server/tests/Gallery_Plots_Tests.cpp @@ -37,7 +37,7 @@ TEST(Gallery_Plots, Concrete_Scenes_Submit_Input_Through_Final_Storage) { const auto* definition = find_gallery_plot_definition(id); ASSERT_NE(definition, nullptr); auto build = definition->create(); - auto event = std::make_shared( + auto event = std::make_unique( Event_Type::pointer_move, Event_Timeline_Time{}); std::visit([event = std::move(event)](auto& scene) mutable { scene->submit_event(std::move(event)); diff --git a/web_server/tests/Gallery_Video_Stream_Tests.cpp b/web_server/tests/Gallery_Video_Stream_Tests.cpp index 40638c2..dcccb9c 100644 --- a/web_server/tests/Gallery_Video_Stream_Tests.cpp +++ b/web_server/tests/Gallery_Video_Stream_Tests.cpp @@ -1,12 +1,14 @@ #include #include #include -#include +#include +#include #include #include #include #include #include +#include #include #include @@ -15,58 +17,54 @@ namespace { using Clock = std::chrono::steady_clock; -void corun_until_or_throw(std::string_view operation, - const std::function& completed) { +void corun_until_or_throw(std::string_view operation, const std::function& completed) { std::string failure; - Task_Graph coordinator{"gallery.video.test.await"}; - coordinator.add("await", [&] { - const auto deadline = Clock::now() + std::chrono::seconds{10}; - Task_Graph::corun_until([&] { + std::exception_ptr runtime_failure; + std::binary_semaphore finished{0}; + auto coordinator = task_flow::make_task_graph("gallery.video.test.await"); + const auto deadline = Clock::now() + std::chrono::seconds{10}; + if (!coordinator->add("await", [&] { + coordinator->corun_until([&] { if (completed()) return true; if (Clock::now() < deadline) return false; failure = std::string{operation} + " timed out"; return true; }); + })) throw std::runtime_error("failed to create gallery test await task"); + const auto submitted = coordinator->run([&](std::exception_ptr current_failure) noexcept { + runtime_failure = current_failure; + finished.release(); }); - aethera::detail::run_taskflow(coordinator); + if (submitted != Run_Taskflow_Result::submitted) throw std::runtime_error("failed to submit gallery test await task"); + /* Test-only join boundary: production paths remain completion-driven and never block a caller. */ + finished.acquire(); + if (runtime_failure) std::rethrow_exception(runtime_failure); if (!failure.empty()) throw std::runtime_error(failure); } -void configure_manual_pixel_delivery( - const std::shared_ptr& control, - std::string_view id) { - ASSERT_TRUE(control->write_plot_property(id, - "frame-analysis", "pacing_mode", "manual").value( - "success", false)); - ASSERT_NO_THROW(corun_until_or_throw("manual Plot policy", [&] { +void configure_frame_policy(const std::shared_ptr& control, std::string_view id) { + ASSERT_TRUE(control->write_plot_property(id, "frame-analysis", "user_frames_per_second", 100.0).value("success", false)); + ASSERT_NO_THROW(corun_until_or_throw("Plot policy", [&] { const auto diagnostics = control->plot_diagnostics(id); - if (!diagnostics.contains("frame_policy")) return false; - const auto& configuration = diagnostics.at("frame_policy").at( - "configuration"); - return configuration.at("mode") == "manual" && - configuration.at("pixel_delivery_enabled").get(); + return diagnostics.value("available", false) && diagnostics.contains("frame_policy"); })); } -void schedule_manual_frame( - const std::shared_ptr& control, - std::string_view id, - std::uint64_t correlation_sequence) { +void schedule_manual_frame(const std::shared_ptr& control, std::string_view id, std::uint64_t correlation_sequence) { const auto now = Clock::now(); - control->request_frame(id, Frame_Request{ + control->request_frame(id, Gallery_Frame_Request{ .issued_at = now, - .sequence = correlation_sequence, .time_milliseconds = std::chrono::duration( now.time_since_epoch()).count(), + .correlation_id = correlation_sequence, .width = 720, - .height = 420, - .source = Frame_Request_Source::immediate + .height = 420 }); } -TEST(Gallery_Video_Stream, - Publication_Feedback_Task_Releases_Each_Plots_Independent_Frame) { +TEST(Gallery_Video_Stream, Sink_Completion_Task_Releases_Each_Plots_Independent_Frame) { + ASSERT_EQ(initialize_task_runtime(), Initialize_Task_Runtime_Result::initialized); const auto definitions = gallery_plot_definitions(); const auto found = std::ranges::find_if( definitions, [](const Plot_Definition& definition) { @@ -80,41 +78,20 @@ TEST(Gallery_Video_Stream, ASSERT_NE(second, definitions.end()); const std::string plot_a{found->id}; const std::string plot_b{second->id}; - auto control_slot = std::make_shared>>(); - auto stream = Gallery_Video_Stream::create({ - {plot_a, [control_slot, plot_a](Frame_Publication_Feedback feedback) { - if (const auto control = control_slot->load()) - control->submit_publication_feedback( - plot_a, std::move(feedback)); - }}, - {plot_b, [control_slot, plot_b](Frame_Publication_Feedback feedback) { - if (const auto control = control_slot->load()) - control->submit_publication_feedback( - plot_b, std::move(feedback)); - }}}); + auto stream = Gallery_Video_Stream::create({{plot_a}, {plot_b}}); ASSERT_TRUE(stream); auto control = mcp::Control_Service::create(mcp::Gallery_Output{ - .publish = [stream](std::string_view id, - std::shared_ptr frame) { - return stream->accept_frame(id, std::move(frame)); - }}); - control_slot->store(control); - configure_manual_pixel_delivery(control, plot_a); - configure_manual_pixel_delivery(control, plot_b); + .make_sink = [stream](std::string_view id) { return stream->frame_sink(id); }}); + configure_frame_policy(control, plot_a); + configure_frame_policy(control, plot_b); constexpr std::size_t frame_count{5}; - std::array sequences{}; std::atomic_size_t received{}; std::atomic> outstanding{}; const auto subscription = stream->subscribe( "gallery-video-test", [&](Gallery_Stream_Frame frame) { if (!frame.video) return false; - const auto index = received.fetch_add( - 1, std::memory_order_acq_rel); - if (index < sequences.size()) - sequences[index].store( - frame.video->sequence, std::memory_order_release); + received.fetch_add(1, std::memory_order_acq_rel); outstanding.store(std::move(frame.video), std::memory_order_release); return true; @@ -126,31 +103,30 @@ TEST(Gallery_Video_Stream, plot_a, plot_a, plot_b, plot_a, plot_b}; stream->request_taskflow_trace(1); for (std::size_t index = 0; index < schedule.size(); ++index) { - const auto published_before = control->plot_diagnostics(schedule[index]) - .at("frame_policy").at("lifecycle") - .at("published_frame_count").get(); + const auto published_before = control->plot_diagnostics(schedule[index]).at("frame_policy").at("completed_frames").get(); + const auto received_before = received.load(std::memory_order_acquire); schedule_manual_frame(control, schedule[index], index + 1U); ASSERT_NO_THROW(corun_until_or_throw("encoded Plot frame", [&] { - return received.load(std::memory_order_acquire) > index; + return received.load(std::memory_order_acquire) > received_before; })); ASSERT_NO_THROW(corun_until_or_throw( "frame policy publication feedback", [&] { const auto diagnostics = control->plot_diagnostics(schedule[index]); - return diagnostics.at("frame_lifecycle") == "running" && - diagnostics.at("frame_policy").at("lifecycle") - .at("published_frame_count") > published_before; + return diagnostics.at("frame_policy").at("completed_frames") > published_before; })); outstanding.store({}, std::memory_order_release); } - EXPECT_EQ(received.load(std::memory_order_acquire), frame_count); - for (std::size_t index = 0; index < frame_count; ++index) - EXPECT_EQ(sequences[index].load(std::memory_order_acquire), - index + 1U); - const auto diagnostics = stream->diagnostics(); - EXPECT_EQ(diagnostics.at("received_frame_count"), frame_count); - EXPECT_EQ(diagnostics.at("processed_frame_count"), frame_count); - EXPECT_EQ(diagnostics.at("encoded_frame_count"), frame_count); + nlohmann::json diagnostics; + ASSERT_NO_THROW(corun_until_or_throw("gallery media drain", [&] { + auto current = stream->diagnostics(); + if (current.at("pending_frame_count") != 0U || current.at("received_frame_count") != current.at("processed_frame_count") || current.at("received_frame_count") != current.at("encoded_frame_count")) return false; + diagnostics = std::move(current); + return true; + })); + EXPECT_GE(diagnostics.at("received_frame_count"), frame_count); + EXPECT_EQ(diagnostics.at("processed_frame_count"), diagnostics.at("received_frame_count")); + EXPECT_EQ(diagnostics.at("encoded_frame_count"), diagnostics.at("received_frame_count")); EXPECT_EQ(diagnostics.at("pending_frame_count"), 0U); ASSERT_NO_THROW(corun_until_or_throw("gallery Taskflow trace", [&] { @@ -165,14 +141,20 @@ TEST(Gallery_Video_Stream, ASSERT_NE(graph, graphs.end()); const auto& nodes = graph->at("nodes"); const auto feedback = std::ranges::find_if(nodes, [](const auto& value) { - return value.value("name", std::string{}) == - "frame.policy.publish.feedback"; + return value.value("name", std::string{}) == "frame.policy.sink.complete"; }); ASSERT_NE(feedback, nodes.end()); EXPECT_EQ(feedback->at("attributes").at("owner"), "frame_policy"); stream->unsubscribe(subscription); stream->shutdown(); + std::weak_ptr stream_lifetime = stream; + stream.reset(); + EXPECT_FALSE(stream_lifetime.expired()); + std::atomic_bool stopped{}; + control->stop([&stopped] { stopped.store(true, std::memory_order_release); }); + ASSERT_NO_THROW(corun_until_or_throw("Frame Policy Sink retirement", [&] { return stopped.load(std::memory_order_acquire); })); + EXPECT_TRUE(stream_lifetime.expired()); } } diff --git a/web_server/tests/Sliding_Statistics_Tests.cpp b/web_server/tests/Sliding_Statistics_Tests.cpp index bbb804a..a3047c2 100644 --- a/web_server/tests/Sliding_Statistics_Tests.cpp +++ b/web_server/tests/Sliding_Statistics_Tests.cpp @@ -1,4 +1,4 @@ -#include +#include #include #include #include @@ -7,64 +7,44 @@ namespace aethera::web { TEST(Sliding_Statistics, Publishes_Derived_Window_Statistics_On_Submit) { Sliding_Statistics statistics{20}; for (int value = 1; value <= 20; ++value) - static_cast(statistics.submit(static_cast(value))); + statistics.add(static_cast(value)); - const auto& state = statistics.state(); - EXPECT_EQ(state.count, 20U); - EXPECT_DOUBLE_EQ(state.latest, 20.0); - EXPECT_DOUBLE_EQ(state.minimum, 1.0); - EXPECT_DOUBLE_EQ(state.maximum, 20.0); - EXPECT_DOUBLE_EQ(state.average, 10.5); - EXPECT_DOUBLE_EQ(state.trimmed_average, 10.5); - EXPECT_NEAR(state.p50, 10.0, 2.0); - EXPECT_GE(state.p95, 15.0); - EXPECT_LE(state.p95, 20.0); - EXPECT_GE(state.p99, state.p95); - EXPECT_LE(state.p99, 20.0); + const auto state = statistics.summary(); + EXPECT_EQ(state.sample_count, 20U); + EXPECT_DOUBLE_EQ(state.average, 10.5); + EXPECT_DOUBLE_EQ(state.p95, 19.0); + EXPECT_GT(state.standard_deviation, 0.0); } TEST(Sliding_Statistics, Overwrites_Oldest_And_Ignores_Nonfinite_Values) { Sliding_Statistics statistics{3}; - static_cast(statistics.submit(1.0)); - static_cast(statistics.submit(2.0)); - static_cast(statistics.submit(std::numeric_limits::infinity())); - static_cast(statistics.submit(3.0)); - static_cast(statistics.submit(4.0)); + statistics.add(1.0); + statistics.add(2.0); + statistics.add(std::numeric_limits::infinity()); + statistics.add(3.0); + statistics.add(4.0); - const auto& state = statistics.state(); - EXPECT_EQ(state.count, 3U); - EXPECT_DOUBLE_EQ(state.latest, 4.0); - EXPECT_DOUBLE_EQ(state.minimum, 2.0); - EXPECT_DOUBLE_EQ(state.maximum, 4.0); - EXPECT_DOUBLE_EQ(state.average, 3.0); + const auto state = statistics.summary(); + EXPECT_EQ(state.sample_count, 3U); + EXPECT_DOUBLE_EQ(state.average, 3.0); + EXPECT_DOUBLE_EQ(state.p95, 4.0); } -TEST(Sliding_Statistics, Estimates_Quantiles_Without_Growing_The_Window) { +TEST(Sliding_Statistics, Computes_Quantiles_From_The_Authoritative_Window) { Sliding_Statistics statistics{64}; for (int value = 1; value <= 1000; ++value) - static_cast(statistics.submit(static_cast(value))); + statistics.add(static_cast(value)); - const auto& state = statistics.state(); - EXPECT_EQ(state.count, 64U); + const auto state = statistics.summary(); + EXPECT_EQ(state.sample_count, 64U); EXPECT_NEAR(state.average, 968.5, 0.001); - EXPECT_DOUBLE_EQ(state.minimum, 937.0); - EXPECT_DOUBLE_EQ(state.maximum, 1000.0); - EXPECT_NEAR(state.p50, 500.0, 10.0); - EXPECT_NEAR(state.p95, 950.0, 15.0); - EXPECT_NEAR(state.p99, 990.0, 15.0); + EXPECT_NEAR(state.p95, 997.0, 1.0); } -TEST(Sliding_Statistics, Keeps_Approximate_Quantiles_Ordered_For_Nonstationary_Input) { +TEST(Sliding_Statistics, Clear_Drops_All_Window_State) { Sliding_Statistics statistics{16}; - for (std::size_t phase = 0; phase < 200; ++phase) { - const double baseline = phase % 2 == 0 ? 1'000'000.0 : -1'000'000.0; - for (std::size_t sample = 0; sample < 17; ++sample) { - const auto state = statistics.submit( - baseline + static_cast(sample * sample)); - EXPECT_TRUE(std::isfinite(state.trimmed_average)); - EXPECT_LE(state.p50, state.p95); - EXPECT_LE(state.p95, state.p99); - } - } + statistics.add(1.0); + statistics.clear(); + EXPECT_EQ(statistics.summary().sample_count, 0U); } } diff --git a/webapp_gallery/src/app.tsx b/webapp_gallery/src/app.tsx index 8d5b782..c3dbeae 100644 --- a/webapp_gallery/src/app.tsx +++ b/webapp_gallery/src/app.tsx @@ -27,25 +27,12 @@ type Plot_Execution_Policies = Record; type Frame_Analysis = Omit & {data_generator?: Data_Generator}; type Schema = {protocol: "aethera.plot.inspector"; version: 3; components: Component[]; frame_analysis: Frame_Analysis}; type Stream_Status = "IDLE" | "CONNECTING" | "LIVE" | "OFFLINE"; -type Frame_Pacing_Mode = "manual" | "fixed_rate" | "maximum_rate"; -type Frame_Delivery = "gallery-pixels" | "diagnostics"; -type Pixel_Format = "rgba8" | "bgra8_premultiplied"; -type Frame_Stage_Values = Record; -type Statistic_Snapshot = {count: number; latest: number; minimum: number; maximum: number; average: number; - trimmed_average: number; variability: number; p50: number; p95: number; p99: number}; -type Event_Statistics = Record>; type Browser_Input_Statistic = {count: number; latest_ms: number; average_ms: number; maximum_ms: number; sent_count: number; disconnected_count: number; websocket_buffered_bytes: number}; type Browser_Input_Statistics = Record; -type Plot_Diagnostics = {protocol: "aethera.plot.diagnostics"; version: 4; dimension: "2D" | "3D"; - sequence: number; correlation_id: number; rendered_sequence: number; rendered_correlation_id: number; - generated_time_unix_ms: number; delivery: Frame_Delivery; frame_rate_fps: number; dropped_sequence_count: number; - window_capacity: number; pixel: {width: number; height: number; format: Pixel_Format; native_format: Pixel_Format; - supported_formats: Pixel_Format[]; byte_length: number}; - frame_policy: Frame_Policy_State; - frame_statistics: Record; input_statistics: Event_Statistics}; -type Frame_Sample = {sequence: number; generated_at_ms: number; received_at_ms: number; values: Frame_Stage_Values}; +type Plot_Diagnostics = {protocol: "aethera.plot.diagnostics"; version: 6; available: boolean; plot: string; dimension: "2D" | "3D"; + frame_policy: Frame_Policy_State; datoviz?: Datoviz_Frame_Observation}; type Video_Presentation_Metrics = {frame_rate_fps: number; presented_frames: number; dropped_frames: number; jitter_buffer_ms: number; decode_processing_ms: number; estimated_playout_delay_ms: number; server_pipeline_ms: number; websocket_arrival_ms: number; source_arrival_ms: number; @@ -75,11 +62,7 @@ type Gallery_Video_State = {status: Stream_Status; layout: Gallery_Layout | null video_group: Gallery_H264_Group | null; presentation: Video_Presentation_Metrics; transport: Gallery_Video_Metrics | null; error: string | null}; type Gallery_Video_States = Record; -type Frame_Diagnostics = {server: Plot_Diagnostics; samples: Frame_Sample[]; pixel_presentation: Video_Presentation_Metrics; - browser_input_statistics: Browser_Input_Statistics}; -type Frame_Metrics = {sequence: number; generated_time_unix_ms: number; server_completion_ms: number; average_server_completion_ms: number; - p95_server_completion_ms: number; p99_server_completion_ms: number; frame_rate_fps: number; p95_frame_interval_jitter_ms: number; - pacing_mode: Frame_Pacing_Mode; fixed_rate_fps: number; delivery: Frame_Delivery; pixel_presentation: Video_Presentation_Metrics}; +type Frame_Metrics = {sequence: number; sampled_time_unix_ms: number; frame_rate_fps: number}; type Taskflow_Node_Trace = {native_id: string; id: string; parent_id: string; name: string; type: string; predecessors: string[]; successors: string[]; attributes?: Record; owner?: {component: string; label: string; kind: string}; prop?: Record; state?: Record}; @@ -91,17 +74,12 @@ type Taskflow_Execution_Trace = {native_id: string; node_id: string; worker_id: cooperative_waits: Array<{started_ms: number; finished_ms: number}>; observer_entry_ms: number; observer_exit_ms: number; queue_wait_ms: number}; type Frame_Policy_State = { - generation: number; - configuration: {mode: Frame_Pacing_Mode; render_enabled: boolean; pixel_delivery_enabled: boolean; fixed_rate_fps: number}; - observation: {duration_ms: number; request_count: number; submitted_frame_count: number; completed_frame_count: number; active_frame_count: number}; - throughput: {request_rate_fps: number; submission_rate_fps: number; completion_rate_fps: number; target_achievement_ratio: number; - latest_frame_interval_ms: number; average_frame_interval_ms: number; frame_interval_jitter_ms: number}; - requests: {periodic: number; immediate: number; maximum_rate: number; accepted: number; coalesced: number; policy_rejected: number; - frame_slot_backpressure: number; scene_rejected: number; acceptance_ratio: number; coalescing_ratio: number; backpressure_ratio: number}; - latency: {latest_tick_queue_ms: number; average_tick_queue_ms: number; maximum_tick_queue_ms: number; - latest_completion_ms: number; average_completion_ms: number; maximum_completion_ms: number}; - last_frame: {sequence: number; request_source: "unspecified" | "periodic" | "immediate" | "maximum_rate"}; + timer_ticks: number; dropped_timer_ticks: number; completed_frames: number; effective_frames_per_second: number; + render_capacity_fps: number | null; send_capacity_fps: number | null; + render_time: Kernel_Statistic; send_time: Kernel_Statistic; end_to_end_time: Kernel_Statistic; + configuration_error: string | null; render_submission_error: string | null; }; +type Kernel_Statistic = {sample_count: number; average_ns: number; p95_ns: number; standard_deviation_ns: number}; type Datoviz_Frame_Observation = { render_sequence: number; path: "recorded" | "reused"; @@ -125,21 +103,6 @@ type Taskflow_Frame_Response = {protocol: "aethera.taskflow.frames"; version: 1; captured: number; complete: boolean; frames: Taskflow_Frame_Trace[]; media_requested?: number; media_captured?: number; media_remaining?: number}; type Gallery_Pipeline_State = Record; -type Taskflow_Worker_State = {id: number; task_count: number; entry_queue_size: number; entry_queue_capacity: number; - peak_queue_size: number; max_queue_capacity: number; active_task: {native_id: string; type: string; time_ns: number}; - task_time_ns: number; busy_time_ns: number; cpu_time_ns: number; cooperative_wait_count: number; - cooperative_wait_time_ns: number; idle_time_ns: number; - min_task_time_ns: number; max_task_time_ns: number; utilization: number; cpu_utilization: number}; -type Taskflow_Type_State = {name: string; count: number; total_time_ns: number; min_time_ns: number; max_time_ns: number}; -type Taskflow_Runtime_State = {protocol: "aethera.taskflow.runtime"; version: 2; worker_count: number; active_topologies: number; - active_taskflows: number; peak_active_taskflows: number; completed_taskflows: number; failed_taskflows: number; - active_tasks: number; peak_active_tasks: number; active_workers: number; peak_active_workers: number; observed_tasks: number; - named_tasks: number; peak_worker_queue_size: number; max_worker_queue_capacity: number; - longest_task: {native_id: string; name: string; type: string; time_ns: number}; total_task_time_ns: number; - worker_busy_time_ns: number; worker_cpu_time_ns: number; observed_wall_time_ns: number; - cooperative_wait_count: number; cooperative_wait_time_ns: number; - worker_utilization: number; worker_cpu_utilization: number; - task_types: Taskflow_Type_State[]; workers: Taskflow_Worker_State[]}; const default_plot_execution_policy = (): Plot_Execution_Policy => ({visible: true}); @@ -159,9 +122,8 @@ function gallery_socket_url(path: string) { function valid_plot_diagnostics(value: unknown): value is Plot_Diagnostics { if (!value || typeof value !== "object") return false; const diagnostics = value as Partial; - return diagnostics.protocol === "aethera.plot.diagnostics" && diagnostics.version === 4 && - typeof diagnostics.sequence === "number" && Boolean(diagnostics.frame_statistics) && - Boolean(diagnostics.input_statistics) && Boolean(diagnostics.frame_policy); + return diagnostics.protocol === "aethera.plot.diagnostics" && diagnostics.version === 6 && + diagnostics.available === true && Boolean(diagnostics.frame_policy); } function valid_gallery_layout(value: unknown): value is Gallery_Layout { @@ -713,7 +675,6 @@ function use_gallery_video(plots: Plot[]): Gallery_Video_States { function use_plot_stream(plot: Plot, surface_ref: React.RefObject, - shared_presentation: Video_Presentation_Metrics, tile_width = 720, tile_height = 420) { const [status, set_status] = useState("CONNECTING"); const [metrics, set_metrics] = useState(null); @@ -721,13 +682,11 @@ function use_plot_stream(plot: Plot, const [error, set_error] = useState(null); const socket_ref = useRef(null); const viewport_ref = useRef({width: tile_width, height: tile_height}); - const presentation_ref = useRef(shared_presentation); const browser_input_accumulators = useRef>({}); const [browser_input_statistics, set_browser_input_statistics] = useState({}); - useEffect(() => { presentation_ref.current = shared_presentation; }, [shared_presentation]); useEffect(() => { viewport_ref.current = {width: tile_width, height: tile_height}; }, [tile_width, tile_height]); const transmit = useCallback((kind: "input", event?: Record) => { @@ -773,7 +732,6 @@ function use_plot_stream(plot: Plot, useEffect(() => { let stopped = false; - let samples: Frame_Sample[] = []; const socket = new ReconnectingWebSocket(socket_url(plot.websocket), [], { minReconnectionDelay: 300, maxReconnectionDelay: 5000, reconnectionDelayGrowFactor: 1.6, maxRetries: Number.POSITIVE_INFINITY @@ -786,40 +744,14 @@ function use_plot_stream(plot: Plot, const server = detail?.diagnostics; if (stopped || detail?.plot_id !== plot.id || !valid_plot_diagnostics(server)) return; - const now = performance.now(); - const values = Object.fromEntries(Object.entries(server.frame_statistics) - .map(([key, statistic]) => [key, statistic.latest])); - if (server.sequence !== 0 && - samples.at(-1)?.sequence !== server.sequence) - samples = [...samples, {sequence: server.sequence, - generated_at_ms: server.generated_time_unix_ms, - received_at_ms: now, values}].slice(-600); const browser_inputs = read_browser_input_statistics(); - const diagnostics: Frame_Diagnostics = { - server, samples, pixel_presentation: {...presentation_ref.current}, - browser_input_statistics: browser_inputs - }; set_browser_input_statistics(browser_inputs); set_server_diagnostics(server); - const completion = server.frame_statistics.server_completion_ms; - const interval = server.frame_statistics.frame_interval_ms; set_metrics({ - sequence: server.sequence, - generated_time_unix_ms: server.generated_time_unix_ms, - server_completion_ms: completion?.latest ?? 0, - average_server_completion_ms: completion?.average ?? 0, - p95_server_completion_ms: completion?.p95 ?? 0, - p99_server_completion_ms: completion?.p99 ?? 0, - frame_rate_fps: server.frame_rate_fps, - p95_frame_interval_jitter_ms: interval?.p95 ?? 0, - pacing_mode: server.frame_policy.configuration.mode, - fixed_rate_fps: server.frame_policy.configuration.fixed_rate_fps, - delivery: server.delivery, - pixel_presentation: diagnostics.pixel_presentation + sequence: server.frame_policy.completed_frames, + sampled_time_unix_ms: Date.now(), + frame_rate_fps: server.frame_policy.effective_frames_per_second }); - window.dispatchEvent(new CustomEvent("aethera-frame-diagnostics", { - detail: {plot_id: plot.id, diagnostics} - })); }; window.addEventListener("aethera:plot-diagnostics", receive_diagnostics); const on_manual_frame = (event: Event) => { @@ -841,7 +773,6 @@ function use_plot_stream(plot: Plot, const on_diagnostics_reset = (event: Event) => { const detail = (event as CustomEvent<{plot_id: string}>).detail; if (detail?.plot_id !== plot.id) return; - samples = []; browser_input_accumulators.current = {}; set_browser_input_statistics({}); set_metrics(null); @@ -1376,37 +1307,35 @@ function Frame_Policy_Pane({plot, analysis, busy, on_refresh, on_update, on_manu }; const frame = response?.frames[frame_index]; const state = frame?.frame_policy; - const percent = (value: number) => `${(value * 100).toFixed(1)}%`; const rate = (value: number) => `${value.toFixed(2)} FPS`; - const duration = (value: number) => `${value.toFixed(value < 10 ? 3 : 2)} ms`; + const duration_ns = (value: number) => `${(value / 1_000_000).toFixed(value < 10_000_000 ? 3 : 2)} ms`; return
{plot.dimension} 后端帧策略配置和运行统计由 Kernel Frame_Policy 双缓冲状态发布；查询只读取已发布状态，不扫描任务图，也不锁住渲染线程。
{analysis ?
{fields.map(field => on_update(analysis, field, value)}/>)}
:

正在读取帧策略…

}
-
按完成帧采样策略状态策略状态绑定在对应帧上，与 Taskflow 帧捕获使用同一请求、帧槽和序列号。每个样本表示该帧创建时已经发布的策略状态。
+
按完成帧采样策略状态策略状态与 Scene Taskflow 观察共同绑定到完成帧；FFmpeg Sink 使用自己的媒体 Taskflow，并在媒体节点完成后释放策略帧。
- {response?.requested ? `${response.captured}/${response.requested} 帧${response.complete ? " · 已完成" : ` · 还需 ${response.remaining} 帧`}` : "只在手动请求后采集；manual 模式需要手动生成相应帧数。"}
+ {response?.requested ? `${response.captured}/${response.requested} 帧${response.complete ? " · 已完成" : ` · 还需 ${response.remaining} 帧`}` : "只在请求后采集；可手动生成帧加快采样。"}
{capture_error ?

{capture_error}

: null} {state && frame ? <>
- 策略代次 {state.generation} · 观察 {duration(state.observation.duration_ms)}
+ 累计完成 {state.completed_frames.toLocaleString("zh-CN")} 帧 · 调度 {state.timer_ticks.toLocaleString("zh-CN")} 次
-
实际完成帧率{rate(state.throughput.completion_rate_fps)}由完成帧间隔计算；评价最终吞吐
-
提交速率{rate(state.throughput.submission_rate_fps)}进入 Scene 的帧速率
-
目标达成率{state.configuration.mode === "fixed_rate" ? percent(state.throughput.target_achievement_ratio) : "不适用"}实际完成 FPS ÷ fixed_rate 目标
-
帧间隔抖动{duration(state.throughput.frame_interval_jitter_ms)}完成间隔标准差，越低越稳定
-
请求接受率{percent(state.requests.acceptance_ratio)}{state.requests.accepted}/{state.observation.request_count} 个请求进入准入
-
请求合并率{percent(state.requests.coalescing_ratio)}忙碌期间被 latest 合并的比例
-
帧槽背压率{percent(state.requests.backpressure_ratio)}{state.requests.frame_slot_backpressure} 次物理帧槽耗尽
-
平均 tick 排队{duration(state.latency.average_tick_queue_ms)}请求发出至 Scene 提交
-
平均完成延迟{duration(state.latency.average_completion_ms)}Scene 提交至完成帧退役
-
在途帧{state.observation.active_frame_count}已提交但尚未完成退役
+
有效限速{rate(state.effective_frames_per_second)}用户上限、Scene 产能和 Sink 产能中的最小值
+
Scene 理论产能{state.render_capacity_fps === null ? "未启用" : rate(state.render_capacity_fps)}由 Scene render 滑动统计推导
+
FFmpeg Sink 理论产能{state.send_capacity_fps === null ? "未启用" : rate(state.send_capacity_fps)}包含媒体 Taskflow 到策略 completion 的耗时
+
Scene 平均耗时{duration_ns(state.render_time.average_ns)}P95 {duration_ns(state.render_time.p95_ns)} · {state.render_time.sample_count} 样本
+
Sink 平均耗时{duration_ns(state.send_time.average_ns)}P95 {duration_ns(state.send_time.p95_ns)} · {state.send_time.sample_count} 样本
+
端到端平均耗时{duration_ns(state.end_to_end_time.average_ns)}P95 {duration_ns(state.end_to_end_time.p95_ns)}
+
丢弃调度{state.dropped_timer_ticks.toLocaleString("zh-CN")}三个物理帧槽均在途时丢弃的 tick
+
配置错误{state.configuration_error ?? "无"}最近一次动态配置检查结果
+
Scene 提交错误{state.render_submission_error ?? "无"}下一次成功提交后清空
-
- {response?.frames.map((sample, index) => { const policy = sample.frame_policy; return policy ? set_frame_index(index)}> : null;})} +
帧来源完成 FPS间隔抖动排队完成延迟合并 / 背压
#{sample.sequence}{sample.request_source ?? "unspecified"}{rate(policy.throughput.completion_rate_fps)}{duration(policy.throughput.latest_frame_interval_ms)}{duration(policy.throughput.frame_interval_jitter_ms)}{duration(policy.latency.latest_tick_queue_ms)}{duration(policy.latency.latest_completion_ms)}{policy.requests.coalesced} / {policy.requests.frame_slot_backpressure}
+ {response?.frames.map((sample, index) => { const policy = sample.frame_policy; return policy ? set_frame_index(index)}> : null;})}
帧有效限速Scene 平均Sink 平均端到端 P95累计完成丢弃 tick
#{sample.sequence}{rate(policy.effective_frames_per_second)}{duration_ns(policy.render_time.average_ns)}{duration_ns(policy.send_time.average_ns)}{duration_ns(policy.end_to_end_time.p95_ns)}{policy.completed_frames}{policy.dropped_timer_ticks}
查看该帧绑定的完整 Frame_Policy State
{JSON.stringify(state, null, 2)}
:

尚未采样帧策略状态。

} @@ -1476,10 +1405,11 @@ function taskflow_node_name(name: string, node?: Taskflow_Node_Trace) { if (name === "render_3d.backend.collect") return "Datoviz · GPU 完成后读回"; if (name === "render_3d.frame.retire") return "3D Scene · 完成帧退役"; if (name === "plot.frame.publish") return "Plot · 发布完成帧"; - if (name === "gallery.sample.capture") return "Gallery · 30 FPS 最近帧采样"; + if (name === "gallery.atlas.compose") return "Gallery Sink · 合成 Plot 完成帧"; if (name === "FFmpeg.H264.encode") return "FFmpeg · Taskflow CPU H.264 编码"; if (name === "gallery.h264.websocket.publish") return "Drogon · H.264 WebSocket 发布"; - if (name === "gallery.sample.complete") return "Gallery · 采样完成"; + if (name === "frame.policy.sink.complete") return "Frame Policy · Sink 完成"; + if (name === "gallery.media.complete") return "Gallery Sink · 媒体帧退役"; if (name === "taskflow.executor.finalize") return "Taskflow · Executor topology 收尾"; if (name === "taskflow.runtime.bookkeeping") return "Taskflow · 运行计数更新"; if (name === "taskflow.runtime.publish_state") return "Taskflow · 运行时状态发布"; @@ -2709,52 +2639,7 @@ function Taskflow_Frame_Pane({plot, components}: {plot: Plot; components: Compon } function Taskflow_Runtime_Pane() { - const [state, set_state] = useState(null); - const [error, set_error] = useState(""); - const load = useCallback(async () => { - const request = await fetch("/taskflow/diagnostics", { - signal: AbortSignal.timeout(3000)}); - const value = await request.json() as Taskflow_Runtime_State & {error?: string}; - if (!request.ok) throw new Error(value.error ?? "读取 Taskflow 总体状态失败"); - set_state(value); set_error(""); - }, []); - useEffect(() => { - let stopped = false; - let timer = 0; - const poll = async () => { - try { await load(); } - catch (failure) { set_error(failure instanceof Error ? failure.message : "读取 Taskflow 总体状态失败"); } - if (!stopped) timer = window.setTimeout(() => void poll(), 1000); - }; - void poll(); - return () => { stopped = true; window.clearTimeout(timer); }; - }, [load]); - return
全局执行域

Taskflow 总体观测

每秒读取一次累计 Observer 状态；逐帧拓扑捕获在各图的“Taskflow 帧分析”页按需开启。

-
{error ?

{error}

: null}{state ? <> -
-
Worker
{state.worker_count}
任务驻留墙钟
{state.worker_utilization.toFixed(1)}%
-
线程 CPU 占用
{state.worker_cpu_utilization.toFixed(1)}%
-
活跃 Worker
{state.active_workers}/{state.worker_count}
活跃任务
{state.active_tasks}
-
活跃 Topology
{state.active_topologies}
活跃 Taskflow
{state.active_taskflows}
-
累计任务
{state.observed_tasks.toLocaleString("zh-CN")}
失败 Taskflow
{state.failed_taskflows}
-
队列峰值
{state.peak_worker_queue_size}
最长任务
{nanoseconds(state.longest_task.time_ns)}
-
协作让出
{state.cooperative_wait_count.toLocaleString("zh-CN")} 次 · {nanoseconds(state.cooperative_wait_time_ns)}
-
-

任务驻留墙钟统计任务函数的生命周期，显式协作让出已经扣除，但 Windows 抢占、等待 CPU 和任务内部未声明的阻塞仍在其中。线程 CPU 才是 Worker 真正获得的处理器时间。两者差值不能直接叫浪费：要结合逐帧节点、协作等待和队列判断。当前 32 个 Worker 同时争用处理器时，高驻留、低 CPU 主要表示调度排队。

-
Worker 任务驻留、实际 CPU、协作让出与本地队列累计值，不在 GET 时重新计算任务样本。
{state.workers.map(worker =>
-
Worker {worker.id}驻留 {worker.utilization.toFixed(1)}% · CPU {worker.cpu_utilization.toFixed(1)}%
-
-
任务
{worker.task_count}
进入队列/峰值
{worker.active_task.time_ns ? worker.entry_queue_size : "--"}/{worker.peak_queue_size}
-
最长
{nanoseconds(worker.max_task_time_ns)}
活跃持续
{worker.active_task.time_ns ? nanoseconds(worker.active_task.time_ns) : "空闲"}
-
线程总 CPU
{nanoseconds(worker.cpu_time_ns)}
任务驻留墙钟
{nanoseconds(worker.busy_time_ns)}
-
协作让出
{worker.cooperative_wait_count.toLocaleString("zh-CN")} 次
让出墙钟
{nanoseconds(worker.cooperative_wait_time_ns)}
- {worker.active_task.time_ns ? {worker.active_task.type} · {worker.active_task.native_id} : null} -
)}
-
Taskflow 原生任务类型按 Observer TaskType 累计执行次数与耗时。
- {state.task_types.map(type => - - )}
类型次数累计最短最长
{type.name}{type.count.toLocaleString("zh-CN")}{nanoseconds(type.total_time_ns)}{nanoseconds(type.min_time_ns)}{nanoseconds(type.max_time_ns)}
- :
正在读取 Taskflow 总体状态
}
; + return
按需执行域

Taskflow 观察入口

当前 Runtime 不发布全局累计状态；避免维护第二份 Worker 状态源。请选择 Plot，在“Taskflow 帧分析”中按需捕获 Scene 与独立 FFmpeg Sink DAG。

; } const Plot_Card = memo(function Plot_Card({plot, selected, policy, gallery, on_policy, on_select}: { @@ -2767,29 +2652,18 @@ const Plot_Card = memo(function Plot_Card({plot, selected, policy, gallery, on_p const source = gallery.layout?.plots[plot.id] ?? null; const tile_width = gallery.layout?.tile_width ?? 720; const tile_height = gallery.layout?.tile_height ?? 420; - const graph = use_plot_stream(plot, surface_ref, gallery.presentation, tile_width, tile_height); + const graph = use_plot_stream(plot, surface_ref, tile_width, tile_height); const status: Stream_Status = graph.status === "OFFLINE" || gallery.status === "OFFLINE" ? "OFFLINE" : graph.status === "LIVE" && gallery.status === "LIVE" ? "LIVE" : "CONNECTING"; const metrics = graph.metrics; const backend_policy = graph.server_diagnostics?.frame_policy ?? null; const source_transport = gallery.transport?.sources[plot.id] ?? null; - const event_statistics = Object.values( - graph.server_diagnostics?.input_statistics ?? {}); - const event_statistic_maximum = ( - key: "queue_wait_ms" | "dispatch_ms" | "total_ms", - member: "latest" | "p95") => event_statistics.reduce( - (maximum, event) => Math.max(maximum, - event[key]?.[member] ?? 0), 0); const browser_inputs = Object.values(graph.browser_input_statistics); const browser_input_latest_ms = browser_inputs.reduce( (maximum, input) => Math.max(maximum, input.latest_ms), 0); const browser_input_buffered_bytes = browser_inputs.reduce( (maximum, input) => Math.max( maximum, input.websocket_buffered_bytes), 0); - const event_queue_p95_ms = event_statistic_maximum( - "queue_wait_ms", "p95"); - const event_dispatch_p95_ms = event_statistic_maximum( - "dispatch_ms", "p95"); useEffect(() => { const canvas = pixel_canvas_ref.current; if (!policy.visible || !gallery.video_group || !source || !canvas) @@ -2797,7 +2671,7 @@ const Plot_Card = memo(function Plot_Card({plot, selected, policy, gallery, on_p return gallery.video_group.register(plot.id, canvas); }, [policy.visible, gallery.video_group, plot.id, source?.column, source?.row]); - const generated_time = metrics ? new Date(metrics.generated_time_unix_ms).toLocaleTimeString("zh-CN", {hour: "2-digit", minute: "2-digit", second: "2-digit", hour12: false, fractionalSecondDigits: 3}) : "尚未生成帧"; + const generated_time = metrics ? new Date(metrics.sampled_time_unix_ms).toLocaleTimeString("zh-CN", {hour: "2-digit", minute: "2-digit", second: "2-digit", hour12: false, fractionalSecondDigits: 3}) : "尚未采样"; return
on_select(plot)} @@ -2805,10 +2679,10 @@ const Plot_Card = memo(function Plot_Card({plot, selected, policy, gallery, on_p
⠿
绘图组件 · {plot.dimension}

{plot.title}

-
{{IDLE: "已停止", CONNECTING: "重连中", LIVE: "实时", OFFLINE: "已离线"}[status]}{metrics?.delivery === "diagnostics" ? "无像素传输" : metrics ? enum_label(metrics.pacing_mode) : "等待策略"}
+
{{IDLE: "已停止", CONNECTING: "重连中", LIVE: "实时", OFFLINE: "已离线"}[status]}{metrics ? `限速 ${metrics.frame_rate_fps.toFixed(1)} FPS` : "等待策略"}
- 后台完成 {backend_policy ? backend_policy.throughput.completion_rate_fps.toFixed(1) : source_transport ? source_transport.logical_completion_rate_fps.toFixed(1) : "--.-"} FPS - 后台延迟 {backend_policy ? backend_policy.latency.latest_completion_ms.toFixed(1) : "--.-"} ms + 后台完成 {source_transport ? source_transport.logical_completion_rate_fps.toFixed(1) : "--.-"} FPS + 后台延迟 {backend_policy ? (backend_policy.end_to_end_time.average_ns / 1_000_000).toFixed(1) : "--.-"} ms 画面产出 {source_transport ? source_transport.rendered_frame_rate_fps.toFixed(1) : metrics ? metrics.frame_rate_fps.toFixed(1) : "--.-"} FPS 策略后像素 {gallery.transport ? gallery.transport.received_frame_rate_fps.toFixed(1) : "--.-"} FPS 媒体处理 {gallery.transport ? gallery.transport.processed_frame_rate_fps.toFixed(1) : "--.-"} FPS @@ -2818,8 +2692,6 @@ const Plot_Card = memo(function Plot_Card({plot, selected, policy, gallery, on_p H.264 编码 {gallery.transport ? gallery.transport.encode_average_ms.toFixed(1) : "--.-"} ms 画面落后 {source_transport?.has_rendered_frame ? source_transport.frame_lag : "--"} 帧 浏览器输入 {browser_inputs.length ? browser_input_latest_ms.toFixed(1) : "--.-"} ms - 事件排队 P95 {event_statistics.length ? event_queue_p95_ms.toFixed(1) : "--.-"} ms - 事件分发 P95 {event_statistics.length ? event_dispatch_p95_ms.toFixed(1) : "--.-"} ms