Fitter.cpp

#include "Fitter.h"
#include "../src/FitStrategy.h"
#include "FitterWorkerGroup.h"
#include "ThresholdFitAlgorithm.h"
#include "ThresholdFitAlgorithm2.h"
#include "FitAlgorithmMap.h"

#include "Sct/SctNames.h"
#include "Sct/SctParameters.h"
#include "Sct/DestroyingDeleter.h"
#include "Sct/StdExceptionWrapper.h"
#include "Sct/IS/IONameIS.h"
#include "Sct/OmniMarshalling.h"

#include <iostream>
#include <vector>
#include <cassert>
#include <ctime>
#include <string>
#include <TMath.h>
#include <boost/timer.hpp>
#include <is/info.h>
#include <is/infodictionary.h>
#include <is/infoiterator.h>
#include <is/inforeceiver.h>
#include <is/infoT.h>

using namespace std;
using namespace Sct;
using namespace SctData;
using namespace boost;

namespace SctFitter {
void doFits(ISCallbackInfo * isc);

Fitter* Fitter::fitter=0;

Fitter& Fitter::instance() {
    if (!fitter)
      throw Sct::IllegalStateError("Attempt to use uninitialised Fitter", __FILE__, __LINE__);
    return *fitter;
}

Fitter& Fitter::initialize(FitterArguments args) throw(ConfigurationException) {
    fitter = new Fitter(args);
    return *fitter;
}

  void Fitter::shutdown(){
    if (workergroup) workergroup->stop();
    getFitterServer().stop();
  }

  Fitter::Fitter(FitterArguments args) throw(ConfigurationException) : 
    IPCNamedObject<POA_FitterI::FitterInterface>(
                         SctNames::getPartition(),
                         args.instanceName()),
    m_nFitsDone(0), m_nFitErrors(0), m_nScans(0), m_scanTimeTaken(0), m_ioTimeTaken(0), m_scan("No last scan"), 
    m_fitStrategy(0), workergroup(0), m_args(args)  
  {
    std::cout << "Fitting Service " << m_args.print() << endl;
    Sct::SctNames::Mrs() << "FITTER_START" << MRS_PARAM<const char*>("Name",m_args.instanceName().c_str()) 
             << MRS_INFORMATION << ENDM;
    auto_ptr<ISInfoReceiver> ir(new ISInfoReceiver(SctNames::getPartition()));
    if (!ir.get())
        throw ConfigurationException("Fitter::initialize can't make infoReceiver ", __FILE__, __LINE__) ;
    m_infoReceiver  = ir;
    try{
      internal_setFitStrategy(m_args.getStrategyName());
    }catch(Sct::Throwable& e){
      std::string defaultStrategy("RootFitStrategy");
      e.sendToMrs(MRS_DIAGNOSTIC);
      internal_setFitStrategy(defaultStrategy);
    }
    workergroup = new FitterWorkerGroup();
    workergroup->reportTo(m_args.getISStatusName());
    updateSubscriptions();
    internal_setFitOptions(m_args.getFitOptions());
  }

Fitter::~Fitter() throw() {}


std::string Fitter::getOutputIsServer() const{
  return m_args.getOutputISServer();
}

IPCServer& Fitter::getFitterServer() throw() {
  static IPCServer fitterServer;
    return fitterServer;
}



void Fitter::go() throw(IsException) {
  workergroup->go(m_args.getNWorkers());
  if (m_args.recoveryMode()) fitAll();
}

void Fitter::updateSubscriptions(){
  if (workergroup) workergroup->setPaused(true);
  recursive_mutex::scoped_lock scoped_lock(counterMutex);
 
  const std::list<SctService::Arguments::Subscription>& input=m_args.getInputISServers();
  for (std::list<SctService::Arguments::Subscription>::const_iterator i=input.begin(); 
       i!=input.end(); ++i){
    try{
      ISInfo::Status s=m_infoReceiver->subscribe((*i).server.c_str(),
                         (*i).regexp.c_str(), doFits);
      if (s!=ISInfo::Success) {
    ostringstream os;
    os <<"Fitter Could not subscribe to IS server ["<<(*i).server 
       <<"] with regexp [" << (*i).regexp << "]";
    IsException(s, os.str(), __FILE__, __LINE__).sendToMrs();
      }else{
    ostringstream os;
    os <<"Fitter subscribed to IS server [" << (*i).server
       <<"] with regexp [" << (*i).regexp << "]";
    Sct::SctNames::Mrs() << "FIT_SUBSCRIBE" << MRS_TEXT(os.str()) << MRS_INFORMATION << ENDM;
      }
    }catch(Throwable & e){
      e.sendToMrs();
    }catch(std::exception & e){
      Sct::StdExceptionWrapper(e, __FILE__, __LINE__).sendToMrs();
    }
  }
  if (workergroup) workergroup->setPaused(false);
}

