source: Sophya/trunk/ArchTOIPipe/Kernel/toiseqbuff.cc@ 1480

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

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


TOISeqBuffered::TOISeqBuffered(int wsz) {
  AllocBuffer(wsz);
}

TOISeqBuffered::TOISeqBuffered(string nm, int wsz) {
  AllocBuffer(wsz);
  setName(nm);
}

TOISeqBuffered::~TOISeqBuffered() {
  delete[] data;
  delete[] flags;
}

void TOISeqBuffered::AllocBuffer(int wsz) 
{
  if (wsz < 128) wsz = 128;
  wsize = wsz;
  buffsize = 2*wsz;
  data = new double[buffsize];
  flags = new int_8[buffsize];
  for(int k=0; k<buffsize; k++) {
    data[k] = defaultValue;
    flags[k] = 0;
  }
  next_in = next_out = -1;
  first_in = first_out = -1;
  started = false;
  dbglev = 0;
}

void TOISeqBuffered::PrintStatus(ostream & os) const
{
  os << "---TOISeqBuffered::PrintStatus() - Name=" << getName() << endl;
  os << "Index: FirstIn= " << getFirstIn() << " LastIn= " << getLastIn()
     << "  Total= " << getLastIn()-getFirstIn()+1 << endl;
  os << "Index: FirstOut= " << getFirstOut() << " LastOut= " << getLastOut()
     << "  Total= " << getLastOut()-getFirstOut()+1 << endl;
  os << " WaitStatus: Put/" ;
  if (isPutWaiting()) os << "Waiting " ;
  else os << "Running ";
  os << " PutCountWait= " << getCountWaitPut() << endl;
  os << " WaitStatus: Get/" ;
  if (isGetWaiting()) os << "Waiting " ;
  else os << "Running ";
  os << " GetCountWait= " << getCountWaitGet() << endl;
}

TOI::DataStatus TOISeqBuffered::isDataAvailNL(int iStart, int iEnd) {
  if (iEnd < iStart) 
    throw RangeCheckError("TOISeqBuffered::isDataAvailNL : iEnd<iStart !");
  if (!started) return DATA_NOT_YET;
  if (iEnd >= next_in) return DATA_NOT_YET;
  if (isDataDeleted(iStart)) return DATA_DELETED;
  return DATA_OK;
}

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

void TOISeqBuffered::doWontNeedBefore(int i) {
  next_out = i;
}


#ifndef NO_SOPHYA
/* ---- l'interface va etre modifiee, NE PAS UTILISER
Array TOISeqBuffered::doGetData(int iStart, int iEnd) {
  //  if (iEnd < iStart) 
  //    throw RangeCheckError("TOI::getData : iEnd<iStart !");
  //  if (iStart <= out_last)
  if (!started)   waitGet(); 
  if (!isDataAvailNL(iStart, iEnd)) 
    throw RangeCheckError("TOISeqBuffered::getData(iS,iE) : data not available");
  cleanWaitGet();
  Vector dat(iEnd - iStart + 1);
  for (int i=0; i<iEnd-iStart+1; i++) {
    dat(i) = dataRef(i+iStart);
  }
  if (first_out < 0)  first_out = iStart;
  if ((iEnd+1) > next_out)  next_out = iEnd+1;
  if (isPutWaiting() && (next_in-next_out < wsize/2 )) signalPut();
  return dat;
}
  l'interface va etre modifiee, NE PAS UTILISER ---- */
#endif

