source: Sophya/trunk/AddOn/TAcq/racqurw.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: 35.1 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 "racqurw.h"
8
9#include <stdlib.h>
10#include <unistd.h>
11#include <fstream>
12#include <signal.h>
13#include "pexceptions.h"
14#include "timestamp.h"
15
16#include "pciewrap.h"
17#include "brpaqu.h"
18#include "minifits.h"
19
20#include "resusage.h" // Pour mesure temps elapsed/CPU ...
21#include "datatable.h" // Pour sauver les entetes de paquet
22#include <sys/time.h> // pour gettimeofday
23
24// Si on veut que MultiDataSaver cree des fichiers avec le numero des FrameCounters...
25// define DEBUGPAQHDR
26
27//-------------------------------------------------------
28// Classe thread de lecture PCI-Express
29//-------------------------------------------------------
30
31PCIEReader::PCIEReader(PCIEWrapperInterface &pciw,uint_4 sizeFrame,uint_4 packSize ,RAcqMemZoneMgr& mem,
32 uint_4 nmax, BRDataFmtConv swapall)
33 : memgr(mem) , pciw_ (pciw)
34{
35 nmax_ = nmax;
36 swapall_ = swapall; // select data swap/format conversion for BRPaquet
37 stop_ = false;
38 packSize_ = packSize;
39 sizeFr_ =sizeFrame;
40 // Pour la logique de gestion des paquets ds zone memoire
41 mid_ = -2;
42 targ_npaq_ = 0;
43
44 max_targ_npaq = memgr.NbPaquets();
45 mmbuf_ = NULL;
46}
47
48bool PCIEReader::MZoneManage(bool fgclean) // Retourne true si probleme
49{
50 /* Pour debug
51 cout << " PCIEReader::MZoneManage() mid_=" << mid_ << " arg_npaq_= " << targ_npaq_
52 << " max_targ_npaq=" << max_targ_npaq << endl;
53 */
54 if (mid_ >= 0) memgr.FreeMemZone(mid_, MemZS_Filled);
55 mmbuf_ = NULL; targ_npaq_ = 0; mid_ = -2;
56 if (fgclean) return false;
57 mid_ = memgr.FindMemZoneId(MemZA_Fill);
58 mmbuf_ = memgr.GetMemZone(mid_);
59 if (mmbuf_==NULL) return true;
60 return false;
61}
62
63void PCIEReader::run()
64{
65 //Precision insuffisante ResourceUsage ru; ru.Update(); // Pour recuperer le temps passe
66 struct timeval tv1, tv2;
67 gettimeofday(&tv1, NULL);
68
69 try{
70 cout << " PCIEReader::run() - Starting , NMaxMemZones=" << nmax_
71 << "memgr.NbPaquets()=" << memgr.NbPaquets() << endl;
72 setRC(1);
73
74
75 // sigaddset(&act.sa_mask,SIGINT); // pour proteger le transfert DMA
76 //sigaction(SIGINT,&act,NULL);
77
78 uint_4 paqsz = memgr.PaqSize();
79 uint_4 dmasz = pciw_.TransferSize();
80 pciw_.StartTransfers();
81
82 Byte* Datas = NULL;
83 Byte* tampon = new Byte[paqsz];
84 Byte* nextpaq = NULL;
85
86 uint_4 off_acheval = 0;
87
88 int nerrdma = 0;
89 int maxerrdma = 10;
90 bool fgarret = false;
91
92 uint_4 npaqfait = 0; // Nb total de paquets traites (DMA + decode)
93
94 /// do{ si boucle infini
95 // for (uint_4 kmz=0; kmz<nmax_; kmz++) {
96 while (npaqfait < nmax_*memgr.NbPaquets()) {
97 if (stop_) break;
98 if (fgarret) break;
99 // On pointe vers le debut de la zone a remplir aver le prochain DMA
100 Datas=pciw_.GetData();
101 if (Datas == NULL) { // No data Read in DMA
102 nerrdma ++;
103 cout << "PCIEReaderChecker/Erreur Waiting for datas ..." << endl;
104 pciw_.PrintStatus(cout);
105 if (nerrdma>=maxerrdma) { fgarret = true; break; }
106 }
107 else { // DMA reussi
108 uint_4 curoff = 0;
109 //1- On traite le paquet a cheval, rempli partiellement avec le DMA d'avant si necessaire
110 if (off_acheval > 0) {
111 if ((paqsz-off_acheval)< dmasz) {
112 memcpy((void *)(tampon+off_acheval), (void *)Datas, paqsz-off_acheval);
113 curoff = paqsz-off_acheval; off_acheval = 0;
114
115 if ((nextpaq=NextPaqTarget()) == NULL) {
116 cout << "2 PCIEReader::run()/Error NextPaqTarget() returned NULL ->STOP 9" << endl;
117 setRC(9); fgarret=true; break;
118 }
119 BRPaquet paq(tampon, nextpaq, paqsz, swapall_);
120 npaqfait++; // Ne pas oublier le compteur de paquets faits
121 }
122 else {
123 memcpy((void *)(tampon+off_acheval), (void *)Datas, dmasz);
124 curoff =dmasz;
125 off_acheval = (dmasz+off_acheval);
126 }
127 }
128 //2- On traite les paquets complet qui se trouvent dans la zone du DMA
129 while((curoff+paqsz)<=dmasz) {
130 // BRPaquet paq((Byte*)(Datas)+((paqsz*j)), nextdma+j*paqsz, paqsz, swapall_);
131 if ((nextpaq=NextPaqTarget()) == NULL) {
132 cout << "3 PCIEReader::run()/Error NextPaqTarget() returned NULL ->STOP 9" << endl;
133 setRC(9); fgarret=true; break;
134 }
135 BRPaquet paq(Datas+curoff, nextpaq, paqsz, swapall_);
136 curoff += paqsz; // On avance l'index dans le buffer du DMA
137 npaqfait++; // Ne pas oublier le compteur de paquets faits
138 } // -- FIN traitement des paquets complets ds un DMA
139 //3- On copie si besoin la fin du DMA dans la zone tampon
140 if (curoff < dmasz) {
141 if (fgarret) break; // pour sortir si l'on est passe par un STOP 9
142 off_acheval = dmasz-curoff;
143 memcpy(tampon, (void*)(Datas+curoff), off_acheval);
144 curoff += off_acheval;
145 }
146 } // Traitement d'un DMA OK
147 }
148// }while(!stop_);
149
150
151 gettimeofday(&tv2, NULL);
152 double tmelaps2 = (tv2.tv_sec-tv1.tv_sec)*1000.+(tv2.tv_usec-tv1.tv_usec)/1000.;
153 if (tmelaps2<0.1) tmelaps2=0.1;
154 cout << " ------------------ PCIEReader::run()-End summary -------------------" << endl;
155 cout << " PCIEReader/Info TotTransfer=" << pciw_.TotTransferBytes()/1024
156 << " kb , ElapsTime=" << tmelaps2 << " ms ->"
157 << (double)pciw_.TotTransferBytes()/tmelaps2 << " kb/s" << endl;
158 cout << " --------------------------------------------------------------------" << endl;
159
160 MZoneManage(true);
161 delete [] tampon;
162
163 }catch (PException& exc) {
164 cout << " PCIEREADER::run()/catched PException " << exc.Msg() << endl;
165 setRC(3);
166 return;
167 }
168 catch(...) {
169 cout << " PCIEREADER::run()/catched unknown ... exception " << endl;
170 setRC(4);
171 return;
172 }
173 setRC(0);
174
175 return;
176}
177
178
179void PCIEReader::Stop()
180{
181 // cout << " PCIEReader::Stop() -------------> STOP" <<endl;
182 stop_ = true;
183 return;
184}
185
186
187//-------------------------------------------------------
188// Classe thread de sauvegarde sur fichiers
189//-------------------------------------------------------
190
191DataSaver::DataSaver(RAcqMemZoneMgr& mem, string& path, uint_4 nfiles, uint_4 nblocperfile, bool savesig)
192 : memgr(mem)
193{
194 nfiles_ = nfiles;
195 nblocperfile_ = nblocperfile;
196 nmax_ = nblocperfile_*nfiles_;
197 savesig_ = savesig; // Si false, pas d'ecriture des fichiers FITS du signal
198 stop_ = false;
199 path_ = path;
200}
201void DataSaver::Stop()
202{
203 // cout<< " DataSaver:Stop ........ " << endl;
204 stop_=true;
205
206}
207void DataSaver::run()
208{
209 setRC(1);
210 BRPaqChecker pcheck; // Verification/comptage des paquets
211
212 try {
213 TimeStamp ts;
214 cout << " DataSaver::run() - Starting " << ts << " NbFiles=" << nfiles_ << " NBloc/File="
215 << nblocperfile_ << " NMaxMemZones=" << nmax_ << endl;
216 char fname[512];
217
218 sprintf(fname,"%s/saver.log",path_.c_str());
219 ofstream filog(fname);
220 filog << " DataSaver::run() - starting log file " << ts << endl;
221 filog << " NbFiles=" << nfiles_ << " NBloc/File=" << nblocperfile_ << " NMaxMemZones=" << nmax_ << endl;
222
223 DataTable dt;
224 dt.AddLongColumn("TimeTag");
225 dt.AddIntegerColumn("FrameCounter");
226 dt.AddIntegerColumn("FrameLength");
227 dt.AddIntegerColumn("Num");
228 DataTableRow dtr = dt.EmptyRow();
229 uint_8 timtag = 0;
230 uint_4 numpaq = 0;
231 bool dthead = false; // Mettre a false pour ne pas remplir DataTable Headers
232
233
234 uint_4 fnum=0;
235 uint_4 paqsz = memgr.PaqSize();
236 cout << " ============================ DataSaver::run() PaqSize " << paqsz <<endl;
237 bool fgnulldev = false;
238 if (path_ == "/dev/null") {
239 cout << " DataSaver::run()/Warning /dev/null path specified, filenames=/dev/null" << endl;
240 fgnulldev = true;
241 }
242 for (uint_4 nbFile=0;nbFile<nfiles_ ;nbFile++) {
243 if (stop_ ) break;
244 if (fgnulldev) strcpy(fname,"/dev/null");
245 else sprintf(fname,"%s/HDRfits%d.txt",path_.c_str(),fnum);
246 ofstream header(fname);
247
248 BRPaquet paq0(NULL, NULL, paqsz);
249 uint_4 npaqperfile = memgr.NbPaquets()*nblocperfile_; // Nombre de paquets ecrits dans un fichier
250
251 MiniFITSFile mff;
252 if (savesig_) { //Reza - Ouverture conditionnel fichier
253 if (fgnulldev) strcpy(fname,"/dev/null");
254 else sprintf(fname,"%s/signal%d.fits",path_.c_str(),(int)fnum++);
255 mff.Open(fname,MF_Write); //Reza - Ouverture conditionnel fichier
256 // Entete correspondant a l'ecriture tout le paquet - trailer compris (modif Mai 2009)
257 mff.setDTypeNaxis(MF_Byte, paq0.PaquetSize(), npaqperfile);
258
259 // Sans TRAILER de paquet mff.setDTypeNaxis(MF_Byte, paq0.DataSize()+paq0.HeaderSize(), npaqperfile);
260 }
261 else sprintf(fname,"MemDataBloc[%d]-NoDataFile",(int)fnum++);
262
263 for (uint_4 kmz=0; kmz<nblocperfile_; kmz++) {
264 if (stop_) break;
265 //DBG cout << " DataSaver::run()- nbFile=" << nbFile << " kmz=" << kmz << endl;
266 int mid = memgr.FindMemZoneId(MemZA_Save);
267 Byte* buff = memgr.GetMemZone(mid);
268 if (buff == NULL) {
269 cout << " DataSaver::run()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
270 setRC(2);
271 return;
272 }
273 for(uint_4 i=0; i<memgr.NbPaquets(); i++) {
274
275 BRPaquet paq(NULL, buff+i*paqsz, paqsz);
276 pcheck.Check(paq); // Verification du paquet / FrameCounter
277 if (savesig_)
278 header << hex << paq.HDRMarker() << " " << paq.TRLMarker() << " "
279 << paq.TimeTag2()<< " "<< paq.TimeTag1()<< " "
280 << paq.FrameCounter() << " " << paq.PaqLen() << endl;
281
282 if (dthead) { // Remplissage DataTable entetes paquets
283 timtag = (uint_8)paq.TimeTag2()*0x100000000ULL+paq.TimeTag1();
284 dtr[0] = timtag;
285 dtr[1] = paq.FrameCounter();
286 dtr[2] = paq.PaqLen();
287 dtr[3] = numpaq++;
288 dt.AddRow(dtr);
289 }
290
291 if (savesig_) // Reza - Ecriture conditionnel fichier fits signal
292 mff.WriteB(paq.Header(),paq.PaquetSize()); // ecriture tout le paquet (modif Mai 2009)
293 // mff.WriteB(paq.Data1(), paq.DataSize());
294 // mff.WriteB(paq.Header(),paq.DataSize()+paq.HeaderSize()); // ecriture datas + header
295 }
296 memgr.FreeMemZone(mid, MemZS_Saved);
297 }
298 ts.SetNow();
299 filog << ts << " : OK data file " << fname << endl;
300 cout << " DataSaver::run() " << ts << " : OK data file " << fname << endl;
301 }
302 if (dthead) {
303 cout << dt;
304 char fname2[256];
305 sprintf(fname2,"%s/headers.ppf",path_.c_str());
306 POutPersist po(fname2);
307 po << dt;
308 }
309 cout << " -------------------- DataSaver::run() -------------------- " << endl;
310 pcheck.Print(cout);
311 cout << " ---------------------------------------------------------- " << endl;
312 ts.SetNow();
313 pcheck.Print(filog);
314 filog << " DataSaver::run() - End of processing/run() " << ts << endl;
315
316 }
317 catch (MiniFITSException& exc) {
318 cout << " DataSaver::run()/catched MiniFITSException " << exc.Msg() << endl;
319 setRC(3);
320 return;
321 }
322 catch(...) {
323 cout << " DataSaver::run()/catched unknown ... exception " << endl;
324 setRC(4);
325 return;
326 }
327 setRC(0);
328 return;
329}
330
331
332
333
334//----------------------------------------------------------------------------------------------------------
335// Classe thread de lecture PCI-Express + Check pour tests de verification de debit/etc avec un seul thread
336//----------------------------------------------------------------------------------------------------------
337
338PCIEReaderChecker::PCIEReaderChecker(PCIEWrapperInterface &pciw, uint_4 sizeFrame,uint_4 packSize, RAcqMemZoneMgr& mem,
339 uint_4 nmax, BRDataFmtConv swapall)
340 : memgr(mem) , pciw_ (pciw)
341{
342 nmax_ = nmax;
343 swapall_ = swapall; // select data swap/format conversion for BRPaquet
344 stop_ = false;
345 packSize_ = packSize;
346 sizeFr_ =sizeFrame;
347}
348void PCIEReaderChecker::run()
349{
350
351 struct timeval tv1, tv2;
352 gettimeofday(&tv1, NULL);
353
354 cout << " PCIEReaderChecker::run() - Starting , NMaxMemZones=" << nmax_
355 << " memgr.NbPaquets()=" << memgr.NbPaquets() << endl;
356 setRC(1);
357 cout << " ... RAcqMemZoneMgr not used - using s fixed memory location for packets decoding ..." << endl;
358
359
360 // sigaddset(&act.sa_mask,SIGINT); // pour proteger le transfert DMA
361 //sigaction(SIGINT,&act,NULL);
362 uint_4 paqsz = memgr.PaqSize();
363 uint_4 dmasz = pciw_.TransferSize();
364 pciw_.StartTransfers();
365
366 BRPaqChecker pcheck; // Verification/comptage des paquets
367
368 Byte* Datas = NULL;
369 Byte* locdata = new Byte[paqsz*memgr.NbPaquets()*memgr.NbZones()];
370 Byte* tampon = new Byte[paqsz];
371
372 uint_4 off_acheval = 0;
373
374 int nerrdma = 0;
375 int maxerrdma = 10;
376 bool fgarret = false;
377 for (uint_4 kmz=0; kmz<nmax_; kmz++) {
378 if (stop_) break;
379 if (fgarret) break;
380
381 Byte* nextdma = locdata+((kmz%memgr.NbZones())*(paqsz*memgr.NbPaquets()));
382 uint_4 npaqfait = 0;
383 // for (uint_4 i=0; i<memgr.NbPaquets(); i += pktInDMATr) { // attention pktInDMATr paquets dans 1 seul DMA
384 while (npaqfait < memgr.NbPaquets()) {
385 if (fgarret) break;
386 // On pointe vers le debut de la zone a remplir aver le prochain DMA
387 //-- Zone memoire locale Byte* nextdma = buff+i*paqsz;
388 Datas=pciw_.GetData();
389
390 if (Datas == NULL) { // No data Read in DMA
391 nerrdma ++;
392 cout << "PCIEReaderChecker/Erreur Waiting for datas ..." << endl;
393 pciw_.PrintStatus(cout);
394 if (nerrdma>=maxerrdma) { fgarret = true; break; }
395 }
396 else {
397 uint_4 curoff = 0;
398 //1- On traite le paquet a cheval, rempli partiellement avec le DMA d'avant si necessaire
399 // if (off_acheval > 0) {
400 // memcpy((void *)(tampon+off_acheval), (void *)Datas, paqsz-off_acheval);
401 // curoff = paqsz-off_acheval; off_acheval = 0;
402
403 // BRPaquet paq(tampon, locdata, paqsz, swapall_);
404 // npaqfait++; // Ne pas oublier le compteur de paquets faits
405 // pcheck.Check(paq); // Verification du paquet / FrameCounter
406 //}
407 if (off_acheval > 0) {
408 if ((paqsz-off_acheval)< dmasz) {
409 memcpy((void *)(tampon+off_acheval), (void *)Datas, paqsz-off_acheval);
410 curoff = paqsz-off_acheval; off_acheval = 0;
411
412 BRPaquet paq(tampon, locdata, paqsz, swapall_);
413 npaqfait++; // Ne pas oublier le compteur de paquets faits
414 pcheck.Check(paq); // Verification du paquet / FrameCounter
415 }
416 else {
417 memcpy((void *)(tampon+off_acheval), (void *)Datas, dmasz);
418 curoff =dmasz;
419 off_acheval = (dmasz+off_acheval);
420 }
421 }
422 //2- On traite les paquets complet qui se trouvent dans la zone du DMA
423 while((curoff+paqsz)<=dmasz) {
424 // BRPaquet paq((Byte*)(Datas)+((paqsz*j)), nextdma+j*paqsz, paqsz, swapall_);
425 // BRPaquet paq(Datas+curoff, locdata, paqsz, swapall_);
426 BRPaquet paq(Datas+curoff, nextdma+npaqfait*paqsz, paqsz, swapall_);
427 curoff += paqsz; // On avance l'index dans le buffer du DMA
428 npaqfait++; // Ne pas oublier le compteur de paquets faits
429 pcheck.Check(paq); // Verification du paquet / FrameCounter
430 } // -- FIN traitement des paquets complets ds un DMA
431 //3- On copie si besoin la fin du DMA dans la zone tampon
432 if (curoff < dmasz) {
433 off_acheval = dmasz-curoff;
434 memcpy(tampon, (void*)(Datas+curoff), off_acheval);
435 curoff += off_acheval;
436 }
437 } // Traitement d'un DMA OK
438
439 } // Fin boucle de remplissage d'une zone memoire
440 } // Fin boucle sur les zones
441
442 setRC(0);
443 gettimeofday(&tv2, NULL);
444 double tmelaps2 = (tv2.tv_sec-tv1.tv_sec)*1000.+(tv2.tv_usec-tv1.tv_usec)/1000.;
445 if (tmelaps2<0.1) tmelaps2=0.1;
446 cout << " ------------------ PCIEReaderChecker::run()-End summary -------------------" << endl;
447 cout << " PCIEReaderChecker/Info TotTransfer=" << pciw_.TotTransferBytes()/1024
448 << " kb , ElapsTime=" << tmelaps2 << " ms ->"
449 << (double)pciw_.TotTransferBytes()/tmelaps2 << " kb/s" << endl;
450 pcheck.Print(cout);
451 cout << " --------------------------------------------------------------------" << endl;
452
453 delete [] locdata;
454 delete [] tampon;
455
456 return;
457}
458void PCIEReaderChecker::Stop()
459{
460 // cout << " PCIEReaderChecker::stop() ........ STOP" <<endl;
461 stop_ = true;
462
463}
464////////////////////////////////////////////////////////////////////////////////////////////////////////
465//----------------------------------------------------------------------------------------------------------
466// Classe thread de lecture PCI-Express + Check pour tests de verification de debit/etc avec un seul thread
467//----------------------------------------------------------------------------------------------------------
468
469/* --Methode-- */
470PCIEMultiReader::PCIEMultiReader(vector<PCIEWrapperInterface*> vec_pciw, RAcqMemZoneMgr& mem, BRParList const& par)
471 : memgr(mem), par_(par), vec_pciw_ (vec_pciw)
472{
473 nmax_ = par_.MaxNbBlocs();
474 swapall_ = par_.GetDataConvFg(); // select data swap/format conversion for BRPaquet
475 stop_ = false;
476 packSize_ = par_.RecvPaquetSize();
477 packSizeInMgr_=memgr.PaqSize();
478 sizeFr_=par_.DMASizeBytes();
479 if (vec_pciw.size() != memgr.NbFibres()) {
480 cout << " PCIEMultiReader()PbArgs: vec_pciw.size()= " << vec_pciw.size() << " memgr.NbFibres()=" <<memgr.NbFibres()<< endl;
481 throw ParmError("PCIEMultiReader:ERROR/ arguments incompatibles vec_pciw.size() != memgr.NbFibres() ");
482 }
483 if (vec_pciw.size() > MAXNBFIB)
484 throw ParmError("PCIEMultiReader:ERROR/ vec_pciw.size() > MAXNBFIB ");
485 nbDma_= vec_pciw.size();
486 mid_=-2;
487 mmbuf_=NULL;
488 max_targ_npaq = memgr.NbPaquets();
489 for (int fid=0 ; fid<(int)nbDma_ ;fid++) {
490 memgr.FiberId(fid)=vec_pciw[fid]->FiberId();
491 mmbufib_[fid]=NULL;
492 }
493}
494
495/* --Methode-- */
496void PCIEMultiReader::run()
497{
498
499 struct timeval tv1,tv2;
500 gettimeofday(&tv1, NULL);
501
502 cout << " PCIEMultiReader::run() - Starting , NMaxMemZones=" << nmax_
503 << " memgr.NbPaquets(),PaqSz=" << memgr.NbPaquets() << " ," << memgr.PaqSize()
504 << " DMA-Paqsize " << packSize_ << " " << BRPaquet::FmtConvToString(swapall_) << endl;
505 setRC(1);
506
507 // sigaddset(&act.sa_mask,SIGINT); // pour proteger le transfert DMA
508 //sigaction(SIGINT,&act,NULL);
509 uint_4 paqszmm = memgr.PaqSize();
510 uint_4 paqsz = packSize_;
511 uint_4 dmasz = vec_pciw_[0]->TransferSize();
512 //DEL vec_pciw_[0]->StartTransfers();
513
514 BRPaqChecker pcheck[MAXNBFIB]; // Verification/comptage des paquets
515 Byte* Datas[MAXNBFIB];
516 Byte* tampon[MAXNBFIB] ;
517 Byte* predtampon=NULL; // tampon de recopie pour la reduction des tailles de paquets
518 Byte* nextpaq=NULL;
519 uint_4 off_acheval=0;
520
521 bool fgarret = false;
522
523 // Initialisation des tampons pour recopie des paquets a cheval pour chaque DMA
524 for (int i=0;i< (int)nbDma_ ;i++) {
525 tampon[i]= new Byte[paqsz];
526 vec_pciw_[i]->SetMaxWaitEndDMA(par_.first_maxkwedma_,par_.first_nretrydma_);
527 }
528 bool fgredpaq=par_.fgreducpsize;
529 if (fgredpaq) {
530 cout << " PCIEMultiReader::run() - PaquetSizeReduction - RedSize=" << par_.redpqsize
531 << " Offset=" << par_.reducoffset << " " << ((par_.reducneedcopy)?"NeedCopy":"NOCopy")
532 << " " << BRPaquet::ReducActionToString(par_.pqreducmode) << endl;
533 predtampon = new Byte[paqsz];
534 }
535
536#ifdef DEBUGPAQHDR
537 ofstream header[MAXNBFIB];
538 for(uint_4 fib=0; fib<nbDma_; fib++) {
539 char hfnm[128];
540 sprintf(hfnm, "./HDRCountPaqs%d.txt", fib);
541 header[fib].open(hfnm);
542 }
543#endif
544 uint_4 npaqfait[MAXNBFIB] ;
545 for (int i=0;i< (int)nbDma_ ;i++) npaqfait[i]=0;
546 // Byte* nextdma = locdata+((kmz%memgr.NbZones())*(paqsz*memgr.NbPaquets()));
547 uint_4 npaqfaitg = 0;
548 // for (uint_4 i=0; i<memgr.NbPaquets(); i += pktInDMATr) { // attention pktInDMATr paquets dans 1 seul DMA
549 while (npaqfaitg < nmax_*memgr.NbPaquets()) { // Boucle global G
550 if (fgarret) break;
551 if (stop_) break;
552
553 // Lancement des DMA
554 for (int dma=0; dma < (int)nbDma_ ;dma++) vec_pciw_[dma]->StartTransfers();
555 if (npaqfaitg==1) {
556 for (int i=0;i< (int)nbDma_ ;i++)
557 vec_pciw_[i]->SetMaxWaitEndDMA(par_.maxkwedma_,par_.nretrydma_);
558 }
559 // On pointe vers le debut de la zone a remplir aver le prochain DMA
560 //-- Zone memoire locale Byte* nextdma = buff+i*paqsz;
561
562 bool fgbaddma=false;
563 // On boucle sur les nbDma_ en attente de leurs terminaison
564 for (int dma=0; dma <(int) nbDma_ ;dma++) {
565 Datas[dma]=vec_pciw_[dma]->GetData();
566 if (Datas[dma] == NULL) { // No data Read in DMA
567 fgbaddma=true;
568 cout << "PCIEMultiReaderChecker/Error Waiting for datas ..." << endl;
569 vec_pciw_[dma]->PrintStatus(cout);
570 fgarret = true; break;
571 }
572 }
573 if (fgbaddma) continue;
574 uint_4 curoff=0;
575 //1- On traite le paquet a cheval, rempli partiellement avec le DMA d'avant si necessaire pour les n fibres
576 if (off_acheval > 0) { // IF Numero B
577 if ((paqsz-off_acheval)< dmasz) { // IF Numero A
578 for(uint_4 fib=0; fib<nbDma_; fib++)
579 memcpy((void *)((tampon[fib])+off_acheval), (void *)Datas[fib], paqsz-off_acheval);
580 curoff = paqsz-off_acheval; off_acheval = 0;
581 if ( MoveToNextTarget() ) {
582 cout << "PCIEMultiReader::run()/Error-A- MoveToNextTarget() returned true ->STOP 9" << endl;
583 setRC(9); fgarret=true; break;
584 }
585 for(uint_4 fib=0; fib<nbDma_; fib++) {
586 nextpaq=GetPaquetTarget(fib);
587 if (nextpaq == NULL) { // Cela ne devrait pas arriver
588 cout << "PCIEReader::run()/Error-A2- GetPaquetTarget(fib) returned NULL ->STOP 9" << endl;
589 setRC(9); fgarret=true; break;
590 }
591 // CHECK S'il faut faire une reduction de taille de paquet
592 if (fgredpaq) { // reduction taille de paquet
593 if (par_.reducneedcopy) {
594 BRPaquet paqc1(tampon[fib], predtampon, paqsz, swapall_);
595 BRPaquet paqc2(nextpaq, par_.redpqsize);
596 paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
597 }
598 else {
599 BRPaquet paqc1(tampon[fib], paqsz);
600 BRPaquet paqc2(nextpaq, par_.redpqsize);
601 paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
602 }
603 }
604 else {
605 BRPaquet paqc(tampon[fib], nextpaq, paqsz, swapall_);
606 }
607 BRPaquet paq(nextpaq, packSizeInMgr_);
608 npaqfait[fib]++;
609 if (fib==nbDma_-1) npaqfaitg++; // Ne pas oublier le compteur de paquets faits
610 pcheck[fib].Check(paq); // Verification du paquet / FrameCounter
611#ifdef DEBUGPAQHDR
612 header[fib] << dec << paq.FrameCounter()<< endl;
613#endif
614 }
615 }
616 else { // se rapporte au IF numero A
617 for(uint_4 fib=0; fib<nbDma_; fib++)
618 memcpy((void *)(tampon[fib]+off_acheval), (void *)Datas[fib], dmasz);
619 curoff =dmasz;
620 off_acheval = (dmasz+off_acheval);
621 }
622 } // Fin IF Numero B
623
624 //2- On traite les paquets complets qui se trouvent dans la zone du DMA
625 while ((curoff+paqsz)<=dmasz) { // while numero C
626 // if ((dma==nbDma_-1)&&(npaqfait >= nmax_* memgr.NbPaquets())) break;
627 if ( MoveToNextTarget() ) {
628 cout << "PCIEMultiReader::run()/Error-B- MoveToNextTarget() returned true ->STOP 9" << endl;
629 setRC(9); fgarret=true; break;
630 }
631 for(uint_4 fib=0; fib<nbDma_; fib++) {
632 if (npaqfait[fib] >= nmax_*memgr.NbPaquets()) continue;
633 nextpaq=GetPaquetTarget(fib);
634 if (nextpaq == NULL) { // Cela ne devrait pas arriver
635 cout << "PCIEReader::run()/Error-B2- GetPaquetTarget(fib) returned NULL ->STOP 9" << endl;
636 setRC(9); fgarret=true; break;
637 }
638 // CHECK S'il faut faire une reduction de taille de paquet
639 if (fgredpaq) { // reduction taille de paquet
640 if (par_.reducneedcopy) {
641 BRPaquet paqc1(Datas[fib]+curoff, predtampon, paqsz, swapall_);
642 BRPaquet paqc2(nextpaq, par_.redpqsize);
643 paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
644 }
645 else {
646 BRPaquet paqc1(Datas[fib]+curoff, paqsz);
647 BRPaquet paqc2(nextpaq, par_.redpqsize);
648 paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
649 }
650 }
651 else {
652 BRPaquet paqc(Datas[fib]+curoff, nextpaq, paqsz, swapall_);
653 }
654 BRPaquet paq(nextpaq, packSizeInMgr_);
655 npaqfait[fib]++;
656 if (fib==nbDma_-1) npaqfaitg++; // Ne pas oublier le compteur de paquets faits
657 pcheck[fib].Check(paq); // Verification du paquet / FrameCounter
658#ifdef DEBUGPAQHDR
659 header[fib] << dec << paq.FrameCounter()<< endl;
660#endif
661 }
662 curoff += paqsz; // On avance l'index dans le buffer du DMA
663 } // -- FIN traitement des paquets complets ds un DMA - FIN du while numero C
664 //3- On copie si besoin la fin du DMA dans la zone tampon
665 if (curoff < dmasz) { // IF numero D
666 off_acheval = dmasz-curoff;
667 for(uint_4 fib=0; fib<nbDma_; fib++)
668 memcpy(tampon[fib], (void*)(Datas[fib]+curoff), off_acheval);
669 // ne sert a rien curoff += off_acheval;
670 } // FIN du if numero D
671 } // FIN Boucle global G
672
673
674 setRC(0);
675 gettimeofday(&tv2, NULL);
676 double tmelaps2 = (tv2.tv_sec-tv1.tv_sec)*1000.+(tv2.tv_usec-tv1.tv_usec)/1000.;
677 if (tmelaps2<0.1) tmelaps2=0.1;
678 cout << " ---------- PCIEMultiReader::run()-End summary NPaqFait=" << npaqfaitg << "------------- " << endl;
679 for (int dma=0; dma < (int)nbDma_ ;dma++) {
680 cout << " --Fib=" << dma << " NPaqFait=" << npaqfait[dma] << " TotTransfer="
681 << vec_pciw_[dma]->TotTransferBytes()/1024
682 << " kb , ElapsTime=" << tmelaps2 << " ms ->"
683 << (double)vec_pciw_[dma]->TotTransferBytes()/tmelaps2 << " kb/s" << endl;
684 pcheck[dma].Print(cout);
685 }
686 cout << " --------------------------------------------------------------------" << endl;
687
688 //---- Nettoyage final et arret
689 MZoneManage(true);
690 memgr.Stop(); // Pour arreter les autres threads
691
692 for (int i=0;i< (int)nbDma_ ;i++) delete[] tampon[i];
693 if ((fgredpaq)&&predtampon) delete[] predtampon;
694#ifdef DEBUGPAQHDR
695 for(uint_4 fib=0; fib<nbDma_; fib++) header[fib].close();
696#endif
697
698 //DBG cout << " fin thread ========================" <<endl;
699 return;
700}
701
702/* --Methode-- */
703bool PCIEMultiReader::MZoneManage(bool fgclean) // Retourne true si probleme
704{
705 /* Pour debug
706 cout << " PCIEReader::MZoneManage() mid_=" << mid_ << " arg_npaq_= " << targ_npaq_
707 << " max_targ_npaq=" << max_targ_npaq << endl;
708 */
709 if (mid_ >= 0) {
710 memgr.GetAuxData(mid_)->filltime_.SetNow();
711 memgr.FreeMemZone(mid_, MemZS_Filled);
712 }
713 mmbuf_ = NULL; targ_npaq_ = 0; mid_ = -2;
714 for (int fid=0;fid<(int)nbDma_ ;fid++) mmbufib_[fid]=NULL;
715 if (fgclean) return false;
716 mid_ = memgr.FindMemZoneId(MemZA_Fill);
717 mmbuf_ = memgr.GetMemZone(mid_);
718 if (mmbuf_==NULL) return true;
719 for (int fid=0;fid<(int)nbDma_ ;fid++) mmbufib_[fid]=memgr.GetMemZone(mid_,fid);
720 return false;
721}
722
723/*
724bool PCIEMultiReader::MZoneManage(int zone,bool fgclean) // Retourne true si probleme
725{
726 // Pour debug
727 //cout << " PCIEReader::MZoneManage() mid_=" << mid_ << " arg_npaq_= " << targ_npaq_
728 << " max_targ_npaq=" << max_targ_npaq << endl;
729 if (mid_[zone] >= 0) memgr[zone]->FreeMemZone(mid_[zone], MemZS_Filled);
730 mmbuf_[zone] = NULL; targ_npaq_[zone] = 0; mid_[zone] = -2;
731 if (fgclean) return false;
732 mid_[zone] = memgr[zone]->FindMemZoneId(MemZA_Fill);
733 mmbuf_[zone] = memgr[zone]->GetMemZone(mid_[zone]);
734 if (mmbuf_[zone]==NULL) return true;
735 return false;
736}
737*/
738
739/* --Methode-- */
740void PCIEMultiReader::Stop()
741{
742 // cout << " PCIEReaderChecker::stop() ........ STOP" <<endl;
743 stop_ = true;
744
745}
746
747
748//--------------------------------------------------------------------
749// Classe thread de sauvegarde sur fichiers avec gestion multifibres
750//--------------------------------------------------------------------
751
752/* --Methode-- */
753MultiDataSaver::MultiDataSaver(RAcqMemZoneMgr& mem)
754 : memgr(mem)
755{
756 BRAcqConfig bpar;
757 nfiles_ = bpar.MaxNbFiles();
758 nblocperfile_ = bpar.BlocPerFile();
759 nmax_ = nblocperfile_*nfiles_;
760 savesig_ = bpar.GetSaveFits(); // Si false, pas d'ecriture des fichiers FITS du signal
761 stop_ = false;
762}
763
764/* --Methode-- */
765void MultiDataSaver::Stop()
766{
767 // cout<< " MultiDataSaver:Stop ........ " << endl;
768 stop_=true;
769}
770
771/* --Methode-- */
772void MultiDataSaver::run()
773{
774 setRC(1);
775 BRPaqChecker pcheck[MAXNBFIB]; // Verification/comptage des paquets
776 BRAcqConfig bpar;
777 try {
778 TimeStamp ts;
779 cout << " MultiDataSaver::run() - Starting " << ts << " \n NbFiles=" << nfiles_ << " NBloc/File="
780 << nblocperfile_ << " NMaxMemZones=" << nmax_ << endl;
781 char fname[512];
782
783 sprintf(fname,"%s/msaver.log",bpar.OutputDirectory().c_str());
784 ofstream filog(fname);
785 filog << " MultiDataSaver::run() - starting log file " << ts << " NFibres= " << memgr.NbFibres() << endl;
786 filog << " NbFiles=" << nfiles_ << " NBloc/File=" << nblocperfile_ << " NMaxMemZones=" << nmax_ << endl;
787
788 // Fichiers entete ascii et signal FITS
789 ofstream header[MAXNBFIB];
790 MiniFITSFile mff[MAXNBFIB];
791
792 uint_4 fnum=0;
793 uint_4 paqsz = memgr.PaqSize();
794 cout << " ============================ MultiDataSaver::run() PaqSize " << paqsz <<endl;
795 bool fgfirstfctt=false;
796 bool fgnulldev = bpar.GetFileDevNull();
797 if (fgnulldev) {
798 cout << " MultiDataSaver::run()/Warning /dev/null path specified, filenames=/dev/null" << endl;
799 fgnulldev = true;
800 }
801 BRPaquet paq0(NULL, NULL, paqsz);
802 TimeStamp tsmz;
803 for (uint_4 nbFile=0;nbFile<nfiles_ ;nbFile++) {
804 if (stop_ ) break;
805 if (memgr.GetRunState() == MemZR_Stopped) break;
806
807 if (savesig_)
808 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
809 if (fgnulldev) strcpy(fname,"/dev/null");
810 else sprintf(fname,"%sHDRfits%d.txt",bpar.OutputDirectoryFib(fib).c_str(),fnum);
811 header[fib].open(fname);
812 }
813
814 uint_4 npaqperfile = memgr.NbPaquets()*nblocperfile_; // Nombre de paquets ecrits dans un fichier
815
816 if (savesig_) { //Reza - Ouverture conditionnel fichier
817 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
818 if (fgnulldev) strcpy(fname,"/dev/null");
819 else sprintf(fname,"%ssignal%d.fits",bpar.OutputDirectoryFib(fib).c_str(),(int)fnum);
820 // cout << " ***DBG** Opening file " << fname << endl;
821 mff[fib].Open(fname,MF_Write); //Reza - Ouverture conditionnel fichier
822 // Entete correspondant a l'ecriture tout le paquet - trailer compris (modif Mai 2009)
823 mff[fib].setDTypeNaxis(MF_Byte, paq0.PaquetSize(), npaqperfile);
824 // Sans TRAILER de paquet mff.setDTypeNaxis(MF_Byte, paq0.DataSize()+paq0.HeaderSize(), npaqperfile);
825 }
826 fnum++; fgfirstfctt=true;
827 }
828 else sprintf(fname,"MemDataBloc[%d]-NoDataFile",(int)fnum++);
829
830 for (uint_4 kmz=0; kmz<nblocperfile_; kmz++) {
831 if (stop_) break;
832 //DBG cout << " MultiDataSaver::run()- nbFile=" << nbFile << " kmz=" << kmz << endl;
833 int mid = memgr.FindMemZoneId(MemZA_Save);
834 Byte* buffg = memgr.GetMemZone(mid);
835 if (buffg == NULL) {
836 cout << " MultiDataSaver::run()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
837 setRC(21);
838 return;
839 }
840 tsmz = memgr.GetAuxData(mid)->filltime_;
841 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) { // Boucle sur les fibres
842 Byte* buff = memgr.GetMemZone(mid,fib);
843 if (buff == NULL) { // Ceci ne devrait pas arriver - suite au test buffg ci-dessus
844 cout << " MultiDataSaver::run()/ERROR memgr.GetMemZone(" << mid << "," << fib << ") -> NULL" << endl;
845 setRC(22);
846 return;
847 }
848 for(uint_4 i=0; i<memgr.NbPaquets(); i++) { // boucle sur les paquets
849 BRPaquet paq(NULL, buff+i*paqsz, paqsz);
850 bool cpaqok=pcheck[fib].Check(paq); // Verification du paquet / FrameCounter
851 if (cpaqok && fgfirstfctt) {
852 framecnt_first_[fib] = pcheck[fib].LastFrameNum();
853 timetag_first_[fib] = paq.TimeTag();
854 AddFitsKWStart(mff,bpar,tsmz);
855 fgfirstfctt=false;
856 }
857 if (savesig_)
858 header[fib] << hex << paq.HDRMarker() << " " << paq.TRLMarker() << " "
859 << paq.TimeTag2()<< " "<< paq.TimeTag1()<< " "
860 << paq.FrameCounter() << " " << paq.PaqLen() << endl;
861 if (savesig_) // Reza - Ecriture conditionnel fichier fits signal
862 mff[fib].WriteB(paq.Header(),paq.PaquetSize()); // ecriture tout le paquet (modif Mai 2009)
863 } // Fin de la boucle sur les paquets
864 } // Fin de la boucle sur les fibres
865 memgr.FreeMemZone(mid, MemZS_Saved);
866 } // Boucle sur les blocs dans un meme fichier
867 ts.SetNow();
868 filog << ts << " : OK data files " << endl;
869 cout << " MultiDataSaver::run() " << ts << " : OK data files " << endl;
870 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
871 if (savesig_) {
872 if (fgnulldev) strcpy(fname,"/dev/null");
873 else sprintf(fname,"%ssignal%d.fits",bpar.OutputDirectoryFib(fib).c_str(),(int)fnum);
874 }
875 else sprintf(fname,"MemDataBloc[%d]-NoDataFile",(int)fnum-1);
876 string pcsum = pcheck[fib].Summary();
877 filog << " Fib " << fib << " -> " << fname << " Stat:" << pcsum << endl;
878 cout << " Fib " << fib << " -> " << fname << " Stat:" << pcsum << endl;
879 }
880 if (savesig_) {
881 // Ajout mots-cle additionnels a tous les fichiers FITS
882 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++)
883 framecnt_last_[fib] = pcheck[fib].LastFrameNum();
884
885 AddFitsKWEnd(mff,bpar,tsmz);
886 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
887 header[fib].close();
888 mff[fib].Close();
889 }
890 }
891
892 } // Fin de boucle sur les fichiers
893 cout << " -------------------- MultiDataSaver::run() -------------------- " << endl;
894 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
895 cout << " MultiDataSaver/Summary Fib " << fib << endl;
896 pcheck[fib].Print(cout);
897 filog << " MultiDataSaver/Summary Fib " << fib << endl;
898 pcheck[fib].Print(filog);
899 }
900 cout << " ---------------------------------------------------------- " << endl;
901 ts.SetNow();
902 filog << " MultiDataSaver::run() - End of processing/run() " << ts << endl;
903
904 }
905 catch (MiniFITSException& exc) {
906 cout << " MultiDataSaver::run()/catched MiniFITSException " << exc.Msg() << endl;
907 setRC(3);
908 return;
909 }
910 catch(...) {
911 cout << " MultiDataSaver::run()/catched unknown ... exception " << endl;
912 setRC(4);
913 return;
914 }
915 setRC(0);
916 return;
917}
918
919
920/* --Methode-- */
921int MultiDataSaver::AddFitsKWStart(MiniFITSFile* mff, BRAcqConfig& acpar, TimeStamp& ts)
922{
923 string cdtu=ts.ToString();
924 string& skysrc=acpar.SkySource();
925 bool hassrc=false;
926 if (skysrc.length()>0) hassrc=true;
927 bool fgredpsz = acpar.GetParams().fgreducpsize;
928 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
929 mff[fib].AddKeyS("DATEOBS", cdtu.c_str(), " Observation Time (YYYY-MM-DDThh:mm:ss UT) ");
930 mff[fib].AddKeyS("TMSTART", cdtu.c_str(), " File Acqu. Start Time/Date ");
931 mff[fib].AddKeyD("ACQVER", acpar.AcqVersion(), " BAORadio Acq Software version ") ;
932 mff[fib].AddKeyS("ACQMODE", acpar.GetParams().AcqMode, " BAORadio Acq run mode" );
933 mff[fib].AddKeyS("BRPAQCFMT", BRPaquet::FmtConvToString(acpar.GetParams().GetDataConvFg()),
934 " BAORadio BRPaquet DataFormatConversion" );
935 mff[fib].AddKeyI("FIBERNUM", acpar.GetParams().FiberNum[fib], " Fiber number") ;
936 mff[fib].AddKeyI("FIBERID",memgr.FiberId(fib), " Fiber identifier (absolute id)");
937 if (hassrc)
938 mff[fib].AddKeyS("SKYSOURC", skysrc, " Source identification" );
939 if (fgredpsz) {
940 mff[fib].AddKeyS("REDPSZMOD", BRPaquet::ReducActionToString(acpar.GetParams().pqreducmode),
941 "PaquetSize Reduction Mode") ;
942 mff[fib].AddKeyI("REDPSZOF", acpar.GetParams().reducoffset," PaquetSize Reduction Offset") ;
943 }
944 }
945 return 0;
946}
947
948/* --Methode-- */
949int MultiDataSaver::AddFitsKWEnd(MiniFITSFile* mff, BRAcqConfig& acpar, TimeStamp& ts)
950{
951 string cdtu=ts.ToString();
952 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
953 mff[fib].AddKeyS("TMEND", cdtu.c_str(), " File Acqu. End Time/Date ");
954 mff[fib].AddKeyI("FCFIRST", framecnt_first_[fib], " First valid frame counter in file") ;
955 mff[fib].AddKeyI("FCLAST", framecnt_last_[fib], " Last valid frame counter in file") ;
956 mff[fib].AddKeyI("TTFIRST", timetag_first_[fib], " First valid timetag in file") ;
957 }
958 return 0;
959}
960
Note: See TracBrowser for help on using the repository browser.