AnalysisWorkerGroup.cpp

#include "AnalysisWorkerGroup.h"
#include "AnalysisAlgorithm.h"
#include "AnalysisAlgorithmMap.h"
#include "Sct/IS/IOManagerIS.h"
#include "Sct/ISProxy/IOManagerISProxy.h"
#include "Sct/IS/IONameIS.h"
#include "Sct/StdExceptionWrapper.h"
#include "SctData/UniqueID.h"
#include "SctData/TestResult.h"
#include "CalibrationController/IS/TestData.h"
#include <boost/timer.hpp>

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

namespace SctAnalysis {

//TestAlgs methods


AnalysisWorkerGroup::TestAlgs::~TestAlgs() {
    boost::recursive_mutex::scoped_lock lock(m_access);

    //We need to explicitly delete the list here so the lock is in force
    //Note that this just removes the pointer..actual deletion will only occur
    //if noone else has grabbed a reference to the object
    m_algorithms.clear();
}

void AnalysisWorkerGroup::TestAlgs::removeAlgorithm(shared_ptr<AnalysisAlgorithm> alg) {
    boost::recursive_mutex::scoped_lock lock(m_access);
    list<shared_ptr<AnalysisAlgorithm> >::iterator it =
        find(m_algorithms.begin(), m_algorithms.end(), alg);

    if ( it != m_algorithms.end() )
        m_algorithms.erase(it);
}

ostream& AnalysisWorkerGroup::TestAlgs::printStatus(ostream& os) const throw() {
    boost::recursive_mutex::scoped_lock lock(m_access);
    os << "   Test name: "  << getTest().testName 
       << ", run number = " << getTest().runNumber
       << ", first scan number = " << getTest().startScanNumber
       << ", nscans = " << getTest().nScans 
       <<  endl;

    return os;
}

shared_ptr<AnalysisAlgorithm> AnalysisWorkerGroup::TestAlgs::findAlgorithm(const string& modulename) {
    boost::recursive_mutex::scoped_lock lock(m_access);
    shared_ptr<AnalysisAlgorithm> alg;
    for (list<shared_ptr<AnalysisAlgorithm> >::const_iterator it=m_algorithms.begin();
            it != m_algorithms.end();
            ++it) {
        if ( (*it)->getModuleName()==modulename ) {
            alg=(*it);
            break;
        }
    }
    if (!alg) {
        alg=addAlgorithm(modulename);
    }
    return alg;
}

shared_ptr<AnalysisAlgorithm> AnalysisWorkerGroup::TestAlgs::addAlgorithm(const string& modulename) {
    boost::recursive_mutex::scoped_lock lock(m_access);
    shared_ptr<AnalysisAlgorithm> alg = AnalysisAlgorithmMap::instance().getAlgorithm(getTest(), modulename);
    m_algorithms.push_back(alg);
    return alg;
}



//AnalysisWorkerGroup methods


ostream&  AnalysisWorkerGroup::printStatus(ostream& os) const throw() {
    cout << "Getting tests lock" << endl;
    boost::recursive_mutex::scoped_lock lock(m_tests_access);
    cout << "got tests lock" << endl;
    os << "===============================================\n" << endl;
    os << "AnalysisWorkerGroup tests:" << endl;
    unsigned ntest=0;
    for (list<shared_ptr<TestAlgs> >::const_iterator it = m_tests.begin();
            it != m_tests.end() ; ++it ) {
        (*it)->printStatus(os);
        os << "-------------------------------------------------"<<endl;
        ++ntest;
    }
    os << endl << "Total number of tests = " << ntest << endl;
    cout << "done ostreaming" << endl;
    return os;
}

void AnalysisWorkerGroup::addTest(shared_ptr<const TestData> testdata) {
    boost::recursive_mutex::scoped_lock lock(m_tests_access);
    m_tests.push_back(shared_ptr<TestAlgs>(new TestAlgs(testdata) ) );
}

void AnalysisWorkerGroup::removeTestsUpTo(shared_ptr<const TestData> testdata) {
    boost::recursive_mutex::scoped_lock lock(m_tests_access);
    list<shared_ptr<TestAlgs> >::iterator it = m_tests.begin();

    while( it != m_tests.end() ) { // no ++it here since we need to delete it in the function!
        if ((*it)->getTest().runNumber==testdata->runNumber
                && (*it)->getTest().startScanNumber<=testdata->startScanNumber) {
            
        cout << "Deleting old test for run "
         << (*it)->getTest().runNumber << ", start scan "
                 << (*it)->getTest().startScanNumber
                 << endl;
            it=m_tests.erase(it);   // erase returns ++it
        } else {
            ++it;                   // increment ourselves
        }
    }
}

shared_ptr<AnalysisWorkerGroup::TestAlgs> AnalysisWorkerGroup::findTest(const TestData& testdata) const throw() {
    return findTest(testdata.runNumber, testdata.startScanNumber);
}

shared_ptr<AnalysisWorkerGroup::TestAlgs> AnalysisWorkerGroup::findTest(const unsigned long runno, const unsigned long scanno) const throw() {
    boost::recursive_mutex::scoped_lock lock(m_tests_access);
    shared_ptr<AnalysisWorkerGroup::TestAlgs> talgs;
    for (list<shared_ptr<TestAlgs> >::const_iterator it = m_tests.begin();
            it != m_tests.end();
            ++it ) {
        const TestData& td=(*it)->getTest();
        if ( runno == td.runNumber &&
                scanno >= td.startScanNumber &&
                scanno < td.startScanNumber + td.nScans) {
            talgs=*it;
            //      cout << "found test for run " << runno<< ", scan " << scanno << endl;
        }
    }
    return talgs;
}




void AnalysisWorkerGroup::purge() throw() {
    //Stop the workers from trying to get on with things
    this->setPaused(true);
    while (paused() != nWorkers()) sleep(1);
    
    // Clear the queue.
    shared_ptr<IOName> s;
    do {s=pop(); } while(s);
    boost::recursive_mutex::scoped_lock lock(m_tests_access);

    // Clear the test data.
    cout << "Clearing tests ... " << endl;
    m_tests.clear();
    cout << " ... done clearing tests" << endl;
    this->setPaused(false);    
}

void AnalysisWorkerGroup::work(shared_ptr<IOName> name) throw() {
    using namespace SctData;
    timer t;
    try {
        //      cout << "AnalysisWorker: name = "<< *name << " Q="<<wg->queueSize()<<", busy="<< wg->busy() << endl;        
    UniqueID id (name->getUniqueID());
        const unsigned runno = id.getRunNumber();
        const unsigned scanno = id.getScanNumber();
        const string modulename= id.getModule();

        //      cout << "run="<< runno << ", scan="<< scanno << ", modulename="<<modulename
        //       << ", classname="<<isname.getClassName()<<endl;

        shared_ptr<AnalysisWorkerGroup::TestAlgs> talgs = findTest(runno, scanno);

        //      cout << "Found test" << endl;

        if (!talgs) {
           std::ostringstream oss; oss << "AnalysisWorker::work. No test found for run " 
                               << runno << " scan" << scanno;
            throw IllegalStateError(oss.str(), __FILE__, __LINE__);
        }
        // Get a testresult. Add a new testresult if one dosen't exist for this module
        shared_ptr<AnalysisAlgorithm> algorithm = talgs->findAlgorithm(modulename);

        //      cout << "algorithm.get()=" << algorithm.get() << endl;

        bool isfit=(name->getClassName()=="SctData::FitScanResult");

        // lock the approprate result and do your stuff, analysis-algorithm.
        boost::recursive_mutex::scoped_lock lock (algorithm->getMutex());
        if (isfit) {
            //      cout << "got fit from queue" << endl;
            algorithm->addFitScanResult(name);
        } else {
            //      cout << "got raw from queue" << endl;
            algorithm->addRawScanResult(name);
        }
    if (algorithm->canAnalyze()) {
        if (algorithm->isDone()) throw IllegalStateError("AnalysisWorker::work.  Algorithm has additional, unnecessary Scan - analysis already done", __FILE__, __LINE__);
        algorithm->checkTestResultExists();
        algorithm->addOverheadTime(t.elapsed());
        
        t.restart();
        algorithm->loadData();
        algorithm->addIOTime(t.elapsed());
        
        t.restart();
        algorithm->analyze();
        algorithm->addAnalysisTime(t.elapsed());
        
        t.restart();
        algorithm->finish();
        algorithm->addIOTime(t.elapsed());
        talgs->removeAlgorithm(algorithm);
    } else {
        algorithm->addOverheadTime(t.elapsed());
    }
    
    } catch (InvalidArgumentError& e) {
    e.sendToMrs();
    } catch(Throwable& e) {
        e.sendToMrs(MRS_ERROR);
    } catch(std::exception& e) {
    StdExceptionWrapper sew(e);
        sew.sendToMrs(MRS_ERROR);        
    } catch(...) {
    Error e("uncaught unknown exception", __FILE__, __LINE__);
    e.sendToMrs(MRS_ERROR);
    }
    //  cout << "Work done" << endl;
}


}// end of namespace SctAnalysis

---