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

//----------------------------------------------------------------
// ---- 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
//-------------------------------------------------------

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
//-------------------------------------------------------

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
//----------------------------------------------------------------------------------------------------------

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
//----------------------------------------------------------------------------------------------------------

/* --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++) mmbufib_[fid]=NULL;
}

/* --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();
  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;
  
  int nerrdma = 0;
  int maxerrdma = 10;
  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];
  }
  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;      
  //      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();
    
    // 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 
        nerrdma ++;   fgbaddma=true;
        cout << "PCIEMultiReaderChecker/Erreur Waiting for datas ..." << endl;
        vec_pciw_[dma]->PrintStatus(cout);
        if (nerrdma>=maxerrdma) { 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 

    
  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 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;

  // //// Nettoyage final
  MZoneManage(true);
  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.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 
//--------------------------------------------------------------------

/* --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); 
    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); 
        }
        AddFitsKWStart(mff,bpar);
        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;       
        }
        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();
              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);
        }
        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);
        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.GetParams().FiberNum[fib], " Fiber number/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;
}