  //infoReceiver callback function must take a static function as an argument:
void Fitter::doFits(ISCallbackInfo * isc) throw(IsException, LogicError) {
    // find out why the callback function has been called:
    if (isc->reason() != is::Created && isc->reason() != is::Updated )
    return;
    
    // put it in the queue
    Fitter::instance().workergroup->push(isc->name());
}

  //CORBA
void Fitter::fit(const char* name) throw() {
    // put it in the queue
    workergroup->push(name);
}

  //CORBA
void Fitter::fitAll() throw() {
  std::list<string> alreadyDone;
  {
    ISInfoIterator iter(SctNames::getPartition(), getOutputIsServer(), "SctData::FitScanResult.*");
    while (iter()) {
      Sct::IS::IONameIS name(iter.name());
      alreadyDone.push_back(name.getUniqueID());
    }
  }

  const std::list<SctService::Arguments::Subscription>& input=m_args.getInputISServers();
  unsigned nToRecover=0;
  for (std::list<SctService::Arguments::Subscription>::const_iterator i=input.begin(); i!=input.end(); ++i){
    ISInfoIterator iter(SctNames::getPartition(), (*i).server.c_str(), 
            (*i).regexp.c_str());
    while (iter()) {
      Sct::IS::IONameIS name(iter.name());
      std::string id=name.getUniqueID();
      if (std::find(alreadyDone.begin(), alreadyDone.end(), id)==alreadyDone.end()){
    nToRecover++;
    workergroup->push(iter.name());
      }
    }
  }
  std::ostringstream msg;
  msg << "Need to fit total of " << nToRecover << " objects";
  SctNames::Mrs() << "FITTING_RECOVERY" << MRS_TEXT(msg.str()) << MRS_DIAGNOSTIC << ENDM;
}

  //CORBA
char* Fitter::getFitOptions() throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    return Sct::copyStringToCorba(internal_getFitStrategy().getOptions());
}

  //CORBA
char* Fitter::status() throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    return Sct::copyStringToCorba(internal_status());
}

void Fitter::internal_setFitOptions(const string & opt) throw (LogicError) {
  recursive_mutex::scoped_lock scoped_lock(counterMutex);
  internal_getFitStrategy().setOptions(opt);
}

  //CORBA
void Fitter::setFitOptions(const char *n) throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    string name( n );
    try {
        internal_setFitOptions(name);
    } catch (LogicError &e) {
        return;
    }
}

char* Fitter::getFitStrategy() throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    return Sct::copyStringToCorba(internal_getFitStrategy().getName());
}

FitterI::StringList* Fitter::listFitAlgorithms(){
  return Sct::copyStringListToCorba<FitterI::StringList>(FitAlgorithmMap::instance().listAlgorithms());
}

FitterI::StringList* Fitter::listFitStrategies(){
  return Sct::copyStringListToCorba<FitterI::StringList>(FitStrategyFactory::instance().listStrategies());
}

  //CORBA
void Fitter::setFitStrategy(const char *n) throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    string name( n );
    cout <<"trying to set fit strategy "<<name<<endl;
    try {
        internal_setFitStrategy(name);
    } catch (LogicError& e) {
      std::string message("failed  to set fit strategy ");
      message += name;
      Sct::InvalidArgumentError(message, __FILE__, __LINE__).sendToMrs();
      return;
    }
    cout <<"set fit strategy "<<name<<endl;
}

