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