Changeset 3897 in Sophya for trunk/AddOn/TAcq/racqueth.cc

Timestamp:

Oct 4, 2010, 4:19:59 PM (15 years ago)

Author:

ansari

Message:

Amelioration des classes PCIEToEthernet et EthernetReader pour eviter des situations de blocage lorsque lecture avec ForceSameFrameCounter, Reza 04/10/2010

File:

• trunk/AddOn/TAcq/racqueth.cc (modified) (18 diffs)

trunk/AddOn/TAcq/racqueth.cc

@@ -26 +26 @@
 //----------------------------------------------------------------------------------------------------------
 
+// Si on veut avoir le protocole de controle (Hand-shake) entre l'emetteur et le recepteur,
+// decommenter la ligne suivante - Reza , 4 octobre 2010 
+// #define  BR_DO_HANDSHAKE  1
 
 /* --Methode-- */
@@ -47 +50 @@
   for(size_t i=0; i<vec_pciw_.size(); i++) {
     vector<ClientSocket> vskt;
+    vector<uint_8> verrcnt;
     for(size_t j=0; j<destname_.size(); j++) {
       ClientSocket sok(destname_[j], tcpportid_);
@@ -58 +62 @@
         throw SocketException("PCIEToEthernet:ERROR/  Connection to EthernetReader not established ");
       cout << " PCIEToEthernet: Ethernet connection established for DMA/fiber" << i << " with " << destname_[j] << endl;
-      vskt.push_back(sok);    
+      vskt.push_back(sok);
+      verrcnt.push_back(0);
     }
     vvec_skt_.push_back(vskt); 
+    vvec_errorcnt_.push_back(verrcnt);
     vec_cntpaq_.push_back(0);
   }
   totrdsnd_ = 0;
-  SetPrintLevel();
+  SetPrintLevel(par.prtlevel_,par.prtmodulo_);
 }
 
@@ -72 +78 @@
   for(size_t i=0; i<vec_pciw_.size(); i++) {
     vector<ClientSocket>& vskt = vvec_skt_[i];
-    for(size_t j=0; j<destname_.size(); j++) vskt[j].Close();
+    vector<uint_8>& verrcnt = vvec_errorcnt_[i];
+    for(size_t j=0; j<destname_.size(); j++) { 
+      cout << " ~PCIEToEthernet() closing socket, fiber/pcieNum=" << i << " ethernet_destNum=" << j 
+           << " ErrorCount=" << verrcnt[j] << endl;
+      vskt[j].Close();
+    }
   }
 }
@@ -154 +165 @@
     // Byte* nextdma = locdata+((kmz%memgr.NbZones())*(paqsz*memgr.NbPaquets()));
   uint_4 npaqfaitg = 0; 
-  uint_4 prtmod = par_.BlocPerFile()*par_.MMgrNbPaquet();
-  if (prtmod < 500) prtmod=500;
 
   while (npaqfaitg < nmaxpaq_) {  // Boucle global G 
@@ -179 +188 @@
     }
     if (fgbaddma) continue;  
