Merge pull request #15108 from Filip722/perf/trace-close-output-20261102

perf(trace): format closed captures on the output worker

core/libs/prx/libSceAgcDriver/Execution/include/PerformanceTrace.hpp

@@ -442,6 +442,41 @@ struct TraceClockSample {
     std::chrono::steady_clock::time_point cpu{};
 };
 
+class TraceCapture {
+public:
+    TraceCapture() = default;
+    TraceCapture(const TraceCapture&) = delete;
+    TraceCapture& operator=(const TraceCapture&) = delete;
+
+    void Add(TraceThread::Snapshot snapshot) {
+        slices.emplace_back(snapshot.slices.begin(), snapshot.slices.end());
+        for (auto& slice : slices.back()) {
+            slice.scope = own(slice.scope);
+            slice.stage = own(slice.stage);
+        }
+        for (auto& total : snapshot.totals) {
+            total.scope = own(total.scope);
+            total.stage = own(total.stage);
+        }
+        snapshot.slices = slices.back();
+        threads.push_back(std::move(snapshot));
+    }
+
+    TraceSummary summary;
+    std::vector<TraceThread::Clock::time_point> presents;
+    std::vector<TraceThread::Snapshot> threads;
+    std::uint64_t process = 0;
+    std::size_t presenter = 0;
+
+private:
+    const char* own(const char* text) {
+        return text != nullptr ? labels.try_emplace(text, text).first->second.c_str() : nullptr;
+    }
+
+    std::vector<std::vector<TraceThread::Slice>> slices;
+    std::map<const char*, std::string> labels;
+};
+
 class TraceFormat {
 public:
     using Clock = std::chrono::steady_clock;

core/libs/prx/libSceAgcDriver/Execution/src/PerformanceTrace.cpp

@@ -398,12 +398,21 @@ void PerformanceTrace::Present(std::uint32_t width, std::uint32_t height, const
     summary.firstPresent = state.firstPresent;
     summary.gpuBusyMs = VulkanDevice::PresentCounts().gpuBusyMs - state.gpuBusyAtOpen;
     summary.trace = "aps5-trace-" + summary.commit + "-" + std::to_string(process) + ".json.gz";
-    auto events = TraceFormat::Events(process, state.presents.front(), snapshots, presenter, state.presents, state.firstPresent);
+    auto capture = std::make_shared<TraceCapture>();
+    capture->summary = std::move(summary);
+    capture->presents = state.presents;
+    capture->process = process;
+    capture->presenter = presenter;
+    for (auto& snapshot : snapshots) capture->Add(std::move(snapshot));
+    const auto name = capture->summary.trace;
     auto& output = writer();
-    output = std::make_unique<ProfileOutput>([directory = traceDirectory(), name = summary.trace, line = TraceFormat::Summary(summary, snapshots, state.presents)](std::string_view text) {
+    output = std::make_unique<ProfileOutput>([directory = traceDirectory(), name, capture = std::move(capture)](std::string_view) mutable {
+        auto closed = std::move(capture);
         try {
+            const auto events = TraceFormat::Events(closed->process, closed->presents.front(), closed->threads, closed->presenter, closed->presents, closed->summary.firstPresent);
+            const auto line = TraceFormat::Summary(closed->summary, closed->threads, closed->presents);
             std::filesystem::create_directories(directory);
-            TraceFile::WriteGzip(directory / name, text);
+            TraceFile::WriteGzip(directory / name, events);
             TraceFile::AppendLine(directory / "aps5-summary.jsonl", line);
             std::fprintf(stderr, "[trace] wrote %s and its line in aps5-summary.jsonl\n", name.c_str());
             std::fflush(stderr);
@@ -412,9 +421,9 @@ void PerformanceTrace::Present(std::uint32_t width, std::uint32_t height, const
             std::fflush(stderr);
             throw;
         }
-    }, std::numeric_limits<std::size_t>::max());
-    ProfilePrint_nid_no_patch("[trace] closed at present %llu, writing %s (%zu bytes of JSON)\n", static_cast<unsigned long long>(present), summary.trace.c_str(), events.size());
-    output->Write(std::move(events));
+    });
+    ProfilePrint_nid_no_patch("[trace] closed at present %llu, formatting %s off-thread\n", static_cast<unsigned long long>(present), name.c_str());
+    if (!output->Write(name)) throw std::runtime_error("trace output rejected the closed capture");
 }
 
 void PerformanceTrace::StopOutput() {

core/libs/prx/libSceAgcDriver/tests/PerformanceTrace.cpp

@@ -1,5 +1,6 @@
 #include "prx/libSceAgcDriver/Execution/include/PerformanceTrace.hpp"
 #include "prx/libSceAgcDriver/Execution/include/TraceFile.hpp"
+#include "prx/libSceAgcDriver/Execution/include/ProfileOutput.hpp"
 #include "zlib.h"
 #include <atomic>
 #include <condition_variable>
@@ -550,6 +551,66 @@ void BenchSwitch() {
     }
 }
 
+void CaptureOwnsSnapshots() {
+    auto capture = std::make_shared<AgcDriver::TraceCapture>();
+    const auto origin = Clock::now();
+    capture->process = 7;
+    capture->summary.commit = "capture";
+    capture->summary.trace = "aps5-trace-capture-7.json.gz";
+    capture->summary.firstPresent = 5;
+    capture->presents = {origin, origin + milliseconds(16)};
+    std::string expectedEvents, expectedSummary;
+    {
+        std::string scope = "Driver.Packet", stage = "draw", gpuName = "batch";
+        std::atomic<std::uint64_t> presents{1};
+        TraceThread thread("queue worker 0x0", presents, 2);
+        thread.Add(scope.c_str(), stage.c_str(), origin + milliseconds(2), origin + milliseconds(2) + microseconds(30));
+        thread.Add(scope.c_str(), nullptr, origin + milliseconds(1), origin + milliseconds(1) + microseconds(40));
+        thread.Add(scope.c_str(), nullptr, origin + milliseconds(3), origin + milliseconds(3) + microseconds(30));
+        auto snapshot = thread.Collect();
+        std::vector<TraceThread::Slice> gpu{{gpuName.c_str(), nullptr, origin + microseconds(5), microseconds(10), 42}};
+        std::vector<TraceThread::Snapshot> snapshots{snapshot, {"gpu", gpu, {}, false}};
+        expectedEvents = TraceFormat::Events(7, origin, snapshots, 0, capture->presents, 5);
+        expectedSummary = TraceFormat::Summary(capture->summary, snapshots, capture->presents);
+        for (auto& value : snapshots) capture->Add(std::move(value));
+        thread.Add(scope.c_str(), nullptr, origin + milliseconds(4), origin + milliseconds(4) + microseconds(30));
+        gpu.clear();
+        scope.assign("changed scope");
+        stage.assign("changed stage");
+        gpuName.assign("changed gpu");
+    }
+    Check(TraceFormat::Events(capture->process, capture->presents.front(), capture->threads, capture->presenter, capture->presents, capture->summary.firstPresent) == expectedEvents, "closed capture borrowed slices or labels, changed ordering or included a late append");
+    Check(TraceFormat::Summary(capture->summary, capture->threads, capture->presents) == expectedSummary, "closed capture changed totals or loss cutoff");
+}
+
+void CaptureWriterDrainAndFailure() {
+    for (const bool fail : {false, true}) {
+        auto capture = std::make_shared<AgcDriver::TraceCapture>();
+        capture->process = 7;
+        capture->summary.trace = "aps5-trace-capture-7.json.gz";
+        const auto origin = Clock::now();
+        capture->presents = {origin, origin + milliseconds(16)};
+        std::weak_ptr<AgcDriver::TraceCapture> released = capture;
+        unsigned calls = 0;
+        std::string events, line;
+        AgcDriver::ProfileOutput output([capture = std::move(capture), &calls, &events, &line, fail](std::string_view) mutable {
+            auto closed = std::move(capture);
+            ++calls;
+            const auto text = TraceFormat::Events(closed->process, closed->presents.front(), closed->threads, closed->presenter, closed->presents, closed->summary.firstPresent);
+            if (fail) throw std::runtime_error("closed capture write failed");
+            events = text;
+            line = TraceFormat::Summary(closed->summary, closed->threads, closed->presents);
+        });
+        Check(output.Write("aps5-trace-capture-7.json.gz"), "closed capture was not queued");
+        output.Stop();
+        bool failed = false;
+        try { output.Flush(); }
+        catch (const std::runtime_error& error) { failed = std::string_view(error.what()) == "closed capture write failed"; }
+        Check(failed == fail && calls == 1 && released.expired(), "closed capture stop lost or repeated its job, hid a failure or retained its payload");
+        Check(fail ? events.empty() && line.empty() : !events.empty() && line.find("aps5-trace-capture-7.json.gz") != std::string::npos, "closed capture events and summary lost their pairing");
+    }
+}
+
 void GzipFile() {
     std::string text;
     for (int index = 0; index < 20000; ++index) text += "{\"name\":\"Driver.Packet\",\"ph\":\"X\",\"ts\":" + std::to_string(index) + "}\n";
@@ -625,6 +686,8 @@ int main() {
         BenchWriterKeepsNewest();
         BenchWriterStopWrites();
         BenchSwitch();
+        CaptureOwnsSnapshots();
+        CaptureWriterDrainAndFailure();
         GzipFile();
         ReplaceGzipFile();
         AppendLines();