AutonomyThread.hpp

Go to the documentation of this file.

 
#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 <random>
#include <sstream>
#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
        };
 
        // Define an enum for storing thread priority levels. These are used as an abstraction layer for the BS::pr enum, so we don't have to include the thread pool
        // header in child classes just to set priorities.
        enum class AutonomyThreadPriority
        {
            eLowest = BS::pr::lowest,     // Thread will be scheduled less often and may not be able to reach the max IPS if the system is under heavy load.
            eLow    = BS::pr::low,        // Thread will be scheduled less often and may not be able to reach the max IPS if the system is under heavy load.
            eNormal = BS::pr::normal,     // Default. Thread will be scheduled normally and should be able to reach the max IPS if the system is not under heavy load.
            eHigh   = BS::pr::high,       // Thread will be scheduled more often and should be able to reach the max IPS even if the system is under
                                          // heavy load, but it may starve other threads of CPU time.
            eHighest = BS::pr::highest    // Thread will be scheduled as often as possible and should be able to reach the max IPS even if the system is under heavy load.
        };
 
        // Declare and define public class methods.
 
        AutonomyThread()
        {
            // Initialize member variables.
            m_bStopThreads                     = false;
            m_eThreadState                     = AutonomyThreadState::eStopped;
            m_nMainThreadMaxIterationPerSecond = 0;
 
            // Generate a random UUID for this thread.
            m_szThreadUUID = this->GenerateUUIDV4();
        }
 
 
        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]()
                {
                    // Note: BS::this_thread::set_os_thread_priority was completely removed in BS v5.1.0.
                    // Priority is kept in memory but not applied to OS threads natively by the library anymore.
                    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; }
 
 
        std::string GetThreadUUID() const { return m_szThreadUUID; }
 
 
        IPS& GetIPS() { return m_IPS; }
 
 
        void SetMainThreadPriority(AutonomyThreadPriority ePriority) { m_eMainThreadPriority = static_cast<BS::pr>(ePriority); }
 
 
        void SetPoolThreadPriority(AutonomyThreadPriority ePriority)
        {
            // Update member variable.
            m_ePoolThreadPriority = static_cast<BS::pr>(ePriority);
            // Native OS priority extensions were removed in BS v5.1.0, so we just reset thread count.
            m_thPool.reset(m_thPool.get_thread_count());
        }
 
    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 nIter = 0; nIter < nNumTasksToQueue; ++nIter)
            {
                // 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 nIter = 0; nIter < nNumTasksToQueue; ++nIter)
            {
                // 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 (no specific template flags needed for basic detaching).
            BS::thread_pool<> m_thLoopPool(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::pr m_eMainThreadPriority = BS::pr::normal;
        BS::pr m_ePoolThreadPriority = BS::pr::normal;
 
        // BS::thread_pool requires the BS::tp::pause flag to enable pause/unpause
        BS::thread_pool<BS::tp::pause> m_thMainThread = BS::thread_pool<BS::tp::pause>(1);
        BS::thread_pool<BS::tp::pause> m_thPool       = BS::thread_pool<BS::tp::pause>(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;
        std::string m_szThreadUUID;
 
        // 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();
        }
 
 
        std::string GenerateUUIDV4()
        {
            static std::random_device stdRandomDevice;
            static std::mt19937 stdGen(stdRandomDevice());
            static std::uniform_int_distribution<> stdDist1(0, 15);
            static std::uniform_int_distribution<> stdDist2(8, 11);
 
            std::stringstream stdStream;
            stdStream << std::hex;
            for (int nIter = 0; nIter < 8; nIter++)
            {
                stdStream << stdDist1(stdGen);
            }
            stdStream << "-";
            for (int nIter = 0; nIter < 4; nIter++)
            {
                stdStream << stdDist1(stdGen);
            }
            stdStream << "-4";
            for (int nIter = 0; nIter < 3; nIter++)
            {
                stdStream << stdDist1(stdGen);
            }
            stdStream << "-";
            stdStream << stdDist2(stdGen);
            for (int nIter = 0; nIter < 3; nIter++)
            {
                stdStream << stdDist1(stdGen);
            }
            stdStream << "-";
            for (int nIter = 0; nIter < 12; nIter++)
            {
                stdStream << stdDist1(stdGen);
            }
            return stdStream.str();
        }
};
 
#endif