source: Sophya/trunk/ArchTOIPipe/Kernel/toi.cc@ 1368

#include "toiprocessor.h"
#include "toi.h"
#include <pthread.h>

#ifdef WITH_SOPHYA
#include "pexceptions.h"
#else
#include "apexceptions.h"
#endif

#define CHKSYNC(ctx)  
//  if (((TOIRegularWindow*)this)->data.size() != ((TOIRegularWindow*)this)->flags.size())        \
//    cout << ctx << ((TOIRegularWindow*)this)->data.size() << "  " <<  \
//      ((TOIRegularWindow*)this)->flags.size() << endl;  \

TOI::TOI() {
  TOIInit();
}

TOI::TOI(string n) {
  name = n;
  TOIInit();
}

void TOI::TOIInit() {
  pthread_mutex_init(&mutex, NULL);
//  pthread_mutex_setname_np(&mutex, (name + "_toi_mutex").c_str(), 0);
  defaultValue = 0;
  producer = NULL;
  dbg = false;
}

TOI::~TOI() {
  pthread_mutex_destroy(&mutex);
}

void TOI::setProducer(TOIProcessor* p) {
  if (producer) 
    throw DuplicateIdExc("TOI::setProducer : producer already defined");
  producer = p;
}

void TOI::addConsumer(TOIProcessor* p) {
  consumers.push_back(p);
}

int TOI::getMinSn(){
  return producer->getMinOut();
}

int TOI::getMaxSn(){
  return producer->getMaxOut();
}

#ifndef NO_SOPHYA
Array TOI::getError(int iStart, int iEnd) {
  if (errorTOI == NULL) throw NotFoundExc("TOI::getDataError : no Error TOI");
  return errorTOI->getData(iStart, iEnd);
}
#endif

double TOI::getError(int i) {
  if (errorTOI == NULL) throw NotFoundExc("TOI::getDataError : no Error TOI");
  return errorTOI->getData(i);
}

void TOI::putDataError(int i, double value, double error, int_4 flag) {
  if (errorTOI == NULL) throw NotFoundExc("TOI::getDataError : no Error TOI");
  putData(i, value, flag);
  errorTOI->putData(i, value, flag);
}

#ifndef NO_SOPHYA
Array TOI::getData(int iStart, int iEnd) {
  lock();
  Array a = doGetData(iStart, iEnd);
  unlock();
  return a;
}
#endif

double TOI::getData(int i) {
  lock();
  double dat = doGetData(i);
  unlock();
  return dat;
}


#ifndef NO_SOPHYA
TArray<int_4> TOI::getFlag(int iStart, int iEnd) {
  lock();
  TArray<int_4> a = doGetFlag(iStart, iEnd);
  unlock();
  return a;
}
#endif

int_4 TOI::getFlag(int i) {
  lock();
  int_4 f = doGetFlag(i);
  unlock();
  return f;
}
  
  

void TOI::putData(int i, double value, int_4 flag) {
  lock();
  doPutData(i, value, flag);
  unlock();
}

void TOI::waitForData(int iStart, int iEnd) {
  if (producer == NULL) throw NotFoundExc("TOI has no producer !");
    
  DataStatus s = isDataAvail(iStart, iEnd);
  if (s == DATA_OK) {
      return;
  }
  if (s == DATA_DELETED) {
      throw NotFoundExc("Data has been purged !");
  }

  producer->lock();
  while (isDataAvailNL(iStart, iEnd) == DATA_NOT_YET) {
    producer->wait();
  }
  producer->unlock();
  return;
}

void TOI::waitForData(int i) {
  waitForData(i,i);
}

void TOI::waitForAnyData() {
  if (! hasSomeData()) {
    producer->lock();
    producer->wait();
    producer->unlock();
  }
}

TOI::DataStatus TOI::isDataAvail(int i) {
  lock();
  DataStatus stat = isDataAvailNL(i);
  unlock();
  return stat;
}

TOI::DataStatus TOI::isDataAvail(int i, int j) {
  lock();
  DataStatus stat = isDataAvailNL(i,j);
  unlock();
  return stat;
}

