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

Last change on this file since 4083 was 4016, checked in by ansari, 14 years ago

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