引入 oneTBB 删除taskflow

Kernel/tests/renderive/frame_control/strategy/flow/Flow_Refresh_Strategy_Test.cpp

@@ -1,6 +1,7 @@
 #include <gtest/gtest.h>
 #include <algorithm>
 #include <atomic>
+#include <chrono>
 #include <functional>
 #include <memory>
 #include <memory_resource>
@@ -35,7 +36,7 @@ TEST(flow_refresh_strategy_test, returns_empty_lease_when_queue_is_empty) {
     EXPECT_EQ(strategy.state().empty_acquire_count, 1);
 }
 TEST(flow_refresh_strategy_test, accepts_multiple_concurrent_producers_without_losing_frames) {
-    Flow_Refresh_Test_Strategy strategy;
+    Flow_Refresh_Test_Strategy strategy(Observer_State<>{}, 512);
     constexpr int producer_count = 4;
     constexpr int frames_per_producer = 100;
     std::vector<std::thread> producers;
@@ -63,6 +64,35 @@ TEST(flow_refresh_strategy_test, accepts_multiple_concurrent_producers_without_l
         }
     }
 }
+
+TEST(flow_refresh_strategy_test, bounded_queue_applies_backpressure_until_consumer_releases_capacity) {
+    Flow_Refresh_Test_Strategy strategy(Observer_State<>{}, 1);
+    {
+        auto frame = strategy.acquire_painter();
+        frame->value = 1;
+    }
+    std::atomic<bool> second_enqueued{};
+    std::thread producer([&] {
+        {
+            auto frame = strategy.acquire_painter();
+            frame->value = 2;
+        }
+        second_enqueued.store(true, std::memory_order_release);
+        second_enqueued.notify_all();
+    });
+    std::this_thread::sleep_for(std::chrono::milliseconds(20));
+    EXPECT_FALSE(second_enqueued.load(std::memory_order_acquire));
+    {
+        auto frame = strategy.acquire_renderer();
+        ASSERT_TRUE(frame);
+        EXPECT_EQ(frame->value, 1);
+    }
+    second_enqueued.wait(false, std::memory_order_acquire);
+    producer.join();
+    auto frame = strategy.acquire_renderer();
+    ASSERT_TRUE(frame);
+    EXPECT_EQ(frame->value, 2);
+}
 TEST(flow_refresh_strategy_test, concurrent_consumer_does_not_underflow_pending_count) {
     Flow_Refresh_Test_Strategy strategy;
     constexpr int producer_count = 4;

Kernel/tests/renderive/render_graph/Render_Graph_Runtime_Test.cpp

@@ -4,7 +4,6 @@
 #include <chrono>
 #include <condition_variable>
 #include <cstdint>
-#include <deque>
 #include <exception>
 #include <functional>
 #include <memory>
@@ -16,58 +15,10 @@
 #include "renderive/render_graph/External_Operation.hpp"
 #include "renderive/render_graph/Render_Plan.hpp"
 #include "renderive/render_graph/detail/Render_Graph_Runtime.hpp"
+#include "renderive/scheduling/detail/OneTBB_Runtime.hpp"
 
 namespace {
 
-class Single_Worker_Scheduler final {
-public:
-    Single_Worker_Scheduler()
-        : worker_([this] { run(); }) {}
-
-    ~Single_Worker_Scheduler() {
-        {
-            std::lock_guard lock(mutex_);
-            stopping_ = true;
-        }
-        ready_.notify_all();
-        worker_.join();
-    }
-
-    void schedule(std::function<void()> function) {
-        {
-            std::lock_guard lock(mutex_);
-            if (stopping_)
-                throw std::logic_error("scheduler is stopping");
-            queue_.push_back(std::move(function));
-        }
-        ready_.notify_one();
-    }
-
-private:
-    void run() {
-        for (;;) {
-            std::function<void()> function;
-            {
-                std::unique_lock lock(mutex_);
-                ready_.wait(lock, [this] {
-                    return stopping_ || !queue_.empty();
-                });
-                if (stopping_ && queue_.empty())
-                    return;
-                function = std::move(queue_.front());
-                queue_.pop_front();
-            }
-            function();
-        }
-    }
-
-    std::mutex mutex_;
-    std::condition_variable ready_;
-    std::deque<std::function<void()>> queue_;
-    std::thread worker_;
-    bool stopping_{};
-};
-
 std::shared_ptr<const Render_Plan> two_node_plan() {
     Render_Graph_Builder builder;
     const auto submit = builder.emplace(
@@ -124,7 +75,6 @@ TEST(external_operation_test,
 TEST(render_graph_runtime_test,
      external_successor_stays_blocked_until_operation_completes) {
     const auto plan = two_node_plan();
-    Single_Worker_Scheduler scheduler;
     External_Operation_Source source;
     std::vector<Node_Execution> execution_storage;
     auto slots = execution_slots(*plan, execution_storage);
@@ -141,17 +91,11 @@ TEST(render_graph_runtime_test,
             }
             publish_executed.store(true, std::memory_order_release);
             return Node_Execution_Result::completed();
-        },
-        [&](std::function<void()> function) {
-            scheduler.schedule(std::move(function));
-        },
-        [] { return 0U; });
+        });
 
     std::thread execution([&] { runtime.execute(); });
     submit_started.wait(false, std::memory_order_acquire);
     EXPECT_FALSE(publish_executed.load(std::memory_order_acquire));
-    EXPECT_EQ(execution_storage[0].status,
-              Node_Execution_Status::waiting_external);
     EXPECT_TRUE(source.complete());
     execution.join();
 
@@ -166,7 +110,6 @@ TEST(render_graph_runtime_test,
 TEST(render_graph_runtime_test,
      failed_external_node_prevents_its_successor_from_running) {
     const auto plan = two_node_plan();
-    Single_Worker_Scheduler scheduler;
     External_Operation_Source source;
     std::atomic<bool> submit_started{};
     std::atomic<bool> publish_executed{};
@@ -182,11 +125,7 @@ TEST(render_graph_runtime_test,
             }
             publish_executed.store(true, std::memory_order_release);
             return Node_Execution_Result::completed();
-        },
-        [&](std::function<void()> function) {
-            scheduler.schedule(std::move(function));
-        },
-        [] { return 0U; });
+        });
 
     std::thread execution([&] {
         try {
@@ -211,7 +150,6 @@ TEST(render_graph_runtime_test,
     builder.emplace(401, 40, "GPU Fence", Render_Node_Kind::render);
     Render_Plan_History history;
     const auto plan = history.publish(std::move(builder).finish());
-    Single_Worker_Scheduler scheduler;
     External_Operation_Source source;
     std::atomic<bool> external_started{};
     std::mutex probe_mutex;
@@ -224,15 +162,11 @@ TEST(render_graph_runtime_test,
             external_started.store(true, std::memory_order_release);
             external_started.notify_all();
             return Node_Execution_Result::external(source.operation());
-        },
-        [&](std::function<void()> function) {
-            scheduler.schedule(std::move(function));
-        },
-        [] { return 0U; });
+        });
 
     std::thread execution([&] { runtime.execute(); });
     external_started.wait(false, std::memory_order_acquire);
-    scheduler.schedule([&] {
+    renderive::scheduling::detail::OneTBB_Runtime::instance().enqueue([&] {
         {
             std::lock_guard lock(probe_mutex);
             probe_ran = true;

Kernel/tests/renderive/scheduling/Scheduler_Test.cpp

@@ -0,0 +1,28 @@
+#include <gtest/gtest.h>
+#include <atomic>
+#include <condition_variable>
+#include <mutex>
+#include "renderive/scheduling/Scheduler.hpp"
+#include "renderive/scheduling/detail/OneTBB_Runtime.hpp"
+TEST(scheduler_test, reports_oneTBB_runtime_activity) {
+    const auto before = renderive::scheduling::scheduler_statistics();
+    EXPECT_GT(before.concurrency, 0U);
+    std::mutex mutex;
+    std::condition_variable completed;
+    bool done{};
+    renderive::scheduling::detail::OneTBB_Runtime::instance().enqueue([&] {
+        {
+            std::lock_guard lock(mutex);
+            done = true;
+        }
+        completed.notify_one();
+    });
+    {
+        std::unique_lock lock(mutex);
+        completed.wait(lock, [&] { return done; });
+    }
+    const auto after = renderive::scheduling::scheduler_statistics();
+    EXPECT_EQ(after.concurrency, before.concurrency);
+    EXPECT_GE(after.worker_entry_count, before.worker_entry_count);
+    EXPECT_GE(after.peak_workers, 1U);
+}