source: Sophya/trunk/AddOn/TAcq/racqueth.cc@ 3911

Last change on this file since 3911 was 3911, checked in by ansari, 15 years ago

debug/amelioration print pour les modifs pour preparation Pittsburgh-Nov2010, Reza 21/11/2010
File size: 32.6 KB
Line 
1//----------------------------------------------------------------
2// ---- classes de threads pour lecture (transfert DMA)
3// et ecriture disque pour acquisition BAORadio -----
4// LAL - R. Ansari - Juin/Juillet 2008
5//----------------------------------------------------------------
6
7#include "racqueth.h"
8
9#include <stdlib.h>
10#include <unistd.h>
11#include <fstream>
12#include <signal.h>
13#include "pexceptions.h"
14#include "ctimer.h"
15#include "timestamp.h"
16
17#include "pciewrap.h"
18#include "brpaqu.h"
19
20#include "resusage.h" // Pour mesure temps elapsed/CPU ...
21#include <sys/time.h> // pour gettimeofday
22
23////////////////////////////////////////////////////////////////////////////////////////////////////////
24//----------------------------------------------------------------------------------------------------------
25// Classe thread de lecture PCI-Express et recopie sur interface reseau (Ethernet)
26//----------------------------------------------------------------------------------------------------------
27
28// Si on veut avoir le protocole de controle (Hand-shake) entre l'emetteur et le recepteur,
29// decommenter la ligne suivante - Reza , 4 octobre 2010
30// #define BR_DO_HANDSHAKE 1
31
32/* --Methode-- */
33PCIEToEthernet::PCIEToEthernet(vector<PCIEWrapperInterface*> vec_pciw, vector<string>& destname, BRParList const& par, int portid)
34 : par_(par), vec_pciw_ (vec_pciw), destname_(destname), tcpportid_(portid)
35{
36 nmaxpaq_ = par_.MaxNbPaquets();
37 swapall_ = par_.GetDataConvFg(); // select data swap/format conversion for BRPaquet
38 stop_ = false;
39 packSize_ = par_.RecvPaquetSize();
40 packSizeInMgr_=par_.MMgrPaquetSize();
41 sizeFr_=par_.DMASizeBytes();
42 if (vec_pciw.size() > MAXNBFIB)
43 throw ParmError("PCIEToEthernet:ERROR/ vec_pciw.size() > MAXNBFIB ");
44 nbDma_= vec_pciw.size();
45
46 // true -> direct transfer of data to ethernet
47 fgdirectsend_ = (par_.pci2eth_fgdirect && (swapall_==BR_Copy) ) ? true : false;
48 char msg[BRTCPMSGLEN];
49 cout << "PCIEToEthernet/Info: Establishing TCP/IP connections ... " << endl;
50 for(size_t i=0; i<vec_pciw_.size(); i++) {
51 vector<ClientSocket> vskt;
52 vector<uint_8> verrcnt;
53 for(size_t j=0; j<destname_.size(); j++) {
54 ClientSocket sok(destname_[j], tcpportid_);
55 for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0';
56 uint_4 dmasz = (fgdirectsend_) ? vec_pciw_[0]->TransferSize() : 0;
57 sprintf(msg,"BAORadio-PCIEToEthernet %d %d %d %d %d", par_.MMgrNbPaquet(), par_.MMgrPaquetSize(),
58 vec_pciw_[i]->FiberId(), dmasz, i);
59 // 123456789012345678901234567890
60 sok.SendAll(msg,BRTCPMSGLEN);
61 sok.ReceiveAll(msg,BRTCPMSGLEN);
62 if (strncmp(msg,"BAORadio-EthernetReader-OK",26)!=0) {
63 msg[BRTCPMSGLEN-1]='\0';
64 cout << "PCIEToEthernet:ERROR/ bad Init_Acknowledge message from EthernetReader : \n " << msg << endl;
65 usleep(10000);
66 throw SocketException("PCIEToEthernet:ERROR/ Connection to EthernetReader not established ");
67 }
68 cout << " PCIEToEthernet: Ethernet connection established for DMA/fiber" << i << " with " << destname_[j] << endl;
69 vskt.push_back(sok);
70 verrcnt.push_back(0);
71 }
72 vvec_skt_.push_back(vskt);
73 vvec_errorcnt_.push_back(verrcnt);
74 vec_cntpaq_.push_back(0);
75 vfgmsgfin_.push_back(false);
76
77 }
78 totrdsnd_ = 0;
79 stopreason_="??Unknown??";
80 SetPrintLevel(par.prtlevel_,par.prtmodulo_);
81}
82
83/* --Methode-- */
84PCIEToEthernet::~PCIEToEthernet()
85{
86 for(size_t i=0; i<vec_pciw_.size(); i++) {
87 vector<ClientSocket>& vskt = vvec_skt_[i];
88 vector<uint_8>& verrcnt = vvec_errorcnt_[i];
89 for(size_t j=0; j<destname_.size(); j++) {
90 cout << " ~PCIEToEthernet() closing socket, fiber/pcieNum=" << i << " ethernet_destNum=" << j
91 << " ErrorCount=" << verrcnt[j] << endl;
92 vskt[j].Close();
93 }
94 }
95}
96
97/* --Methode-- */
98void PCIEToEthernet::run()
99{
100
101 struct timeval tv1,tv2;
102 gettimeofday(&tv1, NULL);
103
104 cout << " PCIEToEthernet::run() - Starting , NMaxMemZones=" << par_.MaxNbBlocs()
105 << " MMgrNbPaquet=" << par_.MMgrNbPaquet() << " MMgr.PaqSiz=" << par_.MMgrPaquetSize()
106 << " DMA-Paqsize " << packSize_ << " " << BRPaquet::FmtConvToString(swapall_) << endl;
107 setRC(1);
108
109 // sigaddset(&act.sa_mask,SIGINT); // pour proteger le transfert DMA
110 //sigaction(SIGINT,&act,NULL);
111 uint_4 paqszmm = par_.MMgrPaquetSize();
112 uint_4 paqsz = packSize_;
113 uint_4 dmasz = vec_pciw_[0]->TransferSize();
114 //DEL vec_pciw_[0]->StartTransfers();
115
116 BRPaqChecker pcheck[MAXNBFIB]; // Verification/comptage des paquets
117 Byte* Datas[MAXNBFIB];
118 Byte* tampon[MAXNBFIB] ;
119 Byte* nexpaq[MAXNBFIB] ; // Pour recevevoir les paquets apres reduction de taille
120
121 Byte* predtampon=NULL; // tampon de recopie pour la reduction des tailles de paquets
122 Byte* nextpaq=NULL;
123 uint_4 off_acheval=0;
124
125 bool fgarret = false;
126
127 // Initialisation des tampons pour recopie des paquets a cheval pour chaque DMA
128 for (int i=0;i< (int)nbDma_ ;i++) {
129 // cout << " DEBUG*AB paqsz=" << paqsz << " paqszmm=" << paqszmm << endl;
130 tampon[i]=nexpaq[i]=NULL;
131 tampon[i] = new Byte[paqsz];
132 nexpaq[i] = new Byte[paqszmm];
133 }
134
135 if (fgdirectsend_)
136 cout << " PCIEToEthernet::run() Direct transfer mode DMA to Ethernet ... " << endl;
137 bool fgredpaq=par_.fgreducpsize;
138 if (fgredpaq) {
139 cout << " PCIEToEthernet::run() - PaquetSizeReduction - RedSize=" << par_.redpqsize
140 << " Offset=" << par_.reducoffset << " " << ((par_.reducneedcopy)?"NeedCopy":"NOCopy")
141 << " " << BRPaquet::ReducActionToString(par_.pqreducmode) << endl;
142 predtampon = new Byte[paqsz];
143 }
144
145
146 uint_8 trnbytes[MAXNBFIB];
147 uint_4 npaqfait[MAXNBFIB];
148 for (int i=0;i< (int)nbDma_ ;i++) {
149 npaqfait[i]=0; trnbytes[i]=0;
150 }
151 // Attente du message GO des taches lecteurs
152 cout << " PCIEToEthernet::run() Waiting for GO message from EthernetReader's ..." << endl;
153 char msg[BRTCPMSGLEN];
154 for(size_t i=0; i<vec_pciw_.size(); i++) {
155 vector<ClientSocket>& vskt=vvec_skt_[i];
156 for(size_t j=0; j<vskt.size(); j++) {
157 for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0';
158 vskt[j].ReceiveAll(msg,BRTCPMSGLEN);
159 if (strncmp(msg,"BAORadio-EthernetReader-Ready-GO",32)!=0) {
160 msg[BRTCPMSGLEN-1]='\0';
161 cout << "PCIEToEthernet:ERROR BAD Go for fiber=" << i << " dest=" << j << " msg: " << msg << endl;
162 throw SocketException("PCIEToEthernet:ERROR/ BAD Go message from EthernetReader ");
163 }
164 cout << " PCIEToEthernet::run() Received GO message for Fiber/DMA " << i << " from " << destname_[j] << endl;
165 }
166 }
167
168 cout << " PCIEToEthernet::run() Starting DMA to Ethernet transfers - NbFibers=NbDMA=" << vec_pciw_.size()
169 << " x NbEthDestinations=" << destname_.size() << endl;
170
171 // Byte* nextdma = locdata+((kmz%memgr.NbZones())*(paqsz*memgr.NbPaquets()));
172 uint_4 npaqfaitg = 0;
173
174 for (int i=0;i< (int)nbDma_ ;i++)
175 vec_pciw_[i]->SetMaxWaitEndDMA(par_.first_maxkwedma_,par_.first_nretrydma_);
176
177 while (npaqfaitg < nmaxpaq_) { // Boucle global G
178 if (fgarret) break;
179 if (stop_) break;
180
181 if ((prtlev_>0)&&(npaqfaitg>1)&&((npaqfaitg-1)%prtmodulo_==0))
182 cout << " PCIEToEthernet::run()/Info NPaqFait= " << npaqfaitg << endl;
183
184 // Lancement des DMA
185 for (int dma=0; dma < (int)nbDma_ ;dma++) vec_pciw_[dma]->StartTransfers();
186 if (npaqfaitg==1) {
187 for (int i=0;i< (int)nbDma_ ;i++)
188 vec_pciw_[i]->SetMaxWaitEndDMA(par_.maxkwedma_,par_.nretrydma_);
189 }
190
191 // On pointe vers le debut de la zone a remplir aver le prochain DMA
192 //-- Zone memoire locale Byte* nextdma = buff+i*paqsz;
193
194 bool fgbaddma=false;
195 // On boucle sur les nbDma_ en attente de leurs terminaison
196 for (int dma=0; dma <(int) nbDma_ ;dma++) {
197 Datas[dma]=vec_pciw_[dma]->GetData();
198 if (Datas[dma] == NULL) { // No data Read in DMA
199 fgbaddma=true;
200 cout << "PCIEToEthernetChecker/ERROR - DMA failed !" << endl;
201 vec_pciw_[dma]->PrintStatus(cout);
202 stopreason_="--Failed DMA--";
203 fgarret = true; break;
204 }
205 }
206 if (fgbaddma) continue;
207
208 if (fgdirectsend_) { // Pas de copie / reduction de taille de paquet, on rebalance tel quel ...
209 for (int fib=0; fib<(int) nbDma_ ;fib++) {
210 //DBG cout << " DEBUG*B fib=" << fib << " Datas[fib]=" << hex << Datas[fib] << dec << endl;
211 SendToTargets(fib, Datas[fib], dmasz);
212 trnbytes[fib] += (uint_8)dmasz;
213 npaqfait[fib] += (trnbytes[fib]/(uint_8)paqszmm);
214 if (fib==nbDma_-1) npaqfaitg += (trnbytes[fib]/(uint_8)paqszmm);
215 trnbytes[fib] = trnbytes[fib]%(uint_8)paqszmm;
216 }
217 continue; // On bypasse le reste
218 }
219
220 // Si on arrive ici, c'est qu'il faut copier / reduire les paquets ....
221 uint_4 curoff=0;
222 //1- On traite le paquet a cheval, rempli partiellement avec le DMA d'avant si necessaire pour les n fibres
223 if (off_acheval > 0) { // IF Numero B
224 if ((paqsz-off_acheval)< dmasz) { // IF Numero A
225 for(uint_4 fib=0; fib<nbDma_; fib++)
226 memcpy((void *)((tampon[fib])+off_acheval), (void *)Datas[fib], paqsz-off_acheval);
227 curoff = paqsz-off_acheval; off_acheval = 0;
228 for(uint_4 fib=0; fib<nbDma_; fib++) {
229 // CHECK S'il faut faire une reduction de taille de paquet
230 if (fgredpaq) { // reduction taille de paquet
231 if (par_.reducneedcopy) {
232 BRPaquet paqc1(tampon[fib], predtampon, paqsz, swapall_);
233 BRPaquet paqc2(nexpaq[fib], par_.redpqsize);
234 paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
235 }
236 else {
237 BRPaquet paqc1(tampon[fib], paqsz);
238 BRPaquet paqc2(nexpaq[fib], par_.redpqsize);
239 paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
240 }
241 }
242 else {
243 BRPaquet paqc(tampon[fib], nexpaq[fib], paqsz, swapall_);
244 }
245 BRPaquet paq(nexpaq[fib], packSizeInMgr_);
246 SendToTargets(fib, nexpaq[fib], packSizeInMgr_);
247 npaqfait[fib]++;
248 if (fib==nbDma_-1) npaqfaitg++; // Ne pas oublier le compteur de paquets faits
249 pcheck[fib].Check(paq); // Verification du paquet / FrameCounter
250 }
251 }
252 else { // se rapporte au IF numero A
253 for(uint_4 fib=0; fib<nbDma_; fib++)
254 memcpy((void *)(tampon[fib]+off_acheval), (void *)Datas[fib], dmasz);
255 curoff =dmasz;
256 off_acheval = (dmasz+off_acheval);
257 }
258 } // Fin IF Numero B
259
260 //2- On traite les paquets complets qui se trouvent dans la zone du DMA
261 while ((curoff+paqsz)<=dmasz) { // while numero C
262 // if ((dma==nbDma_-1)&&(npaqfait >= nmax_* memgr.NbPaquets())) break;
263 // CHECK S'il faut faire une reduction de taille de paquet
264 for(uint_4 fib=0; fib<nbDma_; fib++) {
265 if (fgredpaq) { // reduction taille de paquet
266 if (par_.reducneedcopy) {
267 BRPaquet paqc1(Datas[fib]+curoff, predtampon, paqsz, swapall_);
268 BRPaquet paqc2(nexpaq[fib], par_.redpqsize);
269 paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
270 }
271 else {
272 BRPaquet paqc1(Datas[fib]+curoff, paqsz);
273 BRPaquet paqc2(nexpaq[fib], par_.redpqsize);
274 paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
275 }
276 }
277 else {
278 BRPaquet paqc(Datas[fib]+curoff, nexpaq[fib], paqsz, swapall_);
279 }
280 BRPaquet paq(nexpaq[fib], packSizeInMgr_);
281 SendToTargets(fib, nexpaq[fib], packSizeInMgr_);
282 npaqfait[fib]++;
283 if (fib==nbDma_-1) npaqfaitg++; // Ne pas oublier le compteur de paquets faits
284 pcheck[fib].Check(paq); // Verification du paquet / FrameCounter
285 }
286 curoff += paqsz; // On avance l'index dans le buffer du DMA
287 } // -- FIN traitement des paquets complets ds un DMA - FIN du while numero C
288 //3- On copie si besoin la fin du DMA dans la zone tampon
289 if (curoff < dmasz) { // IF numero D
290 off_acheval = dmasz-curoff;
291 for(uint_4 fib=0; fib<nbDma_; fib++)
292 memcpy(tampon[fib], (void*)(Datas[fib]+curoff), off_acheval);
293 // ne sert a rien curoff += off_acheval;
294 } // FIN du if numero D
295 } // FIN Boucle global G
296
297 if (npaqfaitg >= nmaxpaq_) stopreason_="--Max Nb paquets reached--";
298
299 // ----- Nettoyage final et impression de resume -------
300 CleanUpEndSendAllLinks();
301 setRC(0);
302 gettimeofday(&tv2, NULL);
303 double tmelaps2 = (tv2.tv_sec-tv1.tv_sec)*1000.+(tv2.tv_usec-tv1.tv_usec)/1000.;
304 if (tmelaps2<0.1) tmelaps2=0.1;
305 cout << " ---------- PCIEToEthernet::run()-End StopReason: " << stopreason_ << endl
306 << " Summary NPaqFait=" << npaqfaitg
307 << " TotSend (kb)=" << totrdsnd_/1024 << "------ " << endl;
308 for (int dma=0; dma < (int)nbDma_ ;dma++) {
309 cout << " --Fib=" << dma << " NPaqFait=" << npaqfait[dma] << " TotDMATransfer="
310 << vec_pciw_[dma]->TotTransferBytes()/1024
311 << " ElapsTime=" << tmelaps2 << " ms ->"
312 << (double)vec_pciw_[dma]->TotTransferBytes()/tmelaps2 << " kb/s" << endl;
313
314 vector<ClientSocket>& vskt = vvec_skt_[dma];
315 cout << " TotEthTransfer=";
316 for(size_t j=0; j<vskt.size(); j++)
317 cout << vskt[j].NBytesSent()/1024 << " , ";
318 cout << " kb" << endl;
319
320 if (!fgdirectsend_) pcheck[dma].Print(cout);
321 }
322 cout << " --------------------------------------------------------------------" << endl;
323
324 for (int i=0;i< (int)nbDma_ ;i++) {
325 delete[] tampon[i];
326 delete[] nexpaq[i];
327 }
328 if ((fgredpaq)&&predtampon) delete[] predtampon;
329
330 //DBG cout << " fin thread ========================" <<endl;
331 return;
332}
333
334
335/* --Methode-- */
336void PCIEToEthernet::Stop()
337{
338 // cout << " PCIEReaderChecker::stop() ........ STOP" <<endl;
339 stop_ = true;
340
341}
342
343static long cnt_prt=0;
344/* --Methode-- */
345size_t PCIEToEthernet::SendToTargets(int fib, Byte* data, size_t len, bool fgfin)
346{
347 //DBG cout << " SendToTargets/DBG" << cnt_prt << " len=" << len << " data=" << hex << (unsigned long)data << dec << endl; cnt_prt++;
348 vector<ClientSocket>& vskt = vvec_skt_[fib];
349 vector<uint_8>& verrcnt = vvec_errorcnt_[fib];
350 char msg[BRTCPMSGLEN2];
351 size_t rc=0;
352 if (vec_cntpaq_[fib]%par_.MMgrNbPaquet() == 0) {
353 sprintf(msg,"PCIEToEthernet-SendCnt %d",vec_cntpaq_[fib]);
354 for(size_t j=0; j<vskt.size(); j++) {
355 vskt[j].SendAll(msg, BRTCPMSGLEN2);
356 }
357 }
358 // Envoi des donnees (paquets)
359 for(size_t j=0; j<vskt.size(); j++) {
360 rc += vskt[j].SendAll((const char *)data, len);
361 totrdsnd_ += len;
362 }
363 vec_cntpaq_[fib]++;
364#ifdef BR_DO_HANDSHAKE
365 // attente message de hand-shake
366 if (vec_cntpaq_[fib]%par_.MMgrNbPaquet() == 0) {
367 for(size_t j=0; j<vskt.size(); j++) {
368 //DBG cout << "SendToTargets/DEBUG-B attente -RecvCnt fib=" << fib << " npaq=" << vec_cntpaq_[fib] << endl;
369 vskt[j].ReceiveAll(msg, BRTCPMSGLEN2);
370 msg[BRTCPMSGLEN2-1]='\0';
371 if (strncmp(msg,"EthernetReader-RecvCnt",22) != 0) {
372 cout << " PCIEToEthernet::SendToTargets/ BAD HandShake, Fiber=" << fib
373 << " NumSocket=" << j << " msg: " << msg << endl;
374 verrcnt[j]++;
375 }
376 }
377 }
378#endif
379 if ((vec_cntpaq_[fib]==nmaxpaq_)||(fgfin&&(vec_cntpaq_[fib]%par_.MMgrNbPaquet()==0))) { // fin de l'envoi pour cette fibre
380 sprintf(msg,"PCIEToEthernet-Send-END %d",vec_cntpaq_[fib]);
381 for(size_t j=0; j<vskt.size(); j++) {
382 vskt[j].SendAll(msg, BRTCPMSGLEN2);
383 }
384 vfgmsgfin_[fib]=true;
385 }
386 return rc;
387}
388
389/* --Methode-- */
390void PCIEToEthernet::CleanUpEndSendAllLinks()
391{
392 cout << " PCIEToEthernet::CleanUpEndSendAllLinks() - Ending transmission on links ..." << endl;
393 Byte* tampon = new Byte[par_.MMgrPaquetSize()];
394 memset(tampon,0,(size_t)par_.MMgrPaquetSize());
395 for(size_t fib=0; fib<vfgmsgfin_.size(); fib++) {
396 while (!vfgmsgfin_[fib]) SendToTargets(fib,tampon,par_.MMgrPaquetSize(),true);
397 cout << " ... Ended transmission on links for Fiber " << fib << endl;
398 }
399 delete[] tampon;
400}
401
402//------------------------------------------------------------------
403// Classe thread de lecture sur interface reseau (Ethernet)
404//------------------------------------------------------------------
405
406/* --Methode-- */
407EthernetReader::EthernetReader(RAcqMemZoneMgr& mem, BRParList const& par, int portid)
408 : memgr_(mem), par_(par), stop_(false), tcpportid_(portid)
409{
410 totnbytesrd_ = 0;
411 totnpaqrd_ = 0;
412 totsamefc_ = 0;
413 totnbresync_ = 0;
414
415 if (memgr_.NbFibres() > MAXANAFIB)
416 throw BAORadioException("EthernetReader::EthernetReader/ NbFibres>MAXANAFIB ");
417 if (par_.ethr_nlink != memgr_.NbFibres())
418 throw BAORadioException("EthernetReader::EthernetReader/ NbFibres != ethr_nlink");
419 SetPrintLevel(par.prtlevel_,par.prtmodulo_);
420
421 SetReadMode(); // definitition du mode de lecture des liens
422 WaitENDMsg4Terminate();
423
424 packsize_=memgr_.PaqSize();
425 mid_=-2;
426 mmbuf_=NULL;
427 max_targ_npaq = memgr_.NbPaquets();
428 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) mmbufib_[fib]=NULL;
429
430 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
431 vpaq_.push_back(BRPaquet(NULL,memgr_.PaqSize()));
432 vpchk_.push_back(BRPaqChecker(true,0));
433 curfc_.push_back(0);
434 totnpqrd_.push_back(0);
435 totnpqok_.push_back(0);
436 }
437 srv_sokp_ = new ServerSocket(tcpportid_, par_.ethr_nlink);
438 char msg[BRTCPMSGLEN];
439 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
440 Socket sok=srv_sokp_->WaitClientConnection();
441 for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0';
442 sok.ReceiveAll(msg,BRTCPMSGLEN);
443 if (strncmp(msg,"BAORadio-PCIEToEthernet",23)!=0) {
444 cout << " EthernetReader:ERROR, bad init message from PCIEToEthernet client -> throw SocketException" << endl;
445 throw SocketException("EthernetReader:ERROR/ Bad message from PCIEToEthernet client !");
446 }
447 int ia,ib,ic,id,ie;
448 sscanf(msg+25,"%d %d %d %d %d",&ia,&ib,&ic,&id,&ie);
449 // if (((uint_4)ia!=par_.MMgrNbPaquet()) || ((uint_4)ib!=par_.MMgrPaquetSize())) {
450 if (((uint_4)ia!=memgr_.NbPaquets()) || ((uint_4)ib!=memgr_.PaqSize())) {
451 strcpy(msg,"BAORadio-EthernetReader-BAD MMgrNbPaquet/MMgrPaquetSize()");
452 sok.SendAll(msg,BRTCPMSGLEN);
453 cout << " EthernetReader:ERROR, Bad client.MMgrNbPaquet/PaquetSize NbPaq=" << ia << " PaqSz=" << ib
454 << " Reader_PaqSz=" << memgr_.PaqSize() << endl;
455 usleep(10000);
456 throw SocketException("EthernetReader:ERROR/ Bad MMgrNbPaquet/MMgrPaquetSize() from PCIEToEthernet client !");
457 }
458 for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0';
459 strcpy(msg,"BAORadio-EthernetReader-OK");
460 sok.SendAll(msg,BRTCPMSGLEN);
461 vsok_.push_back(sok);
462 vfibid_.push_back(ic);
463 cout << " EthernetReader/Info connection/link" << fib << " established - FiberId= " << ic << endl;
464 vec_cntpaq_.push_back(0);
465 vec_fgsokend_.push_back(false);
466 }
467 if (vsok_.size()>1) {
468 cout << " EthernetReader/Info: sorting socket according to FiberIds ..." << endl;
469 bool encore=true;
470 Socket tmpskt;
471 int tmpid;
472 while(encore) {
473 encore=false;
474 for(size_t ii=0; ii<vsok_.size()-1; ii++) {
475 if (vfibid_[ii]>vfibid_[ii+1]) {
476 tmpskt=vsok_[ii]; tmpid=vfibid_[ii];
477 vsok_[ii]=vsok_[ii+1]; vfibid_[ii]=vfibid_[ii+1];
478 vsok_[ii+1]=tmpskt; vfibid_[ii+1]=tmpid;
479 encore=true;
480 }
481 }
482 }
483 }
484 cout << " EthernetReader/Info: Setting FiberIds in RAcqMemZoneMgr \n ... FiberIds= " ;
485 for(size_t ii=0; ii<vsok_.size(); ii++) {
486 memgr_.FiberId(ii)=vfibid_[ii];
487 cout << vfibid_[ii] << " ; ";
488 }
489 cout << endl;
490
491 gl_fgsokend = true;
492 stopreason_="??Unknown??";
493 dummybuff_ = new char[packsize_];
494}
495
496/* --Methode-- */
497EthernetReader::~EthernetReader()
498{
499 if (prtlev_>0) cout << " ~EthernetReader() , TotNPaqRd=" << totnpaqrd_ << " TotNbReSync=" << totnbresync_ << endl;
500 for(size_t j=0; j<vsok_.size(); j++) {
501 if (prtlev_>1)
502 cout << " ~EthernetReader() closing sok[" << j << "] CntPaq=" << vec_cntpaq_[j]
503 << " TotNPaqRd=" << totnpqrd_[j] << " TotNPaqOK=" << totnpqok_[j] << endl;
504 vsok_[j].Close();
505 }
506 srv_sokp_->Close();
507 delete srv_sokp_;
508 if (dummybuff_) delete[] dummybuff_;
509}
510
511/* --Methode-- */
512void EthernetReader::run()
513{
514 setRC(1);
515 try {
516 TimeStamp ts;
517 Timer tm("EthernetReader", false);
518 cout << " EthernetReader::run() - Starting " << ts << " NbFibres()=" << memgr_.NbFibres()
519 << " PaqSize() = " << memgr_.PaqSize() << endl;
520 cout << " ...ReadMode: " << ((rdsamefc_)?"Paquets With SameFrameCounter":"All OK paquets") << endl;
521
522 cout << " Sending GO message to all PCIEToEthernet clients ... " << endl;
523 char msg[BRTCPMSGLEN];
524 for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0';
525 strcpy(msg,"BAORadio-EthernetReader-Ready-GO");
526 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
527 cout << " EthernetReader/Debug - Sending GO message to sender " << fib << endl;
528 vsok_[fib].SendAll(msg,BRTCPMSGLEN);
529 }
530 Byte* nextpaq=NULL;
531 bool fgok=true;
532 while (fgok&&(totnpaqrd_<par_.MaxNbPaquets())) {
533 if (stop_) {
534 if (prtlev_>0)
535 cout << ">S>S>S> EthernetReader::run() / Stopping, through Stop() method" << endl;
536 stopreason_="EthernetReader::run() / Stopping, through Stop() method";
537 break;
538 }
539 if ( MoveToNextTarget() ) {
540 cout << "EthernetReader::run()/Error-A- MoveToNextTarget() returned true ->STOP 9" << endl;
541 setRC(7); fgok=false; break;
542 }
543 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
544 nextpaq=GetPaquetTarget(fib);
545 if (nextpaq == NULL) { // Cela ne devrait pas arriver
546 cout << "EthernetReader::run()/Error-A2- GetPaquetTarget(fib) returned NULL ->STOP 9" << endl;
547 setRC(9); fgok=false; break;
548 }
549 vpaq_[fib].Set(nextpaq);
550 }
551 if (ReadNextAllFibers()) { fgok=false; break; }
552 totnpaqrd_++;
553 if ((prtlev_>0)&&(totnpaqrd_%prtmodulo_==0)) {
554 cout << "EthernetReader: NbPaq/Link=" << totnpaqrd_ << " NSameFC="
555 << totsamefc_ << " / NPaqFib0Read=" << totnpqrd_[0]
556 << " FracSameFC=" << 100*totsamefc_/totnpqrd_[0] << " %" << endl;
557 }
558 if (totnpaqrd_>=par_.MaxNbPaquets()) {
559 if (prtlev_>0)
560 cout << ">S>S>S> EthernetReader::run() / Stopping, totnpaqrd_>=par_.MaxNbPaquets()" << endl;
561 stopreason_="EthernetReader::run() / Stopping, totnpaqrd_>=par_.MaxNbPaquets()";
562 }
563 }
564
565 MoveToNextTarget(); // Pour faire traiter le dernier paquet si plein
566 MZoneManage(true); // Nettoyage final
567
568 cout << " ---------------------------------------------------------------------------- " << endl;
569 cout << "--- EthernetReader::run() END, StopReason: " << stopreason_ << " ---" << endl;
570 ts.SetNow();
571 tm.SplitQ();
572 cout << " END reading TotNPaq/Link=" << totnpaqrd_ << " :" << ts << endl;
573 cout << "... NbPaq/Link=" << totnpaqrd_ << " NSameFC="
574 << totsamefc_ << " / NPaqLink[0]_Read=" << totnpqrd_[0]
575 << " FracSameFC=" << 100*totsamefc_/totnpqrd_[0] << " %" << endl;
576
577 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
578 int perc=0;
579 if (totnpqrd_[fib]>0) perc=100*totsamefc_/totnpqrd_[fib];
580 cout << " Fiber" << fib << " TotNPaqRd=" << totnpqrd_[fib] << " TotNPaqOK=" << totnpqok_[fib]
581 << " FracSameFC=" << perc << " %" << endl;
582 if (prtlev_ > 0) vpchk_[fib].Print(cout);
583 }
584 cout << " TotalEthernetRead= " << totnbytesrd_/(1024*1024) << " MBytes Ethernet-Read rate= "
585 << (double)(totnbytesrd_)/1024./tm.PartialElapsedTimems() << " MB/s" << endl;
586 cout << " EthernetReader::run()/Timing: \n";
587 tm.Print();
588 cout << " ---------------------------------------------------------------------------- " << endl;
589
590 usleep(50000); // Attente de traitement du dernier paquet
591 memgr_.Stop(); // Arret
592 if (waitendmsg_) {
593 CleanUpAllSockets(); // On lit tous les liens jusqu'a la reception du message END
594 cout << " EthernetReader::run(): done CleanUpAllSockets()" << endl;
595 }
596 } // Fin du bloc try
597 catch (std::exception& exc) {
598 cout << " EthernetReader::run()/catched execption msg= " << exc.what() << endl;
599 setRC(3);
600 return;
601 }
602 catch(...) {
603 cout << " EthernetReader::run()/catched unknown ... exception " << endl;
604 setRC(4);
605 return;
606 }
607 setRC(0);
608 return;
609}
610
611/* --Methode-- */
612bool EthernetReader::ReadNextAllFibers()
613{
614 if (sfc_maxresync_>0) return ReadNextAllFibersWithSync();
615
616 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
617 if (ReadNext(fib)) return true; // probleme
618 }
619 if (!rdsamefc_ || (memgr_.NbFibres()<2)) {
620 uint_8 cfc=curfc_[0];
621 bool fgsamefc=true;
622 for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
623 if (curfc_[fib]!=cfc) fgsamefc=false;
624 }
625 if (fgsamefc) totsamefc_++;
626 return false; // c'est OK
627 }
628 // On va essayer de lire jusqu'a avoir same_frame_counter
629 uint_8 cfc=curfc_[0];
630 bool fgsamefc=true;
631 for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
632 if (curfc_[fib]!=cfc) {
633 fgsamefc=false;
634 if (curfc_[fib] > cfc) cfc=curfc_[fib];
635 }
636 }
637 if (fgsamefc) {
638 totsamefc_++; return false; // c'est OK , same framecounter
639 }
640 else { // else !fgsame
641 bool encorerd=true;
642 uint_4 ntrd=0;
643 while(encorerd) {
644 encorerd=false;
645 for(uint_4 fib=0; fib<memgr_.NbFibres(); fib++) {
646 if (curfc_[fib]>=cfc) continue;
647 if (ReadNext(fib)) return true; // probleme
648 }
649 ntrd++;
650 bool fgsamefc2=true;
651 cfc=curfc_[0];
652 for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
653 if (curfc_[fib]!=cfc) {
654 fgsamefc2=false;
655 if (curfc_[fib]>cfc) cfc=curfc_[fib];
656 }
657 }
658 if (fgsamefc2) {
659 totsamefc_++; return false; // c'est OK , same framecounter
660 }
661 encorerd=true;
662 if ((sfc_maxdpc_>0)&&(ntrd>sfc_maxdpc_)) {
663 if (prtlev_>0)
664 cout << ">S>S>S> EthernetReader::ReadNextAllFibers()/ Stopping, Nb_ReadNext="
665 << ntrd << " > sfc_maxdpc_" << endl;
666 stopreason_="ReadNextAllFibers() StopReason: failed to get SameFC Nb_ReadNext>sfc_maxdpc_";
667 return true;
668 }
669 } // Fin de while encorerd
670
671 } // fin de else !fgsame
672
673 return true; // probleme
674}
675
676/* --Methode-- */
677bool EthernetReader::ReadNext(int fib)
678{
679 bool fggood=false;
680 while(!fggood) {
681 ReceiveFromSocket(fib, (char *)vpaq_[fib].Begin(), packsize_);
682 if (vec_fgsokend_[fib]) { // fin de reception sur le lien
683 if (prtlev_>0)
684 cout << ">S>S>S> EthernetReader::ReadNext(" << fib << ")/ Stopping, reason EndReceive fgsokend_="
685 << vec_fgsokend_[fib] << endl;
686 stopreason_="ReadNext() StopReason: EndReceive signaled on link";
687 return true;
688 }
689 totnbytesrd_+=packsize_;
690 totnpqrd_[fib]++;
691 fggood = vpchk_[fib].Check(vpaq_[fib],curfc_[fib]);
692 }
693 totnpqok_[fib]++;
694 return false;
695}
696
697/* --Methode-- */
698void EthernetReader::CleanUpAllSockets()
699{
700 bool fgallfinished=false;
701 while (!fgallfinished) {
702 fgallfinished=true;
703 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
704 if (vec_fgsokend_[fib]>0) continue;
705 ReceiveFromSocket(fib,dummybuff_,packsize_);
706 fgallfinished=false;
707 }
708 }
709 return;
710}
711
712/* --Methode-- */
713size_t EthernetReader::ReceiveFromSocket(int fib, char* data, size_t len)
714{
715 size_t rc =0;
716 if (vec_fgsokend_[fib]) { // la lecture est finie
717 memset(data, 0, len);
718 return 0;
719 }
720
721 if (vec_cntpaq_[fib]%memgr_.NbPaquets() == 0) {
722 char msg[BRTCPMSGLEN2];
723 msg[0]='\0';
724 vsok_[fib].ReceiveAll(msg, BRTCPMSGLEN2);
725 // if (strncmp(msg,"PCIEToEthernet-SendCnt",22) != 0)
726 // cout << " EthernetReader::ReceiveFromSocket/ BAD HandShake message : " << msg << endl;
727 bool fgbadhsm=true;
728 if (strncmp(msg,"PCIEToEthernet-Send",19)==0) {
729 if (strncmp(msg+19,"Cnt",3)==0) fgbadhsm=false;
730 else if (strncmp(msg+19,"-END",4)==0) {
731 fgbadhsm=false;
732 vec_fgsokend_[fib]=1; gl_fgsokend=true;
733 //DBG cout << "ReceiveFromSocket/DEBUG -END msg received " << endl;
734 if (prtlev_>2)
735 cout << " EthernetReader::ReceiveFromSocket/ Receive on LinkNo" << fib << " Ended OK" << endl;
736 }
737 }
738 if (fgbadhsm) {
739 cout << " EthernetReader::ReceiveFromSocket/ BAD HandShake, LinkNo " << fib << " msg: " << msg << endl;
740 vec_fgsokend_[fib]=2;
741 }
742 }
743
744 if (vec_fgsokend_[fib]>0) { // la lecture est finie
745 memset(data, 0, len);
746 return 0;
747 }
748
749 rc += vsok_[fib].ReceiveAll(data, len);
750 vec_cntpaq_[fib]++;
751
752#ifdef BR_DO_HANDSHAKE
753 // Envoi de message de hand-shake
754 if (vec_cntpaq_[fib]%memgr_.NbPaquets() == 0) {
755 char msg[BRTCPMSGLEN2];
756 sprintf(msg,"EthernetReader-RecvCnt %d",vec_cntpaq_[fib]);
757 vsok_[fib].SendAll(msg, BRTCPMSGLEN2);
758 //DBG cout << " ReceiveFromSocket/DEBUG-B msg=" << msg << endl;
759 }
760#endif
761 return rc;
762}
763
764/* --Methode-- */
765bool EthernetReader::ReadNextAllFibersWithSync()
766{
767 if (rdsamefc_&&(sfc_maxresync_>0)) {
768 if (SkipAndSyncLinks()>8) return true;
769 }
770 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
771 if (ReadNext(fib)) return true; // probleme
772 }
773 if (!rdsamefc_ || (memgr_.NbFibres()<2)) {
774 uint_8 cfc=curfc_[0];
775 bool fgsamefc=true;
776 for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
777 if (curfc_[fib]!=cfc) fgsamefc=false;
778 }
779 if (fgsamefc) totsamefc_++;
780 return false; // c'est OK
781 }
782 // On va essayer de lire jusqu'a avoir same_frame_counter
783 bool echec=true;
784 int nbechec=0;
785 while (echec) {
786 if (nbechec>sfc_maxresync_) {
787 if (prtlev_>0)
788 cout << ">S>S>S> EthernetReader::ReadNextAllFibersWithSync()/ Stopping, reason NbEchec>(NbMax_Resync="
789 << sfc_maxresync_ << ")" << endl;
790 stopreason_="ReadNextAllFibersWithSync() , StopReason: NbEchec>NbMax_Resyn";
791 return true;
792 }
793 uint_8 cfc=curfc_[0];
794 bool fgsamefc=true;
795 for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
796 if (curfc_[fib]!=cfc) {
797 fgsamefc=false;
798 if (curfc_[fib] > cfc) cfc=curfc_[fib];
799 }
800 }
801 if (fgsamefc) {
802 totsamefc_++; echec=false; return false; // c'est OK , same framecounter
803 }
804 else { // else !fgsame
805 if (nbechec>0) {
806 int rcres=SkipAndSyncLinks();
807 if (rcres>8) return true;
808 else if (rcres==1) { // Il faut relire les fibres
809 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
810 if (ReadNext(fib)) return true; // probleme
811 }
812 fgsamefc=true; cfc=curfc_[0];
813 for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
814 if (curfc_[fib]!=cfc) {
815 fgsamefc=false;
816 if (curfc_[fib] > cfc) cfc=curfc_[fib];
817 }
818 }
819 if (fgsamefc) {
820 totsamefc_++; echec=false; return false; // c'est OK , same framecounter
821 }
822 } // fin de if (rcres==1)
823 }
824
825 if (prtlev_>5)
826 cout << " EthernetReader::ReadNextAllFibersWithSync()/DEBUG NbEchec=" << nbechec
827 << " cfc=" << cfc << endl;
828
829 for(uint_4 fib=0; fib<memgr_.NbFibres(); fib++) {
830 for(uint_4 ntryrd=0; ntryrd<sfc_maxdpc_; ntryrd++) { // on tente un maximum de
831 if (curfc_[fib]>=cfc) break;
832 if (ReadNext(fib)) return true; // probleme
833 }
834 }
835
836 nbechec++;
837 } // fin de else !fgsame
838 } // fin de while(echec): lecture jusqu'a same_frame_counter
839
840 return true; // probleme
841}
842
843/* --Methode-- */
844int EthernetReader::SkipAndSyncLinks()
845{
846 uint_8 minnumpaq=vec_cntpaq_[0]; // minimum des nombres de paquets lus sur les differents liens
847 size_t minnp_fid=0; // numero de fibre correspondant au minimum de paquets lus
848 uint_8 maxnumpaq=vec_cntpaq_[0]; // maximum des nombres de paquets lus sur les differents liens
849 size_t maxnp_fid=0; // numero de fibre correspondant au maximum de paquets lus
850 for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
851 if (vec_cntpaq_[fib]<=minnumpaq) {
852 minnumpaq=vec_cntpaq_[fib]; minnp_fid=fib;
853 }
854 if (vec_cntpaq_[fib]>=maxnumpaq) {
855 maxnumpaq=vec_cntpaq_[fib]; maxnp_fid=fib;
856 }
857 }
858 if (!rdsamefc_ && (minnumpaq!=maxnumpaq))
859 cout << " EthernetReader::SkipAndSyncLinks()/BUG or PROBLEM - ReadAllOK and min<>maxnumpaq="
860 << minnumpaq << "," << maxnumpaq << " !!!" << endl;
861 if ((maxnumpaq-minnumpaq)<sfc_maxdpc_) return 0;
862
863 if (prtlev_>4)
864 cout << " EthernetReader::SkipAndSyncLinks() min,maxnumpaq=" << minnumpaq << "," << maxnumpaq
865 << " min,maxnp_fid=" << minnp_fid << "," << maxnp_fid << " sfc_maxdpc_=" << sfc_maxdpc_ << endl;
866
867 for(size_t fib=0; fib<memgr_.NbFibres(); fib++) {
868 while(vec_cntpaq_[fib]<maxnumpaq) {
869 if (vec_fgsokend_[fib]>0) return 9;
870 ReceiveFromSocket(fib,dummybuff_,packsize_);
871 }
872 }
873 totnbresync_++;
874 if (prtlev_>3) {
875 cout << " EthernetReader::SkipAndSyncLinks() NbResync=" << totnbresync_ << " After Sync: vec_cntpaq_[fib]=";
876 for(size_t ii=0; ii<vec_cntpaq_.size(); ii++) cout << vec_cntpaq_[ii] << " , ";
877 cout << endl;
878 }
879
880 return 1;
881}
882
883/* --Methode-- */
884bool EthernetReader::MZoneManage(bool fgclean) // Retourne true si probleme
885{
886 /* Pour debug
887 cout << " EthernetReader::MZoneManage() mid_=" << mid_ << " arg_npaq_= " << targ_npaq_
888 << " max_targ_npaq=" << max_targ_npaq << endl;
889 */
890 if (mid_ >= 0) {
891 memgr_.GetAuxData(mid_)->filltime_.SetNow();
892 if (fgclean) memgr_.FreeMemZone(mid_, MemZS_Free);
893 else memgr_.FreeMemZone(mid_, MemZS_Filled);
894 }
895 mmbuf_ = NULL; targ_npaq_ = 0; mid_ = -2;
896 for (int fib=0;fib<(int)memgr_.NbFibres() ;fib++) mmbufib_[fib]=NULL;
897 if (fgclean) return false;
898 mid_ = memgr_.FindMemZoneId(MemZA_Fill);
899 mmbuf_ = memgr_.GetMemZone(mid_);
900 if (mmbuf_==NULL) return true;
901 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)
902 mmbufib_[fib]=memgr_.GetMemZone(mid_,fib);
903 return false;
904}
Note: See TracBrowser for help on using the repository browser.