AnalysisWorkerGroup.cpp

#include "AnalysisWorkerGroup.h"
#include "AnalysisAlgorithm.h"
#include "AnalysisAlgorithmMap.h"
#include "AnalysisService.h"
#include "DcsInterface.h"

#include "Sct/IS/IOManagerIS.h"
#include "Sct/ISProxy/IOManagerISProxy.h"
#include "Sct/IS/IONameIS.h"
#include "Sct/StdExceptionWrapper.h"
#include "Sct/SctNames.h"
#include "Sct/UniqueID.h"
#include "SctData/TestResult.h"
#include "SctData/DcsData.h"

#include "CalibrationController/IS/TestData.h"
#include <boost/timer.hpp>
#include <iomanip>
#include <is/info.h>
#include <is/infodictionary.h>
#include <is/infoiterator.h>
#include <is/inforeceiver.h>
#include <is/infoT.h>

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::shared_ptr<const TestData> testdata) : m_testdata(testdata){
  bool dcsProblemsReported=false;
  for (unsigned imodule=0; imodule<m_testdata->modules_size; ++imodule){
    // make dcs snapshots of all modules in test
    try{
      const std::string& module = m_testdata->modules[imodule];
      boost::shared_ptr<AnalysisAlgorithm> alg = findAlgorithm(module);
      if (!alg.get()) continue; 
      alg->checkTestResultExists();

      // try to get DcsInformation for this module
      try{
    if (!AnalysisService::instance().getDcsInterface().getConfigurationInterface().empty()) {  
      shared_ptr<SctData::DcsData> dcs (AnalysisService::instance().getDcsInterface().getData(module) );
      
      alg->getTestResult()->setDcsData(dcs);
    }
      }catch(Sct::Throwable& e){
    if (!dcsProblemsReported) {
      e.sendToMrs(MRS_DIAGNOSTIC);
      dcsProblemsReported=true;
    }
      }catch(std::exception& e){
    if (!dcsProblemsReported) {
      StdExceptionWrapper(e,__FILE__,__LINE__).sendToMrs(MRS_DIAGNOSTIC);
      dcsProblemsReported=true;
    }
      }
    }catch(Sct::Throwable& e){
      e.sendToMrs();
    }
  }
}

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);
}

shared_ptr<const TestData>AnalysisWorkerGroup::TestAlgs::getTest() const{
  if (!m_testdata.get()) throw Sct::InvariantViolatedError("Attempt to retrieve null TestData - object is not properly constructed",__FILE__,__LINE__);
  return m_testdata;
}

void AnalysisWorkerGroup::TestAlgs::replaceTest(shared_ptr<const TestData> testdata){
  boost::recursive_mutex::scoped_lock test_lock(m_access);
  m_testdata=testdata;
  // also need to replace the TestData in the Algorithms!
  for (std::list<shared_ptr<AnalysisAlgorithm> >::iterator it = m_algorithms.begin();
       it != m_algorithms.end();
       ++it){
    // lock the algorithm
    boost::recursive_mutex::scoped_lock alg_lock((*it)->getMutex());
    (*it)->setTestData(testdata);
  }
}

boost::recursive_mutex& AnalysisWorkerGroup::TestAlgs::getMutex() const {
  return m_access;
}

