AutonomyThread.hpp

 
#ifndef AUTONOMYTHREAD_H
#define AUTONOMYTHREAD_H
 
#include "../util/IPS.hpp"
 
#include "../../external/threadpool/include/BS_thread_pool.hpp"
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <vector>
 
 
 
template<class T>
class AutonomyThread
{
    public:
        // Define public enumerators specific to this class.
 
        // Define an enum for storing this classes state.
        enum class AutonomyThreadState
        {
            eStarting,
            eRunning,
            eStopping,
            eStopped
        };
 
        // Declare and define public class methods.
 
        AutonomyThread()
        {
            // Initialize member variables.
            m_bStopThreads                     = false;
            m_eThreadState                     = AutonomyThreadState::eStopped;
            m_nMainThreadMaxIterationPerSecond = 0;
        }
 
 
        virtual ~AutonomyThread()
        {
            // Tell all threads to stop executing user code.
            m_bStopThreads = true;
            // Update thread state.
            m_eThreadState = AutonomyThreadState::eStopping;
 
            // Pause and clear pool queues.
            m_thPool.pause();
            m_thPool.purge();
            m_thMainThread.pause();
            m_thMainThread.purge();
 
            // Wait for all pools to finish.
            m_thPool.wait();
            m_thMainThread.wait();
            // Update thread state.
            m_eThreadState = AutonomyThreadState::eStopped;
        }
 
 
        void Start()
        {
            // Tell any open thread to stop.
            m_bStopThreads = true;
            // Update thread state.
            m_eThreadState = AutonomyThreadState::eStopping;
 
            // Pause queuing of new tasks to the threads, then purge them.
            m_thPool.pause();
            m_thPool.purge();
            m_thMainThread.pause();
            m_thMainThread.purge();
 
            // Wait for loop, pool and main thread to join.
            this->Join();
 
            // Update thread state.
            m_eThreadState = AutonomyThreadState::eStarting;
            // Clear results vector.
            m_vPoolReturns.clear();
            // Reset thread stop toggle.
            m_bStopThreads = false;
 
            // Submit single task to pool queue and store resulting future. Still using pool, as it's scheduling is more efficient.
            std::future<void> fuMainReturn = m_thMainThread.submit_task([this]() { this->RunThread(m_bStopThreads); });
 
            // Unpause pool queues.
            m_thPool.unpause();
            m_thMainThread.unpause();
 
            // Block until thread is started or currently stopping if thread start failed.
            std::unique_lock<std::mutex> lkStartLock(m_muThreadRunningConditionMutex);
            m_cdThreadRunningCondition.wait(lkStartLock,
                                            [this]
                                            { return this->m_eThreadState == AutonomyThreadState::eRunning || this->m_eThreadState == AutonomyThreadState::eStopping; });
        }
 
 
        void RequestStop()
        {
            // Signal for any open threads to stop executing,
            m_bStopThreads = true;
            // Update thread state.
            m_eThreadState = AutonomyThreadState::eStopping;
        }
 
 
        void Join()
        {
            // Wait for pool to finish all tasks.
            m_thPool.wait();
            // Wait for main thread to finish.
            m_thMainThread.wait();
 
            // Update thread state.
            m_eThreadState = AutonomyThreadState::eStopped;
        }
 
 
        bool Joinable() const
        {    // Check current number of running and queued tasks.
            return (m_thMainThread.get_tasks_total() <= 0 && m_thPool.get_tasks_total() <= 0);
        }
 
 
        AutonomyThreadState GetThreadState() const { return m_eThreadState; }
 
 
        IPS& GetIPS() { return m_IPS; }
 
    protected:
        // Declare protected objects.
        IPS m_IPS = IPS();
 
