RunController.cc

Go to the documentation of this file.

#include <sstream>
#include <iostream>

#include <boost/lexical_cast.hpp>

#include <fwk/CentralConfig.h>
#include <fwk/RunController.h>
#include <utl/ErrorLogger.h>
#include <fwk/VModule.h>
#include <utl/Reader.h>
#include <evt/Event.h>
#include <utl/AugerException.h>
#include <utl/TabularStream.h>
#include <utl/StringCompare.h>

using namespace std;
using namespace utl;


namespace fwk {

  RunController::RunController() :
    fCurEvent(new evt::Event)
  {
    ErrorLogger::GetInstance();
  }


  VModule&
  RunController::GetModule(const string& moduleName)
    const
  {
    auto* const m = VModuleFactory::Create(moduleName);

    fUsedModuleNames.insert(moduleName);

    if (!m) {
      ostringstream emsg;
      emsg << "No module creator found for module with name '" << moduleName << "'. "
              "Most likely reasons:\n"
              "1) You misspelled the module name, or the module doesn't exist.\n"
              "2) You declared a default constructor for your module\n"
              "   but did not provide an implementation.\n"
              "3) You linked with static libraries (currently not supported)\n"
              "4) Your application was not linked to AugerModules (check with ldd and your compiler options).";
      ERROR(emsg);
      throw utl::ModuleSequenceException("cannot create module");
    }
    return *m;
  }


  bool
  RunController::HasModule(const string& moduleName)
    const
  {
    auto* const m = VModuleFactory::Create(moduleName);
    return m;
  }


  int
  RunController::GetNumberOfRegisteredModules()
    const
  {
    return VModuleFactory::GetNumberOfCreators();
  }


  string
  RunController::GetRegisteredModuleNames()
    const
  {
    ostringstream modules;
    for (auto it = VModuleFactory::Begin(); it != VModuleFactory::End(); ++it) {
      auto* const m = VModuleFactory::Create(it->first);
      if (m)
        modules << it->first << " (version: "
                << m->GetVersionInfo(VModule::eRevisionNumber)
                << ")\n";
    }
    return modules.str();
  }


  // use AttributeMap to return stuff like
  // name : module name
  // version : svn version
  // last mod : last modified
  // last editor : last editor

  void
  RunController::Init()
  {
    // Attempt to find the sequence.

    // catch exceptions from central config here ?

    CentralConfig* cc = nullptr;
    try {
      // assumes central config has already been given name of bootstrap file
      cc = CentralConfig::GetInstance();
    } catch (AugerException& ex) {
      ostringstream msg;
      msg << "Failure configuring the framework. "
             "A exception of type " << ex.GetExceptionName() << " was thrown. "
             "Message: " << ex.GetMessage() << " "
             "Cannot recover. Terminating";
      ERROR(msg);
      exit(EXIT_FAILURE);
    }

    //fUsedModuleNames.clear();

    Branch topB = cc->GetTopBranch("ModuleSequence");
    if (!topB) {
      ostringstream emsg;
      emsg << "Error in RunController. "
              "Unable to find the module sequencing information! "
              "Is the ModuleSequence configLink correctly defined in the bootstrap file?";
      ERROR(emsg);
      throw utl::ModuleSequenceException("missing <ModuleSequence>");
    } else
      InitBranch(topB);
  }


  void
  RunController::InitBranch(Branch& currentB)
  {
    // initialize only those modules requested in the control file
    GetNextModuleName(currentB);

    for (const auto& mn : fUniqueModuleNames) {

      auto& mod = GetModule(mn);

      mod.InitTiming();

      cerr << "--> Running Init() of module '" << mn << "'\n";

      const auto flag = mod.Init();

      cerr << "--> Finished Init() of module '" << mn << "', "
              "status: " << VModule::GetResultFlagByName(flag) << '\n';

      if (flag == VModule::eFailure)
        throw utl::ModuleSequenceException("Init() failed");

    }
  }


  // Find next module name in the XML file (recursive)
  void
  RunController::GetNextModuleName(Branch& currentB)
  {
    if (currentB.GetName() == "module") {
      const string modName = currentB.Get<string>();
      fUniqueModuleNames.insert(modName);
    }

    for (Branch cb = currentB.GetFirstChild(); cb; cb = cb.GetNextSibling())
      GetNextModuleName(cb);
  }


