source: Sophya/trunk/ArchTOIPipe/Kernel/toiprocessor.cc@ 1437

#include "toiprocessor.h"
#include "toimanager.h"
#include <pthread.h>
#include <values.h>

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

#define pthread_mutexattr_setkind_np pthread_mutexattr_settype

TOIProcessor::TOIProcessor() {
  //cout << "TOIProcessor::TOIProcessor" << endl;
  outTOIs = NULL;
  inTOIs = NULL;
  inited=false;

  upExtra  = 0;
  lowExtra = 0;
  minOut = -1;
  maxOut = -1;
  neededHistory = 1000;
  lastAWN = 0;
  wontNeedValue = -1;

}

TOIProcessor::~TOIProcessor() {
  delete[] outTOIs;
  delete[] inTOIs;
  //if (mutex) 
  pthread_mutex_destroy(&mutex);
  //if (dataReady) 
  pthread_cond_destroy(&dataReady);
  pthread_detach(thread);
}


void TOIProcessor::init() {
  //cout << "TOIProcessor::init" << endl;
}

void TOIProcessor::afterinit() {
  int i;
  inTOIs = new (TOI*[inIx.size()]);
  for(i=0; i<inIx.size(); i++) 
    inTOIs[i] = NULL;
  outTOIs = new (TOI*[outIx.size()]);
  for(i=0; i<outIx.size(); i++) 
    outTOIs[i] = NULL;
}

int TOIProcessor::getMinOut() {
  //cout << name << "minout" << endl;
  if (minOut < 0) minOut = calcMinOut();
  //cout << name << "minout=" << minOut << endl;
  return minOut;
}

int TOIProcessor::getMaxOut() {
  //cout << name << "maxout" << endl;
  if (maxOut < 0) maxOut = calcMaxOut();
  //cout << name << "maxout=" << maxOut << endl;
  return maxOut;
}

int TOIProcessor::calcMinOut() {
  return getMinIn() + lowExtra;
}

int TOIProcessor::calcMaxOut() {
  return getMaxIn() - upExtra;
}

int TOIProcessor::getMinIn() {
  int nIn = inIx.size();
  int minIn = 0;
  for (int i=0; i<nIn; i++) {
    TOI* toi = inTOIs[i];
    int x = toi->getMinSn();
    if (x > minIn) minIn = x;
  }
  return minIn;
}

int TOIProcessor::getMaxIn() {
  int nIn = inIx.size();
  int maxIn = MAXINT;
  for (int i=0; i<nIn; i++) {
    TOI* toi = inTOIs[i];
    int x = toi->getMaxSn();
    if (x < maxIn) maxIn = x;
  }
  return maxIn;
}


int TOIProcessor::declareInput(string toi) {
  if (inIx.find(toi) != inIx.end())
    throw DuplicateIdExc("TOIProcessor::declareInput : "+toi+" already declared");
  int i = inIx.size();
  inIx[toi] = i;
  return i;
}

int  TOIProcessor::declareOutput(string toi) {
  if (outIx.find(toi) != outIx.end())
    throw DuplicateIdExc("TOIProcessor::declareInput : "+toi+" already declared");
  int i = outIx.size();
  outIx[toi] = i;
  return i;
}

int TOIProcessor::getInputTOIIndex(string toi) {
  chkinit();
  map<string, int>::iterator i = inIx.find(toi);
  if (i == inIx.end()) return -1;
  return (*i).second;
}

int TOIProcessor::getOutputTOIIndex(string toi) {
  chkinit();
  map<string, int>::iterator i = outIx.find(toi);
  if (i == outIx.end()) return -1;
  return (*i).second;
}

//  Methodes rajoutees par Reza 11/3/2001
TOI* TOIProcessor::getInputTOI(int toiIndex) {
  //  chkinit();
  if (toiIndex >= inIx.size()) 
    throw RangeCheckError("TOIProcessor::getInputTOI() out of bound toiIndex");
  TOI* toi = inTOIs[toiIndex];
  if (toi == NULL) 
    throw NullPtrError("TOIProcessor::getInputTOI() - Not assigned TOI !");
  return(toi);
}

