Refactor out snapshot logic. Add basic infra for async frame conversions.

Author: Mysticial

Common/Cpp/Concurrency/ParallelTaskRunner.cpp

@@ -54,7 +54,7 @@ void ParallelTaskRunner::wait_for_everything(){
     });
 }
 
-std::shared_ptr<AsyncTask> ParallelTaskRunner::dispatch(std::function<void()>&& func){
+std::shared_ptr<AsyncTask> ParallelTaskRunner::blocking_dispatch(std::function<void()>&& func){
     std::shared_ptr<AsyncTask> task(new AsyncTask(std::move(func)));
 
     std::unique_lock<std::mutex> lg(m_lock);
@@ -74,6 +74,26 @@ std::shared_ptr<AsyncTask> ParallelTaskRunner::dispatch(std::function<void()>&&
 
     return task;
 }
+std::shared_ptr<AsyncTask> ParallelTaskRunner::try_dispatch(std::function<void()>& func){
+    std::lock_guard<std::mutex> lg(m_lock);
+
+    if (m_queue.size() + m_busy_count >= m_max_threads){
+        return nullptr;
+    }
+
+    std::shared_ptr<AsyncTask> task(new AsyncTask(std::move(func)));
+
+    //  Enqueue task.
+    m_queue.emplace_back(task);
+
+    if (m_queue.size() + m_busy_count > m_threads.size()){
+        m_threads.emplace_back(run_with_catch, "ParallelTaskRunner::thread_loop()", [this]{ thread_loop(); });
+    }
+
+    m_thread_cv.notify_one();
+
+    return task;
+}
 
 
 void ParallelTaskRunner::run_in_parallel(
@@ -89,7 +109,7 @@ void ParallelTaskRunner::run_in_parallel(
 
     std::vector<std::shared_ptr<AsyncTask>> tasks;
     for (size_t c = 0; c < blocks; c++){
-        tasks.emplace_back(dispatch([=, &func]{
+        tasks.emplace_back(blocking_dispatch([=, &func]{
             size_t s = start + c * block_size;
             size_t e = std::min(s + block_size, end);
 //            cout << "Running: [" << s << "," << e << ")" << endl;
@@ -108,9 +128,6 @@ void ParallelTaskRunner::run_in_parallel(
 
 
 void ParallelTaskRunner::thread_loop(){
-//#if _WIN32
-//    SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_IDLE);
-//#endif
     if (m_new_thread_callback){
         m_new_thread_callback();
     }

Common/Cpp/Concurrency/ParallelTaskRunner.h

@@ -23,7 +23,13 @@ class ParallelTaskRunner{
 
     void wait_for_everything();
 
-    std::shared_ptr<AsyncTask> dispatch(std::function<void()>&& func);
+    //  Dispatch the function. If there are no threads available, it waits until
+    //  there are.
+    std::shared_ptr<AsyncTask> blocking_dispatch(std::function<void()>&& func);
+
+    //  Dispatch the function. Returns null if no threads are available.
+    //  "func" will be moved-from only on success.
+    std::shared_ptr<AsyncTask> try_dispatch(std::function<void()>& func);
 
     void run_in_parallel(
         const std::function<void(size_t index)>& func,

SerialPrograms/CMakeLists.txt

@@ -402,6 +402,8 @@ file(GLOB MAIN_SOURCES
     Source/CommonFramework/Options/ResolutionOption.cpp
     Source/CommonFramework/Options/ResolutionOption.h
     Source/CommonFramework/Options/ScreenshotFormatOption.h
+    Source/CommonFramework/Options/ThreadPoolOption.cpp
+    Source/CommonFramework/Options/ThreadPoolOption.h
     Source/CommonFramework/Panels/PanelDescriptor.cpp
     Source/CommonFramework/Panels/PanelDescriptor.h
     Source/CommonFramework/Panels/PanelInstance.cpp
@@ -449,6 +451,8 @@ file(GLOB MAIN_SOURCES
     Source/CommonFramework/Tools/ErrorDumper.h
     Source/CommonFramework/Tools/FileDownloader.cpp
     Source/CommonFramework/Tools/FileDownloader.h
+    Source/CommonFramework/Tools/GlobalThreadPools.cpp
+    Source/CommonFramework/Tools/GlobalThreadPools.h
     Source/CommonFramework/Tools/ProgramEnvironment.cpp
     Source/CommonFramework/Tools/ProgramEnvironment.h
     Source/CommonFramework/Tools/StatAccumulator.cpp
@@ -464,6 +468,8 @@ file(GLOB MAIN_SOURCES
     Source/CommonFramework/VideoPipeline/Backends/MediaServicesQt6.cpp
     Source/CommonFramework/VideoPipeline/Backends/MediaServicesQt6.h
     Source/CommonFramework/VideoPipeline/Backends/QVideoFrameCache.h
+    Source/CommonFramework/VideoPipeline/Backends/SnapshotManager.cpp
+    Source/CommonFramework/VideoPipeline/Backends/SnapshotManager.h
     Source/CommonFramework/VideoPipeline/Backends/VideoFrameQt.h
     Source/CommonFramework/VideoPipeline/CameraInfo.h
     Source/CommonFramework/VideoPipeline/Stats/CpuUtilizationStats.h

SerialPrograms/Source/CommonFramework/Globals.cpp

@@ -26,7 +26,7 @@ namespace PokemonAutomation{
 const bool IS_BETA_VERSION = true;
 const int PROGRAM_VERSION_MAJOR = 0;
 const int PROGRAM_VERSION_MINOR = 54;
-const int PROGRAM_VERSION_PATCH = 9;
+const int PROGRAM_VERSION_PATCH = 10;
 
 const std::string PROGRAM_VERSION_BASE =
     "v" + std::to_string(PROGRAM_VERSION_MAJOR) +

SerialPrograms/Source/CommonFramework/Options/Environment/PerformanceOptions.h

@@ -9,6 +9,7 @@
 
 #include "Common/Cpp/Options/GroupOption.h"
 #include "Common/Cpp/Options/TimeDurationOption.h"
+#include "CommonFramework/Options/ThreadPoolOption.h"
 #include "ProcessPriorityOption.h"
 #include "ProcessorLevelOption.h"
 
@@ -47,7 +48,7 @@ class PerformanceOptions : public GroupOption{
             DEFAULT_PRIORITY_REALTIME_INFERENCE
         )
         , NORMAL_INFERENCE_PRIORITY(
-            "<b>normal Inference Priority:</b><br>"
+            "<b>Normal Inference Priority:</b><br>"
             "Thread priority of non-realtime inference threads that can be slow "
             "without negatively affecting a program.",
             DEFAULT_PRIORITY_NORMAL_INFERENCE
@@ -57,12 +58,28 @@ class PerformanceOptions : public GroupOption{
             "Thread priority of computation threads.",
             DEFAULT_PRIORITY_COMPUTE
         )
+        , THREAD_POOL_REALTIME_INFERENCE(
+            "Thread Pool: Real-time Inference",
+            DEFAULT_PRIORITY_REALTIME_INFERENCE,
+            0.5
+        )
+        , THREAD_POOL_NORMAL_INFERENCE(
+            "Thread Pool: Normal Inference",
+            DEFAULT_PRIORITY_NORMAL_INFERENCE,
+            1.0
+        )
     {
         PA_ADD_OPTION(PRECISE_WAKE_MARGIN);
+
         PA_ADD_OPTION(REALTIME_THREAD_PRIORITY);
         PA_ADD_OPTION(REALTIME_INFERENCE_PRIORITY);
         PA_ADD_OPTION(NORMAL_INFERENCE_PRIORITY);
         PA_ADD_OPTION(COMPUTE_PRIORITY);
+
+        //  TODO: REMOVE: Enable when these are actually used.
+//        PA_ADD_OPTION(THREAD_POOL_REALTIME_INFERENCE);
+//        PA_ADD_OPTION(THREAD_POOL_NORMAL_INFERENCE);
+
         PA_ADD_OPTION(PROCESSOR_LEVEL);
     }
 
@@ -74,6 +91,9 @@ class PerformanceOptions : public GroupOption{
     ThreadPriorityOption NORMAL_INFERENCE_PRIORITY;
     ThreadPriorityOption COMPUTE_PRIORITY;
 
+    ThreadPoolOption THREAD_POOL_REALTIME_INFERENCE;
+    ThreadPoolOption THREAD_POOL_NORMAL_INFERENCE;
+
     ProcessorLevelOption PROCESSOR_LEVEL;
 };
 

SerialPrograms/Source/CommonFramework/Options/Environment/ProcessPriorityOption.h

@@ -8,7 +8,7 @@
 #define PokemonAutomation_ProcessPriorityOption_H
 
 #include "Common/Cpp/Options/EnumDropdownOption.h"
-#include "Common/Cpp/Options/GroupOption.h"
+//#include "Common/Cpp/Options/GroupOption.h"
 #include "CommonFramework/Environment/Environment.h"
 
 namespace PokemonAutomation{

SerialPrograms/Source/CommonFramework/Options/ThreadPoolOption.cpp

@@ -0,0 +1,54 @@
+/*  Thread Pool Option
+ *
+ *  From: https://github.com/PokemonAutomation/
+ *
+ */
+
+#include <thread>
+#include "ThreadPoolOption.h"
+
+namespace PokemonAutomation{
+
+
+ThreadPoolOption::ThreadPoolOption(
+    std::string label,
+    ThreadPriority default_priority,
+    double default_max_thread_ratio
+)
+    : GroupOption(std::move(label), LockMode::UNLOCK_WHILE_RUNNING)
+    , m_default_max_threads(
+        std::max(
+            (size_t)(std::thread::hardware_concurrency() * default_max_thread_ratio),
+            (size_t)1
+        )
+    )
+    , HARDWARE_THREADS(
+        "<b>Hardware Threads:</b>",
+        LockMode::UNLOCK_WHILE_RUNNING,
+        std::thread::hardware_concurrency()
+    )
+    , PRIORITY("<b>Thread Priority:</b>", default_priority)
+    , MAX_THREADS(
+        "<b>Maximum Threads:</b>",
+        LockMode::UNLOCK_WHILE_RUNNING,
+        m_default_max_threads
+    )
+{
+    PA_ADD_OPTION(HARDWARE_THREADS);
+    PA_ADD_OPTION(PRIORITY);
+    PA_ADD_OPTION(MAX_THREADS);
+    HARDWARE_THREADS.set_visibility(ConfigOptionState::HIDDEN);
+}
+
+void ThreadPoolOption::load_json(const JsonValue& json){
+    GroupOption::load_json(json);
+
+    //  Reset the max threads if the hardware threads has changed.
+    if (HARDWARE_THREADS != std::thread::hardware_concurrency()){
+        MAX_THREADS.set(m_default_max_threads);
+    }
+}
+
+
+
+}

SerialPrograms/Source/CommonFramework/Options/ThreadPoolOption.h

@@ -0,0 +1,39 @@
+/*  Thread Pool Option
+ *
+ *  From: https://github.com/PokemonAutomation/
+ *
+ */
+
+#ifndef PokemonAutomation_Options_ThreadPoolOption_H
+#define PokemonAutomation_Options_ThreadPoolOption_H
+
+#include "Common/Cpp/Options/SimpleIntegerOption.h"
+#include "Common/Cpp/Options/GroupOption.h"
+#include "Environment/ProcessPriorityOption.h"
+
+namespace PokemonAutomation{
+
+
+class ThreadPoolOption : public GroupOption{
+public:
+    ThreadPoolOption(
+        std::string label,
+        ThreadPriority default_priority,
+        double default_max_thread_ratio = 1.0
+    );
+
+    virtual void load_json(const JsonValue& json) override;
+
+private:
+    const size_t m_default_max_threads;
+    SimpleIntegerOption<size_t> HARDWARE_THREADS;
+
+public:
+    ThreadPriorityOption PRIORITY;
+    SimpleIntegerOption<size_t> MAX_THREADS;
+};
+
+
+
+}
+#endif

SerialPrograms/Source/CommonFramework/Tools/GlobalThreadPools.cpp

@@ -0,0 +1,30 @@
+/*  Global Thread Pools
+ *
+ *  From: https://github.com/PokemonAutomation/
+ *
+ */
+
+#include "CommonFramework/GlobalSettingsPanel.h"
+#include "CommonFramework/Options/Environment/PerformanceOptions.h"
+#include "GlobalThreadPools.h"
+
+namespace PokemonAutomation{
+namespace GlobalThreadPools{
+
+
+
+ParallelTaskRunner& realtime_inference(){
+    static ParallelTaskRunner runner(
+        [](){
+            GlobalSettings::instance().PERFORMANCE->THREAD_POOL_REALTIME_INFERENCE.PRIORITY.set_on_this_thread();
+        },
+        0, GlobalSettings::instance().PERFORMANCE->THREAD_POOL_REALTIME_INFERENCE.MAX_THREADS
+    );
+    return runner;
+}
+
+
+
+
+}
+}

SerialPrograms/Source/CommonFramework/Tools/GlobalThreadPools.h

@@ -0,0 +1,23 @@
+/*  Global Thread Pools
+ *
+ *  From: https://github.com/PokemonAutomation/
+ *
+ */
+
+#ifndef PokemonAutomation_CommonTools_GlobalThreadPools_H
+#define PokemonAutomation_CommonTools_GlobalThreadPools_H
+
+#include "Common/Cpp/Concurrency/ParallelTaskRunner.h"
+
+namespace PokemonAutomation{
+namespace GlobalThreadPools{
+
+
+
+ParallelTaskRunner& realtime_inference();
+
+
+
+}
+}
+#endif