src/mainapp/thread_manager.cpp

   1
   2 /***************************************************************************
   3  *  thread_manager.cpp - Thread manager
   4  *
   5  *  Created: Thu Nov  3 19:11:31 2006 (on train to Cologne)
   6  *  Copyright  2006-2009  Tim Niemueller [www.niemueller.de]
   7  *
   8  ****************************************************************************/
   9
  10 /*  This program is free software; you can redistribute it and/or modify
  11  *  it under the terms of the GNU General Public License as published by
  12  *  the Free Software Foundation; either version 2 of the License, or
  13  *  (at your option) any later version. A runtime exception applies to
  14  *  this software (see LICENSE.GPL_WRE file mentioned below for details).
  15  *
  16  *  This program is distributed in the hope that it will be useful,
  17  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  18  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  19  *  GNU Library General Public License for more details.
  20  *
  21  *  Read the full text in the LICENSE.GPL_WRE file in the doc directory.
  22  */
  23
  24 #include <mainapp/thread_manager.h>
  25 #include <core/threading/thread.h>
  26 #include <core/threading/mutex_locker.h>
  27 #include <core/threading/wait_condition.h>
  28 #include <core/threading/thread_initializer.h>
  29 #include <core/threading/thread_finalizer.h>
  30 #include <core/exceptions/software.h>
  31 #include <core/exceptions/system.h>
  32
  33 #include <aspect/blocked_timing.h>
  34
  35 using namespace fawkes;
  36
  37 /** @class FawkesThreadManager mainapp/thread_manager.h
  38  * Thread Manager.
  39  * This class provides a manager for the threads. Threads are memorized by
  40  * their wakeup hook. When the thread manager is deleted, all threads are
  41  * appropriately cancelled, joined and deleted. Thus the thread manager
  42  * can be used for "garbage collection" of threads.
  43  *
  44  * The thread manager allows easy wakeup of threads of a given wakeup hook.
  45  *
  46  * The thread manager needs a thread initializer. Each thread that is added
  47  * to the thread manager is initialized with this. The runtime type information
  48  * (RTTI) supplied by C++ can be used to initialize threads if appropriate
  49  * (if the thread has certain aspects that need special treatment).
  50  *
  51  * @author Tim Niemueller
  52  */
  53
  54 FawkesThreadManager::FawkesThreadManagerAspectCollector::FawkesThreadManagerAspectCollector(FawkesThreadManager *parent_manager)
  55 {
  56   __parent_manager = parent_manager;
  57 }
  58
  59
  60 void
  61 FawkesThreadManager::FawkesThreadManagerAspectCollector::add(ThreadList &tl)
  62 {
  63   BlockedTimingAspect *timed_thread;
  64
  65   for (ThreadList::iterator i = tl.begin(); i != tl.end(); ++i) {
  66     if ( (timed_thread = dynamic_cast<BlockedTimingAspect *>(*i)) != NULL ) {
  67       throw IllegalArgumentException("ThreadProducerAspect may not add threads with BlockedTimingAspect");
  68     }
  69   }
  70
  71   __parent_manager->add_maybelocked(tl, /* lock */ false);
  72 }
  73
  74
  75 void
  76 FawkesThreadManager::FawkesThreadManagerAspectCollector::add(Thread *t)
  77 {
  78   BlockedTimingAspect *timed_thread;
  79
  80   if ( (timed_thread = dynamic_cast<BlockedTimingAspect *>(t)) != NULL ) {
  81     throw IllegalArgumentException("ThreadProducerAspect may not add threads with BlockedTimingAspect");
  82   }
  83
  84   __parent_manager->add_maybelocked(t, /* lock */ false);
  85 }
  86
  87
  88 void
  89 FawkesThreadManager::FawkesThreadManagerAspectCollector::remove(ThreadList &tl)
  90 {
  91   BlockedTimingAspect *timed_thread;
  92
  93   for (ThreadList::iterator i = tl.begin(); i != tl.end(); ++i) {
  94     if ( (timed_thread = dynamic_cast<BlockedTimingAspect *>(*i)) != NULL ) {
  95       throw IllegalArgumentException("ThreadProducerAspect may not remove threads with BlockedTimingAspect");
  96     }
  97   }
  98
  99   __parent_manager->remove_maybelocked(tl, /* lock */ false);
 100 }
 101
 102
 103 void
 104 FawkesThreadManager::FawkesThreadManagerAspectCollector::remove(Thread *t)
 105 {
 106   BlockedTimingAspect *timed_thread;
 107
 108   if ( (timed_thread = dynamic_cast<BlockedTimingAspect *>(t)) != NULL ) {
 109     throw IllegalArgumentException("ThreadProducerAspect may not remove threads with BlockedTimingAspect");
 110   }
 111
 112   __parent_manager->remove_maybelocked(t, /* lock */ false);
 113 }
 114
 115
 116 void
 117 FawkesThreadManager::FawkesThreadManagerAspectCollector::force_remove(fawkes::ThreadList &tl)
 118 {
 119   throw AccessViolationException("ThreadManagerAspect threads may not force removal of threads");
 120 }
 121
 122 void
 123 FawkesThreadManager::FawkesThreadManagerAspectCollector::force_remove(fawkes::Thread *t)
 124 {
 125   throw AccessViolationException("ThreadManagerAspect threads may not force removal of threads");
 126 }
 127
 128
 129 /** Constructor.
 130  */
 131 FawkesThreadManager::FawkesThreadManager()
 132 {
 133   initializer = NULL;
 134   finalizer   = NULL;
 135   threads.clear();
 136   waitcond_timedthreads = new WaitCondition();
 137   __interrupt_timed_thread_wait = false;
 138   __aspect_collector = new FawkesThreadManagerAspectCollector(this);
 139 }
 140
 141
 142 /** Destructor. */
 143 FawkesThreadManager::~FawkesThreadManager()
 144 {
 145   // stop all threads, we call finalize, and we run through it as long as there are
 146   // still running threads, after that, we force the thread's death.
 147   for (tit = threads.begin(); tit != threads.end(); ++tit) {
 148     (*tit).second.force_stop(finalizer);
 149   }
 150   untimed_threads.force_stop(finalizer);
 151   threads.clear();
 152
 153   delete waitcond_timedthreads;
 154   delete __aspect_collector;
 155 }
 156
 157
 158 /** Set initializer/finalizer.
 159  * This method has to be called before any thread is added/removed.
 160  * @param initializer thread initializer
 161  * @param finalizer thread finalizer
 162  */
 163 void
 164 FawkesThreadManager::set_inifin(ThreadInitializer *initializer, ThreadFinalizer *finalizer)
 165 {
 166   this->initializer = initializer;
 167   this->finalizer   = finalizer;
 168 }
 169
 170
 171 /** Remove the given thread from internal structures.
 172  * Thread is removed from the internal structures. If the thread has the
 173  * BlockedTimingAspect then the hook is added to the changed list.
 174  *
 175  * @param t thread to remove
 176  * @param changed list of changed hooks, appropriate hook is added if necessary
 177  */
 178 void
 179 FawkesThreadManager::internal_remove_thread(Thread *t)
 180 {
 181   BlockedTimingAspect *timed_thread;
 182
 183   if ( (timed_thread = dynamic_cast<BlockedTimingAspect *>(t)) != NULL ) {
 184     // find thread and remove
 185     BlockedTimingAspect::WakeupHook hook = timed_thread->blockedTimingAspectHook();
 186     if ( threads.find(hook) != threads.end() ) {
 187       threads[hook].remove_locked(t);
 188       if (threads[hook].empty())  threads.erase(hook);
 189     }
 190   } else {
 191     untimed_threads.remove_locked(t);
 192   }
 193 }
 194
 195
 196 /** Add the given thread to internal structures.
 197  * Thread is added to the internal structures. If the thread has the
 198  * BlockedTimingAspect then the hook is added to the changed list.
 199  *
 200  * @param t thread to add
 201  * @param changed list of changed hooks, appropriate hook is added if necessary
 202  */
 203 void
 204 FawkesThreadManager::internal_add_thread(Thread *t)
 205 {
 206   BlockedTimingAspect *timed_thread;
 207   if ( (timed_thread = dynamic_cast<BlockedTimingAspect *>(t)) != NULL ) {
 208     BlockedTimingAspect::WakeupHook hook = timed_thread->blockedTimingAspectHook();
 209
 210     if ( threads.find(hook) == threads.end() ) {
 211       threads[hook].set_name("FawkesThreadManagerList Hook %i", hook);
 212       threads[hook].set_maintain_barrier(true);
 213     }
 214     threads[hook].push_back_locked(t);
 215
 216     waitcond_timedthreads->wake_all();
 217   } else {
 218     untimed_threads.push_back_locked(t);
 219   }
 220 }
 221
 222
 223 /** Add threads.
 224  * Add the given threads to the thread manager. The threads are initialised
 225  * as appropriate and started. See the class documentation for supported
 226  * specialisations of threads and the performed initialisation steps.
 227  * If the thread initializer cannot initalize one or more threads no thread
 228  * is added. In this regard the operation is atomic, either all threads are
 229  * added or none.
 230  * @param tl thread list with threads to add
 231  * @exception CannotInitializeThreadException thrown if at least one of the
 232  * threads could not be initialised
 233  */
 234 void
 235 FawkesThreadManager::add_maybelocked(ThreadList &tl, bool lock)
 236 {
 237   if ( ! (initializer && finalizer) ) {
 238     throw NullPointerException("FawkesThreadManager: initializer/finalizer not set");
 239   }
 240
 241   if ( tl.sealed() ) {
 242     throw Exception("Not accepting new threads from list that is not fresh, "
 243                     "list '%s' already sealed", tl.name());
 244   }
 245
 246   tl.lock();
 247
 248   // Try to initialise all threads
 249   try {
 250     tl.init(initializer, finalizer);
 251   } catch (Exception &e) {
 252     tl.unlock();
 253     throw;
 254   }
 255
 256   tl.seal();
 257   tl.start();
 258
 259   // All thread initialized, now add threads to internal structure
 260   MutexLocker locker(threads.mutex(), lock);
 261   for (ThreadList::iterator i = tl.begin(); i != tl.end(); ++i) {
 262     internal_add_thread(*i);
 263   }
 264
 265   tl.unlock();
 266 }
 267
 268
 269 /** Add one thread.
 270  * Add the given thread to the thread manager. The thread is initialized
 271  * as appropriate and started. See the class documentation for supported
 272  * specialisations of threads and the performed initialisation steps.
 273  * If the thread initializer cannot initalize the thread it is not added.
 274  * @param thread thread to add
 275  * @param lock if true the environment is locked before adding the thread
 276  * @exception CannotInitializeThreadException thrown if at least the
 277  * thread could not be initialised
 278  */
 279 void
 280 FawkesThreadManager::add_maybelocked(Thread *thread, bool lock)
 281 {
 282   if ( thread == NULL ) {
 283     throw NullPointerException("FawkesThreadMananger: cannot add NULL as thread");
 284   }
 285
 286   if ( ! (initializer && finalizer) ) {
 287     throw NullPointerException("FawkesThreadManager: initializer/finalizer not set");
 288   }
 289
 290   try {
 291     initializer->init(thread);
 292   } catch (CannotInitializeThreadException &e) {
 293     e.append("Adding thread in FawkesThreadManager failed");
 294     throw;
 295   }
 296
 297   thread->start();
 298   MutexLocker locker(threads.mutex(), lock);
 299   internal_add_thread(thread);
 300 }
 301
 302
 303 /** Remove the given threads.
 304  * The thread manager tries to finalize and stop the threads and then removes the
 305  * threads from the internal structures.
 306  *
 307  * This may fail if at least one thread of the given list cannot be finalized, for
 308  * example if prepare_finalize() returns false or if the thread finalizer cannot
 309  * finalize the thread. In this case a CannotFinalizeThreadException is thrown.
 310  *
 311  * @param tl threads to remove.
 312  * @exception CannotFinalizeThreadException At least one thread cannot be safely
 313  * finalized
 314  * @exception ThreadListNotSealedException if the given thread lits tl is not
 315  * sealed the thread manager will refuse to remove it
 316  */
 317 void
 318 FawkesThreadManager::remove_maybelocked(ThreadList &tl, bool lock)
 319 {
 320   if ( ! (initializer && finalizer) ) {
 321     throw NullPointerException("FawkesThreadManager: initializer/finalizer not set");
 322   }
 323
 324
 325   if ( ! tl.sealed() ) {
 326     throw ThreadListNotSealedException("(FawkesThreadManager) Cannot remove unsealed thread "
 327                                        "list. Not accepting unsealed list '%s' for removal",
 328                                        tl.name());
 329   }
 330
 331   tl.lock();
 332   MutexLocker locker(threads.mutex(), lock);
 333
 334   try {
 335     if ( ! tl.prepare_finalize(finalizer) ) {
 336       tl.cancel_finalize();
 337       tl.unlock();
 338       throw CannotFinalizeThreadException("One or more threads in list '%s' cannot be "
 339                                           "finalized", tl.name());
 340     }
 341   } catch (CannotFinalizeThreadException &e) {
 342     tl.unlock();
 343     throw;
 344   } catch (Exception &e) {
 345     tl.unlock();
 346     e.append("One or more threads in list '%s' cannot be finalized", tl.name());
 347     throw CannotFinalizeThreadException(e);
 348   }
 349
 350   tl.stop();
 351   tl.finalize(finalizer);
 352
 353   for (ThreadList::iterator i = tl.begin(); i != tl.end(); ++i) {
 354     internal_remove_thread(*i);
 355   }
 356
 357   tl.unlock();
 358 }
 359
 360
 361 /** Remove the given thread.
 362  * The thread manager tries to finalize and stop the thread and then removes the
 363  * thread from the internal structures.
 364  *
 365  * This may fail if the thread cannot be finalized, for
 366  * example if prepare_finalize() returns false or if the thread finalizer cannot
 367  * finalize the thread. In this case a CannotFinalizeThreadException is thrown.
 368  *
 369  * @param thread thread to remove.
 370  * @exception CannotFinalizeThreadException At least one thread cannot be safely
 371  * finalized
 372  */
 373 void
 374 FawkesThreadManager::remove_maybelocked(Thread *thread, bool lock)
 375 {
 376   if ( thread == NULL ) return;
 377
 378   if ( ! (initializer && finalizer) ) {
 379     throw NullPointerException("FawkesThreadManager: initializer/finalizer not set");
 380   }
 381
 382   MutexLocker locker(threads.mutex(), lock);
 383   try {
 384     if ( ! thread->prepare_finalize() ) {
 385       thread->cancel_finalize();
 386       throw CannotFinalizeThreadException("Thread '%s'cannot be finalized", thread->name());
 387     }
 388   } catch (CannotFinalizeThreadException &e) {
 389     e.append("FawkesThreadManager cannot stop thread '%s'", thread->name());
 390     thread->cancel_finalize();
 391     throw;
 392   }
 393
 394   thread->cancel();
 395   thread->join();
 396   finalizer->finalize(thread);
 397   thread->finalize();
 398
 399   internal_remove_thread(thread);
 400 }
 401
 402
 403
 404
 405 /** Force removal of the given threads.
 406  * The thread manager tries to finalize and stop the threads and then removes the
 407  * threads from the internal structures.
 408  *
 409  * This will succeed even if a thread of the given list cannot be finalized, for
 410  * example if prepare_finalize() returns false or if the thread finalizer cannot
 411  * finalize the thread.
 412  *
 413  * <b>Caution, using this function may damage your robot.</b>
 414  *
 415  * @param tl threads to remove.
 416  * @exception ThreadListNotSealedException if the given thread lits tl is not
 417  * sealed the thread manager will refuse to remove it
 418  * The threads are removed from thread manager control. The threads will be stopped
 419  * before they are removed (may cause unpredictable results otherwise).
 420  */
 421 void
 422 FawkesThreadManager::force_remove(ThreadList &tl)
 423 {
 424   if ( ! tl.sealed() ) {
 425     throw ThreadListNotSealedException("Not accepting unsealed list '%s' for removal",
 426                                        tl.name());
 427   }
 428
 429   tl.lock();
 430   threads.mutex()->stopby();
 431   tl.force_stop(finalizer);
 432
 433   for (ThreadList::iterator i = tl.begin(); i != tl.end(); ++i) {
 434     internal_remove_thread(*i);
 435   }
 436
 437   tl.unlock();
 438 }
 439
 440
 441 /** Force removal of the given thread.
 442  * The thread manager tries to finalize and stop the thread and then removes the
 443  * thread from the internal structures.
 444  *
 445  * This will succeed even if the thread cannot be finalized, for
 446  * example if prepare_finalize() returns false or if the thread finalizer cannot
 447  * finalize the thread.
 448  *
 449  * <b>Caution, using this function may damage your robot.</b>
 450  *
 451  * @param thread thread to remove.
 452  * @exception ThreadListNotSealedException if the given thread lits tl is not
 453  * sealed the thread manager will refuse to remove it
 454  * The threads are removed from thread manager control. The threads will be stopped
 455  * before they are removed (may cause unpredictable results otherwise).
 456  */
 457 void
 458 FawkesThreadManager::force_remove(fawkes::Thread *thread)
 459 {
 460   MutexLocker lock(threads.mutex());
 461   try {
 462     thread->prepare_finalize();
 463   } catch (Exception &e) {
 464     // ignore
 465   }
 466
 467   thread->cancel();
 468   thread->join();
 469   if (finalizer) finalizer->finalize(thread);
 470   thread->finalize();
 471
 472   internal_remove_thread(thread);
 473 }
 474
 475
 476 void
 477 FawkesThreadManager::wakeup_and_wait(BlockedTimingAspect::WakeupHook hook,
 478                                      unsigned int timeout_usec)
 479 {
 480   MutexLocker lock(threads.mutex());
 481
 482   unsigned int timeout_sec = 0;
 483   if (timeout_usec >= 1000000) {
 484     timeout_sec   = timeout_usec / 1000000;
 485     timeout_usec -= timeout_sec  * 1000000;
 486   }
 487
 488   // Note that the following lines might throw an exception, we just pass it on
 489   if ( threads.find(hook) != threads.end() ) {
 490     threads[hook].wakeup_and_wait(timeout_sec, timeout_usec * 1000);
 491   }
 492 }
 493
 494
 495 void
 496 FawkesThreadManager::wakeup(BlockedTimingAspect::WakeupHook hook, Barrier *barrier)
 497 {
 498   MutexLocker lock(threads.mutex());
 499
 500   if ( threads.find(hook) != threads.end() ) {
 501     if ( barrier ) {
 502       threads[hook].wakeup(barrier);
 503     } else {
 504       threads[hook].wakeup();
 505     }
 506     if ( threads[hook].size() == 0 ) {
 507       threads.erase(hook);
 508     }
 509   }
 510 }
 511
 512
 513 void
 514 FawkesThreadManager::try_recover(std::list<std::string> &recovered_threads)
 515 {
 516   threads.lock();
 517   for (tit = threads.begin(); tit != threads.end(); ++tit) {
 518     tit->second.try_recover(recovered_threads);
 519   }
 520   threads.unlock();
 521 }
 522
 523
 524 bool
 525 FawkesThreadManager::timed_threads_exist()
 526 {
 527   return (threads.size() > 0);
 528 }
 529
 530
 531 void
 532 FawkesThreadManager::wait_for_timed_threads()
 533 {
 534   __interrupt_timed_thread_wait = false;
 535   waitcond_timedthreads->wait();
 536   if ( __interrupt_timed_thread_wait ) {
 537     __interrupt_timed_thread_wait = false;
 538     throw InterruptedException("Waiting for timed threads was interrupted");
 539   }
 540 }
 541
 542 void
 543 FawkesThreadManager::interrupt_timed_thread_wait()
 544 {
 545   __interrupt_timed_thread_wait = true;
 546   waitcond_timedthreads->wake_all();
 547 }
 548
 549
 550
 551 /** Get a thread collector to be used for an aspect initializer.
 552  * @return thread collector instance to use for ThreadProducerAspect.
 553  */
 554 ThreadCollector *
 555 FawkesThreadManager::aspect_collector() const
 556 {
 557   return __aspect_collector;
 558 }