stdlib/Thread.h

Go to the documentation of this file.

#ifndef _THREAD_STD_H
#define _THREAD_STD_H
 
 
#include "delegate/IThread.h"
#include "./predef/util/Timer.h"
#include "ThreadMsg.h"
#include <thread>
#include <queue>
#include <atomic>
#include <condition_variable>
#include <future>
#include <optional>
 
// Comparator for priority queue
struct ThreadMsgComparator {
    bool operator()(const std::shared_ptr<ThreadMsg>& a, const std::shared_ptr<ThreadMsg>& b) const {
        return static_cast<int>(a->GetPriority()) < static_cast<int>(b->GetPriority());
    }
};
 
class Thread : public dmq::IThread
{
public:
    Thread(const std::string& threadName, size_t maxQueueSize = 0);
 
    ~Thread();
 
    bool CreateThread(std::optional<dmq::Duration> watchdogTimeout = std::nullopt);
 
    void ExitThread();
 
    std::thread::id GetThreadId();
 
    static std::thread::id GetCurrentThreadId();
 
    virtual bool IsCurrentThread() override;
 
    std::string GetThreadName() { return THREAD_NAME; }
 
    size_t GetQueueSize();
 
    virtual void DispatchDelegate(std::shared_ptr<dmq::DelegateMsg> msg);
 
private:
    Thread(const Thread&) = delete;
    Thread& operator=(const Thread&) = delete;
 
    void Process();
 
    void SetThreadName(std::thread::native_handle_type handle, const std::string& name);
 
    void WatchdogCheck();
 
    void ThreadCheck();
 
    std::optional<std::thread> m_thread;
#ifdef DMQ_ALLOCATOR
    std::priority_queue<std::shared_ptr<ThreadMsg>,
        std::vector<std::shared_ptr<ThreadMsg>, stl_allocator<std::shared_ptr<ThreadMsg>>>,
        ThreadMsgComparator> m_queue;
#else
    std::priority_queue<std::shared_ptr<ThreadMsg>,
        std::vector<std::shared_ptr<ThreadMsg>>,
        ThreadMsgComparator> m_queue;
#endif
    std::mutex m_mutex;
    std::condition_variable m_cv;
 
    // Condition variable to wake up blocked producers when space is available
    std::condition_variable m_cvNotFull;
 
    const std::string THREAD_NAME;
 
    // Max queue size for back pressure (0 = unlimited)
    const size_t MAX_QUEUE_SIZE;
 
    // Promise and future to synchronize thread start (constructed lazily in CreateThread)
    std::optional<std::promise<void>> m_threadStartPromise;
    std::optional<std::future<void>> m_threadStartFuture;
 
    std::atomic<bool> m_exit;
 
    // Watchdog related members
    std::atomic<dmq::TimePoint> m_lastAliveTime;
    std::unique_ptr<Timer> m_watchdogTimer;
    dmq::ScopedConnection m_watchdogTimerConn;
    std::unique_ptr<Timer> m_threadTimer;
    dmq::ScopedConnection m_threadTimerConn;
    std::atomic<dmq::Duration> m_watchdogTimeout;
};
 
#endif