Changeset 3897 in Sophya for trunk/AddOn

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

Location:

trunk/AddOn/TAcq

Files:

• acqparam_exemple.d (modified) (1 diff)
• brparam.cc (modified) (4 diffs)
• brparam.h (modified) (1 diff)
• mfacq.cc (modified) (2 diffs)
• racqueth.cc (modified) (18 diffs)
• racqueth.h (modified) (6 diffs)
• swrapsock.cc (modified) (1 diff)
• swrapsock.h (modified) (3 diffs)

trunk/AddOn/TAcq/acqparam_exemple.d

@@ -53 +53 @@
 #  En mode reception de paquets sur ethernet,  On peut forcer la lecture
 #   de paquets avec meme compteur de paquet (Force SameFrameCounter )
-#  @ethrforcesamefc
-ethrforcesamefc
+#  @ethrforcesamefc  MaxDiffNumPaq   MaxNbResync
+#  Le parametre MaxDiffNumPaq (50 par defaut) est la difference maximum entre nombre de 
+#  paquets lus sur les differents liens et MaxNbResync (def=50) est le nombre maximum de tentatives 
+#  de resynchronisation succesive
+ethrforcesamefc 50 50 
 ## No de port TCP/IP   
 #  @tcpportid 6912

trunk/AddOn/TAcq/brparam.cc

@@ -135 +135 @@
   tcpportid=BRSPORTID;
   pci2eth_fgdirect=false;
-  ethr_nlink = 0;
-  ethr_forcesamefc_ = false;
+  ethr_nlink=0;
+  ethr_forcesamefc_=false;
+  ethr_sfc_maxdpc_=50;
+  ethr_sfc_maxresync_=50;
 
   skysource="";
@@ -208 +210 @@
   ethr_nlink=p.ethr_nlink;
   ethr_forcesamefc_=p.ethr_forcesamefc_;
+  ethr_sfc_maxdpc_=p.ethr_sfc_maxdpc_;
+  ethr_sfc_maxresync_=p.ethr_sfc_maxresync_;
 
   skysource=p.skysource;
@@ -254 +258 @@
   }
   ethr_nlink=conf.IParam("ethrnlink",0,0);   // Nombre de sources de paquets en reception ethernet 
-  if (conf.HasKey("ethrforcesamefc"))   ethr_forcesamefc_=true;   // force SAME FrameCounter   on all links
+  if (conf.HasKey("ethrforcesamefc"))  {
+    ethr_forcesamefc_=true;   // force SAME FrameCounter   on all links
+    ethr_sfc_maxdpc_=conf.IParam("ethrforcesamefc",0,50);
+    ethr_sfc_maxresync_=conf.IParam("ethrforcesamefc",1,50);
+  }
 
   //  Parametre de controle du monitoring 
@@ -387 +395 @@
     cout << endl;
   }
-  cout << " TCP-PortId=" << tcpportid << " EthernetRead NbSources (=NbLinks)= " << ethr_nlink 
-       << ((ethr_forcesamefc_)?" ForceSameFrameCounter read mode":" AllOKPaquets read mode ") << endl;
+  cout << " TCP-PortId=" << tcpportid << " EthernetRead NbSources (=NbLinks)= " << ethr_nlink << endl;
+  if (ethr_forcesamefc_)
+    cout << "EthernetReader mode: ForceSameFrameCounter read mode with Max_PaquetCounterDiff= " 
+         << ethr_sfc_maxdpc_ << " MaxNbResync=" << ethr_sfc_maxresync_ << endl;
+  else
+    cout << "EthernetReader mode: AllOKPaquets read mode " << endl;
 
   if (fgdoProc>0) 

trunk/AddOn/TAcq/brparam.h

@@ -166 +166 @@
   int ethr_nlink;    // Nombre total de source d'envoi (= nb total de fibre de fibre) 
   bool ethr_forcesamefc_;   // true -> on force receptions de paquets avec SAME FrameCounter sur tous les liens 
-
+  uint_4 ethr_sfc_maxdpc_;   // valeur maximum de difference tolere entre compteurs de paquets de differentes fibres 
+  uint_4 ethr_sfc_maxresync_;   // Nombre maximum de tentative de resynchronisation avant echec
+ 
   // Identification source observee dans le ciel
   string skysource;

trunk/AddOn/TAcq/mfacq.cc

@@ -300 +300 @@
   cout <<"mfacq[2] Creating PCIEToEthernet  thread object " << endl;
   PCIEToEthernet pci2eth(vec_pciw, bpar.GetParams().GetEthTargets(), bpar.GetParams(), bpar.GetParams().tcpportid);
-  pci2eth.SetPrintLevel(1);
+  pci2eth.SetPrintLevel(acpar.prtlevel_, acpar.prtmodulo_); 
   //  usleep(200);  attente au cas ou ...
   pPcie2Eth=&pci2eth;
@@ -387 +387 @@
 
   cout << "mfacq[3] Creating   EthernetReader thread object     " << endl;
-  EthernetReader ethrdr(mmgr, bpar.GetParams(), bpar.GetParams().tcpportid, acpar.ethr_forcesamefc_);
+  EthernetReader ethrdr(mmgr, bpar.GetParams(), bpar.GetParams().tcpportid);
+  ethrdr.SetReadMode(acpar.ethr_forcesamefc_, acpar.ethr_sfc_maxdpc_,acpar.ethr_sfc_maxresync_);
   ethrdr.SetPrintLevel(acpar.prtlevel_, acpar.prtmodulo_); 
 

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

trunk/AddOn/TAcq/racqueth.h

@@ -65 +65 @@
   BRParList par_;   // Parametres divers d'acquisition 
 
