core/library/processes/thread_cabinet.cpp

   1 /*****************************************************************************\
   2 *                                                                             *
   3 *  Name   : thread_cabinet                                                    *
   4 *  Author : Chris Koeritz                                                     *
   5 *                                                                             *
   6 *******************************************************************************
   7 * Copyright (c) 2000-$now By Author.  This program is free software; you can  *
   8 * redistribute it and/or modify it under the terms of the GNU General Public  *
   9 * License as published by the Free Software Foundation; either version 2 of   *
  10 * the License or (at your option) any later version.  This is online at:      *
  11 *     http://www.fsf.org/copyleft/gpl.html                                    *
  12 * Please send any updates to: fred@gruntose.com                               *
  13 \*****************************************************************************/
  14
  15 #include "ethread.h"
  16 #include "thread_cabinet.h"
  17
  18 #include <loggers/critical_events.h>
  19 #include <structures/amorph.h>
  20 #include <structures/unique_id.h>
  21 #include <timely/time_control.h>
  22
  23 #undef LOCKIT
  24 #define LOCKIT auto_synchronizer l(*_lock)
  25   // grabs the mutex for access to the list.
  26
  27 #undef LOG
  28 #define LOG(s) CLASS_EMERGENCY_LOG(program_wide_logger::get(), s)
  29
  30 //#define DEBUG_THREAD_CABINET
  31   // uncomment for noisier version.
  32
  33 using namespace basis;
  34 using namespace loggers;
  35 using namespace structures;
  36 using namespace timely;
  37
  38 namespace processes {
  39
  40 class thread_record
  41 {
  42 public:
  43   ethread *_thread;
  44   unique_int _id;
  45
  46   thread_record(const unique_int &id, ethread *t)
  47       : _thread(t), _id(id) {}
  48
  49   ~thread_record() {
  50     _thread->stop();
  51     WHACK(_thread);
  52   }
  53 };
  54
  55 //////////////
  56
  57 class thread_amorph : public amorph<thread_record>
  58 {
  59 public:
  60 };
  61
  62 //////////////
  63
  64 thread_cabinet::thread_cabinet()
  65 : _lock(new mutex),
  66   _threads(new thread_amorph),
  67   _next_id(new int_roller(1, MAXINT32 - 1)),
  68   _no_additions(0)
  69 {
  70 }
  71
  72 thread_cabinet::~thread_cabinet()
  73 {
  74   WHACK(_threads);
  75   WHACK(_lock);
  76   WHACK(_next_id);
  77 }
  78
  79 int thread_cabinet::threads() const { return _threads->elements(); }
  80
  81 unique_int thread_cabinet::add_thread(ethread *to_add, bool start_it,
  82     void *parm)
  83 {
  84 #ifdef DEBUG_THREAD_CABINET
  85   FUNCDEF("add_thread");
  86 #endif
  87   LOCKIT;
  88   if (_no_additions) {
  89 #ifdef DEBUG_THREAD_CABINET
  90     LOG("no additions flag is true; destroying the thread and failing out.");
  91 #endif
  92     // can't just leave it unhooked hanging out in space...
  93     WHACK(to_add);
  94     return 0;
  95   }
  96   int use_id = _next_id->next_id();
  97   if (start_it) {
  98     to_add->start(parm);
  99   } else {
 100 #ifdef DEBUG_THREAD_CABINET
 101     if (to_add->thread_finished())
 102       LOG(a_sprintf("thread %x is not going to be started and it "
 103           "hasn't started yet!", to_add));
 104 #endif
 105   }
 106   _threads->append(new thread_record(use_id, to_add));
 107   return use_id;
 108 }
 109
 110 bool thread_cabinet::any_running() const
 111 {
 112   LOCKIT;
 113   for (int i = 0; i < _threads->elements(); i++) {
 114     if (_threads->borrow(i)->_thread->thread_started()) return true;
 115   }
 116   return false;
 117 }
 118
 119 void thread_cabinet::start_all(void *ptr)
 120 {
 121   LOCKIT;
 122   for (int i = 0; i < _threads->elements(); i++) {
 123     if (_threads->borrow(i)->_thread->thread_finished()) {
 124       _threads->borrow(i)->_thread->start(ptr);
 125     }
 126   }
 127 }
 128
 129 void thread_cabinet::cancel_all()
 130 {
 131   FUNCDEF("cancel_all");
 132   {
 133     LOCKIT;  // short lock.
 134     _no_additions++;  // keep people from adding new threads.
 135     for (int i = 0; i < _threads->elements(); i++) {
 136       _threads->borrow(i)->_thread->cancel();
 137     }
 138   }
 139   LOCKIT;
 140   _no_additions--;  // allow new threads again.
 141   if (_no_additions < 0)
 142     continuable_error(class_name(), func, "error in logic regarding "
 143         "no additions.");
 144 }
 145
 146 void thread_cabinet::stop_all()
 147 {
 148   FUNCDEF("stop_all");
 149   {
 150     LOCKIT;  // short lock.
 151     _no_additions++;  // keep people from adding new threads.
 152   }
 153   cancel_all();  // signal all threads to leave.
 154   // pause to give them a little while to leave.
 155   time_control::sleep_ms(20);
 156   while (true) {
 157     LOCKIT;  // short lock.
 158     if (!_threads->elements()) break;  // done; nothing left.
 159     clean_debris();  // remove any that did stop.
 160     time_control::sleep_ms(20);  // snooze for a short while.
 161   }
 162   LOCKIT;
 163   _no_additions--;  // allow new threads again.
 164   if (_no_additions < 0)
 165     continuable_error(class_name(), func, "error in logic regarding "
 166         "no additions.");
 167 }
 168
 169 bool thread_cabinet::zap_thread(const unique_int &to_whack)
 170 {
 171   LOCKIT;
 172   for (int i = 0; i < _threads->elements(); i++) {
 173     if (_threads->borrow(i)->_id == to_whack) {
 174       // this is the one they want zapped.
 175       _threads->zap(i, i);
 176       return true;
 177     }
 178   }
 179   return false;
 180 }
 181
 182 bool thread_cabinet::cancel_thread(const unique_int &to_cancel)
 183 {
 184   LOCKIT;
 185   for (int i = 0; i < _threads->elements(); i++) {
 186     if (_threads->borrow(i)->_id == to_cancel) {
 187       // this is the one to signal regarding its own demise.
 188       _threads->borrow(i)->_thread->cancel();
 189       return true;
 190     }
 191   }
 192   return false;
 193 }
 194
 195 void thread_cabinet::clean_debris()
 196 {
 197 #ifdef DEBUG_THREAD_CABINET
 198   FUNCDEF("clean_debris");
 199 #endif
 200   LOCKIT;
 201   for (int i = 0; i < _threads->elements(); i++) {
 202     if (_threads->borrow(i)->_thread->thread_finished()) {
 203       // this one's no longer doing anything.
 204 #ifdef DEBUG_THREAD_CABINET
 205       LOG(a_sprintf("clearing thread %x out.", _threads->borrow(i)->_thread));
 206 #endif
 207       _threads->zap(i, i);
 208       i--;  // skip back before the whack.
 209     }
 210   }
 211 }
 212
 213 int_set thread_cabinet::thread_ids() const
 214 {
 215   LOCKIT;
 216   int_set to_return;
 217   for (int i = 0; i < _threads->elements(); i++)
 218     to_return += _threads->borrow(i)->_id.raw_id();
 219   return to_return;
 220 }
 221
 222 ethread *thread_cabinet::get_thread(int index)
 223 {
 224   LOCKIT;
 225   thread_record *rec = _threads->borrow(index);
 226   if (rec) return rec->_thread;
 227   return NIL;
 228 }
 229
 230 } //namespace.
 231