From 20be348a6a783d34385c1445f4faf82295a30f12 Mon Sep 17 00:00:00 2001 From: wyc <1104749580@qq.com> Date: Sat, 29 Aug 2026 20:24:11 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8E=9F=E5=A7=8B=E5=83=8F=E7=B4=A0=E5=A4=AA?= =?UTF-8?q?=E5=8D=A1=20=E8=B5=B0ffmpeg?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Project_detail_specification.md | 7 +- mcp/main.cmake | 24 + mcp/tests/Control_Path_Benchmarks.cpp | 4 +- mcp/tests/Event_Latency_Benchmarks.cpp | 554 ++++++++++++++++++ render_3D/render_3D/scene/Render_Scene_3D.ipp | 8 +- web_server/src/Gallery_Video_Stream.ipp | 84 --- web_server/src/H264_Encoder.hpp | 43 -- web_server/src/Web_Server.cpp | 23 +- .../src/frame_sampling/Frame_Sampler.cpp | 144 ----- .../src/frame_sampling/Frame_Sampler.hpp | 63 -- .../ffmpeg/FFmpeg_Frame_Transport.hpp | 26 - web_server/src/media/Encoded_Video_Frame.hpp | 32 + .../src/{ => media}/Gallery_Video_Stream.cpp | 335 +++++------ .../src/{ => media}/Gallery_Video_Stream.hpp | 4 +- web_server/src/media/Gallery_Video_Stream.ipp | 116 ++++ .../Gallery_Video_WebSocket.cpp} | 57 +- .../Gallery_Video_WebSocket.hpp} | 20 +- .../detail}/FFmpeg_Frame_Transport.cpp | 26 +- .../media/detail/FFmpeg_Frame_Transport.hpp | 36 ++ .../detail/Gallery_Frame_Atlas.cpp | 28 +- .../detail/Gallery_Frame_Atlas.hpp | 5 +- .../src/{ => media/detail}/H264_Encoder.cpp | 14 +- web_server/src/media/detail/H264_Encoder.hpp | 26 + .../tests/Gallery_Frame_Atlas_Tests.cpp | 19 +- .../tests/Gallery_Video_Stream_Tests.cpp | 122 ++++ web_server/tests/H264_Encoder_Tests.cpp | 16 +- webapp_gallery/src/app.tsx | 158 +++-- 27 files changed, 1310 insertions(+), 684 deletions(-) create mode 100644 mcp/tests/Event_Latency_Benchmarks.cpp delete mode 100644 web_server/src/Gallery_Video_Stream.ipp delete mode 100644 web_server/src/H264_Encoder.hpp delete mode 100644 web_server/src/frame_sampling/Frame_Sampler.cpp delete mode 100644 web_server/src/frame_sampling/Frame_Sampler.hpp delete mode 100644 web_server/src/frame_sampling/ffmpeg/FFmpeg_Frame_Transport.hpp create mode 100644 web_server/src/media/Encoded_Video_Frame.hpp rename web_server/src/{ => media}/Gallery_Video_Stream.cpp (73%) rename web_server/src/{ => media}/Gallery_Video_Stream.hpp (96%) create mode 100644 web_server/src/media/Gallery_Video_Stream.ipp rename web_server/src/{Gallery_WebSocket.cpp => media/Gallery_Video_WebSocket.cpp} (86%) rename web_server/src/{Gallery_WebSocket.hpp => media/Gallery_Video_WebSocket.hpp} (70%) rename web_server/src/{frame_sampling/ffmpeg => media/detail}/FFmpeg_Frame_Transport.cpp (58%) create mode 100644 web_server/src/media/detail/FFmpeg_Frame_Transport.hpp rename web_server/src/{ => media}/detail/Gallery_Frame_Atlas.cpp (90%) rename web_server/src/{ => media}/detail/Gallery_Frame_Atlas.hpp (97%) rename web_server/src/{ => media/detail}/H264_Encoder.cpp (96%) create mode 100644 web_server/src/media/detail/H264_Encoder.hpp create mode 100644 web_server/tests/Gallery_Video_Stream_Tests.cpp diff --git a/Project_detail_specification.md b/Project_detail_specification.md index 1d3f68a..61e618e 100644 --- a/Project_detail_specification.md +++ b/Project_detail_specification.md @@ -29,9 +29,10 @@ 层只向帧写入固定语义的单调时间点与原始耗时,不保存平均值、分位数或波动等衍生统计。 * Plot 使用服务端帧时钟调用 `Scene::render(frame*)`;`fixed_rate` 只由 Frame_Scheduler 周期时钟驱动,完成回调不得改变其 deadline。`maximum_rate` 是唯一允许完成回调在释放 Plot 准入后异步投递下一帧请求的特殊模式;回调不得直接重入 `render()` - 、不得同步等待,并且物理帧槽耗尽时只能由帧退役机制解除背压。Scene 回调发布 2D 原生 BGRA 或 3D 原生 RGBA;Gallery - 像素流按固定时钟采样各 Plot 的 latest,只把本周期真实变化的 tile 通过 Drogon WebSocket 原始像素批协议发送;服务端不组装图集,浏览器由各 Plot 的 WebGL Canvas 直接上传自身 tile。每个连接只允许一批在途, - 浏览器实际提交 Canvas 后 ACK,服务端只发送当时最新帧;Plot 输入和控制也统一使用 Drogon WebSocket。 + 、不得同步等待,并且物理帧槽耗尽时只能由帧退役机制解除背压。Scene 回调发布 2D 原生 BGRA 或 3D 原生 RGBA;每个 Plot + 只服从自身 Frame_Policy。Gallery 仅按 Plot 完成帧的到达顺序更新对应图集槽位,不拥有采样时钟、帧准入、合并或丢帧策略; + 每次槽位更新均在唯一串行 Taskflow 媒体 DAG 中组成图集并由 FFmpeg/libx264 编码 H.264 Annex-B。Drogon WebSocket 只传输压缩 + access unit,不得用在途窗口反向跳过已通过 Plot 策略的帧;浏览器由 WebCodecs 解码后按槽位提交各 Plot Canvas。Plot 输入和控制也统一使用 Drogon WebSocket。 * 帧策略及其时钟实现统一归属 `kernel/src/kernel/Frame_Policy`。配置变更和运行事实通过 MPMC 事件流进入唯一 consumer,由 `double_buffer/model.hpp` 发布 `Frame_Policy::State`;查询只读取已发布 State。手动帧捕获必须把帧创建边界的策略 State 绑定到该帧并沿现有帧捕获协议返回,禁止另建策略历史、轮询累加器或重复诊断字段。 diff --git a/mcp/main.cmake b/mcp/main.cmake index 977bbec..88c8a5a 100644 --- a/mcp/main.cmake +++ b/mcp/main.cmake @@ -111,6 +111,30 @@ if (Aethera_BUILD_TESTS) --benchmark_report_aggregates_only=true DEPENDS Aethera_MCP_Benchmarks USES_TERMINAL) + add_executable(Aethera_MCP_Event_Latency_Benchmarks + "${CMAKE_CURRENT_LIST_DIR}/tests/Event_Latency_Benchmarks.cpp") + target_link_libraries(Aethera_MCP_Event_Latency_Benchmarks PRIVATE + Aethera_MCP_Core + benchmark::benchmark + TBB::tbbmalloc_proxy) + renderive_stage_render_3D_runtime(Aethera_MCP_Event_Latency_Benchmarks) + if (MSVC) + target_compile_options(Aethera_MCP_Event_Latency_Benchmarks PRIVATE + /utf-8 /bigobj) + target_link_options(Aethera_MCP_Event_Latency_Benchmarks PRIVATE + "/INCLUDE:__TBB_malloc_proxy") + add_custom_command(TARGET Aethera_MCP_Event_Latency_Benchmarks POST_BUILD + COMMAND "${CMAKE_COMMAND}" -E copy_if_different + "$" + "$" + "$" + VERBATIM) + endif () + add_custom_target(Aethera_MCP_event_latency_benchmark + COMMAND Aethera_MCP_Event_Latency_Benchmarks + --aethera_event_samples=12 + DEPENDS Aethera_MCP_Event_Latency_Benchmarks + USES_TERMINAL) add_custom_target(Aethera_MCP_check COMMAND "${CMAKE_CTEST_COMMAND}" --test-dir "${CMAKE_BINARY_DIR}" diff --git a/mcp/tests/Control_Path_Benchmarks.cpp b/mcp/tests/Control_Path_Benchmarks.cpp index f1e7baa..71cbac3 100644 --- a/mcp/tests/Control_Path_Benchmarks.cpp +++ b/mcp/tests/Control_Path_Benchmarks.cpp @@ -98,8 +98,8 @@ void append_dimension(web::Plot_Dimension dimension) { [[nodiscard]] Parse_Benchmark_Arguments_Result parse_arguments( int& argc, char** argv) { configuration = {}; - if (!select_plots("3d")) { - configuration_error = "the Gallery has no 3D plots"; + if (!select_plots("all")) { + configuration_error = "the Gallery has no plots"; return Parse_Benchmark_Arguments_Result::invalid_argument; } int retained_count{1}; diff --git a/mcp/tests/Event_Latency_Benchmarks.cpp b/mcp/tests/Event_Latency_Benchmarks.cpp new file mode 100644 index 0000000..17d4d98 --- /dev/null +++ b/mcp/tests/Event_Latency_Benchmarks.cpp @@ -0,0 +1,554 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace aethera::mcp::event_latency_benchmarks { +namespace { + +using Clock = std::chrono::steady_clock; +inline constexpr auto event_types = magic_enum::enum_values(); +inline constexpr std::size_t event_count = event_types.size(); + +struct Configuration { + std::uint32_t width{720}; + std::uint32_t height{420}; + std::uint32_t samples{12}; +}; + +struct Summary { + double p50{}; + double p95{}; + double p99{}; + double maximum{}; +}; + +struct Distribution { + std::vector samples{}; + + void add(double value) { samples.push_back(value); } + + [[nodiscard]] Summary summarize() const { + if (samples.empty()) return {}; + auto ordered = samples; + std::ranges::sort(ordered); + const auto percentile = [&](double probability) { + const auto rank = probability * + static_cast(ordered.size() - 1U); + const auto lower = static_cast(rank); + const auto upper = std::min(lower + 1U, ordered.size() - 1U); + const auto fraction = rank - static_cast(lower); + return ordered[lower] + + (ordered[upper] - ordered[lower]) * fraction; + }; + return { + percentile(0.50), percentile(0.95), percentile(0.99), + ordered.back() + }; + } +}; + +struct Event_Result { + Distribution mcp_accept_ms{}; + Distribution scene_queue_ms{}; + Distribution dispatch_ms{}; + Distribution scene_total_ms{}; + Distribution accepted_to_frame_ms{}; + Distribution end_to_end_ms{}; +}; + +struct Plot_Result { + std::string id{}; + web::Plot_Dimension dimension{web::Plot_Dimension::two_d}; + std::array events{}; +}; + +struct Plot_Probe { + std::atomic_uint64_t expected_correlation{}; + std::atomic_uint64_t completed_steady_ns{}; +}; + +struct Active_Plot { + std::shared_ptr plot{}; + web::Plot::Stream_Id stream{}; + std::shared_ptr probe{}; +}; + +struct Batch_Timing { + std::array submitted_ns{}; + std::array accepted_ns{}; +}; + +Configuration configuration{}; +std::vector published_results{}; + +[[nodiscard]] std::uint64_t steady_time_ns() noexcept { + return static_cast( + std::chrono::duration_cast( + Clock::now().time_since_epoch()).count()); +} + +[[nodiscard]] double elapsed_ms(std::uint64_t begin, + std::uint64_t end) noexcept { + return end >= begin + ? static_cast(end - begin) / 1'000'000.0 : 0.0; +} + +[[nodiscard]] web::Plot_Input_Event make_event( + Event_Type type, std::size_t plot_index, std::uint64_t batch) { + web::Plot_Input_Event event{}; + event.type = type; + event.time_milliseconds = static_cast(steady_time_ns()) / + 1'000'000.0; + event.position = { + 120.0 + static_cast((batch * 13U + plot_index * 7U) % 400U), + 80.0 + static_cast((batch * 11U + plot_index * 5U) % 240U)}; + event.global_position = event.position; + event.button = Mouse_Button::left; + event.buttons = type == Event_Type::pointer_release ? 0U : 1U; + event.modifiers = Keyboard_Modifier::control; + event.pixel_delta_x = 3.0; + event.pixel_delta_y = (batch & 1U) == 0U ? 120.0 : -120.0; + event.angle_delta_x = event.pixel_delta_x; + event.angle_delta_y = event.pixel_delta_y; + event.key = Key::space; + event.native_key = 32; + event.auto_repeat = (batch & 1U) != 0U; + return event; +} + +[[nodiscard]] const nlohmann::json& event_statistic( + const nlohmann::json& diagnostics, Event_Type type, + std::string_view statistic) { + const auto type_name = std::string{magic_enum::enum_name(type)}; + const auto& input = diagnostics.at("input_statistics"); + const auto found_type = input.find(type_name); + if (found_type == input.end()) + throw std::runtime_error( + "completed frame did not publish input statistics for " + + type_name); + const auto found_statistic = found_type->find(std::string{statistic}); + if (found_statistic == found_type->end()) + throw std::runtime_error( + "completed frame did not publish " + std::string{statistic} + + " for " + type_name); + return *found_statistic; +} + +void configure_manual_backend_plot(const std::shared_ptr& plot) { + const auto manual = plot->write_prop( + "frame-analysis", "pacing_mode", "manual"); + if (!manual.value("success", false)) + throw std::runtime_error("failed to select manual frame pacing"); + const auto diagnostics_only = plot->write_prop( + "frame-analysis", "pixel_delivery_enabled", false); + if (!diagnostics_only.value("success", false)) + throw std::runtime_error("failed to disable backend pixel readback"); +} + +[[nodiscard]] std::uint64_t correlation_id( + std::uint64_t batch, std::size_t plot_index) noexcept { + return (batch + 1U) * 1'000U + + static_cast(plot_index + 1U); +} + +void add_distribution(Distribution& destination, + const Distribution& source) { + destination.samples.insert(destination.samples.end(), + source.samples.begin(), source.samples.end()); +} + +class Event_Latency_Reporter final : public benchmark::ConsoleReporter { +public: + bool ReportContext(const Context& context) override { + return ConsoleReporter::ReportContext(context); + } + + void ReportRuns(const std::vector& reports) override { + auto& output = GetOutputStream(); + for (const auto& report : reports) { + output << report.benchmark_name(); + if (report.skipped != benchmark::internal::NotSkipped) + output << " ERROR: " << report.skip_message << '\n'; + else + output << " elapsed=" << report.real_accumulated_time + << " s\n"; + } + if (published_results.empty()) return; + const auto flags = output.flags(); + const auto precision = output.precision(); + output << std::fixed << std::setprecision(3) + << "\nPure backend event timeline (ms)\n" + << "MCP = emit -> aethera_plot_input accepted; " + "Queue/Dispatch = authoritative Scene Event timing; " + "Frame = accepted -> correlated backend frame complete.\n\n"; + + output << std::left << std::setw(24) << "Plot" + << std::setw(6) << "Dim" + << std::setw(18) << "Worst event" + << std::right << std::setw(11) << "MCP P95" + << std::setw(11) << "Queue P95" + << std::setw(11) << "Disp P95" + << std::setw(11) << "Frame P95" + << std::setw(11) << "E2E P95" << '\n'; + for (const auto& plot : published_results) { + std::size_t worst_index{}; + double worst_e2e{}; + for (std::size_t index = 0; index < event_count; ++index) { + const auto p95 = plot.events[index].end_to_end_ms.summarize().p95; + if (p95 <= worst_e2e) continue; + worst_e2e = p95; + worst_index = index; + } + const auto& event = plot.events[worst_index]; + output << std::left << std::setw(24) << plot.id + << std::setw(6) << web::plot_dimension_name(plot.dimension) + << std::setw(18) + << magic_enum::enum_name(event_types[worst_index]) + << std::right + << std::setw(11) << event.mcp_accept_ms.summarize().p95 + << std::setw(11) << event.scene_queue_ms.summarize().p95 + << std::setw(11) << event.dispatch_ms.summarize().p95 + << std::setw(11) + << event.accepted_to_frame_ms.summarize().p95 + << std::setw(11) << worst_e2e << '\n'; + } + + output << "\nAll plots by event type\n" + << std::left << std::setw(18) << "Event" + << std::right << std::setw(11) << "MCP P50" + << std::setw(11) << "MCP P95" + << std::setw(11) << "Queue P95" + << std::setw(11) << "Disp P95" + << std::setw(11) << "Scene P95" + << std::setw(11) << "Frame P95" + << std::setw(11) << "E2E P50" + << std::setw(11) << "E2E P95" + << std::setw(11) << "E2E P99" << '\n'; + for (std::size_t event_index = 0; event_index < event_count; + ++event_index) { + Event_Result aggregate{}; + for (const auto& plot : published_results) { + const auto& event = plot.events[event_index]; + add_distribution(aggregate.mcp_accept_ms, event.mcp_accept_ms); + add_distribution(aggregate.scene_queue_ms, event.scene_queue_ms); + add_distribution(aggregate.dispatch_ms, event.dispatch_ms); + add_distribution(aggregate.scene_total_ms, event.scene_total_ms); + add_distribution( + aggregate.accepted_to_frame_ms, + event.accepted_to_frame_ms); + add_distribution(aggregate.end_to_end_ms, event.end_to_end_ms); + } + const auto mcp = aggregate.mcp_accept_ms.summarize(); + const auto queue = aggregate.scene_queue_ms.summarize(); + const auto dispatch = aggregate.dispatch_ms.summarize(); + const auto scene = aggregate.scene_total_ms.summarize(); + const auto frame = aggregate.accepted_to_frame_ms.summarize(); + const auto end_to_end = aggregate.end_to_end_ms.summarize(); + output << std::left << std::setw(18) + << magic_enum::enum_name(event_types[event_index]) + << std::right << std::setw(11) << mcp.p50 + << std::setw(11) << mcp.p95 + << std::setw(11) << queue.p95 + << std::setw(11) << dispatch.p95 + << std::setw(11) << scene.p95 + << std::setw(11) << frame.p95 + << std::setw(11) << end_to_end.p50 + << std::setw(11) << end_to_end.p95 + << std::setw(11) << end_to_end.p99 << '\n'; + } + output.flags(flags); + output.precision(precision); + } +}; + +void run_event_latency(benchmark::State& state) { + published_results.clear(); + auto service = Control_Service::create(); + const auto definitions = web::gallery_plot_definitions(); + std::vector active; + active.reserve(definitions.size()); + published_results.reserve(definitions.size()); + + try { + for (const auto& definition : definitions) { + auto plot = service->find_plot(definition.id); + if (!plot) + throw std::runtime_error( + "Gallery plot is unavailable: " + + std::string{definition.id}); + configure_manual_backend_plot(plot); + auto probe = std::make_shared(); + active.push_back({std::move(plot), 0, std::move(probe)}); + published_results.push_back({ + std::string{definition.id}, definition.dimension, {}}); + } + + std::string configuration_failure; + Task_Graph configuration_barrier{"mcp.event-latency.configure"}; + configuration_barrier.add("await.manual-policy", [&] { + const auto deadline = Clock::now() + std::chrono::seconds{10}; + Task_Graph::corun_until([&] { + const bool ready = std::ranges::all_of( + active, [](const Active_Plot& item) { + const auto diagnostics = item.plot->diagnostics(); + const auto& policy = diagnostics.at("frame_policy"). + at("configuration"); + return policy.at("mode") == "manual" && + !policy.at("pixel_delivery_enabled").get(); + }); + if (ready) return true; + if (Clock::now() < deadline) return false; + configuration_failure = + "timed out applying manual backend frame policy"; + return true; + }); + }); + aethera::detail::run_taskflow(configuration_barrier); + if (!configuration_failure.empty()) + throw std::runtime_error(configuration_failure); + + for (auto& item : active) { + const auto probe = item.probe; + item.stream = item.plot->subscribe( + [probe](std::shared_ptr frame) { + if (!frame || !frame->pixels) return; + const auto expected = probe->expected_correlation.load( + std::memory_order_acquire); + if (expected == 0 || + frame->pixels->correlation_id != expected) + return; + 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)); + }); + item.plot->configure_stream( + item.stream, configuration.width, configuration.height); + } + + for (auto& item : active) item.plot->reset_diagnostics(); + std::vector batch_timing(active.size()); + std::string failure; + Task_Graph coordinator{"mcp.event-latency"}; + coordinator.add("measure.all-plots", [&] { + for (std::uint64_t batch = 0; batch < configuration.samples; + ++batch) { + const auto deadline = Clock::now() + std::chrono::seconds{30}; + for (std::size_t plot_index = 0; + plot_index < active.size(); ++plot_index) { + auto& item = active[plot_index]; + auto& timing = batch_timing[plot_index]; + const auto correlation = correlation_id(batch, plot_index); + item.probe->completed_steady_ns.store( + 0, std::memory_order_relaxed); + item.probe->expected_correlation.store( + correlation, std::memory_order_release); + for (std::size_t event_index = 0; + event_index < event_count; ++event_index) { + auto request = Plot_Input_Request{ + .plot = published_results[plot_index].id, + .event = make_event( + event_types[event_index], plot_index, batch) + }; + timing.submitted_ns[event_index] = steady_time_ns(); + const auto result = service->call_tool( + "aethera_plot_input", + encode_protocol_value(request)); + timing.accepted_ns[event_index] = steady_time_ns(); + if (result.result != Tool_Call_Result::ok) { + failure = "aethera_plot_input rejected " + + published_results[plot_index].id + "/" + + std::string{magic_enum::enum_name( + event_types[event_index])} + ": " + + result.message; + return; + } + } + item.plot->schedule_render(web::Plot_Render_Tick{ + .issued_at = Clock::now(), + .sequence = correlation, + .time_milliseconds = + static_cast(steady_time_ns()) / 1'000'000.0, + .width = configuration.width, + .height = configuration.height, + .source = Frame_Request_Source::immediate + }); + } + + Task_Graph::corun_until([&] { + const bool complete = std::ranges::all_of( + active, [](const Active_Plot& item) { + return item.probe->completed_steady_ns.load( + std::memory_order_acquire) != 0; + }); + if (complete) return true; + if (Clock::now() < deadline) return false; + failure = "timed out waiting for a correlated backend frame"; + return true; + }); + if (!failure.empty()) return; + + for (std::size_t plot_index = 0; + plot_index < active.size(); ++plot_index) { + const auto completed = active[plot_index].probe-> + completed_steady_ns.load(std::memory_order_acquire); + const auto diagnostics = active[plot_index].plot->diagnostics(); + for (std::size_t event_index = 0; + event_index < event_count; ++event_index) { + auto& result = published_results[plot_index]. + events[event_index]; + const auto& timing = batch_timing[plot_index]; + result.mcp_accept_ms.add(elapsed_ms( + timing.submitted_ns[event_index], + timing.accepted_ns[event_index])); + result.accepted_to_frame_ms.add(elapsed_ms( + timing.accepted_ns[event_index], completed)); + result.end_to_end_ms.add(elapsed_ms( + timing.submitted_ns[event_index], completed)); + const auto type = event_types[event_index]; + result.scene_queue_ms.add(event_statistic( + diagnostics, type, "queue_wait_ms").at("latest"). + get()); + result.dispatch_ms.add(event_statistic( + diagnostics, type, "dispatch_ms").at("latest"). + get()); + result.scene_total_ms.add(event_statistic( + diagnostics, type, "total_ms").at("latest"). + get()); + } + active[plot_index].probe->expected_correlation.store( + 0, std::memory_order_release); + } + } + }); + + for ([[maybe_unused]] auto iteration : state) + aethera::detail::run_taskflow(coordinator); + if (!failure.empty()) state.SkipWithError(failure); + + Distribution all_end_to_end; + for (const auto& plot : published_results) { + Distribution plot_end_to_end; + for (std::size_t event_index = 0; + event_index < event_count; ++event_index) { + const auto event_name = std::string{ + magic_enum::enum_name(event_types[event_index])}; + const auto& event = plot.events[event_index]; + add_distribution(plot_end_to_end, event.end_to_end_ms); + add_distribution(all_end_to_end, event.end_to_end_ms); + const auto prefix = plot.id + "/" + event_name + "/"; + state.counters[prefix + "queue_p95_ms"] = + event.scene_queue_ms.summarize().p95; + state.counters[prefix + "dispatch_p95_ms"] = + event.dispatch_ms.summarize().p95; + state.counters[prefix + "e2e_p95_ms"] = + event.end_to_end_ms.summarize().p95; + } + state.counters[plot.id + "/e2e_p95_ms"] = + plot_end_to_end.summarize().p95; + } + const auto end_to_end = all_end_to_end.summarize(); + state.counters["all/e2e_p50_ms"] = end_to_end.p50; + state.counters["all/e2e_p95_ms"] = end_to_end.p95; + state.counters["all/e2e_p99_ms"] = end_to_end.p99; + state.SetItemsProcessed(static_cast( + configuration.samples * active.size() * event_count)); + } + catch (const std::exception& error) { + state.SkipWithError(error.what()); + } + + for (auto& item : active) item.plot->unsubscribe(item.stream); +} + +BENCHMARK(run_event_latency)->Iterations(1)->UseRealTime(); + +[[nodiscard]] bool parse_unsigned(std::string_view value, + std::uint32_t& output) { + const auto* begin = value.data(); + const auto* end = begin + value.size(); + const auto parsed = std::from_chars(begin, end, output); + return parsed.ec == std::errc{} && parsed.ptr == end; +} + +[[nodiscard]] bool configure(int& argc, char** argv) { + int retained{1}; + for (int index = 1; index < argc; ++index) { + const std::string_view argument{argv[index]}; + constexpr std::string_view samples_prefix{ + "--aethera_event_samples="}; + constexpr std::string_view size_prefix{"--aethera_size="}; + if (argument.starts_with(samples_prefix)) { + if (!parse_unsigned(argument.substr(samples_prefix.size()), + configuration.samples) || + configuration.samples == 0 || configuration.samples > 600) { + std::cerr << "--aethera_event_samples must be in [1, 600]\n"; + return false; + } + continue; + } + if (argument.starts_with(size_prefix)) { + const auto value = argument.substr(size_prefix.size()); + const auto separator = value.find('x'); + if (separator == std::string_view::npos || + !parse_unsigned(value.substr(0, separator), + configuration.width) || + !parse_unsigned(value.substr(separator + 1U), + configuration.height) || + configuration.width == 0 || configuration.height == 0) { + std::cerr << "--aethera_size requires WIDTHxHEIGHT\n"; + return false; + } + continue; + } + argv[retained++] = argv[index]; + } + argc = retained; + return true; +} + +} // namespace +} // namespace aethera::mcp::event_latency_benchmarks + +int main(int argc, char** argv) { + if (!aethera::mcp::event_latency_benchmarks::configure(argc, argv)) + return 2; + aethera::initialize_runtime({}); + benchmark::Initialize(&argc, argv); + if (benchmark::ReportUnrecognizedArguments(argc, argv)) return 2; + benchmark::AddCustomContext("aethera_plots", "all"); + benchmark::AddCustomContext( + "aethera_event_samples", + std::to_string( + aethera::mcp::event_latency_benchmarks::configuration.samples)); + benchmark::AddCustomContext( + "aethera_size", + std::to_string( + aethera::mcp::event_latency_benchmarks::configuration.width) + + "x" + std::to_string( + aethera::mcp::event_latency_benchmarks::configuration.height)); + aethera::mcp::event_latency_benchmarks::Event_Latency_Reporter reporter; + benchmark::RunSpecifiedBenchmarks(&reporter); + benchmark::Shutdown(); + return 0; +} diff --git a/render_3D/render_3D/scene/Render_Scene_3D.ipp b/render_3D/render_3D/scene/Render_Scene_3D.ipp index ce5cc24..ed5ef81 100644 --- a/render_3D/render_3D/scene/Render_Scene_3D.ipp +++ b/render_3D/render_3D/scene/Render_Scene_3D.ipp @@ -670,10 +670,12 @@ void Render_Scene_3D::Private::render_datoviz( scene, *visuals, sequence, observe, readback); } if (!prepared) { + frame->mark(Frame_Trace_Marker::event_dispatch_started); for (const auto& event : context->events) { if (event) event->mark_dispatch_started(sequence); dispatch_datoviz(event, scene.viewport); } + frame->mark(Frame_Trace_Marker::event_dispatch_finished); context->events.clear(); frame->mark(Frame_Trace_Marker::backend_prepare_started); prepared = prepare( @@ -1034,9 +1036,7 @@ void Render_Scene_3D::Private::ensure_frame_taskflow(Object* object) { if (!active_context || !active_context->frame) throw std::logic_error("3D frame DAG lost its active frame"); const auto frame = active_context->frame; frame->mark(Frame_Trace_Marker::scene_render_started); - frame->mark(Frame_Trace_Marker::event_dispatch_started); active_context->events = take_events(object); - frame->mark(Frame_Trace_Marker::event_dispatch_finished); camera_component->advance_object(); axes_component->advance_object(); auto context = std::static_pointer_cast< @@ -1070,11 +1070,11 @@ void Render_Scene_3D::Private::ensure_frame_taskflow(Object* object) { if (!detail::prepared_visual_batch_complete(*context->visuals)) throw std::logic_error( "render scene published an incomplete Visual batch"); + render_datoviz(object, execution, context->visuals, + context->parameters); auto& state = static_cast( double_buffer::detail::Internal_Access::pending_state(object)); state.event_statistics = event_statistics.state(); - render_datoviz(object, execution, context->visuals, - context->parameters); }); submit.describe("dimension", "3D") .describe("backend", "Datoviz") diff --git a/web_server/src/Gallery_Video_Stream.ipp b/web_server/src/Gallery_Video_Stream.ipp deleted file mode 100644 index 6850de0..0000000 --- a/web_server/src/Gallery_Video_Stream.ipp +++ /dev/null @@ -1,84 +0,0 @@ -#pragma once -#include "frame_sampling/Frame_Sampler.hpp" -#include -#include -#include -#include -#include -#include -namespace aethera::web { -struct Gallery_Video_Stream::Private : Prev_Private { - using Object = Gallery_Video_Stream; - struct Source { - Plot_Entry entry; /* 布局槽位关联的实际 Plot。 */ - Plot::Stream_Id stream{}; /* Plot 完成帧的唯一订阅标识。 */ - }; - struct Consumer { - Stream_Id id{}; /* 当前网页媒体连接的订阅身份。 */ - std::string connection{}; /* 同一浏览器页重连时保持稳定的协议身份。 */ - std::shared_ptr handler{}; /* 发布期间保持回调存活。 */ - std::shared_ptr readiness{}; /* 采样前查询对应传输窗口。 */ - std::shared_ptr diagnostics{}; /* 连接自己的传输诊断来源。 */ - }; - static constexpr std::size_t maximum_consumers{32}; - Object* object{}; /* Def 机制所属最终对象;bind_private_crtp 后有效。 */ - Sliding_Statistics compose_ms{600}; /* 图集采样与变更槽复制统计计算器。 */ - Sliding_Statistics sample_delay_ms{600}; /* 固定 deadline 到实际采样的调度延迟。 */ - Sliding_Statistics encode_ms{600}; /* H.264 编码统计计算器。 */ - Sliding_Statistics publish_ms{600}; /* H.264 WebSocket 发布统计计算器。 */ - std::vector sources{}; /* 已按业务标识排序的稳定图集来源。 */ - std::unique_ptr sampler{}; /* latest 图像与 30 FPS deadline 的唯一状态源。 */ - std::shared_ptr media_graph{}; /* 独立媒体时钟运行的持久 DAG;异步完成回调延长其生命周期。 */ - Frame_Scheduler::Timer sample_timer{}; /* 与任意单张 Plot 完成速率解耦的 30 FPS 控制时钟。 */ - std::atomic_bool media_busy{}; /* 单个媒体 DAG 的 latest-only 准入。 */ - frame_sampling::ffmpeg::FFmpeg_Frame_Transport ffmpeg_transport; /* FFmpeg 子模型的唯一编码上下文。 */ - std::optional active_sample{}; /* 当前 deadline 取得的原生图集视图。 */ - std::shared_ptr active_video{}; /* 当前编码结果不可变共享所有权。 */ - std::atomic_uint64_t next_consumer_id{1}; /* 订阅身份生成器。 */ - std::array>, - maximum_consumers> consumers{}; /* 每个槽位是订阅关系的唯一权威入口;热路径原子查询,不复制集合。 */ - std::atomic_bool stopping{}; /* 关闭开始后拒绝新工作。 */ - std::atomic_uint64_t skipped_sample_ticks{}; /* 未到 deadline 或无可写消费者的采样触发数。 */ - std::atomic_uint64_t sampling_clock_ticks{}; /* 有消费者期间触发的独立媒体时钟 tick。 */ - std::atomic_bool failed{}; /* 首次 Unknown Failure 后停止热路径。 */ - std::optional terminal_failure{}; /* 首次终止失败文本。 */ - std::uint64_t encoded_frame_count{}; /* 累计编码成功帧数。 */ - std::chrono::steady_clock::time_point metric_started{}; /* 当前指标窗口起点。 */ - std::uint64_t metric_encoded_start{}; /* 窗口起点累计编码帧数。 */ - std::uint64_t metric_sample_start{}; /* 窗口起点累计采样帧数。 */ - std::uint64_t metric_clock_start{}; /* 窗口起点累计时钟数。 */ - std::vector metric_completion_starts{}; /* 各 Plot 窗口起点逻辑完成数。 */ - std::vector metric_rendered_starts{}; /* 各 Plot 窗口起点真实画面数。 */ - static constexpr std::size_t maximum_taskflow_trace_frames{120}; - std::atomic_size_t taskflow_trace_remaining{}; /* 尚待捕获的真实媒体 DAG 次数。 */ - std::atomic_uint64_t taskflow_trace_control{}; /* 高 32 位 requested,低 32 位 captured。 */ - std::array>, - 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; - void publish(std::shared_ptr video, - std::string notification) noexcept; - void fail(std::exception_ptr failure) noexcept; - void accept_frame(std::size_t slot, - std::shared_ptr frame); - void update_metrics(const frame_sampling::Sampled_Gallery_Frame& sample, - std::size_t encoded_bytes, - std::optional encoder_backend); - void sample_media_frame(); - void encode_media_frame(); - void publish_video_frame(); - void complete_media_frame(); - [[nodiscard]] bool mark_taskflow_trace(); - void restore_taskflow_trace() noexcept; - void store_taskflow_trace(const Taskflow_Frame_Trace& trace); - [[nodiscard]] nlohmann::json taskflow_trace_response() const; -}; -} diff --git a/web_server/src/H264_Encoder.hpp b/web_server/src/H264_Encoder.hpp deleted file mode 100644 index 488dca9..0000000 --- a/web_server/src/H264_Encoder.hpp +++ /dev/null @@ -1,43 +0,0 @@ -#pragma once -#include -#include -#include -#include -#include -#include -#include -#include -namespace aethera::web { -enum struct Video_Pixel_Layout : std::uint8_t { - bgra, - rgba -}; -enum struct Video_Encoder_Backend : std::uint8_t { - libx264 -}; -[[nodiscard]] std::string_view video_encoder_backend_name( - Video_Encoder_Backend backend) noexcept; -[[nodiscard]] std::string_view h264_profile_level_id() noexcept; -struct Encoded_Video_Frame { - std::vector annex_b{}; /* 带起始码的完整 H.264 access unit,可直接交给 RTP packetizer。 */ - std::chrono::microseconds presentation_time{}; /* 从 Plot 帧时钟起点累计的媒体时间戳。 */ - std::uint64_t sequence{}; /* 对应 Render_Frame 的单调序号。 */ - Video_Encoder_Backend backend{Video_Encoder_Backend::libx264}; /* 实际产生本 access unit 的 FFmpeg 后端。 */ - bool key_frame{}; /* 本 access unit 是否可独立解码。 */ -}; -struct H264_Encoder final { -public: - explicit H264_Encoder(double frame_rate); - ~H264_Encoder(); - H264_Encoder(const H264_Encoder&) = delete; - H264_Encoder& operator=(const H264_Encoder&) = delete; - void request_key_frame(); - [[nodiscard]] std::optional encode( - std::span pixels, std::uint32_t width, - std::uint32_t height, Video_Pixel_Layout layout, - std::uint64_t sequence, std::chrono::microseconds presentation_time); -private: - struct Private; - std::unique_ptr d; -}; -} diff --git a/web_server/src/Web_Server.cpp b/web_server/src/Web_Server.cpp index 62961b5..0f6142b 100644 --- a/web_server/src/Web_Server.cpp +++ b/web_server/src/Web_Server.cpp @@ -1,6 +1,6 @@ #include "Web_Server.hpp" -#include "Gallery_Video_Stream.hpp" -#include "Gallery_WebSocket.hpp" +#include "media/Gallery_Video_Stream.hpp" +#include "media/Gallery_Video_WebSocket.hpp" #include "Graph_WebSocket.hpp" #include #include @@ -20,7 +20,8 @@ namespace aethera::web { namespace { using Gallery_Stream_Map = - std::unordered_map>; + std::unordered_map>; drogon::HttpResponsePtr json_response(nlohmann::json value) { auto response = drogon::HttpResponse::newHttpResponse(); @@ -108,8 +109,8 @@ 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 = mcp::Control_Service::create(); - std::vector gallery_2d; - std::vector gallery_3d; + std::vector gallery_2d; + std::vector gallery_3d; gallery_2d.reserve(8); gallery_3d.reserve(16); for (const auto& definition : gallery_plot_definitions()) { @@ -131,7 +132,7 @@ int run_web_server(std::uint16_t port, const std::filesystem::path& asset_root) std::unordered_map>(); const auto add_media_group = [&gallery_streams, &plot_media, &plot_media_group]( std::string id, - std::vector entries) { + std::vector entries) { if (entries.empty()) return; const auto path = "/ws/gallery?group=" + id; for (const auto& entry : entries) { @@ -139,18 +140,19 @@ int run_web_server(std::uint16_t port, const std::filesystem::path& asset_root) plot_media_group->emplace(entry.id, id); } gallery_streams->emplace( - std::move(id), Gallery_Video_Stream::create(std::move(entries))); + std::move(id), + media::Gallery_Video_Stream::create(std::move(entries))); }; constexpr std::size_t plots_per_media_group{4}; const auto add_media_groups = [&add_media_group]( std::string_view prefix, - std::vector entries) { + std::vector entries) { std::size_t group_index{}; for (std::size_t offset = 0; offset < entries.size(); offset += plots_per_media_group) { const auto end = std::min( entries.size(), offset + plots_per_media_group); - std::vector group( + std::vector group( std::make_move_iterator(entries.begin() + static_cast(offset)), std::make_move_iterator(entries.begin() + @@ -165,7 +167,8 @@ int run_web_server(std::uint16_t port, const std::filesystem::path& asset_root) return control->find_plot(id); }; auto websocket = std::make_shared(resolve_plot); - auto gallery_websocket = std::make_shared( + auto gallery_websocket = + std::make_shared( [gallery_streams](std::string_view id) { const auto found = gallery_streams->find(std::string{id}); return found == gallery_streams->end() ? nullptr : found->second; diff --git a/web_server/src/frame_sampling/Frame_Sampler.cpp b/web_server/src/frame_sampling/Frame_Sampler.cpp deleted file mode 100644 index f98aea4..0000000 --- a/web_server/src/frame_sampling/Frame_Sampler.cpp +++ /dev/null @@ -1,144 +0,0 @@ -#include "Frame_Sampler.hpp" -#include -#include -#include -#include -#include - -namespace aethera::web::frame_sampling { -namespace { -using Clock = std::chrono::steady_clock; - -std::int64_t clock_nanoseconds(Clock::time_point value) noexcept { - return std::chrono::duration_cast( - value.time_since_epoch()).count(); -} -} - -struct Frame_Sampler::Private { - detail::Gallery_Frame_Atlas atlas; /* 各 Plot 最近原生帧与持久图集的唯一状态源。 */ - std::chrono::nanoseconds period{}; /* 固定采样周期;构造后不变。 */ - double frame_rate_fps{}; /* 对外诊断使用的精确目标频率。 */ - std::size_t sampling_source_slot{}; /* 为样本提供页面媒体时间的固定业务来源。 */ - std::atomic_int64_t next_deadline_ns{}; /* 下一次允许生成样本的单调 deadline。 */ - std::atomic_uint64_t accepted_source_frames{}; /* 成功发布到 latest 的源帧计数。 */ - std::atomic_uint64_t rejected_source_frames{}; /* 无效或过期源帧计数。 */ - std::atomic_uint64_t sampled_frames{}; /* 实际采样计数。 */ - std::atomic_uint64_t early_ticks{}; /* deadline 前被合并的触发计数。 */ - std::atomic_uint64_t missed_periods{}; /* 因迟到直接跳过的完整周期数。 */ - - Private(double target_frame_rate_fps, - std::uint32_t tile_width, - std::uint32_t tile_height, - std::uint32_t columns, - std::vector source_ids, - std::size_t source_slot) - : atlas(tile_width, tile_height, columns, std::move(source_ids)), - period(static_cast(std::llround( - 1'000'000'000.0 / target_frame_rate_fps))), - frame_rate_fps(target_frame_rate_fps), - sampling_source_slot(source_slot) {} -}; - -Frame_Sampler::Frame_Sampler( - double frame_rate_fps, std::uint32_t tile_width, - std::uint32_t tile_height, std::uint32_t columns, - std::vector source_ids, - std::size_t sampling_source_slot) { - if (!std::isfinite(frame_rate_fps) || frame_rate_fps <= 0.0) - throw std::invalid_argument( - "frame sampler rate must be finite and positive"); - if (sampling_source_slot >= source_ids.size()) - throw std::out_of_range("frame sampler source slot is invalid"); - d = std::make_unique( - frame_rate_fps, tile_width, tile_height, columns, - std::move(source_ids), sampling_source_slot); - if (d->period <= std::chrono::nanoseconds::zero()) - throw std::invalid_argument("frame sampler period is too small"); -} - -Frame_Sampler::~Frame_Sampler() = default; - -Frame_Sampler::Accept_Frame_Result Frame_Sampler::accept_frame( - std::size_t slot, std::shared_ptr frame) { - const auto result = d->atlas.accept_frame(slot, std::move(frame)); - if (result == detail::Gallery_Frame_Atlas::Accept_Frame_Result::accepted) { - d->accepted_source_frames.fetch_add(1, std::memory_order_release); - return Accept_Frame_Result::accepted; - } - d->rejected_source_frames.fetch_add(1, std::memory_order_relaxed); - return result == detail::Gallery_Frame_Atlas::Accept_Frame_Result::stale_frame - ? Accept_Frame_Result::stale_frame - : Accept_Frame_Result::invalid_frame; -} - -std::optional Frame_Sampler::sample() { - const auto now = Clock::now(); - const auto now_ns = clock_nanoseconds(now); - const auto compose_sample = [this, now](double deadline_delay_ms, - std::uint64_t sample_sequence) { - auto composition = d->atlas.compose(); - const auto& source = composition.sources[d->sampling_source_slot]; - Plot_Render_Tick tick{ - now, - source.completion_correlation_id, - static_cast(source.completion_presentation_time.count()) / - 1'000.0, - composition.width, - composition.height}; - return Sampled_Gallery_Frame{ - std::move(composition), tick, sample_sequence, - deadline_delay_ms}; - }; - /* 首帧以前没有可编码内容;首帧以后每个 deadline 都编码当前图集。 - * 这不是 changed-only:静态画面仍持续产生 H.264 帧并维持解码参考链。 */ - if (d->accepted_source_frames.load(std::memory_order_acquire) == 0) - return std::nullopt; - auto deadline_ns = d->next_deadline_ns.load(std::memory_order_acquire); - for (;;) { - if (deadline_ns == 0) { - const auto next = now_ns + d->period.count(); - if (!d->next_deadline_ns.compare_exchange_weak( - deadline_ns, next, std::memory_order_acq_rel, - std::memory_order_acquire)) - continue; - const auto sequence = d->sampled_frames.fetch_add( - 1, std::memory_order_relaxed) + 1U; - return compose_sample(0.0, sequence); - } - if (now_ns < deadline_ns) { - d->early_ticks.fetch_add(1, std::memory_order_relaxed); - return std::nullopt; - } - const auto late_ns = now_ns - deadline_ns; - const auto missed = static_cast( - late_ns / d->period.count()); - const auto next = deadline_ns + - static_cast(missed + 1U) * d->period.count(); - if (!d->next_deadline_ns.compare_exchange_weak( - deadline_ns, next, std::memory_order_acq_rel, - std::memory_order_acquire)) - continue; - const auto sequence = d->sampled_frames.fetch_add( - 1, std::memory_order_relaxed) + 1U; - d->missed_periods.fetch_add(missed, std::memory_order_relaxed); - return compose_sample(static_cast(late_ns) / 1'000'000.0, - sequence); - } -} - -detail::Gallery_Atlas_Description Frame_Sampler::describe() const { - return d->atlas.describe(); -} - -Frame_Sampler_State Frame_Sampler::state() const noexcept { - return { - d->frame_rate_fps, - d->accepted_source_frames.load(std::memory_order_relaxed), - d->rejected_source_frames.load(std::memory_order_relaxed), - d->sampled_frames.load(std::memory_order_relaxed), - d->early_ticks.load(std::memory_order_relaxed), - d->missed_periods.load(std::memory_order_relaxed)}; -} - -} diff --git a/web_server/src/frame_sampling/Frame_Sampler.hpp b/web_server/src/frame_sampling/Frame_Sampler.hpp deleted file mode 100644 index 4a0c71d..0000000 --- a/web_server/src/frame_sampling/Frame_Sampler.hpp +++ /dev/null @@ -1,63 +0,0 @@ -#pragma once -#include -#include "../detail/Gallery_Frame_Atlas.hpp" -#include -#include -#include -#include -#include -#include -#include - -namespace aethera::web::frame_sampling { - -struct Sampled_Gallery_Frame { - detail::Gallery_Atlas_Composition composition{}; /* 本次采样得到的原生像素图集;下一次成功采样前有效。 */ - Plot_Render_Tick tick{}; /* 触发采样的服务端帧时钟。 */ - std::uint64_t sample_sequence{}; /* 媒体流自身严格递增的帧序号。 */ - double deadline_delay_ms{}; /* 实际采样相对固定周期 deadline 的延迟。 */ -}; - -struct Frame_Sampler_State { - double target_frame_rate_fps{}; /* 采样器固定目标频率。 */ - std::uint64_t accepted_source_frames{}; /* 已进入权威 latest 的 Plot 完成帧数。 */ - std::uint64_t rejected_source_frames{}; /* 因尺寸、布局或顺序不合法而拒绝的完成帧数。 */ - std::uint64_t sampled_frames{}; /* 到达 deadline 并实际生成的图集样本数。 */ - std::uint64_t early_ticks{}; /* deadline 前到达、未触发采样的外接图次数。 */ - std::uint64_t missed_periods{}; /* 调度延迟跨过的完整采样周期数。 */ -}; - -/* - * Gallery 像素采样的唯一权威入口。Plot 可以按任意帧策略发布,采样器只在 - * 固定 deadline 到达时读取每个槽位的 latest,不排队、不补历史帧。 - */ -struct Frame_Sampler final { -public: - enum struct Accept_Frame_Result : std::uint8_t { - accepted, - invalid_frame, - stale_frame - }; - - Frame_Sampler(double frame_rate_fps, - std::uint32_t tile_width, - std::uint32_t tile_height, - std::uint32_t columns, - std::vector source_ids, - std::size_t sampling_source_slot); - ~Frame_Sampler(); - Frame_Sampler(const Frame_Sampler&) = delete; - Frame_Sampler& operator=(const Frame_Sampler&) = delete; - - [[nodiscard]] Accept_Frame_Result accept_frame( - std::size_t slot, std::shared_ptr frame); - [[nodiscard]] std::optional sample(); - [[nodiscard]] detail::Gallery_Atlas_Description describe() const; - [[nodiscard]] Frame_Sampler_State state() const noexcept; - -private: - struct Private; - std::unique_ptr d; -}; - -} diff --git a/web_server/src/frame_sampling/ffmpeg/FFmpeg_Frame_Transport.hpp b/web_server/src/frame_sampling/ffmpeg/FFmpeg_Frame_Transport.hpp deleted file mode 100644 index e0557e0..0000000 --- a/web_server/src/frame_sampling/ffmpeg/FFmpeg_Frame_Transport.hpp +++ /dev/null @@ -1,26 +0,0 @@ -#pragma once -#include "../../H264_Encoder.hpp" -#include "../Frame_Sampler.hpp" -#include -#include - -namespace aethera::web::frame_sampling::ffmpeg { - -/* 固定采样帧到 FFmpeg H.264 access unit 的单一编码位置。 */ -struct FFmpeg_Frame_Transport final { -public: - explicit FFmpeg_Frame_Transport(double frame_rate_fps); - ~FFmpeg_Frame_Transport(); - FFmpeg_Frame_Transport(const FFmpeg_Frame_Transport&) = delete; - FFmpeg_Frame_Transport& operator=(const FFmpeg_Frame_Transport&) = delete; - - void request_key_frame() noexcept; - [[nodiscard]] std::optional encode( - const Sampled_Gallery_Frame& frame); - -private: - struct Private; - std::unique_ptr d; -}; - -} diff --git a/web_server/src/media/Encoded_Video_Frame.hpp b/web_server/src/media/Encoded_Video_Frame.hpp new file mode 100644 index 0000000..3e06a22 --- /dev/null +++ b/web_server/src/media/Encoded_Video_Frame.hpp @@ -0,0 +1,32 @@ +#pragma once +#include +#include +#include +#include +#include + +namespace aethera::web::media { + +enum struct Video_Pixel_Layout : std::uint8_t { + bgra, + rgba +}; + +enum struct Video_Encoder_Backend : std::uint8_t { + libx264 +}; + +[[nodiscard]] std::string_view video_encoder_backend_name( + Video_Encoder_Backend backend) noexcept; +[[nodiscard]] std::string_view h264_profile_level_id() noexcept; + +struct Encoded_Video_Frame { + std::vector annex_b{}; /* 带起始码的完整 H.264 access unit,可直接交给 WebCodecs。 */ + std::chrono::microseconds presentation_time{}; /* 从媒体流起点累计的显示时间戳。 */ + std::chrono::nanoseconds source_time_unix{}; /* 对应 Plot 像素进入媒体流水线的 Unix 时间。 */ + std::uint64_t sequence{}; /* 图集媒体流的严格单调帧序号。 */ + Video_Encoder_Backend backend{Video_Encoder_Backend::libx264}; /* 实际产生本 access unit 的编码后端。 */ + bool key_frame{}; /* 本 access unit 是否可独立解码。 */ +}; + +} diff --git a/web_server/src/Gallery_Video_Stream.cpp b/web_server/src/media/Gallery_Video_Stream.cpp similarity index 73% rename from web_server/src/Gallery_Video_Stream.cpp rename to web_server/src/media/Gallery_Video_Stream.cpp index 14daf70..8e4042a 100644 --- a/web_server/src/Gallery_Video_Stream.cpp +++ b/web_server/src/media/Gallery_Video_Stream.cpp @@ -2,7 +2,6 @@ #include #include #include -#include "frame_sampling/Frame_Sampler.hpp" #include #include #include @@ -12,13 +11,13 @@ #include #include #include -namespace aethera::web { +namespace aethera::web::media { namespace { // Scene 的业务视口是 720x420;每组四图形成一行 2880x420 H.264 图集。 constexpr std::uint32_t tile_width{720}; constexpr std::uint32_t tile_height{420}; constexpr std::uint32_t atlas_columns{4}; -constexpr double gallery_frame_rate{30.0}; +constexpr double encoder_nominal_frame_rate{30.0}; constexpr auto metric_interval{std::chrono::seconds(1)}; nlohmann::json statistic_json(const Statistic_State& value) { return { @@ -44,7 +43,7 @@ std::string exception_description(const std::exception_ptr& failure) { } } Gallery_Video_Stream::Private::Private() - : ffmpeg_transport(gallery_frame_rate) {} + : ffmpeg_transport(encoder_nominal_frame_rate) {} Gallery_Video_Stream::Private::~Private() = default; void Gallery_Video_Stream::Private::initialize( std::vector plots) { @@ -57,9 +56,8 @@ void Gallery_Video_Stream::Private::initialize( source_ids.push_back(plot.id); sources.push_back(Source{std::move(plot)}); } - sampler = std::make_unique( - gallery_frame_rate, tile_width, tile_height, atlas_columns, - std::move(source_ids), sources.size() - 1U); + atlas = std::make_unique( + tile_width, tile_height, atlas_columns, std::move(source_ids)); metric_completion_starts.resize(sources.size()); metric_rendered_starts.resize(sources.size()); } @@ -167,7 +165,7 @@ void Gallery_Video_Stream::Private::fail( const auto notification = nlohmann::json{ {"kind", "gallery_error"}, {"protocol", "aethera.gallery.video"}, - {"version", 4}, + {"version", 5}, {"message", *terminal_failure} }.dump(); publish({}, notification); @@ -177,37 +175,48 @@ void Gallery_Video_Stream::Private::fail( void 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) + failed.load(std::memory_order_acquire) || !frame || !frame->pixels || + !frame->pixels->pixels || !consumer_accepts()) return; - static_cast(sampler->accept_frame(slot, frame->pixels)); + Pending_Plot_Frame pending{ + slot, std::move(frame->pixels), 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{}; + 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()); } void Gallery_Video_Stream::Private::update_metrics( - const frame_sampling::Sampled_Gallery_Frame& sample, + const Active_Media_Frame& frame, std::size_t encoded_bytes, std::optional encoder_backend) { - const auto& composition = sample.composition; - const auto sampler_state = sampler->state(); + const auto& composition = frame.composition; const auto now = std::chrono::steady_clock::now(); + const auto received_total = received_frame_count.load( + std::memory_order_relaxed); + const auto processed_total = processed_frame_count.load( + std::memory_order_relaxed); + const auto encoded_total = encoded_frame_count.load( + std::memory_order_relaxed); if (metric_started.time_since_epoch().count() == 0) { metric_started = now; - metric_encoded_start = encoded_frame_count; - metric_sample_start = sampler_state.sampled_frames; - metric_clock_start = sampling_clock_ticks.load(std::memory_order_relaxed); + metric_received_start = received_total; + metric_processed_start = processed_total; + metric_encoded_start = encoded_total; for (std::size_t slot = 0; slot < composition.sources.size(); ++slot) { metric_completion_starts[slot] = - composition.sources[slot].completion_count; + composition.sources[slot].completion_count; metric_rendered_starts[slot] = - composition.sources[slot].rendered_frame_count; + composition.sources[slot].rendered_frame_count; } return; } const auto elapsed = now - metric_started; if (elapsed < metric_interval) return; const double seconds = std::chrono::duration(elapsed).count(); - const auto clock_total = - sampling_clock_ticks.load(std::memory_order_relaxed); nlohmann::json source_metrics = nlohmann::json::object(); - const auto description = sampler->describe(); + const auto description = atlas->describe(); for (std::size_t slot = 0; slot < composition.sources.size(); ++slot) { const auto& progress = composition.sources[slot]; const auto completed = progress.completion_count - @@ -244,38 +253,39 @@ void Gallery_Video_Stream::Private::update_metrics( metric_rendered_starts[slot] = progress.rendered_frame_count; } const auto& compose = compose_ms.state(); - const auto& sample_delay = sample_delay_ms.state(); + const auto& queue_delay = queue_delay_ms.state(); const auto& encode = encode_ms.state(); const auto& publish_time = publish_ms.state(); auto output = nlohmann::json{ {"kind", "gallery_metrics"}, {"protocol", "aethera.gallery.video"}, - {"version", 4}, - {"clock_sequence", sample.sample_sequence}, - {"sampled_frame_count", sampler_state.sampled_frames}, - {"encoded_frame_count", encoded_frame_count}, + {"version", 5}, + {"media_sequence", frame.sequence}, + {"received_frame_count", received_total}, + {"processed_frame_count", processed_total}, + {"encoded_frame_count", encoded_total}, {"encoder_backend", encoder_backend ? video_encoder_backend_name(*encoder_backend) : std::string_view{"inactive"}}, {"profile_level_id", h264_profile_level_id()}, - {"target_frame_rate_fps", gallery_frame_rate}, { - "clock_delivery_rate_fps", - static_cast(clock_total - metric_clock_start) / seconds + "received_frame_rate_fps", + static_cast(received_total - metric_received_start) / + seconds }, { - "sampled_frame_rate_fps", - static_cast(sampler_state.sampled_frames - - metric_sample_start) / seconds + "processed_frame_rate_fps", + static_cast(processed_total - + metric_processed_start) / seconds }, { "encoded_frame_rate_fps", - static_cast(encoded_frame_count - metric_encoded_start) / seconds + static_cast(encoded_total - metric_encoded_start) / seconds }, {"compose_average_ms", compose.average}, {"compose_p95_ms", compose.p95}, - {"sample_delay_average_ms", sample_delay.average}, - {"sample_delay_p95_ms", sample_delay.p95}, + {"queue_delay_average_ms", queue_delay.average}, + {"queue_delay_p95_ms", queue_delay.p95}, {"encode_average_ms", encode.average}, {"encode_p95_ms", encode.p95}, {"publish_average_ms", publish_time.average}, @@ -283,57 +293,58 @@ void Gallery_Video_Stream::Private::update_metrics( { "statistics", { {"compose_ms", statistic_json(compose)}, - {"sample_delay_ms", statistic_json(sample_delay)}, + {"queue_delay_ms", statistic_json(queue_delay)}, {"encode_ms", statistic_json(encode)}, {"publish_ms", statistic_json(publish_time)} } }, + {"pending_frame_count", pending_frames.size_approx()}, {"encoded_bytes", encoded_bytes}, {"fresh_tiles", composition.fresh_tile_count}, {"missing_tiles", composition.missing_tile_count}, {"rejected_frames", composition.rejected_frame_count}, - { - "skipped_sample_ticks", - skipped_sample_ticks.load(std::memory_order_relaxed) - }, - {"sampler", { - {"accepted_source_frames", sampler_state.accepted_source_frames}, - {"rejected_source_frames", sampler_state.rejected_source_frames}, - {"early_ticks", sampler_state.early_ticks}, - {"missed_periods", sampler_state.missed_periods}}}, {"sources", std::move(source_metrics)} }; metric_started = now; - metric_encoded_start = encoded_frame_count; - metric_sample_start = sampler_state.sampled_frames; - metric_clock_start = clock_total; + 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(); } -void Gallery_Video_Stream::Private::sample_media_frame() { - active_sample.reset(); +void Gallery_Video_Stream::Private::begin_media_frame() { + active_frame.reset(); active_video.reset(); if (stopping.load(std::memory_order_acquire) || - failed.load(std::memory_order_acquire) || - !consumer_accepts()) { - skipped_sample_ticks.fetch_add(1, std::memory_order_relaxed); - return; - } + failed.load(std::memory_order_acquire)) return; + Pending_Plot_Frame pending; + if (!pending_frames.try_dequeue(pending)) return; + ++processed_frame_count; const auto started = std::chrono::steady_clock::now(); - active_sample = sampler->sample(); - if (!active_sample) { - skipped_sample_ticks.fetch_add(1, std::memory_order_relaxed); - return; - } + const auto accepted = atlas->accept_frame( + pending.slot, std::move(pending.frame)); + if (accepted == + detail::Gallery_Frame_Atlas::Accept_Frame_Result::invalid_frame) + throw std::logic_error( + "Plot published an invalid gallery pixel frame"); + auto presentation_time = std::chrono::duration_cast< + std::chrono::microseconds>(started - media_origin); + if (presentation_time <= last_presentation_time) + presentation_time = last_presentation_time + + std::chrono::microseconds{1}; + last_presentation_time = presentation_time; + active_frame = Active_Media_Frame{ + atlas->compose(), next_media_sequence++, presentation_time, + pending.source_time_unix, + std::chrono::duration( + started - pending.enqueued_at).count()}; static_cast(compose_ms.submit( std::chrono::duration( std::chrono::steady_clock::now() - started).count())); - static_cast(sample_delay_ms.submit( - active_sample->deadline_delay_ms)); + static_cast(queue_delay_ms.submit(active_frame->queue_delay_ms)); } void Gallery_Video_Stream::Private::encode_media_frame() { - if (!active_sample || !consumer_accepts()) - return; + if (!active_frame) return; /* * 直接在 FFmpeg.H264.encode Taskflow 节点中调用 FFmpeg。这样实际 * RGB->YUV420P 与 libx264 send/receive 全部落在同一个业务节点的 @@ -342,7 +353,14 @@ void Gallery_Video_Stream::Private::encode_media_frame() { * H264_Encoder 因而仍只有一个执行位置,不需要额外 mutex 或专用线程。 */ const auto started = std::chrono::steady_clock::now(); - if (auto encoded = ffmpeg_transport.encode(*active_sample)) + if (auto encoded = ffmpeg_transport.encode( + active_frame->composition.pixels, + active_frame->composition.width, + active_frame->composition.height, + active_frame->composition.layout, + active_frame->sequence, + active_frame->presentation_time, + active_frame->source_time_unix)) active_video = std::make_shared( std::move(*encoded)); static_cast(encode_ms.submit( @@ -359,13 +377,13 @@ void Gallery_Video_Stream::Private::publish_video_frame() { std::chrono::steady_clock::now() - started).count())); } void Gallery_Video_Stream::Private::complete_media_frame() { - if (!active_sample) return; + if (!active_frame) return; const auto encoded_bytes = active_video ? active_video->annex_b.size() : 0U; const auto encoder_backend = active_video ? std::optional{active_video->backend} : std::nullopt; - update_metrics(*active_sample, encoded_bytes, encoder_backend); - active_sample.reset(); + update_metrics(*active_frame, encoded_bytes, encoder_backend); + active_frame.reset(); active_video.reset(); } std::shared_ptr Gallery_Video_Stream::create( @@ -383,28 +401,86 @@ Gallery_Video_Stream::Gallery_Video_Stream() = default; Gallery_Video_Stream::~Gallery_Video_Stream() { shutdown(); } +void Gallery_Video_Stream::Private::arm_media_task( + std::weak_ptr lifetime) { + if (stopping.load(std::memory_order_acquire) || + failed.load(std::memory_order_acquire)) + return; + bool idle{}; + if (!media_task_scheduled.compare_exchange_strong( + 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(); + } + } + + try { + auto completion = [lifetime, graph, trace_frame, trace_started] { + static_cast(graph); + if (trace_started) + aethera::detail::finish_taskflow_trace(*trace_frame); + 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()); + data.media_task_scheduled.store(false, + 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)); + } + 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::bind_plots() { auto& data = static_cast(*d); const auto weak = weak_from_this(); if (data.sources.empty()) throw std::invalid_argument("gallery video stream requires a Plot"); - /* - * 图集媒体有自己的固定采样时钟,不再依赖某一张 Plot 完成后才能运行。 - * 每个来源只发布 latest;静态画面仍按 30 FPS 编码,上一轮媒体 DAG - * 尚未完成时合并本次 tick,不排队也不占用 Worker 等待。 - */ + /* 每个 Plot 独立帧策略发布的物理像素帧按到达顺序进入本 DAG。 + * Gallery 不再拥有采样时钟、准入或丢帧规则;持久 DAG 每次消费一个 + * 完成帧,串行保护图集像素和 FFmpeg 上下文,完成后异步续跑下一帧。 */ auto media = std::make_shared("gallery.video.frame"); - auto sample = media->add("gallery.sample.capture", [weak] { + auto compose = media->add("gallery.atlas.compose", [weak] { if (const auto owner = weak.lock()) { auto& owner_data = static_cast(*owner->d); - try { owner_data.sample_media_frame(); } + try { owner_data.begin_media_frame(); } catch (...) { owner_data.fail(std::current_exception()); } } }); - sample.describe("owner", "gallery") - .describe("stage", "30 FPS latest-frame sampling deadline") - .describe("frame_rate_fps", "30"); + compose.describe("owner", "gallery") + .describe("stage", "apply one policy-approved Plot frame") + .describe("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); @@ -413,7 +489,7 @@ void Gallery_Video_Stream::bind_plots() { } }); encode.describe("owner", "gallery") - .describe("stage", "FFmpeg H.264 encode after Plot pixel publish") + .describe("stage", "encode every composed Plot completion") .describe("backend", "FFmpeg") .describe("codec", "H.264") .describe("execution", "Taskflow worker"); @@ -427,7 +503,7 @@ void Gallery_Video_Stream::bind_plots() { publish_video.describe("owner", "gallery") .describe("transport", "Drogon WebSocket") .describe("stage", "H.264 access unit enqueue"); - auto complete = media->add("gallery.sample.complete", [weak] { + auto complete = media->add("gallery.media.complete", [weak] { if (const auto owner = weak.lock()) { auto& owner_data = static_cast(*owner->d); try { owner_data.complete_media_frame(); } @@ -435,80 +511,11 @@ void Gallery_Video_Stream::bind_plots() { } }); complete.describe("owner", "gallery") - .describe("stage", "sample metrics and ownership release"); - sample.precede(encode); + .describe("stage", "metrics and frame ownership release"); + compose.precede(encode); encode.precede(publish_video); publish_video.precede(complete); data.media_graph = media; - data.sample_timer = Frame_Scheduler::instance().make_timer( - [weak](Frame_Scheduler::Tick) { - const auto owner = weak.lock(); - if (!owner) return; - auto& owner_data = static_cast(*owner->d); - if (owner_data.stopping.load(std::memory_order_acquire) || - owner_data.failed.load(std::memory_order_acquire) || - !owner_data.consumer_accepts()) - return; - owner_data.sampling_clock_ticks.fetch_add( - 1, std::memory_order_relaxed); - bool idle{}; - if (!owner_data.media_busy.compare_exchange_strong( - idle, true, std::memory_order_acq_rel, - std::memory_order_acquire)) { - owner_data.skipped_sample_ticks.fetch_add( - 1, std::memory_order_relaxed); - return; - } - const auto graph = owner_data.media_graph; - bool trace_reserved = owner_data.mark_taskflow_trace(); - auto trace_frame = trace_reserved - ? std::make_shared(Frame_Identity{ - owner_data.sampling_clock_ticks.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) { - owner_data.restore_taskflow_trace(); - trace_reserved = false; - trace_frame.reset(); - } - } - try { - auto completion = [weak, graph, trace_frame, trace_started] { - static_cast(graph); - if (trace_started) - aethera::detail::finish_taskflow_trace(*trace_frame); - if (const auto completed = weak.lock()) { - auto& completed_data = - static_cast(*completed->d); - if (trace_started) - completed_data.store_taskflow_trace( - trace_frame->take_taskflow_trace()); - completed_data.media_busy.store( - false, std::memory_order_release); - } - }; - if (trace_frame) - aethera::detail::run_taskflow( - *graph, *trace_frame, "gallery.media", - std::move(completion)); - else - aethera::detail::run_taskflow( - *graph, std::move(completion)); - } - catch (...) { - if (trace_started) - aethera::detail::finish_taskflow_trace(*trace_frame); - if (trace_reserved) owner_data.restore_taskflow_trace(); - owner_data.media_busy.store(false, - std::memory_order_release); - owner_data.fail(std::current_exception()); - } - }); for (std::size_t slot = 0; slot < data.sources.size(); ++slot) { auto& source = data.sources[slot]; @@ -518,8 +525,8 @@ void Gallery_Video_Stream::bind_plots() { if (!owner) return; auto& owner_data = static_cast(*owner->d); if (owner_data.stopping.load(std::memory_order_acquire)) return; - /* 每个 Plot 只更新自己槽位的权威最新完成帧。 */ - owner_data.accept_frame(slot, std::move(frame)); + try { owner_data.accept_frame(slot, std::move(frame)); } + catch (...) { owner_data.fail(std::current_exception()); } }); source.entry.plot->configure_stream( source.stream, tile_width, tile_height); @@ -558,7 +565,6 @@ Gallery_Video_Stream::Stream_Id Gallery_Video_Stream::subscribe( std::memory_order_acquire)) { if (data.stopping.load(std::memory_order_acquire)) { static_cast(data.remove_consumer(id)); - data.sample_timer.cancel(); throw std::logic_error( "gallery video stream is shutting down"); } @@ -579,10 +585,8 @@ Gallery_Video_Stream::Stream_Id Gallery_Video_Stream::subscribe( if (!inserted) throw std::runtime_error( "gallery video stream consumer capacity is exhausted"); - data.sample_timer.start_periodic(gallery_frame_rate); if (data.stopping.load(std::memory_order_acquire)) { static_cast(data.remove_consumer(id)); - data.sample_timer.cancel(); throw std::logic_error("gallery video stream is shutting down"); } return id; @@ -600,7 +604,6 @@ void Gallery_Video_Stream::request_video_key_frame() { void Gallery_Video_Stream::shutdown() noexcept { auto& data = static_cast(*d); if (data.stopping.exchange(true, std::memory_order_acq_rel)) return; - data.sample_timer.cancel(); for (auto& source : data.sources) { if (source.stream == 0) continue; try { @@ -614,15 +617,14 @@ void Gallery_Video_Stream::shutdown() noexcept { } std::string Gallery_Video_Stream::layout_description() const { const auto& data = static_cast(*d); - const auto description = data.sampler->describe(); + 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}}; return nlohmann::json{ {"kind", "gallery_layout"}, {"protocol", "aethera.gallery.video"}, - {"version", 4}, + {"version", 5}, {"transport", "drogon_h264"}, - {"frame_rate_fps", gallery_frame_rate}, {"columns", description.columns}, {"rows", description.rows}, {"tile_width", description.tile_width}, @@ -636,8 +638,8 @@ std::string Gallery_Video_Stream::layout_description() const { {"profile_level_id", h264_profile_level_id()}, {"transport", "Drogon WebSocket binary"}, {"bitstream", "Annex B"}, - {"protocol_version", 1}, - {"header_bytes", 40} + {"protocol_version", 2}, + {"header_bytes", 48} } }, {"plots", std::move(plots)} @@ -647,13 +649,20 @@ nlohmann::json Gallery_Video_Stream::diagnostics() const { nlohmann::json output{ {"kind", "gallery_metrics"}, {"protocol", "aethera.gallery.video"}, - {"version", 4}, + {"version", 5}, {"sources", nlohmann::json::object()} }; this->template access_state([&output](const State& state) { if (!state.diagnostics.empty()) output = state.diagnostics; }); const auto& data = static_cast(*d); + output["received_frame_count"] = data.received_frame_count.load( + std::memory_order_relaxed); + output["processed_frame_count"] = data.processed_frame_count.load( + std::memory_order_relaxed); + output["encoded_frame_count"] = data.encoded_frame_count.load( + std::memory_order_relaxed); + output["pending_frame_count"] = data.pending_frames.size_approx(); std::size_t consumer_count{}; for (const auto& slot : data.consumers) { const auto consumer = slot.load(std::memory_order_acquire); diff --git a/web_server/src/Gallery_Video_Stream.hpp b/web_server/src/media/Gallery_Video_Stream.hpp similarity index 96% rename from web_server/src/Gallery_Video_Stream.hpp rename to web_server/src/media/Gallery_Video_Stream.hpp index 81a896c..5ad6bdf 100644 --- a/web_server/src/Gallery_Video_Stream.hpp +++ b/web_server/src/media/Gallery_Video_Stream.hpp @@ -1,6 +1,6 @@ #pragma once #include -#include "frame_sampling/ffmpeg/FFmpeg_Frame_Transport.hpp" +#include "Encoded_Video_Frame.hpp" #include #include #include @@ -9,7 +9,7 @@ #include #include -namespace aethera::web { +namespace aethera::web::media { struct Gallery_Stream_Frame { std::shared_ptr video{}; /* FFmpeg 订阅者共享的 H.264 access unit。 */ std::string notification{}; /* 仅承载终止错误等低频控制通知。 */ diff --git a/web_server/src/media/Gallery_Video_Stream.ipp b/web_server/src/media/Gallery_Video_Stream.ipp new file mode 100644 index 0000000..49410b1 --- /dev/null +++ b/web_server/src/media/Gallery_Video_Stream.ipp @@ -0,0 +1,116 @@ +#pragma once +#include "detail/FFmpeg_Frame_Transport.hpp" +#include "detail/Gallery_Frame_Atlas.hpp" +#include +#include +#include +#include +#include +#include +#include + +namespace aethera::web::media { + +struct Gallery_Video_Stream::Private : Prev_Private { + using Object = Gallery_Video_Stream; + + struct Source { + Plot_Entry entry; /* 图集槽位关联的实际 Plot。 */ + Plot::Stream_Id stream{}; /* Plot 完成帧的唯一订阅标识。 */ + }; + + struct Consumer { + Stream_Id id{}; /* 当前网页媒体连接的订阅身份。 */ + std::string connection{}; /* 同一页面重连期间稳定的协议身份。 */ + std::shared_ptr handler{}; /* 发布期间保持回调存活。 */ + std::shared_ptr readiness{}; /* 连接和解码器是否可接收媒体。 */ + std::shared_ptr diagnostics{}; /* 连接自身的传输诊断来源。 */ + }; + + struct Pending_Plot_Frame { + std::size_t slot{}; /* 该完成帧所属的稳定图集槽位。 */ + std::shared_ptr frame{}; /* 已通过所属 Plot 帧策略的不可变像素帧。 */ + std::chrono::steady_clock::time_point enqueued_at{}; /* 进入媒体串行队列的单调时刻。 */ + std::chrono::nanoseconds source_time_unix{}; /* 进入媒体串行队列的 Unix 时间。 */ + }; + + struct Active_Media_Frame { + detail::Gallery_Atlas_Composition composition{}; /* 应用一个 Plot 完成帧后的完整图集视图。 */ + std::uint64_t sequence{}; /* 媒体流严格递增的 access unit 序号。 */ + std::chrono::microseconds presentation_time{}; /* 媒体流起点以来的单调显示时间。 */ + std::chrono::nanoseconds source_time_unix{}; /* 对应 Plot 像素进入媒体流水线的 Unix 时间。 */ + double queue_delay_ms{}; /* Plot 发布到开始合成的队列时间。 */ + }; + + 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 编码耗时。 */ + Sliding_Statistics publish_ms{600}; /* H.264 WebSocket 发布调用耗时。 */ + std::vector sources{}; /* 已按业务标识排序的稳定图集来源。 */ + std::unique_ptr atlas{}; /* 图集像素和各槽位进度的唯一所有者。 */ + moodycamel::ConcurrentQueue pending_frames{}; /* 各 Plot 已完成像素帧的无损 MPMC 入口。 */ + std::shared_ptr media_graph{}; /* 每次只消费一个 Plot 完成帧的持久异步 DAG。 */ + std::atomic_bool media_task_scheduled{}; /* 唯一媒体 DAG 准入;不承担丢帧策略。 */ + std::atomic_uint64_t media_work_generation{}; /* 新完成帧入队后推进,封闭任务退出竞争窗口。 */ + detail::FFmpeg_Frame_Transport ffmpeg_transport; /* 仅由媒体 DAG 串行访问的编码上下文。 */ + std::optional active_frame{}; /* 当前媒体 DAG 独占的图集帧。 */ + std::shared_ptr active_video{}; /* 当前编码结果的共享所有权。 */ + std::atomic_uint64_t next_consumer_id{1}; /* 订阅身份生成器。 */ + std::array>, + maximum_consumers> consumers{}; /* 订阅关系的权威原子槽位。 */ + std::atomic_bool stopping{}; /* 关闭开始后拒绝新工作。 */ + std::atomic_bool failed{}; /* 首次 Unknown Failure 后停止媒体热路径。 */ + std::optional terminal_failure{}; /* 首次终止失败文本,仅故障路径访问。 */ + std::atomic_uint64_t received_frame_count{}; /* 活跃消费者期间收到的 Plot 像素帧总数。 */ + std::atomic_uint64_t processed_frame_count{}; /* 媒体 DAG 已按顺序消费的 Plot 像素帧总数。 */ + std::atomic_uint64_t encoded_frame_count{}; /* 成功产生 H.264 access unit 的累计帧数。 */ + std::uint64_t next_media_sequence{1}; /* 仅媒体 DAG 访问的 access unit 序号。 */ + std::chrono::steady_clock::time_point media_origin{std::chrono::steady_clock::now()}; /* 单调 H.264 PTS 原点。 */ + std::chrono::microseconds last_presentation_time{}; /* 仅媒体 DAG 访问的最近 H.264 PTS。 */ + std::chrono::steady_clock::time_point metric_started{}; /* 当前统计窗口起点。 */ + std::uint64_t metric_received_start{}; /* 窗口起点累计收到帧数。 */ + std::uint64_t metric_processed_start{}; /* 窗口起点累计处理帧数。 */ + std::uint64_t metric_encoded_start{}; /* 窗口起点累计编码帧数。 */ + std::vector metric_completion_starts{}; /* 各 Plot 窗口起点逻辑完成数。 */ + std::vector metric_rendered_starts{}; /* 各 Plot 窗口起点真实画面数。 */ + static constexpr std::size_t maximum_taskflow_trace_frames{120}; + std::atomic_size_t taskflow_trace_remaining{}; /* 尚待捕获的媒体 DAG 次数。 */ + std::atomic_uint64_t taskflow_trace_control{}; /* 高 32 位 requested,低 32 位 captured。 */ + std::array>, + 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; + void publish(std::shared_ptr video, + std::string notification) noexcept; + void fail(std::exception_ptr failure) noexcept; + void accept_frame(std::size_t slot, + std::shared_ptr frame); + void arm_media_task(std::weak_ptr lifetime); + void begin_media_frame(); + void encode_media_frame(); + void publish_video_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); + [[nodiscard]] nlohmann::json taskflow_trace_response() const; +}; + +} diff --git a/web_server/src/Gallery_WebSocket.cpp b/web_server/src/media/Gallery_Video_WebSocket.cpp similarity index 86% rename from web_server/src/Gallery_WebSocket.cpp rename to web_server/src/media/Gallery_Video_WebSocket.cpp index fe3449d..e3af176 100644 --- a/web_server/src/Gallery_WebSocket.cpp +++ b/web_server/src/media/Gallery_Video_WebSocket.cpp @@ -1,4 +1,4 @@ -#include "Gallery_WebSocket.hpp" +#include "Gallery_Video_WebSocket.hpp" #include #include #include @@ -13,10 +13,9 @@ #include #include -namespace aethera::web { +namespace aethera::web::media { namespace { -constexpr std::size_t h264_header_bytes{40}; -constexpr std::size_t maximum_outstanding_frames{3}; +constexpr std::size_t h264_header_bytes{48}; template void write_little_endian(std::span output, @@ -39,24 +38,25 @@ std::vector websocket_packet(const Encoded_Video_Frame& frame) { output[2] = static_cast('H'); output[3] = static_cast('1'); const auto bytes = std::span{output}; - write_little_endian(bytes, 4, std::uint16_t{1}); + write_little_endian(bytes, 4, std::uint16_t{2}); write_little_endian(bytes, 6, static_cast(frame.key_frame ? 1U : 0U)); write_little_endian(bytes, 8, frame.sequence); write_little_endian(bytes, 16, frame.presentation_time.count()); + write_little_endian(bytes, 24, frame.source_time_unix.count()); const auto packed_time = std::chrono::duration_cast< std::chrono::nanoseconds>( std::chrono::system_clock::now().time_since_epoch()).count(); - write_little_endian(bytes, 24, packed_time); - write_little_endian(bytes, 32, + write_little_endian(bytes, 32, packed_time); + write_little_endian(bytes, 40, static_cast(frame.annex_b.size())); - write_little_endian(bytes, 36, std::uint32_t{0}); + write_little_endian(bytes, 44, std::uint32_t{0}); std::ranges::copy(frame.annex_b, output.begin() + h264_header_bytes); return output; } } -struct Gallery_WebSocket::Private final { +struct Gallery_Video_WebSocket::Private final { std::weak_ptr connection; std::shared_ptr stream; Gallery_Video_Stream::Stream_Id subscription{}; @@ -71,7 +71,7 @@ struct Gallery_WebSocket::Private final { std::atomic_uint64_t rejected_count{}; }; -Gallery_WebSocket::Gallery_WebSocket( +Gallery_Video_WebSocket::Gallery_Video_WebSocket( drogon::WebSocketConnectionPtr connection, std::shared_ptr stream, std::string connection_id) @@ -84,9 +84,9 @@ Gallery_WebSocket::Gallery_WebSocket( d->connection_id = std::move(connection_id); } -Gallery_WebSocket::~Gallery_WebSocket() { close(); } +Gallery_Video_WebSocket::~Gallery_Video_WebSocket() { close(); } -void Gallery_WebSocket::start() { +void Gallery_Video_WebSocket::start() { if (d->attached.exchange(true, std::memory_order_acq_rel)) return; const auto weak = weak_from_this(); d->subscription = d->stream->subscribe( @@ -102,10 +102,7 @@ void Gallery_WebSocket::start() { return false; const auto connection = socket->d->connection.lock(); return connection && connection->connected() && - socket->d->decoder_ready.load(std::memory_order_acquire) && - socket->d->outstanding_frames.load( - std::memory_order_acquire) < - maximum_outstanding_frames; + socket->d->decoder_ready.load(std::memory_order_acquire); }, [weak] { const auto socket = weak.lock(); @@ -135,7 +132,7 @@ void Gallery_WebSocket::start() { drogon::WebSocketMessageType::Text); } -void Gallery_WebSocket::deliver(Gallery_Stream_Frame frame) { +void Gallery_Video_WebSocket::deliver(Gallery_Stream_Frame frame) { if (!d->attached.load(std::memory_order_acquire)) return; if (frame.video && !queue_video_frame(std::move(frame.video))) { d->rejected_count.fetch_add(1, std::memory_order_relaxed); @@ -149,7 +146,7 @@ void Gallery_WebSocket::deliver(Gallery_Stream_Frame frame) { } } -bool Gallery_WebSocket::queue_video_frame( +bool Gallery_Video_WebSocket::queue_video_frame( std::shared_ptr frame) { if (!frame || !d->attached.load(std::memory_order_acquire)) return false; @@ -173,7 +170,7 @@ bool Gallery_WebSocket::queue_video_frame( } } -void Gallery_WebSocket::acknowledge_video_frame(std::uint64_t sequence) { +void Gallery_Video_WebSocket::acknowledge_video_frame(std::uint64_t sequence) { auto acknowledged = d->latest_acknowledged_sequence.load( std::memory_order_acquire); while (sequence > acknowledged) { @@ -193,7 +190,7 @@ void Gallery_WebSocket::acknowledge_video_frame(std::uint64_t sequence) { } } -void Gallery_WebSocket::receive(std::string_view message) { +void Gallery_Video_WebSocket::receive(std::string_view message) { const auto json = nlohmann::json::parse(message, nullptr, false); if (!json.is_object()) return; const auto kind = json.value("kind", std::string{}); @@ -215,7 +212,7 @@ void Gallery_WebSocket::receive(std::string_view message) { d->stream->request_video_key_frame(); } -void Gallery_WebSocket::close() noexcept { +void Gallery_Video_WebSocket::close() noexcept { if (!d->attached.exchange(false, std::memory_order_acq_rel)) return; try { if (d->subscription != 0) d->stream->unsubscribe(d->subscription); @@ -226,19 +223,19 @@ void Gallery_WebSocket::close() noexcept { d->outstanding_frames.store(0, std::memory_order_release); } -Gallery_WebSocket_Controller::Gallery_WebSocket_Controller( +Gallery_Video_WebSocket_Controller::Gallery_Video_WebSocket_Controller( Stream_Resolver resolver) : resolve_stream(std::move(resolver)) { if (!resolve_stream) throw std::invalid_argument( "gallery WebSocket requires a stream resolver"); } -void Gallery_WebSocket_Controller::initPathRouting() { +void Gallery_Video_WebSocket_Controller::initPathRouting() { drogon::app().registerWebSocketController( "/ws/gallery", classTypeName()); } -void Gallery_WebSocket_Controller::handleNewConnection( +void Gallery_Video_WebSocket_Controller::handleNewConnection( const drogon::HttpRequestPtr& request, const drogon::WebSocketConnectionPtr& connection) { try { @@ -249,7 +246,7 @@ void Gallery_WebSocket_Controller::handleNewConnection( "Invalid gallery H.264 request"); return; } - auto socket = std::make_shared( + auto socket = std::make_shared( connection, std::move(stream), connection_id); connection->setContext(socket); connection->setPingMessage( @@ -257,7 +254,7 @@ void Gallery_WebSocket_Controller::handleNewConnection( socket->start(); } catch (...) { - if (const auto socket = connection->getContext()) + if (const auto socket = connection->getContext()) socket->close(); try { connection->shutdown(drogon::CloseCode::kViolation, @@ -267,19 +264,19 @@ void Gallery_WebSocket_Controller::handleNewConnection( } } -void Gallery_WebSocket_Controller::handleNewMessage( +void Gallery_Video_WebSocket_Controller::handleNewMessage( const drogon::WebSocketConnectionPtr& connection, std::string&& message, const drogon::WebSocketMessageType& type) { if (type != drogon::WebSocketMessageType::Text || message.size() > 4096U) return; - if (const auto socket = connection->getContext()) + if (const auto socket = connection->getContext()) socket->receive(message); } -void Gallery_WebSocket_Controller::handleConnectionClosed( +void Gallery_Video_WebSocket_Controller::handleConnectionClosed( const drogon::WebSocketConnectionPtr& connection) { - if (const auto socket = connection->getContext()) + if (const auto socket = connection->getContext()) socket->close(); connection->clearContext(); } diff --git a/web_server/src/Gallery_WebSocket.hpp b/web_server/src/media/Gallery_Video_WebSocket.hpp similarity index 70% rename from web_server/src/Gallery_WebSocket.hpp rename to web_server/src/media/Gallery_Video_WebSocket.hpp index c1f633e..b970de4 100644 --- a/web_server/src/Gallery_WebSocket.hpp +++ b/web_server/src/media/Gallery_Video_WebSocket.hpp @@ -5,17 +5,17 @@ #include #include -namespace aethera::web { +namespace aethera::web::media { /* 同一个 Drogon WebSocket 承载布局/背压控制文本和 H.264 Annex-B 二进制帧。 */ -struct Gallery_WebSocket final - : std::enable_shared_from_this { - Gallery_WebSocket(drogon::WebSocketConnectionPtr connection, +struct Gallery_Video_WebSocket final + : std::enable_shared_from_this { + Gallery_Video_WebSocket(drogon::WebSocketConnectionPtr connection, std::shared_ptr stream, std::string connection_id); - ~Gallery_WebSocket(); - Gallery_WebSocket(const Gallery_WebSocket&) = delete; - Gallery_WebSocket& operator=(const Gallery_WebSocket&) = delete; + ~Gallery_Video_WebSocket(); + Gallery_Video_WebSocket(const Gallery_Video_WebSocket&) = delete; + Gallery_Video_WebSocket& operator=(const Gallery_Video_WebSocket&) = delete; void start(); void receive(std::string_view message); @@ -30,11 +30,11 @@ private: std::unique_ptr d; }; -struct Gallery_WebSocket_Controller final - : drogon::WebSocketController { +struct Gallery_Video_WebSocket_Controller final + : drogon::WebSocketController { using Stream_Resolver = std::function< std::shared_ptr(std::string_view)>; - explicit Gallery_WebSocket_Controller(Stream_Resolver resolver); + explicit Gallery_Video_WebSocket_Controller(Stream_Resolver resolver); void handleNewMessage(const drogon::WebSocketConnectionPtr& connection, std::string&& message, const drogon::WebSocketMessageType& type) override; diff --git a/web_server/src/frame_sampling/ffmpeg/FFmpeg_Frame_Transport.cpp b/web_server/src/media/detail/FFmpeg_Frame_Transport.cpp similarity index 58% rename from web_server/src/frame_sampling/ffmpeg/FFmpeg_Frame_Transport.cpp rename to web_server/src/media/detail/FFmpeg_Frame_Transport.cpp index 52b480d..8f52d17 100644 --- a/web_server/src/frame_sampling/ffmpeg/FFmpeg_Frame_Transport.cpp +++ b/web_server/src/media/detail/FFmpeg_Frame_Transport.cpp @@ -1,17 +1,14 @@ #include "FFmpeg_Frame_Transport.hpp" #include -#include -namespace aethera::web::frame_sampling::ffmpeg { +namespace aethera::web::media::detail { struct FFmpeg_Frame_Transport::Private { H264_Encoder encoder; /* FFmpeg 编码上下文的唯一所有者。 */ - double frame_rate_fps{}; /* 媒体序号到单调 PTS 的唯一换算基准。 */ std::atomic_bool key_frame_requested{}; /* 下一次实际编码样本是否强制关键帧。 */ explicit Private(double value_frame_rate_fps) - : encoder(value_frame_rate_fps), - frame_rate_fps(value_frame_rate_fps) {} + : encoder(value_frame_rate_fps) {} }; FFmpeg_Frame_Transport::FFmpeg_Frame_Transport(double frame_rate_fps) @@ -24,20 +21,21 @@ void FFmpeg_Frame_Transport::request_key_frame() noexcept { } std::optional FFmpeg_Frame_Transport::encode( - const Sampled_Gallery_Frame& frame) { + std::span pixels, + std::uint32_t width, + std::uint32_t height, + Plot_Pixel_Layout layout, + std::uint64_t sequence, + std::chrono::microseconds presentation_time, + std::chrono::nanoseconds source_time_unix) { if (d->key_frame_requested.exchange(false, std::memory_order_acq_rel)) d->encoder.request_key_frame(); return d->encoder.encode( - frame.composition.pixels, - frame.composition.width, - frame.composition.height, - frame.composition.layout == Plot_Pixel_Layout::bgra8 + pixels, width, height, + layout == Plot_Pixel_Layout::bgra8 ? Video_Pixel_Layout::bgra : Video_Pixel_Layout::rgba, - frame.sample_sequence, - std::chrono::microseconds{static_cast(std::llround( - static_cast(frame.sample_sequence) * - (1'000'000.0 / d->frame_rate_fps)))}); + sequence, presentation_time, source_time_unix); } } diff --git a/web_server/src/media/detail/FFmpeg_Frame_Transport.hpp b/web_server/src/media/detail/FFmpeg_Frame_Transport.hpp new file mode 100644 index 0000000..b9b90d8 --- /dev/null +++ b/web_server/src/media/detail/FFmpeg_Frame_Transport.hpp @@ -0,0 +1,36 @@ +#pragma once +#include "H264_Encoder.hpp" +#include +#include +#include +#include +#include +#include +#include + +namespace aethera::web::media::detail { + +/* 已通过 Plot 帧策略的图集像素到 H.264 access unit 的单一编码位置。 */ +struct FFmpeg_Frame_Transport final { +public: + explicit FFmpeg_Frame_Transport(double frame_rate_fps); + ~FFmpeg_Frame_Transport(); + FFmpeg_Frame_Transport(const FFmpeg_Frame_Transport&) = delete; + FFmpeg_Frame_Transport& operator=(const FFmpeg_Frame_Transport&) = delete; + + void request_key_frame() noexcept; + [[nodiscard]] std::optional encode( + std::span pixels, + std::uint32_t width, + std::uint32_t height, + Plot_Pixel_Layout layout, + std::uint64_t sequence, + std::chrono::microseconds presentation_time, + std::chrono::nanoseconds source_time_unix); + +private: + struct Private; + std::unique_ptr d; +}; + +} diff --git a/web_server/src/detail/Gallery_Frame_Atlas.cpp b/web_server/src/media/detail/Gallery_Frame_Atlas.cpp similarity index 90% rename from web_server/src/detail/Gallery_Frame_Atlas.cpp rename to web_server/src/media/detail/Gallery_Frame_Atlas.cpp index d724f20..9e6be31 100644 --- a/web_server/src/detail/Gallery_Frame_Atlas.cpp +++ b/web_server/src/media/detail/Gallery_Frame_Atlas.cpp @@ -6,7 +6,7 @@ #include #include -namespace aethera::web::detail { +namespace aethera::web::media::detail { struct Gallery_Frame_Atlas::Private { struct Source { std::atomic> @@ -96,15 +96,10 @@ Gallery_Frame_Atlas::Accept_Frame_Result Gallery_Frame_Atlas::accept_frame( } auto current = source.latest_completion.load( std::memory_order_acquire); - bool rendered_advanced{}; + const bool rendered_advanced = frame->rendered_sequence != 0 && + (!current || + frame->rendered_sequence != current->rendered_sequence); for (;;) { - if (current && frame->sequence <= current->sequence) { - d->rejected_frame_count.fetch_add(1, std::memory_order_relaxed); - return Accept_Frame_Result::stale_frame; - } - rendered_advanced = frame->rendered_sequence != 0 && - (!current || - frame->rendered_sequence != current->rendered_sequence); auto desired = frame; if (source.latest_completion.compare_exchange_weak( current, std::move(desired), std::memory_order_release, @@ -115,18 +110,9 @@ Gallery_Frame_Atlas::Accept_Frame_Result Gallery_Frame_Atlas::accept_frame( if (rendered_advanced) source.rendered_frame_count.fetch_add( 1, std::memory_order_release); - if (frame->pixels && frame->rendered_sequence != 0) { - auto pixels = source.latest_pixels.load(std::memory_order_acquire); - while (!pixels || - frame->rendered_sequence > pixels->rendered_sequence) { - auto desired = frame; - if (source.latest_pixels.compare_exchange_weak( - pixels, std::move(desired), - std::memory_order_release, - std::memory_order_acquire)) - break; - } - } + if (frame->pixels && frame->rendered_sequence != 0) + source.latest_pixels.store(std::move(frame), + std::memory_order_release); return Accept_Frame_Result::accepted; } diff --git a/web_server/src/detail/Gallery_Frame_Atlas.hpp b/web_server/src/media/detail/Gallery_Frame_Atlas.hpp similarity index 97% rename from web_server/src/detail/Gallery_Frame_Atlas.hpp rename to web_server/src/media/detail/Gallery_Frame_Atlas.hpp index b943f8a..615873f 100644 --- a/web_server/src/detail/Gallery_Frame_Atlas.hpp +++ b/web_server/src/media/detail/Gallery_Frame_Atlas.hpp @@ -7,7 +7,7 @@ #include #include #include -namespace aethera::web::detail { +namespace aethera::web::media::detail { struct Gallery_Atlas_Source_Description { std::string id; /* 图集槽位对应的 Plot 业务标识。 */ std::size_t slot{}; /* 从左到右、从上到下的稳定槽位序号。 */ @@ -46,8 +46,7 @@ struct Gallery_Frame_Atlas final { public: enum struct Accept_Frame_Result : std::uint8_t { accepted, - invalid_frame, - stale_frame + invalid_frame }; Gallery_Frame_Atlas(std::uint32_t tile_width, std::uint32_t tile_height, diff --git a/web_server/src/H264_Encoder.cpp b/web_server/src/media/detail/H264_Encoder.cpp similarity index 96% rename from web_server/src/H264_Encoder.cpp rename to web_server/src/media/detail/H264_Encoder.cpp index b50427e..2d91516 100644 --- a/web_server/src/H264_Encoder.cpp +++ b/web_server/src/media/detail/H264_Encoder.cpp @@ -16,7 +16,7 @@ extern "C" { #include } -namespace aethera::web { +namespace aethera::web::media { namespace { constexpr std::string_view high_profile_level_5_1{"640033"}; @@ -49,10 +49,13 @@ std::string_view h264_profile_level_id() noexcept { return high_profile_level_5_1; } +namespace detail { + struct H264_Encoder::Private { struct Submitted_Frame { std::uint64_t sequence{}; std::chrono::microseconds presentation_time{}; + std::chrono::nanoseconds source_time_unix{}; }; double frame_rate{}; @@ -199,7 +202,8 @@ void H264_Encoder::request_key_frame() { std::optional H264_Encoder::encode( std::span pixels, std::uint32_t width, std::uint32_t height, Video_Pixel_Layout layout, - std::uint64_t sequence, std::chrono::microseconds presentation_time) { + std::uint64_t sequence, std::chrono::microseconds presentation_time, + std::chrono::nanoseconds source_time_unix) { if (width == 0 || height == 0 || (width & 1U) != 0 || (height & 1U) != 0 || static_cast(width) > @@ -218,7 +222,8 @@ std::optional H264_Encoder::encode( ? AV_PICTURE_TYPE_I : AV_PICTURE_TYPE_NONE; d->key_frame_requested = false; - d->submitted_frames.push_back({sequence, presentation_time}); + d->submitted_frames.push_back( + {sequence, presentation_time, source_time_unix}); const auto submitted = avcodec_send_frame( d->codec_context, d->frame); if (submitted < 0) { @@ -247,9 +252,12 @@ std::optional H264_Encoder::encode( reinterpret_cast( d->packet->data + d->packet->size)); output.presentation_time = submitted_frame.presentation_time; + output.source_time_unix = submitted_frame.source_time_unix; output.sequence = submitted_frame.sequence; output.backend = Video_Encoder_Backend::libx264; output.key_frame = (d->packet->flags & AV_PKT_FLAG_KEY) != 0; return output; } + +} } diff --git a/web_server/src/media/detail/H264_Encoder.hpp b/web_server/src/media/detail/H264_Encoder.hpp new file mode 100644 index 0000000..a974f6f --- /dev/null +++ b/web_server/src/media/detail/H264_Encoder.hpp @@ -0,0 +1,26 @@ +#pragma once +#include "../Encoded_Video_Frame.hpp" +#include +#include +#include +#include +#include +#include +namespace aethera::web::media::detail { +struct H264_Encoder final { +public: + explicit H264_Encoder(double frame_rate); + ~H264_Encoder(); + H264_Encoder(const H264_Encoder&) = delete; + H264_Encoder& operator=(const H264_Encoder&) = delete; + void request_key_frame(); + [[nodiscard]] std::optional encode( + std::span pixels, std::uint32_t width, + std::uint32_t height, Video_Pixel_Layout layout, + std::uint64_t sequence, std::chrono::microseconds presentation_time, + std::chrono::nanoseconds source_time_unix); +private: + struct Private; + std::unique_ptr d; +}; +} diff --git a/web_server/tests/Gallery_Frame_Atlas_Tests.cpp b/web_server/tests/Gallery_Frame_Atlas_Tests.cpp index a1fc18e..24b8373 100644 --- a/web_server/tests/Gallery_Frame_Atlas_Tests.cpp +++ b/web_server/tests/Gallery_Frame_Atlas_Tests.cpp @@ -1,11 +1,11 @@ #include -#include +#include #include #include #include #include -namespace aethera::web::detail { +namespace aethera::web::media::detail { namespace { std::shared_ptr solid_frame( std::uint64_t sequence, std::uint32_t width, @@ -70,20 +70,21 @@ TEST(Gallery_Frame_Atlas, EXPECT_EQ(composition.sources[0].rendered_frame_count, 1U); } -TEST(Gallery_Frame_Atlas, Rejects_Stale_Completion_Without_Mutating_Progress) { +TEST(Gallery_Frame_Atlas, Applies_Out_Of_Order_Completions_In_Arrival_Order) { Gallery_Frame_Atlas atlas(2, 2, 1, {"alpha"}); ASSERT_EQ(atlas.accept_frame( 0, solid_frame(2, 2, 2, std::byte{31})), Gallery_Frame_Atlas::Accept_Frame_Result::accepted); + static_cast(atlas.compose()); ASSERT_EQ(atlas.accept_frame( - 0, completion_frame(1, 1, 2, 2)), - Gallery_Frame_Atlas::Accept_Frame_Result::stale_frame); + 0, solid_frame(1, 2, 2, std::byte{47})), + Gallery_Frame_Atlas::Accept_Frame_Result::accepted); const auto composition = atlas.compose(); ASSERT_EQ(composition.sources.size(), 1U); - EXPECT_EQ(composition.rejected_frame_count, 1U); - EXPECT_EQ(composition.sources[0].completion_sequence, 2U); - EXPECT_EQ(composition.sources[0].completion_count, 1U); - EXPECT_EQ(composition.pixels[0], std::byte{31}); + EXPECT_EQ(composition.rejected_frame_count, 0U); + EXPECT_EQ(composition.sources[0].completion_sequence, 1U); + EXPECT_EQ(composition.sources[0].completion_count, 2U); + EXPECT_EQ(composition.pixels[0], std::byte{47}); } } diff --git a/web_server/tests/Gallery_Video_Stream_Tests.cpp b/web_server/tests/Gallery_Video_Stream_Tests.cpp new file mode 100644 index 0000000..8d55414 --- /dev/null +++ b/web_server/tests/Gallery_Video_Stream_Tests.cpp @@ -0,0 +1,122 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace aethera::web::media { +namespace { + +using Clock = std::chrono::steady_clock; + +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([&] { + if (completed()) return true; + if (Clock::now() < deadline) return false; + failure = std::string{operation} + " timed out"; + return true; + }); + }); + aethera::detail::run_taskflow(coordinator); + if (!failure.empty()) throw std::runtime_error(failure); +} + +void configure_manual_pixel_delivery(const std::shared_ptr& plot) { + ASSERT_TRUE(plot); + ASSERT_TRUE(plot->write_prop( + "frame-analysis", "pacing_mode", "manual").value( + "success", false)); + ASSERT_TRUE(plot->write_prop( + "frame-analysis", "pixel_delivery_enabled", true).value( + "success", false)); + ASSERT_NO_THROW(corun_until_or_throw("manual Plot policy", [&] { + const auto diagnostics = plot->diagnostics(); + const auto& configuration = diagnostics.at("frame_policy").at( + "configuration"); + return configuration.at("mode") == "manual" && + configuration.at("pixel_delivery_enabled").get(); + })); +} + +void schedule_manual_frame(const std::shared_ptr& plot, + std::uint64_t correlation_sequence) { + const auto now = Clock::now(); + plot->schedule_render(Plot_Render_Tick{ + .issued_at = now, + .sequence = correlation_sequence, + .time_milliseconds = + std::chrono::duration( + now.time_since_epoch()).count(), + .width = 720, + .height = 420, + .source = Frame_Request_Source::immediate + }); +} + +TEST(Gallery_Video_Stream, + Encodes_Each_Plots_Independent_Policy_Frames_Without_Grouping) { + const auto definitions = gallery_plot_definitions(); + const auto found = std::ranges::find_if( + definitions, [](const Plot_Definition& definition) { + return definition.dimension == Plot_Dimension::two_d; + }); + ASSERT_NE(found, definitions.end()); + auto plot_a = found->create(); + auto plot_b = found->create(); + configure_manual_pixel_delivery(plot_a); + configure_manual_pixel_delivery(plot_b); + + auto stream = Gallery_Video_Stream::create({ + {"plot-a", plot_a}, {"plot-b", plot_b}}); + ASSERT_TRUE(stream); + constexpr std::size_t frame_count{5}; + std::array sequences{}; + std::atomic_size_t received{}; + const auto subscription = stream->subscribe( + "gallery-video-test", + [&](Gallery_Stream_Frame frame) { + if (!frame.video) return; + 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); + }, + [] { return true; }, + [] { return nlohmann::json::object(); }); + + const std::array, frame_count> schedule{ + plot_a, plot_a, plot_b, plot_a, plot_b}; + for (std::size_t index = 0; index < schedule.size(); ++index) { + schedule_manual_frame(schedule[index], index + 1U); + ASSERT_NO_THROW(corun_until_or_throw("encoded Plot frame", [&] { + return received.load(std::memory_order_acquire) > index; + })); + } + + 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); + EXPECT_EQ(diagnostics.at("pending_frame_count"), 0U); + + stream->unsubscribe(subscription); + stream->shutdown(); +} + +} +} diff --git a/web_server/tests/H264_Encoder_Tests.cpp b/web_server/tests/H264_Encoder_Tests.cpp index 0151400..66f380c 100644 --- a/web_server/tests/H264_Encoder_Tests.cpp +++ b/web_server/tests/H264_Encoder_Tests.cpp @@ -1,12 +1,12 @@ #include -#include +#include #include #include #include #include #include -namespace aethera::web { +namespace aethera::web::media::detail { namespace { std::vector test_pattern(std::uint32_t width, std::uint32_t height) { @@ -34,13 +34,15 @@ TEST(H264_Encoder, Rejects_Invalid_Input_Before_FFmpeg) { const auto pixels = test_pattern(320, 180); EXPECT_THROW(static_cast(encoder.encode( pixels, 319, 180, Video_Pixel_Layout::rgba, - 1, std::chrono::microseconds{33'333})), + 1, std::chrono::microseconds{33'333}, + std::chrono::nanoseconds{1})), std::invalid_argument); EXPECT_THROW(static_cast(encoder.encode( std::span{pixels}.first( pixels.size() - 1U), 320, 180, Video_Pixel_Layout::rgba, - 1, std::chrono::microseconds{33'333})), + 1, std::chrono::microseconds{33'333}, + std::chrono::nanoseconds{1})), std::invalid_argument); } @@ -57,11 +59,15 @@ TEST(H264_Encoder, Libx264_Produces_Ordered_Annex_B_Access_Units) { auto encoded = encoder.encode( pixels, width, height, Video_Pixel_Layout::rgba, sequence, std::chrono::microseconds{ - static_cast(sequence * 33'333)}); + static_cast(sequence * 33'333)}, + std::chrono::nanoseconds{ + static_cast(sequence * 1'000)}); if (!encoded) continue; if (!first) first = encoded; EXPECT_EQ(encoded->sequence, sequence); EXPECT_EQ(encoded->backend, Video_Encoder_Backend::libx264); + EXPECT_EQ(encoded->source_time_unix, + std::chrono::nanoseconds{sequence * 1'000}); EXPECT_FALSE(encoded->annex_b.empty()); ++produced; } diff --git a/webapp_gallery/src/app.tsx b/webapp_gallery/src/app.tsx index ec7d8b8..8d5b782 100644 --- a/webapp_gallery/src/app.tsx +++ b/webapp_gallery/src/app.tsx @@ -48,10 +48,11 @@ type Plot_Diagnostics = {protocol: "aethera.plot.diagnostics"; version: 4; dimen type Frame_Sample = {sequence: number; generated_at_ms: number; received_at_ms: number; values: Frame_Stage_Values}; 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; - websocket_arrival_ms: number; browser_queue_ms: number; browser_render_ms: number; + server_pipeline_ms: number; websocket_arrival_ms: number; source_arrival_ms: number; + browser_queue_ms: number; browser_render_ms: number; frame_latency_ms: number; freeze_count: number; current_time_seconds: number; ready_state: number}; type Gallery_Source = {column: number; row: number}; -type Gallery_Layout = {kind: "gallery_layout"; protocol: "aethera.gallery.video"; version: 4; +type Gallery_Layout = {kind: "gallery_layout"; protocol: "aethera.gallery.video"; version: 5; transport: "drogon_h264"; frame_rate_fps?: number; columns: number; rows: number; width: number; height: number; tile_width: number; tile_height: number; @@ -62,13 +63,13 @@ type Gallery_Layout = {kind: "gallery_layout"; protocol: "aethera.gallery.video" type Gallery_Source_Metrics = {has_rendered_frame: boolean; logical_completion_rate_fps: number; rendered_frame_rate_fps: number; logical_completion_count: number; rendered_frame_count: number; latest_completion_sequence: number; latest_rendered_sequence: number; latest_rendered_clock_sequence: number; frame_lag: number}; -type Gallery_Video_Metrics = {kind: "gallery_metrics"; protocol: "aethera.gallery.video"; version: 4; clock_sequence: number; - sampled_frame_count: number; encoded_frame_count: number; - target_frame_rate_fps: number; clock_delivery_rate_fps: number; sampled_frame_rate_fps: number; - encoded_frame_rate_fps: number; compose_average_ms: number; compose_p95_ms: number; - encode_average_ms: number; encode_p95_ms: number; sample_delay_average_ms: number; sample_delay_p95_ms: number; +type Gallery_Video_Metrics = {kind: "gallery_metrics"; protocol: "aethera.gallery.video"; version: 5; media_sequence: number; + received_frame_count: number; processed_frame_count: number; encoded_frame_count: number; + received_frame_rate_fps: number; processed_frame_rate_fps: number; encoded_frame_rate_fps: number; + compose_average_ms: number; compose_p95_ms: number; queue_delay_average_ms: number; queue_delay_p95_ms: number; + encode_average_ms: number; encode_p95_ms: number; publish_average_ms: number; publish_p95_ms: number; encoded_bytes: number; fresh_tiles: number; missing_tiles: number; - rejected_frames: number; skipped_sample_ticks: number; encoder_backend: string; + rejected_frames: number; pending_frame_count: number; encoder_backend: string; sources: Record}; type Gallery_Video_State = {status: Stream_Status; layout: Gallery_Layout | null; video_group: Gallery_H264_Group | null; presentation: Video_Presentation_Metrics; @@ -167,7 +168,7 @@ function valid_gallery_layout(value: unknown): value is Gallery_Layout { if (!value || typeof value !== "object") return false; const layout = value as Partial; return layout.kind === "gallery_layout" && - layout.protocol === "aethera.gallery.video" && layout.version === 4 && + layout.protocol === "aethera.gallery.video" && layout.version === 5 && layout.transport === "drogon_h264" && typeof layout.width === "number" && typeof layout.height === "number" && Boolean(layout.video) && @@ -179,9 +180,10 @@ function valid_gallery_metrics(value: unknown): value is Gallery_Video_Metrics { if (!value || typeof value !== "object") return false; const metrics = value as Partial; return metrics.kind === "gallery_metrics" && - metrics.protocol === "aethera.gallery.video" && metrics.version === 4 && + metrics.protocol === "aethera.gallery.video" && metrics.version === 5 && typeof metrics.encoded_frame_rate_fps === "number" && - typeof metrics.clock_delivery_rate_fps === "number" && + typeof metrics.received_frame_rate_fps === "number" && + typeof metrics.processed_frame_rate_fps === "number" && Boolean(metrics.sources); } @@ -234,7 +236,8 @@ const empty_video_presentation = (): Video_Presentation_Metrics => ({ frame_rate_fps: 0, presented_frames: 0, dropped_frames: 0, jitter_buffer_ms: 0, decode_processing_ms: 0, estimated_playout_delay_ms: 0, freeze_count: 0, - websocket_arrival_ms: 0, browser_queue_ms: 0, browser_render_ms: 0, + server_pipeline_ms: 0, websocket_arrival_ms: 0, source_arrival_ms: 0, + browser_queue_ms: 0, browser_render_ms: 0, frame_latency_ms: 0, current_time_seconds: 0, ready_state: 0 }); @@ -242,13 +245,14 @@ const empty_video_presentation = (): Video_Presentation_Metrics => ({ type Gallery_H264_Packet = { sequence: number; timestamp_microseconds: number; + source_unix_milliseconds: number; packed_unix_milliseconds: number; key_frame: boolean; annex_b: Uint8Array; }; function parse_gallery_h264_packet(buffer: ArrayBuffer): Gallery_H264_Packet { - if (buffer.byteLength < 40) throw new Error("H.264 帧头不完整"); + if (buffer.byteLength < 48) throw new Error("H.264 帧头不完整"); const bytes = new Uint8Array(buffer); if (bytes[0] !== 0x41 || bytes[1] !== 0x45 || bytes[2] !== 0x48 || bytes[3] !== 0x31) @@ -256,19 +260,23 @@ function parse_gallery_h264_packet(buffer: ArrayBuffer): Gallery_H264_Packet { const view = new DataView(buffer); const version = view.getUint16(4, true); const flags = view.getUint16(6, true); - const payload_bytes = view.getUint32(32, true); - if (version !== 1 || payload_bytes === 0 || - buffer.byteLength !== 40 + payload_bytes) + const payload_bytes = view.getUint32(40, true); + if (version !== 2 || payload_bytes === 0 || + buffer.byteLength !== 48 + payload_bytes) throw new Error("H.264 帧协议字段无效"); - const packed_nanoseconds = view.getBigInt64(24, true); + const source_nanoseconds = view.getBigInt64(24, true); + const packed_nanoseconds = view.getBigInt64(32, true); return { sequence: Number(view.getBigUint64(8, true)), timestamp_microseconds: Number(view.getBigInt64(16, true)), + source_unix_milliseconds: + Number(source_nanoseconds / 1_000_000n) + + Number(source_nanoseconds % 1_000_000n) / 1_000_000, packed_unix_milliseconds: Number(packed_nanoseconds / 1_000_000n) + Number(packed_nanoseconds % 1_000_000n) / 1_000_000, key_frame: (flags & 1) !== 0, - annex_b: new Uint8Array(buffer, 40, payload_bytes) + annex_b: new Uint8Array(buffer, 48, payload_bytes) }; } @@ -281,16 +289,25 @@ type Pending_H264_Frame = { sequence: number; received_performance_ms: number; received_unix_ms: number; + source_unix_milliseconds: number; packed_unix_milliseconds: number; }; +type Decoded_H264_Frame = Pending_H264_Frame & { + decoded_performance_ms: number; +}; + class Gallery_H264_Group { private layout: Gallery_Layout | null = null; private decoder: VideoDecoder | null = null; private disposed = false; private readonly surfaces = new Map(); private readonly pending = new Map(); - private latest_frame: VideoFrame | null = null; + private latest_frame: { + video: VideoFrame; + metadata: Decoded_H264_Frame; + } | null = null; + private readonly presentation_acknowledgements: number[] = []; private animation_frame = 0; private ready = false; private presented_frames = 0; @@ -298,7 +315,9 @@ class Gallery_H264_Group { private freeze_count = 0; private started_performance_ms = 0; private last_presented_performance_ms = 0; + private server_pipeline_ms = 0; private websocket_arrival_ms = 0; + private source_arrival_ms = 0; private browser_queue_ms = 0; private browser_render_ms = 0; private frame_latency_ms = 0; @@ -320,6 +339,12 @@ class Gallery_H264_Group { this.readiness(ready); } + private acknowledge_presentable_frames() { + for (const sequence of this.presentation_acknowledgements) + this.acknowledge(sequence); + this.presentation_acknowledgements.length = 0; + } + private reset_decoder() { if (this.disposed) return; if (this.decoder) { @@ -328,6 +353,9 @@ class Gallery_H264_Group { for (const frame of this.pending.values()) this.acknowledge(frame.sequence); this.pending.clear(); + this.acknowledge_presentable_frames(); + this.latest_frame?.video.close(); + this.latest_frame = null; this.decoder = null; if (!this.layout || typeof VideoDecoder === "undefined") { this.update_readiness(); @@ -403,14 +431,13 @@ class Gallery_H264_Group { sequence: packet.sequence, received_performance_ms, received_unix_ms, + source_unix_milliseconds: packet.source_unix_milliseconds, packed_unix_milliseconds: packet.packed_unix_milliseconds }); try { this.decoder.decode(new EncodedVideoChunk({ type: packet.key_frame ? "key" : "delta", timestamp: packet.timestamp_microseconds, - duration: Math.round(1_000_000 / - (this.layout?.frame_rate_fps ?? 30)), data: packet.annex_b })); } catch (failure) { @@ -423,31 +450,44 @@ class Gallery_H264_Group { private accept_decoded_frame(frame: VideoFrame) { const metadata = this.pending.get(frame.timestamp); - if (metadata) { - this.pending.delete(frame.timestamp); - this.acknowledge(metadata.sequence); - this.websocket_arrival_ms = Math.max(0, - metadata.received_unix_ms - - metadata.packed_unix_milliseconds); - this.decode_processing_ms = Math.max(0, - performance.now() - metadata.received_performance_ms); + if (!metadata) { + frame.close(); + return; } + this.pending.delete(frame.timestamp); + const decoded_performance_ms = performance.now(); + this.presentation_acknowledgements.push(metadata.sequence); + this.server_pipeline_ms = Math.max(0, + metadata.packed_unix_milliseconds - + metadata.source_unix_milliseconds); + this.websocket_arrival_ms = Math.max(0, + metadata.received_unix_ms - + metadata.packed_unix_milliseconds); + this.source_arrival_ms = Math.max(0, + metadata.received_unix_ms - + metadata.source_unix_milliseconds); + this.decode_processing_ms = Math.max(0, + decoded_performance_ms - metadata.received_performance_ms); if (this.latest_frame) { - this.latest_frame.close(); + this.latest_frame.video.close(); ++this.dropped_frames; } - this.latest_frame = frame; + this.latest_frame = { + video: frame, + metadata: {...metadata, decoded_performance_ms} + }; if (this.animation_frame === 0) this.animation_frame = requestAnimationFrame(() => this.present()); } private present() { this.animation_frame = 0; - const frame = this.latest_frame; + const decoded = this.latest_frame; this.latest_frame = null; const layout = this.layout; - if (!frame || !layout) { - frame?.close(); + if (!decoded || !layout) { + decoded?.video.close(); + this.acknowledge_presentable_frames(); return; } const started = performance.now(); @@ -455,20 +495,22 @@ class Gallery_H264_Group { const source = layout.plots[plot_id]; if (!source) continue; surface.context.drawImage( - frame, + decoded.video, source.column * layout.tile_width, source.row * layout.tile_height, layout.tile_width, layout.tile_height, 0, 0, layout.tile_width, layout.tile_height); } - frame.close(); + decoded.video.close(); const completed = performance.now(); this.browser_render_ms = completed - started; this.browser_queue_ms = Math.max(0, - started - (this.last_presented_performance_ms || started)); - this.frame_latency_ms = this.websocket_arrival_ms + - this.decode_processing_ms + this.browser_render_ms; + started - decoded.metadata.decoded_performance_ms); + this.frame_latency_ms = this.source_arrival_ms + + this.decode_processing_ms + this.browser_queue_ms + + this.browser_render_ms; + this.acknowledge_presentable_frames(); if (this.started_performance_ms === 0) this.started_performance_ms = completed; if (this.last_presented_performance_ms !== 0 && @@ -491,7 +533,9 @@ class Gallery_H264_Group { presented_frames: this.presented_frames, dropped_frames: this.dropped_frames, decode_processing_ms: this.decode_processing_ms, + server_pipeline_ms: this.server_pipeline_ms, websocket_arrival_ms: this.websocket_arrival_ms, + source_arrival_ms: this.source_arrival_ms, browser_queue_ms: this.browser_queue_ms, browser_render_ms: this.browser_render_ms, frame_latency_ms: this.frame_latency_ms, @@ -508,11 +552,12 @@ class Gallery_H264_Group { if (this.animation_frame !== 0) cancelAnimationFrame(this.animation_frame); this.animation_frame = 0; - this.latest_frame?.close(); + this.latest_frame?.video.close(); this.latest_frame = null; for (const frame of this.pending.values()) this.acknowledge(frame.sequence); this.pending.clear(); + this.acknowledge_presentable_frames(); if (this.decoder) { try { this.decoder.close(); } catch { /* already closed */ } } @@ -2728,11 +2773,30 @@ const Plot_Card = memo(function Plot_Card({plot, selected, policy, gallery, on_p 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 (!gallery.video_group || !source || !canvas) return; + if (!policy.visible || !gallery.video_group || !source || !canvas) + return; return gallery.video_group.register(plot.id, canvas); - }, [gallery.video_group, plot.id, source?.column, source?.row]); + }, [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}) : "尚未生成帧"; return
后台完成 {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.rendered_frame_rate_fps.toFixed(1) : metrics ? metrics.frame_rate_fps.toFixed(1) : "--.-"} FPS - 采样 {gallery.transport ? gallery.transport.sampled_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 WebCodecs {gallery.presentation.frame_rate_fps.toFixed(1)} FPS - 到达帧龄 {gallery.presentation.websocket_arrival_ms.toFixed(1)} ms - 浏览器帧龄 {gallery.presentation.frame_latency_ms.toFixed(1)} ms + 像素到达 {gallery.presentation.source_arrival_ms.toFixed(1)} ms + 像素到 Canvas {gallery.presentation.frame_latency_ms.toFixed(1)} ms 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