Line data    Source code

       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 "threadpool.hxx"
      11             : 
      12             : #include <algorithm>
      13             : 
      14             : 
      15           0 : class ThreadPool::ThreadWorker : public salhelper::Thread
      16             : {
      17             :     ThreadPool    *mpPool;
      18             :     osl::Condition maNewWork;
      19             : public:
      20           0 :     ThreadWorker( ThreadPool *pPool ) :
      21             :         salhelper::Thread("sheet-import-thread-pool"),
      22           0 :         mpPool( pPool ) {}
      23             : 
      24           0 :     virtual void execute() SAL_OVERRIDE
      25             :     {
      26             :         ThreadTask *pTask;
      27           0 :         while ( ( pTask = waitForWork() ) )
      28             :         {
      29           0 :             pTask->doWork();
      30           0 :             delete pTask;
      31             :         }
      32           0 :     }
      33             : 
      34           0 :     ThreadTask *waitForWork()
      35             :     {
      36           0 :         ThreadTask *pRet = NULL;
      37             : 
      38           0 :         osl::ResettableMutexGuard aGuard( mpPool->maGuard );
      39             : 
      40           0 :         pRet = mpPool->popWork();
      41             : 
      42           0 :         while( !pRet )
      43             :         {
      44           0 :             maNewWork.reset();
      45             : 
      46           0 :             if( mpPool->mbTerminate )
      47           0 :                 break;
      48             : 
      49           0 :             aGuard.clear(); // unlock
      50             : 
      51           0 :             maNewWork.wait();
      52             : 
      53           0 :             aGuard.reset(); // lock
      54             : 
      55           0 :             pRet = mpPool->popWork();
      56             :         }
      57             : 
      58           0 :         return pRet;
      59             :     }
      60             : 
      61             : 
      62             :     // Why a condition per worker thread - you may ask.
      63             :     //
      64             :     // Unfortunately the Windows synchronisation API that we wrap
      65             :     // is horribly inadequate cf.
      66             :     //    http://www.cs.wustl.edu/~schmidt/win32-cv-1.html
      67             :     // The existing osl::Condition API should only ever be used
      68             :     // between one producer and one consumer thread to avoid the
      69             :     // lost wakeup problem.
      70             : 
      71           0 :     void signalNewWork()
      72             :     {
      73           0 :         maNewWork.set();
      74           0 :     }
      75             : };
      76             : 
      77           0 : ThreadPool::ThreadPool( sal_Int32 nWorkers ) :
      78           0 :     mbTerminate( false )
      79             : {
      80           0 :     for( sal_Int32 i = 0; i < nWorkers; i++ )
      81           0 :         maWorkers.push_back( new ThreadWorker( this ) );
      82             : 
      83           0 :     maTasksEmpty.reset();
      84             : 
      85           0 :     osl::MutexGuard aGuard( maGuard );
      86           0 :     for( size_t i = 0; i < maWorkers.size(); i++ )
      87           0 :         maWorkers[ i ]->launch();
      88           0 : }
      89             : 
      90           0 : ThreadPool::~ThreadPool()
      91             : {
      92           0 :     waitUntilWorkersDone();
      93           0 : }
      94             : 
      95             : /// wait until all the workers have completed and
      96             : /// terminate all threads
      97           0 : void ThreadPool::waitUntilWorkersDone()
      98             : {
      99           0 :     waitUntilEmpty();
     100             : 
     101           0 :     osl::ResettableMutexGuard aGuard( maGuard );
     102           0 :     mbTerminate = true;
     103             : 
     104           0 :     while( !maWorkers.empty() )
     105             :     {
     106           0 :         rtl::Reference< ThreadWorker > xWorker = maWorkers.back();
     107           0 :         maWorkers.pop_back();
     108             :         assert(std::find(maWorkers.begin(), maWorkers.end(), xWorker)
     109             :                 == maWorkers.end());
     110           0 :         xWorker->signalNewWork();
     111           0 :         aGuard.clear();
     112             :         { // unlocked
     113           0 :             xWorker->join();
     114           0 :             xWorker.clear();
     115             :         }
     116           0 :         aGuard.reset();
     117           0 :     }
     118           0 : }
     119             : 
     120           0 : void ThreadPool::pushTask( ThreadTask *pTask )
     121             : {
     122           0 :     osl::MutexGuard aGuard( maGuard );
     123           0 :     maTasks.insert( maTasks.begin(), pTask );
     124             :     // horrible beyond belief:
     125           0 :     for( size_t i = 0; i < maWorkers.size(); i++ )
     126           0 :         maWorkers[ i ]->signalNewWork();
     127           0 :     maTasksEmpty.reset();
     128           0 : }
     129             : 
     130           0 : ThreadTask *ThreadPool::popWork()
     131             : {
     132           0 :     if( !maTasks.empty() )
     133             :     {
     134           0 :         ThreadTask *pTask = maTasks.back();
     135           0 :         maTasks.pop_back();
     136           0 :         return pTask;
     137             :     }
     138             :     else
     139           0 :         maTasksEmpty.set();
     140           0 :     return NULL;
     141             : }
     142             : 
     143           0 : void ThreadPool::waitUntilEmpty()
     144             : {
     145           0 :     osl::ResettableMutexGuard aGuard( maGuard );
     146             : 
     147           0 :     if( maWorkers.empty() )
     148             :     { // no threads at all -> execute the work in-line
     149             :         ThreadTask *pTask;
     150           0 :         while ( ( pTask = popWork() ) )
     151             :         {
     152           0 :             pTask->doWork();
     153           0 :             delete pTask;
     154             :         }
     155           0 :         mbTerminate = true;
     156             :     }
     157             :     else
     158             :     {
     159           0 :         aGuard.clear();
     160           0 :         maTasksEmpty.wait();
     161           0 :         aGuard.reset();
     162             :     }
     163           0 :     assert( maTasks.empty() );
     164           0 : }
     165             : 
     166             : /* vim:set shiftwidth=4 softtabstop=4 expandtab: */