source/timex/Engine.cpp

   1 // ANNA - Anna is Not Nothingness Anymore                                                         //
   2 //                                                                                                //
   3 // (c) Copyright 2005-2015 Eduardo Ramos Testillano & Francisco Ruiz Rayo                         //
   4 //                                                                                                //
   5 // See project site at http://redmine.teslayout.com/projects/anna-suite                           //
   6 // See accompanying file LICENSE or copy at http://www.teslayout.com/projects/public/anna.LICENSE //
   7
   8
   9 #include <algorithm>
  10
  11 #include <anna/core/functions.hpp>
  12 #include <anna/core/tracing/TraceMethod.hpp>
  13 #include <anna/core/tracing/TraceFunction.hpp>
  14 #include <anna/core/mt/Thread.hpp>
  15
  16 #include <anna/app/functions.hpp>
  17
  18 #include <anna/xml/Node.hpp>
  19 #include <anna/xml/Attribute.hpp>
  20
  21 #include <anna/comm/Communicator.hpp>
  22
  23 #include <anna/timex/Engine.hpp>
  24 #include <anna/timex/internal/TickConsumer.hpp>
  25 #include <anna/timex/internal/TickProducer.hpp>
  26 #include <anna/timex/TimeEventObserver.hpp>
  27 #include <anna/timex/internal/sccs.hpp>
  28
  29 using namespace std;
  30 using namespace anna;
  31 using namespace anna::comm;
  32
  33 /**
  34    Resolucion minima (en milisegundos) soportada por el controlador de tiempos.
  35 */
  36 //static
  37 const Millisecond anna::timex::Engine::minResolution(100);
  38
  39
  40 anna::timex::Engine::Engine(const Millisecond & maxTimeout, const Millisecond & resolution) :
  41   app::Component(getClassName()),
  42   a_currentQuantum(0),
  43   a_maxQuantum(0),
  44   a_maxTimeout(maxTimeout),
  45   a_resolution(resolution),
  46   a_isActive(false),
  47   a_tickProducer(NULL),
  48   a_tickConsumer(NULL),
  49   a_expiredQuantum(NULL),
  50   a_timeTable(NULL) {
  51   timex::sccs::activate();
  52 }
  53
  54 anna::timex::Engine::~Engine() {
  55   delete a_tickConsumer;
  56   delete [] a_timeTable;
  57 }
  58
  59 //--------------------------------------------------------------------------------------------
  60 // Inicializa la configuracin.
  61 //
  62 // (2) En Solaris el ualarm tiene una deriva constante que parece ser que dependen de la
  63 //     arquitectura de la m�uina. Para calcular esta deriva y poder corregirla en los
  64 //     posteriores ualarm vamos a calcular la diferencia entre el tiempo esperado y el momento
  65 //     en que realmente llega la seal de ualarm.
  66 //--------------------------------------------------------------------------------------------
  67 void anna::timex::Engine::do_initialize()
  68 throw(RuntimeException) {
  69   LOGMETHOD(TraceMethod tm("timex::Engine", "do_initialize", ANNA_FILE_LOCATION));
  70
  71   if(a_maxQuantum > 0) {
  72     Logger::write(Logger::Warning, "Time controller was previously started", ANNA_FILE_LOCATION);
  73     return;
  74   }
  75
  76   if(a_resolution < minResolution)
  77     throw RuntimeException(functions::asString("Resolution must be greater than %d milliseconds", minResolution.getValue()), ANNA_FILE_LOCATION);
  78
  79   if(a_maxTimeout <= a_resolution)
  80     throw RuntimeException(functions::asString("Max-Timeout must be greater than %d milliseconds", a_resolution.getValue()), ANNA_FILE_LOCATION);
  81
  82   a_maxQuantum = a_maxTimeout / a_resolution;
  83
  84   while((a_maxQuantum * a_resolution) <= a_maxTimeout) a_maxQuantum ++;
  85
  86   a_timeTable = new Quantum [a_maxQuantum];
  87   Communicator* communicator = app::functions::component <Communicator> (ANNA_FILE_LOCATION);
  88   communicator->attach(a_tickConsumer = new TickConsumer(this));
  89   /**
  90    * Esto siempre se ejecutará en un thread aparte aunque la librería haya sido generada en modo ST,
  91    * porque así evitamos tener que generar el pulso de reloj mediante una alarma.
  92    */
  93   a_tickProducer = new TickProducer(this, a_tickConsumer->getfdWrite());
  94   pthread_attr_t attr;
  95   pthread_attr_init(&attr);
  96   int errorCode;
  97
  98   if((errorCode = pthread_create(&a_threadProducer, &attr, TickProducer::exec, a_tickProducer)) != 0)
  99     throw RuntimeException(std::string("timex::Engine::do_initialize"), errorCode, ANNA_FILE_LOCATION);
 100
 101   LOGDEBUG(
 102     string msg("Time controller | Max Timeout: ");
 103     msg += functions::asString(a_maxTimeout);
 104     msg += " | Resolution: ";
 105     msg += functions::asString(a_resolution);
 106     msg += " | Max Quantum: ";
 107     msg += functions::asString(a_maxQuantum);
 108     msg += " quantums";
 109     Logger::write(Logger::Debug, msg, ANNA_FILE_LOCATION);
 110   );
 111 }
 112
 113 // Reimplementado de app::Component
 114 void anna::timex::Engine::do_cloneParent()
 115 throw() {
 116 }
 117
 118 /*
 119  * Se invoca desde app::Application::clone -> app::Component::do_cloneChild (ojo EN EL NUEVO PROCESO).
 120  * Instala la senhal de tick en el proceso, ya que la alarma no se hereda directamente.
 121  */
 122 void anna::timex::Engine::do_cloneChild()
 123 throw(RuntimeException) {
 124 }
 125
 126 //----------------------------------------------------------------------------
 127 // No para los hilos de generacion, sino que evita que se escriban los bytes
 128 // en el 'pipe'.
 129 //----------------------------------------------------------------------------
 130 void anna::timex::Engine::pause()
 131 throw(RuntimeException) {
 132   Guard guard(this, "timex::Engine (pause)");
 133
 134   if(a_tickProducer->isInPause() == false) {
 135     LOGWARNING(
 136       string msg("timex::Engine::pause | Pending TimeEvents: ");
 137       msg += functions::asString((const int) a_directory.size());
 138       Logger::warning(msg, ANNA_FILE_LOCATION);
 139     );
 140     a_tickProducer->setIsInPause(true);
 141   }
 142 }
 143
 144 void anna::timex::Engine::play()
 145 throw(RuntimeException) {
 146   Guard guard(this, "timex::Engine (play)");
 147
 148   if(a_tickProducer->isInPause() == true) {
 149     LOGWARNING(
 150       string msg("timex::Engine::play | Pending TimeEvents: ");
 151       msg += functions::asString((const int) a_directory.size());
 152       Logger::warning(msg, ANNA_FILE_LOCATION);
 153     );
 154     a_tickProducer->setIsInPause(false);
 155   }
 156 }
 157
 158 void anna::timex::Engine::activate(timex::TimeEvent* timeEvent)
 159 throw(RuntimeException) {
 160   LOGMETHOD(TraceMethod tm(Logger::Local7, "timex::Engine", "activate", ANNA_FILE_LOCATION));
 161
 162   if(a_maxQuantum == 0)
 163     throw RuntimeException("Engine::initialize was not called", ANNA_FILE_LOCATION);
 164
 165   if(timeEvent == NULL)
 166     throw RuntimeException("Cannot activate a NULL TimeEvent", ANNA_FILE_LOCATION);
 167
 168   if(timeEvent->a_controller != NULL) {
 169     string msg(timeEvent->asString());
 170     msg += " | Already activated";
 171     throw RuntimeException(msg, ANNA_FILE_LOCATION);
 172   }
 173
 174   const Millisecond & timeout(timeEvent->getTimeout());
 175
 176   if(timeout > a_maxTimeout || timeout < a_resolution) {
 177     string msg("Invalid TimeEvent timeout | Max Timeout: ");
 178     msg += functions::asString(a_maxTimeout);
 179     msg += " | Min timeout (resolution): ";
 180     msg += functions::asString(a_resolution);
 181     msg += " | ";
 182     msg += timeEvent->asString();
 183     throw RuntimeException(msg, ANNA_FILE_LOCATION);
 184   }
 185
 186   Guard guard(this, "timex::Engine (activate)");
 187
 188   if(a_tickProducer->isInPause() == true) {
 189     string msg("Cannot activate TimeEvents with timex::Engine in pause | ");
 190     msg += timeEvent->asString();
 191     throw RuntimeException(msg, ANNA_FILE_LOCATION);
 192   }
 193
 194   const TimeEvent::Id id(timeEvent->getId());
 195
 196   if(a_directory.find(id) != a_directory.end())
 197     throw RuntimeException(functions::asString("Id %ld (0x%x) already in use", id, id), ANNA_FILE_LOCATION);
 198
 199   int iq = getQuantum(timeout);
 200   Quantum* quantum = &a_timeTable [iq];
 201   bool delayed;
 202   timeEvent->a_controller = this;
 203   a_directory.insert(Directory::value_type(id, quantum));
 204
 205   if(quantum == a_expiredQuantum) {
 206     a_delayedQuantum.push_back(timeEvent);
 207     delayed = true;
 208   } else {
 209     quantum->push_back(timeEvent);
 210     delayed = false;
 211   }
 212
 213   LOGDEBUG(
 214     string msg("timex::Engine::activate | ");
 215     msg += timeEvent->asString();
 216     msg += " | Current quantum: ";
 217     msg += functions::asString(a_currentQuantum);
 218     msg += " | Quantum programmed: ";
 219     msg += functions::asString(iq);
 220     msg += " (";
 221     msg += functions::asHexString(anna_ptrnumber_cast(quantum));
 222     msg += functions::asText(") | Delayed: ", delayed);
 223     Logger::debug(msg, ANNA_FILE_LOCATION);
 224   );
 225 }
 226
 227 anna::timex::TimeEvent* anna::timex::Engine::getTimeEvent(const timex::TimeEvent::Id eventTimeId)
 228 throw() {
 229   LOGMETHOD(TraceMethod tm(Logger::Local7, "timex::Engine", "getTimeEvent", ANNA_FILE_LOCATION));
 230   Directory::iterator iid;
 231   TimeEvent* result(NULL);
 232   Quantum* quantum;
 233   TimeEvent* aux;
 234   Guard guard(this, "timex::Engine (getTimeEvent)");
 235
 236   if((iid = a_directory.find(eventTimeId)) == a_directory.end())
 237     return NULL;
 238
 239   quantum = iid->second;
 240
 241   for(Quantum::iterator ii = quantum->begin(), maxii = quantum->end(); ii != maxii; ii ++) {
 242     if((aux = *ii)->getId() == eventTimeId) {
 243       result = aux;
 244       break;
 245     }
 246   }
 247
 248   return result;
 249 }
 250
 251 void anna::timex::Engine::cancel(timex::TimeEvent* timeEvent)
 252 throw(RuntimeException) {
 253   LOGMETHOD(TraceMethod tm(Logger::Local7, "timex::Engine", "cancel", ANNA_FILE_LOCATION));
 254
 255   if(timeEvent == NULL)
 256     throw RuntimeException("Cannot cancel a NULL TimeEvent", ANNA_FILE_LOCATION);
 257
 258   if(timeEvent->a_controller == NULL) {
 259     LOGDEBUG(
 260       string msg(timeEvent->asString());
 261       msg += " | Not activated";
 262       Logger::write(Logger::Debug, msg, ANNA_FILE_LOCATION);
 263     );
 264     return;
 265   }
 266
 267   Directory::iterator iid;
 268   Quantum* quantum;
 269   Guard guard(this, "timex::Engine (cancel)");
 270
 271   if((iid = a_directory.find(timeEvent->getId())) == a_directory.end())
 272     return;
 273
 274   if((quantum = iid->second) == a_expiredQuantum) {
 275     LOGWARNING(
 276       string msg("timex::Engine::cancel | ");
 277       msg += timeEvent->asString();
 278       msg += " | Processing already programmed";
 279       Logger::warning(msg, ANNA_FILE_LOCATION);
 280     );
 281     return;
 282   }
 283
 284   a_directory.erase(iid);
 285   quantum->erase(find(quantum->begin(), quantum->end(), timeEvent));
 286   timeEvent->a_controller = NULL;
 287   LOGDEBUG(
 288     string msg("timex::Engine::cancel | ");
 289     msg += timeEvent->asString();
 290     msg += " | Current quantum: ";
 291     msg += functions::asString(a_currentQuantum);
 292     msg += " | Quantum programmed: ";
 293     msg += functions::asHexString(anna_ptrnumber_cast(quantum));
 294     Logger::debug(msg, ANNA_FILE_LOCATION);
 295   );
 296   notifyRelease(timeEvent);
 297 }
 298
 299 void anna::timex::Engine::do_stop()
 300 throw() {
 301   LOGMETHOD(TraceMethod tm("timex::Engine", "do_stop", ANNA_FILE_LOCATION));
 302   Quantum::iterator ii, maxii;
 303   Guard guard(this, "timex::Engine (do_stop)");
 304   app::functions::component <Communicator> (ANNA_FILE_LOCATION)->detach(a_tickConsumer);
 305
 306   if(a_tickProducer)
 307     a_tickProducer->requestStop();
 308
 309   a_directory.clear();
 310   TimeEvent* timeEvent;
 311
 312   for(int iq = 0; iq < a_maxQuantum; iq ++) {
 313     Quantum& quantum(a_timeTable [iq]);
 314
 315     for(ii = quantum.begin(), maxii = quantum.end(); ii != maxii; ii ++) {
 316       timeEvent = Engine::timeEvent(ii);
 317
 318       try {
 319         timeEvent->a_controller = NULL;
 320         timeEvent->stop();
 321         notifyRelease(timeEvent);
 322       } catch(Exception& ex) {
 323         ex.trace();
 324       }
 325     }
 326
 327     quantum.clear();
 328   }
 329 }
 330
 331 void anna::timex::Engine::kill()
 332 throw() {
 333   Guard guard(this, "timex::Engine (kill)");
 334   app::functions::component <Communicator> (ANNA_FILE_LOCATION)->detach(a_tickConsumer);
 335
 336   if(a_tickProducer)
 337     a_tickProducer->requestStop();
 338
 339   string msg("timex::Engine::kill | ");
 340   msg += functions::asString(" | Pending TimeEvents: %d", a_directory.size());
 341   Logger::critical(msg, ANNA_FILE_LOCATION);
 342
 343   for(int iq = 0; iq < a_maxQuantum; iq ++)
 344     a_timeTable [iq].clear();
 345
 346   a_delayedQuantum.clear();
 347   a_directory.clear();
 348 }
 349
 350 //----------------------------------------------------------------------------------
 351 // (1) Podr� ser que al invocar al m�odo 'expire' de una evento se halla originado
 352 //     la cancelacin de algn otro evento. As�que si el id de evento ya no esta en
 353 //     en el directorio => da igual, se ignora ya que luego borraremos todos los
 354 //     eventos del Quantum.
 355 // (2) Si mientas estamos caducando este Quantum hay que activar transacciones que
 356 //     debe ir sobre el => se guardan en un vector temporal y se copian al final
 357 //     del proceso, con lo que evita totalmente la posiblidad de perdida.
 358 // (3) Si el temporizador ha sido reactivado no tiene que liberarlo.
 359 //----------------------------------------------------------------------------------
 360 void anna::timex::Engine::tick()
 361 throw(RuntimeException) {
 362   LOGMETHOD(TraceMethod tm(Logger::Local7, "timex::Engine", "tick", ANNA_FILE_LOCATION));
 363   /*
 364      if (Logger::isActive (Logger::Local6) == true && a_directory.size () > 0) {
 365         string msg ("Tick | Quantum: ");
 366         msg += functions::asString (a_currentQuantum);
 367         msg += " | Tx: ";
 368         msg += functions::asString (functions::millisecond ());
 369         msg += " | N: ";
 370         msg += functions::asString (a_directory.size ());
 371         Logger::write (Logger::Local6, msg, ANNA_FILE_LOCATION);
 372      };
 373   */
 374   Guard guard(this, "timex::Engine (tick)");
 375   int counter(0);
 376   eventBeginQuantum();
 377   a_expiredQuantum = &a_timeTable [a_currentQuantum];
 378   TimeEvent* timeEvent;
 379   Directory::iterator iid;
 380   quantum_iterator ii, maxii;
 381   bool observed;
 382
 383   for(ii = a_expiredQuantum->begin(), maxii = a_expiredQuantum->end(); ii != maxii; ii ++) {
 384     timeEvent = Engine::timeEvent(ii);
 385
 386     if((iid = a_directory.find(timeEvent->getId())) == a_directory.end())            // (1)
 387       continue;
 388
 389     a_directory.erase(iid);
 390
 391     if(Logger::isActive(Logger::Debug)) {
 392       string msg(timeEvent->asString());
 393       msg += " | Quantum programmed: ";
 394       msg += functions::asHexString(anna_ptrnumber_cast(a_expiredQuantum));
 395       msg += " | Expired";
 396       Logger::write(Logger::Debug, msg, ANNA_FILE_LOCATION);
 397     }
 398
 399     timeEvent->a_controller = NULL;
 400     /*
 401      * Si tiene NO tiene observer asociado, seguramente se auto-liberará en el expire
 402      * Si tiene observer, se invocará al método correspondiente.
 403      */
 404     observed = (timeEvent->a_observer != NULL);
 405
 406     try {
 407       timeEvent->expire(this);
 408       counter ++;
 409     } catch(Exception& ex) {
 410       ex.trace();
 411     }
 412
 413     // Si tiene observer => el expire NO puede haber liberado el TimeEvent
 414     if(observed == true) {
 415       if(timeEvent->a_controller == NULL)                                           // (3)
 416         timeEvent->a_observer->release(timeEvent);
 417     }
 418   }
 419
 420   a_expiredQuantum->clear();
 421
 422   if(a_delayedQuantum.size() > 0) {                                                 // (2)
 423     for(ii = a_delayedQuantum.begin(), maxii = a_delayedQuantum.end(); ii != maxii; ii ++)
 424       a_expiredQuantum->push_back(Engine::timeEvent(ii));
 425
 426     a_delayedQuantum.clear();
 427   }
 428
 429   a_expiredQuantum = NULL;
 430
 431   if(++ a_currentQuantum >= a_maxQuantum)
 432     a_currentQuantum = 0;
 433
 434   eventEndQuantum(counter);
 435 }
 436
 437 void anna::timex::Engine::notifyRelease(timex::TimeEvent* timeEvent)
 438 throw() {
 439   if(timeEvent->a_observer != NULL)
 440     timeEvent->a_observer->release(timeEvent);
 441 }
 442
 443 string anna::timex::Engine::asString() const
 444 throw() {
 445   string msg("timex::Engine { ");
 446   msg += app::Component::asString();
 447   msg += " | Max Timeout: ";
 448   msg += functions::asString(a_maxTimeout);
 449   msg += " ms | Resolution: ";
 450   msg += functions::asString(a_resolution);
 451   msg += " ms | Programming: ";
 452   msg += functions::asString(a_maxQuantum);
 453   msg += " quantums | Pending TimeEvents: ";
 454   msg += functions::asString((const int) a_directory.size());
 455   return msg += " }";
 456 }
 457
 458 xml::Node* anna::timex::Engine::asXML(xml::Node* parent) const
 459 throw() {
 460   parent = Component::asXML(parent);
 461   xml::Node* result = parent->createChild("timex.Engine");
 462   result->createAttribute("MaxTimeout", a_maxTimeout);
 463   result->createAttribute("Resolution", a_resolution);
 464   result->createAttribute("N", a_directory.size());
 465   result->createAttribute("Pause", functions::asString(a_tickProducer->isInPause()));
 466   xml::Node* node = result->createChild("Quantum");
 467   node->createAttribute("Current", a_currentQuantum);
 468   node->createAttribute("Max", a_maxQuantum);
 469   return result;
 470 }