Changeset 3762 in Sophya for trunk/AddOn/TAcq

Timestamp:

May 2, 2010, 12:32:43 AM (15 years ago)

Author:

ansari

Message:

ajout de messages de hand-shake dans les echanges , Reza 01/05/2010

Location:

trunk/AddOn/TAcq

Files:

• racqueth.cc (modified) (11 diffs)
• racqueth.h (modified) (3 diffs)

trunk/AddOn/TAcq/racqueth.cc

@@ -41 +41 @@
   nbDma_= vec_pciw.size();
 
+ //  true -> direct transfer of data to ethernet
+  fgdirectsend_ = (par_.pci2eth_fgdirect && (swapall_==BR_Copy) ) ? true : false;  
   char msg[BRTCPMSGLEN];
   cout << "PCIEToEthernet/Info: Establishing TCP/IP connections ... " << endl;
@@ -48 +50 @@
       ClientSocket sok(destname_[j], tcpportid_);
       for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0';
-      sprintf(msg,"BAORadio-PCIEToEthernet %d %d",0,i);
+      uint_4 dmasz = (fgdirectsend_)  ? vec_pciw_[0]->TransferSize() : 0;
+      sprintf(msg,"BAORadio-PCIEToEthernet  %d %d %d %d %d", par_.MMgrNbPaquet(), par_.MMgrPaquetSize(), dmasz, 0,i);
+      //           123456789012345678901234567890
       sok.Send(msg,BRTCPMSGLEN);
       sok.Receive(msg,BRTCPMSGLEN);
@@ -56 +60 @@
       vskt.push_back(sok);    
     }
-    vvec_skt_.push_back(vskt);    
+    vvec_skt_.push_back(vskt); 
+    vec_cntpaq_.push_back(0);
   }
   totrdsnd_ = 0;
@@ -112 +117 @@
   }
 
-  bool fgdirectsend = (par_.pci2eth_fgdirect && (swapall_==BR_Copy) ) ? true : false;   //  true -> direct transfer of data to ethernet
-  if (fgdirectsend) 
+  if (fgdirectsend_) 
     cout << " PCIEToEthernet::run() Direct transfer mode DMA to Ethernet ... " << endl;
   bool fgredpaq=par_.fgreducpsize;
@@ -178 +182 @@
     if ((prtlev_>0)&&(npaqfaitg%prtmod==0)) 
       cout << " PCIEToEthernet::run()/Info NPaqFait= " << npaqfaitg << endl;
-    if (fgdirectsend) {  // Pas de copie / reduction de taille de paquet, on rebalance tel quel ...
+    if (fgdirectsend_) {  // Pas de copie / reduction de taille de paquet, on rebalance tel quel ...
       for (int fib=0; fib<(int) nbDma_ ;fib++) {
         //DBG cout << " DEBUG*B fib=" << fib << " Datas[fib]=" << hex << Datas[fib] << dec << endl; 
@@ -285 +289 @@
         cout << " kb" << endl;
         
-    if (!fgdirectsend) pcheck[dma].Print(cout);
+    if (!fgdirectsend_) pcheck[dma].Print(cout);
   }
   cout << " --------------------------------------------------------------------" << endl;
@@ -315 +319 @@
   vector<ClientSocket>& vskt = vvec_skt_[fib];
   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].Send(msg, BRTCPMSGLEN2);
+    }
+    if (vec_cntpaq_[fib]>0) {
+      for(size_t j=0; j<vskt.size(); j++) {
+        vskt[j].Receive(msg, BRTCPMSGLEN2);
+        msg[BRTCPMSGLEN2-1]='\0';
+        if (strncmp(msg,"EthernetReader-RecvCnt",22) != 0)
+          cout << " PCIEToEthernet::SendToTargets/ BAD HandShake message : " << msg << endl;
+      }
+    }
+  }
+  if (eths_bsz_<16) {
+    for(size_t j=0; j<vskt.size(); j++) {
+      rc += vskt[j].Send((const char *)data, len);
+      totrdsnd_ += eths_bsz_;
+    }
+    vec_cntpaq_[fib]++;
+    return rc;
+  }
   size_t nblk = len/eths_bsz_;
   size_t fblk = len%eths_bsz_;