  void
  RunController::Run()
  {
    Branch topB = CentralConfig::GetInstance()->GetTopBranch("ModuleSequence");
    DoRunSequence(topB);
  }


  void
  RunController::RunBranch(Branch& currentB)
  {
    DoRunSequence(currentB);
  }


  void
  RunController::Finish()
  {
    Branch topB = CentralConfig::GetInstance()->GetTopBranch("ModuleSequence");
    FinishBranch(topB);
  }


  void
  RunController::FinishBranch(Branch& /*currentB*/)
  {
    fStopwatch.Stop();
    const double totalUTime = fStopwatch.GetCPUTime(Stopwatch::eUser)/second;
    const double totalSTime = fStopwatch.GetCPUTime(Stopwatch::eSystem)/second;
    const double totalTime = fStopwatch.GetCPUTime()/second;
    double moduleUTimeSum = 0;
    double moduleSTimeSum = 0;

    ostringstream failureMessage;

    TabularStream tab("r:  .  .  .");

    tab << "Module" << endc << "USR" << endc << "SYS" << endc << "%" << endr
        << hline;

    for (const auto& mn : fUniqueModuleNames) {

      auto& mod = GetModule(mn);

      auto& stopwatch = mod.GetStopwatch();
      const double moduleUTime = stopwatch.GetCPUTime(Stopwatch::eUser) / second;
      const double moduleSTime = stopwatch.GetCPUTime(Stopwatch::eSystem) / second;
      moduleUTimeSum += moduleUTime;
      moduleSTimeSum += moduleSTime;
      const double frac = round(1000*(moduleUTime + moduleSTime) / totalTime) / 10;

      tab << mn << endc << moduleUTime << endc << moduleSTime << endc << frac << endr;

      cerr << "--> Running Finish() of module '" << mn << "'\n";

      const auto flag = mod.Finish();

      cerr << "--> Finished Finish() of module '" << mn << "', "
              "status: " << VModule::GetResultFlagByName(flag) << '\n';

      if (flag == VModule::eFailure)
        failureMessage << (failureMessage.str().empty() ? "" : "\n")
                       << "Received failure message from Finish() of module: " << mn;
    }

    const auto failureStr = failureMessage.str();
    if (!failureStr.empty())
      ERROR(failureStr);

    if (!fRunData.GetNamedCounters().empty()) {
      TabularStream tab("r: r");
      tab << "Item" << endc << "Count" << endr
          << hline;
      for (const auto& c : fRunData.GetNamedCounters())
        tab << c.first << endc << c.second << endr;
      tab << delr;
      INFO("Performance counters:\n" + tab.Str());
    }

    const double frac = round(1000*(moduleUTimeSum + moduleSTimeSum) / totalTime) / 10;
    tab << hline
        << "All modules" << endc << moduleUTimeSum << endc
        << moduleSTimeSum << endc << frac << endr
        << "Total" << endc << totalUTime << endc << totalSTime << endc << 100;
    ostringstream info;
    info << "\n\nCPU user and system time in Module::Run()\n"
         << tab << '\n';
    const double time = fRealTimeStopwatch.Stop();
    info << "Total real time of the run: " << time/second << " s";
    INFO(info);

    if (!failureStr.empty())
      throw utl::ModuleSequenceException(failureStr);
  }


  string
  RunController::GetBreakStatusByName(const BreakStatus status)
  {
    switch (status) {
    case eNoBreak: return "eNoBreak";
    case eBreak: return "eBreak";
    case eContinue: return "eContinue";
    default: return "Unknown";
    }
  }


  // Sequence the modules (call their Run methods)
  // according to the sequence specified in the XML file.
  void
  RunController::DoRunSequence(Branch& currentB)
  {
    Branch modControl = currentB.GetChild("moduleControl");

    if (!modControl) {
      INFO("No <moduleControl> tag was found in the module sequencing file. No sequencing will be done.");
      return;
    }

    for (Branch cb = modControl.GetFirstChild(); cb; cb = cb.GetNextSibling()) {
      const auto status = DoNextInSequence(cb);
      if (status != eNoBreak) {
        ostringstream err;
        err << "Module sequence cannot handle the " << GetBreakStatusByName(status) << " return value. "
               "Please, fix your module sequence.";
        ERROR(err);
        throw utl::ModuleSequenceException(err.str());
      }
    }
  }