TOI::DataStatus TOI::isDataAvailNL(int i) {
  return isDataAvailNL(i,i);
}

void TOI::wontNeedBefore(int i) {
  int j=i;
  for (vector<TOIProcessor*>::iterator k = consumers.begin();
       k != consumers.end(); k++) {
    if ((*k)->wontNeedValue < j) j = (*k)->wontNeedValue;
  }
  lock();
  doWontNeedBefore(j);
  unlock();
}

void TOI::doWontNeedBefore(int i) {
}


TOIRegularWindow::TOIRegularWindow() {
  i0 = -1;
}

TOIRegularWindow::TOIRegularWindow(string nm) {
  i0 = -1;
  setName(nm);
}

TOIRegularWindow::~TOIRegularWindow() {
}

TOI::DataStatus TOIRegularWindow::isDataAvailNL(int iStart, int iEnd) {
  if (iEnd >= i0 + (long)data.size()) return DATA_NOT_YET;
  if (iStart < i0) return DATA_DELETED;
  return DATA_OK;
}

TOI::DataStatus TOIRegularWindow::isDataAvailNL(int i) {
  return TOI::isDataAvailNL(i);
}

void TOIRegularWindow::doWontNeedBefore(int i) {
  if (i>= i0 + (long)data.size())
    i = i0 + (long)data.size() - 1;
  if (i>i0) {  // don't empty list
    int osz = data.size();
    data.erase(data.begin(), data.begin()+(i-i0));
    flags.erase(flags.begin(), flags.begin()+(i-i0));
    i0 = i;
  }
}


#ifndef NO_SOPHYA
Array TOIRegularWindow::doGetData(int iStart, int iEnd) {
  if (!isDataAvailNL(iStart, iEnd)) throw RangeCheckError("TOI::getData : data not available");
  Array dat(iEnd - iStart + 1);
  long j0 = iStart - i0;
  for (int i=0; i<iEnd-iStart+1; i++) {
    dat[i] = data[i+j0];
  }
  return dat;
}
#endif

double TOIRegularWindow::doGetData(int i) {
  if (isDataAvailNL(i) != DATA_OK) throw RangeCheckError("TOI::getData : data not available");
  double dat = data[i - i0];
  return dat;
}


#ifndef NO_SOPHYA
TArray<int_4> TOIRegularWindow::doGetFlag(int iStart, int iEnd) {
  if (isDataAvailNL(iStart, iEnd) != DATA_OK) throw RangeCheckError("TOI::getData : data not available");
  TArray<int_4> dat(iEnd - iStart + 1);
  long j0 = iStart - i0;
  for (int i=0; i<iEnd-iStart+1; i++) {
    dat[i] = flags[i+j0];
  }
  return dat;
}
#endif

int_4 TOIRegularWindow::doGetFlag(int i) {
  if (!isDataAvailNL(i)) throw RangeCheckError("TOI::getData : data not available");
  return flags[i - i0]; 
}


void TOIRegularWindow::doPutData(int i, double value, int_4 flag) {
  if (i0 == -1) {
    data.insert(data.begin(), 1,  defaultValue);
    flags.insert(flags.begin(), 1,  0);
    i0 = i;
  } else if (i<i0) {
    data.insert(data.begin(), i0-i, defaultValue); 
    flags.insert(flags.begin(), i0-i, 0);
    i0 = i;
  } else if (i>=i0+(int)data.size()) {
    data.insert(data.end(), (long)  (i-(i0+data.size())+1), defaultValue);
    flags.insert(flags.end(), (long) (i-(i0+flags.size())+1), 0);
  }
  data[i-i0] = value;
  flags[i-i0] = flag;  
}

bool TOIRegularWindow::hasSomeData() {
  lock();
  bool x = !data.empty();
  unlock();
  return x;
}

int TOIRegularWindow::nextDataAvail(int iAfter) {
  lock();
  if (iAfter >= i0 + (long)data.size()) {unlock(); return -1;}
  if (iAfter < i0) {unlock(); return i0;}
  unlock();
  return iAfter+1;
}

/*  A faire, le nettoyage (heuristique selon demandes ?, guide ? ) */