帧策略职责明确

2026-08-29 22:04:25 +08:00
parent 20be348a6a
commit bb6bcec901
29 changed files with 1743 additions and 333 deletions
@@ -46,17 +46,23 @@ void Graph_WebSocket::start() {
    d->stream = d->plot->subscribe(
        [weak](std::shared_ptr<const Plot_Stream_Frame> frame) {
            if (const auto socket = weak.lock())
-                socket->deliver_frame(std::move(frame));
+                return socket->deliver_frame(std::move(frame))
+                    ? Plot_Frame_Publication::completed
+                    : Plot_Frame_Publication::ignored;
+            return Plot_Frame_Publication::ignored;
        });
 }

-void Graph_WebSocket::deliver_frame(
+bool Graph_WebSocket::deliver_frame(
    std::shared_ptr<const Plot_Stream_Frame> frame) {
    if (!frame || frame->notification.empty() ||
        !d->attached.load(std::memory_order_acquire))
-        return;
-    try { d->send_handler(frame->notification); }
-    catch (...) {}
+        return false;
+    try {
+        d->send_handler(frame->notification);
+        return true;
+    }
+    catch (...) { return false; }
 }

 void Graph_WebSocket::receive(std::string_view message) {
@@ -22,7 +22,8 @@ public:
    void close() noexcept;

 private:
-    void deliver_frame(std::shared_ptr<const Plot_Stream_Frame> frame);
+    [[nodiscard]] bool deliver_frame(
+        std::shared_ptr<const Plot_Stream_Frame> frame);
    struct Private;
    std::unique_ptr<Private> d;
 };
@@ -135,18 +135,24 @@ bool Gallery_Video_Stream::Private::remove_consumer(
    }
    return false;
 }