string Fitter::internal_status() const throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    string strategyName;
    try {
        strategyName=internal_getFitStrategy().getName();
    } catch (LogicError &e) {
        strategyName="NONE";
    }

    ostringstream oss;
    oss << "\t=== FittingService ===\n";
    oss << m_args.print()<<endl;
    oss << "Current: \n\tStrategy=" << strategyName << endl;
    oss << "\tOptions=" << internal_getFitStrategy().getOptions() << endl;
    oss << "\tWorkers=" << workergroup->nWorkers() << endl
    << "\t  (Busy=" << workergroup->busy() << ")" << endl
    << "\tFits Done=" << internal_nFitsDone() << endl
    << "\tErrors=" << internal_nFitErrors() << endl
    << "\tScans Done=" << m_nScans << endl 
    << "\tLast scan: " << internal_lastScan() << endl; 
    oss << "\tQueue=" << workergroup->queueSize();
    if (workergroup->isFifo()){
      oss << " [FIFO]\n";
    }else{
      oss << " [FILO]\n";
    }
    oss << "Debug : " << (workergroup->debug()?"yes":"no") << "\n";
    
    oss << "\nAvailable FitAlgorithms\n\t";
    std::list<string> algs = FitAlgorithmMap::instance().listAlgorithms();
    for (std::list<string>::const_iterator i=algs.begin(); i!=algs.end(); ++i){
      oss << *i << " ";
    }
    
    oss << "\n\nAvailable FitStrategies\n\t";
    std::list<string> strategies = FitStrategyFactory::instance().listStrategies();
    for (std::list<string>::const_iterator i=strategies.begin(); i!=strategies.end(); ++i){
      oss << *i << " ";
    }
    
    oss << "\n\nTests:\n";
    workergroup->printTests(oss);
    
    oss << "\n\nTiming:"<<endl
    << "\tI/O time: " << m_ioTimeTaken << endl
    << "\tScan time: "<< m_scanTimeTaken << endl 
    << "\tApprox average time/scan/thread: " << getAverageTimePerScan() << endl;
    return oss.str();
}

void Fitter::setDebug(CORBA::Boolean value){
  workergroup->setDebug(value);
}

bool Fitter::debug(){
  return workergroup->debug();
}

void Fitter::incrementFitErrors() throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    ++m_nFitErrors;
}

void Fitter::incrementFitsDone() throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    ++m_nFitsDone;
}

void Fitter::scanDone(double time) throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    ++m_nScans;
    m_scanTimeTaken += time;
}

void Fitter::addIOTime(double time) throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    m_ioTimeTaken += time;
}


double Fitter::getAverageTimePerScan() const throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    double time = (m_ioTimeTaken+m_scanTimeTaken)/workergroup->nWorkers();
    if (m_nScans > 0) time /= m_nScans;     
    return time;
}


long Fitter::internal_nFitsDone() const throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    return m_nFitsDone;
}

long Fitter::internal_nFitErrors() const throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    return m_nFitErrors;
}

  //CORBA
long Fitter::nFitsDone() throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    return internal_nFitsDone();
}

  //CORBA
long Fitter::nFitErrors() throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    return internal_nFitErrors();
}

  //CORBA
long Fitter::queueLength() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    return workergroup->queueSize();
}

  //CORBA
long Fitter::busy() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    return workergroup->busy();
}

std::string Fitter::internal_lastScan() const throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    return m_scan;
}

  //CORBA
char* Fitter::lastScan() throw() {    
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    return copyStringToCorba(m_scan);
}

bool Fitter::isFifo(){
  return workergroup->isFifo();
}

void Fitter::setFifo(bool val){
  workergroup->setFifo(val);
}

FitStrategy& Fitter::internal_getFitStrategy() const throw(LogicError) {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    if (!m_fitStrategy)
        throw InvariantViolatedError("Fitter::getStrategy() no fit strategy defined", __FILE__, __LINE__);
    return *m_fitStrategy;
}

void Fitter::internal_setFitStrategy(const string & name) throw(LogicError) {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    m_fitStrategy = FitStrategyFactory::instance().getStrategy(name);
    SctNames::Mrs() << "FIT_STRATEGY" << MRS_TEXT(string("set fit strategy to ")+name) 
            << MRS_INFORMATION << ENDM;
}

} // end of namespace SctFitter

---