ae/Renderive
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

改进

Browse Source
This commit is contained in:
种太阳的热带美短
2026-08-15 17:44:35 +08:00
parent b4094ede44
commit d48f6608ad
18 changed files with 310 additions and 280 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kernel/src/renderive/render_graph/Render_Plan.cpp
+10 -5
View File
@@ -104,21 +104,26 @@ bool Render_Graph_Builder::reaches(std::size_t from, std::size_t target) const {
std::shared_ptr<const Render_Plan> Render_Plan_History::publish(Render_Graph graph) {
normalize(graph);
std::lock_guard lock(mutex_);
if (!plans_.empty() && same_topology(plans_.back()->graph, graph))
return plans_.back();
if (next_version_ == 0 ||
next_version_ == std::numeric_limits<Render_Plan_Version>::max())
const auto existing = std::find_if(plans_.begin(), plans_.end(), [&graph](const auto& plan) {
return same_topology(plan->graph, graph);
});
if (existing != plans_.end()) {
current_ = *existing;
return current_;
}
if (next_version_ == 0 || next_version_ == std::numeric_limits<Render_Plan_Version>::max())
throw std::overflow_error("render plan version exhausted");
auto plan = std::make_shared<Render_Plan>();
plan->version = next_version_++;
plan->graph = std::move(graph);
plans_.push_back(plan);
current_ = plan;
return plan;
}
std::shared_ptr<const Render_Plan> Render_Plan_History::current() const {
std::lock_guard lock(mutex_);
return plans_.empty() ? nullptr : plans_.back();
return current_;
}
std::shared_ptr<const Render_Plan> Render_Plan_History::find(Render_Plan_Version version) const {
Kernel/src/renderive/render_graph/Render_Plan.hpp
+1
View File
@@ -68,6 +68,7 @@ public:
private:
mutable std::mutex mutex_;
std::vector<std::shared_ptr<const Render_Plan>> plans_;
std::shared_ptr<const Render_Plan> current_;
Render_Plan_Version next_version_{1};
};
Kernel/src/renderive/render_graph/detail/Render_Graph_Runtime.cpp
+50 -38
View File
@@ -1,7 +1,6 @@
#include "Render_Graph_Runtime.hpp"
#include <atomic>
#include <exception>
#include <limits>
#include <mutex>
#include <stdexcept>
#include <unordered_map>
@@ -19,14 +18,7 @@ struct Render_Graph_Runtime::State : std::enable_shared_from_this<Render_Graph_R
std::unique_ptr<Execute_Node_Type> execute;
bool root{};
};
State(const Render_Plan& plan, std::span<Node_Execution* const> execution_slots, Execute_Node execute)
: executions(execution_slots.begin(), execution_slots.end()), execute_node(std::move(execute)) {
if (!execute_node)
throw std::invalid_argument("render graph node executor is empty");
if (executions.empty())
executions.resize(plan.graph.nodes.size());
if (executions.size() != plan.graph.nodes.size())
throw std::invalid_argument("render graph execution slot count differs from plan");
explicit State(const Render_Plan& plan) : execution_count(plan.graph.nodes.size()) {
nodes.resize(plan.graph.nodes.size());
std::unordered_map<Render_Node_Id, std::size_t> indices;
indices.reserve(plan.graph.nodes.size());
@@ -46,30 +38,51 @@ struct Render_Graph_Runtime::State : std::enable_shared_from_this<Render_Graph_R
});
oneapi::tbb::flow::make_edge(*nodes[index].ready, *nodes[index].execute);
}
for (const auto& edge : plan.graph.edges) {
const std::size_t from = indices.at(edge.from);
const std::size_t to = indices.at(edge.to);
oneapi::tbb::flow::make_edge(*nodes[from].execute, *nodes[to].ready);
}
for (const auto& edge : plan.graph.edges)
oneapi::tbb::flow::make_edge(*nodes[indices.at(edge.from)].execute, *nodes[indices.at(edge.to)].ready);
}
void execute() {
if (started.exchange(true, std::memory_order_acq_rel))
throw std::logic_error("render graph runtime already executed");
auto& arena = renderive::scheduling::detail::OneTBB_Runtime::instance().arena();
arena.execute([this] {
for (auto& node : nodes) {
if (node.root)
node.ready->try_put(Message{});
void execute(std::span<Node_Execution* const> execution_slots, Execute_Node execute) {
if (running.exchange(true, std::memory_order_acq_rel))
throw std::logic_error("render graph runtime is already executing");
try {
if (!execute)
throw std::invalid_argument("render graph node executor is empty");
if (!execution_slots.empty() && execution_slots.size() != execution_count)
throw std::invalid_argument("render graph execution slot count differs from plan");
graph.reset();
executions.assign(execution_slots.begin(), execution_slots.end());
if (executions.empty())
executions.resize(execution_count);
execute_node = std::move(execute);
failed.store(false, std::memory_order_release);
{
std::lock_guard lock(error_mutex);
first_exception = nullptr;
}
graph.wait_for_all();
});
std::exception_ptr error;
{
std::lock_guard lock(error_mutex);
error = first_exception;
auto& arena = renderive::scheduling::detail::OneTBB_Runtime::instance().arena();
arena.execute([this] {
for (auto& node : nodes) {
if (node.root)
node.ready->try_put(Message{});
}
graph.wait_for_all();
});
std::exception_ptr error;
{
std::lock_guard lock(error_mutex);
error = first_exception;
}
execute_node = {};
executions.clear();
running.store(false, std::memory_order_release);
if (error)
std::rethrow_exception(error);
} catch (...) {
execute_node = {};
executions.clear();
running.store(false, std::memory_order_release);
throw;
}
if (error)
std::rethrow_exception(error);
}
void mark_ready(std::size_t index) noexcept {
if (failed.load(std::memory_order_acquire))
@@ -113,9 +126,9 @@ struct Render_Graph_Runtime::State : std::enable_shared_from_this<Render_Graph_R
auto* gateway_ptr = &gateway;
result.operation().on_complete([self = std::move(self), gateway_ptr, index](std::exception_ptr error) {
const std::uint64_t end = render_clock_now_ns();
if (error) {
if (error)
self->fail(index, std::move(error), end);
} else {
else {
self->complete_external(index, end);
gateway_ptr->try_put(Message{});
}
@@ -158,18 +171,17 @@ struct Render_Graph_Runtime::State : std::enable_shared_from_this<Render_Graph_R
}
oneapi::tbb::flow::graph graph;
std::vector<Node> nodes;
const std::size_t execution_count{};
std::vector<Node_Execution*> executions;
Execute_Node execute_node;
std::mutex error_mutex;
std::exception_ptr first_exception;
std::atomic_bool started{};
std::atomic_bool running{};
std::atomic_bool failed{};
};
Render_Graph_Runtime::Render_Graph_Runtime(const Render_Plan& plan, std::span<Node_Execution* const> executions, Execute_Node execute_node)
: state_(std::make_shared<State>(plan, executions, std::move(execute_node))) {
}
Render_Graph_Runtime::Render_Graph_Runtime(const Render_Plan& plan) : state_(std::make_shared<State>(plan)) {}
Render_Graph_Runtime::~Render_Graph_Runtime() = default;
void Render_Graph_Runtime::execute() {
state_->execute();
void Render_Graph_Runtime::execute(std::span<Node_Execution* const> executions, Execute_Node execute_node) {
state_->execute(executions, std::move(execute_node));
}
}
Kernel/src/renderive/render_graph/detail/Render_Graph_Runtime.hpp
+2 -2
View File
@@ -10,13 +10,13 @@ namespace renderive::render_graph::detail {
class Render_Graph_Runtime final {
public:
using Execute_Node = std::function<Node_Execution_Result(std::size_t execution_index, Node_Execution_Metrics* metrics)>;
Render_Graph_Runtime(const Render_Plan& plan, std::span<Node_Execution* const> executions, Execute_Node execute_node);
explicit Render_Graph_Runtime(const Render_Plan& plan);
~Render_Graph_Runtime();
Render_Graph_Runtime(const Render_Graph_Runtime&) = delete;
Render_Graph_Runtime& operator=(const Render_Graph_Runtime&) = delete;
Render_Graph_Runtime(Render_Graph_Runtime&&) = delete;
Render_Graph_Runtime& operator=(Render_Graph_Runtime&&) = delete;
void execute();
void execute(std::span<Node_Execution* const> executions, Execute_Node execute_node);
private:
struct State;
std::shared_ptr<State> state_;
Kernel/src/renderive/scene/base/Scene_Base.cpp
+86 -32
View File
@@ -84,6 +84,20 @@ private:
Scene_Base* previous_{};
};
struct Scene_Base::Compiled_Render_Plan {
struct Renderable_Key {
Renderable_Id id{};
std::shared_ptr<const Renderable_Graph> graph;
std::vector<Renderable_Id> dependency_parent_ids;
bool prepare_required{};
bool paint_required{};
};
std::vector<Renderable_Key> renderables;
std::vector<Renderable_Id> display_order;
std::shared_ptr<const Render_Plan> plan;
std::vector<Execution_Binding> execution_bindings;
std::unique_ptr<renderive::render_graph::detail::Render_Graph_Runtime> runtime;
};
Scene_Base::Scene_Base() : Scene_Base(*std::pmr::get_default_resource()) {}
Scene_Base::Scene_Base(std::pmr::memory_resource& upstream_memory_resource)
: Scene_Base(upstream_memory_resource, std::make_unique<Impl>()) {}
@@ -358,9 +372,9 @@ void Scene_Base::submit_render(Abstract_Frame* frame) {
if (render_sequence_ == std::numeric_limits<std::uint64_t>::max())
throw std::overflow_error("render sequence exhausted");
snapshot->render_sequence = ++render_sequence_;
auto task = std::make_shared<Render_Task>(memory_resource());
auto task = std::make_shared<Render_Task>();
task->frame = frame ? Render_Task::Frame{std::ref(*frame)} : Render_Task::Frame{std::make_shared<Abstract_Frame>()};
task->plan = compile_render_plan(*snapshot, *task);
task->compiled_plan = compile_render_plan(*snapshot);
task->topology = std::make_shared<Topology_Snapshot>(topology_snapshot());
task->completion = std::make_shared<Render_Completion>();
snapshot->capture_ticket_ = capture_controller_.begin_frame();
@@ -369,7 +383,7 @@ void Scene_Base::submit_render(Abstract_Frame* frame) {
const Observation submitted_observation{
Observation_Event::render_submitted, d_func().now_ns(), task->snapshot->render_sequence,
task->snapshot->scene_state_revision_, task->snapshot->renderables.size(), task->topology,
task->snapshot, task->plan};
task->snapshot, task->compiled_plan->plan};
++pending_operations_;
task_lock.unlock();
Scene_Base* previous_observer_scene = std::exchange(active_submitted_observer_scene_, this);
@@ -931,18 +945,18 @@ void Scene_Base::execute_render_task(std::shared_ptr<Render_Task> task) {
const auto& snapshot = *task->snapshot;
d_func().dispatch({Observation_Event::render_started, d_func().now_ns(), snapshot.render_sequence,
snapshot.scene_state_revision_, snapshot.renderables.size(), task->topology,
task->snapshot, task->plan});
task->snapshot, task->compiled_plan->plan});
std::exception_ptr exception;
try {
execute_render_graph(*task);
d_func().dispatch({Observation_Event::render_completed, d_func().now_ns(), snapshot.render_sequence,
snapshot.scene_state_revision_, snapshot.renderables.size(), task->topology,
task->snapshot, task->plan});
task->snapshot, task->compiled_plan->plan});
} catch (...) {
exception = std::current_exception();
d_func().dispatch({Observation_Event::render_failed, d_func().now_ns(), snapshot.render_sequence,
snapshot.scene_state_revision_, snapshot.renderables.size(), task->topology,
task->snapshot, task->plan});
task->snapshot, task->compiled_plan->plan});
}
{
std::lock_guard lock(task_mutex_);
@@ -952,8 +966,37 @@ void Scene_Base::execute_render_task(std::shared_ptr<Render_Task> task) {
render_completed_.notify_all();
}
std::shared_ptr<const Render_Plan> Scene_Base::compile_render_plan(
const Frame_Render_Snapshot& snapshot, Render_Task& task) {
std::shared_ptr<Scene_Base::Compiled_Render_Plan> Scene_Base::compile_render_plan(
const Frame_Render_Snapshot& snapshot) {
const auto matches = [this, &snapshot] {
if (!compiled_render_plan_ ||
compiled_render_plan_->display_order != snapshot.display_order ||
compiled_render_plan_->renderables.size() != snapshot.renderables.size())
return false;
for (std::size_t index = 0; index < snapshot.renderables.size(); ++index) {
const auto& key = compiled_render_plan_->renderables[index];
const auto& state = snapshot.renderables[index];
if (key.id != state.renderable_id ||
key.graph != state.render_graph ||
key.dependency_parent_ids != state.dependency_parent_ids ||
key.prepare_required != state.prepare_required ||
key.paint_required != state.paint_required)
return false;
}
return true;
};
if (matches())
return compiled_render_plan_;
const auto update_key = [&snapshot](Compiled_Render_Plan& compiled) {
compiled.renderables.clear();
compiled.renderables.reserve(snapshot.renderables.size());
for (const auto& state : snapshot.renderables) {
compiled.renderables.push_back({
state.renderable_id, state.render_graph, state.dependency_parent_ids,
state.prepare_required, state.paint_required});
}
compiled.display_order = snapshot.display_order;
};
struct Active_Renderable {
std::size_t snapshot_index{};
std::vector<Render_Node_Id> prepare_roots;
@@ -999,8 +1042,7 @@ std::shared_ptr<const Render_Plan> Scene_Base::compile_render_plan(
state.render_graph->functions.at(node.execution_index);
std::visit([&](const auto& function) {
functions.emplace(node.node_id,
Execution_Binding{state.owner_, item.snapshot_index,
Execution_Function{function}});
Execution_Binding{item.snapshot_index, Execution_Function{function}});
}, render_function);
selected.insert(node.node_id);
}
@@ -1068,7 +1110,7 @@ std::shared_ptr<const Render_Plan> Scene_Base::compile_render_plan(
graph.nodes.push_back({composite_begin_node_id_, 0, "Composite Frame",
Render_Node_Kind::composite, 0});
functions.emplace(composite_begin_node_id_,
Execution_Binding{{}, std::nullopt,
Execution_Binding{std::nullopt,
Composite_Render_Node_Function{
[compositor](const Composite_Render_Context& context) {
compositor->begin_composite(context);
@@ -1082,7 +1124,7 @@ std::shared_ptr<const Render_Plan> Scene_Base::compile_render_plan(
graph.nodes.push_back({composite_node_id, id,
"Composite", Render_Node_Kind::composite, 0});
functions.emplace(composite_node_id,
Execution_Binding{state.owner_, index,
Execution_Binding{index,
Composite_Render_Node_Function{
[compositor](const Composite_Render_Context& context) {
compositor->composite(context);
@@ -1110,7 +1152,7 @@ std::shared_ptr<const Render_Plan> Scene_Base::compile_render_plan(
Render_Node_Kind::render, 0});
functions.emplace(
scene_render_node_id_,
Execution_Binding{{}, std::nullopt,
Execution_Binding{std::nullopt,
Scene_Render_Node_Function{
[renderer](const Scene_Render_Context& context) {
return renderer->render_scene(context);
@@ -1120,17 +1162,30 @@ std::shared_ptr<const Render_Plan> Scene_Base::compile_render_plan(
}
auto plan = render_plan_history_.publish(std::move(graph));
task.execution_bindings.clear();
task.execution_bindings.resize(plan->graph.nodes.size());
std::vector<Execution_Binding> execution_bindings(plan->graph.nodes.size());
for (const auto& node : plan->graph.nodes) {
auto function = functions.find(node.node_id);
if (function == functions.end())
throw std::logic_error(
"render plan node has no execution binding");
task.execution_bindings[node.execution_index] =
std::move(function->second);
throw std::logic_error("render plan node has no execution binding");
execution_bindings[node.execution_index] = std::move(function->second);
}
return plan;
if (const auto existing = compiled_render_plans_.find(plan->version);
existing != compiled_render_plans_.end()) {
auto& compiled = *existing->second;
update_key(compiled);
compiled.execution_bindings = std::move(execution_bindings);
compiled_render_plan_ = existing->second;
return compiled_render_plan_;
}
auto compiled = std::make_shared<Compiled_Render_Plan>();
update_key(*compiled);
compiled->plan = std::move(plan);
compiled->execution_bindings = std::move(execution_bindings);
compiled->runtime = std::make_unique<
renderive::render_graph::detail::Render_Graph_Runtime>(*compiled->plan);
compiled_render_plans_.emplace(compiled->plan->version, compiled);
compiled_render_plan_ = compiled;
return compiled;
}
void Scene_Base::execute_render_graph(Render_Task& task) {
@@ -1142,7 +1197,7 @@ void Scene_Base::execute_render_graph(Render_Task& task) {
return &value.get();
},
task.frame);
if (!task.snapshot || !task.plan || !frame)
if (!task.snapshot || !task.compiled_plan || !frame)
throw std::logic_error("render task is incomplete");
const auto& snapshot = *task.snapshot;
const Capture_Frame_Ticket capture_ticket = snapshot.capture_ticket_;
@@ -1154,23 +1209,23 @@ void Scene_Base::execute_render_graph(Render_Task& task) {
}
const bool capture = capture_ticket.capture;
frame->begin_render(snapshot.render_sequence, *task.plan, capture,
const auto& compiled = *task.compiled_plan;
frame->begin_render(snapshot.render_sequence, *compiled.plan, capture,
capture ? render_clock_now_ns() : 0);
frame_active = true;
std::vector<Node_Execution*> execution_slots(
task.plan->graph.nodes.size());
for (const auto& node : task.plan->graph.nodes)
compiled.plan->graph.nodes.size());
for (const auto& node : compiled.plan->graph.nodes)
execution_slots[node.execution_index] =
frame->execution_slot(node.execution_index);
renderive::render_graph::detail::Render_Graph_Runtime runtime(
*task.plan, execution_slots,
[this, &task, &snapshot, &execution_slots](
compiled.runtime->execute(
execution_slots,
[this, &compiled, &snapshot, &execution_slots](
std::size_t execution_index,
Node_Execution_Metrics* metrics) -> Node_Execution_Result {
const auto& node = task.plan->graph.nodes.at(execution_index);
const auto& binding =
task.execution_bindings.at(execution_index);
const auto& node = compiled.plan->graph.nodes.at(execution_index);
const auto& binding = compiled.execution_bindings.at(execution_index);
Node_Diagnostic_Sink diagnostics;
if (auto* execution = execution_slots.at(execution_index))
diagnostics = Node_Diagnostic_Sink(execution->attachments);
@@ -1216,7 +1271,6 @@ void Scene_Base::execute_render_graph(Render_Task& task) {
}
throw std::logic_error("unknown render node kind");
});
runtime.execute();
for (const auto& state : snapshot.renderables) {
if (state.prepare_required)
@@ -1240,7 +1294,7 @@ void Scene_Base::execute_render_graph(Render_Task& task) {
if (completed) {
capture_repository_.publish(
capture_ticket.session_id, completed,
analyze_frame(*task.plan, *completed));
analyze_frame(*compiled.plan, *completed));
}
capture_controller_.finish_frame(capture_ticket, true);
} catch (...) {
Kernel/src/renderive/scene/base/Scene_Base.hpp
+7 -10
View File
@@ -221,26 +221,20 @@ private:
Scene_Render_Node_Function>;
struct Execution_Binding {
Renderable owner;
std::optional<std::size_t> renderable_index;
Execution_Function function;
};
struct Compiled_Render_Plan;
struct Render_Task {
using Frame = std::variant<std::shared_ptr<Abstract_Frame>,
std::reference_wrapper<Abstract_Frame>>;
explicit Render_Task(std::pmr::memory_resource& memory_resource)
: execution_bindings(&memory_resource) {}
std::shared_ptr<const Frame_Render_Snapshot> snapshot;
std::shared_ptr<const Render_Plan> plan;
std::shared_ptr<Compiled_Render_Plan> compiled_plan;
std::shared_ptr<const Topology_Snapshot> topology;
std::pmr::vector<Execution_Binding> execution_bindings;
Frame frame;
std::shared_ptr<Render_Completion> completion;
};
class Execution_Context;
class Render_Execution_Scope;
@@ -257,8 +251,8 @@ private:
[[nodiscard]] std::shared_ptr<Frame_Render_Snapshot>
capture_live_frame();
void publish_frame_state_locked();
std::shared_ptr<const Render_Plan> compile_render_plan(
const Frame_Render_Snapshot& snapshot, Render_Task& task);
std::shared_ptr<Compiled_Render_Plan> compile_render_plan(
const Frame_Render_Snapshot& snapshot);
void execute_render_task(std::shared_ptr<Render_Task> task);
void execute_renderable_edit(std::function<void()> edit);
void execute_render_graph(Render_Task& task);
@@ -302,6 +296,9 @@ private:
const Render_Node_Id composite_begin_node_id_;
const Render_Node_Id scene_render_node_id_;
Render_Plan_History render_plan_history_;
std::unordered_map<Render_Plan_Version, std::shared_ptr<Compiled_Render_Plan>>
compiled_render_plans_;
std::shared_ptr<Compiled_Render_Plan> compiled_render_plan_;
Capture_Controller capture_controller_;
Capture_Repository capture_repository_;
};
Kernel/tests/renderive/render_graph/Render_DAG_Test.cpp
+3
View File
@@ -54,6 +54,9 @@ TEST(render_dag_test, one_graph_assigns_dense_execution_slots_and_versions_only_
changed_graph.edges.push_back({100, 101});
const auto changed = history.publish(std::move(changed_graph));
EXPECT_GT(changed->version, first->version);
const auto reused = history.publish(make_parallel_graph());
EXPECT_EQ(reused, first);
EXPECT_EQ(history.current(), first);
EXPECT_EQ(history.find(first->version), first);
EXPECT_EQ(history.find(changed->version), changed);
}
Kernel/tests/renderive/render_graph/Render_Graph_Runtime_Test.cpp
+44 -32
View File
@@ -1,5 +1,6 @@
#include <gtest/gtest.h>
#include <array>
#include <atomic>
#include <chrono>
#include <condition_variable>
@@ -72,6 +73,20 @@ TEST(external_operation_test,
EXPECT_TRUE(subscribed_first.complete());
}
TEST(render_graph_runtime_test, reusable_topology_executes_multiple_frames) {
const auto plan = two_node_plan();
renderive::render_graph::detail::Render_Graph_Runtime runtime(*plan);
std::array<std::size_t, 2> execution_count{};
const auto execute_node = [&](std::size_t index, Node_Execution_Metrics*) {
++execution_count.at(index);
return Node_Execution_Result::completed();
};
runtime.execute({}, execute_node);
runtime.execute({}, execute_node);
EXPECT_EQ(execution_count[0], 2U);
EXPECT_EQ(execution_count[1], 2U);
}
TEST(render_graph_runtime_test,
external_successor_stays_blocked_until_operation_completes) {
const auto plan = two_node_plan();
@@ -81,19 +96,18 @@ TEST(render_graph_runtime_test,
std::atomic<bool> submit_started{};
std::atomic<bool> publish_executed{};
renderive::render_graph::detail::Render_Graph_Runtime runtime(
*plan, slots,
[&](std::size_t index, Node_Execution_Metrics*) {
if (index == 0) {
submit_started.store(true, std::memory_order_release);
submit_started.notify_all();
return Node_Execution_Result::external(source.operation());
}
publish_executed.store(true, std::memory_order_release);
return Node_Execution_Result::completed();
});
renderive::render_graph::detail::Render_Graph_Runtime runtime(*plan);
const auto execute_node = [&](std::size_t index, Node_Execution_Metrics*) {
if (index == 0) {
submit_started.store(true, std::memory_order_release);
submit_started.notify_all();
return Node_Execution_Result::external(source.operation());
}
publish_executed.store(true, std::memory_order_release);
return Node_Execution_Result::completed();
};
std::thread execution([&] { runtime.execute(); });
std::thread execution([&] { runtime.execute(slots, execute_node); });
submit_started.wait(false, std::memory_order_acquire);
EXPECT_FALSE(publish_executed.load(std::memory_order_acquire));
EXPECT_TRUE(source.complete());
@@ -115,21 +129,20 @@ TEST(render_graph_runtime_test,
std::atomic<bool> publish_executed{};
std::exception_ptr graph_error;
renderive::render_graph::detail::Render_Graph_Runtime runtime(
*plan, {},
[&](std::size_t index, Node_Execution_Metrics*) {
if (index == 0) {
submit_started.store(true, std::memory_order_release);
submit_started.notify_all();
return Node_Execution_Result::external(source.operation());
}
publish_executed.store(true, std::memory_order_release);
return Node_Execution_Result::completed();
});
renderive::render_graph::detail::Render_Graph_Runtime runtime(*plan);
const auto execute_node = [&](std::size_t index, Node_Execution_Metrics*) {
if (index == 0) {
submit_started.store(true, std::memory_order_release);
submit_started.notify_all();
return Node_Execution_Result::external(source.operation());
}
publish_executed.store(true, std::memory_order_release);
return Node_Execution_Result::completed();
};
std::thread execution([&] {
try {
runtime.execute();
runtime.execute({}, execute_node);
} catch (...) {
graph_error = std::current_exception();
}
@@ -156,15 +169,14 @@ TEST(render_graph_runtime_test,
std::condition_variable probe_completed;
bool probe_ran{};
renderive::render_graph::detail::Render_Graph_Runtime runtime(
*plan, {},
[&](std::size_t, Node_Execution_Metrics*) {
external_started.store(true, std::memory_order_release);
external_started.notify_all();
return Node_Execution_Result::external(source.operation());
});
renderive::render_graph::detail::Render_Graph_Runtime runtime(*plan);
const auto execute_node = [&](std::size_t, Node_Execution_Metrics*) {
external_started.store(true, std::memory_order_release);
external_started.notify_all();
return Node_Execution_Result::external(source.operation());
};
std::thread execution([&] { runtime.execute(); });
std::thread execution([&] { runtime.execute({}, execute_node); });
external_started.wait(false, std::memory_order_acquire);
renderive::scheduling::detail::OneTBB_Runtime::instance().enqueue([&] {
{
render_3D/render_3D/detail/Gpu_Completion_Service.cpp
+32 -67
View File
@@ -1,33 +1,22 @@
#include "Gpu_Completion_Service.h"
#include <chrono>
#include <stdexcept>
#include <string>
#include <utility>
namespace renderive::render_3d::detail {
Gpu_Completion_Service::Gpu_Completion_Service()
: thread_([this] { run(); }) {}
Gpu_Completion_Service::Gpu_Completion_Service() {
pending_.set_capacity(default_capacity);
thread_ = std::thread([this] { run(); });
}
Gpu_Completion_Service::~Gpu_Completion_Service() {
shutdown();
}
Gpu_Completion_Service::Reservation::Reservation(
std::shared_ptr<Pending_Fence> pending) noexcept
: pending_(std::move(pending)) {}
Gpu_Completion_Service::Reservation::Reservation(std::shared_ptr<Pending_Fence> pending) noexcept : pending_(std::move(pending)) {}
Gpu_Completion_Service::Reservation::~Reservation() {
cancel();
}
Gpu_Completion_Service::Reservation::Reservation(
Reservation&& other) noexcept
: pending_(std::exchange(other.pending_, {})) {}
void Gpu_Completion_Service::Reservation::watch(
VkDevice device, VkFence fence) noexcept {
Gpu_Completion_Service::Reservation::Reservation(Reservation&& other) noexcept : pending_(std::exchange(other.pending_, {})) {}
void Gpu_Completion_Service::Reservation::watch(VkDevice device, VkFence fence) noexcept {
if (!pending_ || device == VK_NULL_HANDLE || fence == VK_NULL_HANDLE)
std::terminate();
auto pending = std::exchange(pending_, {});
@@ -41,11 +30,15 @@ void Gpu_Completion_Service::Reservation::watch(
}
pending->ready.notify_one();
}
void Gpu_Completion_Service::Reservation::cancel() noexcept {
if (!pending_)
return;
auto pending = std::exchange(pending_, {});
cancel_reserved(pending);
}
void Gpu_Completion_Service::cancel_reserved(const std::shared_ptr<Pending_Fence>& pending) noexcept {
if (!pending)
return;
{
std::lock_guard lock(pending->mutex);
if (pending->status == Pending_Fence::Status::reserved)
@@ -53,9 +46,7 @@ void Gpu_Completion_Service::Reservation::cancel() noexcept {
}
pending->ready.notify_one();
}
Gpu_Completion_Service::Reservation Gpu_Completion_Service::prepare(
Completion completion, bool observe) {
Gpu_Completion_Service::Reservation Gpu_Completion_Service::prepare(Completion completion, bool observe) {
if (!completion)
throw std::invalid_argument("GPU completion callback is empty");
auto pending = std::make_shared<Pending_Fence>();
@@ -65,61 +56,46 @@ Gpu_Completion_Service::Reservation Gpu_Completion_Service::prepare(
std::lock_guard lock(mutex_);
if (stopping_)
throw std::runtime_error("GPU completion service is stopping");
pending_.push_back(pending);
pending_.push(pending);
}
ready_.notify_one();
return Reservation(std::move(pending));
}
void Gpu_Completion_Service::shutdown() noexcept {
{
std::lock_guard lock(mutex_);
if (stopping_ && !thread_.joinable())
return;
stopping_ = true;
const auto cancel_reserved = [](const auto& pending) {
if (!pending)
return;
{
std::lock_guard pending_lock(pending->mutex);
if (pending->status == Pending_Fence::Status::reserved)
pending->status = Pending_Fence::Status::canceled;
}
pending->ready.notify_one();
};
cancel_reserved(active_);
for (const auto& pending : pending_)
std::shared_ptr<Pending_Fence> pending;
while (pending_.try_pop(pending))
cancel_reserved(pending);
pending_.push(std::shared_ptr<Pending_Fence>{});
}
ready_.notify_one();
if (thread_.joinable())
thread_.join();
}
void Gpu_Completion_Service::run() noexcept {
for (;;) {
std::shared_ptr<Pending_Fence> pending;
pending_.pop(pending);
if (!pending)
return;
{
std::unique_lock lock(mutex_);
ready_.wait(lock, [this] {
return stopping_ || !pending_.empty();
});
if (stopping_ && pending_.empty())
return;
pending = std::move(pending_.front());
pending_.pop_front();
std::lock_guard lock(mutex_);
if (stopping_) {
cancel_reserved(pending);
continue;
}
active_ = pending;
}
VkDevice device{VK_NULL_HANDLE};
VkFence fence{VK_NULL_HANDLE};
Completion completion;
bool observe{};
{
std::unique_lock lock(pending->mutex);
pending->ready.wait(lock, [&pending] {
return pending->status != Pending_Fence::Status::reserved;
});
pending->ready.wait(lock, [&pending] { return pending->status != Pending_Fence::Status::reserved; });
if (pending->status == Pending_Fence::Status::canceled) {
lock.unlock();
std::lock_guard service_lock(mutex_);
@@ -131,26 +107,16 @@ void Gpu_Completion_Service::run() noexcept {
completion = std::move(pending->completion);
observe = pending->observe;
}
const auto wait_started = observe
? std::chrono::steady_clock::now()
: std::chrono::steady_clock::time_point{};
const VkResult result = vkWaitForFences(
device, 1, &fence, VK_TRUE, UINT64_MAX);
const auto wait_started = observe ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{};
const VkResult result = vkWaitForFences(device, 1, &fence, VK_TRUE, UINT64_MAX);
Result completion_result;
if (observe) {
const auto wait_duration =
std::chrono::duration_cast<std::chrono::nanoseconds>(
std::chrono::steady_clock::now() - wait_started).count();
completion_result.wait_duration_ns = wait_duration > 0
? static_cast<std::uint64_t>(wait_duration)
: 0;
const auto wait_duration = std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::steady_clock::now() - wait_started).count();
completion_result.wait_duration_ns = wait_duration > 0 ? static_cast<std::uint64_t>(wait_duration) : 0;
}
if (result != VK_SUCCESS) {
try {
throw std::runtime_error(
"GPU fence wait failed with Vulkan result " +
std::to_string(static_cast<int>(result)));
throw std::runtime_error("GPU fence wait failed with Vulkan result " + std::to_string(static_cast<int>(result)));
} catch (...) {
completion_result.error = std::current_exception();
}
@@ -165,5 +131,4 @@ void Gpu_Completion_Service::run() noexcept {
}
}
}
} // namespace renderive::render_3d::detail
}
render_3D/render_3D/detail/Gpu_Completion_Service.h
+6 -25
View File
@@ -1,59 +1,43 @@
#pragma once
#include <volk.h>
#include <condition_variable>
#include <cstdint>
#include <deque>
#include <exception>
#include <functional>
#include <mutex>
#include <memory>
#include <mutex>
#include <thread>
#include <oneapi/tbb/concurrent_queue.h>
namespace renderive::render_3d::detail {
class Gpu_Completion_Service final {
struct Pending_Fence;
public:
struct Result {
std::exception_ptr error;
std::uint64_t wait_duration_ns{};
};
using Completion = std::function<void(Result)>;
class Reservation final {
public:
Reservation() = default;
~Reservation();
Reservation(const Reservation&) = delete;
Reservation& operator=(const Reservation&) = delete;
Reservation(Reservation&& other) noexcept;
Reservation& operator=(Reservation&&) = delete;
void watch(VkDevice device, VkFence fence) noexcept;
private:
explicit Reservation(std::shared_ptr<Pending_Fence> pending) noexcept;
void cancel() noexcept;
std::shared_ptr<Pending_Fence> pending_;
friend class Gpu_Completion_Service;
};
Gpu_Completion_Service();
~Gpu_Completion_Service();
Gpu_Completion_Service(const Gpu_Completion_Service&) = delete;
Gpu_Completion_Service& operator=(const Gpu_Completion_Service&) = delete;
[[nodiscard]] Reservation prepare(Completion completion, bool observe);
void shutdown() noexcept;
private:
struct Pending_Fence {
enum class Status {
@@ -61,7 +45,6 @@ private:
watched,
canceled
};
std::mutex mutex;
std::condition_variable ready;
VkDevice device{VK_NULL_HANDLE};
@@ -70,15 +53,13 @@ private:
Status status{Status::reserved};
bool observe{};
};
static void cancel_reserved(const std::shared_ptr<Pending_Fence>& pending) noexcept;
void run() noexcept;
static constexpr std::size_t default_capacity = 64;
std::mutex mutex_;
std::condition_variable ready_;
std::deque<std::shared_ptr<Pending_Fence>> pending_;
oneapi::tbb::concurrent_bounded_queue<std::shared_ptr<Pending_Fence>> pending_;
std::shared_ptr<Pending_Fence> active_;
std::thread thread_;
bool stopping_{};
};
} // namespace renderive::render_3d::detail
}
render_3D/render_3D/detail/Render_Domain.h
+24 -67
View File
@@ -1,6 +1,4 @@
#pragma once
#include <condition_variable>
#include <functional>
#include <future>
#include <memory>
@@ -9,87 +7,66 @@
#include <thread>
#include <type_traits>
#include <utility>
#include <oneapi/tbb/concurrent_queue.h>
namespace renderive::render_3d::detail {
class Render_Domain final {
struct Task {
explicit Task(std::function<void()> value)
: function(std::move(value)) {}
explicit Task(std::function<void()> value) : function(std::move(value)) {}
std::function<void()> function;
std::unique_ptr<Task> next;
};
public:
class Prepared_Task final {
public:
Prepared_Task() = default;
~Prepared_Task() = default;
Prepared_Task(const Prepared_Task&) = delete;
Prepared_Task& operator=(const Prepared_Task&) = delete;
Prepared_Task(Prepared_Task&&) noexcept = default;
Prepared_Task& operator=(Prepared_Task&&) noexcept = default;
private:
explicit Prepared_Task(std::unique_ptr<Task> task) noexcept
: task_(std::move(task)) {}
explicit Prepared_Task(std::unique_ptr<Task> task) noexcept : task_(std::move(task)) {}
std::unique_ptr<Task> task_;
friend class Render_Domain;
};
Render_Domain() : thread_([this] { run(); }) {}
Render_Domain() {
tasks_.set_capacity(default_capacity);
thread_ = std::thread([this] { run(); });
}
~Render_Domain() {
{
std::lock_guard lock(mutex_);
stopping_ = true;
tasks_.push(std::unique_ptr<Task>{});
}
condition_.notify_one();
if (thread_.joinable())
thread_.join();
}
Render_Domain(const Render_Domain&) = delete;
Render_Domain& operator=(const Render_Domain&) = delete;
[[nodiscard]] Prepared_Task prepare(std::function<void()> function) {
if (!function)
throw std::invalid_argument("render domain task is empty");
return Prepared_Task(
std::make_unique<Task>(std::move(function)));
return Prepared_Task(std::make_unique<Task>(std::move(function)));
}
void post(std::function<void()> function) {
auto task = prepare(std::move(function));
{
std::lock_guard lock(mutex_);
if (stopping_)
throw std::runtime_error("Point_Scene render domain is stopping");
enqueue_locked(std::move(task.task_));
}
condition_.notify_one();
std::lock_guard lock(mutex_);
if (stopping_)
throw std::runtime_error("Point_Scene render domain is stopping");
tasks_.push(std::move(task.task_));
}
void post(Prepared_Task task) noexcept {
if (!task.task_)
std::terminate();
{
std::lock_guard lock(mutex_);
if (stopping_)
std::terminate();
enqueue_locked(std::move(task.task_));
}
condition_.notify_one();
std::lock_guard lock(mutex_);
if (stopping_)
std::terminate();
tasks_.push(std::move(task.task_));
}
template <class Function>
auto invoke(Function&& function) -> std::invoke_result_t<Function> {
using Result = std::invoke_result_t<Function>;
auto task = std::make_shared<std::packaged_task<Result()>>(
std::forward<Function>(function));
auto task = std::make_shared<std::packaged_task<Result()>>(std::forward<Function>(function));
auto result = task->get_future();
post([task] { (*task)(); });
if constexpr (std::is_void_v<Result>)
@@ -97,40 +74,20 @@ public:
else
return result.get();
}
private:
void run() {
for (;;) {
std::unique_ptr<Task> task;
{
std::unique_lock lock(mutex_);
condition_.wait(lock, [&] { return stopping_ || first_; });
if (stopping_ && !first_)
return;
task = std::move(first_);
first_ = std::move(task->next);
if (!first_)
last_ = nullptr;
}
tasks_.pop(task);
if (!task)
return;
task->function();
}
}
void enqueue_locked(std::unique_ptr<Task> task) noexcept {
Task* const inserted = task.get();
if (last_)
last_->next = std::move(task);
else
first_ = std::move(task);
last_ = inserted;
}
static constexpr std::size_t default_capacity = 64;
oneapi::tbb::concurrent_bounded_queue<std::unique_ptr<Task>> tasks_;
std::mutex mutex_;
std::condition_variable condition_;
std::unique_ptr<Task> first_;
Task* last_{};
bool stopping_{};
std::thread thread_;
};
} // namespace renderive::render_3d::detail
}
web_server/app/Gallery_Protocol.cpp
+11
View File
@@ -4,6 +4,7 @@
#include "Gallery_Observer_Adminive.h"
#include "Gallery_Renderables.h"
#include "Gallery_Session_Control_Adminive.h"
#include "common/Gallery_Scheduler_Adminive.h"
#include "adminive/adminive.hpp"
#include "adminive/adapters/nlohmann_json.hpp"
@@ -392,6 +393,10 @@ Json dashboard_contract() {
dashboard_field("呈现 P95 ms", "client_performance.display_interval_p95_ms", "fixed", 2),
dashboard_field("呈现 P99 ms", "client_performance.display_interval_p99_ms", "fixed", 2),
dashboard_field("响应负载 MB/s", "performance.pixel_payload_megabytes_per_second", "fixed", 1),
dashboard_field("TBB 并发度", "scheduler.concurrency"),
dashboard_field("活动 Worker", "scheduler.active_workers"),
dashboard_field("峰值 Worker", "scheduler.peak_workers"),
dashboard_field("活动外部线程", "scheduler.active_external_threads"),
described_dashboard_field<renderive::Plot_Frame_Status>(
"kernel_observer", "pending_frame_count"),
described_dashboard_field<renderive::Plot_Frame_Status>(
@@ -433,6 +438,12 @@ Json dashboard_contract() {
{"source", "client_performance"},
{"descriptor", adminive::to_descriptor_json<
Json, Gallery_Client_Performance>()}
},
{
{"title", "oneTBB 调度器"},
{"source", "scheduler"},
{"descriptor", adminive::to_descriptor_json<
Json, renderive::scheduling::Scheduler_Statistics>()}
}
})}
}}
web_server/app/common/Gallery_Scheduler_Adminive.h
+23
View File
@@ -0,0 +1,23 @@
#pragma once
#include "renderive/scheduling/Scheduler.hpp"
#include "adminive/adminive.hpp"
namespace adminive {
template <>
struct Type_Descriptor<renderive::scheduling::Scheduler_Statistics> {
static auto get() {
using T = renderive::scheduling::Scheduler_Statistics;
return object<T>("scheduler_statistics",
ADMINIVE_FIELD_LABEL(T, concurrency, "调度并发度"),
ADMINIVE_FIELD_LABEL(T, active_workers, "活动 Worker"),
ADMINIVE_FIELD_LABEL(T, peak_workers, "峰值 Worker"),
ADMINIVE_FIELD_LABEL(T, active_external_threads, "活动外部线程"),
ADMINIVE_FIELD_LABEL(T, peak_external_threads, "峰值外部线程"),
ADMINIVE_FIELD_LABEL(T, worker_entry_count, "Worker 进入次数"),
ADMINIVE_FIELD_LABEL(T, worker_exit_count, "Worker 退出次数"),
ADMINIVE_FIELD_LABEL(T, external_entry_count, "外部线程进入次数"),
ADMINIVE_FIELD_LABEL(T, external_exit_count, "外部线程退出次数"))
.label("oneTBB 调度器");
}
};
static_assert(Described_Type<renderive::scheduling::Scheduler_Statistics>);
}
web_server/app/render_2D/Gallery_Scene2D.cpp
+3
View File
@@ -3,6 +3,7 @@
#include "../Gallery_Enum.h"
#include "../Gallery_Observer_Adminive.h"
#include "../Gallery_Session_Control_Adminive.h"
#include "../common/Gallery_Scheduler_Adminive.h"
#include "../Pixel_Frame.h"
#include "../Web_Performance_Log.h"
#include "render_2D/export.h"
@@ -653,6 +654,8 @@ public:
{"kernel_observer", std::move(observer_data)},
{"consumer_feedback", std::move(consumer_feedback_data)},
{"client_performance", std::move(client_performance_data)},
{"scheduler", adminive::to_frontend_json<nlohmann::json>(
renderive::scheduling::scheduler_statistics())},
{"renderable_observers", controls_.observers(plot_.render_scene())},
{"performance_capture", gallery_performance_capture_json(plot_.render_scene())},
{"last_action_result", last_action_result_}
web_server/app/render_3D/Gallery_Scene3D.cpp
+3
View File
@@ -2,6 +2,7 @@
#include "../Gallery_Capture_Json.h"
#include "../Gallery_Enum.h"
#include "../common/Gallery_Scheduler_Adminive.h"
#include "../common/Pixel_Frame.h"
#include <render_3D/Point_Scene.h>
@@ -461,6 +462,8 @@ public:
{"presentation_fps", client_performance_.presentation_fps},
{"websocket_buffered_bytes",
client_performance_.websocket_buffered_bytes}}},
{"scheduler", adminive::to_frontend_json<nlohmann::json>(
renderive::scheduling::scheduler_statistics())},
{"session_id", session_id_},
{"performance_capture",
gallery_performance_capture_json(scene_.render_scene())},
web_server/tests/Datoviz_Gallery_Tests.cpp
+1
View File
@@ -240,6 +240,7 @@ TEST(RenderiveWebDatovizGallery, PublishesRealRgbaAndRendersAfterResize) {
const auto observed = response_json(session.handle(gallery_request(
Gallery_Request_Kind::Observe,
R"({"category":"event","type":"gallery_observe","client_metrics":{}})")));
EXPECT_TRUE(observed.at("telemetry").contains("scheduler"));
const auto& performance = observed.at("telemetry").at("performance");
EXPECT_GT(performance.at("measured_fps").get<double>(), 0.0);
EXPECT_GT(performance.at("pixel_response_fps").get<double>(), 0.0);
web_server/tests/Web_Bridge_Tests.cpp
+2 -1
View File
@@ -243,7 +243,7 @@ TEST(RenderiveWebGallery, AdminiveCatalogCoversEveryControlCaseAndThreeModes) {
TEST(RenderiveWebGallery, CatalogOwnsDashboardFieldsAndDynamicBehavior) {
const nlohmann::json catalog = parse_json(Gallery_Protocol::catalog_json());
const auto& dashboard = catalog.at("dashboard");
EXPECT_EQ(dashboard.at("performance").at("fields").size(), 26U);
EXPECT_EQ(dashboard.at("performance").at("fields").size(), 30U);
EXPECT_EQ(dashboard.at("limits").at("fields").size(), 4U);
EXPECT_EQ(dashboard.at("observer").at("sections").size(), 4U);
std::set<std::string> client_sources;
@@ -972,6 +972,7 @@ TEST(RenderiveWebGallery, EveryIndependentCore2CanvasBuildsAndRenders) {
EXPECT_EQ(telemetry.at("frame_mode").at("id"), std::string(mode)) << case_id;
EXPECT_TRUE(telemetry.at("telemetry").contains("kernel_observer")) << case_id;
EXPECT_TRUE(telemetry.at("telemetry").contains("performance")) << case_id;
EXPECT_TRUE(telemetry.at("telemetry").contains("scheduler")) << case_id;
EXPECT_TRUE(telemetry.at("telemetry").contains("data_shape")) << case_id;
EXPECT_GT(telemetry.at("telemetry").at("data_shape").at("rendered_elements")
.get<std::size_t>(), 0U) << case_id;
webapp_gallery/src/protocol/gallery_types.ts
+2 -1
View File
@@ -24,8 +24,9 @@ export interface Gallery_Catalog {type: "catalog"; navigation: Gallery_Navigatio
export interface Gallery_Client_Performance {transport_fps: number; presentation_fps: number; display_interval_latest_ms: number; display_interval_ms: number; display_interval_average_ms: number; display_interval_p95_ms: number; display_interval_p99_ms: number; display_interval_deviation_ms: number; frame_round_trip_ms: number; frame_round_trip_average_ms: number; frame_round_trip_p95_ms: number; frame_round_trip_p99_ms: number; frame_round_trip_deviation_ms: number; changed_pixel_frames: number; duplicate_pixel_frames: number; overwritten_pixel_frames: number; frame_request_timeout_count: number; last_pixel_receive_age_ms: number; last_pixel_change_age_ms: number; websocket_buffered_bytes: number;}
export interface Gallery_Performance extends Record<string, Json_Value> {}
export interface Gallery_Kernel_Observer extends Record<string, Json_Value> {}
export interface Gallery_Scheduler_Statistics {concurrency: number; active_workers: number; peak_workers: number; active_external_threads: number; peak_external_threads: number; worker_entry_count: number; worker_exit_count: number; external_entry_count: number; external_exit_count: number;}
export interface Gallery_Renderable_Observer extends Gallery_Resource {}
export interface Gallery_Telemetry {[key: string]: unknown; renderable_observers?: Gallery_Renderable_Observer[]; performance_capture?: Gallery_Performance_Capture;}
export interface Gallery_Telemetry {[key: string]: unknown; scheduler?: Gallery_Scheduler_Statistics; renderable_observers?: Gallery_Renderable_Observer[]; performance_capture?: Gallery_Performance_Capture;}
export interface Gallery_Render_Node {node_id: number; id?: string | number; name: string; label?: string; owner: string; kind: string;}
export interface Gallery_Render_Edge {from: number | string; to: number | string;}