source: Sophya/trunk/AddOn/TAcq/racqurw.cc@ 4016

//----------------------------------------------------------------
// ---- classes de threads pour lecture (transfert DMA) 
//      et ecriture disque pour acquisition   BAORadio       -----
// LAL -      R. Ansari - Juin/Juillet 2008 
//----------------------------------------------------------------

#include "racqurw.h"

#include <stdlib.h>
#include <unistd.h>
#include <fstream>
#include <signal.h>
#include "pexceptions.h"
#include "timestamp.h"

#include "pciewrap.h"
#include "brpaqu.h"
#include "minifits.h"

#include "resusage.h" // Pour mesure temps elapsed/CPU ...
#include "datatable.h" // Pour sauver les entetes de paquet 
#include <sys/time.h>  // pour gettimeofday

// Si on veut que MultiDataSaver cree des fichiers avec le numero des FrameCounters...
// define DEBUGPAQHDR  

//-------------------------------------------------------
// Classe thread de lecture PCI-Express
//-------------------------------------------------------

/*!
   \class PCIEReader
   \ingroup TAcq

   \brief PCI-Express data read-out thread class \b (deprecated)

   \b Deprecated - Use PCIEMultiReader instead.
   DMA through the PCIEWrapperInterface object interface and tranfer to RAcqMemZoneMgr memory
*/ 

PCIEReader::PCIEReader(PCIEWrapperInterface &pciw,uint_4 sizeFrame,uint_4 packSize ,RAcqMemZoneMgr& mem, 
                       uint_4 nmax, BRDataFmtConv swapall)
  :   memgr(mem) , pciw_ (pciw)
{
  nmax_ = nmax; 
  swapall_ = swapall;   // select data swap/format conversion for BRPaquet 
  stop_ = false;
  packSize_ = packSize;
  sizeFr_ =sizeFrame;
  // Pour la logique de gestion des paquets ds zone memoire
  mid_ = -2;
  targ_npaq_ = 0;
 
 max_targ_npaq = memgr.NbPaquets();
  mmbuf_ = NULL;
}

bool PCIEReader::MZoneManage(bool fgclean)    // Retourne true si probleme 
{
  /* Pour debug 
  cout << " PCIEReader::MZoneManage() mid_=" << mid_ << " arg_npaq_= " << targ_npaq_  
       << " max_targ_npaq=" << max_targ_npaq << endl;
  */
  if (mid_ >= 0)  memgr.FreeMemZone(mid_, MemZS_Filled);
  mmbuf_ = NULL;  targ_npaq_ = 0;  mid_ = -2;
  if (fgclean)  return false;
  mid_ = memgr.FindMemZoneId(MemZA_Fill);
  mmbuf_ = memgr.GetMemZone(mid_);
  if (mmbuf_==NULL) return true;
  return false;
}

void PCIEReader::run()
{
   //Precision insuffisante  ResourceUsage ru; ru.Update();  // Pour recuperer le temps passe
  struct timeval tv1, tv2;
  gettimeofday(&tv1, NULL);
  
  try{
  cout << " PCIEReader::run() - Starting , NMaxMemZones=" << nmax_ 
       << "memgr.NbPaquets()=" << memgr.NbPaquets() << endl;    
  setRC(1);     

 
  // sigaddset(&act.sa_mask,SIGINT);  // pour proteger le transfert DMA
  //sigaction(SIGINT,&act,NULL);  
        
  uint_4 paqsz = memgr.PaqSize();
  uint_4 dmasz = pciw_.TransferSize();
  pciw_.StartTransfers();

  Byte* Datas = NULL;
  Byte* tampon = new Byte[paqsz];
  Byte* nextpaq = NULL;

  uint_4 off_acheval = 0;
  
  int nerrdma = 0;
  int maxerrdma = 10;
  bool fgarret = false;
  
  uint_4 npaqfait = 0;    //  Nb total de paquets traites (DMA + decode)
  
  ///  do{  si boucle infini 
  //  for (uint_4 kmz=0; kmz<nmax_; kmz++)  {
  while (npaqfait < nmax_*memgr.NbPaquets()) {
    if (stop_) break;
    if (fgarret) break;
    // On pointe vers le debut de la zone a remplir aver le prochain DMA
      Datas=pciw_.GetData();
      if (Datas == NULL) { // No data Read in DMA 
        nerrdma ++;
        cout << "PCIEReaderChecker/Erreur Waiting for datas ..." << endl;
        pciw_.PrintStatus(cout);
        if (nerrdma>=maxerrdma) { fgarret = true; break; }
      }
      else {  // DMA reussi 
        uint_4 curoff = 0;
        //1- On traite le paquet a cheval, rempli partiellement avec le DMA d'avant si necessaire
        if (off_acheval > 0) {
          if ((paqsz-off_acheval)< dmasz) { 
            memcpy((void *)(tampon+off_acheval), (void *)Datas, paqsz-off_acheval);
            curoff = paqsz-off_acheval;  off_acheval = 0;  
            
            if ((nextpaq=NextPaqTarget()) == NULL) {
              cout << "2 PCIEReader::run()/Error NextPaqTarget() returned NULL ->STOP 9" << endl;
              setRC(9);  fgarret=true; break;
            }
            BRPaquet paq(tampon, nextpaq, paqsz, swapall_);
            npaqfait++;  // Ne pas oublier le compteur de paquets faits 
          }
          else {
            memcpy((void *)(tampon+off_acheval), (void *)Datas, dmasz);
            curoff =dmasz;
            off_acheval = (dmasz+off_acheval);
          }
        }
        //2- On traite les paquets complet qui se trouvent dans la zone du DMA 
        while((curoff+paqsz)<=dmasz) {
          //          BRPaquet paq((Byte*)(Datas)+((paqsz*j)), nextdma+j*paqsz, paqsz, swapall_);
          if ((nextpaq=NextPaqTarget()) == NULL) {
            cout << "3 PCIEReader::run()/Error NextPaqTarget() returned NULL ->STOP 9" << endl;
            setRC(9);  fgarret=true; break;
          }
          BRPaquet paq(Datas+curoff, nextpaq, paqsz, swapall_);
          curoff += paqsz;  // On avance l'index dans le buffer du DMA
          npaqfait++;  // Ne pas oublier le compteur de paquets faits 
        }  // -- FIN traitement des paquets complets ds un DMA 
        //3- On copie si besoin la fin du DMA dans la zone tampon
        if (curoff < dmasz) {
          if (fgarret) break;  // pour sortir si l'on est passe par un STOP 9
          off_acheval = dmasz-curoff;
          memcpy(tampon, (void*)(Datas+curoff), off_acheval);
          curoff += off_acheval; 
        }
      }  //   Traitement d'un DMA OK 
  }
//  }while(!stop_);     
  
  
  gettimeofday(&tv2, NULL);
  double tmelaps2 = (tv2.tv_sec-tv1.tv_sec)*1000.+(tv2.tv_usec-tv1.tv_usec)/1000.;
  if (tmelaps2<0.1) tmelaps2=0.1; 
  cout << " ------------------ PCIEReader::run()-End summary -------------------" << endl;
  cout << " PCIEReader/Info TotTransfer=" << pciw_.TotTransferBytes()/1024 
       << " kb , ElapsTime=" << tmelaps2 << " ms ->" 
       << (double)pciw_.TotTransferBytes()/tmelaps2 << " kb/s" << endl; 
  cout << " --------------------------------------------------------------------" << endl;
  
  MZoneManage(true);
  delete [] tampon;
 
  }catch (PException& exc) {
    cout << " PCIEREADER::run()/catched PException " << exc.Msg() << endl;
    setRC(3);   
    return; 
  }
  catch(...) {
    cout << "  PCIEREADER::run()/catched unknown ... exception " << endl;
    setRC(4);   
    return; 
  }
  setRC(0);
 
  return;
}


