cppu/source/threadpool/threadpool.cxx

   1 /*************************************************************************
   2  *
   3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   4  *
   5  * Copyright 2008 by Sun Microsystems, Inc.
   6  *
   7  * OpenOffice.org - a multi-platform office productivity suite
   8  *
   9  * $RCSfile: threadpool.cxx,v $
  10  * $Revision: 1.18 $
  11  *
  12  * This file is part of OpenOffice.org.
  13  *
  14  * OpenOffice.org is free software: you can redistribute it and/or modify
  15  * it under the terms of the GNU Lesser General Public License version 3
  16  * only, as published by the Free Software Foundation.
  17  *
  18  * OpenOffice.org is distributed in the hope that it will be useful,
  19  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  20  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  21  * GNU Lesser General Public License version 3 for more details
  22  * (a copy is included in the LICENSE file that accompanied this code).
  23  *
  24  * You should have received a copy of the GNU Lesser General Public License
  25  * version 3 along with OpenOffice.org.  If not, see
  26  * <http://www.openoffice.org/license.html>
  27  * for a copy of the LGPLv3 License.
  28  *
  29  ************************************************************************/
  30
  31 // MARKER(update_precomp.py): autogen include statement, do not remove
  32 #include "precompiled_cppu.hxx"
  33 #include <hash_set>
  34 #include <stdio.h>
  35
  36 #include <osl/diagnose.h>
  37 #include <osl/mutex.hxx>
  38 #include <osl/thread.h>
  39
  40 #include <uno/threadpool.h>
  41
  42 #include "threadpool.hxx"
  43 #include "thread.hxx"
  44
  45 using namespace ::std;
  46 using namespace ::osl;
  47
  48 namespace cppu_threadpool
  49 {
  50     DisposedCallerAdmin *DisposedCallerAdmin::getInstance()
  51     {
  52         static DisposedCallerAdmin *pDisposedCallerAdmin = 0;
  53         if( ! pDisposedCallerAdmin )
  54         {
  55             MutexGuard guard( Mutex::getGlobalMutex() );
  56             if( ! pDisposedCallerAdmin )
  57             {
  58                 static DisposedCallerAdmin admin;
  59                 pDisposedCallerAdmin = &admin;
  60             }
  61         }
  62         return pDisposedCallerAdmin;
  63     }
  64
  65     DisposedCallerAdmin::~DisposedCallerAdmin()
  66     {
  67 #if OSL_DEBUG_LEVEL > 1
  68         if( !m_lst.empty() )
  69         {
  70             printf( "DisposedCallerList : %lu left\n" , static_cast<unsigned long>(m_lst.size( )));
  71         }
  72 #endif
  73     }
  74
  75     void DisposedCallerAdmin::dispose( sal_Int64 nDisposeId )
  76     {
  77         MutexGuard guard( m_mutex );
  78         m_lst.push_back( nDisposeId );
  79     }
  80
  81     void DisposedCallerAdmin::stopDisposing( sal_Int64 nDisposeId )
  82     {
  83         MutexGuard guard( m_mutex );
  84         for( DisposedCallerList::iterator ii = m_lst.begin() ;
  85              ii != m_lst.end() ;
  86              ++ ii )
  87         {
  88             if( (*ii) == nDisposeId )
  89             {
  90                 m_lst.erase( ii );
  91                 break;
  92             }
  93         }
  94     }
  95
  96     sal_Bool DisposedCallerAdmin::isDisposed( sal_Int64 nDisposeId )
  97     {
  98         MutexGuard guard( m_mutex );
  99         for( DisposedCallerList::iterator ii = m_lst.begin() ;
 100              ii != m_lst.end() ;
 101              ++ ii )
 102         {
 103             if( (*ii) == nDisposeId )
 104             {
 105                 return sal_True;
 106             }
 107         }
 108         return sal_False;
 109     }
 110
 111
 112     //-------------------------------------------------------------------------------
 113     ThreadPool::~ThreadPool()
 114     {
 115 #if OSL_DEBUG_LEVEL > 1
 116         if( m_mapQueue.size() )
 117         {
 118             printf( "ThreadIdHashMap : %lu left\n" , static_cast<unsigned long>(m_mapQueue.size()) );
 119         }
 120 #endif
 121     }
 122     ThreadPool *ThreadPool::getInstance()
 123     {
 124         static ThreadPool *pThreadPool = 0;
 125         if( ! pThreadPool )
 126         {
 127             MutexGuard guard( Mutex::getGlobalMutex() );
 128             if( ! pThreadPool )
 129             {
 130                 static ThreadPool pool;
 131                 pThreadPool = &pool;
 132             }
 133         }
 134         return pThreadPool;
 135     }
 136
 137
 138     void ThreadPool::dispose( sal_Int64 nDisposeId )
 139     {
 140         if( nDisposeId )
 141         {
 142             DisposedCallerAdmin::getInstance()->dispose( nDisposeId );
 143
 144             MutexGuard guard( m_mutex );
 145             for( ThreadIdHashMap::iterator ii = m_mapQueue.begin() ;
 146                  ii != m_mapQueue.end();
 147                  ++ii)
 148             {
 149                 if( (*ii).second.first )
 150                 {
 151                     (*ii).second.first->dispose( nDisposeId );
 152                 }
 153                 if( (*ii).second.second )
 154                 {
 155                     (*ii).second.second->dispose( nDisposeId );
 156                 }
 157             }
 158         }
 159         else
 160         {
 161             {
 162                 MutexGuard guard( m_mutexWaitingThreadList );
 163                 for( WaitingThreadList::iterator ii = m_lstThreads.begin() ;
 164                      ii != m_lstThreads.end() ;
 165                      ++ ii )
 166                 {
 167                     // wake the threads up
 168                     osl_setCondition( (*ii)->condition );
 169                 }
 170             }
 171             ThreadAdmin::getInstance()->join();
 172         }
 173     }
 174
 175     void ThreadPool::stopDisposing( sal_Int64 nDisposeId )
 176     {
 177         DisposedCallerAdmin::getInstance()->stopDisposing( nDisposeId );
 178     }
 179
 180     /******************
 181      * This methods lets the thread wait a certain amount of time. If within this timespan
 182      * a new request comes in, this thread is reused. This is done only to improve performance,
 183      * it is not required for threadpool functionality.
 184      ******************/
 185     void ThreadPool::waitInPool( ORequestThread * pThread )
 186     {
 187         struct WaitingThread waitingThread;
 188         waitingThread.condition = osl_createCondition();
 189         waitingThread.thread = pThread;
 190         {
 191             MutexGuard guard( m_mutexWaitingThreadList );
 192             m_lstThreads.push_front( &waitingThread );
 193         }
 194
 195         // let the thread wait 2 seconds
 196         TimeValue time = { 2 , 0 };
 197         osl_waitCondition( waitingThread.condition , &time );
 198
 199         {
 200             MutexGuard guard ( m_mutexWaitingThreadList );
 201             if( waitingThread.thread )
 202             {
 203                 // thread wasn't reused, remove it from the list
 204                 WaitingThreadList::iterator ii = find(
 205                     m_lstThreads.begin(), m_lstThreads.end(), &waitingThread );
 206                 OSL_ASSERT( ii != m_lstThreads.end() );
 207                 m_lstThreads.erase( ii );
 208             }
 209         }
 210
 211         osl_destroyCondition( waitingThread.condition );
 212     }
 213
 214     void ThreadPool::createThread( JobQueue *pQueue ,
 215                                    const ByteSequence &aThreadId,
 216                                    sal_Bool bAsynchron )
 217     {
 218         sal_Bool bCreate = sal_True;
 219         {
 220             // Can a thread be reused ?
 221             MutexGuard guard( m_mutexWaitingThreadList );
 222             if( ! m_lstThreads.empty() )
 223             {
 224                 // inform the thread and let it go
 225                 struct WaitingThread *pWaitingThread = m_lstThreads.back();
 226                 pWaitingThread->thread->setTask( pQueue , aThreadId , bAsynchron );
 227                 pWaitingThread->thread = 0;
 228
 229                 // remove from list
 230                 m_lstThreads.pop_back();
 231
 232                 // let the thread go
 233                 osl_setCondition( pWaitingThread->condition );
 234                 bCreate = sal_False;
 235             }
 236         }
 237
 238         if( bCreate )
 239         {
 240             ORequestThread *pThread =
 241                 new ORequestThread( pQueue , aThreadId, bAsynchron);
 242             // deletes itself !
 243             pThread->create();
 244         }
 245     }
 246
 247     sal_Bool ThreadPool::revokeQueue( const ByteSequence &aThreadId, sal_Bool bAsynchron )
 248     {
 249         MutexGuard guard( m_mutex );
 250
 251         ThreadIdHashMap::iterator ii = m_mapQueue.find( aThreadId );
 252         OSL_ASSERT( ii != m_mapQueue.end() );
 253
 254         if( bAsynchron )
 255         {
 256             if( ! (*ii).second.second->isEmpty() )
 257             {
 258                 // another thread has put something into the queue
 259                 return sal_False;
 260             }
 261
 262             (*ii).second.second = 0;
 263             if( (*ii).second.first )
 264             {
 265                 // all oneway request have been processed, now
 266                 // synchronus requests may go on
 267                 (*ii).second.first->resume();
 268             }
 269         }
 270         else
 271         {
 272             if( ! (*ii).second.first->isEmpty() )
 273             {
 274                 // another thread has put something into the queue
 275                 return sal_False;
 276             }
 277             (*ii).second.first = 0;
 278         }
 279
 280         if( 0 == (*ii).second.first && 0 == (*ii).second.second )
 281         {
 282             m_mapQueue.erase( ii );
 283         }
 284
 285         return sal_True;
 286     }
 287
 288
 289     void ThreadPool::addJob(
 290         const ByteSequence &aThreadId ,
 291         sal_Bool bAsynchron,
 292         void *pThreadSpecificData,
 293         RequestFun * doRequest )
 294     {
 295         sal_Bool bCreateThread = sal_False;
 296         JobQueue *pQueue = 0;
 297         {
 298             MutexGuard guard( m_mutex );
 299
 300             ThreadIdHashMap::iterator ii = m_mapQueue.find( aThreadId );
 301
 302             if( ii == m_mapQueue.end() )
 303             {
 304                 m_mapQueue[ aThreadId ] = pair < JobQueue * , JobQueue * > ( 0 , 0 );
 305                 ii = m_mapQueue.find( aThreadId );
 306                 OSL_ASSERT( ii != m_mapQueue.end() );
 307             }
 308
 309             if( bAsynchron )
 310             {
 311                 if( ! (*ii).second.second )
 312                 {
 313                     (*ii).second.second = new JobQueue();
 314                     bCreateThread = sal_True;
 315                 }
 316                 pQueue = (*ii).second.second;
 317             }
 318             else
 319             {
 320                 if( ! (*ii).second.first )
 321                 {
 322                     (*ii).second.first = new JobQueue();
 323                     bCreateThread = sal_True;
 324                 }
 325                 pQueue = (*ii).second.first;
 326
 327                 if( (*ii).second.second && ( (*ii).second.second->isBusy() ) )
 328                 {
 329                     pQueue->suspend();
 330                 }
 331             }
 332             pQueue->add( pThreadSpecificData , doRequest );
 333         }
 334
 335         if( bCreateThread )
 336         {
 337             createThread( pQueue , aThreadId , bAsynchron);
 338         }
 339     }
 340
 341     void ThreadPool::prepare( const ByteSequence &aThreadId )
 342     {
 343         MutexGuard guard( m_mutex );
 344
 345         ThreadIdHashMap::iterator ii = m_mapQueue.find( aThreadId );
 346
 347         if( ii == m_mapQueue.end() )
 348         {
 349             JobQueue *p = new JobQueue();
 350             m_mapQueue[ aThreadId ] = pair< JobQueue * , JobQueue * > ( p , 0 );
 351         }
 352         else if( 0 == (*ii).second.first )
 353         {
 354             (*ii).second.first = new JobQueue();
 355         }
 356     }
 357
 358     void * ThreadPool::enter( const ByteSequence & aThreadId , sal_Int64 nDisposeId )
 359     {
 360         JobQueue *pQueue = 0;
 361         {
 362             MutexGuard guard( m_mutex );
 363
 364             ThreadIdHashMap::iterator ii = m_mapQueue.find( aThreadId );
 365
 366             OSL_ASSERT( ii != m_mapQueue.end() );
 367             pQueue = (*ii).second.first;
 368         }
 369
 370         OSL_ASSERT( pQueue );
 371         void *pReturn = pQueue->enter( nDisposeId );
 372
 373         if( pQueue->isCallstackEmpty() )
 374         {
 375             if( revokeQueue( aThreadId , sal_False) )
 376             {
 377                 // remove queue
 378                 delete pQueue;
 379             }
 380         }
 381         return pReturn;
 382     }
 383 }
 384
 385
 386 using namespace cppu_threadpool;
 387
 388 struct uno_ThreadPool_Equal
 389 {
 390     sal_Bool operator () ( const uno_ThreadPool &a , const uno_ThreadPool &b ) const
 391         {
 392             return a == b;
 393         }
 394 };
 395
 396 struct uno_ThreadPool_Hash
 397 {
 398     sal_Size operator () ( const uno_ThreadPool &a  )  const
 399         {
 400             return (sal_Size) a;
 401         }
 402 };
 403
 404
 405
 406 typedef ::std::hash_set< uno_ThreadPool, uno_ThreadPool_Hash, uno_ThreadPool_Equal > ThreadpoolHashSet;
 407
 408 static ThreadpoolHashSet *g_pThreadpoolHashSet;
 409
 410 struct _uno_ThreadPool
 411 {
 412     sal_Int32 dummy;
 413 };
 414
 415 extern "C" uno_ThreadPool SAL_CALL
 416 uno_threadpool_create() SAL_THROW_EXTERN_C()
 417 {
 418     MutexGuard guard( Mutex::getGlobalMutex() );
 419     if( ! g_pThreadpoolHashSet )
 420     {
 421         g_pThreadpoolHashSet = new ThreadpoolHashSet();
 422     }
 423
 424     // Just ensure that the handle is unique in the process (via heap)
 425     uno_ThreadPool h = new struct _uno_ThreadPool;
 426     g_pThreadpoolHashSet->insert( h );
 427     return h;
 428 }
 429
 430 extern "C" void SAL_CALL
 431 uno_threadpool_attach( uno_ThreadPool ) SAL_THROW_EXTERN_C()
 432 {
 433     sal_Sequence *pThreadId = 0;
 434     uno_getIdOfCurrentThread( &pThreadId );
 435     ThreadPool::getInstance()->prepare( pThreadId );
 436     rtl_byte_sequence_release( pThreadId );
 437     uno_releaseIdFromCurrentThread();
 438 }
 439
 440 extern "C" void SAL_CALL
 441 uno_threadpool_enter( uno_ThreadPool hPool , void **ppJob )
 442     SAL_THROW_EXTERN_C()
 443 {
 444     sal_Sequence *pThreadId = 0;
 445     uno_getIdOfCurrentThread( &pThreadId );
 446     *ppJob =
 447         ThreadPool::getInstance()->enter(
 448             pThreadId,
 449             sal::static_int_cast< sal_Int64 >(
 450                 reinterpret_cast< sal_IntPtr >(hPool)) );
 451     rtl_byte_sequence_release( pThreadId );
 452     uno_releaseIdFromCurrentThread();
 453 }
 454
 455 extern "C" void SAL_CALL
 456 uno_threadpool_detach( uno_ThreadPool ) SAL_THROW_EXTERN_C()
 457 {
 458     // we might do here some tiding up in case a thread called attach but never detach
 459 }
 460
 461 extern "C" void SAL_CALL
 462 uno_threadpool_putJob(
 463     uno_ThreadPool,
 464     sal_Sequence *pThreadId,
 465     void *pJob,
 466     void ( SAL_CALL * doRequest ) ( void *pThreadSpecificData ),
 467     sal_Bool bIsOneway ) SAL_THROW_EXTERN_C()
 468 {
 469     ThreadPool::getInstance()->addJob( pThreadId, bIsOneway, pJob ,doRequest );
 470 }
 471
 472 extern "C" void SAL_CALL
 473 uno_threadpool_dispose( uno_ThreadPool hPool ) SAL_THROW_EXTERN_C()
 474 {
 475     ThreadPool::getInstance()->dispose(
 476         sal::static_int_cast< sal_Int64 >(
 477             reinterpret_cast< sal_IntPtr >(hPool)) );
 478 }
 479
 480 extern "C" void SAL_CALL
 481 uno_threadpool_destroy( uno_ThreadPool hPool ) SAL_THROW_EXTERN_C()
 482 {
 483     ThreadPool::getInstance()->stopDisposing(
 484         sal::static_int_cast< sal_Int64 >(
 485             reinterpret_cast< sal_IntPtr >(hPool)) );
 486
 487     if( hPool )
 488     {
 489         // special treatment for 0 !
 490         OSL_ASSERT( g_pThreadpoolHashSet );
 491
 492         MutexGuard guard( Mutex::getGlobalMutex() );
 493
 494         ThreadpoolHashSet::iterator ii = g_pThreadpoolHashSet->find( hPool );
 495         OSL_ASSERT( ii != g_pThreadpoolHashSet->end() );
 496         g_pThreadpoolHashSet->erase( ii );
 497         delete hPool;
 498
 499         if( g_pThreadpoolHashSet->empty() )
 500         {
 501             delete g_pThreadpoolHashSet;
 502             g_pThreadpoolHashSet = 0;
 503         }
 504     }
 505 }