source: Sophya/trunk/ArchTOIPipe/Kernel/cgt.cc@ 2331

// ArchTOIPipe           (C)     CEA/DAPNIA/SPP IN2P3/LAL
//                               Eric Aubourg
//                               Christophe Magneville
//                               Reza Ansari

// -------------------------------------------------------------------------
// classe CGT : Compagnie Generale des Tuyaux 
// Gestionnaire de tuyaux (toi) et d'unites de traitements (TOIProcessor)
//                                         R. Ansari / Mai 2002
// -------------------------------------------------------------------------

#include "cgt.h"
#include <typeinfo>
#include "toiseqbuff.h"
#include "toisegment.h"
#include "fitstoiwtr.h"
#include "toimanager.h"

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

#include <pthread.h>

CGT::CGT(bool fgsegmented, int bufsz, int maxseg)
{
  if (fgsegmented) {
    SelectTOISeqBuffered(bufsz);
    SelectTOISegmented(bufsz, maxseg);
  }
  else {
    SelectTOISegmented(bufsz, maxseg);
    SelectTOISeqBuffered(bufsz);
  }
  SetDebugLevel();
}

CGT::~CGT()
{
  vector< tuyau >::iterator it;
  if (dbgLevel > 0) 
    cout << " CGT::~CGT() - deleting TOI's " << endl;
  for(it = lesTuyaux.begin(); it != lesTuyaux.end(); it++) { 
    if (dbgLevel > 1) 
      cout << " ... deleting TOI " << (*it).toi->getName() << endl;
    delete (*it).toi;
  }
}


void CGT::SelectTOISegmented(int bufsz, int maxseg)
{
  fgSegmented = true;
  fgSeqbuff = false;
  segBuffsz = bufsz;
  segMaxseg = maxseg;
}

void CGT::SelectTOISeqBuffered(int wsz)
{
  fgSeqbuff = true;
  fgSegmented = false;
  seqWsz = wsz;
} 

TOI& CGT::Connect(TOIProcessor& prout, string out, 
                  TOIProcessor& prin, string in, string nom, int wbsz, bool withFlag)
{
  int id1 = findProcId(&prout);
  int id2 = findProcId(&prin);
  struct tuyau tuy;
  tuy.pr1id = id1;
  tuy.pr2id = id2;
  TOI* toi;
  if (nom.length() < 1) {
    char buff[128];
    sprintf(buff, "TOI%d_C[%d-%d]", (int)lesTuyaux.size(), id1, id2);
    nom = buff;
  }
  if (wbsz < 16)   wbsz = (fgSegmented) ? segBuffsz : seqWsz;
  if (fgSegmented) toi = new TOISegmented(nom, wbsz, segMaxseg);
  else toi = new TOISeqBuffered(nom, wbsz);
  tuy.toi = toi;
  lesTuyaux.push_back(tuy);
  prout.addOutput(out, toi);
  if (withFlag) { // Si c'est un FITSTOIWriter
    FITSTOIWriter* ftw = dynamic_cast< FITSTOIWriter* >(&prin);
    if (ftw) ftw->addInput(in, toi, withFlag);
    else prin.addInput(in, toi);
  }
  else prin.addInput(in, toi);
  if (dbgLevel > 1)
    cout << " CGT::Connect() TOI " << toi->getName() << " created " << endl;
  return(*toi);
}

TOI& CGT::Connect(TOIProcessor& prout, const char* out, 
                  TOIProcessor& prin, const char* in, string nom, int wbsz, bool withFlag)
{
  string outs = out;  
  string ins = in;
  return Connect(prout, outs, prin, ins, nom, wbsz, withFlag);
}

void CGT::ListProcs(::ostream& os, bool prstat) // const plus tard
{
  os << " --- CGT::ListProcs() , NbProcs=" << lesProcs.size() << " ----- " << endl;
  //  vector< TOIProcessor*  >::const_iterator it;
  //  for(it = lesProcs.begin(); it != lesProcs.end(); it++) { 
  for(int i=0; i< lesProcs.size(); i++) {
    os << " >> Proc Id= " << i << " @ " << hex << lesProcs[i] << dec ;
    string ctyp = typeid(*lesProcs[i]).name();
    os << " Type= " << ctyp << endl;
    if (prstat) lesProcs[i]->PrintStatus(os);
  }
}

void CGT::ListTOIs(::ostream& os, bool prstat)
{
  os << " --- CGT::ListTOIs() , NbTOIs=" << lesTuyaux.size();
  if ( fgSegmented ) os << " Default TOI: TOISegmented " << endl;
  else os << " Default TOI: TOISeqBuffered " << endl;
  for(int i=0; i< lesTuyaux.size(); i++) {
    os << " >> TOI " << lesTuyaux[i].toi->getName() << " Connecting Procs (Id): " 
       << lesTuyaux[i].pr1id << " <---> " << lesTuyaux[i].pr2id << endl;
    if (prstat > 1)  lesTuyaux[i].toi->PrintStatus(os);    
    else if (prstat == 1)  
      os << " PutCountWait= " << lesTuyaux[i].toi->getCountWaitPut() 
         << " GetCountWait= " << lesTuyaux[i].toi->getCountWaitGet() << endl;
  }  
}

void CGT::PrintStatus(::ostream & os, bool prstat)
{
  ListProcs(os, prstat);
  ListTOIs(os, prstat);
}

void CGT::DeleteProcs()
{
  vector< TOIProcessor*  >::iterator it;
  cout << " CGT::DeleteProcs() - deleting TOIProcessors - NProc=" 
       << lesProcs.size() << endl;
  for(it = lesProcs.begin(); it != lesProcs.end(); it++) {
    if (dbgLevel > 1) 
      cout << " ... deleting TOIProcessors @ " << hex << *it << dec << endl;
    delete *it;
  }
}

void CGT::Start()
{
  cout << " CGT::Start() starting TOIProcessors ... NProc=" 
       << lesProcs.size() << endl;
  /*  for(int i=0; i< lesProcs.size(); i++) {
      if (dbgLevel > 1) 
      cout << " Staring Proc Id= " << i << " @ " << hex << lesProcs[i] << dec << endl;
      lesProcs[i]->start();
      }
  */
  TOIManager::getManager()->startAll();
}

int CGT::findProcId(TOIProcessor* pr)
{
  for(int i=0; i< lesProcs.size(); i++) 
    if (lesProcs[i] == pr) return(i);

  lesProcs.push_back(pr);
  return(lesProcs.size()-1);
}