void PCIEReader::Stop()
{ 
  // cout << " PCIEReader::Stop() -------------> STOP" <<endl; 
  stop_ = true;
  return;
}


//-------------------------------------------------------
// Classe thread de sauvegarde sur fichiers
//-------------------------------------------------------

/*!
   \class DataSaver
   \ingroup TAcq

   \brief Data saver in FITS format thread class \b (deprecated)

   \b Deprecated - Use MultiDataSaver instead. 
   BRPaquets are dumped to disk in FITS format.
*/ 

DataSaver::DataSaver(RAcqMemZoneMgr& mem, string& path, uint_4 nfiles, uint_4 nblocperfile, bool savesig)
  :  memgr(mem) 
{
  nfiles_ = nfiles;
  nblocperfile_ = nblocperfile;
  nmax_ = nblocperfile_*nfiles_;
  savesig_ = savesig;  // Si false, pas d'ecriture des fichiers FITS du signal  
  stop_ = false;        
  path_ = path; 
}
void DataSaver::Stop()
{
  // cout<< " DataSaver:Stop ........ " << endl;
  stop_=true;

}
void DataSaver::run()
{
  setRC(1);     
  BRPaqChecker pcheck;  // Verification/comptage des paquets 

  try {
    TimeStamp ts; 
    cout << " DataSaver::run() - Starting " << ts << " NbFiles=" << nfiles_ << " NBloc/File=" 
         << nblocperfile_ << " NMaxMemZones=" << nmax_ << endl; 
    char fname[512];

    sprintf(fname,"%s/saver.log",path_.c_str());
    ofstream filog(fname);
    filog << " DataSaver::run() - starting log file " << ts << endl;                   
    filog << " NbFiles=" << nfiles_ << " NBloc/File="  << nblocperfile_ << " NMaxMemZones=" << nmax_ << endl;
        
    DataTable dt;
    dt.AddLongColumn("TimeTag");
    dt.AddIntegerColumn("FrameCounter");
    dt.AddIntegerColumn("FrameLength");
    dt.AddIntegerColumn("Num");
    DataTableRow dtr = dt.EmptyRow();
    uint_8 timtag = 0;
    uint_4 numpaq = 0;
    bool dthead = false;  // Mettre a false pour ne pas remplir DataTable Headers


    uint_4 fnum=0;
    uint_4 paqsz = memgr.PaqSize();
    cout << " ============================ DataSaver::run() PaqSize " << paqsz <<endl;
    bool fgnulldev = false;
    if (path_ == "/dev/null") { 
      cout << " DataSaver::run()/Warning /dev/null path specified, filenames=/dev/null" << endl;
      fgnulldev = true;
    }
    for (uint_4 nbFile=0;nbFile<nfiles_ ;nbFile++) {
      if (stop_ )   break;
      if (fgnulldev) strcpy(fname,"/dev/null");
      else sprintf(fname,"%s/HDRfits%d.txt",path_.c_str(),fnum);
      ofstream header(fname);

      BRPaquet paq0(NULL, NULL, paqsz); 
      uint_4 npaqperfile = memgr.NbPaquets()*nblocperfile_;  // Nombre de paquets ecrits dans un fichier 

      MiniFITSFile mff; 
      if (savesig_) { //Reza - Ouverture conditionnel fichier 
        if (fgnulldev) strcpy(fname,"/dev/null");
        else sprintf(fname,"%s/signal%d.fits",path_.c_str(),(int)fnum++);
        mff.Open(fname,MF_Write);  //Reza - Ouverture conditionnel fichier 
        // Entete correspondant a l'ecriture tout le paquet - trailer compris (modif Mai 2009)
        mff.setDTypeNaxis(MF_Byte, paq0.PaquetSize(), npaqperfile);

        // Sans TRAILER de paquet mff.setDTypeNaxis(MF_Byte, paq0.DataSize()+paq0.HeaderSize(), npaqperfile); 
      }
      else sprintf(fname,"MemDataBloc[%d]-NoDataFile",(int)fnum++);

      for (uint_4 kmz=0; kmz<nblocperfile_; kmz++) {
        if (stop_) break;
          //DBG cout << " DataSaver::run()- nbFile=" << nbFile << " kmz=" << kmz << endl;       
        int mid = memgr.FindMemZoneId(MemZA_Save);  
        Byte* buff = memgr.GetMemZone(mid);
        if (buff == NULL) {
          cout << " DataSaver::run()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
          setRC(2);     
          return;       
        }
        for(uint_4 i=0; i<memgr.NbPaquets(); i++) {
 
          BRPaquet paq(NULL, buff+i*paqsz, paqsz);
          pcheck.Check(paq);   // Verification du paquet / FrameCounter 
          if (savesig_)
             header << hex << paq.HDRMarker() << " " << paq.TRLMarker() << " " 
                    << paq.TimeTag2()<< " "<< paq.TimeTag1()<< " " 
                    << paq.FrameCounter() << " " << paq.PaqLen()  << endl;

          if (dthead) {  // Remplissage DataTable entetes paquets 
            timtag = (uint_8)paq.TimeTag2()*0x100000000ULL+paq.TimeTag1();
            dtr[0] = timtag;
            dtr[1] = paq.FrameCounter();
            dtr[2] = paq.PaqLen();
            dtr[3] = numpaq++;
            dt.AddRow(dtr);
          }

          if (savesig_) // Reza - Ecriture conditionnel fichier fits signal 
            mff.WriteB(paq.Header(),paq.PaquetSize()); // ecriture tout le paquet (modif Mai 2009)
            //  mff.WriteB(paq.Data1(), paq.DataSize());
            //  mff.WriteB(paq.Header(),paq.DataSize()+paq.HeaderSize()); // ecriture datas + header
        }
        memgr.FreeMemZone(mid, MemZS_Saved);
      }
      ts.SetNow(); 
      filog << ts << " : OK data file  " << fname << endl;                     
      cout << " DataSaver::run() " << ts << " : OK data file  " << fname << endl;
    }
    if (dthead) {
      cout << dt;
      char fname2[256];
      sprintf(fname2,"%s/headers.ppf",path_.c_str());
      POutPersist po(fname2);
      po << dt;
    }
    cout << " --------------------  DataSaver::run() -------------------- " << endl;
    pcheck.Print(cout);
    cout << " ---------------------------------------------------------- " << endl;
    ts.SetNow();
    pcheck.Print(filog);
    filog << " DataSaver::run() - End of processing/run() " << ts << endl;                     

  }
  catch (MiniFITSException& exc) {
    cout << " DataSaver::run()/catched MiniFITSException " << exc.Msg() << endl;
    setRC(3);   
    return; 
  }
  catch(...) {
    cout << " DataSaver::run()/catched unknown ... exception " << endl;
    setRC(4);   
    return; 
  }
  setRC(0);
  return;
}   




