comphelper/source/misc/threadpool.cxx

   1 /* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
   2 /*
   3  * This file is part of the LibreOffice project.
   4  *
   5  * This Source Code Form is subject to the terms of the Mozilla Public
   6  * License, v. 2.0. If a copy of the MPL was not distributed with this
   7  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
   8  */
   9
  10 #include <comphelper/threadpool.hxx>
  11
  12 #include <rtl/instance.hxx>
  13 #include <boost/shared_ptr.hpp>
  14 #include <thread>
  15 #include <algorithm>
  16
  17 namespace comphelper {
  18
  19 class ThreadPool::ThreadWorker : public salhelper::Thread
  20 {
  21     ThreadPool    *mpPool;
  22     osl::Condition maNewWork;
  23     bool           mbWorking;
  24 public:
  25
  26     ThreadWorker( ThreadPool *pPool ) :
  27         salhelper::Thread("thread-pool"),
  28         mpPool( pPool ),
  29         mbWorking( false )
  30     {
  31     }
  32
  33     virtual void execute() SAL_OVERRIDE
  34     {
  35         ThreadTask *pTask;
  36         while ( ( pTask = waitForWork() ) )
  37         {
  38             pTask->doWork();
  39             delete pTask;
  40         }
  41     }
  42
  43     ThreadTask *waitForWork()
  44     {
  45         ThreadTask *pRet = NULL;
  46
  47         osl::ResettableMutexGuard aGuard( mpPool->maGuard );
  48
  49         pRet = mpPool->popWork();
  50
  51         while( !pRet )
  52         {
  53             if (mbWorking)
  54                 mpPool->stopWork();
  55             mbWorking = false;
  56             maNewWork.reset();
  57
  58             if( mpPool->mbTerminate )
  59                 break;
  60
  61             aGuard.clear(); // unlock
  62
  63             maNewWork.wait();
  64
  65             aGuard.reset(); // lock
  66
  67             pRet = mpPool->popWork();
  68         }
  69
  70         if (pRet)
  71         {
  72             if (!mbWorking)
  73                 mpPool->startWork();
  74             mbWorking = true;
  75         }
  76
  77         return pRet;
  78     }
  79
  80     // Why a condition per worker thread - you may ask.
  81     //
  82     // Unfortunately the Windows synchronisation API that we wrap
  83     // is horribly inadequate cf.
  84     //    http://www.cs.wustl.edu/~schmidt/win32-cv-1.html
  85     // The existing osl::Condition API should only ever be used
  86     // between one producer and one consumer thread to avoid the
  87     // lost wakeup problem.
  88
  89     void signalNewWork()
  90     {
  91         maNewWork.set();
  92     }
  93 };
  94
  95 ThreadPool::ThreadPool( sal_Int32 nWorkers ) :
  96     mnThreadsWorking( 0 ),
  97     mbTerminate( false )
  98 {
  99     for( sal_Int32 i = 0; i < nWorkers; i++ )
 100         maWorkers.push_back( new ThreadWorker( this ) );
 101
 102     maTasksComplete.set();
 103
 104     osl::MutexGuard aGuard( maGuard );
 105     for( size_t i = 0; i < maWorkers.size(); i++ )
 106         maWorkers[ i ]->launch();
 107 }
 108
 109 ThreadPool::~ThreadPool()
 110 {
 111     waitAndCleanupWorkers();
 112 }
 113
 114 struct ThreadPoolStatic : public rtl::StaticWithInit< boost::shared_ptr< ThreadPool >,
 115                                                       ThreadPoolStatic >
 116 {
 117     boost::shared_ptr< ThreadPool > operator () () {
 118         sal_Int32 nThreads = std::max( std::thread::hardware_concurrency(), 1U );
 119         return boost::shared_ptr< ThreadPool >( new ThreadPool( nThreads ) );
 120     };
 121 };
 122
 123 ThreadPool& ThreadPool::getSharedOptimalPool()
 124 {
 125     return *ThreadPoolStatic::get().get();
 126 }
 127
 128 void ThreadPool::waitAndCleanupWorkers()
 129 {
 130     waitUntilEmpty();
 131
 132     osl::ResettableMutexGuard aGuard( maGuard );
 133     mbTerminate = true;
 134
 135     while( !maWorkers.empty() )
 136     {
 137         rtl::Reference< ThreadWorker > xWorker = maWorkers.back();
 138         maWorkers.pop_back();
 139         assert(std::find(maWorkers.begin(), maWorkers.end(), xWorker)
 140                 == maWorkers.end());
 141         xWorker->signalNewWork();
 142         aGuard.clear();
 143         { // unlocked
 144             xWorker->join();
 145             xWorker.clear();
 146         }
 147         aGuard.reset();
 148     }
 149 }
 150
 151 void ThreadPool::pushTask( ThreadTask *pTask )
 152 {
 153     osl::MutexGuard aGuard( maGuard );
 154     maTasks.insert( maTasks.begin(), pTask );
 155
 156     // horrible beyond belief:
 157     for( size_t i = 0; i < maWorkers.size(); i++ )
 158         maWorkers[ i ]->signalNewWork();
 159     maTasksComplete.reset();
 160 }
 161
 162 ThreadTask *ThreadPool::popWork()
 163 {
 164     if( !maTasks.empty() )
 165     {
 166         ThreadTask *pTask = maTasks.back();
 167         maTasks.pop_back();
 168         return pTask;
 169     }
 170     else
 171         return NULL;
 172 }
 173
 174 void ThreadPool::startWork()
 175 {
 176     mnThreadsWorking++;
 177 }
 178
 179 void ThreadPool::stopWork()
 180 {
 181     assert( mnThreadsWorking > 0 );
 182     if ( --mnThreadsWorking == 0 )
 183         maTasksComplete.set();
 184 }
 185
 186 void ThreadPool::waitUntilEmpty()
 187 {
 188     osl::ResettableMutexGuard aGuard( maGuard );
 189
 190     if( maWorkers.empty() )
 191     { // no threads at all -> execute the work in-line
 192         ThreadTask *pTask;
 193         while ( ( pTask = popWork() ) )
 194         {
 195             pTask->doWork();
 196             delete pTask;
 197         }
 198     }
 199     else
 200     {
 201         aGuard.clear();
 202         maTasksComplete.wait();
 203         aGuard.reset();
 204     }
 205     assert( maTasks.empty() );
 206 }
 207
 208 } // namespace comphelper
 209
 210 /* vim:set shiftwidth=4 softtabstop=4 expandtab: */