Parallel framework refactor.

Author: Mysticial

Common/Cpp/Concurrency/AsyncDispatcher.cpp

@@ -42,7 +42,8 @@ AsyncDispatcher::~AsyncDispatcher(){
         thread.join();
     }
     for (AsyncTask* task : m_queue){
-        task->signal();
+        task->report_cancelled();
+//        task->signal();
     }
 }
 
@@ -58,7 +59,7 @@ void AsyncDispatcher::dispatch_task(AsyncTask& task){
     std::lock_guard<std::mutex> lg(m_lock);
 
     //  Enqueue task.
-    m_queue.emplace_back(&task);
+    m_queue.emplace_back(&task)->report_started();
 
     //  Make sure a thread is ready for it.
     if (m_queue.size() > m_threads.size() - m_busy_count){
@@ -97,7 +98,7 @@ void AsyncDispatcher::run_in_parallel(
 
         //  Enqueue tasks.
         for (std::unique_ptr<AsyncTask>& task : tasks){
-            m_queue.emplace_back(task.get());
+            m_queue.emplace_back(task.get())->report_started();
         }
 
         //  Make sure there are enough threads.
@@ -153,18 +154,7 @@ void AsyncDispatcher::thread_loop(){
             m_busy_count++;
         }
 
-        try{
-            task->m_task();
-        }catch (...){
-            task->m_exception = std::current_exception();
-            task->m_stopped_with_error.store(true, std::memory_order_release);
-//            cout << "Task threw an exception." << endl;
-//            std::lock_guard<std::mutex> lg(m_lock);
-//            for (AsyncTask* t : m_queue){
-//                t->signal();
-//            }
-        }
-        task->signal();
+        task->run();
     }
 //    cout << "AsyncDispatcher::thread_loop() End (outside) = " << GetCurrentThreadId() << endl;
 }

Common/Cpp/Concurrency/AsyncTask.cpp

@@ -6,38 +6,52 @@
 
 #include "AsyncTask.h"
 
+//#include <iostream>
+//using std::cout;
+//using std::endl;
+
 namespace PokemonAutomation{
 
 
 
 AsyncTask::~AsyncTask(){
-    std::unique_lock<std::mutex> lg(m_lock);
-    m_cv.wait(lg, [this]{ return m_finished; });
-}
-bool AsyncTask::is_finished() const{
-    std::lock_guard<std::mutex> lg(m_lock);
-    return m_finished;
-}
-void AsyncTask::rethrow_exceptions(){
-    if (!m_stopped_with_error.load(std::memory_order_acquire)){
+    State state = m_state.load(std::memory_order_acquire);
+    if (state == State::NOT_STARTED || state == State::SAFE_TO_DESTRUCT){
+//        cout << "Already Done: " << (int)state << endl;
         return;
     }
-    std::unique_lock<std::mutex> lg(m_lock);
-    if (m_exception){
-        std::rethrow_exception(m_exception);
+
+    {
+        std::unique_lock<std::mutex> lg(m_lock);
+        m_cv.wait(lg, [this]{
+            return m_state.load(std::memory_order_relaxed) != State::RUNNING;
+        });
     }
-}
-void AsyncTask::wait_and_rethrow_exceptions(){
-    std::unique_lock<std::mutex> lg(m_lock);
-    m_cv.wait(lg, [this]{ return m_finished; });
-    if (m_exception){
-        std::rethrow_exception(m_exception);
+
+    while (m_state.load(std::memory_order_acquire) != State::SAFE_TO_DESTRUCT){
+        //  pause
     }
+
+//    cout << "Late Finish" << endl;
 }
-void AsyncTask::signal(){
-    std::lock_guard<std::mutex> lg(m_lock);
-    m_finished = true;
+
+
+void AsyncTask::report_cancelled() noexcept{
+    m_state.store(State::FINISHED, std::memory_order_release);
+    {
+        std::lock_guard<std::mutex> lg(m_lock);
+    }
     m_cv.notify_all();
+    m_state.store(State::SAFE_TO_DESTRUCT, std::memory_order_release);
+}
+void AsyncTask::run() noexcept{
+    try{
+        m_task();
+    }catch (...){
+        std::lock_guard<std::mutex> lg(m_lock);
+        m_exception = std::current_exception();
+    }
+    report_cancelled();
 }
 
 

Common/Cpp/Concurrency/AsyncTask.h

@@ -18,36 +18,57 @@ namespace PokemonAutomation{
 
 class AsyncTask{
 public:
-    //  Wait for the task to finish before destructing. Doesn't rethrow exceptions.
-    ~AsyncTask();
-
-    bool is_finished() const;
+    enum class State{
+        NOT_STARTED,
+        RUNNING,
+        FINISHED,
+        SAFE_TO_DESTRUCT,
+    };
 
-    //  If the task ended with an exception, rethrow it here.
-    //  This does not clear the exception.
-    void rethrow_exceptions();
 
-    //  Wait for the task to finish. Will rethrow any exceptions.
-    void wait_and_rethrow_exceptions();
+public:
+    //  If the task has already started, this will wait for it to finish.
+    //  This will not rethrow exceptions.
+    ~AsyncTask();
 
 
-private:
+public:
     template <class... Args>
     AsyncTask(Args&&... args)
         : m_task(std::forward<Args>(args)...)
-        , m_finished(false)
-        , m_stopped_with_error(false)
+        , m_state(State::NOT_STARTED)
     {}
-    void signal();
 
-private:
-    friend class FireForgetDispatcher;
-    friend class AsyncDispatcher;
-    friend class ComputationThreadPoolCore;
+    bool is_finished() const noexcept{
+        State state = m_state.load(std::memory_order_acquire);
+        return state == State::FINISHED || state == State::SAFE_TO_DESTRUCT;
+    }
 
+    //  Wait for the task to finish. Will rethrow any exceptions.
+    void wait_and_rethrow_exceptions(){
+        if (!is_finished()){
+            std::unique_lock<std::mutex> lg(m_lock);
+            m_cv.wait(lg, [this]{ return is_finished(); });
+        }
+        if (m_exception){
+            std::rethrow_exception(m_exception);
+        }
+    }
+
+
+public:
+    //  These should only be called inside a parallel framework.
+    //  These are not thread-safe with each other.
+    void report_started(){
+        m_state.store(State::RUNNING, std::memory_order_release);
+    }
+    void report_cancelled() noexcept;
+    void run() noexcept;
+
+
+private:
     std::function<void()> m_task;
-    bool m_finished;
-    std::atomic<bool> m_stopped_with_error;
+    std::atomic<State> m_state;
     std::exception_ptr m_exception;
     mutable std::mutex m_lock;
     std::condition_variable m_cv;