AlgThread.cxx

Go to the documentation of this file.

// $Id: AlgThread.cxx,v 1.2 2008/01/31 22:18:17 rustem Exp $

// C++
#include <iostream>
#include <iterator>

// Local
#include "AlgThread.h"

using namespace std;

//---------------------------------------------------------------------------------------
void* Anp::run_alg_thread(void *data_)
{
   if(!data_)
   {
      cerr << "Anp::run_alg_thread - null data pointer" << endl;
      return 0;
   }

   AlgThread *data = static_cast<AlgThread *> (data_);

   data -> Run();

   return data;
}

//---------------------------------------------------------------------------------------
Anp::AlgThread::AlgThread(AlgSnarlPtr alg, const string &name)
   :fMutex(),
    fThread(),
    fThreadStatus(1),
    fAlg(alg),
    fName(name),
    fRecord(),
    fNPass(0),
    fNFail(0)
{
}

//---------------------------------------------------------------------------------------
Anp::AlgThread::~AlgThread()
{
}

//----------------------------------------------------------------------
void Anp::AlgThread::Run(Record &record, const bool new_thread)
{   
   
   if(new_thread)
   {
      //
      // Lock mutex ONLY when starting new thread
      //
      Anp::Lock<Mutex> lock(fMutex);

      //
      // Make local copy of input record
      //
      fRecord = record;

      //
      // Run kNN algorithm in this thread in NEW thread
      //
      fThreadStatus = pthread_create(&fThread, NULL, Anp::run_alg_thread, this);
      
      if(fThreadStatus != 0)
      {
         cerr << "AlgThread::Run - invalid thread status: " << fThreadStatus << endl;
      }
   }
   else
   {
      //
      // No thread is created, set thread status to 1,
      // so that we do not try later to join same thread
      //
      fThreadStatus = 1;

      //
      // Make local copy of input record
      //
      fRecord = record;
      
      //
      // Run kNN algorithm in this thread
      //
      Run();
   }
}

//----------------------------------------------------------------------
int Anp::AlgThread::Join()
{
   //
   // If there is parallel thread running then attempt to join to main thread
   //
   if(fThreadStatus == 0)
   {
      return pthread_join(fThread, NULL);
   }

   return fThreadStatus;
}

//------------------------------------------------------------------------------------------
bool Anp::AlgThread::Init(const Header &header)
{
   //
   // Initiliaze FillkNN::AlgSnarl algorithm
   //
   
   if(!fAlg.valid())
   {
      cerr << "AlgThread::Init - invalid AlgSnarl handle" << endl;
      return false;
   }

   return fAlg -> Init(header);
}

//---------------------------------------------------------------------------------------
void Anp::AlgThread::End(const DataBlock &block)
{
   //
   // Send signal to algirithm that we are done here...
   //
   if(fAlg.valid()) fAlg -> End(block);
}

//---------------------------------------------------------------------------------------
void Anp::AlgThread::Run()
{
   //
   // This function _might_ be running in new thread and will 
   // modify internal class data so put lock on current thread.
   //
   Anp::Lock<Mutex> lock(fMutex);

   if(!fAlg.valid())
   {
      cerr << "AlgThread::Run - invalid AlgSnarl handle" << endl;
      return;
   }

   //
   // Store all already existing event and track keys
   //
   map<short, Anp::PrevDataKey> emap, tmap;

   for(EventIterator it = fRecord.EventBegIterator(); it != fRecord.EventEndIterator(); ++it)
   {
      emap[it -> EventIndex()] = Anp::PrevDataKey(it -> DataBeg(), it -> DataEnd());
   }
   for(TrackIterator it = fRecord.TrackBegIterator(); it != fRecord.TrackEndIterator(); ++it)
   {
      tmap[it -> TrackIndex()] = Anp::PrevDataKey(it -> DataBeg(), it -> DataEnd());
   }

   //
   // Run algorithm...
   //
   fAlg -> Run(fRecord);

   //
   // Erase previous data with keys stored earlier
   //
   for(EventIterator it = fRecord.EventBegIterator(); it != fRecord.EventEndIterator(); ++it)
   {
      it -> Erase(std::remove_if(it -> DataBegIterator(),
                                 it -> DataEndIterator(),
                                 emap[it -> EventIndex()]), it -> DataEndIterator());
   }

   //
   // Erase previous data with keys stored earlier
   //
   for(TrackIterator it = fRecord.TrackBegIterator(); it != fRecord.TrackEndIterator(); ++it)
   {
      it -> Erase(std::remove_if(it -> DataBegIterator(),
                                 it -> DataEndIterator(),
                                 tmap[it -> TrackIndex()]), it -> DataEndIterator());
   }
}

//---------------------------------------------------------------------------------------
Anp::PrevDataKey::PrevDataKey()
   :dvec(),
    valid(false)
{
}

//---------------------------------------------------------------------------------------
Anp::PrevDataKey::PrevDataKey(DataIter ibeg, DataIter iend)
   :dvec(),
    valid(true)
{
   //
   // Save input elements: key and data pairs
   //
   dvec.insert(dvec.end(), ibeg, iend);

   //
   // Sort keys
   //
   std::sort(dvec.begin(), dvec.end());
}

//---------------------------------------------------------------------------------------
Anp::PrevDataKey::PrevDataKey(const vector<short> &kvec_)
   :dvec(),
    valid(true)
{
   for(unsigned int i = 0; i < kvec_.size(); ++i)
   {
      dvec.push_back(Anp::Data(kvec_[i], 0.0));
   }

   //
   // Sort vector
   //
   std::sort(dvec.begin(), dvec.end());
}

//---------------------------------------------------------------------------------------
bool Anp::PrevDataKey::operator()(const Data &data) const
{
   if(!valid)
   {
      cerr << "PrevDataKey::operator() - invalid internal state" << endl;
      return false;
   }

   //
   // return true if this key is found
   //
   return std::binary_search(dvec.begin(), dvec.end(), data);
}

//---------------------------------------------------------------------------------------
bool Anp::PrevDataKey::operator() (Event &event) const
{
   if(!valid)
   {
      cerr << "PrevDataKey::operator() - invalid internal state" << endl;
      return false;
   }

   //
   // Add current elements to Event
   //
   for(DataVec::const_iterator dit = dvec.begin(); dit != dvec.end(); ++dit)
   {
      event.Add(dit -> Key(), dit -> Data());
   }

   return true;
}

//---------------------------------------------------------------------------------------
bool Anp::PrevDataKey::operator() (Track &track) const
{
   if(!valid)
   {
      cerr << "PrevDataKey::operator() - invalid internal state" << endl;
      return false;
   }

   //
   // Add current elements to Track
   //
   for(DataVec::const_iterator dit = dvec.begin(); dit != dvec.end(); ++dit)
   {
      track.Add(dit -> Key(), dit -> Data());
   }

   return true;
}

---