-  uint_8 nmaxpaq_;  // Nombre maxi de paquets traites
+  uint_8 nmaxpaq_;  // Nombre maxi de paquets traites - DOIT etre mutiple de mmgr.NbPaquets()
   BRDataFmtConv swapall_;  // select data swap/format conversion for BRPaquet
   bool stop_;  
@@ -73 +73 @@
   int tcpportid_;
   vector< vector<ClientSocket> > vvec_skt_;
-  vector< uint_4 > vec_cntpaq_; 
+  vector< vector<uint_8> > vvec_errorcnt_;
+  vector< uint_8 > vec_cntpaq_; 
   uint_4 packSize_;
   uint_4 packSizeInMgr_;
@@ -89 +90 @@
 class EthernetReader : public ZThread {
 public:
-  EthernetReader(RAcqMemZoneMgr& mem, BRParList const& par, int portid=BRSPORTID, bool rdsamefc=false);
+  EthernetReader(RAcqMemZoneMgr& mem, BRParList const& par, int portid=BRSPORTID);
+  virtual ~EthernetReader(); 
+
+  // Configuration du mode de lecture des paquets 
+  // Si force_samefc=true, on essaye de lire des paquets avec meme FrameCounter, sans depasser une 
+  // difference maximum de maxdiff_paqnum entre nombre de paquets lus sur chaque fibre
+  inline void SetReadMode(bool force_samefc=false, uint_4 maxdiff_paqnum=50, uint_4 maxresync=50)
+  { rdsamefc_=force_samefc; sfc_maxdpc_=maxdiff_paqnum; sfc_maxresync_=maxresync; }
+
   virtual void run();
   inline void Stop() { stop_ = true; }
@@ -100 +109 @@
   bool ReadNextAllFibers();      // Renvoie true si probleme 
   bool ReadNext(int fib);   // Renvoie true si probleme 
+  void CleanUpAllSockets();  // pour finir la lecture de toutes les fibres 
   size_t ReceiveFromSocket(int fib, char* data, size_t len);
+
+// recale les links en cas d'ecart important entre nb de paquets lu sur chaque lien
+  int SkipAndSyncLinks();   // Renvoie 0 si rien fait, 1 si skip fait, 9 si probleme  
 
   // Permet d'avancer d'un paquet dans la zone - renvoie true si probleme
@@ -119 +132 @@
   BRParList par_;   // Parametres divers d'acquisition 
   int tcpportid_;
-  vector< Socket > vsok_;  //  Les sockets de lecture pour chaque fibre/lien 
-  vector< uint_4 > vec_cntpaq_; 
+  ServerSocket* srv_sokp_;  // Le socket serveur  
+  vector< Socket > vsok_;  //  Les sockets de lecture pour chaque lien ethernet (1 lien= 1 fibre a l'envoi) 
+  vector< uint_8 > vec_cntpaq_;   // vecteur de compteurs de paquet ( 1 compteur / lien ) 
+  vector< uint_4 > vec_fgsokend_;    // vecteur de flag, >0 : fin de reception (ou ereur) sur le lien 
+  bool gl_fgsokend;          // true -> au moins l'un de liens a termine 
   bool stop_;
   bool rdsamefc_;  // if true, read paquets with same value of FrameCounter on different fibers/link 
+  uint_4 sfc_maxdpc_;   // difference maxi entre nombre de paquets lu sur les liens 
+  uint_4 sfc_maxresync_;  // nb maximum de tentatives de resynchronisation 
+
   uint_8 totnbytesrd_;
   uint_8 totnpaqrd_;
   uint_8 totsamefc_;   // nombre total de paquets avec meme framecounter  
+
+  uint_8 totnbresync_;   // nombre total de saut+resynchronisation 
 
   vector<BRPaquet> vpaq_; 
@@ -140 +161 @@
   Byte* mmbufib_[MAXANAFIB];  // Pointeurs zone memoire de chaque fibre rendu par RAcqMemZoneMgr 
 
+  // zone tampon contenant zero si lecture sur fibre est fini 
+  char* dummybuff_;
+
   int prtlev_;   // print level 
   uint_8 prtmodulo_;   // print periodicity (modulo) 

trunk/AddOn/TAcq/swrapsock.cc

@@ -214 +214 @@
 
   return Socket(s); 
+}
+
+
+/* --Methode-- */
+int ServerSocket::Close()
+{
+  if (skt<0) return 0;
+  int rc=-1;
+  /*   il ne faut apparemment pas faire shutdown sur un socket serveur (non connected - ENOTCONN ) 
+  rc = shutdown(skt, SHUT_RDWR);
+  if(rc < 0) 
+    cout << "Socket::Close() Erreur: Pb shutdown() ErrNo="
+         << errno << endl;
+  */
+  rc = close(skt);
+  if(rc < 0) 
+    cout << "ServerSocket::Close() Erreur: Pb close() ErrNo="
+         << errno << endl;
+  skt = -1;
+  return(0);
 }
 

trunk/AddOn/TAcq/swrapsock.h

@@ -37 +37 @@
   Socket(int s=-1);
   Socket(Socket const & a);
-  ~Socket();
+  virtual ~Socket();
 
   size_t Send(const char * buff, size_t len, int flag=0);
@@ -44 +44 @@
   size_t ReceiveAll(char * buff, size_t len);
 
-  int Close();
+  virtual int Close();
   
   inline Socket& operator=(Socket const & a)
@@ -82 +82 @@
   ServerSocket(int port, int nconmax);
   Socket WaitClientConnection();
+  virtual int Close();  // redefinition de la methode de Socket::Close()
 
 protected: