---

AnalysisService.cpp

Go to the documentation of this file.

#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 "SctData/TestResult.h"

#include <boost/scoped_ptr.hpp>

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

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

    bool multiThreaded = false; //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.run();
    } catch (Throwable& e) {
        e.sendToMrs(MRS_FATAL);
        terminate();
    }
}

namespace SctAnalysis {

AnalysisService::AnalysisService(const string& fitStrategyName) : IPCObject(AnalysisServiceI_C_AnalysisServiceInterface_instanceName, &getServer()) {
    setFitStrategy(fitStrategyName);
    workergroup = new AnalysisWorkerGroup();
}

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

void AnalysisService::run() {
    ISInfoReceiver r(SctNames::getPartition());

    // start up worker thread.
    workergroup->go(1);

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

char* AnalysisService::ping(AnalysisServiceIStatus* status) throw() {
    status->returnCode = AnalysisServiceIReply_Success;
    // Have to make a new char* because IPC wants to delete it after use!
    char *message = new char[60];
    strcpy(message,"Hello from Analysis Service");
    return message;
}

char* AnalysisService::ipcStatus(AnalysisServiceIStatus* status) throw() {
    status->returnCode = AnalysisServiceIReply_Success;
    // Have to make a new char* because IPC wants to delete it after use!
    char *statusMessage = new char[10000];
    ostringstream os;
    workergroup->printStatus(os);
    strcpy(statusMessage, os.str().c_str());
    return statusMessage;
}

void AnalysisService::ipcAnalyzeModule(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( 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(fitIter.name());
                    } catch ( Sct::Throwable& e) {
                        e.sendToMrs(MRS_ERROR);
                    }
                }
            }
        }
        status->returnCode = AnalysisServiceIReply_Success;
    } catch (Throwable& e) {
        e.sendToMrs(MRS_ERROR);
        return;
    }
}

void AnalysisService::ipcPurge(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::ipcAnalyze(AnalysisServiceIStatus* status, char* name) throw() {
    ipcAnalyzeModule(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(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

---