Merge branch 'PokemonAutomation:main' into main

Author: NympheaR

Common/Cpp/Concurrency/AsyncTask.cpp

@@ -4,8 +4,14 @@
  *
  */
 
+#include <atomic>
+#include <exception>
+#include "Common/Cpp/Containers/Pimpl.tpp"
 #include "SpinPause.h"
+#include "Mutex.h"
+#include "ConditionVariable.h"
 #include "AsyncTask.h"
+#include "AsyncTaskCore.h"
 
 //#include <iostream>
 //using std::cout;
@@ -15,54 +21,55 @@ namespace PokemonAutomation{
 
 
 
+
+AsyncTask::AsyncTask(AsyncTask&&) = default;
+AsyncTask& AsyncTask::operator=(AsyncTask&&) = default;
+
+
+AsyncTask::AsyncTask()
+    : m_sanitizer("AsyncTask")
+{}
+AsyncTask::AsyncTask(std::function<void()> task)
+    : m_core(CONSTRUCT_TOKEN, std::move(task))
+    , m_sanitizer("AsyncTask")
+{}
+
 AsyncTask::~AsyncTask(){
-    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<Mutex> lg(m_lock);
-        m_cv.wait(lg, [this]{
-            return m_state.load(std::memory_order_relaxed) != State::RUNNING;
-        });
-    }
-
-    while (m_state.load(std::memory_order_acquire) != State::SAFE_TO_DESTRUCT){
-        pause();
-    }
-
-//    cout << "Late Finish" << endl;
+    wait_and_ignore_exceptions();
 }
 
 
