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

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

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