AnalysisService.cpp

#include "AnalysisService.h"
#include "AnalysisAlgorithmMap.h"
#include "AnalysisAlgorithm.h"
#include "AnalysisWorkerGroup.h"

#include "Sct/SctNames.h"
#include "Sct/IpcObjectException.h"
#include "Sct/IS/IOManagerIS.h"
#include "Sct/IS/IONameIS.h"
#include "Sct/ConfigurationException.h"
#include "SctData/TestResult.h"
#include "CalibrationController/IS/TestData.h"

#include <pmg/pmg_initSync.h>
#include <boost/scoped_ptr.hpp>

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

void pmgSynch(void *) {
    pmg_initSync();
}

int main(int argc, char** argv) {
    using namespace SctAnalysis;
    setExceptionHandlers(argv[0]);

    bool multiThreaded = true; //Choose single/multi-threaded
    IPCCore::init(multiThreaded);
    AnalysisService& s=AnalysisService::instance();

    try {
        s.setFitStrategy("NagFitStrategy");
    } catch(LogicError& e) {
        //No need to do anything - stick with default
    }

    try {
        s.getFitStrategy().setOptions("NQR");
        
        if (!s.publish()) {
            IpcObjectException e("Failed to publish Analysis Service", __FILE__, __LINE__);
            e.sendToMrs(MRS_ERROR);
        }
        s.getServer().doSoon(pmgSynch, NULL);
        s.run();
        s.getServer().run();
        s.withdraw();
    } catch (Throwable& e) {
        e.sendToMrs(MRS_FATAL);
        terminate();
    }
}

namespace SctAnalysis {

AnalysisService::AnalysisService(const string& fitStrategyName) : IPCObject(AnalysisServiceI_C_AnalysisServiceInterface_instanceName, &getServer()),
    infoReceiver(new ISInfoReceiver(SctNames::getPartition())) {
    if (!infoReceiver.get())
        throw ConfigurationException("AnalysisService::AnalysisService can't make infoReceiver ", __FILE__, __LINE__) ;
    setFitStrategy(fitStrategyName);
    workergroup = new AnalysisWorkerGroup();
}

IPCServer& AnalysisService::getServer() throw() {
    static IPCServer server(AnalysisServiceI_C_AnalysisServiceInterface_serverName, SctNames::getPartition());
    return server;
}

void AnalysisService::run() {
    // start up worker thread.
    workergroup->go(1);

    // subscribe to various IS stuff.
    infoReceiver->subscribe(SctNames::getControlDataName().c_str(), ".*TestData.*", testDataCallback, this);
    infoReceiver->subscribe(SctNames::getFittedDataName().c_str(),  ".*FitScanResult.*", scanResultCallback, this);
    infoReceiver->subscribe(SctNames::getEventDataName().c_str(), ".*RawScanResult.*", scanResultCallback, this);
}

  ilu_ShortInteger AnalysisService::busy(AnalysisServiceIStatus* status) {
    boost::recursive_mutex::scoped_lock lock(m_status_access);
     status->returnCode = AnalysisServiceIReply_Success;
    return workergroup->busy();
  }

  ilu_ShortInteger AnalysisService::queueLength(AnalysisServiceIStatus* status) {
  boost::recursive_mutex::scoped_lock lock(m_status_access);
     status->returnCode = AnalysisServiceIReply_Success;
    return workergroup->busy();
  }

char* AnalysisService::status(AnalysisServiceIStatus* status) throw() {
  boost::recursive_mutex::scoped_lock lock(m_status_access);
    status->returnCode = AnalysisServiceIReply_Success;
    // Have to make a new char* because IPC wants to delete it after use!
    ostringstream os;
    workergroup->printStatus(os);
    os << "\n" << AnalysisAlgorithmMap::instance().getAllStatus() << endl;
    const unsigned length = os.str().length()+1;
    char *statusMessage = new char[length];
    strcpy(statusMessage, os.str().c_str());
    return statusMessage;
}

void AnalysisService::analyzeModule(AnalysisServiceIStatus* status, char* testname, 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__);
        }
        cout << "read " << testname << " which has #scans=" <<  testdata->nScans << endl;
        if (workergroup->findTest(*testdata).get() == 0 ) {
            workergroup->addTest(testdata);
        }

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

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

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

void AnalysisService::scanResultCallback(ISCallbackInfo * isc) {
    try {
        if (isc->reason() != ISInfoCreated )
            return;
        cout << "AnalysisService scanResultCallback on " << string(isc->name())<<endl;
        //cout << "AnalysisService scanResultCallback busy=" << AnalysisWorkerGroupFactory::pointer()->busy()
        //<< " Q=" << AnalysisWorkerGroupFactory::pointer()->queueSize()
        //<< " Nworker=" << AnalysisWorkerGroupFactory::pointer()->nWorkers() << endl;
        instance().workergroup->push(shared_ptr<IOName>(new IONameIS(isc->name())));
        //cout << " scanResultCallback done " << endl;
    } 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() == ISInfoCreated ) {
            cout << "Callback for new testdata object "<<endl;
            if (instance().workergroup->findTest(*testdata).get() == 0 ) {
                //cout << "Listener adding test" << endl;
                instance().workergroup->addTest(testdata);
                //cout << "Listener added test" << endl;
            } else {
                throw Sct::InvalidArgumentError(string("AnalysisService::TestCallback Test already exists: ")+string(isc->name()), __FILE__, __LINE__);
            }

    } else if (isc->reason() == ISInfoUpdated && testdata->status == TestData::ABORTED) {
        instance().workergroup->removeTestsUpTo(testdata);
        
    } else if (isc->reason() == ISInfoDeleted ) {
            instance().workergroup->removeTestsUpTo(testdata);
        }
    } catch (Sct::Throwable& e) {
        e.sendToMrs(MRS_ERROR);
    }
}

void AnalysisService::setFitStrategy(const string& name) throw(LogicError) {
    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)
        return initialize();
    return *service;
}

AnalysisService& AnalysisService::initialize(const string& fitStrategyName) {
    if (service) {
        service->setFitStrategy(fitStrategyName);
    } else {
        service = new AnalysisService(fitStrategyName);
    }
    return *service;
}

AnalysisService* AnalysisService::service = 0;

}// end of namespace SctAnalysis

---