//----------------------------------------------------------------------------------------------------------
// Classe thread de lecture PCI-Express + Check pour tests de verification de debit/etc avec un seul thread
//----------------------------------------------------------------------------------------------------------

/*!
   \class PCIEReaderChecker
   \ingroup TAcq

   \brief PCI-Express data read-out and BRPaquet structure check 

*/ 


PCIEReaderChecker::PCIEReaderChecker(PCIEWrapperInterface &pciw, uint_4 sizeFrame,uint_4 packSize, RAcqMemZoneMgr& mem, 
                                     uint_4 nmax, BRDataFmtConv swapall) 
  :   memgr(mem) , pciw_ (pciw)
{
  nmax_ = nmax; 
  swapall_ = swapall;   // select data swap/format conversion for BRPaquet
  stop_ = false;
  packSize_ = packSize;
  sizeFr_ =sizeFrame;
}
void PCIEReaderChecker::run()
{
  
  struct timeval tv1, tv2;
  gettimeofday(&tv1, NULL);
  
  cout << " PCIEReaderChecker::run() - Starting , NMaxMemZones=" << nmax_ 
       << " memgr.NbPaquets()=" << memgr.NbPaquets() << endl;   
  setRC(1);     
  cout << " ... RAcqMemZoneMgr not used - using s fixed memory location for packets decoding ..." << endl;

 
  // sigaddset(&act.sa_mask,SIGINT);  // pour proteger le transfert DMA
  //sigaction(SIGINT,&act,NULL);        
  uint_4 paqsz = memgr.PaqSize();
  uint_4 dmasz = pciw_.TransferSize();
  pciw_.StartTransfers();

  BRPaqChecker pcheck;  // Verification/comptage des paquets 

  Byte* Datas = NULL;
  Byte* locdata = new Byte[paqsz*memgr.NbPaquets()*memgr.NbZones()];
  Byte* tampon = new Byte[paqsz];

  uint_4 off_acheval = 0;

  int nerrdma = 0;
  int maxerrdma = 10;
  bool fgarret = false;
  for (uint_4 kmz=0; kmz<nmax_; kmz++)  {
    if (stop_) break;
    if (fgarret) break;

    Byte* nextdma = locdata+((kmz%memgr.NbZones())*(paqsz*memgr.NbPaquets()));
    uint_4 npaqfait = 0;      
    //      for (uint_4 i=0; i<memgr.NbPaquets(); i += pktInDMATr) {  // attention pktInDMATr paquets dans 1 seul DMA
    while (npaqfait < memgr.NbPaquets()) {
      if (fgarret) break;
      // On pointe vers le debut de la zone a remplir aver le prochain DMA
      //-- Zone memoire locale Byte* nextdma = buff+i*paqsz;
        Datas=pciw_.GetData();
        
        if (Datas == NULL) { // No data Read in DMA 
          nerrdma ++;
          cout << "PCIEReaderChecker/Erreur Waiting for datas ..." << endl;
          pciw_.PrintStatus(cout);
          if (nerrdma>=maxerrdma) { fgarret = true; break; }
        }
        else {
          uint_4 curoff = 0;
          //1- On traite le paquet a cheval, rempli partiellement avec le DMA d'avant si necessaire
          //      if (off_acheval > 0) {  
          //  memcpy((void *)(tampon+off_acheval), (void *)Datas, paqsz-off_acheval);
          //  curoff = paqsz-off_acheval;  off_acheval = 0;  
            
          // BRPaquet paq(tampon, locdata, paqsz, swapall_);
          //  npaqfait++;  // Ne pas oublier le compteur de paquets faits 
          //  pcheck.Check(paq);   // Verification du paquet / FrameCounter 
          //}
        if (off_acheval > 0) {
          if ((paqsz-off_acheval)< dmasz) { 
            memcpy((void *)(tampon+off_acheval), (void *)Datas, paqsz-off_acheval);
            curoff = paqsz-off_acheval;  off_acheval = 0;  
            
            BRPaquet paq(tampon, locdata, paqsz, swapall_);
            npaqfait++;  // Ne pas oublier le compteur de paquets faits
            pcheck.Check(paq);   // Verification du paquet / FrameCounter 
          }
          else {
            memcpy((void *)(tampon+off_acheval), (void *)Datas, dmasz);
            curoff =dmasz;
            off_acheval = (dmasz+off_acheval);
          }
        }
          //2- On traite les paquets complet qui se trouvent dans la zone du DMA 
          while((curoff+paqsz)<=dmasz) {
            //        BRPaquet paq((Byte*)(Datas)+((paqsz*j)), nextdma+j*paqsz, paqsz, swapall_);
            //  BRPaquet paq(Datas+curoff, locdata, paqsz, swapall_);
            BRPaquet paq(Datas+curoff, nextdma+npaqfait*paqsz, paqsz, swapall_);
            curoff += paqsz;  // On avance l'index dans le buffer du DMA
            npaqfait++;  // Ne pas oublier le compteur de paquets faits 
            pcheck.Check(paq);   // Verification du paquet / FrameCounter 
          }  // -- FIN traitement des paquets complets ds un DMA 
          //3- On copie si besoin la fin du DMA dans la zone tampon
          if (curoff < dmasz) {
            off_acheval = dmasz-curoff;
            memcpy(tampon, (void*)(Datas+curoff), off_acheval);
            curoff += off_acheval; 
          }
        }  //   Traitement d'un DMA OK 
        
    }  // Fin boucle de remplissage d'une zone memoire
  } //  Fin boucle sur les zones 
  
  setRC(0);
  gettimeofday(&tv2, NULL);
  double tmelaps2 = (tv2.tv_sec-tv1.tv_sec)*1000.+(tv2.tv_usec-tv1.tv_usec)/1000.;
  if (tmelaps2<0.1) tmelaps2=0.1; 
  cout << " ------------------ PCIEReaderChecker::run()-End summary -------------------" << endl;
  cout << " PCIEReaderChecker/Info TotTransfer=" << pciw_.TotTransferBytes()/1024 
       << " kb , ElapsTime=" << tmelaps2 << " ms ->" 
       << (double)pciw_.TotTransferBytes()/tmelaps2 << " kb/s" << endl; 
  pcheck.Print(cout);
  cout << " --------------------------------------------------------------------" << endl;

  delete [] locdata;
  delete [] tampon;

  return;
}
void PCIEReaderChecker::Stop()
{ 
  // cout << " PCIEReaderChecker::stop()  ........ STOP" <<endl; 
  stop_ = true;

}
////////////////////////////////////////////////////////////////////////////////////////////////////////
//----------------------------------------------------------------------------------------------------------
// Classe thread de lecture PCI-Express + Check pour tests de verification de debit/etc avec un seul thread
//----------------------------------------------------------------------------------------------------------
/*!
   \class PCIEMultiReader
   \ingroup TAcq

   \brief PCI-Express data read-out thread class from multiple fibres (sources)

   Perform DMA through the set PCIEWrapperInterface object interfaces and tranfer date to RAcqMemZoneMgr memory.
*/ 