TOI* TOIProcessor::getOutputTOI(int toiIndex) {
  //  chkinit();
  if (toiIndex >= outIx.size()) 
    throw RangeCheckError("TOIProcessor::getOutputTOI() out of bound toiIndex");
  TOI* toi = outTOIs[toiIndex];
  if (toi == NULL) 
    throw NullPtrError("TOIProcessor::getOutputTOI() - Not assigned TOI !");
  return(toi);
}

bool TOIProcessor::checkInputTOIIndex(int toiIndex) {
  if (toiIndex >= inIx.size()) return false;
  if (inTOIs[toiIndex] == NULL) return false;
  return true;
}

bool TOIProcessor::checkOutputTOIIndex(int toiIndex) {
  if (toiIndex >= outIx.size()) return false;
  if (outTOIs[toiIndex] == NULL) return false;
  return true;
}

void TOIProcessor::PrintStatus(ostream & os) 
{
  os << " TOIProcessor::PrintStatus() - Name= " << name 
     << " MinIn=" << getMinIn() << " MaxIn=" << getMaxIn() << endl;
  os << " --- Inputs N= " << inIx.size() << endl;
  int k;
  for(k=0; k<inIx.size(); k++) {
    os << "Input[" << k << "] : " <<  getInName(k) ;
    if (inTOIs[k] != NULL) 
      os << " Connected TOI " << inTOIs[k]->getName() << endl;
    else os << " NO TOI " << endl;
  }
  os << " --- Outputs N= " << outIx.size() << endl;
  for(k=0; k<outIx.size(); k++) {
    os << "Output[" << k << "] : " <<  getOutName(k) ;
    if (outTOIs[k] != NULL) 
      os << " Connected TOI " << outTOIs[k]->getName() << endl;
    else os << " NO TOI " << endl;
  }
  os << endl;
  return;
}

//  Fin rajout Reza 11/3/2001 

void TOIProcessor::addInput(string name, TOI* toi) {
  chkinit();
  map<string, int>::iterator i = inIx.find(name);
  if (i == inIx.end()) throw NotFoundExc("TOIProcessor::addInput "+
                                         name+" not declared");
  inTOIs[(*i).second] = toi; 
}

void TOIProcessor::addOutput(string name, TOI* toi) {
  chkinit();
  map<string, int>::iterator i = outIx.find(name);
  if (i == outIx.end()) throw NotFoundExc("TOIProcessor::addOutput "+
                                          name+" not declared");
  toi->setProducer(this);
  outTOIs[(*i).second] = toi; 
}

string TOIProcessor::getOutName(int i) {
  if (i > outIx.size()) throw RangeCheckError("TOIProcessor::getOutName "
                                             " out of bound");
  map<string, int>::iterator j;
  for(j=outIx.begin(); j!= outIx.end(); j++) 
    if ((*j).second == i)  return (*j).first;

  throw RangeCheckError("TOIProcessor::getOutName  Not found index !");
}

string TOIProcessor::getInName(int i) {
  if (i > inIx.size()) throw RangeCheckError("TOIProcessor::getInName "
                                             " out of bound");
  map<string, int>::iterator j;
  for(j=inIx.begin(); j!= inIx.end(); j++) 
    if ((*j).second == i)  return (*j).first;

  throw RangeCheckError("TOIProcessor::getOutName  Not found index !");
}

void TOIProcessor::run() {
  
}

void* TOIProcessor::ThreadStart(void* arg) {
  TOIProcessor* p = (TOIProcessor*) arg;
  //  cout << p->name << " new thread running " << pthread_self() << endl;
  p->run();
  //  cout << p->name << " thread done " << pthread_self() << endl;
  return NULL;
}

