WorkerThread.h

Go to the documentation of this file.

 
#pragma once
 
#ifndef __WORKER_THREAD_H_
#define __WORKER_THREAD_H_
 
#include <atomic>
#include <mutex>
#include <thread>
 
namespace Utils
{

template <class ThreadsPool, class Job> class WorkerThread
{
  public:
    explicit WorkerThread(ThreadsPool *threadsPool) : threadsPool(threadsPool), StopFlag(false)
    {
        Thread = std::thread(&WorkerThread::Run, this);
    }

    ~WorkerThread()
    {
        Stop();
    }

    void Start()
    {
        if (Thread.joinable())
            return;
 
        {
            std::lock_guard lock(threadsPool->Mutex);
            StopFlag = false;
        }
        Thread = std::thread(&WorkerThread::Run, this);
    }

    void Stop()
    {
        if (!Thread.joinable())
            return;
 
        {
            std::lock_guard lock(threadsPool->Mutex);
 
            StopFlag = true;
        }
 
        Join();
    }

    void SetStopUnlocked()
    {
        StopFlag = true;
    }

    inline void Join()
    {
        if (Thread.joinable())
            Thread.join();
    }

    void Abandon()
    {
        if (Thread.joinable())
            (void)new std::thread(std::move(Thread));
    }
 
  private:
    inline bool TerminateWait() const
    {
        return threadsPool->HasWork() || StopFlag;
    }

    void Run()
    {
        for (;;)
        {
            std::unique_lock lock(threadsPool->Mutex);
            if (!TerminateWait())
                threadsPool->Condition.wait(lock, [this] { return TerminateWait(); });
 
            while (threadsPool->HasWork() && !StopFlag)
            {
                std::shared_ptr<Job> job = std::move(threadsPool->JobsQueue.front());
                threadsPool->JobsQueue.pop();
                ++threadsPool->Running;
                lock.unlock();
 
                job->DoWork();
                const size_t count = job->GetJobCount();
                // A job can own the pool's owner (execution jobs hold their
                // network). Release it before it counts as finished, so the
                // owner is never destroyed on one of its own workers.
                job.reset();
 
                // Idle before the waiter can see the job finished, so a fork()
                // right after WaitForFinish() finds the pool idle.
                lock.lock();
                --threadsPool->Running;
                {
                    std::lock_guard lockCount(threadsPool->FinishMutex);
                    threadsPool->FinishCount += count;
                }
            }
 
            if (StopFlag)
                break;
            lock.unlock();
            threadsPool->NotifyFinish();
        }
    }
 
    ThreadsPool *threadsPool; 
    std::thread Thread;       
    bool StopFlag;            
};
 
} // namespace Utils
 
#endif // !__WORKER_THREAD_H_