Fitter.cpp

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

#include "Sct/SctNames.h"
#include "Sct/SctParameters.h"

#include <iostream>
#include <vector>
#include <cassert>
#include <ctime>
#include <TMath.h>
#include <boost/timer.hpp>

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)
        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())
    ,m_nFitsDone(0), m_nFitErrors(0), m_nScans(0), m_scanTimeTaken(0), m_ioTimeTaken(0), m_scan("No last scan"), m_fitStrategy(0) {

    auto_ptr<ISInfoReceiver> ir(new ISInfoReceiver(SctNames::getPartition()));
    if (!ir.get())
        throw ConfigurationException("Fitter::initialize can't make infoReceiver ", __FILE__, __LINE__) ;
    m_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=m_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) {
    // find out why the callback function has been called:
    if (isc->reason() != ISInfoCreated && isc->reason() != ISInfoUpdated )
    return;
    
    // put it in the queue
    Fitter::instance().workergroup->push(isc->name());
}

void Fitter::fit(FitterIStatus* status, char* name) throw() {
    // put it in the queue
    workergroup->push(name);
    status->returnCode=FitterIReply_Success;
}

void Fitter::fitAll(FitterIStatus* status) throw() {

    ISInfoIterator iter(SctNames::getPartition(), SctNames::getEventDataName().c_str(), "SctData::RawScanResult.*");
    while (iter()) {
    // put it in the queue
    workergroup->push(iter.name());
    status->returnCode=FitterIReply_Success;
    }
}

char* Fitter::getFitOptions(FitterIStatus* status) throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    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::status(FitterIStatus* stat) throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    stat->returnCode = FitterIReply_Success;
    // Have to make a new char* because IPC wants to delete it after use!
    string msg = status();
    char *statusMessage = new char[msg.length()+1];
    strcpy(statusMessage, msg.c_str());
    return statusMessage;
}

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

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

void Fitter::useAnalyticAlgorithm (FitterIStatus *_status, ilu_Boolean use) {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    if (use) {
    ThresholdFitAlgorithm2::putInMap();
    } else {
    ThresholdFitAlgorithm::putInMap();
    }
}

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

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

char* Fitter::getFitStrategy(FitterIStatus* status) throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    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::setFitStrategy(FitterIStatus* status, char *n) throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    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::status() const throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    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=" << nFitsDone() << ", Errors=";
    oss << nFitErrors() << ", Scans Done=" << m_nScans << "\nLast scan: " << lastScan(); 
    oss << "\n Timing.  I/O time: " << m_ioTimeTaken << " Scan time: "; 
    oss << m_scanTimeTaken << " Approx average time/scan/thread: " << getAverageTimePerScan() << endl;
    m_status = oss.str();
    return m_status.c_str();
}

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::nFitsDone() const throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    return m_nFitsDone;
}

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

long Fitter::nFitsDone(FitterIStatus* status) throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    status->returnCode = FitterIReply_Success;
    return this->nFitsDone();
}

long Fitter::nFitErrors(FitterIStatus* status) throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    status->returnCode = FitterIReply_Success;
    return this->nFitErrors();
}

long Fitter::queueLength(FitterIStatus* status) {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    status->returnCode = FitterIReply_Success;
    return workergroup->queueSize();
}

long Fitter::busy(FitterIStatus* status) {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    status->returnCode = FitterIReply_Success;
    return workergroup->busy();
}

const char* Fitter::lastScan() const throw() {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    return m_scan.c_str();
}

char* Fitter::lastScan(FitterIStatus* status) throw() {    
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    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) {
    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::setFitStrategy(string name) throw(LogicError) {
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    m_fitStrategy = FitStrategyFactory::instance().getStrategy(name);
}

} // end of namespace SctFitter

---