-    if ((prtlev_>0)&&(npaqfaitg%prtmod==0)) 
+    if ((prtlev_>0)&&(npaqfaitg%prtmodulo_==0)) 
       cout << " PCIEToEthernet::run()/Info NPaqFait= " << npaqfaitg << endl;
     if (fgdirectsend_) {  // Pas de copie / reduction de taille de paquet, on rebalance tel quel ...
@@ -317 +326 @@
   //DBG    cout << " SendToTargets/DBG" << cnt_prt << " len=" << len << " data=" << hex << (unsigned long)data << dec << endl; cnt_prt++; 
   vector<ClientSocket>& vskt = vvec_skt_[fib];
+  vector<uint_8>& verrcnt = vvec_errorcnt_[fib];
+  char msg[BRTCPMSGLEN2];
   size_t rc=0;
   if (vec_cntpaq_[fib]%par_.MMgrNbPaquet() == 0) {
-    char msg[BRTCPMSGLEN2];
     sprintf(msg,"PCIEToEthernet-SendCnt %d",vec_cntpaq_[fib]);
     for(size_t j=0; j<vskt.size(); j++) {
       vskt[j].SendAll(msg, BRTCPMSGLEN2);
     }
-    if (vec_cntpaq_[fib]>0) {
-      for(size_t j=0; j<vskt.size(); j++) {
-        vskt[j].ReceiveAll(msg, BRTCPMSGLEN2);
-        msg[BRTCPMSGLEN2-1]='\0';
-        if (strncmp(msg,"EthernetReader-RecvCnt",22) != 0)
-          cout << " PCIEToEthernet::SendToTargets/ BAD HandShake message : " << msg << endl;
-      }
-    }
-  }
+  }
+  //  Envoi des donnees (paquets)
   for(size_t j=0; j<vskt.size(); j++) {
     rc += vskt[j].SendAll((const char *)data, len);
@@ -338 +341 @@
   }
   vec_cntpaq_[fib]++;
+#ifdef BR_DO_HANDSHAKE
+  // attente message de hand-shake 
+  if (vec_cntpaq_[fib]%par_.MMgrNbPaquet() == 0) {
+    for(size_t j=0; j<vskt.size(); j++) {
+      //DBG      cout << "SendToTargets/DEBUG-B attente -RecvCnt fib=" << fib << " npaq=" << vec_cntpaq_[fib] << endl;
+      vskt[j].ReceiveAll(msg, BRTCPMSGLEN2);
+      msg[BRTCPMSGLEN2-1]='\0';
+      if (strncmp(msg,"EthernetReader-RecvCnt",22) != 0) {
+        cout << " PCIEToEthernet::SendToTargets/ BAD HandShake, Fiber=" << fib 
+             << " NumSocket=" << j << " msg: " << msg << endl;
+        verrcnt[j]++;
+      }
+    }    
+  }
+#endif
+  if (vec_cntpaq_[fib]==nmaxpaq_) {  // fin de l'envoi pour cette fibre 
+    sprintf(msg,"PCIEToEthernet-Send-END %d",vec_cntpaq_[fib]);
+    for(size_t j=0; j<vskt.size(); j++) {
+      vskt[j].SendAll(msg, BRTCPMSGLEN2);
+    }
+    
+  }
   return rc;
 }
@@ -345 +370 @@
 
 /* --Methode-- */
-EthernetReader::EthernetReader(RAcqMemZoneMgr& mem, BRParList const& par, int portid, bool rdsamefc)
-  :  memgr_(mem), par_(par), stop_(false), rdsamefc_(rdsamefc), tcpportid_(portid)
+EthernetReader::EthernetReader(RAcqMemZoneMgr& mem, BRParList const& par, int portid)
+  :  memgr_(mem), par_(par), stop_(false), tcpportid_(portid)
 {
   totnbytesrd_ = 0;
   totnpaqrd_ = 0;
   totsamefc_ = 0;
+  totnbresync_ = 0;
+
   if (memgr_.NbFibres() > MAXANAFIB) 
     throw BAORadioException("EthernetReader::EthernetReader/ NbFibres>MAXANAFIB "); 
   if (par_.ethr_nlink != memgr_.NbFibres()) 
     throw BAORadioException("EthernetReader::EthernetReader/ NbFibres != ethr_nlink"); 
+  SetPrintLevel(par.prtlevel_,par.prtmodulo_);
+
+  SetReadMode();  // definitition du mode de lecture des liens 
 
   packsize_=memgr_.PaqSize();
@@ -369 +399 @@
     totnpqok_.push_back(0);    
   }
-  ServerSocket srv(tcpportid_, par_.ethr_nlink);
+  srv_sokp_ = new ServerSocket(tcpportid_, par_.ethr_nlink);
   char msg[BRTCPMSGLEN];
   for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
