00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022
00023
00024
00025 #ifndef WTHREADEDFUNCTION_H
00026 #define WTHREADEDFUNCTION_H
00027
00028 #include <memory.h>
00029 #include <iostream>
00030
00031 #include <string>
00032 #include <vector>
00033 #include <boost/thread.hpp>
00034
00035 #include "WAssert.h"
00036 #include "WWorkerThread.h"
00037 #include "WSharedObject.h"
00038 #include "WExportCommon.h"
00039
00040
00041
00042
00043 enum WThreadedFunctionStatus
00044 {
00045 W_THREADS_INITIALIZED,
00046 W_THREADS_RUNNING,
00047 W_THREADS_STOP_REQUESTED,
00048 W_THREADS_ABORTED,
00049 W_THREADS_FINISHED
00050 };
00051
00052
00053
00054
00055 enum WThreadedFunctionNbThreads
00056 {
00057 W_AUTOMATIC_NB_THREADS = 0
00058 };
00059
00060
00061
00062
00063
00064
00065 class OWCOMMON_EXPORT WThreadedFunctionBase
00066 {
00067
00068 typedef boost::signal< void ( WException const& ) > ExceptionSignal;
00069
00070 public:
00071
00072
00073 typedef boost::function< void ( WException const& ) > ExceptionFunction;
00074
00075
00076
00077
00078 WThreadedFunctionBase();
00079
00080
00081
00082
00083
00084
00085 virtual ~WThreadedFunctionBase();
00086
00087
00088
00089
00090 virtual void run() = 0;
00091
00092
00093
00094
00095
00096 virtual void stop() = 0;
00097
00098
00099
00100
00101 virtual void wait() = 0;
00102
00103
00104
00105
00106
00107
00108 WThreadedFunctionStatus status();
00109
00110
00111
00112
00113
00114
00115 boost::shared_ptr< WCondition > getThreadsDoneCondition();
00116
00117
00118
00119
00120
00121
00122 void subscribeExceptionSignal( ExceptionFunction func );
00123
00124 protected:
00125
00126
00127
00128 WThreadedFunctionBase( WThreadedFunctionBase const& );
00129
00130
00131
00132
00133
00134
00135 WThreadedFunctionBase& operator = ( WThreadedFunctionBase const& );
00136
00137
00138 boost::shared_ptr< WCondition > m_doneCondition;
00139
00140
00141 ExceptionSignal m_exceptionSignal;
00142
00143
00144 WSharedObject< WThreadedFunctionStatus > m_status;
00145 };
00146
00147
00148
00149
00150
00151
00152
00153
00154
00155
00156
00157
00158
00159
00160
00161
00162
00163
00164
00165
00166
00167
00168
00169
00170
00171
00172
00173
00174
00175 template< class Function_T >
00176 class WThreadedFunction : public WThreadedFunctionBase
00177 {
00178
00179 typedef boost::signal< void ( WException const& ) > ExceptionSignal;
00180
00181 public:
00182
00183
00184 typedef boost::function< void ( WException const& ) > ExceptionFunction;
00185
00186
00187
00188
00189
00190
00191
00192
00193
00194 WThreadedFunction( std::size_t numThreads, boost::shared_ptr< Function_T > function );
00195
00196
00197
00198
00199
00200
00201 virtual ~WThreadedFunction();
00202
00203
00204
00205
00206 virtual void run();
00207
00208
00209
00210
00211
00212 virtual void stop();
00213
00214
00215
00216
00217 virtual void wait();
00218
00219 private:
00220
00221
00222
00223 WThreadedFunction( WThreadedFunction const& );
00224
00225
00226
00227
00228
00229
00230 WThreadedFunction& operator = ( WThreadedFunction const& );
00231
00232
00233
00234
00235 void handleThreadDone();
00236
00237
00238
00239
00240
00241
00242 void handleThreadException( WException const& e );
00243
00244
00245 std::size_t m_numThreads;
00246
00247
00248
00249 std::vector< boost::shared_ptr< WWorkerThread< Function_T > > > m_threads;
00250
00251
00252 boost::shared_ptr< Function_T > m_func;
00253
00254
00255 WSharedObject< std::size_t > m_threadsDone;
00256 };
00257
00258 template< class Function_T >
00259 WThreadedFunction< Function_T >::WThreadedFunction( std::size_t numThreads, boost::shared_ptr< Function_T > function )
00260 : WThreadedFunctionBase(),
00261 m_numThreads( numThreads ),
00262 m_threads(),
00263 m_func( function ),
00264 m_threadsDone()
00265 {
00266 if( !m_func )
00267 {
00268 throw WException( std::string( "No valid thread function pointer." ) );
00269 }
00270
00271
00272 if( m_numThreads == W_AUTOMATIC_NB_THREADS )
00273 {
00274 m_numThreads = 1;
00275 while( m_numThreads < boost::thread::hardware_concurrency() / 2 && m_numThreads < 1024 )
00276 {
00277 m_numThreads *= 2;
00278 }
00279 }
00280
00281
00282 m_threadsDone.getWriteTicket()->get() = 0;
00283
00284
00285 for( std::size_t k = 0; k < m_numThreads; ++k )
00286 {
00287 boost::shared_ptr< WWorkerThread< Function_T > > t( new WWorkerThread< Function_T >( m_func, k, m_numThreads ) );
00288 t->subscribeStopSignal( boost::bind( &WThreadedFunction::handleThreadDone, this ) );
00289 t->subscribeExceptionSignal( boost::bind( &WThreadedFunction::handleThreadException, this, _1 ) );
00290 m_threads.push_back( t );
00291 }
00292 }
00293
00294 template< class Function_T >
00295 WThreadedFunction< Function_T >::~WThreadedFunction()
00296 {
00297 stop();
00298 }
00299
00300 template< class Function_T >
00301 void WThreadedFunction< Function_T >::run()
00302 {
00303
00304 m_threadsDone.getWriteTicket()->get() = 0;
00305
00306 m_status.getWriteTicket()->get() = W_THREADS_RUNNING;
00307
00308 for( std::size_t k = 0; k < m_numThreads; ++k )
00309 {
00310 m_threads[ k ]->run();
00311 }
00312 }
00313
00314 template< class Function_T >
00315 void WThreadedFunction< Function_T >::stop()
00316 {
00317
00318 m_status.getWriteTicket()->get() = W_THREADS_STOP_REQUESTED;
00319
00320 typename std::vector< boost::shared_ptr< WWorkerThread< Function_T > > >::iterator it;
00321
00322 for( it = m_threads.begin(); it != m_threads.end(); ++it )
00323 {
00324 ( *it )->requestStop();
00325 }
00326 }
00327
00328 template< class Function_T >
00329 void WThreadedFunction< Function_T >::wait()
00330 {
00331 typename std::vector< boost::shared_ptr< WWorkerThread< Function_T > > >::iterator it;
00332
00333 for( it = m_threads.begin(); it != m_threads.end(); ++it )
00334 {
00335 ( *it )->wait();
00336 }
00337 }
00338
00339 template< class Function_T >
00340 void WThreadedFunction< Function_T >::handleThreadDone()
00341 {
00342 typedef typename WSharedObject< std::size_t >::WriteTicket WT;
00343
00344 WT t = m_threadsDone.getWriteTicket();
00345 WAssert( t->get() < m_numThreads, "" );
00346 ++t->get();
00347 std::size_t k = t->get();
00348 t = WT();
00349
00350 if( m_numThreads == k )
00351 {
00352 typedef typename WSharedObject< WThreadedFunctionStatus >::WriteTicket ST;
00353 ST s = m_status.getWriteTicket();
00354 if( s->get() == W_THREADS_RUNNING )
00355 {
00356 s->get() = W_THREADS_FINISHED;
00357 }
00358 else if( s->get() == W_THREADS_STOP_REQUESTED )
00359 {
00360 s->get() = W_THREADS_ABORTED;
00361 }
00362 else
00363 {
00364 throw WException( std::string( "Invalid status change." ) );
00365 }
00366 m_doneCondition->notify();
00367 }
00368 }
00369
00370 template< class Function_T >
00371 void WThreadedFunction< Function_T >::handleThreadException( WException const& e )
00372 {
00373
00374 typedef typename WSharedObject< WThreadedFunctionStatus >::WriteTicket WT;
00375 WT w = m_status.getWriteTicket();
00376 WAssert( w->get() != W_THREADS_FINISHED &&
00377 w->get() != W_THREADS_ABORTED, "" );
00378 if( w->get() == W_THREADS_RUNNING )
00379 {
00380 w->get() = W_THREADS_STOP_REQUESTED;
00381 }
00382
00383 w = WT();
00384
00385 handleThreadDone();
00386
00387 m_exceptionSignal( e );
00388 }
00389
00390 #endif // WTHREADEDFUNCTION_H