/* --Methode-- */
PCIEMultiReader::PCIEMultiReader(vector<PCIEWrapperInterface*> vec_pciw, RAcqMemZoneMgr& mem, BRParList const& par)
  :   memgr(mem), par_(par), vec_pciw_ (vec_pciw)
{
  nmax_ = par_.MaxNbBlocs();    
  swapall_ = par_.GetDataConvFg();      // select data swap/format conversion for BRPaquet
  stop_ = false;
  packSize_ = par_.RecvPaquetSize();
  packSizeInMgr_=memgr.PaqSize();
  sizeFr_=par_.DMASizeBytes();
  if (vec_pciw.size() != memgr.NbFibres()) {
    cout << " PCIEMultiReader()PbArgs: vec_pciw.size()= " << vec_pciw.size() << " memgr.NbFibres()=" <<memgr.NbFibres()<< endl;
    throw ParmError("PCIEMultiReader:ERROR/ arguments incompatibles vec_pciw.size() != memgr.NbFibres() ");
  }
  if (vec_pciw.size() > MAXNBFIB) 
    throw ParmError("PCIEMultiReader:ERROR/  vec_pciw.size() > MAXNBFIB ");
  nbDma_= vec_pciw.size();
  mid_=-2; 
  mmbuf_=NULL;
  max_targ_npaq = memgr.NbPaquets();
  for (int fid=0 ; fid<(int)nbDma_ ;fid++) { 
    memgr.FiberId(fid)=vec_pciw[fid]->FiberId(); 
    mmbufib_[fid]=NULL;
  }
  stopreason_="??Unknown??";
}

