AnalysisService.cpp

#include "AnalysisService.h"
#include "AnalysisAlgorithmMap.h"
#include "AnalysisAlgorithm.h"
#include "AnalysisWorkerGroup.h"
#include "AnalysisArguments.h"
#include "DcsInterface.h"
#include "ConfigurationInterface.h"

#include "Sct/SctNames.h"
#include "Sct/StdExceptionWrapper.h"
#include "Sct/IpcObjectException.h"
#include "Sct/IS/IOManagerIS.h"
#include "Sct/IS/IONameIS.h"
#include "Sct/ConfigurationException.h"
#include "Sct/DestroyingDeleter.h"
#include "Sct/UnsupportedOperationError.h"
#include "Sct/OmniMarshalling.h"
#include "sctConf/configuration.h"
#include "SctData/TestResult.h"
#include "sctConfIPC/configipc.h"
#include "CalibrationController/IS/TestData.h"

#include <ipc/core.h>
#include <boost/scoped_ptr.hpp>
#include <is/info.h>
#include <is/infodictionary.h>
#include <is/infoiterator.h>
#include <is/inforeceiver.h>
#include <is/infoT.h>

using namespace Sct;
using namespace Sct::IS;
using namespace SctData;
using boost::scoped_ptr;

namespace SctAnalysis {

AnalysisService::~AnalysisService(){
}

void AnalysisService::shutdown(){
  if (workergroup) workergroup->stop();
  getServer().stop();
}

AnalysisService::AnalysisService(AnalysisArguments args) : 
      IPCNamedObject<POA_AnalysisServiceI::AnalysisServiceInterface,
             ipc::multi_thread>(
                    SctNames::getPartition(),
                    args.instanceName()),
      dcsinterface(0), infoReceiver(new ISInfoReceiver(SctNames::getPartition())), m_args(args) {
  if (!infoReceiver.get())
    throw ConfigurationException("AnalysisService::AnalysisService can't make infoReceiver ", __FILE__, __LINE__) ;
  std::cout << "Making DcsInterface ... " << std::endl;
    try{
      boost::shared_ptr<SctConfiguration::Configuration> configipc(new SctConfiguration::ConfigIPC);
      s_configuration=configipc;
      boost::shared_ptr<ConfigurationInterface> config_local (new ConfigurationInterface(configipc));
      dcsinterface = new DcsInterface(config_local);
      config_local->update();
    }catch(std::exception &e){
      SctNames::Mrs() << MRS_TEXT("I wont be able to save DCS information as I dont know the configuration (std::exception)")
              << "ANALYSIS_NO_DCS" << MRS_WARNING << ENDM;
    }catch(CORBA::Exception &e){
      SctNames::Mrs() << MRS_TEXT("I wont be able to save DCS information as I dont know the configuration (CORBA::Exception)")
              << "ANALYSIS_NO_DCS" << MRS_WARNING << ENDM;
    }catch(...){
      SctNames::Mrs() << MRS_TEXT("I wont be able to save DCS information as I dont know the configuration (unknown exception)")
              << "ANALYSIS_NO_DCS" << MRS_WARNING << ENDM;
    }
    try{
      setFitStrategy(m_args.getStrategyName());
    }catch(Throwable& e){
      e.sendToMrs();
      setFitStrategy("RootFitStrategy");
    }
    getFitStrategy().setOptions(m_args.getFitOptions());
    workergroup = new AnalysisWorkerGroup();
    workergroup->reportTo(m_args.getISStatusName());
}

IPCServer& AnalysisService::getServer() throw() {
    static IPCServer server;
    return server;
}

const std::string AnalysisService::getOutputISServerName(){
  return m_args.getOutputISServer();
}

boost::shared_ptr<SctConfiguration::Configuration> AnalysisService::getConfiguration(){
  return s_configuration;
}

void AnalysisService::run() {
    // start up worker thread.
    workergroup->go(m_args.getNWorkers());
    
    if (m_args.recoveryMode()) recover();
    const list<SctService::Arguments::Subscription>& theList=m_args.getInputISServers();

    // subscribe to various IS stuff.
    for (list<SctService::Arguments::Subscription>::const_iterator i=theList.begin(); 
     i!=theList.end(); ++i){
      ISInfo::Status s = infoReceiver->subscribe((*i).server.c_str(), 
                         (*i).regexp.c_str(), generalCallback);
      if (s!=ISInfo::Success){
    std::ostringstream oss;
    oss << "Analysis service could not subscribed to [" << (*i).server 
        << "] to receive [" << (*i).regexp << "]";
    SctNames::Mrs() << "ANALYSIS_SUBSCRIBE" << MRS_TEXT(oss.str()) << MRS_ERROR << ENDM;
      }else{
    std::ostringstream oss;
    oss << "Analysis service subscribed to [" << (*i).server 
        << "] to receive [" << (*i).regexp << "]";
    SctNames::Mrs() << "ANALYSIS_SUBSCRIBE" << MRS_TEXT(oss.str()) << MRS_INFORMATION << ENDM;
      }
    }
}

