Add v210 upload worker probe

AppConfig.cpp

@@ -577,6 +577,10 @@ bool applyConfigFile(
     {
         config.paceUploads = boolValue;
     }
+    if (jsonBool(objectField(*root, "v210UploadWorker"), boolValue))
+    {
+        config.v210UploadWorker = boolValue;
+    }
 
     if (const std::string* present =
             jsonString(objectField(*root, "present")))
@@ -809,6 +813,7 @@ void printUsage(const char* executableName)
         << "  --perf                Log per-frame CPU timing breakdown\n"
         << "  --verbose             Enable detailed startup and SDK logs\n"
         << "  --pace-uploads        Pace v210 uploads by feed frame rate\n"
+        << "  --v210-upload-worker  Copy direct v210 payloads on a worker thread\n"
         << "  --config <path>       Load JSON feed configuration\n"
         << "  --grid <cols>x<rows>  Set multiview grid (default 2x2, max 16 feeds)\n"
         << "  --fps-cap <fps>       Limit render loop FPS (default 60)\n"
@@ -955,6 +960,12 @@ ConfigParseResult parseAppConfig(
             continue;
         }
 
+        if (arg == "--v210-upload-worker")
+        {
+            result.config.v210UploadWorker = true;
+            continue;
+        }
+
         if (arg == "--v210-staging-memory")
         {
             if (i + 1 >= argc)

AppConfig.hpp

@@ -50,6 +50,7 @@ struct AppConfig
     bool logPerf = false;
     bool verbose = false;
     bool paceUploads = false;
+    bool v210UploadWorker = false;
     uint32_t fpsCap = 60;
     uint32_t maxV210UploadsPerFrame = 0;
     uint32_t gridCols = DefaultGridCols;

main.cpp

@@ -18,7 +18,9 @@
 
 #include <algorithm>
 #include <atomic>
+#include <condition_variable>
 #include <cstring>
+#include <deque>
 #include <csignal>
 #include <filesystem>
 #include <fstream>
@@ -26,7 +28,9 @@
 #include <iostream>
 #include <map>
 #include <memory>
+#include <mutex>
 #include <sstream>
+#include <thread>
 #include <vector>
 
 #if defined(__linux__)
@@ -49,6 +53,8 @@ struct PerfStats
     uint64_t stageCopyTicks = 0;
     uint64_t directV210SourceReadNs = 0;
     uint64_t directV210PayloadCopyNs = 0;
+    uint64_t workerV210SourceReadNs = 0;
+    uint64_t workerV210PayloadCopyNs = 0;
     uint64_t uploadRecordTicks = 0;
     uint64_t submitPresentTicks = 0;
     uint64_t idleDelayTicks = 0;
@@ -69,6 +75,8 @@ struct FramePerfStats
     uint64_t stageCopyTicks = 0;
     uint64_t directV210SourceReadNs = 0;
     uint64_t directV210PayloadCopyNs = 0;
+    uint64_t workerV210SourceReadNs = 0;
+    uint64_t workerV210PayloadCopyNs = 0;
     uint64_t maxStageCopyTicks = 0;
     uint64_t maxStageCopyBytes = 0;
     uint32_t maxStageCopyFeed = 0;
@@ -95,6 +103,129 @@ struct FramePerfStats
     uint32_t maxProcessEventType = 0;
 };
 
+struct V210UploadJob
+{
+    uint32_t feedIndex = 0;
+    uint32_t frameCounter = 0;
+    IVideoFeed* feed = nullptr;
+    void* destination = nullptr;
+    size_t destinationSize = 0;
+};
+
+struct V210UploadResult
+{
+    uint32_t feedIndex = 0;
+    bool copied = false;
+    uint64_t version = 0;
+    V210ReadTiming timing{};
+};
+
+class V210UploadWorker
+{
+public:
+    V210UploadWorker()
+        : mThread(&V210UploadWorker::run, this)
+    {
+    }
+
+    ~V210UploadWorker()
+    {
+        stop();
+    }
+
+    V210UploadWorker(const V210UploadWorker&) = delete;
+    V210UploadWorker& operator=(const V210UploadWorker&) = delete;
+
+    void enqueue(V210UploadJob job)
+    {
+        {
+            std::lock_guard<std::mutex> lock(mMutex);
+            mJobs.push_back(job);
+        }
+        mCondition.notify_one();
+    }
+
+    void drainResults(std::vector<V210UploadResult>& results)
+    {
+        std::lock_guard<std::mutex> lock(mMutex);
+        while (!mResults.empty())
+        {
+            results.push_back(mResults.front());
+            mResults.pop_front();
+        }
+    }
+
+    void stop()
+    {
+        {
+            std::lock_guard<std::mutex> lock(mMutex);
+            if (mStopping)
+            {
+                return;
+            }
+            mStopping = true;
+        }
+        mCondition.notify_one();
+        if (mThread.joinable())
+        {
+            mThread.join();
+        }
+    }
+
+private:
+    void run()
+    {
+        while (true)
+        {
+            V210UploadJob job{};
+            {
+                std::unique_lock<std::mutex> lock(mMutex);
+                mCondition.wait(
+                    lock,
+                    [&]
+                    {
+                        return mStopping || !mJobs.empty();
+                    });
+
+                if (mStopping && mJobs.empty())
+                {
+                    return;
+                }
+
+                job = mJobs.front();
+                mJobs.pop_front();
+            }
+
+            V210UploadResult result{};
+            result.feedIndex = job.feedIndex;
+            if (job.feed != nullptr && job.destination != nullptr)
+            {
+                result.copied = job.feed->readV210FrameInto(
+                    job.frameCounter,
+                    job.destination,
+                    job.destinationSize,
+                    &result.timing);
+                if (result.copied)
+                {
+                    result.version = job.feed->frameVersion();
+                }
+            }
+
+            {
+                std::lock_guard<std::mutex> lock(mMutex);
+                mResults.push_back(result);
+            }
+        }
+    }
+
+    std::mutex mMutex;
+    std::condition_variable mCondition;
+    std::deque<V210UploadJob> mJobs;
+    std::deque<V210UploadResult> mResults;
+    bool mStopping = false;
+    std::thread mThread;
+};
+
 struct FeedPerfStats
 {
     uint64_t updates = 0;
@@ -842,8 +973,16 @@ int main(int argc, char* argv[])
     std::vector<bool> pendingUploadNeeded(feedCount);
     std::vector<bool> postFenceDirectV210Read(feedCount);
     std::vector<double> feedUploadCredits(feedCount, 1.0);
+    std::vector<bool> v210WorkerInFlight(feedCount);
+    std::unique_ptr<V210UploadWorker> v210UploadWorker;
+    std::vector<V210UploadResult> v210WorkerResults;
     uint64_t lastUploadPaceTicks = SDL_GetPerformanceCounter();
 
+    if (config.v210UploadWorker)
+    {
+        v210UploadWorker = std::make_unique<V210UploadWorker>();
+    }
+
     std::fill(
         uploadedVersions.begin(),
         uploadedVersions.end(),
@@ -1432,6 +1571,40 @@ int main(int argc, char* argv[])
             }
         };
 
+        auto recordFeedUploadStats =
+            [&](uint32_t feedIndex)
+        {
+            if (!config.logPerf)
+            {
+                return;
+            }
+
+            ++feedPerfStats[feedIndex].uploads;
+
+            const uint64_t uploadTicks =
+                SDL_GetPerformanceCounter();
+            if (feedPerfStats[feedIndex].lastUploadTicks != 0)
+            {
+                const uint64_t interval =
+                    uploadTicks -
+                    feedPerfStats[feedIndex].lastUploadTicks;
+                feedPerfStats[feedIndex].uploadIntervalTicks +=
+                    interval;
+                feedPerfStats[feedIndex].minUploadIntervalTicks =
+                    std::min(
+                        feedPerfStats[feedIndex].minUploadIntervalTicks,
+                        interval
+                    );
+                feedPerfStats[feedIndex].maxUploadIntervalTicks =
+                    std::max(
+                        feedPerfStats[feedIndex].maxUploadIntervalTicks,
+                        interval
+                    );
+                ++feedPerfStats[feedIndex].uploadIntervals;
+            }
+            feedPerfStats[feedIndex].lastUploadTicks = uploadTicks;
+        };
+
         for (uint32_t i = 0; i < feedCount; ++i)
         {
             if (v210FeedReady[i] &&
@@ -1500,6 +1673,54 @@ int main(int argc, char* argv[])
             framePerf.submitPresentTicks += elapsed;
         }
 
+        if (config.v210UploadWorker)
+        {
+            v210WorkerResults.clear();
+            v210UploadWorker->drainResults(v210WorkerResults);
+
+            for (const V210UploadResult& result : v210WorkerResults)
+            {
+                const uint32_t i = result.feedIndex;
+                if (i >= feedCount)
+                {
+                    continue;
+                }
+
+                v210WorkerInFlight[i] = false;
+
+                if (!result.copied)
+                {
+                    continue;
+                }
+
+                const bool uploadNeeded =
+                    uploadedVersions[i] != result.version;
+                recordFeedVersionStats(i, uploadNeeded);
+
+                if (!uploadNeeded)
+                {
+                    continue;
+                }
+
+                pendingVersions[i] = result.version;
+                uploadedVersions[i] = result.version;
+                feedUploadNeeded[i] = true;
+                recordFeedUploadStats(i);
+
+                if (config.logPerf)
+                {
+                    perfStats.workerV210SourceReadNs +=
+                        result.timing.sourceReadNs;
+                    perfStats.workerV210PayloadCopyNs +=
+                        result.timing.payloadCopyNs;
+                    framePerf.workerV210SourceReadNs +=
+                        result.timing.sourceReadNs;
+                    framePerf.workerV210PayloadCopyNs +=
+                        result.timing.payloadCopyNs;
+                }
+            }
+        }
+
         ++frameCounter;
 
         if (config.logFps)
@@ -1615,6 +1836,42 @@ int main(int argc, char* argv[])
 
                 if (directV210Read)
                 {
+                    if (config.v210UploadWorker)
+                    {
+                        if (!v210WorkerInFlight[i])
+                        {
+                            const size_t v210Bytes =
+                                static_cast<size_t>(
+                                    feeds[i]->v210Stride()
+                                ) *
+                                static_cast<size_t>(
+                                    feeds[i]->v210Height()
+                                );
+
+                            v210UploadWorker->enqueue(
+                                V210UploadJob{
+                                    i,
+                                    frameCounter,
+                                    feeds[i].get(),
+                                    v210Feeds[i].v210MappedData,
+                                    v210Bytes
+                                });
+                            v210WorkerInFlight[i] = true;
+                            ++v210UploadsThisFrame;
+
+                            if (config.paceUploads &&
+                                feeds[i]->hasFrameRate())
+                            {
+                                feedUploadCredits[i] =
+                                    std::max(
+                                        0.0,
+                                        feedUploadCredits[i] - 1.0
+                                    );
+                            }
+                        }
+                        continue;
+                    }
+
                     V210ReadTiming directReadTiming{};
                     const size_t v210Bytes =
                         static_cast<size_t>(
@@ -1720,30 +1977,7 @@ int main(int argc, char* argv[])
             }
             if (config.logPerf)
             {
-                ++feedPerfStats[i].uploads;
+                recordFeedUploadStats(i);
-
-                const uint64_t uploadTicks =
-                    SDL_GetPerformanceCounter();
-                if (feedPerfStats[i].lastUploadTicks != 0)
-                {
-                    const uint64_t interval =
-                        uploadTicks -
-                        feedPerfStats[i].lastUploadTicks;
-                    feedPerfStats[i].uploadIntervalTicks +=
-                        interval;
-                    feedPerfStats[i].minUploadIntervalTicks =
-                        std::min(
-                            feedPerfStats[i].minUploadIntervalTicks,
-                            interval
-                        );
-                    feedPerfStats[i].maxUploadIntervalTicks =
-                        std::max(
-                            feedPerfStats[i].maxUploadIntervalTicks,
-                            interval
-                        );
-                    ++feedPerfStats[i].uploadIntervals;
-                }
-                feedPerfStats[i].lastUploadTicks = uploadTicks;
             }
             feedUploadNeeded[i] = true;
         }
@@ -2144,6 +2378,14 @@ int main(int argc, char* argv[])
                     << (static_cast<double>(
                             framePerf.directV210PayloadCopyNs) /
                         1000000.0)
+                    << "ms v210WorkerRead="
+                    << (static_cast<double>(
+                            framePerf.workerV210SourceReadNs) /
+                        1000000.0)
+                    << "ms v210WorkerCopy="
+                    << (static_cast<double>(
+                            framePerf.workerV210PayloadCopyNs) /
+                        1000000.0)
                     << "ms stageCopies="
                     << framePerf.stageCopies
                     << " v210Copies="
@@ -2203,6 +2445,14 @@ int main(int argc, char* argv[])
                     << (static_cast<double>(
                             perfStats.directV210PayloadCopyNs) /
                         1000000.0 / frames)
+                    << "ms v210WorkerRead="
+                    << (static_cast<double>(
+                            perfStats.workerV210SourceReadNs) /
+                        1000000.0 / frames)
+                    << "ms v210WorkerCopy="
+                    << (static_cast<double>(
+                            perfStats.workerV210PayloadCopyNs) /
+                        1000000.0 / frames)
                     << "ms uploadRecord="
                     << (perfStats.uploadRecordTicks * invMs / frames)
                     << "ms submitPresent="
@@ -2301,6 +2551,11 @@ int main(int argc, char* argv[])
     // Cleanup
     // ----------------------------------------
 
+    if (v210UploadWorker)
+    {
+        v210UploadWorker->stop();
+    }
+
     vkDeviceWaitIdle(ctx.device());
 
     ImGui_ImplVulkan_Shutdown();