/* --Methode-- */
void PCIEMultiReader::run()
{
  
  struct timeval tv1,tv2;
  gettimeofday(&tv1, NULL);
  
  cout << " PCIEMultiReader::run() - Starting , NMaxMemZones=" << nmax_ 
       << " memgr.NbPaquets(),PaqSz=" << memgr.NbPaquets() << " ," << memgr.PaqSize() 
       << " DMA-Paqsize " << packSize_ << "  " << BRPaquet::FmtConvToString(swapall_) << endl;  
  setRC(1);     

  // sigaddset(&act.sa_mask,SIGINT);  // pour proteger le transfert DMA
  //sigaction(SIGINT,&act,NULL);        
  uint_4 paqszmm = memgr.PaqSize();
  uint_4 paqsz =  packSize_;
  uint_4 dmasz = vec_pciw_[0]->TransferSize();
  //DEL   vec_pciw_[0]->StartTransfers();

  BRPaqChecker pcheck[MAXNBFIB];  // Verification/comptage des paquets  
  Byte* Datas[MAXNBFIB];
  Byte* tampon[MAXNBFIB] ; 
  Byte* predtampon=NULL;  // tampon de recopie pour la reduction des tailles de paquets 
  Byte* nextpaq=NULL;
  uint_4 off_acheval=0;
  
  bool fgarret = false;
  
  // Initialisation des tampons pour recopie des paquets a cheval pour chaque DMA
  for (int i=0;i< (int)nbDma_ ;i++) {  
    tampon[i]=   new Byte[paqsz];
    vec_pciw_[i]->SetMaxWaitEndDMA(par_.first_maxkwedma_,par_.first_nretrydma_);
  }
  bool fgredpaq=par_.fgreducpsize;
  if (fgredpaq) {
    cout << " PCIEMultiReader::run() - PaquetSizeReduction - RedSize=" << par_.redpqsize 
         << " Offset=" << par_.reducoffset << " " << ((par_.reducneedcopy)?"NeedCopy":"NOCopy") 
         << "  " << BRPaquet::ReducActionToString(par_.pqreducmode) << endl;
    predtampon = new Byte[paqsz];
  }
  
#ifdef DEBUGPAQHDR  
  ofstream header[MAXNBFIB];
  for(uint_4 fib=0; fib<nbDma_; fib++) {
    char hfnm[128];
    sprintf(hfnm, "./HDRCountPaqs%d.txt", fib);
    header[fib].open(hfnm);
  }
#endif   
  uint_4 npaqfait[MAXNBFIB] ;
  for (int i=0;i< (int)nbDma_ ;i++) npaqfait[i]=0;
    // Byte* nextdma = locdata+((kmz%memgr.NbZones())*(paqsz*memgr.NbPaquets()));
  uint_4 npaqfaitg = 0; 
  bool fg_change_timeout=true;     
  //      for (uint_4 i=0; i<memgr.NbPaquets(); i += pktInDMATr) {  // attention pktInDMATr paquets dans 1 seul DMA
  while (npaqfaitg < nmax_*memgr.NbPaquets()) {  // Boucle global G 
    if (fgarret) break;
    if (stop_) break;
    
    // Lancement des DMA 
    for (int dma=0; dma < (int)nbDma_ ;dma++) vec_pciw_[dma]->StartTransfers();
    if ((npaqfaitg>1)&&fg_change_timeout)  { 
      for (int i=0;i< (int)nbDma_ ;i++)
        vec_pciw_[i]->SetMaxWaitEndDMA(par_.maxkwedma_,par_.nretrydma_);
      fg_change_timeout=false;
    }
    // On pointe vers le debut de la zone a remplir aver le prochain DMA
    //-- Zone memoire locale Byte* nextdma = buff+i*paqsz;
    
    bool fgbaddma=false;
    // On boucle sur les nbDma_ en attente de leurs terminaison    
    for (int dma=0; dma <(int) nbDma_ ;dma++)  {
      Datas[dma]=vec_pciw_[dma]->GetData();
      if (Datas[dma] == NULL) { // No data Read in DMA 
        fgbaddma=true;
        cout << "PCIEMultiReaderChecker/ERROR - DMA failed !" << endl;
        vec_pciw_[dma]->PrintStatus(cout);
        stopreason_="--Failed DMA--";
        fgarret = true;  break; 
      }
    }
    if (fgbaddma) continue;  
    uint_4 curoff=0;
    //1- On traite le paquet a cheval, rempli partiellement avec le DMA d'avant si necessaire pour les n fibres
    if (off_acheval  > 0) {  // IF Numero B
      if ((paqsz-off_acheval)< dmasz) {  // IF Numero A
        for(uint_4 fib=0; fib<nbDma_; fib++) 
          memcpy((void *)((tampon[fib])+off_acheval), (void *)Datas[fib], paqsz-off_acheval);
        curoff = paqsz-off_acheval;  off_acheval = 0;
        if ( MoveToNextTarget() ) {
          cout << "PCIEMultiReader::run()/Error-A- MoveToNextTarget() returned true ->STOP 9" << endl;
          setRC(9);  fgarret=true; break;
        }
        for(uint_4 fib=0; fib<nbDma_; fib++) {
          nextpaq=GetPaquetTarget(fib);
          if (nextpaq == NULL) { // Cela ne devrait pas arriver 
            cout << "PCIEReader::run()/Error-A2- GetPaquetTarget(fib)  returned NULL ->STOP 9" << endl;
            setRC(9);  fgarret=true; break;
          }
          // CHECK  S'il faut faire une reduction de taille de paquet
          if (fgredpaq) { // reduction taille de paquet
            if (par_.reducneedcopy) {
              BRPaquet paqc1(tampon[fib], predtampon, paqsz, swapall_);
              BRPaquet paqc2(nextpaq, par_.redpqsize);
              paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
            }
            else {
              BRPaquet paqc1(tampon[fib], paqsz);
              BRPaquet paqc2(nextpaq, par_.redpqsize);
              paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
            }
          }
          else  { 
            BRPaquet paqc(tampon[fib], nextpaq, paqsz, swapall_);
          }
          BRPaquet paq(nextpaq, packSizeInMgr_);
          npaqfait[fib]++;
          if (fib==nbDma_-1) npaqfaitg++;  // Ne pas oublier le compteur de paquets faits 
          pcheck[fib].Check(paq);   // Verification du paquet / FrameCounter 
#ifdef DEBUGPAQHDR  
          header[fib] << dec << paq.FrameCounter()<< endl; 
#endif
        }
      }
      else {  // se rapporte au IF numero A
        for(uint_4 fib=0; fib<nbDma_; fib++) 
          memcpy((void *)(tampon[fib]+off_acheval), (void *)Datas[fib], dmasz);
        curoff =dmasz;
        off_acheval = (dmasz+off_acheval);
      } 
    }  // Fin IF Numero B

    //2- On traite les paquets complets qui se trouvent dans la zone du DMA 
    while ((curoff+paqsz)<=dmasz) {  // while numero C 
      //          if ((dma==nbDma_-1)&&(npaqfait >= nmax_* memgr.NbPaquets())) break;
      if ( MoveToNextTarget() ) {
        cout << "PCIEMultiReader::run()/Error-B- MoveToNextTarget() returned true ->STOP 9" << endl;
        setRC(9);  fgarret=true; break;
      }
      for(uint_4 fib=0; fib<nbDma_; fib++) {
        if (npaqfait[fib] >= nmax_*memgr.NbPaquets())  continue;
        nextpaq=GetPaquetTarget(fib);
        if (nextpaq == NULL) { // Cela ne devrait pas arriver 
          cout << "PCIEReader::run()/Error-B2- GetPaquetTarget(fib)  returned NULL ->STOP 9" << endl;
          setRC(9);  fgarret=true; break;
        }
        // CHECK  S'il faut faire une reduction de taille de paquet
        if (fgredpaq) { // reduction taille de paquet
          if (par_.reducneedcopy) {
            BRPaquet paqc1(Datas[fib]+curoff, predtampon, paqsz, swapall_);
            BRPaquet paqc2(nextpaq, par_.redpqsize);
            paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
          }
          else {
            BRPaquet paqc1(Datas[fib]+curoff, paqsz);
            BRPaquet paqc2(nextpaq, par_.redpqsize);
            paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
          }
        }
        else { 
          BRPaquet paqc(Datas[fib]+curoff, nextpaq, paqsz, swapall_);
        }
        BRPaquet paq(nextpaq, packSizeInMgr_);
        npaqfait[fib]++;
        if (fib==nbDma_-1) npaqfaitg++;  // Ne pas oublier le compteur de paquets faits 
        pcheck[fib].Check(paq);   // Verification du paquet / FrameCounter 
#ifdef DEBUGPAQHDR  
        header[fib] << dec << paq.FrameCounter()<< endl; 
#endif
      }
      curoff += paqsz;  // On avance l'index dans le buffer du DMA
    } // -- FIN traitement des paquets complets ds un DMA - FIN du while numero C 
      //3- On copie si besoin la fin du DMA dans la zone tampon
    if (curoff < dmasz) {  // IF numero D 
      off_acheval = dmasz-curoff;
      for(uint_4 fib=0; fib<nbDma_; fib++) 
        memcpy(tampon[fib], (void*)(Datas[fib]+curoff), off_acheval);
      // ne sert a rien         curoff += off_acheval; 
    } // FIN du if numero D 
  }  //   FIN  Boucle global G 

  if (npaqfaitg >= nmax_*memgr.NbPaquets())   stopreason_="--Max Nb paquets reached--";

  MoveToNextTarget();  // Pour faire traiter le dernier paquet si plein 
  MZoneManage(true);  //---- Nettoyage final
    
  setRC(0);
  gettimeofday(&tv2, NULL);
  double tmelaps2 = (tv2.tv_sec-tv1.tv_sec)*1000.+(tv2.tv_usec-tv1.tv_usec)/1000.;
  if (tmelaps2<0.1) tmelaps2=0.1; 
  cout << " ---------- PCIEMultiReader::run()-End StopReason: " << stopreason_ << endl
       << " Summary NPaqFait=" << npaqfaitg << "------------- " << endl;
  for (int dma=0; dma < (int)nbDma_ ;dma++)   {
    cout << " --Fib=" << dma << " NPaqFait=" << npaqfait[dma] <<  " TotTransfer=" 
         << vec_pciw_[dma]->TotTransferBytes()/1024 
         << " kb , ElapsTime=" << tmelaps2 << " ms ->" 
         << (double)vec_pciw_[dma]->TotTransferBytes()/tmelaps2 << " kb/s" << endl;     
    pcheck[dma].Print(cout);
  }
  cout << " --------------------------------------------------------------------" << endl;

  usleep(250000);       // Attente de traitement du dernier paquet
  memgr.Stop();   // Pour arreter les autres threads 

  for (int i=0;i< (int)nbDma_ ;i++) delete[] tampon[i]; 
  if ((fgredpaq)&&predtampon) delete[] predtampon;
#ifdef DEBUGPAQHDR  
  for(uint_4 fib=0; fib<nbDma_; fib++)  header[fib].close();
#endif

  //DBG  cout << " fin thread ========================" <<endl;  
  return;
}