void TOISeqBuffered::doGetData(int i, double & val, int_8 & flg) {
  if (!started) { 
    cout << " TOISeqBuffered::doGetData() - waitGet() Waiting for start ... " << endl;
    waitGet(); 
  }
  cleanWaitGet();
  if (isDataDeleted(i)) {
    if (dbglev > 0) 
      cout << " TOISeqBuffered::doGetData() - DataDeleted() name=" << getName()  
           << " i=" << i << " next_in= " << next_in 
           << " next_out=" << next_out << endl;
    throw RangeCheckError("TOISeqBuffered::doGetData(i) : data deleted");
  }
  while (i >= next_in) { 
    if (i>next_out) next_out = i; 
    if (dbglev > 0) 
      cout << " TOISeqBuffered::doGetData() - waitGet() name=" << getName()  
           << " i=" << i << " next_in= " << next_in 
           << " next_out=" << next_out << endl;
    waitGet(); 
    if (dbglev > 0) 
      cout << " ... Out of waitGet() i=" << i
           << " next_in= " << next_in << " next_out=" << next_out << endl;
  }
  cleanWaitGet();
  val = dataRef(i);
  flg = flagRef(i);
  if (first_out < 0)  first_out = i;
  if ((i+1) > next_out)  next_out = i+1;
  if (isPutWaiting() && (next_in-next_out < wsize/2 )) { 
    if (dbglev > 0) 
      cout << " TOISeqBuffered::doGetData() - signalPut() name=" << getName()  
           << " i=" << i << " next_in= " << next_in 
           << " next_out=" << next_out << endl;
    signalPut();
  }
  return;
}


#ifndef NO_SOPHYA
/* ---- l'interface va etre modifiee, NE PAS UTILISER
TArray<int_4> TOISeqBuffered::doGetFlag(int iStart, int iEnd) {
  if (!started) waitGet(); 
  cleanWaitGet();
  if (!isDataAvailNL(iStart, iEnd)) 
    throw RangeCheckError("TOISeqBuffered::getFlag(iS,iE) : data not available");
  TVector<int_4> dat(iEnd - iStart + 1);
  for (int i=0; i<iEnd-iStart+1; i++) {
    dat[i] = flagRef(i+iStart);
  }
  return dat;
}
  l'interface va etre modifiee, NE PAS UTILISER ---- */
#endif

/*RZCMV
int_4 TOISeqBuffered::doGetFlag(int i) {
  if (!started) waitGet(); 
  cleanWaitGet();
  if (isDataDeleted(i))
    throw RangeCheckError("TOISeqBuffered::doGetFlag(i) : data deleted");
  while (i >= next_in) waitGet();
  int_4 dat = flagRef(i);
  return dat;
}
*/


void TOISeqBuffered::doPutData(int i, double value, int_8 flag) {
  if (!started) {
    first_in = next_in = i;
    next_out = next_in;
    started = true;
  }
  else {
    if (i != next_in) {
      if (dbglev > 0) 
        cout << " TOISeqBuffered::doPutData() - i!=next_in() name=" << getName()  
             << " i=" << i << " next_in= " << next_in 
             << " next_out=" << next_out << endl;
      string msg = "TOISeqBuffered::doPutData() : i!=next_in TOIname="  + getName();
      throw RangeCheckError(msg);
    }
    if (next_in-next_out >= wsize) { 
      if (dbglev > 0) 
        cout << " TOISeqBuffered::doPutData() - waitPut() " << getName() 
             << " i=" << i 
             << " next_in= " << next_in << " next_out=" << next_out << endl;
      waitPut();
      if (dbglev > 0) 
        cout << " ... Out of waitPut() i=" << i
             << " next_in= " << next_in << " next_out=" << next_out << endl;
    }
    cleanWaitPut();
  }
  dataRef(i) = value;
  flagRef(i) = flag;
  next_in = i+1;
  if (isGetWaiting() && (next_in-next_out > wsize/8))  { 
    if (dbglev > 0) 
      cout << " TOISeqBuffered::doPutData() - signalGet() name=" << getName()
           << " i=" << i << " next_in= " << next_in 
           << " next_out=" << next_out << endl;
    signalGet();
  }
}

bool TOISeqBuffered::hasSomeData() {
  lock();
  bool x =  started;
  unlock();
  return x;
}

int TOISeqBuffered::nextDataAvail(int iAfter) {
  lock();
  if (iAfter >= next_in ) {unlock(); return -1;}
  if (iAfter < (next_in-buffsize)) {unlock(); return (next_in-buffsize);}
  unlock();
  return iAfter+1;
}