ostream& AnalysisWorkerGroup::TestAlgs::printStatus(ostream& os) const throw() {
    boost::recursive_mutex::scoped_lock lock(m_access);
    os << setw(26) << getTest()->testName 
       << setw(8) << getTest()->runNumber
       << setw(8) << getTest()->startScanNumber
       << setw(8) << getTest()->nScans 
       << setw(26) << getTest()->analysisAlgorithm
       << "\n\t options = [" << getTest()->options << "]"
       << 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
AnalysisWorkerGroup::AnalysisWorkerGroup() : m_debug(false){
}

AnalysisWorkerGroup::~AnalysisWorkerGroup() {
}

bool AnalysisWorkerGroup::debug() {
  return m_debug;
}

ostream&  AnalysisWorkerGroup::printStatus(ostream& os) const throw() {
    boost::recursive_mutex::scoped_lock lock(m_tests_access);
    os << "\t === Analysis Service ===\n" << endl;
    os << "AnalysisWorkerGroup tests:" << endl;
    unsigned ntest=0;
    os << setw(26) << "TestName" 
       << setw(8) << "Run"
       << setw(8) << "Scan"
       << setw(8) << "NScans" 
       << setw(26) << "AnalysisAlg"
       <<  endl;
    os << "----------------------------------------------------------------"<<endl;
    for (list<shared_ptr<TestAlgs> >::const_iterator it = m_tests.begin();
            it != m_tests.end() ; ++it ) {
        (*it)->printStatus(os);
        ++ntest;
    }
    os << "----------------------------------------------------------------"<<endl;
    os << endl;
    os << "\tWorkers=" << nWorkers() << "\t  (Busy=" << busy() << ")"
       << "\tQueue =" << queueSize();
    if (isFifo()) {
      os << " [FIFO]\n";
    }else{
      os << " [FILO]\n";
    }
    os <<"Total number of tests = " << ntest << endl;
    return os;
}

void AnalysisWorkerGroup::addTest(shared_ptr<const TestData> testdata) {
    boost::recursive_mutex::scoped_lock lock(m_tests_access);
    boost::shared_ptr<TestAlgs> talgs;
    talgs=findTest(*testdata);
    if (talgs) {
      std::cout << "WARNING  << adding rather than replacing test" << std::endl;
      replaceTest(testdata);
    }else{
      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!
      // lock the TestAlgs
      boost::recursive_mutex::scoped_lock lock((*it)->getMutex());
        if ((*it)->getTest()->runNumber==testdata->runNumber
                && (*it)->getTest()->startScanNumber<=testdata->startScanNumber) {
            
      std::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);
}

void AnalysisWorkerGroup::replaceTest(shared_ptr<const TestData> testdata){
  boost::recursive_mutex::scoped_lock lock(m_tests_access);
  if (!testdata) throw Sct::IllegalStateError("Couldnt replace non-existant test",__FILE__,__LINE__);
  shared_ptr<AnalysisWorkerGroup::TestAlgs> talgs = findTest(*testdata);
  if (!talgs) {
    std::cout << "WARNING - adding rather than replacing test!" << std::endl;
    addTest(testdata);
    talgs = findTest(*testdata);
  }
  if (!talgs) throw Sct::IllegalStateError("Couldn't find or add test",__FILE__,__LINE__);
  talgs->replaceTest(testdata);
}

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 ) {
        boost::recursive_mutex::scoped_lock lock((*it)->getMutex());
        shared_ptr<const TestData> td=(*it)->getTest();
    
        if ( runno == td->runNumber ){
      // if full mode, there are three results per scan
#warning "Second number below should be 3!"
      int results_per_scan = (td->options.find("full")==std::string::npos) ? 1 : 3;
      if (scanno >= td->startScanNumber &&
          scanno < td->startScanNumber + td->nScans * results_per_scan) talgs=*it;
    }
    }
    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.
    std::cout << "Clearing tests ... " << endl;
    m_tests.clear();
    std::cout << " ... done clearing tests" << endl;
    this->setPaused(false);    
}

void AnalysisWorkerGroup::work(shared_ptr<IOName> name) throw() {
    using namespace SctData;

    try {
    timer t;
    UniqueID id (name->getUniqueID());
        const unsigned runno = id.getRunNumber();
        const unsigned scanno = id.getScanNumber();
        const string modulename= id.getModule();

    if (debug()) std::cout << "run="<< runno << ", scan="<< scanno << ", modulename="<<modulename << std::endl;

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

    if (debug()) std::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;
    
    boost::recursive_mutex::scoped_lock talgs_lock(talgs->getMutex());
    algorithm =  talgs->findAlgorithm(modulename);

    if (!algorithm.get()) return;
    
        if (debug()) std::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 algorithm_lock (algorithm->getMutex());
        if (isfit) {
      if (debug()) std::cout << "got fit from queue" << endl;
      boost::recursive_mutex::scoped_lock lock(m_tests_access);
      algorithm->addFitScanResult(name);
        } else {
      if (debug()) std::cout << "got raw from queue" << endl;
      boost::recursive_mutex::scoped_lock lock(m_tests_access);
      algorithm->addRawScanResult(name);
        }
    if (debug()) std::cout << "checking if I can analyse algorithm" << endl;
    if (algorithm->canAnalyze()) {
        if (algorithm->isDone()) throw IllegalStateError("AnalysisWorker::work.  Algorithm has additional, unnecessary Scan - analysis already done", __FILE__, __LINE__);
        if (debug()) std::cout << "checking existance" << std::endl;
        algorithm->checkTestResultExists();
        algorithm->addOverheadTime(t.elapsed());
        
        t.restart();
        if (debug()) std::cout << "loading data" << std::endl;
        algorithm->loadData();
        algorithm->addIOTime(t.elapsed());
        
        t.restart();
        if (debug()) std::cout << "analyzing" << std::endl;
        algorithm->analyze();
        algorithm->addAnalysisTime(t.elapsed());
        
        t.restart();
        if (debug()) std::cout << "finishing" << std::endl;
        algorithm->finish();
        algorithm->addIOTime(t.elapsed());
        if (debug()) std::cout << "removing algorithm" << std::endl;
        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);
    }
}


}// end of namespace SctAnalysis

---