观察者效应优化
This commit is contained in:
@@ -23,14 +23,10 @@ struct Taskflow_Frame_Access {
|
||||
Render_Frame& frame, std::size_t worker, std::uint64_t native_id,
|
||||
std::size_t queue_size, std::size_t queue_capacity,
|
||||
Clock::time_point entered, Clock::time_point started,
|
||||
Clock::time_point finished, std::uint64_t cpu_entered_ns,
|
||||
std::uint64_t cpu_started_ns, std::uint64_t cpu_duration_ns,
|
||||
std::uint64_t cpu_cycles, std::uint64_t cooperative_wait_ns,
|
||||
bool cpu_time_coarse);
|
||||
Clock::time_point finished, std::uint64_t cooperative_wait_ns);
|
||||
static void finish_task_observer(
|
||||
Render_Frame& frame, std::size_t worker, std::size_t task,
|
||||
Clock::time_point completed, std::uint64_t cpu_finished_ns,
|
||||
std::uint64_t cpu_completed_ns) noexcept;
|
||||
Clock::time_point completed) noexcept;
|
||||
[[nodiscard]] static Taskflow_Graph_Token begin_graph(
|
||||
Render_Frame& frame, Task_Graph& graph, std::string_view stage);
|
||||
static void finish_graph(Taskflow_Graph_Token token,
|
||||
|
||||
+15
-19
@@ -51,7 +51,7 @@ void Render_Frame::begin(Frame_Identity identity,
|
||||
for (auto& measurement : d->measurements)
|
||||
measurement.store(0, std::memory_order_relaxed);
|
||||
d->taskflow_trace_requested.store(false, std::memory_order_relaxed);
|
||||
d->taskflow_workers.clear();
|
||||
for (auto& worker : d->taskflow_workers) worker.tasks.clear();
|
||||
d->taskflow_graphs.clear();
|
||||
d->markers[static_cast<std::size_t>(Frame_Trace_Marker::created)].store(encode_present_value(0), std::memory_order_relaxed);
|
||||
}
|
||||
@@ -177,7 +177,10 @@ Taskflow_Frame_Trace Render_Frame::take_taskflow_trace() {
|
||||
static_cast<Frame_Trace_Measurement>(index),
|
||||
decode_present_value(encoded)});
|
||||
}
|
||||
result.graphs = std::move(d->taskflow_graphs);
|
||||
result.graphs.reserve(d->taskflow_graphs.size());
|
||||
for (auto& graph : d->taskflow_graphs)
|
||||
result.graphs.push_back(std::move(graph));
|
||||
d->taskflow_graphs.clear();
|
||||
for (auto& worker : d->taskflow_workers) {
|
||||
for (auto& task : worker.tasks)
|
||||
result.tasks.push_back(std::move(task));
|
||||
@@ -341,9 +344,11 @@ Taskflow_Frame_Trace Render_Frame::take_taskflow_trace() {
|
||||
|
||||
void detail::Taskflow_Frame_Access::begin_capture(Render_Frame& frame, std::size_t workers) {
|
||||
auto& data = *frame.d;
|
||||
data.taskflow_workers.clear();
|
||||
data.taskflow_workers.resize(workers);
|
||||
for (auto& worker : data.taskflow_workers) worker.tasks.reserve(64);
|
||||
for (auto& worker : data.taskflow_workers) {
|
||||
worker.tasks.clear();
|
||||
if (worker.tasks.capacity() < 64) worker.tasks.reserve(64);
|
||||
}
|
||||
data.taskflow_graphs.clear();
|
||||
}
|
||||
|
||||
@@ -356,10 +361,7 @@ std::size_t detail::Taskflow_Frame_Access::append_task(
|
||||
Render_Frame& frame, std::size_t worker, std::uint64_t native_id,
|
||||
std::size_t queue_size, std::size_t queue_capacity,
|
||||
Clock::time_point entered, Clock::time_point started,
|
||||
Clock::time_point finished, std::uint64_t cpu_entered_ns,
|
||||
std::uint64_t cpu_started_ns, std::uint64_t cpu_duration_ns,
|
||||
std::uint64_t cpu_cycles, std::uint64_t cooperative_wait_ns,
|
||||
bool cpu_time_coarse) {
|
||||
Clock::time_point finished, std::uint64_t cooperative_wait_ns) {
|
||||
auto& data = *frame.d;
|
||||
if (worker >= data.taskflow_workers.size())
|
||||
return std::numeric_limits<std::size_t>::max();
|
||||
@@ -376,22 +378,15 @@ std::size_t detail::Taskflow_Frame_Access::append_task(
|
||||
trace.finished_ms = elapsed_ms(finished);
|
||||
trace.completed_ms = trace.finished_ms;
|
||||
trace.duration_ms = std::max(0.0, trace.finished_ms - trace.started_ms);
|
||||
trace.cpu_duration_ms = static_cast<double>(cpu_duration_ns) / 1'000'000.0;
|
||||
trace.cpu_cycles = cpu_cycles;
|
||||
trace.cooperative_wait_ms = static_cast<double>(cooperative_wait_ns) / 1'000'000.0;
|
||||
trace.cpu_time_coarse = cpu_time_coarse;
|
||||
trace.observer_entry_ms = std::max(0.0, trace.started_ms - trace.entered_ms);
|
||||
trace.observer_entry_cpu_ms = cpu_started_ns >= cpu_entered_ns
|
||||
? static_cast<double>(cpu_started_ns - cpu_entered_ns) / 1'000'000.0
|
||||
: 0.0;
|
||||
data.taskflow_workers[worker].tasks.push_back(std::move(trace));
|
||||
return data.taskflow_workers[worker].tasks.size() - 1;
|
||||
}
|
||||
|
||||
void detail::Taskflow_Frame_Access::finish_task_observer(
|
||||
Render_Frame& frame, std::size_t worker, std::size_t task,
|
||||
Clock::time_point completed, std::uint64_t cpu_finished_ns,
|
||||
std::uint64_t cpu_completed_ns) noexcept {
|
||||
Clock::time_point completed) noexcept {
|
||||
auto& data = *frame.d;
|
||||
if (worker >= data.taskflow_workers.size() ||
|
||||
task >= data.taskflow_workers[worker].tasks.size()) return;
|
||||
@@ -400,9 +395,6 @@ void detail::Taskflow_Frame_Access::finish_task_observer(
|
||||
completed - data.created_at).count();
|
||||
trace.observer_exit_ms = std::max(
|
||||
0.0, trace.completed_ms - trace.finished_ms);
|
||||
trace.observer_exit_cpu_ms = cpu_completed_ns >= cpu_finished_ns
|
||||
? static_cast<double>(cpu_completed_ns - cpu_finished_ns) / 1'000'000.0
|
||||
: 0.0;
|
||||
}
|
||||
|
||||
detail::Taskflow_Graph_Token detail::Taskflow_Frame_Access::begin_graph(
|
||||
@@ -411,6 +403,10 @@ detail::Taskflow_Graph_Token detail::Taskflow_Frame_Access::begin_graph(
|
||||
graph.stage = stage;
|
||||
graph.taskflow_name = taskflow.name();
|
||||
graph.nodes = detail::Task_Graph_Access::nodes(taskflow);
|
||||
for (auto& worker : frame.d->taskflow_workers) {
|
||||
const auto required = worker.tasks.size() + graph.nodes.size();
|
||||
if (worker.tasks.capacity() < required) worker.tasks.reserve(required);
|
||||
}
|
||||
graph.submitted_ms = std::chrono::duration<double, std::milli>(
|
||||
Clock::now() - frame.d->created_at).count();
|
||||
frame.d->taskflow_graphs.push_back(std::move(graph));
|
||||
|
||||
@@ -109,14 +109,9 @@ struct Taskflow_Task_Trace {
|
||||
double finished_ms{}; /* Observer on_exit 进入，即任务体已经结束的时间。 */
|
||||
double completed_ms{}; /* Observer on_exit 与按帧追踪写入全部结束的时间。 */
|
||||
double duration_ms{}; /* 仅任务体 started 到 finished 的持续时间。 */
|
||||
double cpu_duration_ms{}; /* 任务体独占当前 worker 片段的线程 CPU 时间；cooperative corun 期间不计入。 */
|
||||
std::uint64_t cpu_cycles{}; /* Windows QueryThreadCycleTime 的独占 CPU 周期；用于短任务 CPU 活动判定，不直接换算秒。 */
|
||||
double cooperative_wait_ms{}; /* Task_Graph::corun/corun_until 主动让出 Worker 的墙钟时间。 */
|
||||
bool cpu_time_coarse{}; /* 当前平台的线程 CPU 时间源是否为低分辨率计费时钟（Windows GetThreadTimes）。 */
|
||||
double observer_entry_ms{}; /* on_entry 诊断本身的耗时。 */
|
||||
double observer_exit_ms{}; /* on_exit 诊断与按帧追踪写入的耗时。 */
|
||||
double observer_entry_cpu_ms{}; /* on_entry 诊断实际消耗的 worker CPU 时间。 */
|
||||
double observer_exit_cpu_ms{}; /* on_exit 诊断实际消耗的 worker CPU 时间。 */
|
||||
double ready_ms{}; /* 前驱完成或根 run 提交后的估算就绪时间。 */
|
||||
double queue_wait_ms{}; /* ready 到 entered 的估算 Executor 排队时间。 */
|
||||
};
|
||||
|
||||
@@ -72,32 +72,6 @@ std::uint64_t thread_cpu_cycles(HANDLE thread) noexcept {
|
||||
: 0;
|
||||
}
|
||||
#endif
|
||||
std::uint64_t current_thread_cpu_ns() noexcept {
|
||||
#if defined(_WIN32)
|
||||
return thread_cpu_ns(GetCurrentThread());
|
||||
#elif defined(CLOCK_THREAD_CPUTIME_ID)
|
||||
timespec value{};
|
||||
if (clock_gettime(CLOCK_THREAD_CPUTIME_ID, &value) != 0) return 0;
|
||||
return static_cast<std::uint64_t>(value.tv_sec) * 1'000'000'000ULL +
|
||||
static_cast<std::uint64_t>(value.tv_nsec);
|
||||
#else
|
||||
return 0;
|
||||
#endif
|
||||
}
|
||||
std::uint64_t current_thread_cpu_cycles() noexcept {
|
||||
#if defined(_WIN32)
|
||||
return thread_cpu_cycles(GetCurrentThread());
|
||||
#else
|
||||
return 0;
|
||||
#endif
|
||||
}
|
||||
constexpr bool thread_cpu_time_is_coarse() noexcept {
|
||||
#if defined(_WIN32)
|
||||
return true;
|
||||
#else
|
||||
return false;
|
||||
#endif
|
||||
}
|
||||
struct Task_Observer : public tf::ObserverInterface {
|
||||
private:
|
||||
using Clock = std::chrono::steady_clock;
|
||||
@@ -120,9 +94,9 @@ private:
|
||||
std::atomic_uint64_t active_task_hash{};
|
||||
std::atomic_uint64_t active_task_started_ns{};
|
||||
std::atomic_uint64_t active_segment_started_ns{};
|
||||
std::atomic_uint64_t active_segment_cpu_started_ns{};
|
||||
#if defined(_WIN32)
|
||||
std::atomic_uintptr_t native_thread_handle{}; /* Watchdog 只读的真实 Worker 线程句柄。 */
|
||||
std::atomic_uint64_t cpu_time_origin_ns{}; /* 首次进入 Worker 时的累计线程 CPU，读取侧据此计算观测期增量。 */
|
||||
std::atomic_uint64_t active_cpu_cycles{}; /* 连续片段最近一次采样的 Worker CPU 周期。 */
|
||||
std::atomic_uint64_t active_cpu_progress_ns{}; /* CPU 周期最后前进的墙钟时刻。 */
|
||||
#endif
|
||||
@@ -130,32 +104,19 @@ private:
|
||||
std::atomic_bool active_task_reported{};
|
||||
std::atomic_uint64_t task_time_ns{};
|
||||
std::atomic_uint64_t busy_time_ns{};
|
||||
std::atomic_uint64_t cpu_time_ns{};
|
||||
std::atomic_uint64_t min_task_time_ns{std::numeric_limits<std::uint64_t>::max()};
|
||||
std::atomic_uint64_t max_task_time_ns{};
|
||||
std::atomic_size_t max_predecessors{};
|
||||
std::atomic_size_t max_successors{};
|
||||
std::atomic_size_t max_strong_dependencies{};
|
||||
std::atomic_size_t max_weak_dependencies{};
|
||||
std::atomic_uint64_t first_task_time_ns{};
|
||||
std::atomic_uint64_t last_task_time_ns{};
|
||||
std::atomic_uint64_t longest_task_time_ns{};
|
||||
std::atomic_size_t longest_task_hash{};
|
||||
std::atomic<tf::TaskType> longest_task_type{tf::TaskType::UNDEFINED};
|
||||
std::atomic<std::shared_ptr<const std::string>> longest_task_name{};
|
||||
};
|
||||
struct Start_Record {
|
||||
Clock::time_point entered{}; /* Observer on_entry 进入时间。 */
|
||||
Clock::time_point started{}; /* on_entry 完成、任务体即将执行的时间。 */
|
||||
Clock::time_point segment_started{}; /* 当前连续独占 Worker 片段的起点。 */
|
||||
std::uint64_t maximum_segment_ns{}; /* 已结束连续独占片段的最大墙钟。 */
|
||||
std::uint64_t cpu_entered_ns{}; /* on_entry 进入时的 worker CPU 时间。 */
|
||||
std::uint64_t cpu_started_ns{}; /* 任务体开始前的 worker CPU 时间，仅用于 Observer entry 统计。 */
|
||||
std::uint64_t cpu_segment_started_ns{}; /* 当前任务独占 Worker 片段的线程 CPU 起点。 */
|
||||
bool cpu_segment_active{}; /* CPU 累计值本身允许为 0，不能拿 0 当未启动哨兵。 */
|
||||
std::uint64_t cpu_duration_ns{}; /* 已累计的任务独占 Worker CPU；嵌套 corun/子任务不计入。 */
|
||||
std::uint64_t cpu_cycle_segment_started{}; /* Windows 当前独占片段的 QueryThreadCycleTime 起点。 */
|
||||
std::uint64_t cpu_cycles{}; /* Windows 已累计的任务独占 CPU 周期。 */
|
||||
Clock::time_point cooperative_wait_started{}; /* 主动 corun 让出 Worker 的墙钟起点。 */
|
||||
std::uint64_t cooperative_wait_ns{}; /* 已累计 cooperative wait 墙钟。 */
|
||||
Render_Frame* frame{}; /* 进入任务时唯一活动的按帧捕获。 */
|
||||
@@ -167,7 +128,6 @@ private:
|
||||
};
|
||||
std::vector<std::vector<Start_Record>> starts;
|
||||
std::vector<Clock::time_point> worker_busy_starts;
|
||||
std::vector<std::uint64_t> worker_cpu_starts;
|
||||
std::unique_ptr<Worker_Statistics[]> worker_statistics;
|
||||
std::size_t worker_statistics_count{};
|
||||
std::uint64_t worker_occupation_limit_ns{}; /* 单节点连续非 CPU 等待 Worker 的上限。 */
|
||||
@@ -278,41 +238,6 @@ private:
|
||||
}
|
||||
}
|
||||
}
|
||||
static void close_cpu_segment(
|
||||
Start_Record& active, std::uint64_t cpu_now_ns,
|
||||
std::uint64_t cycle_now) noexcept {
|
||||
if (!active.frame || !active.cpu_segment_active) return;
|
||||
if (cpu_now_ns >= active.cpu_segment_started_ns)
|
||||
active.cpu_duration_ns +=
|
||||
cpu_now_ns - active.cpu_segment_started_ns;
|
||||
#if defined(_WIN32)
|
||||
if (cycle_now >= active.cpu_cycle_segment_started)
|
||||
active.cpu_cycles += cycle_now - active.cpu_cycle_segment_started;
|
||||
#else
|
||||
static_cast<void>(cycle_now);
|
||||
#endif
|
||||
active.cpu_segment_started_ns = 0;
|
||||
active.cpu_cycle_segment_started = 0;
|
||||
active.cpu_segment_active = false;
|
||||
}
|
||||
static void close_cpu_segment(Start_Record& active) noexcept {
|
||||
if (!active.frame || !active.cpu_segment_active) return;
|
||||
close_cpu_segment(active, current_thread_cpu_ns(),
|
||||
current_thread_cpu_cycles());
|
||||
}
|
||||
static void open_cpu_segment(
|
||||
Start_Record& active, std::uint64_t cpu_now_ns,
|
||||
std::uint64_t cycle_now) noexcept {
|
||||
if (!active.frame) return;
|
||||
active.cpu_segment_started_ns = cpu_now_ns;
|
||||
active.cpu_cycle_segment_started = cycle_now;
|
||||
active.cpu_segment_active = true;
|
||||
}
|
||||
static void open_cpu_segment(Start_Record& active) noexcept {
|
||||
if (!active.frame) return;
|
||||
open_cpu_segment(active, current_thread_cpu_ns(),
|
||||
current_thread_cpu_cycles());
|
||||
}
|
||||
void pause_worker(std::size_t worker) noexcept {
|
||||
if (worker >= starts.size() || starts[worker].empty()) return;
|
||||
const auto now = Clock::now();
|
||||
@@ -324,15 +249,12 @@ private:
|
||||
active.maximum_segment_ns = std::max(
|
||||
active.maximum_segment_ns, elapsed);
|
||||
}
|
||||
close_cpu_segment(active);
|
||||
active.segment_started = {};
|
||||
if (active.cooperative_wait_started == Clock::time_point{})
|
||||
active.cooperative_wait_started = now;
|
||||
active.cooperatively_suspended = true;
|
||||
worker_statistics[worker].active_segment_started_ns.store(
|
||||
0, std::memory_order_release);
|
||||
worker_statistics[worker].active_segment_cpu_started_ns.store(
|
||||
0, std::memory_order_release);
|
||||
#if defined(_WIN32)
|
||||
worker_statistics[worker].active_cpu_cycles.store(
|
||||
0, std::memory_order_release);
|
||||
@@ -352,17 +274,11 @@ private:
|
||||
}
|
||||
active.cooperatively_suspended = false;
|
||||
active.segment_started = now;
|
||||
open_cpu_segment(active);
|
||||
worker_statistics[worker].active_segment_started_ns.store(
|
||||
clock_ns(now), std::memory_order_release);
|
||||
worker_statistics[worker].active_segment_cpu_started_ns.store(
|
||||
current_thread_cpu_ns(), std::memory_order_release);
|
||||
#if defined(_WIN32)
|
||||
const auto handle = worker_statistics[worker].native_thread_handle.load(
|
||||
std::memory_order_acquire);
|
||||
worker_statistics[worker].active_cpu_cycles.store(
|
||||
thread_cpu_cycles(reinterpret_cast<HANDLE>(handle)),
|
||||
std::memory_order_release);
|
||||
worker_statistics[worker].active_cpu_cycles.store(0,
|
||||
std::memory_order_release);
|
||||
worker_statistics[worker].active_cpu_progress_ns.store(
|
||||
clock_ns(now), std::memory_order_release);
|
||||
#endif
|
||||
@@ -392,10 +308,9 @@ public:
|
||||
void set_up(std::size_t workers) override {
|
||||
starts.resize(workers);
|
||||
worker_busy_starts.resize(workers);
|
||||
worker_cpu_starts.resize(workers);
|
||||
worker_statistics = std::make_unique<Worker_Statistics[]>(workers);
|
||||
worker_statistics_count = workers;
|
||||
for (auto& worker : starts) worker.reserve(8);
|
||||
for (auto& worker : starts) worker.reserve(32);
|
||||
watchdog_thread = std::thread([this] {
|
||||
run_watchdog();
|
||||
});
|
||||
@@ -412,12 +327,11 @@ public:
|
||||
GetCurrentProcess(), GetCurrentThread(),
|
||||
GetCurrentProcess(), &duplicated, 0, FALSE,
|
||||
DUPLICATE_SAME_ACCESS)) {
|
||||
std::uintptr_t expected{};
|
||||
if (!worker_state.native_thread_handle.compare_exchange_strong(
|
||||
expected, reinterpret_cast<std::uintptr_t>(duplicated),
|
||||
std::memory_order_release,
|
||||
std::memory_order_relaxed))
|
||||
CloseHandle(duplicated);
|
||||
worker_state.cpu_time_origin_ns.store(
|
||||
thread_cpu_ns(duplicated), std::memory_order_relaxed);
|
||||
worker_state.native_thread_handle.store(
|
||||
reinterpret_cast<std::uintptr_t>(duplicated),
|
||||
std::memory_order_release);
|
||||
}
|
||||
}
|
||||
#endif
|
||||
@@ -429,19 +343,16 @@ public:
|
||||
now - parent.segment_started).count());
|
||||
parent.maximum_segment_ns = std::max(
|
||||
parent.maximum_segment_ns, elapsed);
|
||||
close_cpu_segment(parent);
|
||||
parent.segment_started = {};
|
||||
}
|
||||
}
|
||||
if (worker_starts.empty()) {
|
||||
worker_busy_starts[worker.id()] = now;
|
||||
worker_cpu_starts[worker.id()] = current_thread_cpu_ns();
|
||||
}
|
||||
if (worker_starts.empty()) worker_busy_starts[worker.id()] = now;
|
||||
const auto queue_size = worker.queue_size();
|
||||
const auto queue_capacity = worker.queue_capacity();
|
||||
Start_Record record{};
|
||||
record.entered = now;
|
||||
record.cpu_entered_ns = current_thread_cpu_ns();
|
||||
record.queue_size = worker.queue_size();
|
||||
record.queue_capacity = worker.queue_capacity();
|
||||
record.queue_size = queue_size;
|
||||
record.queue_capacity = queue_capacity;
|
||||
record.native_id = static_cast<std::uint64_t>(task.hash_value());
|
||||
record.type = task.type();
|
||||
worker_starts.push_back(std::move(record));
|
||||
@@ -450,55 +361,30 @@ public:
|
||||
const auto active_depth = worker_state.active_depth.fetch_add(
|
||||
1, std::memory_order_relaxed) + 1;
|
||||
update_max(worker_state.peak_active_depth, active_depth);
|
||||
worker_state.current_queue_size.store(worker.queue_size(), std::memory_order_relaxed);
|
||||
worker_state.current_queue_capacity.store(worker.queue_capacity(), std::memory_order_relaxed);
|
||||
worker_state.current_queue_size.store(queue_size, std::memory_order_relaxed);
|
||||
worker_state.current_queue_capacity.store(queue_capacity, std::memory_order_relaxed);
|
||||
worker_state.active_task_hash.store(task.hash_value(), std::memory_order_relaxed);
|
||||
worker_state.active_task_started_ns.store(clock_ns(now), std::memory_order_relaxed);
|
||||
worker_state.active_segment_started_ns.store(clock_ns(now),
|
||||
std::memory_order_release);
|
||||
worker_state.active_segment_cpu_started_ns.store(
|
||||
current_thread_cpu_ns(), std::memory_order_release);
|
||||
#if defined(_WIN32)
|
||||
const auto native_handle = worker_state.native_thread_handle.load(
|
||||
std::memory_order_acquire);
|
||||
worker_state.active_cpu_cycles.store(
|
||||
thread_cpu_cycles(reinterpret_cast<HANDLE>(native_handle)),
|
||||
std::memory_order_release);
|
||||
worker_state.active_cpu_cycles.store(0, std::memory_order_release);
|
||||
worker_state.active_cpu_progress_ns.store(
|
||||
clock_ns(now), std::memory_order_release);
|
||||
#endif
|
||||
worker_state.active_task_type.store(task.type(), std::memory_order_relaxed);
|
||||
worker_state.active_task_reported.store(false, std::memory_order_relaxed);
|
||||
update_max(worker_state.peak_queue_size, worker.queue_size());
|
||||
update_max(worker_state.max_queue_capacity, worker.queue_capacity());
|
||||
update_max(worker_state.max_predecessors, task.num_predecessors());
|
||||
update_max(worker_state.max_successors, task.num_successors());
|
||||
update_max(worker_state.max_strong_dependencies,
|
||||
task.num_strong_dependencies());
|
||||
update_max(worker_state.max_weak_dependencies,
|
||||
task.num_weak_dependencies());
|
||||
update_max(worker_state.peak_queue_size, queue_size);
|
||||
update_max(worker_state.max_queue_capacity, queue_capacity);
|
||||
if (!task.name().empty()) worker_state.named_task_count.fetch_add(1, std::memory_order_relaxed);
|
||||
update_first(worker_state.first_task_time_ns, clock_ns(now));
|
||||
const auto task_cpu_started_ns = current_thread_cpu_ns();
|
||||
const auto task_cpu_cycle_started = worker_starts.back().frame
|
||||
? current_thread_cpu_cycles() : 0;
|
||||
worker_starts.back().cpu_started_ns = task_cpu_started_ns;
|
||||
worker_starts.back().cpu_segment_started_ns = task_cpu_started_ns;
|
||||
worker_starts.back().cpu_cycle_segment_started =
|
||||
task_cpu_cycle_started;
|
||||
worker_starts.back().cpu_segment_active =
|
||||
worker_starts.back().frame != nullptr;
|
||||
worker_starts.back().started = Clock::now();
|
||||
worker_starts.back().segment_started = worker_starts.back().started;
|
||||
}
|
||||
void on_exit(tf::WorkerView worker, tf::TaskView task) override {
|
||||
const auto finished = Clock::now();
|
||||
const auto cpu_finished_ns = current_thread_cpu_ns();
|
||||
auto& worker_starts = starts[worker.id()];
|
||||
auto& active = worker_starts.back();
|
||||
const auto cpu_finished_cycles = active.frame
|
||||
? current_thread_cpu_cycles() : 0;
|
||||
close_cpu_segment(active, cpu_finished_ns, cpu_finished_cycles);
|
||||
if (active.cooperative_wait_started != Clock::time_point{}) {
|
||||
active.cooperative_wait_ns += static_cast<std::uint64_t>(
|
||||
std::chrono::duration_cast<std::chrono::nanoseconds>(
|
||||
@@ -538,9 +424,6 @@ public:
|
||||
longest, elapsed, std::memory_order_relaxed)) {
|
||||
worker_state.longest_task_hash.store(task.hash_value(), std::memory_order_relaxed);
|
||||
worker_state.longest_task_type.store(task.type(), std::memory_order_relaxed);
|
||||
worker_state.longest_task_name.store(
|
||||
std::make_shared<const std::string>(task.name()),
|
||||
std::memory_order_release);
|
||||
}
|
||||
worker_state.active_depth.fetch_sub(1, std::memory_order_relaxed);
|
||||
std::optional<std::size_t> trace_task;
|
||||
@@ -550,39 +433,27 @@ public:
|
||||
*start.frame, worker.id(),
|
||||
static_cast<std::uint64_t>(task.hash_value()),
|
||||
start.queue_size, start.queue_capacity, start.entered,
|
||||
start.started, finished, start.cpu_entered_ns,
|
||||
start.cpu_started_ns, start.cpu_duration_ns,
|
||||
start.cpu_cycles, start.cooperative_wait_ns,
|
||||
thread_cpu_time_is_coarse());
|
||||
start.started, finished, start.cooperative_wait_ns);
|
||||
}
|
||||
catch (...) {
|
||||
/* Observer 不能让按需诊断分配失败改变渲染任务的完成语义。 */
|
||||
}
|
||||
}
|
||||
const auto cpu_completed_ns = current_thread_cpu_ns();
|
||||
const auto completed = Clock::now();
|
||||
if (start.frame) {
|
||||
if (trace_task)
|
||||
detail::Taskflow_Frame_Access::finish_task_observer(
|
||||
*start.frame, worker.id(), *trace_task, completed,
|
||||
cpu_finished_ns, cpu_completed_ns);
|
||||
*start.frame, worker.id(), *trace_task, completed);
|
||||
}
|
||||
if (worker_starts.empty()) {
|
||||
auto busy = static_cast<std::uint64_t>(
|
||||
std::chrono::duration_cast<std::chrono::nanoseconds>(
|
||||
completed - worker_busy_starts[worker.id()]).count());
|
||||
worker_state.busy_time_ns.fetch_add(busy, std::memory_order_relaxed);
|
||||
const auto cpu_started = worker_cpu_starts[worker.id()];
|
||||
const auto cpu = cpu_completed_ns >= cpu_started
|
||||
? cpu_completed_ns - cpu_started
|
||||
: 0;
|
||||
worker_state.cpu_time_ns.fetch_add(cpu, std::memory_order_relaxed);
|
||||
worker_state.active_task_hash.store(0, std::memory_order_relaxed);
|
||||
worker_state.active_task_started_ns.store(0, std::memory_order_relaxed);
|
||||
worker_state.active_segment_started_ns.store(0,
|
||||
std::memory_order_release);
|
||||
worker_state.active_segment_cpu_started_ns.store(
|
||||
0, std::memory_order_release);
|
||||
#if defined(_WIN32)
|
||||
worker_state.active_cpu_cycles.store(0, std::memory_order_release);
|
||||
worker_state.active_cpu_progress_ns.store(0,
|
||||
@@ -601,8 +472,6 @@ public:
|
||||
if (parent.cooperatively_suspended) {
|
||||
worker_state.active_segment_started_ns.store(
|
||||
0, std::memory_order_release);
|
||||
worker_state.active_segment_cpu_started_ns.store(
|
||||
0, std::memory_order_release);
|
||||
#if defined(_WIN32)
|
||||
worker_state.active_cpu_cycles.store(0,
|
||||
std::memory_order_release);
|
||||
@@ -611,20 +480,13 @@ public:
|
||||
#endif
|
||||
}
|
||||
else {
|
||||
const auto parent_cpu_started_ns = current_thread_cpu_ns();
|
||||
const auto parent_cpu_cycle_started = parent.frame
|
||||
? current_thread_cpu_cycles() : 0;
|
||||
const auto parent_resumed = Clock::now();
|
||||
parent.segment_started = parent_resumed;
|
||||
open_cpu_segment(parent, parent_cpu_started_ns,
|
||||
parent_cpu_cycle_started);
|
||||
worker_state.active_segment_started_ns.store(
|
||||
clock_ns(parent_resumed), std::memory_order_release);
|
||||
worker_state.active_segment_cpu_started_ns.store(
|
||||
parent_cpu_started_ns, std::memory_order_release);
|
||||
#if defined(_WIN32)
|
||||
worker_state.active_cpu_cycles.store(
|
||||
parent_cpu_cycle_started, std::memory_order_release);
|
||||
worker_state.active_cpu_cycles.store(0,
|
||||
std::memory_order_release);
|
||||
worker_state.active_cpu_progress_ns.store(
|
||||
clock_ns(parent_resumed), std::memory_order_release);
|
||||
#endif
|
||||
@@ -657,10 +519,6 @@ public:
|
||||
state.named_task_count = 0;
|
||||
state.peak_observed_worker_queue_size = 0;
|
||||
state.max_observed_worker_queue_capacity = 0;
|
||||
state.max_predecessors = 0;
|
||||
state.max_successors = 0;
|
||||
state.max_strong_dependencies = 0;
|
||||
state.max_weak_dependencies = 0;
|
||||
state.total_task_time_ns = 0;
|
||||
state.worker_busy_time_ns = 0;
|
||||
state.worker_cpu_time_ns = 0;
|
||||
@@ -696,7 +554,17 @@ public:
|
||||
: std::string{};
|
||||
target.task_time_ns = source.task_time_ns.load(std::memory_order_relaxed);
|
||||
target.busy_time_ns = source.busy_time_ns.load(std::memory_order_relaxed);
|
||||
target.cpu_time_ns = source.cpu_time_ns.load(std::memory_order_relaxed);
|
||||
#if defined(_WIN32)
|
||||
const auto native_handle = source.native_thread_handle.load(
|
||||
std::memory_order_acquire);
|
||||
const auto cpu_origin = source.cpu_time_origin_ns.load(
|
||||
std::memory_order_acquire);
|
||||
const auto cpu_now = thread_cpu_ns(
|
||||
reinterpret_cast<HANDLE>(native_handle));
|
||||
target.cpu_time_ns = cpu_now >= cpu_origin ? cpu_now - cpu_origin : 0;
|
||||
#else
|
||||
target.cpu_time_ns = 0;
|
||||
#endif
|
||||
target.non_cpu_time_ns = target.busy_time_ns > target.cpu_time_ns
|
||||
? target.busy_time_ns - target.cpu_time_ns
|
||||
: 0;
|
||||
@@ -724,16 +592,6 @@ public:
|
||||
state.max_observed_worker_queue_capacity = std::max(
|
||||
state.max_observed_worker_queue_capacity,
|
||||
target.max_observed_queue_capacity);
|
||||
state.max_predecessors = std::max(state.max_predecessors,
|
||||
source.max_predecessors.load(std::memory_order_relaxed));
|
||||
state.max_successors = std::max(state.max_successors,
|
||||
source.max_successors.load(std::memory_order_relaxed));
|
||||
state.max_strong_dependencies = std::max(
|
||||
state.max_strong_dependencies,
|
||||
source.max_strong_dependencies.load(std::memory_order_relaxed));
|
||||
state.max_weak_dependencies = std::max(
|
||||
state.max_weak_dependencies,
|
||||
source.max_weak_dependencies.load(std::memory_order_relaxed));
|
||||
state.total_task_time_ns += target.task_time_ns;
|
||||
state.worker_busy_time_ns += target.busy_time_ns;
|
||||
state.worker_cpu_time_ns += target.cpu_time_ns;
|
||||
@@ -765,8 +623,8 @@ public:
|
||||
state.longest_task_time_ns = longest;
|
||||
state.longest_task_hash = source.longest_task_hash.load(
|
||||
std::memory_order_relaxed);
|
||||
const auto name = source.longest_task_name.load(
|
||||
std::memory_order_acquire);
|
||||
const auto name = detail::taskflow_node_name(
|
||||
state.longest_task_hash);
|
||||
state.longest_task_name = name ? *name : std::string{};
|
||||
state.longest_task_type = std::string(tf::to_string(
|
||||
source.longest_task_type.load(std::memory_order_relaxed)));
|
||||
|
||||
@@ -80,10 +80,6 @@ struct Task_Runtime_State : State_Type<Task_Runtime_State_Tag> {
|
||||
std::size_t named_task_count{};
|
||||
std::size_t peak_observed_worker_queue_size{};
|
||||
std::size_t max_observed_worker_queue_capacity{};
|
||||
std::size_t max_predecessors{};
|
||||
std::size_t max_successors{};
|
||||
std::size_t max_strong_dependencies{};
|
||||
std::size_t max_weak_dependencies{};
|
||||
std::size_t longest_task_hash{};
|
||||
std::string longest_task_name;
|
||||
std::string longest_task_type;
|
||||
|
||||
@@ -341,14 +341,9 @@ nlohmann::json taskflow_trace_json(
|
||||
{"started_ms", task.started_ms}, {"finished_ms", task.finished_ms},
|
||||
{"completed_ms", task.completed_ms},
|
||||
{"duration_ms", task.duration_ms},
|
||||
{"cpu_duration_ms", task.cpu_duration_ms},
|
||||
{"cpu_cycles", task.cpu_cycles},
|
||||
{"cooperative_wait_ms", task.cooperative_wait_ms},
|
||||
{"cpu_time_coarse", task.cpu_time_coarse},
|
||||
{"observer_entry_ms", task.observer_entry_ms},
|
||||
{"observer_exit_ms", task.observer_exit_ms},
|
||||
{"observer_entry_cpu_ms", task.observer_entry_cpu_ms},
|
||||
{"observer_exit_cpu_ms", task.observer_exit_cpu_ms},
|
||||
{"queue_wait_ms", task.queue_wait_ms}});
|
||||
}
|
||||
return {
|
||||
|
||||
@@ -84,10 +84,6 @@ nlohmann::json taskflow_runtime_json() {
|
||||
{"named_tasks", state.named_task_count},
|
||||
{"peak_worker_queue_size", state.peak_observed_worker_queue_size},
|
||||
{"max_worker_queue_capacity", state.max_observed_worker_queue_capacity},
|
||||
{"max_predecessors", state.max_predecessors},
|
||||
{"max_successors", state.max_successors},
|
||||
{"max_strong_dependencies", state.max_strong_dependencies},
|
||||
{"max_weak_dependencies", state.max_weak_dependencies},
|
||||
{"longest_task", {{"native_id", std::to_string(state.longest_task_hash)},
|
||||
{"name", state.longest_task_name},
|
||||
{"type", state.longest_task_type},
|
||||
|
||||
+32
-69
@@ -81,9 +81,8 @@ type Taskflow_Graph_Trace = {stage: string; name: string; submitted_ms: number;
|
||||
completed: boolean; nodes: Taskflow_Node_Trace[]};
|
||||
type Taskflow_Execution_Trace = {native_id: string; node_id: string; worker_id: number; worker_queue_size: number;
|
||||
worker_queue_capacity: number; ready_ms: number; entered_ms: number; started_ms: number; finished_ms: number;
|
||||
completed_ms: number; duration_ms: number; cpu_duration_ms: number; cpu_cycles: number; cooperative_wait_ms: number;
|
||||
cpu_time_coarse: boolean; observer_entry_ms: number; observer_exit_ms: number;
|
||||
observer_entry_cpu_ms: number; observer_exit_cpu_ms: number; queue_wait_ms: number};
|
||||
completed_ms: number; duration_ms: number; cooperative_wait_ms: number;
|
||||
observer_entry_ms: number; observer_exit_ms: number; queue_wait_ms: number};
|
||||
type Frame_Policy_State = {
|
||||
generation: number;
|
||||
configuration: {mode: Frame_Pacing_Mode; render_enabled: boolean; video_enabled: boolean; fixed_rate_fps: number};
|
||||
@@ -113,8 +112,7 @@ type Taskflow_Type_State = {name: string; count: number; total_time_ns: number;
|
||||
type Taskflow_Runtime_State = {protocol: "aethera.taskflow.runtime"; version: 1; worker_count: number; active_topologies: number;
|
||||
active_taskflows: number; peak_active_taskflows: number; completed_taskflows: number; failed_taskflows: number;
|
||||
active_tasks: number; peak_active_tasks: number; active_workers: number; peak_active_workers: number; observed_tasks: number;
|
||||
named_tasks: number; peak_worker_queue_size: number; max_worker_queue_capacity: number; max_predecessors: number;
|
||||
max_successors: number; max_strong_dependencies: number; max_weak_dependencies: number;
|
||||
named_tasks: number; peak_worker_queue_size: number; max_worker_queue_capacity: number;
|
||||
longest_task: {native_id: string; name: string; type: string; time_ns: number}; total_task_time_ns: number;
|
||||
worker_busy_time_ns: number; worker_cpu_time_ns: number; observed_wall_time_ns: number;
|
||||
worker_utilization: number; worker_cpu_utilization: number;
|
||||
@@ -1203,20 +1201,6 @@ function nanoseconds(value: number) {
|
||||
return milliseconds(value / 1_000_000);
|
||||
}
|
||||
|
||||
function cpu_cycles(value: number) {
|
||||
if (!Number.isFinite(value) || value <= 0) return "--";
|
||||
if (value >= 1_000_000_000) return `${(value / 1_000_000_000).toFixed(2)} Gcy`;
|
||||
if (value >= 1_000_000) return `${(value / 1_000_000).toFixed(2)} Mcy`;
|
||||
if (value >= 1_000) return `${(value / 1_000).toFixed(1)} Kcy`;
|
||||
return `${Math.round(value)} cy`;
|
||||
}
|
||||
|
||||
function task_cpu_label(sample: Taskflow_Execution_Trace) {
|
||||
if (sample.cpu_time_coarse && sample.cpu_duration_ms === 0 && sample.cpu_cycles > 0)
|
||||
return "低于系统 CPU 计时分辨率";
|
||||
return milliseconds(sample.cpu_duration_ms);
|
||||
}
|
||||
|
||||
type Taskflow_Node_Phase = "prepare" | "paint" | "control" | "completion" | "other";
|
||||
|
||||
function taskflow_node_phase(name: string): Taskflow_Node_Phase {
|
||||
@@ -1340,7 +1324,7 @@ function Taskflow_Node_Label({node, sample, summary, state, level}: {
|
||||
node, execution: sample ?? null,
|
||||
aggregate: summary ? {
|
||||
present: summary.present, executed: summary.executed, frames: summary.frames,
|
||||
duration: summary.duration, queue: summary.queue, cpu: summary.cpu, observer: summary.observer,
|
||||
duration: summary.duration, queue: summary.queue, observer: summary.observer,
|
||||
workers: summary.workers, stability: summary.stability
|
||||
} : null,
|
||||
state: state ?? null
|
||||
@@ -1375,8 +1359,7 @@ function Taskflow_Node_Label({node, sample, summary, state, level}: {
|
||||
<span>{node.type} · 样本 {summary.executed}/{summary.frames} · W {summary.workers.join(", ") || "--"}</span>
|
||||
<span>执行 {milliseconds(summary.duration.average)} ± {milliseconds(summary.duration.variability)} · P95 {milliseconds(summary.duration.p95)}</span>
|
||||
<span>排队 {milliseconds(summary.queue.average)} ± {milliseconds(summary.queue.variability)} · P99 {milliseconds(summary.queue.p99)}</span>
|
||||
<span>线程 CPU{summary.cpu_time_coarse ? "（低分辨率）" : ""} {milliseconds(summary.cpu.average)} · CPU 活动 {cpu_cycles(summary.cycles.average)}</span>
|
||||
<span>协作等待 {milliseconds(summary.cooperative_wait.average)} · Observer {milliseconds(summary.observer.average)}</span>
|
||||
<span>协作等待 {milliseconds(summary.cooperative_wait.average)} · Observer 墙钟 {milliseconds(summary.observer.average)}</span>
|
||||
</> : node.type === "diagnostic" ? <>
|
||||
<span>运行时收尾</span>
|
||||
<span>开始 {sample ? `+${sample.started_ms.toFixed(3)} ms` : "--"} · 结束 {sample ? `+${sample.finished_ms.toFixed(3)} ms` : "--"}</span>
|
||||
@@ -1385,8 +1368,7 @@ function Taskflow_Node_Label({node, sample, summary, state, level}: {
|
||||
<span>{node.type} · W{sample?.worker_id ?? "--"}</span>
|
||||
<span>就绪 {sample ? `+${sample.ready_ms.toFixed(3)} ms` : "--"} · 开始 {sample ? `+${sample.started_ms.toFixed(3)} ms` : "--"}</span>
|
||||
<span>任务体 {sample ? milliseconds(duration) : "未执行"} · 排队 {sample ? milliseconds(wait) : "--"}</span>
|
||||
<span>线程 CPU {sample ? task_cpu_label(sample) : "--"}{sample?.cpu_time_coarse ? "（低分辨率）" : ""} · CPU 活动 {sample ? cpu_cycles(sample.cpu_cycles) : "--"}</span>
|
||||
<span>协作等待 {sample ? milliseconds(sample.cooperative_wait_ms) : "--"} · 未归因墙时 {sample ? `${sample.cpu_time_coarse ? "≈ " : ""}${milliseconds(Math.max(0, duration - sample.cooperative_wait_ms - sample.cpu_duration_ms))}` : "--"}</span>
|
||||
<span>协作等待 {sample ? milliseconds(sample.cooperative_wait_ms) : "--"} · 独占任务体 {sample ? milliseconds(Math.max(0, duration - sample.cooperative_wait_ms)) : "--"}</span>
|
||||
<span>Observer {sample ? milliseconds(sample.observer_entry_ms + sample.observer_exit_ms) : "--"}</span></>}
|
||||
{Object.entries(node.attributes ?? {}).map(([key, value]) => <span key={key}>{key}：{value}</span>)}
|
||||
</div>;
|
||||
@@ -1434,23 +1416,19 @@ function taskflow_graph_analysis(graph: Taskflow_Graph_Trace, executions: Taskfl
|
||||
const execution_time = leaf_rows.reduce((sum, row) => sum + row.duration_ms, 0);
|
||||
const module_envelope_time = module_rows.reduce((sum, row) => sum + row.duration_ms, 0);
|
||||
const diagnostic_time = diagnostic_rows.reduce((sum, row) => sum + row.duration_ms, 0);
|
||||
const cpu_execution_time = leaf_rows.reduce((sum, row) => sum + row.cpu_duration_ms, 0);
|
||||
const cooperative_wait_time = leaf_rows.reduce((sum, row) => sum + row.cooperative_wait_ms, 0);
|
||||
const cpu_cycles_total = leaf_rows.reduce((sum, row) => sum + row.cpu_cycles, 0);
|
||||
const cpu_time_coarse = leaf_rows.some(row => row.cpu_time_coarse);
|
||||
const unattributed_wall_time = Math.max(0, execution_time - cooperative_wait_time - cpu_execution_time);
|
||||
const exclusive_body_time = Math.max(0, execution_time - cooperative_wait_time);
|
||||
const observer_entry_time = task_rows.reduce((sum, row) => sum + row.observer_entry_ms, 0);
|
||||
const observer_exit_time = task_rows.reduce((sum, row) => sum + row.observer_exit_ms, 0);
|
||||
const observer_cpu_time = task_rows.reduce((sum, row) => sum + row.observer_entry_cpu_ms + row.observer_exit_cpu_ms, 0);
|
||||
const queue_time = task_rows.reduce((sum, row) => sum + row.queue_wait_ms, 0);
|
||||
const first_entered = task_rows.length ? Math.min(...task_rows.map(row => row.entered_ms)) : graph.submitted_ms;
|
||||
const last_completed = task_rows.length ? Math.max(...task_rows.map(row => row.completed_ms)) : graph.finished_ms;
|
||||
const longest = task_rows.reduce<Taskflow_Execution_Trace | null>(
|
||||
(result, row) => !result || row.duration_ms > result.duration_ms ? row : result, null);
|
||||
const longest_unattributed = task_rows.reduce<Taskflow_Execution_Trace | null>((result, row) => {
|
||||
const value = Math.max(0, row.duration_ms - row.cooperative_wait_ms - row.cpu_duration_ms);
|
||||
const longest_exclusive = task_rows.reduce<Taskflow_Execution_Trace | null>((result, row) => {
|
||||
const value = Math.max(0, row.duration_ms - row.cooperative_wait_ms);
|
||||
const previous = result
|
||||
? Math.max(0, result.duration_ms - result.cooperative_wait_ms - result.cpu_duration_ms)
|
||||
? Math.max(0, result.duration_ms - result.cooperative_wait_ms)
|
||||
: -1;
|
||||
return value > previous ? row : result;
|
||||
}, null);
|
||||
@@ -1510,8 +1488,8 @@ function taskflow_graph_analysis(graph: Taskflow_Graph_Trace, executions: Taskfl
|
||||
for (const level of levels.values()) width_by_level.set(level, (width_by_level.get(level) ?? 0) + 1);
|
||||
return {
|
||||
rows, task_rows, leaf_rows, module_rows, diagnostic_rows, levels, wall_time, execution_time,
|
||||
module_envelope_time, diagnostic_time, diagnostic_wall_time, cpu_execution_time, cooperative_wait_time,
|
||||
cpu_cycles_total, cpu_time_coarse, unattributed_wall_time, observer_entry_time, observer_cpu_time,
|
||||
module_envelope_time, diagnostic_time, diagnostic_wall_time, cooperative_wait_time,
|
||||
exclusive_body_time, observer_entry_time,
|
||||
observer_exit_time, queue_time, body_wall_time, observer_wall_time, idle_wall_time, initial_wait,
|
||||
completion_tail, unattributed_completion_tail,
|
||||
internal_idle_time: Math.max(0, idle_wall_time - initial_wait - unattributed_completion_tail),
|
||||
@@ -1521,12 +1499,11 @@ function taskflow_graph_analysis(graph: Taskflow_Graph_Trace, executions: Taskfl
|
||||
diagnostic_node_count: graph.nodes.filter(node => node.type.toLowerCase() === "diagnostic").length,
|
||||
layer_count: width_by_level.size,
|
||||
parallel_layer_count: [...width_by_level.values()].filter(width => width > 1).length,
|
||||
longest, longest_unattributed,
|
||||
longest, longest_exclusive,
|
||||
longest_duration: longest?.duration_ms ?? 0,
|
||||
longest_unattributed_duration: longest_unattributed
|
||||
? Math.max(0, longest_unattributed.duration_ms -
|
||||
longest_unattributed.cooperative_wait_ms -
|
||||
longest_unattributed.cpu_duration_ms)
|
||||
longest_exclusive_duration: longest_exclusive
|
||||
? Math.max(0, longest_exclusive.duration_ms -
|
||||
longest_exclusive.cooperative_wait_ms)
|
||||
: 0
|
||||
};
|
||||
}
|
||||
@@ -1539,15 +1516,12 @@ const taskflow_graph_metric_definitions = [
|
||||
{key: "module_envelope_time", label: "Module 包络总和", unit: "milliseconds", description: "每帧所有 Taskflow Module Observer 包络之和。"},
|
||||
{key: "diagnostic_time", label: "收尾诊断总和", unit: "milliseconds", description: "Executor 返回后的运行计数、状态发布、Observer graph 退役与完成记录墙钟之和。"},
|
||||
{key: "diagnostic_wall_time", label: "收尾诊断并集", unit: "milliseconds", description: "Topology 尾部被明确归因到诊断收尾节点的时间区间并集。"},
|
||||
{key: "cpu_execution_time", label: "叶子任务线程 CPU", unit: "milliseconds", description: "每帧叶子任务独占 Worker 的线程 CPU 时间之和。"},
|
||||
{key: "cpu_cycles_total", label: "叶子任务 CPU 活动", unit: "cycles", description: "每帧叶子任务 QueryThreadCycleTime 活动量之和。"},
|
||||
{key: "cooperative_wait_time", label: "累计协作等待", unit: "milliseconds", description: "每帧 Task_Graph 协作等待墙钟之和。"},
|
||||
{key: "unattributed_wall_time", label: "未归因墙时", unit: "milliseconds", description: "任务体墙钟扣除协作等待和线程 CPU 后的守恒余量。"},
|
||||
{key: "exclusive_body_time", label: "独占任务体墙钟", unit: "milliseconds", description: "任务体墙钟扣除显式协作让出后的总和。"},
|
||||
{key: "body_wall_time", label: "任务体墙钟并集", unit: "milliseconds", description: "每帧至少有一个任务体执行的时间区间并集。"},
|
||||
{key: "observer_wall_time", label: "Observer 独占墙钟", unit: "milliseconds", description: "只有 Observer 活动、没有任务体执行的时间区间。"},
|
||||
{key: "observer_entry_time", label: "Observer entry 墙钟", unit: "milliseconds", description: "每帧所有 Observer entry 墙钟之和。"},
|
||||
{key: "observer_exit_time", label: "Observer exit 墙钟", unit: "milliseconds", description: "每帧所有 Observer exit 墙钟之和。"},
|
||||
{key: "observer_cpu_time", label: "Observer 实际 CPU", unit: "milliseconds", description: "每帧 Observer entry/exit 真正占用的 Worker CPU 时间。"},
|
||||
{key: "idle_wall_time", label: "无任务墙钟", unit: "milliseconds", description: "Topology 中没有捕获任务体或 Observer 活动的区间。"},
|
||||
{key: "queue_time", label: "累计节点排队", unit: "milliseconds", description: "各节点从前驱完成到进入 Worker 的估算等待之和。"},
|
||||
{key: "initial_wait", label: "首次准入等待", unit: "milliseconds", description: "Topology 提交后首节点进入 Worker 前的等待。"},
|
||||
@@ -1559,20 +1533,20 @@ const taskflow_graph_metric_definitions = [
|
||||
{key: "layer_count", label: "依赖层", unit: "count", description: "每帧 DAG 的依赖层数量。"},
|
||||
{key: "parallel_layer_count", label: "并行层", unit: "count", description: "每帧包含多个可并行节点的依赖层数量。"},
|
||||
{key: "longest_duration", label: "最长节点执行", unit: "milliseconds", description: "每帧最长执行节点的任务体墙钟。"},
|
||||
{key: "longest_unattributed_duration", label: "最长节点未归因墙时", unit: "milliseconds", description: "每帧单节点最大的未归因墙钟。"}
|
||||
{key: "longest_exclusive_duration", label: "最长独占任务体", unit: "milliseconds", description: "每帧单节点扣除显式协作让出后的最大任务体墙钟。"}
|
||||
] as const;
|
||||
type Taskflow_Graph_Metric_Key = typeof taskflow_graph_metric_definitions[number]["key"];
|
||||
type Taskflow_Node_Stability = "stable" | "variable" | "missing" | "long_tail";
|
||||
type Taskflow_Node_Aggregate = {
|
||||
node: Taskflow_Node_Trace; present: number; executed: number; frames: number;
|
||||
duration: Distribution_Statistics; queue: Distribution_Statistics; cpu: Distribution_Statistics;
|
||||
cycles: Distribution_Statistics; cooperative_wait: Distribution_Statistics; cpu_time_coarse: boolean;
|
||||
duration: Distribution_Statistics; queue: Distribution_Statistics;
|
||||
cooperative_wait: Distribution_Statistics;
|
||||
observer: Distribution_Statistics; workers: number[]; stability: Taskflow_Node_Stability;
|
||||
};
|
||||
type Taskflow_Graph_Aggregate = {
|
||||
key: string; graph: Taskflow_Graph_Trace; frames: number; samples: number; topology_variants: number;
|
||||
metrics: Record<Taskflow_Graph_Metric_Key, Distribution_Statistics>;
|
||||
completed_frames: number; cpu_time_coarse_frames: number;
|
||||
completed_frames: number;
|
||||
nodes: Map<string, Taskflow_Node_Aggregate>; edge_presence: Map<string, number>;
|
||||
};
|
||||
|
||||
@@ -1602,7 +1576,7 @@ function aggregate_taskflow_graph(frames: Taskflow_Frame_Trace[], key: string):
|
||||
.map(graph => ({frame, graph})));
|
||||
if (!samples.length) return null;
|
||||
type Mutable_Node = {template: Taskflow_Node_Trace; present: number; executed: number; duration: number[];
|
||||
queue: number[]; cpu: number[]; cycles: number[]; cooperative_wait: number[]; cpu_time_coarse: boolean;
|
||||
queue: number[]; cooperative_wait: number[];
|
||||
observer: number[]; workers: Set<number>};
|
||||
const accumulated = new Map<string, Mutable_Node>();
|
||||
const edges = new Map<string, number>();
|
||||
@@ -1610,7 +1584,6 @@ function aggregate_taskflow_graph(frames: Taskflow_Frame_Trace[], key: string):
|
||||
const metric_samples = Object.fromEntries(taskflow_graph_metric_definitions.map(
|
||||
definition => [definition.key, [] as number[]])) as Record<Taskflow_Graph_Metric_Key, number[]>;
|
||||
let completed_frames = 0;
|
||||
let cpu_time_coarse_frames = 0;
|
||||
for (const {frame, graph} of samples) {
|
||||
const native_to_id = new Map(graph.nodes.map(node => [node.native_id, node.id]));
|
||||
const execution = new Map(frame.executions.map(value => [value.native_id, value]));
|
||||
@@ -1618,11 +1591,10 @@ function aggregate_taskflow_graph(frames: Taskflow_Frame_Trace[], key: string):
|
||||
const analysis = taskflow_graph_analysis(graph, frame.executions);
|
||||
for (const definition of taskflow_graph_metric_definitions)
|
||||
metric_samples[definition.key].push(analysis[definition.key]);
|
||||
if (analysis.cpu_time_coarse) ++cpu_time_coarse_frames;
|
||||
if (graph.completed) ++completed_frames;
|
||||
for (const node of graph.nodes) {
|
||||
const current = accumulated.get(node.id) ?? {template: node, present: 0, executed: 0,
|
||||
duration: [], queue: [], cpu: [], cycles: [], cooperative_wait: [], cpu_time_coarse: false,
|
||||
duration: [], queue: [], cooperative_wait: [],
|
||||
observer: [], workers: new Set<number>()};
|
||||
++current.present;
|
||||
const sample = execution.get(node.native_id);
|
||||
@@ -1630,10 +1602,7 @@ function aggregate_taskflow_graph(frames: Taskflow_Frame_Trace[], key: string):
|
||||
++current.executed;
|
||||
current.duration.push(sample.duration_ms);
|
||||
current.queue.push(sample.queue_wait_ms);
|
||||
current.cpu.push(sample.cpu_duration_ms);
|
||||
current.cycles.push(sample.cpu_cycles);
|
||||
current.cooperative_wait.push(sample.cooperative_wait_ms);
|
||||
current.cpu_time_coarse = current.cpu_time_coarse || sample.cpu_time_coarse;
|
||||
current.observer.push(sample.observer_entry_ms + sample.observer_exit_ms);
|
||||
current.workers.add(sample.worker_id);
|
||||
}
|
||||
@@ -1671,8 +1640,7 @@ function aggregate_taskflow_graph(frames: Taskflow_Frame_Trace[], key: string):
|
||||
const node = {...value.template, native_id: id, id,
|
||||
predecessors: predecessors.get(id) ?? [], successors: successors.get(id) ?? []};
|
||||
nodes.set(id, {node, present: value.present, executed: value.executed, frames: frames.length,
|
||||
duration, queue, cpu: distribution_statistics(value.cpu), cycles: distribution_statistics(value.cycles),
|
||||
cooperative_wait: distribution_statistics(value.cooperative_wait), cpu_time_coarse: value.cpu_time_coarse,
|
||||
duration, queue, cooperative_wait: distribution_statistics(value.cooperative_wait),
|
||||
observer: distribution_statistics(value.observer),
|
||||
workers: [...value.workers].sort((left, right) => left - right), stability});
|
||||
return node;
|
||||
@@ -1682,24 +1650,23 @@ function aggregate_taskflow_graph(frames: Taskflow_Frame_Trace[], key: string):
|
||||
definition => [definition.key, distribution_statistics(metric_samples[definition.key])])) as
|
||||
Record<Taskflow_Graph_Metric_Key, Distribution_Statistics>;
|
||||
return {key, frames: frames.length, samples: samples.length, topology_variants: topology_signatures.size,
|
||||
metrics, completed_frames, cpu_time_coarse_frames, nodes, edge_presence: edges,
|
||||
metrics, completed_frames, nodes, edge_presence: edges,
|
||||
graph: {...first, submitted_ms: 0, finished_ms: metrics.wall_time.average, nodes: graph_nodes}};
|
||||
}
|
||||
|
||||
function taskflow_metric_value(value: number, unit: "milliseconds" | "cycles" | "count") {
|
||||
function taskflow_metric_value(value: number, unit: "milliseconds" | "count") {
|
||||
if (unit === "milliseconds") return milliseconds(value);
|
||||
if (unit === "cycles") return cpu_cycles(value);
|
||||
return value.toLocaleString("zh-CN", {maximumFractionDigits: 3});
|
||||
}
|
||||
|
||||
function taskflow_variance_value(value: number, unit: "milliseconds" | "cycles" | "count") {
|
||||
function taskflow_variance_value(value: number, unit: "milliseconds" | "count") {
|
||||
if (!Number.isFinite(value)) return "--";
|
||||
const suffix = unit === "milliseconds" ? " ms²" : unit === "cycles" ? " cy²" : "";
|
||||
const suffix = unit === "milliseconds" ? " ms²" : "";
|
||||
return `${value.toLocaleString("zh-CN", {maximumFractionDigits: 6})}${suffix}`;
|
||||
}
|
||||
|
||||
function taskflow_distribution_value(statistic: Distribution_Statistics,
|
||||
unit: "milliseconds" | "cycles" | "count") {
|
||||
unit: "milliseconds" | "count") {
|
||||
return <>均值 {taskflow_metric_value(statistic.average, unit)} · 方差 {taskflow_variance_value(statistic.variance, unit)} · 范围 [{taskflow_metric_value(statistic.minimum, unit)}, {taskflow_metric_value(statistic.maximum, unit)}]</>;
|
||||
}
|
||||
|
||||
@@ -1935,7 +1902,6 @@ function Taskflow_Dag({graph, executions, frame, aggregate, components, gallery_
|
||||
topology_variants: aggregate.topology_variants,
|
||||
metrics: aggregate.metrics,
|
||||
completed_frames: aggregate.completed_frames,
|
||||
cpu_time_coarse_frames: aggregate.cpu_time_coarse_frames,
|
||||
nodes: Object.fromEntries(aggregate.nodes)} : undefined,
|
||||
executions: executions.filter(value => native_ids.has(value.native_id)),
|
||||
all_executions: frame?.executions ?? executions
|
||||
@@ -2414,7 +2380,7 @@ function Taskflow_Frame_Pane({plot, components}: {plot: Plot; components: Compon
|
||||
<div><dt>波动节点</dt><dd>{[...aggregate.nodes.values()].filter(node => node.stability === "variable").length}</dd></div>
|
||||
<div><dt>缺帧节点</dt><dd>{[...aggregate.nodes.values()].filter(node => node.stability === "missing").length}</dd></div>
|
||||
<div><dt>长尾节点</dt><dd>{[...aggregate.nodes.values()].filter(node => node.stability === "long_tail").length}</dd></div>
|
||||
<div><dt>完成帧 / CPU 粗粒度帧</dt><dd>{aggregate.completed_frames} / {aggregate.cpu_time_coarse_frames}</dd></div>
|
||||
<div><dt>完成帧</dt><dd>{aggregate.completed_frames}</dd></div>
|
||||
{taskflow_graph_metric_definitions.map(definition => {
|
||||
const statistic = aggregate.metrics[definition.key];
|
||||
return <div key={definition.key} title={`${definition.description} 样本 ${statistic.count}；标准差 ${taskflow_metric_value(statistic.variability, definition.unit)}；P95 ${taskflow_metric_value(statistic.p95, definition.unit)}；P99 ${taskflow_metric_value(statistic.p99, definition.unit)}。`}>
|
||||
@@ -2427,15 +2393,12 @@ function Taskflow_Frame_Pane({plot, components}: {plot: Plot; components: Compon
|
||||
<div title="提交 Taskflow 到整个 Topology 完成的墙钟时间，包含 Executor 排队和当前既有的完成尾部。"><dt>Topology 墙钟</dt><dd>{milliseconds(graph_analysis.wall_time)}</dd></div>
|
||||
<div title="只累计叶子任务；Taskflow Module 是包住子图的区间，不能与子节点重复相加。"><dt>叶子任务墙钟总和</dt><dd>{milliseconds(graph_analysis.execution_time)}</dd></div>
|
||||
<div title="所有 Module Observer 区间之和，仅显示子图包络；不计入叶子任务总和。"><dt>Module 包络总和</dt><dd>{milliseconds(graph_analysis.module_envelope_time)}</dd></div>
|
||||
<div title="叶子任务独占 Taskflow Worker 片段的线程 CPU 时间；corun/corun_until 期间执行的其他任务不会归到外层节点。Windows GetThreadTimes 仍是低分辨率计费时钟。"><dt>叶子任务线程 CPU{graph_analysis.cpu_time_coarse ? "（低分辨率）" : ""}</dt><dd>{milliseconds(graph_analysis.cpu_execution_time)}</dd></div>
|
||||
<div title="Windows 使用 QueryThreadCycleTime 记录独占片段 CPU 活动量。周期数不直接换算为秒，但短任务即使 GetThreadTimes 显示 0 也能确认确实消耗过 CPU。"><dt>叶子任务 CPU 活动</dt><dd>{cpu_cycles(graph_analysis.cpu_cycles_total)}</dd></div>
|
||||
<div title="Task_Graph::corun/corun_until 主动让出当前 Worker、由同一 Worker 协作执行其他 Taskflow 工作的墙钟时间。"><dt>累计协作等待</dt><dd>{milliseconds(graph_analysis.cooperative_wait_time)}</dd></div>
|
||||
<div title="任务体墙钟减去显式 cooperative wait 和线程 CPU 时间。它可能包含锁/驱动/系统调用等待以及 Windows CPU 计时量化误差，因此不是 OS 抢占时间。"><dt>未归因墙时{graph_analysis.cpu_time_coarse ? "（估算）" : ""}</dt><dd>{milliseconds(graph_analysis.unattributed_wall_time)}</dd></div>
|
||||
<div title="任务体墙钟减去显式 cooperative wait；不在任务热路径调用线程计时系统 API。"><dt>独占任务体墙钟</dt><dd>{milliseconds(graph_analysis.exclusive_body_time)}</dd></div>
|
||||
<div title="Topology 墙钟中至少有一个节点任务体正在执行的区间并集。"><dt>任务体墙钟并集</dt><dd>{milliseconds(graph_analysis.body_wall_time)}</dd></div>
|
||||
<div title="Topology 墙钟中只有 Observer on_entry/on_exit 在执行、没有任务体执行的区间。"><dt>Observer 独占墙钟</dt><dd>{milliseconds(graph_analysis.observer_wall_time)}</dd></div>
|
||||
<div title="所有节点 Observer entry 墙钟耗时之和，包含并行重叠。"><dt>Observer entry 墙钟</dt><dd>{milliseconds(graph_analysis.observer_entry_time)}</dd></div>
|
||||
<div title="所有节点 Observer exit 和按帧追踪写入墙钟耗时之和，包含并行重叠。"><dt>Observer exit 墙钟</dt><dd>{milliseconds(graph_analysis.observer_exit_time)}</dd></div>
|
||||
<div title="Observer entry 和 exit 真正占用的 worker CPU 时间之和。"><dt>Observer 实际 CPU</dt><dd>{milliseconds(graph_analysis.observer_cpu_time)}</dd></div>
|
||||
<div title="Topology 墙钟中没有捕获节点任务体或 Observer 活动的守恒余量。"><dt>无任务墙钟</dt><dd>{milliseconds(graph_analysis.idle_wall_time)}</dd></div>
|
||||
<div title="各节点从前驱完成到 on_entry 的估算等待之和，用于定位 Executor 调度长尾。"><dt>累计节点排队</dt><dd>{milliseconds(graph_analysis.queue_time)}</dd></div>
|
||||
<div title="Topology 提交后，首个节点真正进入 Worker 前的等待。"><dt>首次准入等待</dt><dd>{milliseconds(graph_analysis.initial_wait)}</dd></div>
|
||||
@@ -2451,7 +2414,7 @@ function Taskflow_Frame_Pane({plot, components}: {plot: Plot; components: Compon
|
||||
<div title="Observer 时间区间内同时执行的最大节点数量。"><dt>实际最大并行</dt><dd>{graph_analysis.maximum_parallelism}</dd></div>
|
||||
<div title="同一层节点没有相互依赖，可以并行；串行图的并行层数量为零。"><dt>依赖层 / 并行层</dt><dd>{graph_analysis.layer_count} / {graph_analysis.parallel_layer_count}</dd></div>
|
||||
<div title={graph_analysis.longest?.node_id}><dt>最长执行节点</dt><dd>{graph_analysis.longest ? milliseconds(graph_analysis.longest.duration_ms) : "--"}</dd></div>
|
||||
<div title={graph_analysis.longest_unattributed?.node_id}><dt>最长未归因墙时节点</dt><dd>{graph_analysis.longest_unattributed ? `${taskflow_node_name(graph_analysis.longest_unattributed.node_id.split("/").at(-1) ?? graph_analysis.longest_unattributed.node_id)} · ${graph_analysis.longest_unattributed.cpu_time_coarse ? "≈ " : ""}${milliseconds(Math.max(0, graph_analysis.longest_unattributed.duration_ms - graph_analysis.longest_unattributed.cooperative_wait_ms - graph_analysis.longest_unattributed.cpu_duration_ms))}` : "--"}</dd></div>
|
||||
<div title={graph_analysis.longest_exclusive?.node_id}><dt>最长独占任务体节点</dt><dd>{graph_analysis.longest_exclusive ? `${taskflow_node_name(graph_analysis.longest_exclusive.node_id.split("/").at(-1) ?? graph_analysis.longest_exclusive.node_id)} · ${milliseconds(Math.max(0, graph_analysis.longest_exclusive.duration_ms - graph_analysis.longest_exclusive.cooperative_wait_ms))}` : "--"}</dd></div>
|
||||
<div><dt>状态</dt><dd>{graph.completed ? "完成" : "未完成"}</dd></div>
|
||||
</dl><Taskflow_Dag graph={graph} executions={frame.executions} frame={frame} components={components} gallery_state={gallery_state}
|
||||
fullscreen={fullscreen_view === "dag"} on_fullscreen_change={value => set_fullscreen_view(value ? "dag" : null)}
|
||||
|
||||
Reference in New Issue
Block a user