juce_ThreadPool.h

/*
  ==============================================================================
 
   This file is part of the JUCE library.
   Copyright (c) 2022 - Raw Material Software Limited
 
   JUCE is an open source library subject to commercial or open-source
   licensing.
 
   The code included in this file is provided under the terms of the ISC license
   http://www.isc.org/downloads/software-support-policy/isc-license. Permission
   To use, copy, modify, and/or distribute this software for any purpose with or
   without fee is hereby granted provided that the above copyright notice and
   this permission notice appear in all copies.
 
   JUCE IS PROVIDED "AS IS" WITHOUT ANY WARRANTY, AND ALL WARRANTIES, WHETHER
   EXPRESSED OR IMPLIED, INCLUDING MERCHANTABILITY AND FITNESS FOR PURPOSE, ARE
   DISCLAIMED.
 
  ==============================================================================
*/
 
namespace juce
{
 
class ThreadPool;
 
//==============================================================================
class JUCE_API  ThreadPoolJob
{
public:
    //==============================================================================
    explicit ThreadPoolJob (const String& name);

    virtual ~ThreadPoolJob();
 
    //==============================================================================
    String getJobName() const;

    void setJobName (const String& newName);
 
    //==============================================================================
    enum JobStatus
    {
        jobHasFinished = 0,     
 
        jobNeedsRunningAgain    
    };

    virtual JobStatus runJob() = 0;
 
 
    //==============================================================================
    bool isRunning() const noexcept                     { return isActive; }

    bool shouldExit() const noexcept                    { return shouldStop; }

    void signalJobShouldExit();

    void addListener (Thread::Listener*);

    void removeListener (Thread::Listener*);
 
    //==============================================================================
    static ThreadPoolJob* getCurrentThreadPoolJob();
 
    //==============================================================================
private:
    friend class ThreadPool;
    String jobName;
    ThreadPool* pool = nullptr;
    std::atomic<bool> shouldStop { false }, isActive { false }, shouldBeDeleted { false };
    ListenerList<Thread::Listener, Array<Thread::Listener*, CriticalSection>> listeners;
 
    JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR (ThreadPoolJob)
};
 
//==============================================================================
struct ThreadPoolOptions
{
    [[nodiscard]] ThreadPoolOptions withThreadName (String newThreadName) const
    {
        return withMember (*this, &ThreadPoolOptions::threadName, newThreadName);
    }

    [[nodiscard]] ThreadPoolOptions withNumberOfThreads (int newNumberOfThreads) const
    {
        return withMember (*this, &ThreadPoolOptions::numberOfThreads, newNumberOfThreads);
    }

    [[nodiscard]] ThreadPoolOptions withThreadStackSizeBytes (size_t newThreadStackSizeBytes) const
    {
        return withMember (*this, &ThreadPoolOptions::threadStackSizeBytes, newThreadStackSizeBytes);
    }

    [[nodiscard]] ThreadPoolOptions withDesiredThreadPriority (Thread::Priority newDesiredThreadPriority) const
    {
        return withMember (*this, &ThreadPoolOptions::desiredThreadPriority, newDesiredThreadPriority);
    }
 
    String threadName { "Pool" };
    int numberOfThreads { SystemStats::getNumCpus() };
    size_t threadStackSizeBytes { Thread::osDefaultStackSize };
    Thread::Priority desiredThreadPriority { Thread::Priority::normal };
};
 
 
//==============================================================================
class JUCE_API  ThreadPool
{
public:
    using Options = ThreadPoolOptions;
 
    //==============================================================================
    explicit ThreadPool (const Options& options);

    ThreadPool() : ThreadPool { Options{} } {}

    ThreadPool (int numberOfThreads,
                size_t threadStackSizeBytes = Thread::osDefaultStackSize,
                Thread::Priority desiredThreadPriority = Thread::Priority::normal);

    ~ThreadPool();
 
    //==============================================================================
    class JUCE_API  JobSelector
    {
    public:
        virtual ~JobSelector() = default;

        virtual bool isJobSuitable (ThreadPoolJob* job) = 0;
    };
 
    //==============================================================================
    void addJob (ThreadPoolJob* job,
                 bool deleteJobWhenFinished);

    void addJob (std::function<ThreadPoolJob::JobStatus()> job);

    void addJob (std::function<void()> job);

    bool removeJob (ThreadPoolJob* job,
                    bool interruptIfRunning,
                    int timeOutMilliseconds);

    bool removeAllJobs (bool interruptRunningJobs,
                        int timeOutMilliseconds,
                        JobSelector* selectedJobsToRemove = nullptr);

    int getNumJobs() const noexcept;

    int getNumThreads() const noexcept;

    ThreadPoolJob* getJob (int index) const noexcept;

    bool contains (const ThreadPoolJob* job) const noexcept;

    bool isJobRunning (const ThreadPoolJob* job) const noexcept;

    bool waitForJobToFinish (const ThreadPoolJob* job,
                             int timeOutMilliseconds) const;

    void moveJobToFront (const ThreadPoolJob* jobToMove) noexcept;

    StringArray getNamesOfAllJobs (bool onlyReturnActiveJobs) const;
 
private:
    //==============================================================================
    Array<ThreadPoolJob*> jobs;
 
    struct ThreadPoolThread;
    friend class ThreadPoolJob;
    OwnedArray<ThreadPoolThread> threads;
 
    CriticalSection lock;
    WaitableEvent jobFinishedSignal;
 
    bool runNextJob (ThreadPoolThread&);
    ThreadPoolJob* pickNextJobToRun();
    void addToDeleteList (OwnedArray<ThreadPoolJob>&, ThreadPoolJob*) const;
    void stopThreads();
 
    // Note that this method has changed, and no longer has a parameter to indicate
    // whether the jobs should be deleted - see the new method for details.
    void removeAllJobs (bool, int, bool);
 
    JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR (ThreadPool)
};
 
} // namespace juce