Move QVideoFrame -> QImage conversion off the video pivot thread and allow parallelization.

Author: Mysticial

Common/Cpp/Concurrency/AsyncDispatcher.h

@@ -49,7 +49,7 @@ class AsyncTask{
 private:
     friend class FireForgetDispatcher;
     friend class AsyncDispatcher;
-    friend class ParallelTaskRunner;
+    friend class ParallelTaskRunnerCore;
 
     std::function<void()> m_task;
     bool m_finished;

Common/Cpp/Concurrency/ParallelTaskRunner.cpp

@@ -8,6 +8,8 @@
 #include <Windows.h>
 #endif
 #include "Common/Cpp/PanicDump.h"
+#include "Common/Cpp/Containers/Pimpl.tpp"
+#include "Common/Cpp/CpuUtilization/CpuUtilization.h"
 #include "ParallelTaskRunner.h"
 
 //#include <iostream>
@@ -17,44 +19,201 @@
 namespace PokemonAutomation{
 
 
+class StopWatch{
+public:
+    StopWatch()
+        : m_total(WallDuration::zero())
+        , m_last_start(WallClock::max())
+    {}
+
+    void start(){
+//        cout << "start" << endl;
+        if (m_last_start != WallClock::max()){
+            return;
+        }
+        m_last_start = current_time();
+    }
+    void stop(){
+//        cout << "stop" << endl;
+        if (m_last_start == WallClock::max()){
+            return;
+        }
+        m_total += current_time() - m_last_start;
+        m_last_start = WallClock::max();
+    }
+    WallDuration total() const{
+        WallDuration ret = m_total;
+        if (m_last_start != WallClock::max()){
+            ret += current_time() - m_last_start;
+        }
+//        cout << "total: " << std::chrono::duration_cast<Milliseconds>(ret).count() << endl;
+        return ret;
+    }
+
+private:
+    WallDuration m_total;
+    WallClock m_last_start;
+};
+
+
+
+
+
+class ParallelTaskRunnerCore final{
+public:
+    ParallelTaskRunnerCore(
+        std::function<void()>&& new_thread_callback,
+        size_t starting_threads,
+        size_t max_threads
+    );
+    ~ParallelTaskRunnerCore();
+
+    size_t max_threads() const{
+        return m_max_threads;
+    }
+    WallDuration cpu_time() const;
+
+    void wait_for_everything();
+
+    //  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,
+        size_t start, size_t end,
+        size_t block_size = 0
+    );
+
+
+private:
+    struct ThreadData{
+        std::thread thread;
+        ThreadHandle handle;
+        StopWatch runtime;
+    };
+
+    void spawn_thread();
+    void thread_loop(ThreadData& data);
+
+
+private:
+    struct Data;
+
+    std::function<void()> m_new_thread_callback;
+    size_t m_max_threads;
+    std::deque<std::shared_ptr<AsyncTask>> m_queue;
+
+    std::deque<ThreadData> m_threads;
+
+    bool m_stopping;
+    size_t m_busy_count;
+    mutable std::mutex m_lock;
+    std::condition_variable m_thread_cv;
+    std::condition_variable m_dispatch_cv;
+};
+
+
+
+
+
+
 
 ParallelTaskRunner::ParallelTaskRunner(
     std::function<void()>&& new_thread_callback,
     size_t starting_threads,
     size_t max_threads
+)
+    : m_core(
+        CONSTRUCT_TOKEN,
+        std::move(new_thread_callback),
+        starting_threads,
+        max_threads
+    )
+{}
+ParallelTaskRunner::~ParallelTaskRunner() = default;
+size_t ParallelTaskRunner::max_threads() const{
+    return m_core->max_threads();
+}
+WallDuration ParallelTaskRunner::cpu_time() const{
+    return m_core->cpu_time();
+}
+void ParallelTaskRunner::wait_for_everything(){
+    m_core->wait_for_everything();
+}
+std::shared_ptr<AsyncTask> ParallelTaskRunner::blocking_dispatch(std::function<void()>&& func){
+    return m_core->blocking_dispatch(std::move(func));
+}
+std::shared_ptr<AsyncTask> ParallelTaskRunner::try_dispatch(std::function<void()>& func){
+    return m_core->try_dispatch(func);
+}
+void ParallelTaskRunner::run_in_parallel(
+    const std::function<void(size_t index)>& func,
+    size_t start, size_t end,
+    size_t block_size
+){
+    m_core->run_in_parallel(func, start, end, block_size);
+}
+
+
+
+
+
+
+
+
+ParallelTaskRunnerCore::ParallelTaskRunnerCore(
+    std::function<void()>&& new_thread_callback,
+    size_t starting_threads,
+    size_t max_threads
 )
     : m_new_thread_callback(std::move(new_thread_callback))
     , m_max_threads(max_threads == 0 ? std::thread::hardware_concurrency() : max_threads)
     , m_stopping(false)
     , m_busy_count(0)
 {
     for (size_t c = 0; c < starting_threads; c++){
-        m_threads.emplace_back(run_with_catch, "ParallelTaskRunner::thread_loop()", [this]{ thread_loop(); });
+        spawn_thread();
     }
 }
-ParallelTaskRunner::~ParallelTaskRunner(){
+ParallelTaskRunnerCore::~ParallelTaskRunnerCore(){
     {
         std::lock_guard<std::mutex> lg(m_lock);
         m_stopping = true;
         m_thread_cv.notify_all();
 //        m_dispatch_cv.notify_all();
     }
-    for (std::thread& thread : m_threads){
-        thread.join();
+    for (ThreadData& thread : m_threads){
+        thread.thread.join();
     }
     for (auto& task : m_queue){
         task->signal();
     }
 }
 
-void ParallelTaskRunner::wait_for_everything(){
+WallDuration ParallelTaskRunnerCore::cpu_time() const{
+    //  TODO: Don't lock the entire queue.
+    WallDuration ret = WallDuration::zero();
+    std::lock_guard<std::mutex> lg(m_lock);
+    for (const ThreadData& thread : m_threads){
+//        ret += thread_cpu_time(thread.handle);
+        ret += thread.runtime.total();
+    }
+    return ret;
+}
+
+void ParallelTaskRunnerCore::wait_for_everything(){
     std::unique_lock<std::mutex> lg(m_lock);
     m_dispatch_cv.wait(lg, [this]{
         return m_queue.size() + m_busy_count == 0;
     });
 }
 
-std::shared_ptr<AsyncTask> ParallelTaskRunner::blocking_dispatch(std::function<void()>&& func){
+std::shared_ptr<AsyncTask> ParallelTaskRunnerCore::blocking_dispatch(std::function<void()>&& func){
     std::shared_ptr<AsyncTask> task(new AsyncTask(std::move(func)));
 
     std::unique_lock<std::mutex> lg(m_lock);
@@ -67,14 +226,14 @@ std::shared_ptr<AsyncTask> ParallelTaskRunner::blocking_dispatch(std::function<v
     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(); });
+        spawn_thread();
     }
 
     m_thread_cv.notify_one();
 
     return task;
 }
-std::shared_ptr<AsyncTask> ParallelTaskRunner::try_dispatch(std::function<void()>& func){
+std::shared_ptr<AsyncTask> ParallelTaskRunnerCore::try_dispatch(std::function<void()>& func){
     std::lock_guard<std::mutex> lg(m_lock);
 
     if (m_queue.size() + m_busy_count >= m_max_threads){
@@ -87,7 +246,7 @@ std::shared_ptr<AsyncTask> ParallelTaskRunner::try_dispatch(std::function<void()
     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(); });
+        spawn_thread();
     }
 
     m_thread_cv.notify_one();
@@ -96,7 +255,7 @@ std::shared_ptr<AsyncTask> ParallelTaskRunner::try_dispatch(std::function<void()
 }
 
 
-void ParallelTaskRunner::run_in_parallel(
+void ParallelTaskRunnerCore::run_in_parallel(
     const std::function<void(size_t index)>& func,
     size_t start, size_t end,
     size_t block_size
@@ -126,12 +285,23 @@ void ParallelTaskRunner::run_in_parallel(
 
 
 
+void ParallelTaskRunnerCore::spawn_thread(){
+    ThreadData& handle = m_threads.emplace_back();
+    handle.thread = std::thread(
+        run_with_catch,
+        "ParallelTaskRunner::thread_loop()",
+        [&, this]{ thread_loop(handle); }
+    );
+}
+void ParallelTaskRunnerCore::thread_loop(ThreadData& data){
+    data.handle = current_thread_handle();
 
-void ParallelTaskRunner::thread_loop(){
     if (m_new_thread_callback){
         m_new_thread_callback();
     }
 
+    data.runtime.start();
+
     bool busy = false;
     while (true){
         std::shared_ptr<AsyncTask> task;
@@ -147,7 +317,9 @@ void ParallelTaskRunner::thread_loop(){
                 return;
             }
             if (m_queue.empty()){
+                data.runtime.stop();
                 m_thread_cv.wait(lg);
+                data.runtime.start();
                 continue;
             }
 
@@ -173,4 +345,12 @@ void ParallelTaskRunner::thread_loop(){
 
 
 
+
+//template class Pimpl<ParallelTaskRunnerCore>;
+
+
+
+
+
+
 }

Common/Cpp/Concurrency/ParallelTaskRunner.h

@@ -7,12 +7,17 @@
 #ifndef PokemonAutomation_ParallelTaskRunner_H
 #define PokemonAutomation_ParallelTaskRunner_H
 
+#include "Common/Cpp/Time.h"
+#include "Common/Cpp/Containers/Pimpl.h"
 #include "AsyncDispatcher.h"
 
 namespace PokemonAutomation{
 
 
-class ParallelTaskRunner{
+class ParallelTaskRunnerCore;
+
+
+class ParallelTaskRunner final{
 public:
     ParallelTaskRunner(
         std::function<void()>&& new_thread_callback,
@@ -21,6 +26,9 @@ class ParallelTaskRunner{
     );
     ~ParallelTaskRunner();
 
+    size_t max_threads() const;
+    WallDuration cpu_time() const;
+
     void wait_for_everything();
 
     //  Dispatch the function. If there are no threads available, it waits until
@@ -39,21 +47,12 @@ class ParallelTaskRunner{
 
 
 private:
-//    void dispatch_task(AsyncTask& task);
-    void thread_loop();
+    Pimpl<ParallelTaskRunnerCore> m_core;
+};
+
+
 
 
-private:
-    std::function<void()> m_new_thread_callback;
-    size_t m_max_threads;
-    std::deque<std::shared_ptr<AsyncTask>> m_queue;
-    std::vector<std::thread> m_threads;
-    bool m_stopping;
-    size_t m_busy_count;
-    std::mutex m_lock;
-    std::condition_variable m_thread_cv;
-    std::condition_variable m_dispatch_cv;
-};
 
 
 

Common/Cpp/Concurrency/PeriodicScheduler.cpp

@@ -148,7 +148,7 @@ void PeriodicRunner::thread_loop(){
     bool is_back_to_back = false;
     std::unique_lock<std::mutex> lg(m_lock);
     WallClock last_check_timestamp = current_time();
-    WallClock::duration idle_since_last_check = WallClock::duration(0);
+    WallDuration idle_since_last_check = WallDuration(0);
     while (true){
         if (cancelled()){
             return;
@@ -166,7 +166,7 @@ void PeriodicRunner::thread_loop(){
             m_utilization.push_event(now - last_check_timestamp - idle_since_last_check, now);
         }
         last_check_timestamp = now;
-        idle_since_last_check = WallClock::duration(0);
+        idle_since_last_check = WallDuration(0);
 //        cout << m_utilization.utilization() << endl;
 
         void* event = m_scheduler.request_next_event(now);

Common/Cpp/CpuUtilization/CpuUtilization.cpp

@@ -0,0 +1,16 @@
+/*  CPU Utilization
+ *
+ *  From: https://github.com/PokemonAutomation/
+ *
+ */
+
+#include "CpuUtilization.h"
+
+
+#if _WIN32
+#include "CpuUtilization_Windows.tpp"
+#endif
+
+#if defined(__linux) || defined(__APPLE__)
+#include "CpuUtilization_Linux.tpp"
+#endif