cppu/source/threadpool/thread.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: thread.cxx,v $
  10  * $Revision: 1.13 $
  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 <stdio.h>
  34 #include <osl/diagnose.h>
  35 #include <uno/threadpool.h>
  36
  37 #include "thread.hxx"
  38 #include "jobqueue.hxx"
  39 #include "threadpool.hxx"
  40
  41
  42 using namespace osl;
  43 extern "C" {
  44
  45 void SAL_CALL cppu_requestThreadWorker( void *pVoid )
  46 {
  47     ::cppu_threadpool::ORequestThread *pThread = ( ::cppu_threadpool::ORequestThread * ) pVoid;
  48
  49     pThread->run();
  50     pThread->onTerminated();
  51 }
  52
  53 }
  54
  55 namespace cppu_threadpool {
  56
  57 // ----------------------------------------------------------------------------------
  58     ThreadAdmin::~ThreadAdmin()
  59     {
  60 #if OSL_DEBUG_LEVEL > 1
  61         if( m_lst.size() )
  62         {
  63             fprintf( stderr, "%lu Threads left\n" , static_cast<unsigned long>(m_lst.size()) );
  64         }
  65 #endif
  66     }
  67
  68     void ThreadAdmin::add( ORequestThread *p )
  69     {
  70         MutexGuard aGuard( m_mutex );
  71         m_lst.push_back( p );
  72     }
  73
  74     void ThreadAdmin::remove( ORequestThread * p )
  75     {
  76         MutexGuard aGuard( m_mutex );
  77         ::std::list< ORequestThread * >::iterator ii = ::std::find( m_lst.begin(), m_lst.end(), p );
  78         OSL_ASSERT( ii != m_lst.end() );
  79         m_lst.erase( ii );
  80     }
  81
  82     void ThreadAdmin::join()
  83     {
  84         ORequestThread *pCurrent;
  85         do
  86         {
  87             pCurrent = 0;
  88             {
  89                 MutexGuard aGuard( m_mutex );
  90                 if( ! m_lst.empty() )
  91                 {
  92                     pCurrent = m_lst.front();
  93                     pCurrent->setDeleteSelf( sal_False );
  94                 }
  95             }
  96             if ( pCurrent )
  97             {
  98                 pCurrent->join();
  99                 delete pCurrent;
 100             }
 101         } while( pCurrent );
 102     }
 103
 104     ThreadAdmin* ThreadAdmin::getInstance()
 105     {
 106         static ThreadAdmin *pThreadAdmin = 0;
 107         if( ! pThreadAdmin )
 108         {
 109             MutexGuard guard( Mutex::getGlobalMutex() );
 110             if( ! pThreadAdmin )
 111             {
 112                 static ThreadAdmin admin;
 113                 pThreadAdmin = &admin;
 114             }
 115         }
 116         return pThreadAdmin;
 117
 118     }
 119
 120 // ----------------------------------------------------------------------------------
 121     ORequestThread::ORequestThread( JobQueue *pQueue,
 122                                     const ByteSequence &aThreadId,
 123                                     sal_Bool bAsynchron )
 124         : m_thread( 0 )
 125         , m_pQueue( pQueue )
 126         , m_aThreadId( aThreadId )
 127         , m_bAsynchron( bAsynchron )
 128         , m_bDeleteSelf( sal_True )
 129     {
 130         ThreadAdmin::getInstance()->add( this );
 131     }
 132
 133
 134     ORequestThread::~ORequestThread()
 135     {
 136         if (m_thread != 0)
 137         {
 138             osl_destroyThread(m_thread);
 139         }
 140     }
 141
 142
 143     void ORequestThread::setTask( JobQueue *pQueue,
 144                                   const ByteSequence &aThreadId,
 145                                   sal_Bool bAsynchron )
 146     {
 147         m_pQueue = pQueue;
 148         m_aThreadId = aThreadId;
 149         m_bAsynchron = bAsynchron;
 150     }
 151
 152     sal_Bool ORequestThread::create()
 153     {
 154         OSL_ASSERT(m_thread == 0);      // only one running thread per instance
 155
 156         m_thread = osl_createSuspendedThread( cppu_requestThreadWorker, (void*)this);
 157         if ( m_thread )
 158         {
 159             osl_resumeThread( m_thread );
 160         }
 161
 162         return m_thread != 0;
 163     }
 164
 165     void ORequestThread::join()
 166     {
 167         osl_joinWithThread( m_thread );
 168     }
 169
 170     void ORequestThread::onTerminated()
 171     {
 172         ThreadAdmin::getInstance()->remove( this );
 173         if( m_bDeleteSelf )
 174         {
 175             delete this;
 176         }
 177     }
 178
 179     void ORequestThread::run()
 180     {
 181         while ( m_pQueue )
 182         {
 183             if( ! m_bAsynchron )
 184             {
 185                 if ( !uno_bindIdToCurrentThread( m_aThreadId.getHandle() ) )
 186                 {
 187                     OSL_ASSERT( false );
 188                 }
 189             }
 190
 191             while( ! m_pQueue->isEmpty() )
 192             {
 193                 // Note : Oneways should not get a disposable disposeid,
 194                 //        It does not make sense to dispose a call in this state.
 195                 //        That's way we put it an disposeid, that can't be used otherwise.
 196                 m_pQueue->enter(
 197                     sal::static_int_cast< sal_Int64 >(
 198                         reinterpret_cast< sal_IntPtr >(this)),
 199                     sal_True );
 200
 201                 if( m_pQueue->isEmpty() )
 202                 {
 203                     ThreadPool::getInstance()->revokeQueue( m_aThreadId , m_bAsynchron );
 204                     // Note : revokeQueue might have failed because m_pQueue.isEmpty()
 205                     //        may be false (race).
 206                 }
 207             }
 208
 209             delete m_pQueue;
 210             m_pQueue = 0;
 211
 212             if( ! m_bAsynchron )
 213             {
 214                 uno_releaseIdFromCurrentThread();
 215             }
 216
 217             cppu_threadpool::ThreadPool::getInstance()->waitInPool( this );
 218         }
 219     }
 220 }