YiThreadPool.h

Go to the documentation of this file.

// © You i Labs Inc. 2000-2017. All rights reserved.
#ifndef _YI_THREAD_POOL_H_
#define _YI_THREAD_POOL_H_

#include "thread/YiMutex.h"
#include "thread/YiStaticTask.h"
#include "thread/YiThread.h"
#include "thread/YiWaitCondition.h"

#ifdef YI_DEBUG
#define YI_THREAD_POOL_INSTRUMENTED
#endif

template <class ResultType>
class CYIFuture;
template <class ResultType>
class CYIStaticTask;
class CYITaskBase;

class CYIThreadPool
{
public:
    CYIThreadPool(const CYIString &sName = "ThreadPool", uint32_t uMaxSize = 4, uint32_t uMaxSleepingSize = 4, uint32_t uInitialSize = 2, uint32_t uExpiryTimeMs = 30000, YI_PRIORITY ePriority = YI_PRIORITY_DEFAULT, uint32_t uStackSize = CYIThread::DEFAULT_STACK_SIZE);
    virtual ~CYIThreadPool();
    
    const CYIString &GetName() const;
    
    uint32_t GetSleepingThreadsCount() const;

    uint32_t GetThreadsCount() const;

    uint32_t GetPendingTasksCount() const;

    bool Queue(std::unique_ptr<CYITaskBase> pTask);

    void Shutdown();

    template<typename ResultType>
    std::unique_ptr<CYIFuture<ResultType>> Queue(ResultType (*const pFunc)());

    template<typename ResultType, typename P1>
    std::unique_ptr<CYIFuture<ResultType>> Queue(ResultType (*const pFunc)(P1), P1 const &funcParam1);

    template<typename ResultType, typename P1, typename P2>
    std::unique_ptr<CYIFuture<ResultType>> Queue(ResultType (*const pFunc)(P1, P2), P1 const &funcParam1, P2 const &funcParam2);

    template<typename ResultType, typename P1, typename P2, typename P3>
    std::unique_ptr<CYIFuture<ResultType>> Queue(ResultType (*const pFunc)(P1, P2, P3), P1 const &funcParam1, P2 const &funcParam2, P3 const &funcParam3);

    template<typename ResultType, typename P1, typename P2, typename P3, typename P4>
    std::unique_ptr<CYIFuture<ResultType>> Queue(ResultType (*const pFunc)(P1, P2, P3, P4), P1 const &funcParam1, P2 const &funcParam2, P3 const &funcParam3, P4 const &funcParam4);

    template <typename ResultType>
    bool QueueAndForget(ResultType (* const pFunc)());
    
    template <typename ResultType, typename P1>
    bool QueueAndForget(ResultType (* const pFunc)(P1), P1 const &funcParam1);
    
    template <typename ResultType, typename P1, typename P2>
    bool QueueAndForget(ResultType (* const pFunc)(P1, P2), P1 const &funcParam1, P2 const &funcParam2);
    
    template <typename ResultType, typename P1, typename P2, typename P3>
    bool QueueAndForget(ResultType (* const pFunc)(P1, P2, P3), P1 const &funcParam1, P2 const &funcParam2, P3 const &funcParam3);
    
    template <typename ResultType, typename P1, typename P2, typename P3, typename P4>
    bool QueueAndForget(ResultType (* const pFunc)(P1, P2, P3, P4), P1 const &funcParam1, P2 const &funcParam2, P3 const &funcParam3, P4 const &funcParam4);

#ifdef YI_THREAD_POOL_INSTRUMENTED
    struct Statistics
    {
        uint64_t m_uQueuedTasks; 
        uint64_t m_uExecutedTasks; 
        uint64_t m_uCancelledTasks; 
        float m_fCancelledTasksPercentage; // The percentage of queued tasks that have been cancelled (0..1).

        uint32_t m_uCreatedThreads; 
        uint32_t m_uDestroyedThreads; 

        uint64_t m_uTasksThatMayHaveRunImmediately; 
        uint64_t m_uTasksThatLikelyHadToWaitForExecution; 
        float m_fBlockedTasksPercentage; // The percentage of queued tasks that may have had to wait until a worker thread was available (0..1).

        uint64_t m_uThreadTimeSpentExecutinguS; 
        uint64_t m_uThreadTimeSpentSleepinguS; 
        uint64_t m_uThreadTimeSpentOtheruS; 
        float m_fThreadWastePercentage; 
        float m_fThreadExecutionPercentage; 
        float m_fAverageTaskExecutionTimeuS; 

        uint64_t m_uMinTaskWaitTimeuS; 
        uint64_t m_uMaxTaskWaitTimeuS; 
        float m_fAverageTaskWaitTimeuS; 

        uint32_t m_uCurrentThreadsCount; 
        uint32_t m_uCurrentSleepingThreadsCount; 
        uint32_t m_uCurrentPendingTasksCount; 
    };
    
    Statistics GetStatistics();
    
private:
    std::atomic<uint64_t> m_nQueuedTasks{ 0 };
    std::atomic<uint64_t> m_nExecutedTasks{ 0 };
    std::atomic<uint64_t> m_nCancelledTasks{ 0 };

    std::atomic<uint32_t> m_nCreatedThreads{ 0 };
    std::atomic<uint32_t> m_nDestroyedThreads{ 0 };

    std::atomic<uint64_t> m_nTasksThatMayHaveRunImmediately{ 0 };

    std::atomic<uint64_t> m_nThreadTimeSpentExecutinguS{ 0 };
    std::atomic<uint64_t> m_nThreadTimeSpentSleepinguS{ 0 };
    std::atomic<uint64_t> m_nThreadTimeSpentOtheruS{ 0 };

    std::atomic<uint64_t> m_nMinTaskWaitTimeuS{ 0 };
    std::atomic<uint64_t> m_nMaxTaskWaitTimeuS{ 0 };
    std::atomic<uint64_t> m_nTotalTaskWaitTimeuS{ 0 };

    mutable CYIMutex    m_statisticsMutex;
    std::map<const CYITaskBase *, uint64_t> m_taskQueueingTimes;
#endif
    
private:
    class WorkerThread : public CYIThread
    {
    public:
        WorkerThread(CYIThreadPool *pParent, const CYIString &sName, YI_PRIORITY ePriority, uint32_t uStackSize);
        virtual ~WorkerThread();
        virtual void Run() override;
        void CancelRunningTask();
        
    private:
        CYIThreadPool  *m_pParent;
        std::unique_ptr<CYITaskBase> m_pCurrentTask;
        CYIMutex        m_taskMutex;
    };
    
    WorkerThread *CreateWorker(); 
    void AddWorker(); 
    void AddWorkersIfNeeded(); 

    std::list<WorkerThread*>        m_workers;
    std::list<std::unique_ptr<CYITaskBase>> m_pendingTasks;
    mutable CYIMutex                m_mutex;
    mutable CYIWaitCondition        m_workAvailable;
    mutable CYIWaitCondition        m_workerCompleted;
    bool                         m_bShutDown;
    uint32_t m_uSleepingThreadsCount;
    uint32_t m_uNextThreadNumber;
    CYIThread                      *m_pWorkerPendingDeletion;
    
    CYIString                       m_sName;
    uint32_t m_uMaxSize;
    uint32_t m_uMaxSleepingSize;
    uint64_t m_uExpiryTimeMs;
    YI_PRIORITY                     m_ePriority;
    uint32_t m_uStackSize;
};

#include "YiThreadPool.inl"

#endif // _YI_THREAD_POOL_H_