-void AsyncTask::report_cancelled() noexcept{
-    {
+bool AsyncTask::is_finished() const noexcept{
 #ifdef PA_SANITIZE_AsyncTask
-        auto scope = m_sanitizer.check_scope();
+    auto scope = m_sanitizer.check_scope();
 #endif
-        m_state.store(State::FINISHED, std::memory_order_release);
-        {
-            std::lock_guard<Mutex> lg(m_lock);
-        }
-        m_cv.notify_all();
-    }
-    m_state.store(State::SAFE_TO_DESTRUCT, std::memory_order_release);
+    return m_core->is_finished();
 }
-void AsyncTask::run() noexcept{
-    {
+
+void AsyncTask::wait_and_ignore_exceptions() noexcept{
+    m_core.clear();
+}
+void AsyncTask::wait_and_rethrow_exceptions(){
 #ifdef PA_SANITIZE_AsyncTask
-        auto scope = m_sanitizer.check_scope();
+    auto scope = m_sanitizer.check_scope();
 #endif
-        try{
-            m_task();
-        }catch (...){
-            std::lock_guard<Mutex> lg(m_lock);
-            m_exception = std::current_exception();
-        }
-    }
-    report_cancelled();
+    m_core->wait_and_rethrow_exceptions();
+}
+
+
+void AsyncTask::report_started(){
+#ifdef PA_SANITIZE_AsyncTask
+    auto scope = m_sanitizer.check_scope();
+#endif
+    m_core->report_started();
+}
+void AsyncTask::report_cancelled() noexcept{
+    m_sanitizer.check_usage();
+    m_core->report_cancelled();
+}
+void AsyncTask::run() noexcept{
+    m_sanitizer.check_usage();
+    m_core->run();
 }
 
 

Common/Cpp/Concurrency/AsyncTask.h

@@ -8,10 +8,7 @@
 #define PokemonAutomation_AsyncTask_H
 
 #include <functional>
-#include <atomic>
-#include <exception>
-#include "Mutex.h"
-#include "ConditionVariable.h"
+#include "Common/Cpp/Containers/Pimpl.h"
 
 #define PA_SANITIZE_AsyncTask
 
@@ -21,72 +18,50 @@
 
 namespace PokemonAutomation{
 
+class AsyncTaskCore;
+
 
 class AsyncTask{
 public:
-    enum class State{
-        NOT_STARTED,
-        RUNNING,
-        FINISHED,
-        SAFE_TO_DESTRUCT,
-    };
-
+    AsyncTask(const AsyncTask&) = delete;
+    void operator=(const AsyncTask&) = delete;
+    AsyncTask(AsyncTask&&);
+    AsyncTask& operator=(AsyncTask&&);
 
-public:
     //  If the task has already started, this will wait for it to finish.
     //  This will not rethrow exceptions.
     ~AsyncTask();
 
 
 public:
-    template <class... Args>
-    AsyncTask(Args&&... args)
-        : m_task(std::forward<Args>(args)...)
-        , m_state(State::NOT_STARTED)
-    {}
+    AsyncTask();
+    AsyncTask(std::function<void()> task);
 
-    bool is_finished() const noexcept{
-#ifdef PA_SANITIZE_AsyncTask
-        auto scope = m_sanitizer.check_scope();
-#endif
-        State state = m_state.load(std::memory_order_acquire);
-        return state == State::FINISHED || state == State::SAFE_TO_DESTRUCT;
+    operator bool() const{
+        return m_core;
+    }
+    AsyncTaskCore* core(){
+        return m_core.get();
     }
+    bool is_finished() const noexcept;
+
+    //  Wait for the task to finish. Will ignore any exceptions.
+    void wait_and_ignore_exceptions() noexcept;
 
     //  Wait for the task to finish. Will rethrow any exceptions.
-    void wait_and_rethrow_exceptions(){
-#ifdef PA_SANITIZE_AsyncTask
-        auto scope = m_sanitizer.check_scope();
-#endif
-        if (!is_finished()){
-            std::unique_lock<Mutex> lg(m_lock);
-            m_cv.wait(lg, [this]{ return is_finished(); });
-        }
-        if (m_exception){
-            std::rethrow_exception(m_exception);
-        }
-    }
+    void wait_and_rethrow_exceptions();
 
 
 public:
     //  These should only be called inside a parallel framework.
     //  These are not thread-safe with each other.
-    void report_started(){
-#ifdef PA_SANITIZE_AsyncTask
-        auto scope = m_sanitizer.check_scope();
-#endif
-        m_state.store(State::RUNNING, std::memory_order_release);
-    }
+    void report_started();
     void report_cancelled() noexcept;
     void run() noexcept;
 
 
 private:
-    std::function<void()> m_task;
-    std::atomic<State> m_state;
-    std::exception_ptr m_exception;
-    mutable Mutex m_lock;
-    ConditionVariable m_cv;
+    Pimpl<AsyncTaskCore> m_core;
 
 #ifdef PA_SANITIZE_AsyncTask
     LifetimeSanitizer m_sanitizer;

Common/Cpp/Concurrency/AsyncTaskCore.h

@@ -0,0 +1,116 @@
+/*  Async Task
+ *
+ *  From: https://github.com/PokemonAutomation/
+ *
+ */
+
+#ifndef PokemonAutomation_AsyncTaskCore_H
+#define PokemonAutomation_AsyncTaskCore_H
+
+#include <functional>
+#include <atomic>
+#include <exception>
+#include "SpinPause.h"
+#include "Mutex.h"
+#include "ConditionVariable.h"
+
+namespace PokemonAutomation{
+
+
+
+class AsyncTaskCore{
+public:
+    enum class State{
+        NOT_STARTED,
+        RUNNING,
+        FINISHED,
+        SAFE_TO_DESTRUCT,
+    };
+
+
+public:
+    //  If the task has already started, this will wait for it to finish.
+    //  This will not rethrow exceptions.
+    ~AsyncTaskCore(){
+        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<Mutex> lg(m_lock);
+            m_cv.wait(lg, [this]{
+                return m_state.load(std::memory_order_relaxed) != State::RUNNING;
+            });
+        }
+
+        while (m_state.load(std::memory_order_acquire) != State::SAFE_TO_DESTRUCT){
+            pause();
+        }
+    }
+    AsyncTaskCore(std::function<void()> task)
+        : m_task(std::move(task))
+        , m_state(State::NOT_STARTED)
+    {}
+
+    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<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{
+        {
+            m_state.store(State::FINISHED, std::memory_order_release);
+            {
+                std::lock_guard<Mutex> lg(m_lock);
+            }
+            m_cv.notify_all();
+        }
+        m_state.store(State::SAFE_TO_DESTRUCT, std::memory_order_release);
+    }
+    void run() noexcept{
+        {
+            try{
+                m_task();
+            }catch (...){
+                std::lock_guard<Mutex> lg(m_lock);
+                m_exception = std::current_exception();
+            }
+        }
+        report_cancelled();
+    }
+
+
+private:
+    std::function<void()> m_task;
+    std::atomic<State> m_state;
+
+    std::exception_ptr m_exception;
+
+    mutable Mutex m_lock;
+    ConditionVariable m_cv;
+};
+
+
+
+}
+#endif

Common/Cpp/Concurrency/PeriodicScheduler.cpp

@@ -196,9 +196,9 @@ void PeriodicRunner::thread_loop(){
         idle_since_last_check += end - start;
     }
 }
-void PeriodicRunner::stop_thread(){
+void PeriodicRunner::stop_thread() noexcept{
     PeriodicRunner::cancel(nullptr);
-    m_runner.reset();
+    m_runner.wait_and_ignore_exceptions();
 }
 
 double PeriodicRunner::current_utilization() const{

Common/Cpp/Concurrency/PeriodicScheduler.h

@@ -88,7 +88,7 @@ class PeriodicRunner : public Cancellable{
 private:
     void thread_loop();
 protected:
-    void stop_thread();
+    void stop_thread() noexcept;
 
 private:
     ThreadPool& m_thread_pool;
@@ -102,7 +102,7 @@ class PeriodicRunner : public Cancellable{
 
     PeriodicScheduler m_scheduler;
 
-    std::unique_ptr<AsyncTask> m_runner;
+    AsyncTask m_runner;
 };
 
 

Common/Cpp/Concurrency/ScheduledTaskRunner.cpp

@@ -25,7 +25,7 @@ ScheduledTaskRunner::ScheduledTaskRunner(ThreadPool& thread_pool)
 ScheduledTaskRunner::~ScheduledTaskRunner(){
     stop();
 }
-void ScheduledTaskRunner::stop(){
+void ScheduledTaskRunner::stop() noexcept{
     if (!m_runner){
         return;
     }
@@ -36,7 +36,7 @@ void ScheduledTaskRunner::stop(){
         m_stopped = true;
     }
     m_cv.notify_all();
-    m_runner.reset();
+    m_runner.wait_and_ignore_exceptions();
 //    cout << "ScheduledTaskRunner: (Destructor - end):   " << this << endl;
 }