src/AsyncFramework/JobScheduler.cpp

   1 /***************************************************************************
   2  *   Copyright (C) 2009-2011 by Stefan Fuhrmann                            *
   3  *   stefanfuhrmann@alice-dsl.de                                           *
   4  *                                                                         *
   5  *   This program is free software; you can redistribute it and/or modify  *
   6  *   it under the terms of the GNU General Public License as published by  *
   7  *   the Free Software Foundation; either version 2 of the License, or     *
   8  *   (at your option) any later version.                                   *
   9  *                                                                         *
  10  *   This program is distributed in the hope that it will be useful,       *
  11  *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
  12  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
  13  *   GNU General Public License for more details.                          *
  14  *                                                                         *
  15  *   You should have received a copy of the GNU General Public License     *
  16  *   along with this program; if not, write to the                         *
  17  *   Free Software Foundation, Inc.,                                       *
  18  *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
  19  ***************************************************************************/
  20
  21 #include "stdafx.h"
  22 #include "JobScheduler.h"
  23 #include "IJob.h"
  24 #include "Thread.h"
  25
  26 namespace async
  27 {
  28
  29 // queue size management
  30
  31 void CJobScheduler::CQueue::Grow (size_t newSize)
  32 {
  33     TJob* newData = new TJob[newSize];
  34
  35     size_t count = size();
  36     if (first)
  37         memmove (newData, first, count * sizeof (TJob[1]));
  38     delete[] data;
  39
  40     data = newData;
  41     end = newData + newSize;
  42     first = newData;
  43     last = newData + count;
  44 }
  45
  46 void CJobScheduler::CQueue::AutoGrow()
  47 {
  48     size_t count = size();
  49
  50     if (count * 2 <= capacity())
  51     {
  52         memmove (data, first, count * sizeof (TJob[1]));
  53         first = data;
  54         last = data + count;
  55     }
  56     else
  57     {
  58         Grow (capacity() * 2);
  59     }
  60 }
  61
  62 // remove one entry from \ref starving container.
  63 // Return NULL, if container was empty
  64
  65 CJobScheduler* CJobScheduler::CThreadPool::SelectStarving()
  66 {
  67     CCriticalSectionLock lock (mutex);
  68     if (starving.empty() || pool.empty())
  69         return NULL;
  70
  71     std::vector<CJobScheduler*>::iterator begin = starving.begin();
  72
  73     CJobScheduler* scheduler = *begin;
  74     starving.erase (begin);
  75
  76     return scheduler;
  77 }
  78
  79 // create empty thread pool
  80
  81 CJobScheduler::CThreadPool::CThreadPool()
  82     : yetToCreate (0)
  83     , allocCount (0)
  84     , maxCount (0)
  85 {
  86 }
  87
  88 // release threads
  89
  90 CJobScheduler::CThreadPool::~CThreadPool()
  91 {
  92     SetThreadCount(0);
  93 }
  94
  95 // Meyer's singleton
  96
  97 CJobScheduler::CThreadPool& CJobScheduler::CThreadPool::GetInstance()
  98 {
  99     static CThreadPool instance;
 100     return instance;
 101 }
 102
 103 // pool interface
 104
 105 CJobScheduler::SThreadInfo* CJobScheduler::CThreadPool::TryAlloc()
 106 {
 107     CCriticalSectionLock lock (mutex);
 108     if (pool.empty())
 109     {
 110         if (yetToCreate == 0)
 111             return NULL;
 112
 113         // lazy thread creation
 114
 115         SThreadInfo* info = new SThreadInfo;
 116         info->owner = NULL;
 117         info->thread = new CThread (&ThreadFunc, info, true);
 118
 119         --yetToCreate;
 120         ++allocCount;
 121
 122         return info;
 123     }
 124
 125     CJobScheduler::SThreadInfo* thread = pool.back();
 126     pool.pop_back();
 127     ++allocCount;
 128
 129     return thread;
 130 }
 131
 132 void CJobScheduler::CThreadPool::Release (SThreadInfo* thread)
 133 {
 134     {
 135         // put back into pool, unless its capacity has been exceeded
 136
 137         CCriticalSectionLock lock (mutex);
 138         if (pool.size() + --allocCount < maxCount)
 139         {
 140             pool.push_back (thread);
 141
 142             // shortcut
 143
 144             if (starving.empty())
 145                 return;
 146         }
 147         else
 148         {
 149             // pool overflow -> terminate thread
 150
 151             delete thread->thread;
 152             delete thread;
 153         }
 154     }
 155
 156     // notify starved schedulers that there may be some threads to alloc
 157
 158     for ( CJobScheduler* scheduler = SelectStarving()
 159         ; scheduler != NULL
 160         ; scheduler = SelectStarving())
 161     {
 162         scheduler->ThreadAvailable();
 163     }
 164 }
 165
 166 // set max. number of concurrent threads
 167
 168 void CJobScheduler::CThreadPool::SetThreadCount (size_t count)
 169 {
 170     CCriticalSectionLock lock (mutex);
 171
 172     maxCount = count;
 173     if (pool.size() + allocCount < maxCount)
 174         yetToCreate = maxCount - pool.size() + allocCount;
 175
 176     while ((pool.size() + allocCount > maxCount) && !pool.empty())
 177     {
 178         SThreadInfo* info = pool.back();
 179         pool.pop_back();
 180
 181         delete info->thread;
 182         delete info;
 183     }
 184 }
 185
 186 size_t CJobScheduler::CThreadPool::GetThreadCount() const
 187 {
 188     return maxCount;
 189 }
 190
 191 // manage starving schedulers
 192
 193 void CJobScheduler::CThreadPool::AddStarving (CJobScheduler* scheduler)
 194 {
 195     CCriticalSectionLock lock (mutex);
 196     starving.push_back (scheduler);
 197 }
 198
 199 bool CJobScheduler::CThreadPool::RemoveStarving (CJobScheduler* scheduler)
 200 {
 201     CCriticalSectionLock lock (mutex);
 202
 203     typedef std::vector<CJobScheduler*>::iterator TI;
 204     TI begin = starving.begin();
 205     TI end = starving.end();
 206
 207     TI newEnd = std::remove_copy (begin, end, begin, scheduler);
 208     if (newEnd == end)
 209         return false;
 210
 211     starving.erase (newEnd, end);
 212     return true;
 213 }
 214
 215 // job execution helpers
 216
 217 CJobScheduler::TJob CJobScheduler::AssignJob (SThreadInfo* info)
 218 {
 219     CCriticalSectionLock lock (mutex);
 220
 221     // wake up threads that waited for some work to finish
 222
 223     if (waitingThreads > 0)
 224         threadIsIdle.Set();
 225
 226     // suspend this thread if there is no work left
 227
 228     if (queue.empty() || (threads.stopCount != 0))
 229     {
 230         // suspend
 231
 232         info->thread->Suspend();
 233
 234         // remove from "running" list and put it back either to
 235         // "suspended" pool or global shared pool
 236
 237         bool terminateThread = false;
 238         for (size_t i = 0, count = threads.running.size(); i < count; ++i)
 239             if (threads.running[i] == info)
 240             {
 241                 threads.running[i] = threads.running[count-1];
 242                 threads.running.pop_back();
 243
 244                 // still in debt of shared pool?
 245
 246                 if (threads.fromShared > 0)
 247                 {
 248                     // if we are actually idle, we aren't starving anymore
 249
 250                     StopStarvation();
 251
 252                     // put back to global pool
 253
 254                     info->owner = NULL;
 255                     CThreadPool::GetInstance().Release (info);
 256
 257                     --threads.fromShared;
 258                 }
 259                 else if (aggressiveThreading)
 260                 {
 261                     // don't keep private idle threads
 262
 263                     delete info;
 264                     ++threads.yetToCreate;
 265                     terminateThread = true;
 266                 }
 267                 else
 268                 {
 269
 270                     // add to local pool
 271
 272                     threads.suspended.push_back (info);
 273                     ++threads.suspendedCount;
 274                 }
 275
 276                 // signal empty queue, if necessary
 277
 278                 ++threads.unusedCount;
 279                 if (--threads.runningCount == 0)
 280                     emptyEvent.Set();
 281
 282                 break;
 283             }
 284
 285         return TJob ((IJob*)NULL, terminateThread);
 286     }
 287
 288     // extract one job
 289
 290     return queue.pop();
 291 }
 292
 293 // try to get a thread from the shared pool.
 294 // register as "starving" if that failed
 295
 296 bool CJobScheduler::AllocateSharedThread()
 297 {
 298     if (!threads.starved)
 299     {
 300         SThreadInfo* info = CThreadPool::GetInstance().TryAlloc();
 301         if (info != NULL)
 302         {
 303             ++threads.fromShared;
 304             --threads.unusedCount;
 305             ++threads.runningCount;
 306
 307             threads.running.push_back (info);
 308             info->owner = this;
 309             info->thread->Resume();
 310
 311             return true;
 312         }
 313         else
 314         {
 315             threads.starved = true;
 316             CThreadPool::GetInstance().AddStarving (this);
 317         }
 318     }
 319
 320     return false;
 321 }
 322
 323 bool CJobScheduler::AllocateThread()
 324 {
 325     if (threads.suspendedCount > 0)
 326     {
 327         // recycle suspended, private thread
 328
 329         SThreadInfo* info = threads.suspended.back();
 330         threads.suspended.pop_back();
 331
 332         --threads.suspendedCount;
 333         --threads.unusedCount;
 334         ++threads.runningCount;
 335
 336         threads.running.push_back (info);
 337         info->thread->Resume();
 338     }
 339     else if (threads.yetToCreate > 0)
 340     {
 341         // time to start a new private thread
 342
 343         --threads.yetToCreate;
 344
 345         --threads.unusedCount;
 346         ++threads.runningCount;
 347
 348         SThreadInfo* info = new SThreadInfo;
 349         info->owner = this;
 350         info->thread = new CThread (&ThreadFunc, info, true);
 351         threads.running.push_back (info);
 352
 353         info->thread->Resume();
 354     }
 355     else
 356     {
 357         // try to allocate a shared thread
 358
 359         if (threads.fromShared < threads.maxFromShared)
 360             return AllocateSharedThread();
 361
 362         return false;
 363     }
 364
 365     return true;
 366 }
 367
 368 // unregister from "starving" list.
 369 // This may race with CThreadPool::Release -> must loop here.
 370
 371 void CJobScheduler::StopStarvation()
 372 {
 373     while (threads.starved)
 374         if (CThreadPool::GetInstance().RemoveStarving (this))
 375         {
 376             threads.starved = false;
 377             break;
 378         }
 379 }
 380
 381 // worker thread function
 382
 383 bool CJobScheduler::ThreadFunc (void* arg)
 384 {
 385     SThreadInfo* info = reinterpret_cast<SThreadInfo*>(arg);
 386
 387     TJob job = info->owner->AssignJob (info);
 388     if (job.first != NULL)
 389     {
 390         // run the job
 391
 392         job.first->Execute();
 393
 394         // it is no longer referenced by the scheduler
 395
 396         job.first->OnUnSchedule (info->owner);
 397
 398         // is it our responsibility to clean up this job?
 399
 400         if (job.second)
 401             delete job.first;
 402
 403         // continue
 404
 405         return false;
 406     }
 407     else
 408     {
 409         // maybe, auto-delete thread object
 410
 411         return job.second;
 412     }
 413 }
 414
 415 // Create & remove threads
 416
 417 CJobScheduler::CJobScheduler
 418     ( size_t threadCount
 419     , size_t sharedThreads
 420     , bool aggressiveThreading
 421     , bool fifo)
 422     : waitingThreads (0)
 423     , aggressiveThreading (aggressiveThreading)
 424 {
 425     threads.runningCount = 0;
 426     threads.suspendedCount = 0;
 427
 428     threads.fromShared = 0;
 429     threads.maxFromShared = sharedThreads;
 430
 431     threads.unusedCount = threadCount + sharedThreads;
 432     threads.yetToCreate = threadCount;
 433
 434     threads.starved = false;
 435     threads.stopCount = 0;
 436
 437     queue.set_fifo (fifo);
 438
 439     // auto-initialize shared threads
 440
 441     if (GetSharedThreadCount() == 0)
 442         UseAllCPUs();
 443 }
 444
 445 CJobScheduler::~CJobScheduler(void)
 446 {
 447     StopStarvation();
 448     WaitForEmptyQueue();
 449
 450     assert (threads.running.empty());
 451     assert (threads.fromShared == 0);
 452
 453     for (size_t i = 0, count = threads.suspended.size(); i < count; ++i)
 454     {
 455         SThreadInfo* info = threads.suspended[i];
 456         delete info->thread;
 457         delete info;
 458     }
 459 }
 460
 461 // Meyer's singleton
 462
 463 CJobScheduler* CJobScheduler::GetDefault()
 464 {
 465     static CJobScheduler instance (0, SIZE_MAX);
 466     return &instance;
 467 }
 468
 469 // job management:
 470 // add new job
 471
 472 void CJobScheduler::Schedule (IJob* job, bool transferOwnership)
 473 {
 474     TJob toAdd (job, transferOwnership);
 475     job->OnSchedule (this);
 476
 477     CCriticalSectionLock lock (mutex);
 478
 479     if (queue.empty())
 480         emptyEvent.Reset();
 481
 482     queue.push (toAdd);
 483     if (threads.stopCount != 0)
 484         return;
 485
 486     bool addThread = aggressiveThreading
 487                  || (   (queue.size() > 2 * threads.runningCount)
 488                      && (threads.unusedCount > 0));
 489
 490     if (addThread)
 491         AllocateThread();
 492 }
 493
 494 // notification that a new thread may be available
 495
 496 void CJobScheduler::ThreadAvailable()
 497 {
 498     CCriticalSectionLock lock (mutex);
 499     threads.starved = false;
 500
 501     while (   (threads.stopCount != 0)
 502            && (queue.size() > 2 * threads.runningCount)
 503            && (threads.suspendedCount == 0)
 504            && (threads.yetToCreate == 0)
 505            && (threads.fromShared < threads.maxFromShared))
 506     {
 507          if (!AllocateSharedThread())
 508             return;
 509     }
 510 }
 511
 512 // wait for all current and follow-up jobs to terminate
 513
 514 void CJobScheduler::WaitForEmptyQueue()
 515 {
 516     WaitForEmptyQueueOrTimeout(INFINITE);
 517 }
 518
 519 bool CJobScheduler::WaitForEmptyQueueOrTimeout(DWORD milliSeconds)
 520 {
 521     for (;;)
 522     {
 523         {
 524             CCriticalSectionLock lock (mutex);
 525
 526             // if the scheduler has been stopped, we need to remove
 527             // the waiting jobs manually
 528
 529             if (threads.stopCount != 0)
 530                 while (!queue.empty())
 531                 {
 532                     const TJob& job = queue.pop();
 533
 534                     job.first->OnUnSchedule(this);
 535                     if (job.second)
 536                         delete job.first;
 537                 }
 538
 539             // empty queue and no jobs still being processed?
 540
 541             if ((threads.runningCount == 0) && queue.empty())
 542                 return true;
 543
 544             // we will be woken up as soon as both containers are empty
 545
 546             emptyEvent.Reset();
 547         }
 548
 549         if (!emptyEvent.WaitForEndOrTimeout(milliSeconds))
 550             return false;
 551     }
 552     //return true;
 553 }
 554
 555 // wait for some jobs to be finished.
 556
 557 void CJobScheduler::WaitForSomeJobs()
 558 {
 559     {
 560         CCriticalSectionLock lock (mutex);
 561         if (  static_cast<size_t>(InterlockedIncrement (&waitingThreads))
 562             < threads.runningCount)
 563         {
 564             // there are enough running job threads left
 565             // -> wait for one of them to run idle *or*
 566             //    for too many of them to enter this method
 567
 568             threadIsIdle.Reset();
 569         }
 570         else
 571         {
 572             // number of waiting jobs nor equal to or greater than
 573             // the number of running job threads
 574             // -> wake them all
 575
 576             threadIsIdle.Set();
 577             InterlockedDecrement (&waitingThreads);
 578
 579             return;
 580         }
 581     }
 582
 583     threadIsIdle.WaitFor();
 584     InterlockedDecrement (&waitingThreads);
 585 }
 586
 587 // Returns the number of jobs waiting for execution.
 588
 589 size_t CJobScheduler::GetQueueDepth() const
 590 {
 591     CCriticalSectionLock lock (mutex);
 592     return queue.size();
 593 }
 594
 595 // Returns the number of threads that currently execute
 596 // jobs for this scheduler
 597
 598 size_t CJobScheduler::GetRunningThreadCount() const
 599 {
 600     CCriticalSectionLock lock (mutex);
 601     return threads.runningCount;
 602 }
 603
 604 // remove waiting entries from the queue until their
 605 // number drops to or below the given watermark.
 606
 607 std::vector<IJob*> CJobScheduler::RemoveJobFromQueue
 608     ( size_t watermark
 609     , bool oldest)
 610 {
 611     std::vector<IJob*> removed;
 612
 613     {
 614         CCriticalSectionLock lock (mutex);
 615         if (queue.size() > watermark)
 616         {
 617             size_t toRemove = queue.size() - watermark;
 618             removed.reserve (toRemove);
 619
 620             // temporarily change the queue extraction strategy
 621             // such that we remove jobs from the requested end
 622             // (fifo -> oldest are at front, otherwise they are
 623             // at the back)
 624
 625             bool fifo = queue.get_fifo();
 626             queue.set_fifo (oldest == fifo);
 627
 628             // remove 'em
 629
 630             for (size_t i = 0; i < toRemove; ++i)
 631             {
 632                 IJob* job = queue.pop().first;
 633                 job->OnUnSchedule (this);
 634
 635                 removed.push_back (job);
 636             }
 637
 638             // restore job execution order
 639
 640             queue.set_fifo (fifo);
 641         }
 642     }
 643
 644     return removed;
 645 }
 646
 647 long CJobScheduler::Stop()
 648 {
 649     CCriticalSectionLock lock (mutex);
 650     return ++threads.stopCount;
 651 }
 652
 653 long CJobScheduler::Resume()
 654 {
 655     CCriticalSectionLock lock (mutex);
 656     assert (threads.stopCount > 0);
 657
 658     if (--threads.stopCount == 0)
 659     {
 660         while (   (    (queue.size() > threads.runningCount)
 661                     && aggressiveThreading)
 662                || (    (queue.size() > 4 * threads.runningCount)
 663                     && (threads.unusedCount > 0)))
 664         {
 665             if (!AllocateThread())
 666                 break;
 667         }
 668     }
 669
 670     return threads.stopCount;
 671 }
 672
 673
 674
 675 // set max. number of concurrent threads
 676
 677 void CJobScheduler::SetSharedThreadCount (size_t count)
 678 {
 679     CThreadPool::GetInstance().SetThreadCount(count);
 680 }
 681
 682 size_t CJobScheduler::GetSharedThreadCount()
 683 {
 684     return CThreadPool::GetInstance().GetThreadCount();
 685 }
 686
 687 void CJobScheduler::UseAllCPUs()
 688 {
 689     SetSharedThreadCount (GetHWThreadCount());
 690 }
 691
 692 size_t CJobScheduler::GetHWThreadCount()
 693 {
 694 #ifdef _WIN32
 695     SYSTEM_INFO si;
 696     GetSystemInfo(&si);
 697
 698     size_t sysNumProcs = si.dwNumberOfProcessors;
 699 #else
 700     size_t sysNumProcs = sysconf (_SC_NPROCESSORS_CONF);
 701 #endif
 702
 703     return sysNumProcs;
 704 }
 705
 706 }