@@ -333 +360 @@
     }
   }
+  vec_cntpaq_[fib]++;
   return rc;
 }
@@ -372 +400 @@
     if (strncmp(msg,"BAORadio-PCIEToEthernet",23)!=0) 
       throw SocketException("EthernetReader:ERROR/  Bad message from PCIEToEthernet client !");
+    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())) {
+      strcpy(msg,"BAORadio-EthernetReader-BAD MMgrNbPaquet/MMgrPaquetSize()");
+      sok.Send(msg,BRTCPMSGLEN);
+      throw SocketException("EthernetReader:ERROR/  Bad MMgrNbPaquet/MMgrPaquetSize() from PCIEToEthernet client !");
+    }
     for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0';
     strcpy(msg,"BAORadio-EthernetReader-OK");
@@ -377 +412 @@
     vsok_.push_back(sok);
     cout << " EthernetReader/Info connection/link" << fib << " established." << endl; 
+    vec_cntpaq_.push_back(0);
   }
 
@@ -510 +546 @@
 size_t EthernetReader::ReceiveFromSocket(int fib, char* data, size_t len)
 {
-  size_t nblk = len/ethr_bsz_;
-  size_t fblk = len%ethr_bsz_;
-  size_t off = 0;
   size_t rc =0;
-  if (nblk>0) {
-    for(size_t i=0; i<nblk; i++)  {
-      rc += vsok_[fib].Receive(data+off, ethr_bsz_);
-      off += ethr_bsz_;  
-    }
-  }
-  if (fblk>0) {
-    rc += vsok_[fib].Receive(data+off, fblk);
-  }
+  if (vec_cntpaq_[fib]%par_.MMgrNbPaquet() == 0) {
+    char msg[BRTCPMSGLEN2];
+    msg[0]='\0';
+    vsok_[fib].Receive(msg, BRTCPMSGLEN2);
+    if (strncmp(msg,"PCIEToEthernet-SendCnt",22) != 0) 
+      cout << " EthernetReader::ReceiveFromSocket/ BAD HandShake message : " << msg << endl;
+  }  
+
+  if (ethr_bsz_<16) {
+    rc += vsok_[fib].Receive(data, len);
+  }
+  else {
+    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].Receive(data+off, ethr_bsz_);
+        off += ethr_bsz_;  
+      }
+    }
+    if (fblk>0) {
+      rc += vsok_[fib].Receive(data+off, fblk);
+    } 
+  }
+  vec_cntpaq_[fib]++;
+
+  if (vec_cntpaq_[fib]%par_.MMgrNbPaquet() == 0) {
+    char msg[BRTCPMSGLEN2];
+    sprintf(msg,"EthernetReader-RecvCnt %d",vec_cntpaq_[fib]);
+    vsok_[fib].Send(msg, BRTCPMSGLEN2);
+  }  
   return rc;
 }

trunk/AddOn/TAcq/racqueth.h

@@ -40 +40 @@
 #define BRTCPMSGLEN 128
 #endif
+#ifndef BRTCPMSGLEN2
+#define BRTCPMSGLEN2 64
+#endif
 
 
@@ -71 +74 @@
   int tcpportid_;
   vector< vector<ClientSocket> > vvec_skt_;
+  vector< uint_4 > vec_cntpaq_; 
   uint_4 packSize_;
   uint_4 packSizeInMgr_;
   uint_4 nbDma_;
   uint_8 totrdsnd_;
+  bool fgdirectsend_;  //  true -> direct transfer of DMA data to ethernet
   size_t eths_bsz_;
   int prtlev_;
@@ -116 +121 @@
   int tcpportid_;
   vector< Socket > vsok_;  //  Les sockets de lecture pour chaque fibre/lien 
+  vector< uint_4 > vec_cntpaq_; 
   bool stop_;
   bool rdsamefc_;  // if true, read paquets with same value of FrameCounter on different fibers/link