src/OSspecific/POSIX/multiThreader/threadHandlerI.H

   1 /*---------------------------------------------------------------------------*\
   2   =========                 |
   3   \\      /  F ield         | foam-extend: Open Source CFD
   4    \\    /   O peration     | Version:     3.2
   5     \\  /    A nd           | Web:         http://www.foam-extend.org
   6      \\/     M anipulation  | For copyright notice see file Copyright
   7 -------------------------------------------------------------------------------
   8 License
   9     This file is part of foam-extend.
  10
  11     foam-extend is free software: you can redistribute it and/or modify it
  12     under the terms of the GNU General Public License as published by the
  13     Free Software Foundation, either version 3 of the License, or (at your
  14     option) any later version.
  15
  16     foam-extend is distributed in the hope that it will be useful, but
  17     WITHOUT ANY WARRANTY; without even the implied warranty of
  18     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  19     General Public License for more details.
  20
  21     You should have received a copy of the GNU General Public License
  22     along with foam-extend.  If not, see <http://www.gnu.org/licenses/>.
  23
  24 \*---------------------------------------------------------------------------*/
  25
  26 // * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
  27
  28 namespace Foam
  29 {
  30
  31 // * * * * * * * * * * * * * * * * Constructors  * * * * * * * * * * * * * * //
  32
  33 // Constructor for threadHandler
  34 template<class T>
  35 threadHandler<T>::threadHandler
  36 (
  37     T& tRef,
  38     const multiThreader& threader
  39 )
  40 :
  41     tRef_(tRef),
  42     threader_(threader),
  43     argList_(0),
  44     nThreads_(threader.getNumThreads()),
  45     pthreadID_(pthread_self()),
  46     master_(false),
  47     predicate_(false)
  48 {}
  49
  50
  51 // * * * * * * * * * * * * * * * *  Destructors  * * * * * * * * * * * * * * //
  52
  53 template<class T>
  54 threadHandler<T>::~threadHandler()
  55 {
  56     // Null the argument list pointer copies
  57     // to avoid multiple deallocations.
  58     forAll(argList_, indexI)
  59     {
  60         argList_[indexI] = NULL;
  61     }
  62 }
  63
  64
  65 // * * * * * * * * * * * * * * * Member Functions  * * * * * * * * * * * * * //
  66
  67 // Return a reference to the template class
  68 template<class T>
  69 inline T& threadHandler<T>::reference()
  70 {
  71     return tRef_;
  72 }
  73
  74
  75 // Set a size for the argument list
  76 template<class T>
  77 inline void threadHandler<T>::setSize
  78 (
  79     const label size
  80 )
  81 {
  82     argList_.setSize(size, NULL);
  83 }
  84
  85
  86 // Set a argument pointer for a particular index
  87 template<class T>
  88 inline void threadHandler<T>::set
  89 (
  90     const label index,
  91     void* argPtr
  92 )
  93 {
  94     if (!argList_.size())
  95     {
  96         FatalErrorIn("threadHandler<T>::set")
  97             << "Attempt to access element from zero sized list"
  98             << abort(FatalError);
  99     }
 100     else
 101     if (index < 0 || index >= argList_.size())
 102     {
 103         FatalErrorIn("threadHandler<T>::set")
 104             << "index " << index << " out of range 0 ... "
 105             << argList_.size()-1
 106             << abort(FatalError);
 107     }
 108
 109     argList_[index] = argPtr;
 110 }
 111
 112
 113 // Return a reference to the multiThreader
 114 template<class T>
 115 inline const multiThreader& threadHandler<T>::threader() const
 116 {
 117     return threader_;
 118 }
 119
 120
 121 // Return the number of threads
 122 template<class T>
 123 inline label threadHandler<T>::nThreads() const
 124 {
 125     return nThreads_;
 126 }
 127
 128
 129 // Designate as master thread
 130 template<class T>
 131 inline void threadHandler<T>::setMaster() const
 132 {
 133     master_ = true;
 134 }
 135
 136
 137 // Designate as slave thread
 138 template<class T>
 139 inline void threadHandler<T>::setSlave() const
 140 {
 141     master_ = false;
 142 }
 143
 144
 145 // Is this a master thread?
 146 template<class T>
 147 inline bool threadHandler<T>::master() const
 148 {
 149     return (master_ == true);
 150 }
 151
 152
 153 // Is this a slave thread?
 154 template<class T>
 155 inline bool threadHandler<T>::slave() const
 156 {
 157     return !master();
 158 }
 159
 160
 161 // Lock this thread
 162 template<class T>
 163 inline void threadHandler<T>::lock
 164 (
 165     const signalType sType
 166 ) const
 167 {
 168     if (sType == START)
 169     {
 170         startMutex_.lock();
 171     }
 172     else
 173     if (sType == STOP)
 174     {
 175         stopMutex_.lock();
 176     }
 177 }
 178
 179
 180 // Unlock this thread
 181 template<class T>
 182 inline void threadHandler<T>::unlock
 183 (
 184     const signalType sType
 185 ) const
 186 {
 187     if (sType == START)
 188     {
 189         startMutex_.unlock();
 190     }
 191     else
 192     if (sType == STOP)
 193     {
 194         stopMutex_.unlock();
 195     }
 196 }
 197
 198
 199 // Send signal to a waiting conditional
 200 template<class T>
 201 inline void threadHandler<T>::sendSignal
 202 (
 203     const signalType sType
 204 ) const
 205 {
 206     lock(sType);
 207
 208     if (predicate(sType))
 209     {
 210         InfoIn("threadHandler::sendSignal()")
 211             << "Predicate is already set."
 212             << endl;
 213     }
 214     else
 215     {
 216         // Set predicate before signalling
 217         setPredicate(sType);
 218     }
 219
 220     if (sType == START)
 221     {
 222         threader().signal
 223         (
 224             startConditional_
 225         );
 226     }
 227     else
 228     if (sType == STOP)
 229     {
 230         threader().signal
 231         (
 232             stopConditional_
 233         );
 234     }
 235     else
 236     {
 237         FatalErrorIn("threadHandler::sendSignal()")
 238             << "Undefined enumerant."
 239             << abort(FatalError);
 240     }
 241
 242     unlock(sType);
 243 }
 244
 245
 246 // Wait for signal
 247 template<class T>
 248 inline void threadHandler<T>::waitForSignal
 249 (
 250     const signalType sType
 251 ) const
 252 {
 253     if (sType == START)
 254     {
 255         threader().waitForCondition
 256         (
 257             startConditional_,
 258             startMutex_
 259         );
 260     }
 261     else
 262     if (sType == STOP)
 263     {
 264         threader().waitForCondition
 265         (
 266             stopConditional_,
 267             stopMutex_
 268         );
 269     }
 270     else
 271     {
 272         FatalErrorIn("threadHandler::waitForSignal()")
 273             << "Undefined enumerant."
 274             << abort(FatalError);
 275     }
 276
 277     if (!predicate(sType))
 278     {
 279         FatalErrorIn("threadHandler::waitForSignal()")
 280             << "Spurious wake-up."
 281             << abort(FatalError);
 282     }
 283
 284     unsetPredicate(sType);
 285
 286     // Unlock the acquired mutex
 287     unlock(sType);
 288 }
 289
 290
 291 // Return state of the predicate variable
 292 template<class T>
 293 inline bool threadHandler<T>::predicate
 294 (
 295     const signalType sType
 296 ) const
 297 {
 298     return predicate_[sType];
 299 }
 300
 301
 302 // Set the predicate variable
 303 template<class T>
 304 inline void threadHandler<T>::setPredicate
 305 (
 306     const signalType sType
 307 ) const
 308 {
 309     predicate_[sType] = true;
 310 }
 311
 312
 313 // Unset the predicate variable
 314 template<class T>
 315 inline void threadHandler<T>::unsetPredicate
 316 (
 317     const signalType sType
 318 ) const
 319 {
 320     predicate_[sType] = false;
 321 }
 322
 323
 324 // Return the ID
 325 template<class T>
 326 inline void threadHandler<T>::setID(const pthread_t& pt)
 327 {
 328     pthreadID_ = pt;
 329 }
 330
 331
 332 // Return the ID
 333 template<class T>
 334 inline pthread_t threadHandler<T>::ID() const
 335 {
 336     return pthreadID_;
 337 }
 338
 339
 340 // Does the calling thread correspond to this handler?
 341 template<class T>
 342 inline bool threadHandler<T>::self() const
 343 {
 344     return pthread_equal(ID(), pthread_self());
 345 }
 346
 347
 348 // Return an argument pointer at a particular index
 349 template<class T>
 350 inline void * threadHandler<T>::operator()
 351 (
 352     const label index
 353 )
 354 {
 355     if (!argList_.size())
 356     {
 357         FatalErrorIn("threadHandler<T>::operator()")
 358             << "Attempt to access element from zero sized list"
 359             << abort(FatalError);
 360     }
 361     else
 362     if (index < 0 || index >= argList_.size())
 363     {
 364         FatalErrorIn("threadHandler<T>::operator()")
 365             << "index " << index << " out of range 0 ... "
 366             << argList_.size()-1
 367             << abort(FatalError);
 368     }
 369     else
 370     if (!argList_[index])
 371     {
 372         FatalErrorIn("threadHandler<T>::operator()")
 373             << "Hanging pointer. This is not allowed."
 374             << abort(FatalError);
 375     }
 376
 377     return argList_[index];
 378 }
 379
 380
 381 // * * * * * * * * * * * * * * * Global Functions  * * * * * * * * * * * * * //
 382
 383 // Lock all threads provided by sequence
 384 template <class T>
 385 void lockThreads
 386 (
 387     const List<label>& sequence,
 388     const PtrList<threadHandler<T> >& handler
 389 )
 390 {
 391     forAll(sequence, i)
 392     {
 393         handler[sequence[i]].lock(threadHandler<T>::START);
 394         handler[sequence[i]].lock(threadHandler<T>::STOP);
 395
 396         handler[sequence[i]].unsetPredicate(threadHandler<T>::START);
 397         handler[sequence[i]].unsetPredicate(threadHandler<T>::STOP);
 398     }
 399 }
 400
 401
 402 // Synchronize all threads provided by sequence
 403 template <class T>
 404 void synchronizeThreads
 405 (
 406     const List<label>& sequence,
 407     const PtrList<threadHandler<T> >& handler
 408 )
 409 {
 410     forAll(sequence, i)
 411     {
 412         // Wait for a signal from this thread before moving on.
 413         handler[sequence[i]].waitForSignal(threadHandler<T>::STOP);
 414     }
 415 }
 416
 417
 418 // Execute threads for the submitted static function by sequence
 419 template <class T>
 420 void executeThreads
 421 (
 422     const List<label>& sequence,
 423     PtrList<threadHandler<T> >& handler,
 424     void (*tFunction)(void*)
 425 )
 426 {
 427     if (!handler.size())
 428     {
 429         FatalErrorIn
 430         (
 431             "\n\n"
 432             "template <class T>\n"
 433             "void executeThreads\n"
 434             "(\n"
 435             "    const List<label>& sequence,\n"
 436             "    PtrList<threadHandler<T> >& handler,\n"
 437             "    void (*tFunction)(void*)\n"
 438             ")\n"
 439         )
 440             << "Empty handler list."
 441             << abort(FatalError);
 442     }
 443
 444     // Fetch threader reference
 445     const multiThreader& threader = handler[0].threader();
 446
 447     // Lock slave threads by sequence
 448     lockThreads(sequence, handler);
 449
 450     forAll(sequence, i)
 451     {
 452         // Submit jobs to the work queue
 453         threader.addToWorkQueue(tFunction, &(handler[sequence[i]]));
 454
 455         // Wait for a signal from this thread before moving on.
 456         handler[sequence[i]].waitForSignal(threadHandler<T>::START);
 457     }
 458
 459     // Synchronize threads
 460     synchronizeThreads(sequence, handler);
 461 }
 462
 463 // * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
 464
 465 } // End namespace Foam
 466
 467 // ************************************************************************* //