  // Invoke Run method the appropriate number of times
  // for the next module in the sequence (recursive)
  RunController::BreakStatus
  RunController::DoNextInSequence(Branch& currentB)
  {
    if (!currentB)
      return eNoBreak;  // null branch

    if (currentB.GetName() == "loop" ||
        currentB.GetName() == "try") {

      bool unboundedLoopFlag = false;  // true if user requests an unbounded loop,
      int loopInt = 1;                 // otherwise, loop this number of times.
      const auto& attMap = currentB.GetAttributes();
      const auto attEnd = attMap.end();

      const auto numTimesIt = attMap.find("numTimes");
      if (numTimesIt != attEnd) {
        if (StringEquivalent(numTimesIt->second, "unbounded"))
          unboundedLoopFlag = true;
        else
          loopInt = boost::lexical_cast<int>(numTimesIt->second);
      }

      // Allow the tag to be called either save or pushEventToStack
      const auto pushEventToStackIt = attMap.find("pushEventToStack");
      const auto saveIt = attMap.find("save");

      bool saveFlag = false;

      if (saveIt != attEnd) {

        ostringstream warn;
        warn << "The 'save' loop attribute is deprecated (people find it unclear). "
                "Please use 'pushEventToStack' instead.";
        WARNING(warn);

        // this error can probably be encoded with xsd
        if (pushEventToStackIt != attEnd) {
          ostringstream err;
          err << "You can have either a 'save' or 'pushEventToStack' attribute"
                 "in the loop tag, but not both.";
          ERROR(err);
          throw utl::ModuleSequenceException(err.str());
        }

        if (StringEquivalent(saveIt->second, "yes"))
          saveFlag = true;

      }

      if (pushEventToStackIt != attEnd &&
          StringEquivalent(pushEventToStackIt->second, "yes"))
        saveFlag = true;

      evt::Event* const savedEvent = saveFlag ? new evt::Event(*fCurEvent) : nullptr;

      for (int iLoop = 0; unboundedLoopFlag || iLoop < loopInt; ++iLoop) {

        if (iLoop && savedEvent)
          *fCurEvent = *savedEvent;

        BreakStatus innerBreakStatus = eNoBreak;
        for (Branch cb = currentB.GetFirstChild(); cb; cb = cb.GetNextSibling()) {
          const auto outerBreakStatus = DoNextInSequence(cb);  // doing next in the sequence (recursive)
          if (outerBreakStatus == eBreak) {
            innerBreakStatus = eBreak;
            break;
          } else if (outerBreakStatus == eContinue)
            break;
        }

        if (innerBreakStatus == eBreak)
          break;

      }

      delete savedEvent;
      return eNoBreak;

    } else if (currentB.GetName() == "module") {

      string modName;
      currentB.GetData(modName);

      cerr << "--> Running Run() of module '" << modName << "'\n";

      // store the module name so that it can be queried
      fCurrentModule = modName;

      // run the module
      const auto status = GetModule(modName).RunWithTiming(*fCurEvent);

      cerr << "--> Finished running module '" << modName << "' "
              "with status " << VModule::GetResultFlagByName(status) << '\n';

      if (status == VModule::eFailure) {
        cerr << "  +--> Calling Finish() methods of all modules\n";
        Finish();
        throw utl::ModuleSequenceException("Module sequence failure");
      }

      switch (status) {
      case VModule::eBreakLoop: return eBreak;
      case VModule::eContinueLoop: return eContinue;
      default: return eNoBreak;
      }

    } else {

      ostringstream err;
      err << "Encountered a Branch in XML file with name <" << currentB.GetName() << ">. "
             "RunController can only process Branches called <loop> or <module>.";
      ERROR(err);
      throw utl::ModuleSequenceException("unknown branch");

    }

    return eNoBreak;
  }


  RunController::~RunController()
  {
    delete fCurEvent;

    ErrorLogger::GetInstance().WriteErrorMessagesToStream();
  }

}