        // Declare and define protected class methods.
 
 
        void RunPool(const unsigned int nNumTasksToQueue, const unsigned int nNumThreads = 2, const bool bForceStopCurrentThreads = false)
        {
            // Check if the pools need to be resized.
            if (m_thPool.get_thread_count() != nNumThreads)
            {
                // Pause queuing of new tasks to the threads, then purge them.
                m_thPool.pause();
                m_thPool.purge();
                // Wait for open threads to terminate, then resize the pool.
                m_thPool.reset(nNumThreads);
                // Unpause queue.
                m_thPool.unpause();
 
                // Clear results vector.
                m_vPoolReturns.clear();
            }
            // Check if the current pool tasks should be stopped before queueing more tasks.
            else if (bForceStopCurrentThreads)
            {
                // Pause queuing of new tasks to the threads, then purge them.
                m_thPool.pause();
                m_thPool.purge();
                // Wait for threadpool to join.
                m_thPool.wait();
                // Unpause queue.
                m_thPool.unpause();
            }
 
            // Loop nNumThreads times and queue tasks.
            for (unsigned int i = 0; i < nNumTasksToQueue; ++i)
            {
                // Submit single task to pool queue.
                m_vPoolReturns.emplace_back(m_thPool.submit_task(
                    [this]()
                    {
                        // Run user pool code without lock.
                        this->PooledLinearCode();
                    }));
            }
        }
 
 
        void RunDetachedPool(const unsigned int nNumTasksToQueue, const unsigned int nNumThreads = 2, const bool bForceStopCurrentThreads = false)
        {
            // Check if the pools need to be resized.
            if (m_thPool.get_thread_count() != nNumThreads)
            {
                // Pause queuing of new tasks to the threads, then purge them.
                m_thPool.pause();
                m_thPool.purge();
                // Wait for open threads to terminate, then resize the pool.
                m_thPool.reset(nNumThreads);
                // Unpause queue.
                m_thPool.unpause();
 
                // Clear results vector.
                m_vPoolReturns.clear();
            }
            // Check if the current pool tasks should be stopped before queueing more tasks.
            else if (bForceStopCurrentThreads)
            {
                // Pause queuing of new tasks to the threads, then purge them.
                m_thPool.pause();
                m_thPool.purge();
                // Wait for threadpool to join.
                m_thPool.wait();
                // Unpause queue.
                m_thPool.unpause();
            }
 
            // Loop nNumThreads times and queue tasks.
            for (unsigned int i = 0; i < nNumTasksToQueue; ++i)
            {
                // Push single task to pool queue. No return value no control.
                m_thPool.detach_task(
                    [this]()
                    {
                        // Run user code without lock.
                        this->PooledLinearCode();
                    });
            }
        }
 
 
        template<typename N, typename F>
        void ParallelizeLoop(const int nNumThreads, const N tTotalIterations, F&& tLoopFunction)
        {
            // Create new thread pool.
            BS::thread_pool m_thLoopPool = BS::thread_pool(nNumThreads);
 
            m_thLoopPool.detach_blocks(0,
                                       tTotalIterations,
                                       [&tLoopFunction](const int nStart, const int nEnd)
                                       {
                                           // Call loop function without lock.
                                           tLoopFunction(nStart, nEnd);
                                       });
 
            // Wait for loop to finish.
            m_thLoopPool.wait();
        }
 
 
        void ClearPoolQueue() { m_thPool.purge(); }
 
 
        void JoinPool() { m_thPool.wait(); }
 
 
        bool PoolJoinable() const
        {
            // Check current number of running and queued tasks.
            return (m_thPool.get_tasks_total() <= 0);
        }
 
 
        void SetMainThreadIPSLimit(int nMaxIterationsPerSecond = 0)
        {
            // Assign member variable.
            m_nMainThreadMaxIterationPerSecond = nMaxIterationsPerSecond;
        }
 
 
        int GetPoolNumOfThreads() { return m_thPool.get_thread_count(); }
 
 
        int GetPoolQueueLength() { return m_thPool.get_tasks_queued(); }
 
 
        std::vector<T> GetPoolResults()
        {
            // Create instance variable.
            std::vector<T> vResults;
 
            // Loop the pool futures and get result.
            for (std::future<T> fResult : m_vPoolReturns)
            {
                // Store returned value.
                vResults.emplace_back(fResult.get());
            }
 
            // Clear pool returns member variable.
            m_vPoolReturns.clear();
 
            return vResults;
        }
 
 
        int GetMainThreadMaxIPS() const
        {
            // Return member variable value.
            return m_nMainThreadMaxIterationPerSecond;
        }
 
    private:
        // Declare private class member variables.
 
        BS::thread_pool m_thMainThread = BS::thread_pool(1);
        BS::thread_pool m_thPool       = BS::thread_pool(2);
        std::vector<std::future<T>> m_vPoolReturns;
        std::atomic_bool m_bStopThreads;
        std::atomic<AutonomyThreadState> m_eThreadState;
        std::mutex m_muThreadRunningConditionMutex;
        std::condition_variable m_cdThreadRunningCondition;
        int m_nMainThreadMaxIterationPerSecond;
 
        // Declare and/or define private methods.
 
        // Declare interface class pure virtual functions. (These must be overriden by inheritor.)
        virtual void ThreadedContinuousCode() = 0;    // This is where user's main single threaded and continuously looping code will go.
        virtual T PooledLinearCode()          = 0;    // This is where user's offshoot, highly parallelizable code will go. Helpful for intensive short-lived tasks.
                                                      // Can be ran from inside the ThreadedContinuousCode() method.
 
        // Declare and define private interface methods.
 
        void RunThread(std::atomic_bool& bStopThread)
        {
            // Declare instance variables.
            std::chrono::_V2::system_clock::time_point tmStartTime;
 
            // Loop until stop flag is set.
            while (!bStopThread)
            {
                // Check if max IPS limit has been set.
                if (m_nMainThreadMaxIterationPerSecond > 0)
                {
                    // Get start execution time.
                    tmStartTime = std::chrono::high_resolution_clock::now();
                }
 
                // Call method containing user code.
                this->ThreadedContinuousCode();
 
                // Check if max IPS limit has been set.
                if (m_nMainThreadMaxIterationPerSecond > 0)
                {
                    // Get end execution time.
                    std::chrono::_V2::system_clock::time_point tmEndTime = std::chrono::high_resolution_clock::now();
                    // Get execution time of user code.
                    std::chrono::microseconds tmElapsedTime = std::chrono::duration_cast<std::chrono::microseconds>(tmEndTime - tmStartTime);
                    // Check if the elapsed time is slower than the max iterations per seconds.
                    if (tmElapsedTime.count() < (1.0 / m_nMainThreadMaxIterationPerSecond) * 1000000)
                    {
                        // Calculate the time to wait to stay under IPS cap.
                        int nSleepTime = ((1.0 / m_nMainThreadMaxIterationPerSecond) * 1000000) - tmElapsedTime.count();
                        // Make this thread sleep for the remaining time.
                        std::this_thread::sleep_for(std::chrono::microseconds(nSleepTime));
                    }
                }
 
                // Check if thread state needs to be updated.
                if (m_eThreadState != AutonomyThreadState::eRunning && m_eThreadState != AutonomyThreadState::eStopping)
                {
                    // Update thread state to running.
                    m_eThreadState = AutonomyThreadState::eRunning;
                    // Notify waiting start method that thread is now running.
                    m_cdThreadRunningCondition.notify_all();
                }
 
                // Call iteration per second tracking tick.
                m_IPS.Tick();
            }
 
            // Notify waiting start method that thread is now stopping.
            m_cdThreadRunningCondition.notify_all();
        }
};
 
#endif