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

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

1/ Mise en place d'informations auxiliaires au niveau de RAcqMemZoneMgr ,

FillTime (DATEOBS) pour chaque zone et identificateur des fibres

2/ remplissage de FillTime(DATEOBS) Et FiberId's de MemZoneMgr par

les classes PCIEMultiReader et EthernetReader

3/ Ajout de mots-cle (ds les datacards) pour numeros d'identification des

fibres et parametres de controles du DMA (@fiberids @waitenddma)
et corrections/ameliorations des modes acq (raw1c ordfft2c ...)

4/ Parametrage des WaitEndDMA ds PCIEWrapperInterface et implementation

au niveau de DMAMgrV6 (dmamgrv6.h .cc)

5/ Gestion d'echec DMA -> arret acquisition ds PCIEMultiReader,EthernetReader

modification TimeOut DMA entre le premier DMA et les suivants pour permettre
un arret propre rapide

6/ Adaptation de la classe BRVisibilityCalculator pour utilisation

FillTime (DATEOBS) et FiberIds de MemZoneMgr - amelioration de
l'identification des numeros de canaux pour les matrices de visibilites

7/ Petites ameliorations (ecriture mot-cles) pour MultiDataSaver et

amelioration classe BRMultiFitsReader pour lecture FillTime (DATEOBS)
et FiberIds depuis les fichiers FITS et remplissage de RAcqMemZoneMgr

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