cppu/source/threadpool/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  * This file incorporates work covered by the following license notice:
  10  *
  11  *   Licensed to the Apache Software Foundation (ASF) under one or more
  12  *   contributor license agreements. See the NOTICE file distributed
  13  *   with this work for additional information regarding copyright
  14  *   ownership. The ASF licenses this file to you under the Apache
  15  *   License, Version 2.0 (the "License"); you may not use this file
  16  *   except in compliance with the License. You may obtain a copy of
  17  *   the License at http://www.apache.org/licenses/LICENSE-2.0 .
  18  */
  19
  20 #include <sal/config.h>
  21
  22 #include <cassert>
  23 #include <chrono>
  24 #include <algorithm>
  25 #include <unordered_map>
  26
  27 #include <osl/diagnose.h>
  28 #include <osl/mutex.hxx>
  29 #include <rtl/instance.hxx>
  30 #include <sal/log.hxx>
  31
  32 #include <uno/threadpool.h>
  33
  34 #include "threadpool.hxx"
  35 #include "thread.hxx"
  36
  37 using namespace ::std;
  38 using namespace ::osl;
  39 using namespace ::rtl;
  40
  41 namespace cppu_threadpool
  42 {
  43     WaitingThread::WaitingThread(
  44         rtl::Reference<ORequestThread> const & theThread): thread(theThread)
  45     {}
  46
  47     namespace {
  48
  49     struct theDisposedCallerAdmin :
  50         public rtl::StaticWithInit< DisposedCallerAdminHolder, theDisposedCallerAdmin >
  51     {
  52         DisposedCallerAdminHolder operator () () {
  53             return std::make_shared<DisposedCallerAdmin>();
  54         }
  55     };
  56
  57     }
  58
  59     DisposedCallerAdminHolder const & DisposedCallerAdmin::getInstance()
  60     {
  61         return theDisposedCallerAdmin::get();
  62     }
  63
  64     DisposedCallerAdmin::~DisposedCallerAdmin()
  65     {
  66         SAL_WARN_IF( !m_vector.empty(), "cppu.threadpool", "DisposedCallerList :  " << m_vector.size() << " left");
  67     }
  68
  69     void DisposedCallerAdmin::dispose( void const * nDisposeId )
  70     {
  71         MutexGuard guard( m_mutex );
  72         m_vector.push_back( nDisposeId );
  73     }
  74
  75     void DisposedCallerAdmin::destroy( void const * nDisposeId )
  76     {
  77         MutexGuard guard( m_mutex );
  78         m_vector.erase(std::remove(m_vector.begin(), m_vector.end(), nDisposeId), m_vector.end());
  79     }
  80
  81     bool DisposedCallerAdmin::isDisposed( void const * nDisposeId )
  82     {
  83         MutexGuard guard( m_mutex );
  84         return (std::find(m_vector.begin(), m_vector.end(), nDisposeId) != m_vector.end());
  85     }
  86
  87
  88     ThreadPool::ThreadPool() :
  89         m_DisposedCallerAdmin( DisposedCallerAdmin::getInstance() )
  90     {
  91     }
  92
  93     ThreadPool::~ThreadPool()
  94     {
  95         SAL_WARN_IF( m_mapQueue.size(), "cppu.threadpool", "ThreadIdHashMap:  " << m_mapQueue.size() << " left");
  96     }
  97
  98     void ThreadPool::dispose( void const * nDisposeId )
  99     {
 100         m_DisposedCallerAdmin->dispose( nDisposeId );
 101
 102         MutexGuard guard( m_mutex );
 103         for (auto const& item :  m_mapQueue)
 104         {
 105             if( item.second.first )
 106             {
 107                 item.second.first->dispose( nDisposeId );
 108             }
 109             if( item.second.second )
 110             {
 111                 item.second.second->dispose( nDisposeId );
 112             }
 113         }
 114     }
 115
 116     void ThreadPool::destroy( void const * nDisposeId )
 117     {
 118         m_DisposedCallerAdmin->destroy( nDisposeId );
 119     }
 120
 121     /******************
 122      * This methods lets the thread wait a certain amount of time. If within this timespan
 123      * a new request comes in, this thread is reused. This is done only to improve performance,
 124      * it is not required for threadpool functionality.
 125      ******************/
 126     void ThreadPool::waitInPool( rtl::Reference< ORequestThread > const & pThread )
 127     {
 128         WaitingThread waitingThread(pThread);
 129         {
 130             MutexGuard guard( m_mutexWaitingThreadList );
 131             m_dequeThreads.push_front( &waitingThread );
 132         }
 133
 134         // let the thread wait 2 seconds
 135         waitingThread.condition.wait( std::chrono::seconds(2) );
 136
 137         {
 138             MutexGuard guard ( m_mutexWaitingThreadList );
 139             if( waitingThread.thread.is() )
 140             {
 141                 // thread wasn't reused, remove it from the list
 142                 WaitingThreadDeque::iterator ii = find(
 143                     m_dequeThreads.begin(), m_dequeThreads.end(), &waitingThread );
 144                 OSL_ASSERT( ii != m_dequeThreads.end() );
 145                 m_dequeThreads.erase( ii );
 146             }
 147         }
 148     }
 149
 150     void ThreadPool::joinWorkers()
 151     {
 152         {
 153             MutexGuard guard( m_mutexWaitingThreadList );
 154             for (auto const& thread : m_dequeThreads)
 155             {
 156                 // wake the threads up
 157                 thread->condition.set();
 158             }
 159         }
 160         m_aThreadAdmin.join();
 161     }
 162
 163     bool ThreadPool::createThread( JobQueue *pQueue ,
 164                                    const ByteSequence &aThreadId,
 165                                    bool bAsynchron )
 166     {
 167         {
 168             // Can a thread be reused ?
 169             MutexGuard guard( m_mutexWaitingThreadList );
 170             if( ! m_dequeThreads.empty() )
 171             {
 172                 // inform the thread and let it go
 173                 struct WaitingThread *pWaitingThread = m_dequeThreads.back();
 174                 pWaitingThread->thread->setTask( pQueue , aThreadId , bAsynchron );
 175                 pWaitingThread->thread = nullptr;
 176
 177                 // remove from list
 178                 m_dequeThreads.pop_back();
 179
 180                 // let the thread go
 181                 pWaitingThread->condition.set();
 182                 return true;
 183             }
 184         }
 185
 186         rtl::Reference pThread(
 187             new ORequestThread( this, pQueue , aThreadId, bAsynchron) );
 188         return pThread->launch();
 189     }
 190
 191     bool ThreadPool::revokeQueue( const ByteSequence &aThreadId, bool bAsynchron )
 192     {
 193         MutexGuard guard( m_mutex );
 194
 195         ThreadIdHashMap::iterator ii = m_mapQueue.find( aThreadId );
 196         OSL_ASSERT( ii != m_mapQueue.end() );
 197
 198         if( bAsynchron )
 199         {
 200             if( ! (*ii).second.second->isEmpty() )
 201             {
 202                 // another thread has put something into the queue
 203                 return false;
 204             }
 205
 206             (*ii).second.second = nullptr;
 207             if( (*ii).second.first )
 208             {
 209                 // all oneway request have been processed, now
 210                 // synchronous requests may go on
 211                 (*ii).second.first->resume();
 212             }
 213         }
 214         else
 215         {
 216             if( ! (*ii).second.first->isEmpty() )
 217             {
 218                 // another thread has put something into the queue
 219                 return false;
 220             }
 221             (*ii).second.first = nullptr;
 222         }
 223
 224         if( nullptr == (*ii).second.first && nullptr == (*ii).second.second )
 225         {
 226             m_mapQueue.erase( ii );
 227         }
 228
 229         return true;
 230     }
 231
 232
 233     bool ThreadPool::addJob(
 234         const ByteSequence &aThreadId ,
 235         bool bAsynchron,
 236         void *pThreadSpecificData,
 237         RequestFun * doRequest,
 238         void const * disposeId )
 239     {
 240         bool bCreateThread = false;
 241         JobQueue *pQueue = nullptr;
 242         {
 243             MutexGuard guard( m_mutex );
 244             if (m_DisposedCallerAdmin->isDisposed(disposeId)) {
 245                 return true;
 246             }
 247
 248             ThreadIdHashMap::iterator ii = m_mapQueue.find( aThreadId );
 249
 250             if( ii == m_mapQueue.end() )
 251             {
 252                 m_mapQueue[ aThreadId ] = pair < JobQueue * , JobQueue * > ( nullptr , nullptr );
 253                 ii = m_mapQueue.find( aThreadId );
 254                 OSL_ASSERT( ii != m_mapQueue.end() );
 255             }
 256
 257             if( bAsynchron )
 258             {
 259                 if( ! (*ii).second.second )
 260                 {
 261                     (*ii).second.second = new JobQueue();
 262                     bCreateThread = true;
 263                 }
 264                 pQueue = (*ii).second.second;
 265             }
 266             else
 267             {
 268                 if( ! (*ii).second.first )
 269                 {
 270                     (*ii).second.first = new JobQueue();
 271                     bCreateThread = true;
 272                 }
 273                 pQueue = (*ii).second.first;
 274
 275                 if( (*ii).second.second && ( (*ii).second.second->isBusy() ) )
 276                 {
 277                     pQueue->suspend();
 278                 }
 279             }
 280             pQueue->add( pThreadSpecificData , doRequest );
 281         }
 282
 283         return !bCreateThread || createThread( pQueue , aThreadId , bAsynchron);
 284     }
 285
 286     void ThreadPool::prepare( const ByteSequence &aThreadId )
 287     {
 288         MutexGuard guard( m_mutex );
 289
 290         ThreadIdHashMap::iterator ii = m_mapQueue.find( aThreadId );
 291
 292         if( ii == m_mapQueue.end() )
 293         {
 294             JobQueue *p = new JobQueue();
 295             m_mapQueue[ aThreadId ] = pair< JobQueue * , JobQueue * > ( p , nullptr );
 296         }
 297         else if( nullptr == (*ii).second.first )
 298         {
 299             (*ii).second.first = new JobQueue();
 300         }
 301     }
 302
 303     void * ThreadPool::enter( const ByteSequence & aThreadId , void const * nDisposeId )
 304     {
 305         JobQueue *pQueue = nullptr;
 306         {
 307             MutexGuard guard( m_mutex );
 308
 309             ThreadIdHashMap::iterator ii = m_mapQueue.find( aThreadId );
 310
 311             OSL_ASSERT( ii != m_mapQueue.end() );
 312             pQueue = (*ii).second.first;
 313         }
 314
 315         OSL_ASSERT( pQueue );
 316         void *pReturn = pQueue->enter( nDisposeId );
 317
 318         if( pQueue->isCallstackEmpty() )
 319         {
 320             if( revokeQueue( aThreadId , false) )
 321             {
 322                 // remove queue
 323                 delete pQueue;
 324             }
 325         }
 326         return pReturn;
 327     }
 328 }
 329
 330 // All uno_ThreadPool handles in g_pThreadpoolHashSet with overlapping life
 331 // spans share one ThreadPool instance.  When g_pThreadpoolHashSet becomes empty
 332 // (within the last uno_threadpool_destroy) all worker threads spawned by that
 333 // ThreadPool instance are joined (which implies that uno_threadpool_destroy
 334 // must never be called from a worker thread); afterwards, the next call to
 335 // uno_threadpool_create (if any) will lead to a new ThreadPool instance.
 336
 337 using namespace cppu_threadpool;
 338
 339 namespace {
 340
 341 struct uno_ThreadPool_Equal
 342 {
 343     bool operator () ( const uno_ThreadPool &a , const uno_ThreadPool &b ) const
 344         {
 345             return a == b;
 346         }
 347 };
 348
 349 struct uno_ThreadPool_Hash
 350 {
 351     std::size_t operator () ( const uno_ThreadPool &a  )  const
 352         {
 353             return reinterpret_cast<std::size_t>( a );
 354         }
 355 };
 356
 357 }
 358
 359 typedef std::unordered_map< uno_ThreadPool, ThreadPoolHolder, uno_ThreadPool_Hash, uno_ThreadPool_Equal > ThreadpoolHashSet;
 360
 361 static ThreadpoolHashSet *g_pThreadpoolHashSet;
 362
 363 struct _uno_ThreadPool
 364 {
 365     sal_Int32 dummy;
 366 };
 367
 368 namespace {
 369
 370 ThreadPoolHolder getThreadPool( uno_ThreadPool hPool )
 371 {
 372     MutexGuard guard( Mutex::getGlobalMutex() );
 373     assert( g_pThreadpoolHashSet != nullptr );
 374     ThreadpoolHashSet::iterator i( g_pThreadpoolHashSet->find(hPool) );
 375     assert( i != g_pThreadpoolHashSet->end() );
 376     return i->second;
 377 }
 378
 379 }
 380
 381 extern "C" uno_ThreadPool SAL_CALL
 382 uno_threadpool_create() SAL_THROW_EXTERN_C()
 383 {
 384     MutexGuard guard( Mutex::getGlobalMutex() );
 385     ThreadPoolHolder p;
 386     if( ! g_pThreadpoolHashSet )
 387     {
 388         g_pThreadpoolHashSet = new ThreadpoolHashSet;
 389         p = new ThreadPool;
 390     }
 391     else
 392     {
 393         assert( !g_pThreadpoolHashSet->empty() );
 394         p = g_pThreadpoolHashSet->begin()->second;
 395     }
 396
 397     // Just ensure that the handle is unique in the process (via heap)
 398     uno_ThreadPool h = new struct _uno_ThreadPool;
 399     g_pThreadpoolHashSet->emplace( h, p );
 400     return h;
 401 }
 402
 403 extern "C" void SAL_CALL
 404 uno_threadpool_attach( uno_ThreadPool hPool ) SAL_THROW_EXTERN_C()
 405 {
 406     sal_Sequence *pThreadId = nullptr;
 407     uno_getIdOfCurrentThread( &pThreadId );
 408     getThreadPool( hPool )->prepare( pThreadId );
 409     rtl_byte_sequence_release( pThreadId );
 410     uno_releaseIdFromCurrentThread();
 411 }
 412
 413 extern "C" void SAL_CALL
 414 uno_threadpool_enter( uno_ThreadPool hPool , void **ppJob )
 415     SAL_THROW_EXTERN_C()
 416 {
 417     sal_Sequence *pThreadId = nullptr;
 418     uno_getIdOfCurrentThread( &pThreadId );
 419     *ppJob =
 420         getThreadPool( hPool )->enter(
 421             pThreadId,
 422             hPool );
 423     rtl_byte_sequence_release( pThreadId );
 424     uno_releaseIdFromCurrentThread();
 425 }
 426
 427 extern "C" void SAL_CALL
 428 uno_threadpool_detach(SAL_UNUSED_PARAMETER uno_ThreadPool) SAL_THROW_EXTERN_C()
 429 {
 430     // we might do here some tidying up in case a thread called attach but never detach
 431 }
 432
 433 extern "C" void SAL_CALL
 434 uno_threadpool_putJob(
 435     uno_ThreadPool hPool,
 436     sal_Sequence *pThreadId,
 437     void *pJob,
 438     void ( SAL_CALL * doRequest ) ( void *pThreadSpecificData ),
 439     sal_Bool bIsOneway ) SAL_THROW_EXTERN_C()
 440 {
 441     if (!getThreadPool(hPool)->addJob( pThreadId, bIsOneway, pJob ,doRequest, hPool ))
 442     {
 443         SAL_WARN(
 444             "cppu.threadpool",
 445             "uno_threadpool_putJob in parallel with uno_threadpool_destroy");
 446     }
 447 }
 448
 449 extern "C" void SAL_CALL
 450 uno_threadpool_dispose( uno_ThreadPool hPool ) SAL_THROW_EXTERN_C()
 451 {
 452     getThreadPool(hPool)->dispose(
 453         hPool );
 454 }
 455
 456 extern "C" void SAL_CALL
 457 uno_threadpool_destroy( uno_ThreadPool hPool ) SAL_THROW_EXTERN_C()
 458 {
 459     ThreadPoolHolder p( getThreadPool(hPool) );
 460     p->destroy(
 461         hPool );
 462
 463     bool empty;
 464     {
 465         OSL_ASSERT( g_pThreadpoolHashSet );
 466
 467         MutexGuard guard( Mutex::getGlobalMutex() );
 468
 469         ThreadpoolHashSet::iterator ii = g_pThreadpoolHashSet->find( hPool );
 470         OSL_ASSERT( ii != g_pThreadpoolHashSet->end() );
 471         g_pThreadpoolHashSet->erase( ii );
 472         delete hPool;
 473
 474         empty = g_pThreadpoolHashSet->empty();
 475         if( empty )
 476         {
 477             delete g_pThreadpoolHashSet;
 478             g_pThreadpoolHashSet = nullptr;
 479         }
 480     }
 481
 482     if( empty )
 483     {
 484         p->joinWorkers();
 485     }
 486 }
 487
 488 /* vim:set shiftwidth=4 softtabstop=4 expandtab: */