-    Socket sok=srv.WaitClientConnection();
+    Socket sok=srv_sokp_->WaitClientConnection();
     for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0';
     sok.ReceiveAll(msg,BRTCPMSGLEN);
@@ -379 +409 @@
     int ia,ib,ic,id,ie;
     sscanf(msg+25,"%d %d %d %d %d",&ia,&ib,&ic,&id,&ie);
-    if (((uint_4)ia!=par_.MMgrNbPaquet()) || ((uint_4)ib!=par_.MMgrPaquetSize())) {
+    //    if (((uint_4)ia!=par_.MMgrNbPaquet()) || ((uint_4)ib!=par_.MMgrPaquetSize())) { 
+    if (((uint_4)ia!=memgr_.NbPaquets()) || ((uint_4)ib!=memgr_.PaqSize())) { 
       strcpy(msg,"BAORadio-EthernetReader-BAD MMgrNbPaquet/MMgrPaquetSize()");
       sok.SendAll(msg,BRTCPMSGLEN);
@@ -390 +421 @@
     cout << " EthernetReader/Info connection/link" << fib << " established." << endl; 
     vec_cntpaq_.push_back(0);
-  }
-
-  SetPrintLevel();
-}
-
+    vec_fgsokend_.push_back(false);
+  }
+  gl_fgsokend = true;
+  dummybuff_ = new char[packsize_];
+}
+
+/* --Methode-- */
+EthernetReader::~EthernetReader()
+{
+  if (prtlev_>0) cout << " ~EthernetReader() , TotNPaqRd=" << totnpaqrd_ << " TotNbReSync=" << totnbresync_ << endl;
+  for(size_t j=0; j<vsok_.size(); j++) {
+    if (prtlev_>1)  
+      cout << " ~EthernetReader() closing sok[" << j << "] CntPaq=" << vec_cntpaq_[j] 
+           << " TotNPaqRd=" << totnpqrd_[j] << " TotNPaqOK=" << totnpqok_[j] << endl;
+    vsok_[j].Close();
+  }
+  srv_sokp_->Close();
+  delete srv_sokp_;
+  if (dummybuff_) delete[] dummybuff_;
+}
 
 /* --Methode-- */
@@ -466 +512 @@
     usleep(50000);       // Attente de traitement du dernier paquet
     memgr_.Stop();        // Arret 
-
+    CleanUpAllSockets();  // On lit tous les liens jusqu'a la reception du message END
+    cout << " EthernetReader::run(): done CleanUpAllSockets()" << endl;
   }  // Fin du bloc try 
   catch (std::exception& exc) {
@@ -485 +532 @@
 bool EthernetReader::ReadNextAllFibers()
 {
+  if (SkipAndSyncLinks()>8)  return true;
+ 
   for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)  {
     if (ReadNext(fib)) return true;  // probleme 
@@ -499 +548 @@
   //  On va essayer de lire jusqu'a avoir same_frame_counter 
   bool echec=true;
+  int nbechec=0;
   while (echec) {
+    if (nbechec>sfc_maxresync_)  { 
+      if (prtlev_>0)  
+        cout << " EthernetReader::ReadNextAllFibers()/ Stopping, reason NbEchec>(NbMax_Resync=" << sfc_maxresync_ << ")" << endl; 
+      return true;
+    }
     uint_8 cfc=curfc_[0];
     bool fgsamefc=true;
@@ -512 +567 @@
     }
     else {  // else !fgsame
+      if (nbechec>sfc_maxresync_)  return true;
+      if (nbechec>0)   {
+        int rcres=SkipAndSyncLinks();
+        if (rcres>8)  return true;
+        else if (rcres==1) {  // Il faut relire les fibres 
+          for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)  {
+            if (ReadNext(fib)) return true;  // probleme
+          }
+          fgsamefc=true;   cfc=curfc_[0];
+          for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
+            if (curfc_[fib]!=cfc) {
+              fgsamefc=false;
+              if (curfc_[fib] > cfc)  cfc=curfc_[fib]; 
+            }
+          }
+          if (fgsamefc) { 
+            totsamefc_++;  echec=false; return false;  // c'est OK  , same framecounter
+          }
+        }  // fin de if (rcres==1) 
+      }
+      
+      if (prtlev_>5) 
+        cout << " EthernetReader::ReadNextAllFibers()/DEBUG NbEchec=" << nbechec  
+             << " cfc=" << cfc << endl;
+      
       for(uint_4 fib=0; fib<memgr_.NbFibres(); fib++) {
-        while (curfc_[fib]<cfc) {
-          if (ReadNext(fib)) return true;  // probleme 
+        for(uint_4 ntryrd=0; ntryrd<sfc_maxdpc_; ntryrd++) {   // on tente un maximum de 
+          if (curfc_[fib]>=cfc) break; 
+          if (ReadNext(fib)) return true;  // probleme    
         }
       }
+      nbechec++;
     }   // fin de  else !fgsame 
   }  // fin de while(echec): lecture jusqu'a same_frame_counter 
