JobSystemThreadPool.h

Go to the documentation of this file.

// Jolt Physics Library (https://github.com/jrouwe/JoltPhysics)
// SPDX-FileCopyrightText: 2021 Jorrit Rouwe
// SPDX-License-Identifier: MIT
 
#pragma once
 
#include <Jolt/Core/JobSystemWithBarrier.h>
#include <Jolt/Core/FixedSizeFreeList.h>
#include <Jolt/Core/Semaphore.h>
 
JPH_SUPPRESS_WARNINGS_STD_BEGIN
#include <thread>
JPH_SUPPRESS_WARNINGS_STD_END
 
JPH_NAMESPACE_BEGIN
 
// Things we're using from STL
using std::thread;

class JPH_EXPORT JobSystemThreadPool final : public JobSystemWithBarrier
{
public:
    JPH_OVERRIDE_NEW_DELETE

                            JobSystemThreadPool(uint inMaxJobs, uint inMaxBarriers, int inNumThreads = -1);
                            JobSystemThreadPool() = default;
    virtual                 ~JobSystemThreadPool() override;

    using InitExitFunction = function<void(int)>;
    void                    SetThreadInitFunction(const InitExitFunction &inInitFunction)   { mThreadInitFunction = inInitFunction; }
    void                    SetThreadExitFunction(const InitExitFunction &inExitFunction)   { mThreadExitFunction = inExitFunction; }

    void                    Init(uint inMaxJobs, uint inMaxBarriers, int inNumThreads = -1);
 
    // See JobSystem
    virtual int             GetMaxConcurrency() const override              { return int(mThreads.size()) + 1; }
    virtual JobHandle       CreateJob(const char *inName, ColorArg inColor, const JobFunction &inJobFunction, uint32 inNumDependencies = 0) override;

    void                    SetNumThreads(int inNumThreads)                 { StopThreads(); StartThreads(inNumThreads); }
 
protected:
    // See JobSystem
    virtual void            QueueJob(Job *inJob) override;
    virtual void            QueueJobs(Job **inJobs, uint inNumJobs) override;
    virtual void            FreeJob(Job *inJob) override;
 
private:
    void                    StartThreads(int inNumThreads);
    void                    StopThreads();

    void                    ThreadMain(int inThreadIndex);

    inline uint             GetHead() const;

    inline void             QueueJobInternal(Job *inJob);

    InitExitFunction        mThreadInitFunction = [](int) { };
    InitExitFunction        mThreadExitFunction = [](int) { };

    using AvailableJobs = FixedSizeFreeList<Job>;
    AvailableJobs           mJobs;

    Array<thread>           mThreads;
 
    // The job queue
    static constexpr uint32 cQueueLength = 1024;
    static_assert(IsPowerOf2(cQueueLength));                                // We do bit operations and require queue length to be a power of 2
    atomic<Job *>           mQueue[cQueueLength];
 
    // Head and tail of the queue, do this value modulo cQueueLength - 1 to get the element in the mQueue array
    atomic<uint> *          mHeads = nullptr;                               
    alignas(JPH_CACHE_LINE_SIZE) atomic<uint> mTail = 0;                    
 
    // Semaphore used to signal worker threads that there is new work
    Semaphore               mSemaphore;

    atomic<bool>            mQuit = false;
};
 
JPH_NAMESPACE_END