#ifdef Linux
#define pthread_mutexattr_settype  pthread_mutexattr_setkind_np
#define  PTHREAD_MUTEX_ERRORCHECK PTHREAD_MUTEX_ERRORCHECK_NP
#define pthread_mutex_setname_np(a,b,c)
#define pthread_cond_setname_np(a,b,c)
#endif

void TOIProcessor::start() {
  pthread_cond_init(&dataReady, NULL);
  pthread_mutexattr_init(&mutattr);
  //  pthread_mutexattr_settype(&mutattr, PTHREAD_MUTEX_ERRORCHECK);
  pthread_mutex_init(&mutex, &mutattr);
  //pthread_mutex_setname_np(&mutex, (name + "_proc_mutex").c_str(), 0);
  //pthread_cond_setname_np(&dataReady, (name + "_proc_cond").c_str(), 0);
  //cout << name << " starting thread " << &thread << endl;
  pthread_create(&thread, NULL, ThreadStart, this);
  TOIManager::getManager()->addThread(&thread);
}

#ifndef NO_SOPHYA
Array TOIProcessor::getData(int toiIndex, int iStart, int iEnd) {
  TOI* toi = getInputTOI(toiIndex);
  toi->waitForData(iStart, iEnd);
  return toi->getData(iStart, iEnd);
}

Array TOIProcessor::getError(int toiIndex, int iStart, int iEnd) {
  TOI* toi = getInputTOI(toiIndex);
  toi->waitForData(iStart, iEnd);
  return toi->getError(iStart, iEnd);
}

TArray<int_4> TOIProcessor::getFlag(int toiIndex, int iStart, int iEnd) {
  TOI* toi = getInputTOI(toiIndex);
  toi->waitForData(iStart, iEnd);
  return toi->getFlag(iStart, iEnd);
}
#endif

double TOIProcessor::getData(int toiIndex, int i) {
  TOI* toi = getInputTOI(toiIndex);
  toi->waitForData(i);
  return toi->getData(i);
}

double TOIProcessor::getError(int toiIndex, int i) {
  TOI* toi = getInputTOI(toiIndex);
  toi->waitForData(i);
  return toi->getError(i);
}

int_4 TOIProcessor::getFlag(int toiIndex, int i) {
  TOI* toi = getInputTOI(toiIndex);
  toi->waitForData(i);
  return toi->getFlag(i);
}

void TOIProcessor::setNeededHistory(int nsamples) {
  neededHistory =  nsamples;
}

void TOIProcessor::wontNeedBefore(int i) {
  if (i<wontNeedValue) return;
  wontNeedValue = i;
  for (int j=0; j< (int) inIx.size();  j++) {
    inTOIs[j]->wontNeedBefore(i);
  }
}

void TOIProcessor::autoWontNeed(int iCur) {
  if (neededHistory <=0) return;
  if (iCur < lastAWN + neededHistory) return;
  lastAWN = iCur;
  //cout << name << " wontNeedBefore "  << iCur-neededHistory << endl;
  wontNeedBefore(iCur-neededHistory);
}

void TOIProcessor::notify() {
  lock();
/*  if (mutex.__m_lock.__status == 0) {
    cout << "wait without lock" << endl; abort();
  }*/
  
  pthread_cond_broadcast(&dataReady);
  unlock();
}


void TOIProcessor::putData(int toiIndex, int i, double value, int_4 flg) {
  TOI* toi = getOutputTOI(toiIndex);
  toi->putData(i, value, flg);
  autoWontNeed(i);
  notify();
}


void TOIProcessor::putDataError(int toiIndex, int i, double value, 
                                double error, int_4 flg) { 
  TOI* toi = getOutputTOI(toiIndex);
  if (toi == NULL) 
    throw NullPtrError("TOIProcessor::putDataError() - Not assigned TOI !");
  toi->putDataError(i, value, error, flg);
  autoWontNeed(i);
  notify();
}