source: Sophya/trunk/ArchTOIPipe/Kernel/ringprocessor.cc@ 2386

// ArchTOIPipe           (C)     CEA/DAPNIA/SPP IN2P3/LAL
//                               Eric Aubourg
//                               Christophe Magneville
//                               Reza Ansari
// $Id: ringprocessor.cc,v 1.2 2003-05-20 10:10:09 aubourg Exp $

#include "ringprocessor.h"
#include "ringpipe.h"
#include "toimanager.h"

#include <iostream>
#include <typeinfo>        // ajout pour linux
#ifdef HAVE_VALUES_H
#include <values.h>
#endif
#ifndef MAXINT
#define MAXINT 2147483647
#endif

#ifdef HAVE_STDINT_H
#include <stdint.h>
#endif


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

RingProcessor::RingProcessor() {
  outRings = NULL;
  inRings = NULL;
  inited = false;

  ringBegin = 0;
  ringEnd = MAXINT;

  wontNeedRing = -1;
  neededRingHistory = 3;
  lastAWNR = -1;

  TOIManager::getManager()->registerProcessor(this);
}

RingProcessor::~RingProcessor() {
  delete[] inRings;
  delete[] outRings;
}


void RingProcessor::init()
{}

void RingProcessor::afterinit() {
  inRings = new (RingPipe*[inRingIx.size()]);
  for (int i=0; i<inRingIx.size(); i++)
    inRings[i] = NULL;
  outRings = new (RingPipe*[outRingIx.size()]);
  for (int i=0; i<outRingIx.size(); i++)
    outRings[i] = NULL;
}

void RingProcessor::setNeededRingHistory(int nr) {
  neededRingHistory = nr;
}

int RingProcessor::declareRingInput(string ring) {
  if (inRingIx.find(ring) != inRingIx.end())
    throw DuplicateIdExc("RingProcessor::declareRingInput : "+ring+
                         " already declared");
  int i = inRingIx.size();
  inRingIx[ring] = i;
  return i;
}


int RingProcessor::declareRingOutput(string ring) {
  if (outRingIx.find(ring) != outRingIx.end())
    throw DuplicateIdExc("RingProcessor::declareRingOutput : "+ring+
                         " already declared");
  int i = outRingIx.size();
  outRingIx[ring] = i;
  return i;
}

void RingProcessor::addRingInput(string name, RingPipe* ring) {
  chkinit();
  map<string, int>::iterator i = inRingIx.find(name);
  if (i == inRingIx.end()) throw NotFoundExc("RingProcessor::addInput "+
                                         name+" not declared");
  inRings[(*i).second] = ring;
  ring->addConsumer(this);  
}

void RingProcessor::addRingOutput(string name, RingPipe* ring) {
  chkinit();
  map<string, int>::iterator i = outRingIx.find(name);
  if (i == outRingIx.end()) throw NotFoundExc("RingProcessor::addOutput "+
                                          name+" not declared");
  ring->setProducer(this);
  outRings[(*i).second] = ring;
}

void RingProcessor::run()
{}

void* RingProcessor::ThreadStart(void* arg) {
  RingProcessor* r = (RingProcessor*) arg;
  try {
    r->run();
  }
  catch (PThrowable & exc) {
    cerr << "\n RingProcessor::ThreadStart() Catched Exception RingProcessor@" 
         << hex << r << dec << "\n"  
         << (string)typeid(exc).name() 
         << " - Msg= " << exc.Msg() << endl;
  }
  catch (const std::exception & sex) {
    cerr << "\n RingProcessor::ThreadStart() Catched std::exception RingProcessor@" 
         << hex << r << dec << "\n"  
         << (string)typeid(sex).name() << endl; 
  }
  catch (...) {
    cerr << "\n RingProcessor::ThreadStart() Catched ... exception RingProcessor@" 
         << hex << r << dec << endl;
  }
  //  r->warnPutDone();
  pthread_exit(NULL);
  return NULL;
}


void RingProcessor::start() {
  pthread_create(&thread, NULL, ThreadStart, this);
}

RingPipe* RingProcessor::getInputRing(int index) {
  if (index >= inRingIx.size())
    throw RangeCheckError("RingProcessor::getInputRing() out of bound");
  RingPipe* ring = inRings[index];
  if (ring == NULL)
    throw NullPtrError("RingProcessor::getInputRing() - Not assigned Ring !");
  return(ring);
}

RingPipe* RingProcessor::getOutputRing(int index) {
  if (index >= outRingIx.size())
    throw RangeCheckError("RingProcessor::getOutputRing() out of bound");
  RingPipe* ring = outRings[index];
  if (ring == NULL)
    throw NullPtrError("RingProcessor::getOutputRing() - Not assigned Ring !");
  return(ring);
}

RingPipe* RingProcessor::getOutRing(string out)
{
  // recherche du nom de la sortie et verification si le toi existe deja
  map<string, int>::iterator i = outRingIx.find(out);
  if (i == outRingIx.end()) {
    return NULL;
  } else {
    return outRings[(*i).second];
  }
}



Ring const* RingProcessor::getRing(int index, int i) {
  RingPipe* pipe = getInputRing(index);
  pipe->waitForRing(i);
  return pipe->getRing(i);
}


void RingProcessor::putRing(int index, int i, Ring const* ring) {
  RingPipe* pipe = getOutputRing(index);
  if (pipe == NULL) return;
  pipe->putRing(i, ring);
}

void RingProcessor::wontNeedRingBefore(int i) {
  if (i<wontNeedRing) return;
  wontNeedRing = i;
  for (int j=0; j< (int) inRingIx.size();  j++) {
    if (inRings[j])  inRings[j]->wontNeedRingBefore(i);
  }
}

void RingProcessor::autoWontNeedRing(int iCur) {
  if (neededRingHistory <=0) return;
  if (iCur <= lastAWNR + neededRingHistory/10) return;
  lastAWNR = iCur;
  wontNeedRingBefore(iCur-neededRingHistory);
}


void RingProcessor::getRingRange(int& min, int&max) {
  if (min < ringBegin) min = ringBegin;
  if (max > ringEnd)   max = ringEnd;

  for (int i=0; i<inRingIx.size(); i++)
    inRings[i]->getRingRange(min, max);

  ringBegin = min;
  ringEnd   = max;
}