---

Fitter.cpp

Go to the documentation of this file.

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

#include "SctData/RawScanResult.h"
#include "Sct/IS/IOManagerIS.h"
#include "Sct/SctNames.h"
#include "Sct/SctParameters.h"
#include "Sct/Timer.h"

#include <iostream>
#include <vector>
#include <cassert>
#include <ctime>
#include <TMath.h>

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

namespace SctFitter {
void doFits(ISCallbackInfo * isc);

Fitter* Fitter::fitter = 0;

Fitter& Fitter::instance() {
    if (!fitter)
        return initialize();
    return *fitter;
}

Fitter& Fitter::initialize(const string& fitStrategyName) throw(ConfigurationException) {
    if (fitter) {
        fitter->setFitStrategy(fitStrategyName);
        return *fitter;
    }
    fitter = new Fitter(fitStrategyName);
    return *fitter;
}

Fitter::Fitter(const string& fitStrategyName) throw(ConfigurationException) : IPCObject(FitterI_C_FitterInterface_instanceName, &getFitterServer())
    ,nFitsDone(0), nFitErrors(0), nScans(0), scanTimeTaken(0), ioTimeTaken(0), m_scan("No last scan"), fitStrategy(0) {

    auto_ptr<ISInfoReceiver> ir(new ISInfoReceiver(SctNames::getPartition()));
    if (!ir.get())
        throw ConfigurationException("Fitter::initialize can't make infoReceiver ", __FILE__, __LINE__) ;
    infoReceiver  = ir;

    setFitStrategy(fitStrategyName);
    workergroup = new FitterWorkerGroup();
}

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

IPCServer& Fitter::getFitterServer() throw() {
    static IPCServer fitterServer(FitterI_C_FitterInterface_serverName, SctNames::getPartition());
    return fitterServer;
}



void Fitter::go(unsigned nWorker) throw(IsException) {
    ISInfo::Status s=infoReceiver->subscribe(SctNames::getEventDataName().c_str(),"SctData::RawScanResult.*", doFits);
    if (s!=ISInfo::Success) {
        ostringstream os;
        os <<"Fitter::go() Could not subscribe to "<<SctNames::getEventDataName();
        throw IsException(s, os.str(), __FILE__, __LINE__);
    }
    workergroup->go(nWorker);

    //Not done since Fitter is an IPC server, and so IPCServer.run() is called elsewhere.
    //infoReceiver->run();
}

// The infoReceiver callback function must take a static function as an argument:
void Fitter::doFits(ISCallbackInfo * isc) throw(IsException, LogicError) {
    using namespace SctData;
    try {
    double start = timer();
        // find out why the callback function has been called:
        if (isc->reason() != ISInfoCreated && isc->reason() != ISInfoUpdated )
            return;

        shared_ptr<const Serializable> ob ( IOManagerIS::instance().read(*isc) );
        shared_ptr<const RawScanResult> raw = dynamic_pointer_cast<const RawScanResult>(ob);
        if (!raw)
            throw InvalidArgumentError("Fitter::doFits() not a RawScanResult", __FILE__, __LINE__);
        // put it in the queue
        Fitter::instance().workergroup->push(raw);
    Fitter::instance().addIOTime(timer() - start);
    } catch (Throwable& e) {
        e.sendToMrs(MRS_ERROR);
    }
}

void Fitter::ipcFit(FitterIStatus* status, char* name) throw() {
    try {
    double start = timer();
        shared_ptr<const Serializable> ob = IOManagerIS::instance().read(string(name));
        shared_ptr<const RawScanResult> raw = dynamic_pointer_cast<const RawScanResult>(ob);
        if (!raw)
            throw InvalidArgumentError("Fitter::ipcFitAll() not a RawScanResult", __FILE__, __LINE__);
        // put it in the queue
        workergroup->push(raw);
        status->returnCode=FitterIReply_Success;
    addIOTime(timer() - start);
    } catch (Throwable& e) {
        e.sendToMrs(MRS_ERROR);
        return;
    }
}
void Fitter::ipcFitAll(FitterIStatus* status) throw() {

    ISInfoIterator iter(SctNames::getPartition(), SctNames::getEventDataName().c_str(), "SctData::RawScanResult.*");
    while (iter()) {
        try {
        double start = timer();
            shared_ptr<const Serializable> ob = IOManagerIS::instance().read(iter);
            shared_ptr<const RawScanResult> raw = dynamic_pointer_cast<const RawScanResult>(ob);
            if (!raw)
                throw InvalidArgumentError("Fitter::ipcFitAll() not a RawScanResult", __FILE__, __LINE__);
            // put it in the queue
            workergroup->push(raw);
            status->returnCode=FitterIReply_Success;
        addIOTime(timer() - start);
        } catch (Throwable& e) {
            e.sendToMrs(MRS_ERROR);
            return;
        }
    }
}

char* Fitter::ipcGetFitOptions(FitterIStatus* status) throw() {
    string options = "";
    try {
        options = getFitStrategy().getOptions();        
    } catch (LogicError &e) {
    //Um don't do anything!
    }
    status->returnCode = FitterIReply_Success;
    char *newOpt=new char[options.length()+1];
    strcpy(newOpt, options.c_str());
    return newOpt;
}

char* Fitter::ipcFitterStatus(FitterIStatus* status) throw() {
    status->returnCode = FitterIReply_Success;
    // Have to make a new char* because IPC wants to delete it after use!
    string msg = getStatus();
    char *statusMessage = new char[msg.length()+1];
    strcpy(statusMessage, msg.c_str());
    return statusMessage;
}

void Fitter::setFitOptions(string opt) throw (LogicError) {
    getFitStrategy().setOptions(opt);
}

void Fitter::ipcSetFitOptions(FitterIStatus* status, char *n) throw() {
    string name( n );
    try {
        setFitOptions(n);
    } catch (LogicError &e) {
        return;
    }
    status->returnCode = FitterIReply_Success;
}

void Fitter::ipcUseAnalyticAlgorithm (FitterIStatus *_status, ilu_Boolean use) {
    if (use) {
    ThresholdFitAlgorithm2::putInMap();
    } else {
    ThresholdFitAlgorithm::putInMap();
    }
}

bool Fitter::isUsingAnalyticAlgorithm() const {
    return ThresholdFitAlgorithm2::isInMap();
}

ilu_Boolean Fitter::ipcIsUsingAnalyticAlgorithm(FitterIStatus *_status) {
    _status->returnCode = FitterIReply_Success;
    return isUsingAnalyticAlgorithm();
}

char* Fitter::ipcGetFitStrategy(FitterIStatus* status) throw() {
    string strat;
    try {
        strat = getFitStrategy().getName();
    } catch (LogicError &e) {
    //Do nothing!
    }    
    status->returnCode = FitterIReply_Success;
    char *newOpt=new char[strat.length()+1];
    strcpy(newOpt, strat.c_str());
    return newOpt;
}

void Fitter::ipcSetFitStrategy(FitterIStatus* status, char *n) throw() {
    string name( n );
    cout <<"trying to set fit strategy "<<name<<endl;
    try {
        setFitStrategy(name);
    } catch (LogicError& e) {
        cout <<"failed  to set fit strategy "<<name<<endl;
        return;
    }
    cout <<"set fit strategy "<<name<<endl;
    status->returnCode = FitterIReply_Success;
}

const char* Fitter::getStatus() const throw() {
    string strategyName;
    try {
        strategyName=getFitStrategy().getName();
    } catch (LogicError &e) {
        strategyName="NONE";
    }
    ostringstream oss;
    oss << "Fitter Strategy=" << strategyName <<  " Fit Options=" << getFitStrategy().getOptions()
    << " Using analytic fits=" << isUsingAnalyticAlgorithm() << endl;
    oss << "Workers=" << workergroup->nWorkers();
    oss << ", (Busy=" << workergroup->busy() << "), Fits Done=" << getNFitsDone() << ", Errors=";
    oss << getNFitErrors() << ", Scans Done=" << nScans << "\nLast scan: " << lastScan(); 
    oss << "\n Timing.  Read I/O time: " << ioTimeTaken << " Scan time: "; 
    oss << scanTimeTaken << " Approx average time/scan/thread: " << getAverageTimePerScan() << endl;
    status = oss.str();
    return status.c_str();
}

void Fitter::incrementFitErrors() throw() {
    mutex::scoped_lock scoped_lock(counterMutex);
    ++nFitErrors;
}

void Fitter::incrementFitsDone() throw() {
    mutex::scoped_lock scoped_lock(counterMutex);
    ++nFitsDone;
}

void Fitter::scanDone(double time) throw() {
    mutex::scoped_lock scoped_lock(counterMutex);
    ++nScans;
    scanTimeTaken += time;
}

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


double Fitter::getAverageTimePerScan() const throw() {
    double time = ioTimeTaken/(1+workergroup->nWorkers()) + scanTimeTaken/workergroup->nWorkers();
    if (nScans > 0) time /= nScans;     
    return time;
}

long Fitter::getNFitsDone() const throw() {
    return nFitsDone;
}

long Fitter::getNFitErrors() const throw() {
    return nFitErrors;
}

long Fitter::ipcNFitsDone(FitterIStatus* status) throw() {
    status->returnCode = FitterIReply_Success;
    return this->getNFitsDone();
}

long Fitter::ipcNFitErrors(FitterIStatus* status) throw() {
    status->returnCode = FitterIReply_Success;
    return this->getNFitErrors();
}

const char* Fitter::lastScan() const throw() {
    return m_scan.c_str();
}
char* Fitter::ipcLastScan(FitterIStatus* status) throw() {    
    char * message = new char[m_scan.length()+1];
    strcpy(message, m_scan.c_str() );
    status->returnCode = FitterIReply_Success;
    return message;
}

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

void Fitter::setFitStrategy(string name) throw(LogicError) {
    fitStrategy = FitStrategyFactory::instance().getStrategy(name);
}

} // end of namespace SctFitter

---