@@ -529 +611 @@
   while(!fggood) { 
     ReceiveFromSocket(fib, (char *)vpaq_[fib].Begin(), packsize_);
+    if (vec_fgsokend_[fib])   {   // fin de reception sur le lien 
+      if (prtlev_>0)  
+        cout << " EthernetReader::ReadNext(" << fib << ")/ Stopping, reason EndReceive fgsokend_=" << vec_fgsokend_[fib] << endl; 
+      return true; 
+    }
     totnbytesrd_+=packsize_;
     totnpqrd_[fib]++;
@@ -538 +625 @@
 
 /* --Methode-- */
+void EthernetReader::CleanUpAllSockets()
+{
+  bool fgallfinished=false;
+  while (!fgallfinished) {
+    fgallfinished=true;
+    for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)  {
+      if (vec_fgsokend_[fib]>0) continue;
+      ReceiveFromSocket(fib,dummybuff_,packsize_);
+      fgallfinished=false;
+    }
+  }
+  return;
+}
+
+/* --Methode-- */
 size_t EthernetReader::ReceiveFromSocket(int fib, char* data, size_t len)
 {
   size_t rc =0;
-  if (vec_cntpaq_[fib]%par_.MMgrNbPaquet() == 0) {
+  if (vec_fgsokend_[fib]) { // la lecture est finie 
+    memset(data, 0, len);
+    return 0;
+  }
+
+  if (vec_cntpaq_[fib]%memgr_.NbPaquets() == 0) { 
     char msg[BRTCPMSGLEN2];
     msg[0]='\0';
     vsok_[fib].ReceiveAll(msg, BRTCPMSGLEN2);
-    if (strncmp(msg,"PCIEToEthernet-SendCnt",22) != 0) 
-      cout << " EthernetReader::ReceiveFromSocket/ BAD HandShake message : " << msg << endl;
-  }  
+    //    if (strncmp(msg,"PCIEToEthernet-SendCnt",22) != 0) 
+    //      cout << " EthernetReader::ReceiveFromSocket/ BAD HandShake message : " << msg << endl;
+    bool fgbadhsm=true;
+    if (strncmp(msg,"PCIEToEthernet-Send",19)==0) {
+      if (strncmp(msg+19,"Cnt",3)==0) fgbadhsm=false;
+      else if (strncmp(msg+19,"-END",4)==0) {
+        fgbadhsm=false;
+        vec_fgsokend_[fib]=1;  gl_fgsokend=true;
+        //DBG   cout << "ReceiveFromSocket/DEBUG -END msg received  " << endl;
+        if (prtlev_>2)  
+          cout << " EthernetReader::ReceiveFromSocket/ Receive on LinkNo" << fib << " Ended OK" << endl;
+      }
+    }
+    if (fgbadhsm) {
+      cout << " EthernetReader::ReceiveFromSocket/ BAD HandShake, LinkNo " <<  fib << " msg: " << msg << endl;
+      vec_fgsokend_[fib]=2; 
+    }
+  }
+
+  if (vec_fgsokend_[fib]>0) { // la lecture est finie 
+    memset(data, 0, len);
+    return 0;
+  }
+
   rc += vsok_[fib].ReceiveAll(data, len);
   vec_cntpaq_[fib]++;
 
-  /*     
-  size_t nblk = len/ethr_bsz_;
-  size_t fblk = len%ethr_bsz_;
-  size_t off = 0;
-  if (nblk>0) {
-    for(size_t i=0; i<nblk; i++)  {
-      rc += vsok_[fib].ReceiveAll(data+off, ethr_bsz_);
-      off += ethr_bsz_;  
-    }
-  }
-  if (fblk>0) {
-    rc += vsok_[fib].ReceiveAll(data+off, fblk);
-  } 
-  vec_cntpaq_[fib]++;
-  */
-
-  if (vec_cntpaq_[fib]%par_.MMgrNbPaquet() == 0) {
+#ifdef BR_DO_HANDSHAKE
+  // Envoi de message de hand-shake 
+  if (vec_cntpaq_[fib]%memgr_.NbPaquets() == 0) {
     char msg[BRTCPMSGLEN2];
     sprintf(msg,"EthernetReader-RecvCnt %d",vec_cntpaq_[fib]);
     vsok_[fib].SendAll(msg, BRTCPMSGLEN2);
-  }  
+    //DBG    cout << " ReceiveFromSocket/DEBUG-B msg=" << msg << endl;
+  }
+#endif  
   return rc;
+}
+
+/* --Methode-- */
+int EthernetReader::SkipAndSyncLinks()
+{
+  uint_8 minnumpaq=vec_cntpaq_[0];   // minimum des nombres de paquets lus sur les differents liens 
+  size_t minnp_fid=0;   // numero de fibre correspondant au minimum de paquets lus
+  uint_8 maxnumpaq=vec_cntpaq_[0];   // maximum des nombres de paquets lus sur les differents liens 
+  size_t maxnp_fid=0;   // numero de fibre correspondant au maximum de paquets lus
+  for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
+    if (vec_cntpaq_[fib]<=minnumpaq) { 
+      minnumpaq=vec_cntpaq_[fib];    minnp_fid=fib;
+    }
+    if (vec_cntpaq_[fib]>=maxnumpaq) { 
+      maxnumpaq=vec_cntpaq_[fib];    maxnp_fid=fib;
+    }
+  }
+  if (!rdsamefc_ && (minnumpaq!=maxnumpaq)) 
+    cout << " EthernetReader::SkipAndSyncLinks()/BUG or PROBLEM - ReadAllOK and min<>maxnumpaq=" 
+         << minnumpaq << "," << maxnumpaq << " !!!" << endl;
+  if ((maxnumpaq-minnumpaq)<sfc_maxdpc_)  return 0;
+
+  if (prtlev_>4) 
+    cout << " EthernetReader::SkipAndSyncLinks() min,maxnumpaq=" << minnumpaq << "," << maxnumpaq
+         << " min,maxnp_fid=" << minnp_fid << "," << maxnp_fid << " sfc_maxdpc_=" << sfc_maxdpc_ << endl;
+
+  for(size_t fib=0; fib<memgr_.NbFibres(); fib++)  {
+    while(vec_cntpaq_[fib]<maxnumpaq) {
+      if (vec_fgsokend_[fib]>0) return 9;
+      ReceiveFromSocket(fib,dummybuff_,packsize_);
+    }
+  }
+  totnbresync_++; 
+  if (prtlev_>3) {
+    cout << " EthernetReader::SkipAndSyncLinks() NbResync=" << totnbresync_ << " After Sync: vec_cntpaq_[fib]=";
+    for(size_t ii=0; ii<vec_cntpaq_.size(); ii++) cout << vec_cntpaq_[ii] << " , ";
+    cout << endl;
+  }
+
+  return 1;
 }