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

//----------------------------------------------------------------
// ---- 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
//-------------------------------------------------------
// 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()
{
  struct sigaction act;
  struct sigaction old_act;
   //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);     

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

  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_.DoTransferData();
      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 << " DataProc::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();
    for (uint_4 nbFile=0;nbFile<nfiles_ ;nbFile++) {
      if (stop_ )   break;
      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 
        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 

          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;

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

  BRPaqChecker pcheck;  // Verification/comptage des paquets 

  Byte* Datas = NULL;
  Byte* locdata = new Byte[paqsz];
  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;
    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_.DoTransferData();
        
        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_);
            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;

}