/* --Methode-- */
bool PCIEMultiReader::MZoneManage(bool fgclean)    // Retourne true si probleme 
{
  /* Pour debug 
  cout << " PCIEReader::MZoneManage() mid_=" << mid_ << " arg_npaq_= " << targ_npaq_  
       << " max_targ_npaq=" << max_targ_npaq << endl;
  */
  if (mid_ >= 0)  { 
    memgr.GetAuxData(mid_)->filltime_.SetNow(); 
    if (fgclean) memgr.FreeMemZone(mid_, MemZS_Free);
    else memgr.FreeMemZone(mid_, MemZS_Filled);
  }
  mmbuf_ = NULL;  targ_npaq_ = 0;  mid_ = -2;
  for (int fid=0;fid<(int)nbDma_ ;fid++) mmbufib_[fid]=NULL;
  if (fgclean)  return false;
  mid_ = memgr.FindMemZoneId(MemZA_Fill);
  mmbuf_ = memgr.GetMemZone(mid_);
  if (mmbuf_==NULL)   return true;
  for (int fid=0;fid<(int)nbDma_ ;fid++) mmbufib_[fid]=memgr.GetMemZone(mid_,fid);
  return false;
}

/*
bool PCIEMultiReader::MZoneManage(int zone,bool fgclean)    // Retourne true si probleme 
{
  // Pour debug 
  //cout << " PCIEReader::MZoneManage() mid_=" << mid_ << " arg_npaq_= " << targ_npaq_  
       << " max_targ_npaq=" << max_targ_npaq << endl;
  if (mid_[zone] >= 0)  memgr[zone]->FreeMemZone(mid_[zone], MemZS_Filled);
  mmbuf_[zone] = NULL;  targ_npaq_[zone] = 0;  mid_[zone] = -2;
  if (fgclean)  return false;
  mid_[zone] = memgr[zone]->FindMemZoneId(MemZA_Fill);
  mmbuf_[zone] = memgr[zone]->GetMemZone(mid_[zone]);
  if (mmbuf_[zone]==NULL) return true;
  return false;
}
*/

