---

AnalysisWorkerGroup.cpp

Go to the documentation of this file.

#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"

using namespace SctService;
using namespace boost;

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;

    for ( list<shared_ptr<AnalysisAlgorithm> >::const_iterator it = m_algorithms.begin();
            it != m_algorithms.end(); ++it ) {
        os << "      " << (*it)->getModuleName()
        << "  # of scans: " << (*it)->getTestResult()->getNScans() << " of " << 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.
    std::string s;
    do {s=pop(); } while( s.length() != 0 );
    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(std::string name) throw() {
    using namespace SctData;
    try {
        //      cout << "AnalysisWorker: name = "<< *name << " Q="<<wg->queueSize()<<", busy="<< wg->busy() << endl;

        Sct::IS::IONameIS isname(name);
        const unsigned runno = isname.getRunNumber();
        const unsigned scanno = isname.getScanNumber();
        const string modulename= isname.getModuleName();

        //      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)
            throw InvalidArgumentError("AnalysisWorker::work. No test found for scan", __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=(isname.getClassName()==string("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->canAddFitScanResult(name);
        } else {
            //      cout << "got raw from queue" << endl;
            algorithm->canAddRawScanResult(name);
        }
        if ( algorithm->isDone() ) {
            //      cout << "Removing algorithm" << endl;
        talgs->removeAlgorithm(algorithm);
        }
    } 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

---