29 :
Thread (
"Pool", stackSize), pool (p)
36 if (! pool.runNextJob (*
this))
40 std::atomic<ThreadPoolJob*> currentJob {
nullptr };
43 JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR (ThreadPoolThread)
55 jassert (pool ==
nullptr || ! pool->
contains (
this));
76 listeners.add (listener);
81 listeners.remove (listener);
87 return t->currentJob.load();
95 jassert (numThreads > 0);
97 createThreads (numThreads, threadStackSize);
111void ThreadPool::createThreads (
int numThreads,
size_t threadStackSize)
113 for (
int i = jmax (1, numThreads); --i >= 0;)
114 threads.add (
new ThreadPoolThread (*
this, threadStackSize));
116 for (
auto* t : threads)
120void ThreadPool::stopThreads()
122 for (
auto* t : threads)
123 t->signalThreadShouldExit();
125 for (
auto* t : threads)
131 jassert (job !=
nullptr);
132 jassert (job->pool ==
nullptr);
134 if (job->pool ==
nullptr)
137 job->shouldStop =
false;
138 job->isActive =
false;
139 job->shouldBeDeleted = deleteJobWhenFinished;
146 for (
auto* t : threads)
156 JobStatus runJob()
override {
return job(); }
161 addJob (
new LambdaJobWrapper (jobToRun),
true);
168 LambdaJobWrapper (std::function<
void()> j) :
ThreadPoolJob (
"lambda"), job (j) {}
171 std::function<void()> job;
174 addJob (
new LambdaJobWrapper (jobToRun),
true);
185 return threads.size();
203 return jobs.contains (
const_cast<ThreadPoolJob*
> (job)) && job->isActive;
210 auto index = jobs.indexOf (
const_cast<ThreadPoolJob*
> (job));
212 if (index > 0 && ! job->isActive)
213 jobs.move (index, 0);
227 jobFinishedSignal.
wait (2);
236 bool dontWait =
true;
243 if (jobs.contains (job))
247 if (interruptIfRunning)
254 jobs.removeFirstMatchingValue (job);
255 addToDeleteList (deletionList, job);
274 for (
int i = jobs.size(); --i >= 0;)
276 auto* job = jobs.getUnchecked(i);
278 if (selectedJobsToRemove ==
nullptr || selectedJobsToRemove->
isJobSuitable (job))
282 jobsToWaitFor.
add (job);
284 if (interruptRunningJobs)
285 job->signalJobShouldExit();
290 addToDeleteList (deletionList, job);
301 for (
int i = jobsToWaitFor.
size(); --i >= 0;)
309 if (jobsToWaitFor.
size() == 0)
315 jobFinishedSignal.
wait (20);
326 for (
auto* job : jobs)
327 if (job->isActive || ! onlyReturnActiveJobs)
328 s.
add (job->getJobName());
337 for (
auto* t : threads)
338 if (! t->setPriority (newPriority))
351 for (
int i = 0; i < jobs.size(); ++i)
353 if (
auto* job = jobs[i])
360 addToDeleteList (deletionList, job);
365 job->isActive =
true;
375bool ThreadPool::runNextJob (ThreadPoolThread& thread)
377 if (
auto* job = pickNextJobToRun())
380 thread.currentJob = job;
384 result = job->runJob();
391 thread.currentJob =
nullptr;
393 OwnedArray<ThreadPoolJob> deletionList;
396 const ScopedLock sl (lock);
398 if (jobs.contains (job))
400 job->isActive =
false;
404 jobs.removeFirstMatchingValue (job);
405 addToDeleteList (deletionList, job);
407 jobFinishedSignal.
signal();
412 jobs.move (jobs.indexOf (job), -1);
423void ThreadPool::addToDeleteList (OwnedArray<ThreadPoolJob>& deletionList, ThreadPoolJob* job)
const
425 job->shouldStop =
true;
428 if (job->shouldBeDeleted)
429 deletionList.add (job);
Holds a resizable array of primitive or copy-by-value objects.
ElementType getUnchecked(int index) const
Returns one of the elements in the array, without checking the index passed in.
int size() const noexcept
Returns the current number of elements in the array.
void remove(int indexToRemove)
Removes an element from the array.
void add(const ElementType &newElement)
Appends a new element at the end of the array.
Automatically locks and unlocks a mutex object.
An array designed for holding objects.
A special array for holding a list of strings.
void add(String stringToAdd)
Appends a string at the end of the array.
static int getNumCpus() noexcept
Returns the number of logical CPU cores.
A task that is executed by a ThreadPool object.
void signalJobShouldExit()
Calling this will cause the shouldExit() method to return true, and the job should (if it's been impl...
JobStatus
These are the values that can be returned by the runJob() method.
@ jobHasFinished
indicates that the job has finished and can be removed from the pool.
@ jobNeedsRunningAgain
indicates that the job would like to be called again when a thread is free.
void setJobName(const String &newName)
Changes the job's name.
String getJobName() const
Returns the name of this job.
void addListener(Thread::Listener *)
Add a listener to this thread job which will receive a callback when signalJobShouldExit was called o...
virtual ~ThreadPoolJob()
Destructor.
static ThreadPoolJob * getCurrentThreadPoolJob()
If the calling thread is being invoked inside a runJob() method, this will return the ThreadPoolJob t...
void removeListener(Thread::Listener *)
Removes a listener added with addListener.
ThreadPoolJob(const String &name)
Creates a thread pool job object.
A callback class used when you need to select which ThreadPoolJob objects are suitable for some kind ...
virtual bool isJobSuitable(ThreadPoolJob *job)=0
Should return true if the specified thread matches your criteria for whatever operation that this obj...
A set of threads that will run a list of jobs.
void moveJobToFront(const ThreadPoolJob *jobToMove) noexcept
If the given job is in the queue, this will move it to the front so that it is the next one to be exe...
void addJob(ThreadPoolJob *job, bool deleteJobWhenFinished)
Adds a job to the queue.
int getNumThreads() const noexcept
Returns the number of threads assigned to this thread pool.
ThreadPoolJob * getJob(int index) const noexcept
Returns one of the jobs in the queue.
int getNumJobs() const noexcept
Returns the number of jobs currently running or queued.
bool removeAllJobs(bool interruptRunningJobs, int timeOutMilliseconds, JobSelector *selectedJobsToRemove=nullptr)
Tries to remove all jobs from the pool.
StringArray getNamesOfAllJobs(bool onlyReturnActiveJobs) const
Returns a list of the names of all the jobs currently running or queued.
bool isJobRunning(const ThreadPoolJob *job) const noexcept
Returns true if the given job is currently being run by a thread.
bool setThreadPriorities(int newPriority)
Changes the priority of all the threads.
bool removeJob(ThreadPoolJob *job, bool interruptIfRunning, int timeOutMilliseconds)
Tries to remove a job from the pool.
bool contains(const ThreadPoolJob *job) const noexcept
Returns true if the given job is currently queued or running.
bool waitForJobToFinish(const ThreadPoolJob *job, int timeOutMilliseconds) const
Waits until a job has finished running and has been removed from the pool.
ThreadPool()
Creates a thread pool with one thread per CPU core.
Used to receive callbacks for thread exit calls.
virtual void exitSignalSent()=0
Called if Thread::signalThreadShouldExit was called.
static Thread *JUCE_CALLTYPE getCurrentThread()
Finds the thread object that is currently running.
bool wait(int timeOutMilliseconds) const
Suspends the execution of this thread until either the specified timeout period has elapsed,...
Thread(const String &threadName, size_t threadStackSize=0)
Creates a thread.
bool threadShouldExit() const
Checks whether the thread has been told to stop running.
static uint32 getMillisecondCounter() noexcept
Returns the number of millisecs since a fixed event (usually system startup).
bool wait(int timeOutMilliseconds=-1) const
Suspends the calling thread until the event has been signalled.
void signal() const
Wakes up any threads that are currently waiting on this object.
void run() override
Must be implemented to perform the thread's actual code.