-void Gallery_Video_Stream::Private::publish(
+bool Gallery_Video_Stream::Private::publish(
    std::shared_ptr<const Encoded_Video_Frame> video,
    std::string notification) noexcept {
-    if (!video && notification.empty()) return;
+    if (!video && notification.empty()) return false;
+    bool succeeded{};
    try {
        std::vector<Stream_Id> failed_consumers;
        for (const auto& slot : consumers) {
            const auto consumer = slot.load(std::memory_order_acquire);
            if (!consumer || !consumer->handler) continue;
            try {
-                (*consumer->handler)(Gallery_Stream_Frame{
-                    video, notification});
+                auto accepted_video = video;
+                if (accepted_video && consumer->readiness &&
+                    !(*consumer->readiness)())
+                    accepted_video.reset();
+                if (!accepted_video && notification.empty()) continue;
+                succeeded |= (*consumer->handler)(Gallery_Stream_Frame{
+                    std::move(accepted_video), notification});
            }
            catch (...) {
                failed_consumers.push_back(consumer->id);
@@ -156,10 +162,39 @@ void Gallery_Video_Stream::Private::publish(
            static_cast<void>(remove_consumer(id));
    }
    catch (...) {}
+    return succeeded;
+}
+void Gallery_Video_Stream::Private::release_pending_frames() noexcept {
+    Pending_Plot_Frame pending;
+    while (pending_frames.try_dequeue(pending)) {
+        try {
+            if (pending.slot < sources.size() && pending.frame &&
+                pending.frame->pixels)
+                sources[pending.slot].entry.plot->submit_publication_feedback({
+                    .completed_at = std::chrono::steady_clock::now(),
+                    .sequence = pending.frame->pixels->sequence,
+                    .succeeded = false});
+        }
+        catch (...) {}
+        pending = {};
+    }
 }
 void Gallery_Video_Stream::Private::fail(
    std::exception_ptr failure) noexcept {
    if (failed.exchange(true, std::memory_order_acq_rel)) return;
+    if (active_frame && !active_frame->publication_feedback_submitted &&
+        active_frame->source_slot < sources.size()) {
+        try {
+            sources[active_frame->source_slot].entry.plot
+                ->submit_publication_feedback({
+                    .completed_at = std::chrono::steady_clock::now(),
+                    .sequence = active_frame->plot_frame_sequence,
+                    .succeeded = false});
+            active_frame->publication_feedback_submitted = true;
+        }
+        catch (...) {}
+    }
+    release_pending_frames();
    try {
        terminal_failure = exception_description(failure);
        const auto notification = nlohmann::json{
@@ -168,24 +203,31 @@ void Gallery_Video_Stream::Private::fail(
                    {"version", 5},
                    {"message", *terminal_failure}
                }.dump();
-        publish({}, notification);
+        static_cast<void>(publish({}, notification));
    }
    catch (...) {}
 }
-void Gallery_Video_Stream::Private::accept_frame(
+Plot_Frame_Publication Gallery_Video_Stream::Private::accept_frame(
    std::size_t slot, std::shared_ptr<const Plot_Stream_Frame> frame) {
    if (stopping.load(std::memory_order_acquire) ||
        failed.load(std::memory_order_acquire) || !frame || !frame->pixels ||
        !frame->pixels->pixels || !consumer_accepts())
-        return;
+        return Plot_Frame_Publication::ignored;
    Pending_Plot_Frame pending{
-        slot, std::move(frame->pixels), std::chrono::steady_clock::now(),
+        slot, std::move(frame),
+        std::chrono::steady_clock::now(),
        std::chrono::duration_cast<std::chrono::nanoseconds>(
            std::chrono::system_clock::now().time_since_epoch())};
    if (!pending_frames.enqueue(std::move(pending))) throw std::bad_alloc{};
+    if (stopping.load(std::memory_order_acquire) ||
+        failed.load(std::memory_order_acquire)) {
+        release_pending_frames();
+        return Plot_Frame_Publication::ignored;
+    }
    received_frame_count.fetch_add(1, std::memory_order_relaxed);
    media_work_generation.fetch_add(1, std::memory_order_release);
    arm_media_task(object->weak_from_this());
+    return Plot_Frame_Publication::asynchronous;
 }
 void Gallery_Video_Stream::Private::update_metrics(
    const Active_Media_Frame& frame,
@@ -322,11 +364,16 @@ void Gallery_Video_Stream::Private::begin_media_frame() {
    ++processed_frame_count;
    const auto started = std::chrono::steady_clock::now();
    const auto accepted = atlas->accept_frame(
-        pending.slot, std::move(pending.frame));
+        pending.slot, pending.frame->pixels);
    if (accepted ==
-        detail::Gallery_Frame_Atlas::Accept_Frame_Result::invalid_frame)
+        detail::Gallery_Frame_Atlas::Accept_Frame_Result::invalid_frame) {
+        sources[pending.slot].entry.plot->submit_publication_feedback({
+            .completed_at = std::chrono::steady_clock::now(),
+            .sequence = pending.frame->pixels->sequence,
+            .succeeded = false});
        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)
@@ -337,7 +384,8 @@ void Gallery_Video_Stream::Private::begin_media_frame() {
        atlas->compose(), next_media_sequence++, presentation_time,
        pending.source_time_unix,
        std::chrono::duration<double, std::milli>(
-            started - pending.enqueued_at).count()};
+            started - pending.enqueued_at).count(),
+        pending.slot, pending.frame->pixels->sequence, false, false};
    static_cast<void>(compose_ms.submit(
        std::chrono::duration<double, std::milli>(
            std::chrono::steady_clock::now() - started).count()));
@@ -371,13 +419,27 @@ void Gallery_Video_Stream::Private::publish_video_frame() {
    if (!active_video) return;
    ++encoded_frame_count;
    const auto started = std::chrono::steady_clock::now();
-    publish(active_video, {});
+    active_frame->publication_succeeded = publish(active_video, {});
    static_cast<void>(publish_ms.submit(
        std::chrono::duration<double, std::milli>(
            std::chrono::steady_clock::now() - started).count()));
 }
+void Gallery_Video_Stream::Private::submit_publication_feedback() {
+    if (!active_frame || active_frame->publication_feedback_submitted) return;
+    if (active_frame->source_slot >= sources.size())
+        throw std::logic_error(
+            "gallery publication feedback source is invalid");
+    sources[active_frame->source_slot].entry.plot->submit_publication_feedback({
+        .completed_at = std::chrono::steady_clock::now(),
+        .sequence = active_frame->plot_frame_sequence,
+        .succeeded = active_frame->publication_succeeded});
+    active_frame->publication_feedback_submitted = true;
+}
 void Gallery_Video_Stream::Private::complete_media_frame() {
    if (!active_frame) return;
+    if (!active_frame->publication_feedback_submitted)
+        throw std::logic_error(
+            "gallery media completion preceded policy feedback task");
    const auto encoded_bytes = active_video ? active_video->annex_b.size() : 0U;
    const auto encoder_backend = active_video
        ? std::optional{active_video->backend}
@@ -503,6 +565,17 @@ void Gallery_Video_Stream::bind_plots() {
    publish_video.describe("owner", "gallery")
                   .describe("transport", "Drogon WebSocket")
                   .describe("stage", "H.264 access unit enqueue");
+    auto publication_feedback = media->add(
+        "frame.policy.publish.feedback", [weak] {
+        if (const auto owner = weak.lock()) {
+            auto& owner_data = static_cast<Private&>(*owner->d);
+            try { owner_data.submit_publication_feedback(); }
+            catch (...) { owner_data.fail(std::current_exception()); }
+        }
+    });
+    publication_feedback.describe("owner", "frame_policy")
+                        .describe("stage", "media publication feedback")
+                        .describe("execution", "Taskflow worker");
    auto complete = media->add("gallery.media.complete", [weak] {
        if (const auto owner = weak.lock()) {
            auto& owner_data = static_cast<Private&>(*owner->d);
@@ -511,10 +584,12 @@ void Gallery_Video_Stream::bind_plots() {
        }
    });
    complete.describe("owner", "gallery")
-            .describe("stage", "metrics and frame ownership release");
+            .describe("stage", "metrics and Gallery stage ownership release")
+            .describe("policy_lifecycle", "already reported by preceding task");
    compose.precede(encode);
    encode.precede(publish_video);
-    publish_video.precede(complete);
+    publish_video.precede(publication_feedback);
+    publication_feedback.precede(complete);
    data.media_graph = media;

    for (std::size_t slot = 0; slot < data.sources.size(); ++slot) {
@@ -522,11 +597,17 @@ void Gallery_Video_Stream::bind_plots() {
        source.stream = source.entry.plot->subscribe(
            [weak, slot](std::shared_ptr<const Plot_Stream_Frame> frame) {
                const auto owner = weak.lock();
-                if (!owner) return;
+                if (!owner) return Plot_Frame_Publication::ignored;
                auto& owner_data = static_cast<Private&>(*owner->d);
-                if (owner_data.stopping.load(std::memory_order_acquire)) return;
-                try { owner_data.accept_frame(slot, std::move(frame)); }
-                catch (...) { owner_data.fail(std::current_exception()); }
+                if (owner_data.stopping.load(std::memory_order_acquire))
+                    return Plot_Frame_Publication::ignored;
+                try {
+                    return owner_data.accept_frame(slot, std::move(frame));
+                }
+                catch (...) {
+                    owner_data.fail(std::current_exception());
+                    return Plot_Frame_Publication::ignored;
+                }
            });
        source.entry.plot->configure_stream(
            source.stream, tile_width, tile_height);
@@ -614,6 +695,7 @@ void Gallery_Video_Stream::shutdown() noexcept {
    }
    for (auto& consumer : data.consumers)
        consumer.store({}, std::memory_order_release);
+    data.release_pending_frames();
 }
 std::string Gallery_Video_Stream::layout_description() const {
    const auto& data = static_cast<const Private&>(*d);
@@ -29,7 +29,7 @@ struct Gallery_Video_Stream : Def<Gallery_Video_Stream, Root>,
    struct Private;

    using Stream_Id = std::uint64_t;
-    using Stream_Handler = std::function<void(Gallery_Stream_Frame)>;
+    using Stream_Handler = std::function<bool(Gallery_Stream_Frame)>;
    using Transport_Readiness = std::function<bool()>;
    using Transport_Diagnostics = std::function<nlohmann::json()>;

@@ -29,7 +29,7 @@ struct Gallery_Video_Stream::Private : Prev_Private {

    struct Pending_Plot_Frame {
        std::size_t slot{};                                  /* 该完成帧所属的稳定图集槽位。 */
-        std::shared_ptr<const Plot_Pixel_Frame> frame{};      /* 已通过所属 Plot 帧策略的不可变像素帧。 */
+        std::shared_ptr<const Plot_Stream_Frame> frame{};     /* 持有像素帧以及其隐藏的 Plot 交付生命周期。 */
        std::chrono::steady_clock::time_point enqueued_at{};  /* 进入媒体串行队列的单调时刻。 */
        std::chrono::nanoseconds source_time_unix{};          /* 进入媒体串行队列的 Unix 时间。 */
    };
@@ -40,6 +40,10 @@ struct Gallery_Video_Stream::Private : Prev_Private {
        std::chrono::microseconds presentation_time{};   /* 媒体流起点以来的单调显示时间。 */
        std::chrono::nanoseconds source_time_unix{};      /* 对应 Plot 像素进入媒体流水线的 Unix 时间。 */
        double queue_delay_ms{};                         /* Plot 发布到开始合成的队列时间。 */
+        std::size_t source_slot{};                        /* 发布反馈路由到唯一来源 Plot 的槽位。 */
+        std::uint64_t plot_frame_sequence{};              /* Frame_Policy 当前管理的来源帧序号。 */
+        bool publication_succeeded{};                     /* 媒体传输是否至少成功接收一次发布。 */
+        bool publication_feedback_submitted{};            /* 防止失败清理重复提交策略反馈。 */
    };

    static constexpr std::size_t maximum_consumers{32};
@@ -57,7 +61,7 @@ struct Gallery_Video_Stream::Private : Prev_Private {
    std::atomic_uint64_t media_work_generation{}; /* 新完成帧入队后推进，封闭任务退出竞争窗口。 */
    detail::FFmpeg_Frame_Transport ffmpeg_transport; /* 仅由媒体 DAG 串行访问的编码上下文。 */
    std::optional<Active_Media_Frame> active_frame{}; /* 当前媒体 DAG 独占的图集帧。 */
-    std::shared_ptr<const Encoded_Video_Frame> active_video{}; /* 当前编码结果的共享所有权。 */
+    std::shared_ptr<const Encoded_Video_Frame> active_video{}; /* 当前编码结果的共享媒体所有权，不承担帧策略状态。 */
    std::atomic_uint64_t next_consumer_id{1}; /* 订阅身份生成器。 */
    std::array<std::atomic<std::shared_ptr<const Consumer>>,
               maximum_consumers> consumers{}; /* 订阅关系的权威原子槽位。 */
@@ -94,15 +98,18 @@ struct Gallery_Video_Stream::Private : Prev_Private {
    void initialize(std::vector<Plot_Entry> plots);
    [[nodiscard]] bool consumer_accepts() const noexcept;
    [[nodiscard]] bool remove_consumer(Stream_Id stream) noexcept;
-    void publish(std::shared_ptr<const Encoded_Video_Frame> video,
-                 std::string notification) noexcept;
+    [[nodiscard]] bool publish(
+        std::shared_ptr<const Encoded_Video_Frame> video,
+        std::string notification) noexcept;
+    void release_pending_frames() noexcept;
    void fail(std::exception_ptr failure) noexcept;
-    void accept_frame(std::size_t slot,
-                      std::shared_ptr<const Plot_Stream_Frame> frame);
+    [[nodiscard]] Plot_Frame_Publication accept_frame(
+        std::size_t slot, std::shared_ptr<const Plot_Stream_Frame> frame);
    void arm_media_task(std::weak_ptr<Gallery_Video_Stream> lifetime);
    void begin_media_frame();
    void encode_media_frame();
    void publish_video_frame();
+    void submit_publication_feedback();
    void complete_media_frame();
    void update_metrics(const Active_Media_Frame& frame,
                        std::size_t encoded_bytes,
@@ -1,4 +1,5 @@
 #include "Gallery_Video_WebSocket.hpp"
+#include <concurrentqueue-1.0.5/concurrentqueue.h>
 #include <nlohmann/json.hpp>
 #include <algorithm>
 #include <atomic>
@@ -63,12 +64,19 @@ struct Gallery_Video_WebSocket::Private final {
    std::string connection_id{};
    std::atomic_bool attached{};
    std::atomic_bool decoder_ready{};
-    std::atomic_size_t outstanding_frames{};
+    moodycamel::ConcurrentQueue<
+        std::shared_ptr<const Encoded_Video_Frame>> outstanding_frames{};
    std::atomic_uint64_t latest_sent_sequence{};
    std::atomic_uint64_t latest_acknowledged_sequence{};
    std::atomic_uint64_t sent_count{};
    std::atomic_uint64_t acknowledged_count{};
    std::atomic_uint64_t rejected_count{};
+
+    void release_outstanding_frames() noexcept {
+        std::shared_ptr<const Encoded_Video_Frame> outstanding;
+        while (outstanding_frames.try_dequeue(outstanding))
+            outstanding.reset();
+    }
 };

 Gallery_Video_WebSocket::Gallery_Video_WebSocket(
@@ -93,7 +101,8 @@ void Gallery_Video_WebSocket::start() {
        d->connection_id,
        [weak](Gallery_Stream_Frame frame) {
            if (const auto socket = weak.lock())
-                socket->deliver(std::move(frame));
+                return socket->deliver(std::move(frame));
+            return false;
        },
        [weak] {
            const auto socket = weak.lock();
@@ -111,8 +120,8 @@ void Gallery_Video_WebSocket::start() {
                {"kind", "drogon_h264_state"},
                {"decoder_ready", socket->d->decoder_ready.load(
                    std::memory_order_acquire)},
-                {"outstanding_frames", socket->d->outstanding_frames.load(
-                    std::memory_order_acquire)},
+                {"outstanding_frames",
+                    socket->d->outstanding_frames.size_approx()},
                {"latest_sent_sequence", socket->d->latest_sent_sequence.load(
                    std::memory_order_relaxed)},
                {"latest_acknowledged_sequence",
@@ -132,30 +141,44 @@ void Gallery_Video_WebSocket::start() {
                         drogon::WebSocketMessageType::Text);
 }

-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))) {
+bool Gallery_Video_WebSocket::deliver(Gallery_Stream_Frame frame) {
+    if (!d->attached.load(std::memory_order_acquire)) return false;
+    bool succeeded{};
+    const bool has_video = static_cast<bool>(frame.video);
+    if (frame.video) {
+        succeeded = queue_video_frame(std::move(frame.video));
+    }
+    if (has_video && !succeeded) {
        d->rejected_count.fetch_add(1, std::memory_order_relaxed);
        d->stream->request_video_key_frame();
    }
    if (!frame.notification.empty()) {
        if (const auto connection = d->connection.lock();
-            connection && connection->connected())
+            connection && connection->connected()) {
            connection->send(std::move(frame.notification),
                             drogon::WebSocketMessageType::Text);
+            succeeded = true;
+        }
    }
+    return succeeded;
 }

 bool Gallery_Video_WebSocket::queue_video_frame(
    std::shared_ptr<const Encoded_Video_Frame> frame) {
-    if (!frame || !d->attached.load(std::memory_order_acquire))
+    if (!frame || !d->attached.load(std::memory_order_acquire) ||
+        !d->decoder_ready.load(std::memory_order_acquire))
        return false;
    const auto connection = d->connection.lock();
    if (!connection || !connection->connected()) return false;
    const auto sequence = frame->sequence;
    try {
        auto packet = websocket_packet(*frame);
-        d->outstanding_frames.fetch_add(1, std::memory_order_acq_rel);
+        if (!d->outstanding_frames.enqueue(frame)) throw std::bad_alloc{};
+        if (!d->attached.load(std::memory_order_acquire) ||
+            !d->decoder_ready.load(std::memory_order_acquire)) {
+            d->release_outstanding_frames();
+            return false;
+        }
        connection->send(
            reinterpret_cast<const char*>(packet.data()), packet.size(),
            drogon::WebSocketMessageType::Binary);
@@ -164,30 +187,25 @@ bool Gallery_Video_WebSocket::queue_video_frame(
        return true;
    }
    catch (...) {
-        d->outstanding_frames.fetch_sub(1, std::memory_order_release);
        close();
        return false;
    }
 }

 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) {
-        if (!d->latest_acknowledged_sequence.compare_exchange_weak(
-                acknowledged, sequence, std::memory_order_acq_rel,
-                std::memory_order_acquire))
-            continue;
-        auto outstanding = d->outstanding_frames.load(
-            std::memory_order_acquire);
-        while (outstanding != 0 &&
-            !d->outstanding_frames.compare_exchange_weak(
-                outstanding, outstanding - 1U,
-                std::memory_order_acq_rel,
-                std::memory_order_acquire)) {}
-        d->acknowledged_count.fetch_add(1, std::memory_order_relaxed);
+    if (sequence == 0 || sequence <= d->latest_acknowledged_sequence.load(
+            std::memory_order_acquire))
+        return;
+    std::shared_ptr<const Encoded_Video_Frame> acknowledged;
+    if (!d->outstanding_frames.try_dequeue(acknowledged) ||
+        !acknowledged || acknowledged->sequence != sequence) {
+        d->rejected_count.fetch_add(1, std::memory_order_relaxed);
+        close();
        return;
    }
+    d->latest_acknowledged_sequence.store(
+        sequence, std::memory_order_release);
+    d->acknowledged_count.fetch_add(1, std::memory_order_relaxed);
 }

 void Gallery_Video_WebSocket::receive(std::string_view message) {
@@ -202,6 +220,7 @@ void Gallery_Video_WebSocket::receive(std::string_view message) {
    }
    if (kind == "h264_stream_not_ready") {
        d->decoder_ready.store(false, std::memory_order_release);
+        d->release_outstanding_frames();
        return;
    }
    if (kind == "h264_frame_accepted") {
@@ -213,14 +232,16 @@ void Gallery_Video_WebSocket::receive(std::string_view message) {
 }

 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);
+    if (d->attached.exchange(false, std::memory_order_acq_rel)) {
+        try {
+            if (d->subscription != 0)
+                d->stream->unsubscribe(d->subscription);
+        }
+        catch (...) {}
+        d->subscription = 0;
    }
-    catch (...) {}
-    d->subscription = 0;
    d->decoder_ready.store(false, std::memory_order_release);
-    d->outstanding_frames.store(0, std::memory_order_release);
+    d->release_outstanding_frames();
 }

 Gallery_Video_WebSocket_Controller::Gallery_Video_WebSocket_Controller(
@@ -22,7 +22,7 @@ struct Gallery_Video_WebSocket final
    void close() noexcept;

 private:
-    void deliver(Gallery_Stream_Frame frame);
+    [[nodiscard]] bool deliver(Gallery_Stream_Frame frame);
    [[nodiscard]] bool queue_video_frame(
        std::shared_ptr<const Encoded_Video_Frame> frame);
    void acknowledge_video_frame(std::uint64_t sequence);