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

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

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