/* --Methode-- */
void PCIEMultiReader::Stop()
{ 
  // cout << " PCIEReaderChecker::stop()  ........ STOP" <<endl; 
  stop_ = true;

}


//--------------------------------------------------------------------
// Classe thread de sauvegarde sur fichiers avec gestion multifibres 
//--------------------------------------------------------------------
/*!
   \class MultiDataSaver
   \ingroup TAcq

   \brief Data saver in FITS format thread class for multiple BRPaquet sources (fibres)

   BRPaquets from each source (fibre) are dumped to disk in separate files in FITS format.
*/ 

/* --Methode-- */
MultiDataSaver::MultiDataSaver(RAcqMemZoneMgr& mem)
  :  memgr(mem)
{  
  BRAcqConfig bpar;
  nfiles_ = bpar.MaxNbFiles();
  nblocperfile_ = bpar.BlocPerFile();
  nmax_ = nblocperfile_*nfiles_;
  savesig_ = bpar.GetSaveFits();  // Si false, pas d'ecriture des fichiers FITS du signal       
  stop_ = false;        
}

/* --Methode-- */
void MultiDataSaver::Stop()
{
  // cout<< " MultiDataSaver:Stop ........ " << endl;
  stop_=true;
}

/* --Methode-- */
void MultiDataSaver::run()
{
  setRC(1);     
  BRPaqChecker pcheck[MAXNBFIB];  // Verification/comptage des paquets 
  BRAcqConfig bpar;
  try {
    TimeStamp ts; 
    cout << " MultiDataSaver::run() - Starting " << ts << " \n   NbFiles=" << nfiles_ << " NBloc/File=" 
         << nblocperfile_ << " NMaxMemZones=" << nmax_ << endl; 
    char fname[512];

    sprintf(fname,"%s/msaver.log",bpar.OutputDirectory().c_str());
    ofstream filog(fname);
    filog << " MultiDataSaver::run() - starting log file " << ts << " NFibres= " << memgr.NbFibres() << endl;                  
    filog << " NbFiles=" << nfiles_ << " NBloc/File="  << nblocperfile_ << " NMaxMemZones=" << nmax_ << endl;
        
    // Fichiers entete ascii et signal FITS
    ofstream header[MAXNBFIB];
    MiniFITSFile mff[MAXNBFIB]; 

    uint_4 fnum=0;
    uint_4 paqsz = memgr.PaqSize();
    cout << " ============================ MultiDataSaver::run() PaqSize " << paqsz <<endl;
    bool fgfirstfctt=false;
    bool fgnulldev = bpar.GetFileDevNull(); 
    if (fgnulldev) { 
      cout << " MultiDataSaver::run()/Warning /dev/null path specified, filenames=/dev/null" << endl;
      fgnulldev = true;
    }
    BRPaquet paq0(NULL, NULL, paqsz); 
    TimeStamp tsmz;
    for (uint_4 nbFile=0;nbFile<nfiles_ ;nbFile++) {
      if (stop_ )   break;
      if (memgr.GetRunState() == MemZR_Stopped) break;

      if (savesig_)
        for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
          if (fgnulldev) strcpy(fname,"/dev/null");
          else sprintf(fname,"%sHDRfits%d.txt",bpar.OutputDirectoryFib(fib).c_str(),fnum);
          header[fib].open(fname);
        }
    
      uint_4 npaqperfile = memgr.NbPaquets()*nblocperfile_;  // Nombre de paquets ecrits dans un fichier 

      if (savesig_) { //Reza - Ouverture conditionnel fichier 
        for(uint_4 fib=0; fib<memgr.NbFibres(); fib++)    {
          if (fgnulldev) strcpy(fname,"/dev/null");
          else sprintf(fname,"%ssignal%d.fits",bpar.OutputDirectoryFib(fib).c_str(),(int)fnum);
          //      cout << " ***DBG** Opening file " << fname << endl;
          mff[fib].Open(fname,MF_Write);  //Reza - Ouverture conditionnel fichier 
          // Entete correspondant a l'ecriture tout le paquet - trailer compris (modif Mai 2009)
          mff[fib].setDTypeNaxis(MF_Byte, paq0.PaquetSize(), npaqperfile);      
        // Sans TRAILER de paquet mff.setDTypeNaxis(MF_Byte, paq0.DataSize()+paq0.HeaderSize(), npaqperfile); 
        }
        fnum++;   fgfirstfctt=true;
      }
      else sprintf(fname,"MemDataBloc[%d]-NoDataFile",(int)fnum++);

      for (uint_4 kmz=0; kmz<nblocperfile_; kmz++) {
        if (stop_) break;
          //DBG cout << " MultiDataSaver::run()- nbFile=" << nbFile << " kmz=" << kmz << endl;  
        int mid = memgr.FindMemZoneId(MemZA_Save);  
        Byte* buffg = memgr.GetMemZone(mid);
        if (buffg == NULL) {
          cout << " MultiDataSaver::run()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
          setRC(21);    
          return;       
        }
        tsmz = memgr.GetAuxData(mid)->filltime_;
        for(uint_4 fib=0; fib<memgr.NbFibres(); fib++)  {  // Boucle sur les fibres 
          Byte* buff = memgr.GetMemZone(mid,fib);
          if (buff == NULL) {  // Ceci ne devrait pas arriver - suite au test buffg ci-dessus
            cout << " MultiDataSaver::run()/ERROR memgr.GetMemZone(" << mid << "," << fib << ") -> NULL" << endl;
            setRC(22);  
            return;     
          }
          for(uint_4 i=0; i<memgr.NbPaquets(); i++) {  // boucle sur les paquets 
            BRPaquet paq(NULL, buff+i*paqsz, paqsz);
            bool cpaqok=pcheck[fib].Check(paq);   // Verification du paquet / FrameCounter 
            if (cpaqok && fgfirstfctt) {
              framecnt_first_[fib] = pcheck[fib].LastFrameNum();
              timetag_first_[fib] = paq.TimeTag();
              AddFitsKWStart(mff,bpar,tsmz);
              fgfirstfctt=false;
            }
            if (savesig_)
              header[fib] << hex << paq.HDRMarker() << " " << paq.TRLMarker() << " " 
                          << paq.TimeTag2()<< " "<< paq.TimeTag1()<< " " 
                          << paq.FrameCounter() << " " << paq.PaqLen()  << endl;
            if (savesig_) // Reza - Ecriture conditionnel fichier fits signal 
              mff[fib].WriteB(paq.Header(),paq.PaquetSize()); // ecriture tout le paquet (modif Mai 2009)
          } // Fin de la boucle sur les paquets 
        }  // Fin de la boucle sur les fibres 
        memgr.FreeMemZone(mid, MemZS_Saved);
      }  // Boucle sur les blocs dans un meme fichier
      ts.SetNow(); 
      filog << ts << " : OK data files " << endl;
      cout << " MultiDataSaver::run() " << ts << " : OK data files  " << endl;
      for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
        if (savesig_) {
          if (fgnulldev) strcpy(fname,"/dev/null");
          else sprintf(fname,"%ssignal%d.fits",bpar.OutputDirectoryFib(fib).c_str(),(int)fnum-1);
        }
        else sprintf(fname,"MemDataBloc[%d]-NoDataFile",(int)fnum-1);
        string pcsum = pcheck[fib].Summary();
        filog << "    Fib " << fib << " -> " << fname << " Stat:" << pcsum << endl;
        cout  << "    Fib " << fib << " -> " << fname << " Stat:" << pcsum << endl;
      }
      if (savesig_)  {
        // Ajout mots-cle additionnels a tous les fichiers FITS
        for(uint_4 fib=0; fib<memgr.NbFibres(); fib++)
          framecnt_last_[fib] = pcheck[fib].LastFrameNum(); 

        AddFitsKWEnd(mff,bpar,tsmz);
        for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
          header[fib].close();
          mff[fib].Close();       
        }
      }
     
    }  // Fin de boucle sur les fichiers 
    cout << " --------------------  MultiDataSaver::run() -------------------- " << endl;
    for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
      cout << " MultiDataSaver/Summary Fib " << fib << endl;
      pcheck[fib].Print(cout);
      filog << " MultiDataSaver/Summary Fib " << fib << endl;
      pcheck[fib].Print(filog);
    }
    cout << " ---------------------------------------------------------- " << endl;
    ts.SetNow();
    filog << " MultiDataSaver::run() - End of processing/run() " << ts << endl;                

  }
  catch (MiniFITSException& exc) {
    cout << " MultiDataSaver::run()/catched MiniFITSException " << exc.Msg() << endl;
    setRC(3);   
    return; 
  }
  catch(...) {
    cout << " MultiDataSaver::run()/catched unknown ... exception " << endl;
    setRC(4);   
    return; 
  }
  setRC(0);
  return;
}   