  CORBA::Short AnalysisService::busy() {
    boost::recursive_mutex::scoped_lock lock(m_status_access);
    return workergroup->busy();
  }

  CORBA::Short AnalysisService::queueLength() {
  boost::recursive_mutex::scoped_lock lock(m_status_access);
    return workergroup->busy();
  }

char* AnalysisService::status() throw() {
  boost::recursive_mutex::scoped_lock lock(m_status_access);
    std::ostringstream os;
    workergroup->printStatus(os);
    os << "\n" << AnalysisAlgorithmMap::instance().getAllStatus() << std::endl;
    if (dcsinterface){
      os << "AnalysisService : DCS INFORMATION" << std::endl;
      dcsinterface->printStatus(os);
    }

    const unsigned length = os.str().length()+1;
    char *statusMessage = new char[length];
    // Have to make a new char* because IPC wants to delete it after use!
    strcpy(statusMessage, os.str().c_str());
    return statusMessage;
}

void AnalysisService::analyzeModule(const char* testname, const char* modulename) throw() {
    try {
        shared_ptr<TestData> testdata( new TestData() );
        ISInfoDictionary& id = SctNames::getISDictionary();
        ISInfo::Status result = id.findValue(testname, *testdata);

        if (result != ISInfo::Success) {
            string os = "Error reading from IS server.  Couldn't get: ";
            os += testname;
            throw IsException(result, os, __FILE__, __LINE__);
        }
        if (workergroup->debug()) std::cout << "read " << testname << " which has #scans=" <<  testdata->nScans << std::endl;
        if (workergroup->findTest(*testdata).get() == 0 ) {
            workergroup->addTest(testdata);
        }

        for (unsigned iscan=testdata->startScanNumber;
                iscan<(testdata->startScanNumber + testdata->nScans);
                ++iscan ) {
            {
                std::ostringstream name_raw;
                name_raw << "SctData::RawScanResult." <<  testdata->runNumber << "." << iscan << "." << modulename;
                if (workergroup->debug()) std::cout << "Looking for " << name_raw.str() << std::endl;
                ISInfoIterator rawIter(SctNames::getPartition(), SctNames::getEventDataName().c_str(), name_raw.str().c_str());
                while ( rawIter() ) {
                    try {
                        if (workergroup->debug())std::cout << "trying to add " << rawIter.name() << std::endl;
                        workergroup->push(shared_ptr<IOName>(new IONameIS(rawIter.name())));
                    } catch ( Sct::Throwable& e) {
                        e.sendToMrs(MRS_ERROR);
                    }
                }
            } {
                std::ostringstream name_fit;
                name_fit << "SctData::FitScanResult." <<  testdata->runNumber << "." << iscan << "." << modulename;
                if(workergroup->debug())std::cout << "Looking for " << name_fit.str() << std::endl;
                ISInfoIterator fitIter(SctNames::getPartition(), SctNames::getFittedDataName().c_str(), name_fit.str().c_str());
                while ( fitIter() ) {
                    try {
                        if (workergroup->debug()) std::cout << "trying to add " << fitIter.name() << std::endl;
                        workergroup->push(shared_ptr<IOName>(new IONameIS(fitIter.name())));
                    } catch ( Sct::Throwable& e) {
                        e.sendToMrs(MRS_ERROR);
                    }
                }
            }
        }
    } catch (Throwable& e) {
        e.sendToMrs(MRS_ERROR);
        return;
    }
}

void AnalysisService::purge() throw() {
    try {
        std::cout << "Doing purge" << std::endl;
        workergroup->purge();
        std::cout << "Memory purge completed" << std::endl;
    } catch (Throwable& e) {
        e.sendToMrs(MRS_ERROR);
        return;
    }
}

void AnalysisService::analyze(const char* name) throw() {
    analyzeModule(name, "*");
}

AnalysisServiceI::StringList* AnalysisService::listAlgorithms(){
  return Sct::copyStringListToCorba<AnalysisServiceI::StringList>(AnalysisAlgorithmMap::instance().listAlgorithms());
}

void AnalysisService::generalCallback(ISCallbackInfo * isc){
  try{
    IONameIS name(isc->name());
    if (name.getClassName()=="TestData"){
      testDataCallback(isc);
    }else if (name.getClassName()=="SctData::RawScanResult"
          || name.getClassName()=="SctData::FitScanResult") {
      scanResultCallback(isc);
    }
  }catch(Sct::Throwable& e){
    e.sendToMrs();
  }
}

void AnalysisService::scanResultCallback(ISCallbackInfo * isc) {
    try {
        if (isc->reason() != is::Created )
            return;
        instance().workergroup->push(shared_ptr<IOName>(new IONameIS(isc->name())));
    } catch(Sct::Throwable& e) {
        e.sendToMrs(MRS_ERROR);
    }
}

void AnalysisService::testDataCallback(ISCallbackInfo * isc) {
    try {
        shared_ptr<TestData> testdata( new TestData() );
        isc->value(*testdata);

        if (isc->reason() == is::Created ) { 
            std::cout << "adding " << isc->name() << std::endl; 
            instance().workergroup->addTest(testdata);
    } else if (isc->reason() == is::Updated && testdata->status == TestData::ABORTED) {
            std::cout << "removing aborted test " << isc->name() << std::endl; 
      instance().workergroup->removeTestsUpTo(testdata);
    } else if (isc->reason() == is::Updated) {
            std::cout << "replacing " << isc->name() << std::endl; 
        instance().workergroup->replaceTest(testdata);
    } else if (isc->reason() == is::Deleted ) {
            std::cout << "removing " << isc->name() << std::endl; 
            instance().workergroup->removeTestsUpTo(testdata);
        }
    } catch (Sct::Throwable& e) {
        e.sendToMrs(MRS_ERROR);
    }
}

void AnalysisService::setFitStrategy(const string& name) throw(LogicError) {
    SctNames::Mrs() << "ANALYSIS_FIT_STRATEGY"  << MRS_PARAM<const char*>("Strategy", name.c_str()) 
            << MRS_INFORMATION << MRS_TEXT("Fitting method is being set") << ENDM;
    fitStrategy = SctFitter::FitStrategyFactory::instance().getStrategy(name);
}

SctFitter::FitStrategy& AnalysisService::getFitStrategy() const throw(LogicError) {
    if (!fitStrategy)
        throw InvariantViolatedError("Fitter::getStrategy() no fit strategy defined", __FILE__, __LINE__);
    return *fitStrategy;
}

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

bool AnalysisService::isFifo(){
  return instance().workergroup->isFifo();
}

void AnalysisService::setFifo(bool val){
  instance().workergroup->setFifo(val);
}

AnalysisService& AnalysisService::initialize(AnalysisArguments args) {
    service = new AnalysisService(args);
    return *service;
}

DcsInterface& AnalysisService::getDcsInterface(){
  if (!dcsinterface){
    throw ConfigurationException("Cant return dcsinterface",__FILE__, __LINE__);
  }else{
    return *dcsinterface;
  }
}

void AnalysisService::recover(){
  SctNames::Mrs() << "ANALYSIS_RECOVERY" << MRS_TEXT("Analysis recovery started in recovery mode") << ENDM;
  Sct::UnsupportedOperationError("Recovery not implimented", __FILE__, __LINE__).sendToMrs(MRS_DIAGNOSTIC);
}


AnalysisService* AnalysisService::service=0;

}// end of namespace SctAnalysis

---