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

// ArchTOIPipe           (C)     CEA/DAPNIA/SPP IN2P3/LAL
//                               Eric Aubourg
//                               Christophe Magneville
//                               Reza Ansari
// $Id: toiseqbuff.cc,v 1.16 2003-11-14 12:34:55 aubourg Exp $

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

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


TOISeqBuffered::TOISeqBuffered(int wsz) {
  data = NULL;
  flags = NULL;
  AllocBuffer(wsz);
  setName("toiseqbuff");
  syncOldWay = false;
}

TOISeqBuffered::TOISeqBuffered(string nm, int wsz) {
  data = NULL;
  flags = NULL;
  AllocBuffer(wsz);
  setName(nm);
  syncOldWay = false;
}

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

void TOISeqBuffered::AllocBuffer(int wsz) 
{
  if (wsz < 128) wsz = 128;
  wsize = wsz;
  buffsize = 2*wsz;
  if (data)   delete[] data;
  if (flags)  delete[] flags;
  data = new double[buffsize];
  flags = new uint_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() 
     << "\n  WindowSize= " << wsize << " BufferSize= " << buffsize << 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(long iStart, long 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(long i) {
  return TOI::isDataAvailNL(i);
}

void TOISeqBuffered::wontNeedBefore(long i) {
  // $CHECK$  Reza 30/4/2001 - Je ne sais pas a quoi ca sert !
  //  next_out = i; $CHECK$  Reza 30/4/2001
}


#ifndef NO_SOPHYA
/* ---- l'interface va etre modifiee, NE PAS UTILISER
Array TOISeqBuffered::doGetData(long iStart, long 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 (long 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

double TOISeqBuffered::getData(long i)
{
  double val; 
  uint_8 flg;
  getData(i, val, flg);
  return(val);
}


void TOISeqBuffered::getData(long i, double & val, uint_8 & flg) {
  lock();
  if (!started) { 
    cout << " TOISeqBuffered::getData() - waitGet() Waiting for start ... " << endl;
    waitGet(); 
  }
  cleanWaitGet();
  if (isDataDeleted(i)) {
    if (dbglev > 0) 
      cout << " TOISeqBuffered::getData() - DataDeleted() name=" << getName()  
           << " i=" << i << " next_in= " << next_in 
           << " next_out=" << next_out << endl;
    unlock();
    throw RangeCheckError("TOISeqBuffered::getData(i) : data deleted");
  }
  while (i >= next_in) { 
    if (i>next_out) next_out = i; 
    if (dbglev > 0) 
      cout << " TOISeqBuffered::getData() - 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::getData() - signalPut() name=" << getName()  
           << " i=" << i << " next_in= " << next_in 
           << " next_out=" << next_out << endl;
    signalPut(); 
  }
  if (fgsigput)  broadcast();
  unlock();
  return;
}


void TOISeqBuffered::getData(long i, int n, double* data, uint_8* flg)
{
  lock();
  if (!started) { 
    cout << " TOISeqBuffered::getData(i,n ...) - waitGet() Waiting for start ... " << endl;
    waitGet(); 
  }
  cleanWaitGet();
  if (isDataDeleted(i)) {
    if (dbglev > 0) 
      cout << " TOISeqBuffered::getData(i,n ...) - DataDeleted() name=" << getName()  
           << " i=" << i << " next_in= " << next_in 
           << " next_out=" << next_out << endl;
    unlock();
    throw RangeCheckError("TOISeqBuffered::getData(i) : data deleted");
  }
  for(long j=i; j<i+n; j++) {
    while (j >= next_in) { 
      if (j>next_out) next_out = j; 
      if (dbglev > 0) 
        cout << " TOISeqBuffered::getData(i,n ...) - waitGet() name=" << getName()  
             << " j=" << j << " next_in= " << next_in 
             << " next_out=" << next_out << endl;
      waitGet(); 
      if (dbglev > 0) 
        cout << " ... Out of waitGet() j=" << j
             << " next_in= " << next_in << " next_out=" << next_out << endl;
      cleanWaitGet();
    }
    data[j-i] = dataRef(j);
    if (flg) flg[j-i] = flagRef(j);
    if (first_out < 0)  first_out = j;
    if ((j+1) > next_out)  next_out = j+1;
    if (isPutWaiting() && (next_in-next_out < wsize/2 )) { 
      if (dbglev > 0) 
        cout << " TOISeqBuffered::getData(i,n ...) - signalPut() name=" << getName()  
             << " i=" << i << " next_in= " << next_in 
             << " next_out=" << next_out << endl;
      //      signalPut(); 
      broadcast();
    }
  }
  unlock();
  //  if (fgsigput)  signal();
  return;  
}


#ifndef NO_SOPHYA
/* ---- l'interface va etre modifiee, NE PAS UTILISER
TArray<int_4> TOISeqBuffered::doGetFlag(long iStart, long 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 (long 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(long 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::putData(long i, double value, uint_8 flag) {
  lock();
  if (!started) {
    first_in = next_in = i;
    next_out = next_in;
    started = true;
  }
  else {
    if (i != next_in) {
      if (dbglev > 0) 
        cout << " TOISeqBuffered::putData() - i!=next_in() name=" << getName()  
             << " i=" << i << " next_in= " << next_in 
             << " next_out=" << next_out << endl;
      string msg = "TOISeqBuffered::putData() : i!=next_in TOIname="  + getName();
      unlock();
      throw RangeCheckError(msg);
    }
    if (next_in-next_out >= wsize) { 
      if (dbglev > 0) 
        cout << " TOISeqBuffered::putData() - 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::putData() - signalGet() name=" << getName()
           << " i=" << i << " next_in= " << next_in 
           << " next_out=" << next_out << endl;
    broadcast();
    //    signalGet(); 
  }
  //  if (fgsigget)  broadcast();
  unlock();
  return;
}

void TOISeqBuffered::putData(long i, int n, double const* val, uint_8 const* flg) 
{
  lock();
  if (!started) {
    first_in = next_in = i;
    next_out = next_in;
    started = true;
  }
  else {
    if (i != next_in) {
      if (dbglev > 0) 
        cout << " TOISeqBuffered::putData(i,n ...) - i!=next_in() name=" << getName()  
             << " i=" << i << " next_in= " << next_in 
             << " next_out=" << next_out << endl;
      string msg = "TOISeqBuffered::putData() : i!=next_in TOIname="  + getName();
      unlock();
      throw RangeCheckError(msg);
    }
  }
  for(long j=i; j<i+n; j++) {
    if (next_in-next_out >= wsize) { 
      if (dbglev > 0) 
        cout << " TOISeqBuffered::putData(i,n ...) - waitPut() " << getName() 
             << " j=" << j 
             << " next_in= " << next_in << " next_out=" << next_out << endl;
      broadcast();
      waitPut();
      if (dbglev > 0) 
        cout << " ... Out of waitPut() j=" << j
             << " next_in= " << next_in << " next_out=" << next_out << endl;
      cleanWaitPut();
    }
    dataRef(j) = val[j-i];
    if (flg) flagRef(j) = flg[j-i];
    else flagRef(j) = 0;
    next_in = j+1;
    if (isGetWaiting() && (next_in-next_out > wsize/8))  { 
      if (dbglev > 0) 
        cout << " TOISeqBuffered::putData(i,n ...) - signalGet() name=" << getName()
             << " i=" << i << " next_in= " << next_in 
             << " next_out=" << next_out << endl;
      // signalGet(); 
      broadcast();
    }
  }
  /*
  if (isGetWaiting() && (next_in-next_out > wsize/8))  { 
    if (dbglev > 0) 
      cout << " TOISeqBuffered::putData(i,n ...) - signalGet() name=" << getName()
           << " i=" << i << " next_in= " << next_in 
           << " next_out=" << next_out << endl;
    //signalGet(); 
    broadcast();
  }
  */
  //  if (fgsigget)  broadcast();
  unlock();
  return;
}

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

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

void TOISeqBuffered::doGetData(long i, double & val, uint_8 & flg)
{
  cerr << " TOISeqBuffered::doGetData() not implemented !"
       << " \n A quoi ca set ??? Reza - Mai 2002 " << endl;
  throw NotAvailableOperation("TOISeqBuffered::doGetData() not implemented !");
}

void TOISeqBuffered::doPutData(long i, double value, uint_8 flag)
{
  cerr << " TOISeqBuffered::doGetData() not implemented !"
       << " \n A quoi ca set ??? Reza - Mai 2002 " << endl;
  throw NotAvailableOperation("TOISeqBuffered::doGetData() not implemented !");
}