/* --Methode-- */
int MultiDataSaver::AddFitsKWStart(MiniFITSFile* mff, BRAcqConfig& acpar, TimeStamp& ts)
{ 
  string cdtu=ts.ToString();
  string& skysrc=acpar.SkySource();
  bool hassrc=false;
  if (skysrc.length()>0)  hassrc=true;
  bool fgredpsz = acpar.GetParams().fgreducpsize;
  for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
    mff[fib].AddKeyS("DATEOBS", cdtu.c_str(), " Observation Time (YYYY-MM-DDThh:mm:ss UT) ");
    mff[fib].AddKeyS("TMSTART", cdtu.c_str(), " File Acqu. Start Time/Date  ");   
    mff[fib].AddKeyD("ACQVER", acpar.AcqVersion(), " BAORadio Acq Software version ") ;
    mff[fib].AddKeyS("ACQMODE", acpar.GetParams().AcqMode, " BAORadio Acq run  mode" );
    mff[fib].AddKeyS("BRPAQCFMT", BRPaquet::FmtConvToString(acpar.GetParams().GetDataConvFg()), 
                         " BAORadio BRPaquet DataFormatConversion" );
    mff[fib].AddKeyI("FIBERNUM", acpar.FiberNum(fib), " Fiber number") ;
    mff[fib].AddKeyI("FIBERID",memgr.FiberId(fib), " Fiber identifier (absolute id)"); 
    if (hassrc) 
      mff[fib].AddKeyS("SKYSOURC", skysrc, " Source identification" );
    if (fgredpsz) {
      mff[fib].AddKeyS("REDPSZMOD", BRPaquet::ReducActionToString(acpar.GetParams().pqreducmode),
                          "PaquetSize Reduction Mode") ; 
      mff[fib].AddKeyI("REDPSZOF", acpar.GetParams().reducoffset," PaquetSize Reduction Offset") ; 
    }
  }
  return 0;
}

/* --Methode-- */
int MultiDataSaver::AddFitsKWEnd(MiniFITSFile* mff, BRAcqConfig& acpar, TimeStamp& ts)
{ 
  string cdtu=ts.ToString();
  for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
    mff[fib].AddKeyS("TMEND", cdtu.c_str(), " File Acqu. End Time/Date  ");       
    mff[fib].AddKeyI("FCFIRST", framecnt_first_[fib], " First valid frame counter in file") ;
    mff[fib].AddKeyI("FCLAST", framecnt_last_[fib], " Last valid frame counter in file") ;
    mff[fib].AddKeyI("TTFIRST", timetag_first_[fib], " First valid timetag in file") ;
  }
  return 0;
}