juce/source/src/threads/juce_ThreadPool.cpp

   1 /*
   2   ==============================================================================
   3
   4    This file is part of the JUCE library - "Jules' Utility Class Extensions"
   5    Copyright 2004-11 by Raw Material Software Ltd.
   6
   7   ------------------------------------------------------------------------------
   8
   9    JUCE can be redistributed and/or modified under the terms of the GNU General
  10    Public License (Version 2), as published by the Free Software Foundation.
  11    A copy of the license is included in the JUCE distribution, or can be found
  12    online at www.gnu.org/licenses.
  13
  14    JUCE is distributed in the hope that it will be useful, but WITHOUT ANY
  15    WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
  16    A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
  17
  18   ------------------------------------------------------------------------------
  19
  20    To release a closed-source product which uses JUCE, commercial licenses are
  21    available: visit www.rawmaterialsoftware.com/juce for more information.
  22
  23   ==============================================================================
  24 */
  25
  26 #include "../core/juce_StandardHeader.h"
  27
  28 BEGIN_JUCE_NAMESPACE
  29
  30 #include "juce_ThreadPool.h"
  31 #include "../core/juce_Time.h"
  32
  33
  34 //==============================================================================
  35 ThreadPoolJob::ThreadPoolJob (const String& name)
  36     : jobName (name),
  37       pool (nullptr),
  38       shouldStop (false),
  39       isActive (false),
  40       shouldBeDeleted (false)
  41 {
  42 }
  43
  44 ThreadPoolJob::~ThreadPoolJob()
  45 {
  46     // you mustn't delete a job while it's still in a pool! Use ThreadPool::removeJob()
  47     // to remove it first!
  48     jassert (pool == nullptr || ! pool->contains (this));
  49 }
  50
  51 String ThreadPoolJob::getJobName() const
  52 {
  53     return jobName;
  54 }
  55
  56 void ThreadPoolJob::setJobName (const String& newName)
  57 {
  58     jobName = newName;
  59 }
  60
  61 void ThreadPoolJob::signalJobShouldExit()
  62 {
  63     shouldStop = true;
  64 }
  65
  66
  67 //==============================================================================
  68 class ThreadPool::ThreadPoolThread  : public Thread
  69 {
  70 public:
  71     ThreadPoolThread (ThreadPool& pool_)
  72         : Thread ("Pool"),
  73           pool (pool_),
  74           busy (false)
  75     {
  76     }
  77
  78     void run()
  79     {
  80         while (! threadShouldExit())
  81         {
  82             if (! pool.runNextJob())
  83                 wait (500);
  84         }
  85     }
  86
  87 private:
  88     ThreadPool& pool;
  89     bool volatile busy;
  90
  91     JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR (ThreadPoolThread);
  92 };
  93
  94 //==============================================================================
  95 ThreadPool::ThreadPool (const int numThreads,
  96                         const bool startThreadsOnlyWhenNeeded,
  97                         const int stopThreadsWhenNotUsedTimeoutMs)
  98     : threadStopTimeout (stopThreadsWhenNotUsedTimeoutMs),
  99       priority (5)
 100 {
 101     jassert (numThreads > 0); // not much point having one of these with no threads in it.
 102
 103     for (int i = jmax (1, numThreads); --i >= 0;)
 104         threads.add (new ThreadPoolThread (*this));
 105
 106     if (! startThreadsOnlyWhenNeeded)
 107         for (int i = threads.size(); --i >= 0;)
 108             threads.getUnchecked(i)->startThread (priority);
 109 }
 110
 111 ThreadPool::~ThreadPool()
 112 {
 113     removeAllJobs (true, 4000);
 114
 115     int i;
 116     for (i = threads.size(); --i >= 0;)
 117         threads.getUnchecked(i)->signalThreadShouldExit();
 118
 119     for (i = threads.size(); --i >= 0;)
 120         threads.getUnchecked(i)->stopThread (500);
 121 }
 122
 123 void ThreadPool::addJob (ThreadPoolJob* const job)
 124 {
 125     jassert (job != nullptr);
 126     jassert (job->pool == nullptr);
 127
 128     if (job->pool == nullptr)
 129     {
 130         job->pool = this;
 131         job->shouldStop = false;
 132         job->isActive = false;
 133
 134         {
 135             const ScopedLock sl (lock);
 136             jobs.add (job);
 137
 138             int numRunning = 0;
 139
 140             for (int i = threads.size(); --i >= 0;)
 141                 if (threads.getUnchecked(i)->isThreadRunning() && ! threads.getUnchecked(i)->threadShouldExit())
 142                     ++numRunning;
 143
 144             if (numRunning < threads.size())
 145             {
 146                 bool startedOne = false;
 147                 int n = 1000;
 148
 149                 while (--n >= 0 && ! startedOne)
 150                 {
 151                     for (int i = threads.size(); --i >= 0;)
 152                     {
 153                         if (! threads.getUnchecked(i)->isThreadRunning())
 154                         {
 155                             threads.getUnchecked(i)->startThread (priority);
 156                             startedOne = true;
 157                             break;
 158                         }
 159                     }
 160
 161                     if (! startedOne)
 162                         Thread::sleep (2);
 163                 }
 164             }
 165         }
 166
 167         for (int i = threads.size(); --i >= 0;)
 168             threads.getUnchecked(i)->notify();
 169     }
 170 }
 171
 172 int ThreadPool::getNumJobs() const
 173 {
 174     return jobs.size();
 175 }
 176
 177 ThreadPoolJob* ThreadPool::getJob (const int index) const
 178 {
 179     const ScopedLock sl (lock);
 180     return jobs [index];
 181 }
 182
 183 bool ThreadPool::contains (const ThreadPoolJob* const job) const
 184 {
 185     const ScopedLock sl (lock);
 186     return jobs.contains (const_cast <ThreadPoolJob*> (job));
 187 }
 188
 189 bool ThreadPool::isJobRunning (const ThreadPoolJob* const job) const
 190 {
 191     const ScopedLock sl (lock);
 192     return jobs.contains (const_cast <ThreadPoolJob*> (job)) && job->isActive;
 193 }
 194
 195 bool ThreadPool::waitForJobToFinish (const ThreadPoolJob* const job,
 196                                      const int timeOutMs) const
 197 {
 198     if (job != nullptr)
 199     {
 200         const uint32 start = Time::getMillisecondCounter();
 201
 202         while (contains (job))
 203         {
 204             if (timeOutMs >= 0 && Time::getMillisecondCounter() >= start + timeOutMs)
 205                 return false;
 206
 207             jobFinishedSignal.wait (2);
 208         }
 209     }
 210
 211     return true;
 212 }
 213
 214 bool ThreadPool::removeJob (ThreadPoolJob* const job,
 215                             const bool interruptIfRunning,
 216                             const int timeOutMs)
 217 {
 218     bool dontWait = true;
 219
 220     if (job != nullptr)
 221     {
 222         const ScopedLock sl (lock);
 223
 224         if (jobs.contains (job))
 225         {
 226             if (job->isActive)
 227             {
 228                 if (interruptIfRunning)
 229                     job->signalJobShouldExit();
 230
 231                 dontWait = false;
 232             }
 233             else
 234             {
 235                 jobs.removeValue (job);
 236                 job->pool = nullptr;
 237             }
 238         }
 239     }
 240
 241     return dontWait || waitForJobToFinish (job, timeOutMs);
 242 }
 243
 244 bool ThreadPool::removeAllJobs (const bool interruptRunningJobs,
 245                                 const int timeOutMs,
 246                                 const bool deleteInactiveJobs,
 247                                 ThreadPool::JobSelector* selectedJobsToRemove)
 248 {
 249     Array <ThreadPoolJob*> jobsToWaitFor;
 250
 251     {
 252         const ScopedLock sl (lock);
 253
 254         for (int i = jobs.size(); --i >= 0;)
 255         {
 256             ThreadPoolJob* const job = jobs.getUnchecked(i);
 257
 258             if (selectedJobsToRemove == nullptr || selectedJobsToRemove->isJobSuitable (job))
 259             {
 260                 if (job->isActive)
 261                 {
 262                     jobsToWaitFor.add (job);
 263
 264                     if (interruptRunningJobs)
 265                         job->signalJobShouldExit();
 266                 }
 267                 else
 268                 {
 269                     jobs.remove (i);
 270
 271                     if (deleteInactiveJobs)
 272                         delete job;
 273                     else
 274                         job->pool = nullptr;
 275                 }
 276             }
 277         }
 278     }
 279
 280     const uint32 start = Time::getMillisecondCounter();
 281
 282     for (;;)
 283     {
 284         for (int i = jobsToWaitFor.size(); --i >= 0;)
 285             if (! isJobRunning (jobsToWaitFor.getUnchecked (i)))
 286                 jobsToWaitFor.remove (i);
 287
 288         if (jobsToWaitFor.size() == 0)
 289             break;
 290
 291         if (timeOutMs >= 0 && Time::getMillisecondCounter() >= start + timeOutMs)
 292             return false;
 293
 294         jobFinishedSignal.wait (20);
 295     }
 296
 297     return true;
 298 }
 299
 300 StringArray ThreadPool::getNamesOfAllJobs (const bool onlyReturnActiveJobs) const
 301 {
 302     StringArray s;
 303     const ScopedLock sl (lock);
 304
 305     for (int i = 0; i < jobs.size(); ++i)
 306     {
 307         const ThreadPoolJob* const job = jobs.getUnchecked(i);
 308         if (job->isActive || ! onlyReturnActiveJobs)
 309             s.add (job->getJobName());
 310     }
 311
 312     return s;
 313 }
 314
 315 bool ThreadPool::setThreadPriorities (const int newPriority)
 316 {
 317     bool ok = true;
 318
 319     if (priority != newPriority)
 320     {
 321         priority = newPriority;
 322
 323         for (int i = threads.size(); --i >= 0;)
 324             if (! threads.getUnchecked(i)->setPriority (newPriority))
 325                 ok = false;
 326     }
 327
 328     return ok;
 329 }
 330
 331 bool ThreadPool::runNextJob()
 332 {
 333     ThreadPoolJob* job = nullptr;
 334
 335     {
 336         const ScopedLock sl (lock);
 337
 338         for (int i = 0; i < jobs.size(); ++i)
 339         {
 340             job = jobs[i];
 341
 342             if (job != nullptr && ! (job->isActive || job->shouldStop))
 343                 break;
 344
 345             job = nullptr;
 346         }
 347
 348         if (job != nullptr)
 349             job->isActive = true;
 350
 351     }
 352
 353     if (job != nullptr)
 354     {
 355         JUCE_TRY
 356         {
 357             ThreadPoolJob::JobStatus result = job->runJob();
 358
 359             lastJobEndTime = Time::getApproximateMillisecondCounter();
 360
 361             const ScopedLock sl (lock);
 362
 363             if (jobs.contains (job))
 364             {
 365                 job->isActive = false;
 366
 367                 if (result != ThreadPoolJob::jobNeedsRunningAgain || job->shouldStop)
 368                 {
 369                     job->pool = nullptr;
 370                     job->shouldStop = true;
 371                     jobs.removeValue (job);
 372
 373                     if (result == ThreadPoolJob::jobHasFinishedAndShouldBeDeleted)
 374                         delete job;
 375
 376                     jobFinishedSignal.signal();
 377                 }
 378                 else
 379                 {
 380                     // move the job to the end of the queue if it wants another go
 381                     jobs.move (jobs.indexOf (job), -1);
 382                 }
 383             }
 384         }
 385 #if JUCE_CATCH_UNHANDLED_EXCEPTIONS
 386         catch (...)
 387         {
 388             const ScopedLock sl (lock);
 389             jobs.removeValue (job);
 390         }
 391 #endif
 392     }
 393     else
 394     {
 395         if (threadStopTimeout > 0
 396              && Time::getApproximateMillisecondCounter() > lastJobEndTime + threadStopTimeout)
 397         {
 398             const ScopedLock sl (lock);
 399
 400             if (jobs.size() == 0)
 401                 for (int i = threads.size(); --i >= 0;)
 402                     threads.getUnchecked(i)->signalThreadShouldExit();
 403         }
 404         else
 405         {
 406             return false;
 407         }
 408     }
 409
 410     return true;
 411 }
 412
 413 END_JUCE_NAMESPACE