Changeset 3658 in Sophya for trunk/AddOn/TAcq/racqurw.cc

Timestamp:

Oct 18, 2009, 11:10:33 AM (16 years ago)

Author:

ansari

Message:

1/ Gestion multi-fibre ds RAcqMemZoneMgr et les DMAReader/DiskWriter (brproc.cc)
2/ Possibilite d'ajout de mot cle ds l'entete FITS par la classe MiniFITSFile

Reza , 18/10/2009

File:

• trunk/AddOn/TAcq/racqurw.cc (modified) (9 diffs)

trunk/AddOn/TAcq/racqurw.cc

@@ -59 +59 @@
 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;
@@ -70 +68 @@
   setRC(1);     
 
-  uint_4 cpt=0;
+ 
   // sigaddset(&act.sa_mask,SIGINT);  // pour proteger le transfert DMA
   //sigaction(SIGINT,&act,NULL);  
@@ -232 +230 @@
     uint_4 fnum=0;
     uint_4 paqsz = memgr.PaqSize();
+    cout << " ============================ DataSaver::run() PaqSize " << paqsz <<endl;
     bool fgnulldev = false;
     if (path_ == "/dev/null") { 
@@ -253 +252 @@
         // 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); 
       }
@@ -353 +353 @@
   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);        
@@ -363 +363 @@
 
   Byte* Datas = NULL;
-  Byte* locdata = new Byte[paqsz];
+  Byte* locdata = new Byte[paqsz*memgr.NbPaquets()*memgr.NbZones()];
   Byte* tampon = new Byte[paqsz];
 
@@ -375 +375 @@
     if (fgarret) break;
 
-    Byte* nextdma = locdata;
+    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
@@ -419 +419 @@
           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, 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 
@@ -457 +458 @@
 
 }
+////////////////////////////////////////////////////////////////////////////////////////////////////////
+//----------------------------------------------------------------------------------------------------------
+// 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,uint_4 sizeFrame,uint_4 packSize ,RAcqMemZoneMgr& mem, uint_4 nmax, BRDataFmtConv swapall,int binMin, int nbBin)
+  :   memgr(mem) , vec_pciw_ (vec_pciw)
+{
+  nmax_ = nmax; 
+  swapall_ = swapall;   // select data swap/format conversion for BRPaquet
+  stop_ = false;
+  packSize_ = packSize;
+  packSizeInMgr_=memgr.PaqSize();
+  sizeFr_ =sizeFrame;
+  binMin_= binMin;
+  nbBin_= nbBin;
+  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()=" << memgr.NbPaquets() << "Paqsize " << packSize_<< endl; 
+  setRC(1);     
+
+  // sigaddset(&act.sa_mask,SIGINT);  // pour proteger le transfert DMA
+  //sigaction(SIGINT,&act,NULL);        
+  // uint_4 paqsz = memgr[0]->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* 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];
+  }
+ 
+  ofstream header[MAXNBFIB];
+  for(uint_4 fib=0; fib<nbDma_; fib++) {
+    char hfnm[128];
+    sprintf(hfnm, "./HDRCountPaqs%d.txt", fib);
+    header[fib].open(hfnm);
+  }
+   
+  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;
+          }
+          BRPaquet paq(tampon[fib], nextpaq, paqsz, swapall_,binMin_,nbBin_);
+          npaqfait[fib]++;
+          if (fib==nbDma_-1) npaqfaitg++;  // Ne pas oublier le compteur de paquets faits 
+          pcheck[fib].Check(paq);   // Verification du paquet / FrameCounter 
+          header[fib] << dec << paq.FrameCounter()<< endl; ;
+        }
+      }
+      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;
+        }
+        BRPaquet paq(Datas[fib]+curoff, nextpaq, paqsz, swapall_,binMin_,nbBin_);
+        npaqfait[fib]++;
+        if (fib==nbDma_-1) npaqfaitg++;  // Ne pas oublier le compteur de paquets faits 
+        pcheck[fib].Check(paq);   // Verification du paquet / FrameCounter 
+        header[fib] << dec << paq.FrameCounter()<< endl; ;
+      }
+      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(uint_4 fib=0; fib<nbDma_; fib++)  header[fib].close();
+  for (int i=0;i< (int)nbDma_ ;i++) delete[] tampon[i]; 
+  //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 
+//--------------------------------------------------------------------
+
+MultiDataSaver::MultiDataSaver(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 MultiDataSaver::Stop()
+{
+  // cout<< " MultiDataSaver:Stop ........ " << endl;
+  stop_=true;
+
+}
+void MultiDataSaver::run()
+{
+  setRC(1);     
+  BRPaqChecker pcheck[MAXNBFIB];  // Verification/comptage des paquets 
+
+  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",path_.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 fgnulldev = false;
+    if (path_ == "/dev/null") { 
+      cout << " MultiDataSaver::run()/Warning /dev/null path specified, filenames=/dev/null" << endl;
+      fgnulldev = true;
+    }
+    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,"%s/Fibre%d/HDRfits%d.txt",path_.c_str(),fib+1,fnum);
+          header[fib].open(fname);
+        }
+      BRPaquet paq0(NULL, NULL, paqsz); 
+      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,"%s/Fibre%d/signal%d.fits",path_.c_str(),fib+1,(int)fnum);
+          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);
+        }
+        fnum++;
+        // 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 << " 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(2);     
+          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(2);   
+            return;     
+          }
+          for(uint_4 i=0; i<memgr.NbPaquets(); i++) {  // boucle sur les paquets 
+            BRPaquet paq(NULL, buff+i*paqsz, paqsz);
+            pcheck[fib].Check(paq);   // Verification du paquet / FrameCounter 
+            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,"%s/Fibre%d/signal%d.fits",path_.c_str(),fib+1,(int)fnum-1);
+        }
+        else sprintf(fname,"MemDataBloc[%d]-NoDataFile",(int)fnum-1);
+        filog << "    Fib " << fib << " -> " << fname << endl;
+        cout << "    Fib " << fib << " -> " << fname << endl;
+      }
+      if (savesig_)
+        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;
+}   
+