source: Sophya/trunk/AddOn/TAcq/racqurw.cc@ 3991

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

Petite correction dans le lecteur PCIEMultiReader pour assurer le traitement des derniers paquets recus,
correction print de MultiDataSaver, ajout mode acq=hdcorrel (acquisition H.Deschamps) et MAJ numero de
version, Reza 02/03/2011
File size: 35.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 "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 stopreason_="??Unknown??";
494}
495
496/* --Methode-- */
497void PCIEMultiReader::run()
498{
499
500 struct timeval tv1,tv2;
501 gettimeofday(&tv1, NULL);
502
503 cout << " PCIEMultiReader::run() - Starting , NMaxMemZones=" << nmax_
504 << " memgr.NbPaquets(),PaqSz=" << memgr.NbPaquets() << " ," << memgr.PaqSize()
505 << " DMA-Paqsize " << packSize_ << " " << BRPaquet::FmtConvToString(swapall_) << endl;
506 setRC(1);
507
508 // sigaddset(&act.sa_mask,SIGINT); // pour proteger le transfert DMA
509 //sigaction(SIGINT,&act,NULL);
510 uint_4 paqszmm = memgr.PaqSize();
511 uint_4 paqsz = packSize_;
512 uint_4 dmasz = vec_pciw_[0]->TransferSize();
513 //DEL vec_pciw_[0]->StartTransfers();
514
515 BRPaqChecker pcheck[MAXNBFIB]; // Verification/comptage des paquets
516 Byte* Datas[MAXNBFIB];
517 Byte* tampon[MAXNBFIB] ;
518 Byte* predtampon=NULL; // tampon de recopie pour la reduction des tailles de paquets
519 Byte* nextpaq=NULL;
520 uint_4 off_acheval=0;
521
522 bool fgarret = false;
523
524 // Initialisation des tampons pour recopie des paquets a cheval pour chaque DMA
525 for (int i=0;i< (int)nbDma_ ;i++) {
526 tampon[i]= new Byte[paqsz];
527 vec_pciw_[i]->SetMaxWaitEndDMA(par_.first_maxkwedma_,par_.first_nretrydma_);
528 }
529 bool fgredpaq=par_.fgreducpsize;
530 if (fgredpaq) {
531 cout << " PCIEMultiReader::run() - PaquetSizeReduction - RedSize=" << par_.redpqsize
532 << " Offset=" << par_.reducoffset << " " << ((par_.reducneedcopy)?"NeedCopy":"NOCopy")
533 << " " << BRPaquet::ReducActionToString(par_.pqreducmode) << endl;
534 predtampon = new Byte[paqsz];
535 }
536
537#ifdef DEBUGPAQHDR
538 ofstream header[MAXNBFIB];
539 for(uint_4 fib=0; fib<nbDma_; fib++) {
540 char hfnm[128];
541 sprintf(hfnm, "./HDRCountPaqs%d.txt", fib);
542 header[fib].open(hfnm);
543 }
544#endif
545 uint_4 npaqfait[MAXNBFIB] ;
546 for (int i=0;i< (int)nbDma_ ;i++) npaqfait[i]=0;
547 // Byte* nextdma = locdata+((kmz%memgr.NbZones())*(paqsz*memgr.NbPaquets()));
548 uint_4 npaqfaitg = 0;
549 bool fg_change_timeout=true;
550 // for (uint_4 i=0; i<memgr.NbPaquets(); i += pktInDMATr) { // attention pktInDMATr paquets dans 1 seul DMA
551 while (npaqfaitg < nmax_*memgr.NbPaquets()) { // Boucle global G
552 if (fgarret) break;
553 if (stop_) break;
554
555 // Lancement des DMA
556 for (int dma=0; dma < (int)nbDma_ ;dma++) vec_pciw_[dma]->StartTransfers();
557 if ((npaqfaitg>1)&&fg_change_timeout) {
558 for (int i=0;i< (int)nbDma_ ;i++)
559 vec_pciw_[i]->SetMaxWaitEndDMA(par_.maxkwedma_,par_.nretrydma_);
560 fg_change_timeout=false;
561 }
562 // On pointe vers le debut de la zone a remplir aver le prochain DMA
563 //-- Zone memoire locale Byte* nextdma = buff+i*paqsz;
564
565 bool fgbaddma=false;
566 // On boucle sur les nbDma_ en attente de leurs terminaison
567 for (int dma=0; dma <(int) nbDma_ ;dma++) {
568 Datas[dma]=vec_pciw_[dma]->GetData();
569 if (Datas[dma] == NULL) { // No data Read in DMA
570 fgbaddma=true;
571 cout << "PCIEMultiReaderChecker/ERROR - DMA failed !" << endl;
572 vec_pciw_[dma]->PrintStatus(cout);
573 stopreason_="--Failed DMA--";
574 fgarret = true; break;
575 }
576 }
577 if (fgbaddma) continue;
578 uint_4 curoff=0;
579 //1- On traite le paquet a cheval, rempli partiellement avec le DMA d'avant si necessaire pour les n fibres
580 if (off_acheval > 0) { // IF Numero B
581 if ((paqsz-off_acheval)< dmasz) { // IF Numero A
582 for(uint_4 fib=0; fib<nbDma_; fib++)
583 memcpy((void *)((tampon[fib])+off_acheval), (void *)Datas[fib], paqsz-off_acheval);
584 curoff = paqsz-off_acheval; off_acheval = 0;
585 if ( MoveToNextTarget() ) {
586 cout << "PCIEMultiReader::run()/Error-A- MoveToNextTarget() returned true ->STOP 9" << endl;
587 setRC(9); fgarret=true; break;
588 }
589 for(uint_4 fib=0; fib<nbDma_; fib++) {
590 nextpaq=GetPaquetTarget(fib);
591 if (nextpaq == NULL) { // Cela ne devrait pas arriver
592 cout << "PCIEReader::run()/Error-A2- GetPaquetTarget(fib) returned NULL ->STOP 9" << endl;
593 setRC(9); fgarret=true; break;
594 }
595 // CHECK S'il faut faire une reduction de taille de paquet
596 if (fgredpaq) { // reduction taille de paquet
597 if (par_.reducneedcopy) {
598 BRPaquet paqc1(tampon[fib], predtampon, paqsz, swapall_);
599 BRPaquet paqc2(nextpaq, par_.redpqsize);
600 paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
601 }
602 else {
603 BRPaquet paqc1(tampon[fib], paqsz);
604 BRPaquet paqc2(nextpaq, par_.redpqsize);
605 paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
606 }
607 }
608 else {
609 BRPaquet paqc(tampon[fib], nextpaq, paqsz, swapall_);
610 }
611 BRPaquet paq(nextpaq, packSizeInMgr_);
612 npaqfait[fib]++;
613 if (fib==nbDma_-1) npaqfaitg++; // Ne pas oublier le compteur de paquets faits
614 pcheck[fib].Check(paq); // Verification du paquet / FrameCounter
615#ifdef DEBUGPAQHDR
616 header[fib] << dec << paq.FrameCounter()<< endl;
617#endif
618 }
619 }
620 else { // se rapporte au IF numero A
621 for(uint_4 fib=0; fib<nbDma_; fib++)
622 memcpy((void *)(tampon[fib]+off_acheval), (void *)Datas[fib], dmasz);
623 curoff =dmasz;
624 off_acheval = (dmasz+off_acheval);
625 }
626 } // Fin IF Numero B
627
628 //2- On traite les paquets complets qui se trouvent dans la zone du DMA
629 while ((curoff+paqsz)<=dmasz) { // while numero C
630 // if ((dma==nbDma_-1)&&(npaqfait >= nmax_* memgr.NbPaquets())) break;
631 if ( MoveToNextTarget() ) {
632 cout << "PCIEMultiReader::run()/Error-B- MoveToNextTarget() returned true ->STOP 9" << endl;
633 setRC(9); fgarret=true; break;
634 }
635 for(uint_4 fib=0; fib<nbDma_; fib++) {
636 if (npaqfait[fib] >= nmax_*memgr.NbPaquets()) continue;
637 nextpaq=GetPaquetTarget(fib);
638 if (nextpaq == NULL) { // Cela ne devrait pas arriver
639 cout << "PCIEReader::run()/Error-B2- GetPaquetTarget(fib) returned NULL ->STOP 9" << endl;
640 setRC(9); fgarret=true; break;
641 }
642 // CHECK S'il faut faire une reduction de taille de paquet
643 if (fgredpaq) { // reduction taille de paquet
644 if (par_.reducneedcopy) {
645 BRPaquet paqc1(Datas[fib]+curoff, predtampon, paqsz, swapall_);
646 BRPaquet paqc2(nextpaq, par_.redpqsize);
647 paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
648 }
649 else {
650 BRPaquet paqc1(Datas[fib]+curoff, paqsz);
651 BRPaquet paqc2(nextpaq, par_.redpqsize);
652 paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
653 }
654 }
655 else {
656 BRPaquet paqc(Datas[fib]+curoff, nextpaq, paqsz, swapall_);
657 }
658 BRPaquet paq(nextpaq, packSizeInMgr_);
659 npaqfait[fib]++;
660 if (fib==nbDma_-1) npaqfaitg++; // Ne pas oublier le compteur de paquets faits
661 pcheck[fib].Check(paq); // Verification du paquet / FrameCounter
662#ifdef DEBUGPAQHDR
663 header[fib] << dec << paq.FrameCounter()<< endl;
664#endif
665 }
666 curoff += paqsz; // On avance l'index dans le buffer du DMA
667 } // -- FIN traitement des paquets complets ds un DMA - FIN du while numero C
668 //3- On copie si besoin la fin du DMA dans la zone tampon
669 if (curoff < dmasz) { // IF numero D
670 off_acheval = dmasz-curoff;
671 for(uint_4 fib=0; fib<nbDma_; fib++)
672 memcpy(tampon[fib], (void*)(Datas[fib]+curoff), off_acheval);
673 // ne sert a rien curoff += off_acheval;
674 } // FIN du if numero D
675 } // FIN Boucle global G
676
677 if (npaqfaitg >= nmax_*memgr.NbPaquets()) stopreason_="--Max Nb paquets reached--";
678
679 MoveToNextTarget(); // Pour faire traiter le dernier paquet si plein
680 MZoneManage(true); //---- Nettoyage final
681
682 setRC(0);
683 gettimeofday(&tv2, NULL);
684 double tmelaps2 = (tv2.tv_sec-tv1.tv_sec)*1000.+(tv2.tv_usec-tv1.tv_usec)/1000.;
685 if (tmelaps2<0.1) tmelaps2=0.1;
686 cout << " ---------- PCIEMultiReader::run()-End StopReason: " << stopreason_ << endl
687 << " Summary NPaqFait=" << npaqfaitg << "------------- " << endl;
688 for (int dma=0; dma < (int)nbDma_ ;dma++) {
689 cout << " --Fib=" << dma << " NPaqFait=" << npaqfait[dma] << " TotTransfer="
690 << vec_pciw_[dma]->TotTransferBytes()/1024
691 << " kb , ElapsTime=" << tmelaps2 << " ms ->"
692 << (double)vec_pciw_[dma]->TotTransferBytes()/tmelaps2 << " kb/s" << endl;
693 pcheck[dma].Print(cout);
694 }
695 cout << " --------------------------------------------------------------------" << endl;
696
697 usleep(250000); // Attente de traitement du dernier paquet
698 memgr.Stop(); // Pour arreter les autres threads
699
700 for (int i=0;i< (int)nbDma_ ;i++) delete[] tampon[i];
701 if ((fgredpaq)&&predtampon) delete[] predtampon;
702#ifdef DEBUGPAQHDR
703 for(uint_4 fib=0; fib<nbDma_; fib++) header[fib].close();
704#endif
705
706 //DBG cout << " fin thread ========================" <<endl;
707 return;
708}
709
710/* --Methode-- */
711bool PCIEMultiReader::MZoneManage(bool fgclean) // Retourne true si probleme
712{
713 /* Pour debug
714 cout << " PCIEReader::MZoneManage() mid_=" << mid_ << " arg_npaq_= " << targ_npaq_
715 << " max_targ_npaq=" << max_targ_npaq << endl;
716 */
717 if (mid_ >= 0) {
718 memgr.GetAuxData(mid_)->filltime_.SetNow();
719 if (fgclean) memgr.FreeMemZone(mid_, MemZS_Free);
720 else memgr.FreeMemZone(mid_, MemZS_Filled);
721 }
722 mmbuf_ = NULL; targ_npaq_ = 0; mid_ = -2;
723 for (int fid=0;fid<(int)nbDma_ ;fid++) mmbufib_[fid]=NULL;
724 if (fgclean) return false;
725 mid_ = memgr.FindMemZoneId(MemZA_Fill);
726 mmbuf_ = memgr.GetMemZone(mid_);
727 if (mmbuf_==NULL) return true;
728 for (int fid=0;fid<(int)nbDma_ ;fid++) mmbufib_[fid]=memgr.GetMemZone(mid_,fid);
729 return false;
730}
731
732/*
733bool PCIEMultiReader::MZoneManage(int zone,bool fgclean) // Retourne true si probleme
734{
735 // Pour debug
736 //cout << " PCIEReader::MZoneManage() mid_=" << mid_ << " arg_npaq_= " << targ_npaq_
737 << " max_targ_npaq=" << max_targ_npaq << endl;
738 if (mid_[zone] >= 0) memgr[zone]->FreeMemZone(mid_[zone], MemZS_Filled);
739 mmbuf_[zone] = NULL; targ_npaq_[zone] = 0; mid_[zone] = -2;
740 if (fgclean) return false;
741 mid_[zone] = memgr[zone]->FindMemZoneId(MemZA_Fill);
742 mmbuf_[zone] = memgr[zone]->GetMemZone(mid_[zone]);
743 if (mmbuf_[zone]==NULL) return true;
744 return false;
745}
746*/
747
748/* --Methode-- */
749void PCIEMultiReader::Stop()
750{
751 // cout << " PCIEReaderChecker::stop() ........ STOP" <<endl;
752 stop_ = true;
753
754}
755
756
757//--------------------------------------------------------------------
758// Classe thread de sauvegarde sur fichiers avec gestion multifibres
759//--------------------------------------------------------------------
760
761/* --Methode-- */
762MultiDataSaver::MultiDataSaver(RAcqMemZoneMgr& mem)
763 : memgr(mem)
764{
765 BRAcqConfig bpar;
766 nfiles_ = bpar.MaxNbFiles();
767 nblocperfile_ = bpar.BlocPerFile();
768 nmax_ = nblocperfile_*nfiles_;
769 savesig_ = bpar.GetSaveFits(); // Si false, pas d'ecriture des fichiers FITS du signal
770 stop_ = false;
771}
772
773/* --Methode-- */
774void MultiDataSaver::Stop()
775{
776 // cout<< " MultiDataSaver:Stop ........ " << endl;
777 stop_=true;
778}
779
780/* --Methode-- */
781void MultiDataSaver::run()
782{
783 setRC(1);
784 BRPaqChecker pcheck[MAXNBFIB]; // Verification/comptage des paquets
785 BRAcqConfig bpar;
786 try {
787 TimeStamp ts;
788 cout << " MultiDataSaver::run() - Starting " << ts << " \n NbFiles=" << nfiles_ << " NBloc/File="
789 << nblocperfile_ << " NMaxMemZones=" << nmax_ << endl;
790 char fname[512];
791
792 sprintf(fname,"%s/msaver.log",bpar.OutputDirectory().c_str());
793 ofstream filog(fname);
794 filog << " MultiDataSaver::run() - starting log file " << ts << " NFibres= " << memgr.NbFibres() << endl;
795 filog << " NbFiles=" << nfiles_ << " NBloc/File=" << nblocperfile_ << " NMaxMemZones=" << nmax_ << endl;
796
797 // Fichiers entete ascii et signal FITS
798 ofstream header[MAXNBFIB];
799 MiniFITSFile mff[MAXNBFIB];
800
801 uint_4 fnum=0;
802 uint_4 paqsz = memgr.PaqSize();
803 cout << " ============================ MultiDataSaver::run() PaqSize " << paqsz <<endl;
804 bool fgfirstfctt=false;
805 bool fgnulldev = bpar.GetFileDevNull();
806 if (fgnulldev) {
807 cout << " MultiDataSaver::run()/Warning /dev/null path specified, filenames=/dev/null" << endl;
808 fgnulldev = true;
809 }
810 BRPaquet paq0(NULL, NULL, paqsz);
811 TimeStamp tsmz;
812 for (uint_4 nbFile=0;nbFile<nfiles_ ;nbFile++) {
813 if (stop_ ) break;
814 if (memgr.GetRunState() == MemZR_Stopped) break;
815
816 if (savesig_)
817 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
818 if (fgnulldev) strcpy(fname,"/dev/null");
819 else sprintf(fname,"%sHDRfits%d.txt",bpar.OutputDirectoryFib(fib).c_str(),fnum);
820 header[fib].open(fname);
821 }
822
823 uint_4 npaqperfile = memgr.NbPaquets()*nblocperfile_; // Nombre de paquets ecrits dans un fichier
824
825 if (savesig_) { //Reza - Ouverture conditionnel fichier
826 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
827 if (fgnulldev) strcpy(fname,"/dev/null");
828 else sprintf(fname,"%ssignal%d.fits",bpar.OutputDirectoryFib(fib).c_str(),(int)fnum);
829 // cout << " ***DBG** Opening file " << fname << endl;
830 mff[fib].Open(fname,MF_Write); //Reza - Ouverture conditionnel fichier
831 // Entete correspondant a l'ecriture tout le paquet - trailer compris (modif Mai 2009)
832 mff[fib].setDTypeNaxis(MF_Byte, paq0.PaquetSize(), npaqperfile);
833 // Sans TRAILER de paquet mff.setDTypeNaxis(MF_Byte, paq0.DataSize()+paq0.HeaderSize(), npaqperfile);
834 }
835 fnum++; fgfirstfctt=true;
836 }
837 else sprintf(fname,"MemDataBloc[%d]-NoDataFile",(int)fnum++);
838
839 for (uint_4 kmz=0; kmz<nblocperfile_; kmz++) {
840 if (stop_) break;
841 //DBG cout << " MultiDataSaver::run()- nbFile=" << nbFile << " kmz=" << kmz << endl;
842 int mid = memgr.FindMemZoneId(MemZA_Save);
843 Byte* buffg = memgr.GetMemZone(mid);
844 if (buffg == NULL) {
845 cout << " MultiDataSaver::run()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
846 setRC(21);
847 return;
848 }
849 tsmz = memgr.GetAuxData(mid)->filltime_;
850 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) { // Boucle sur les fibres
851 Byte* buff = memgr.GetMemZone(mid,fib);
852 if (buff == NULL) { // Ceci ne devrait pas arriver - suite au test buffg ci-dessus
853 cout << " MultiDataSaver::run()/ERROR memgr.GetMemZone(" << mid << "," << fib << ") -> NULL" << endl;
854 setRC(22);
855 return;
856 }
857 for(uint_4 i=0; i<memgr.NbPaquets(); i++) { // boucle sur les paquets
858 BRPaquet paq(NULL, buff+i*paqsz, paqsz);
859 bool cpaqok=pcheck[fib].Check(paq); // Verification du paquet / FrameCounter
860 if (cpaqok && fgfirstfctt) {
861 framecnt_first_[fib] = pcheck[fib].LastFrameNum();
862 timetag_first_[fib] = paq.TimeTag();
863 AddFitsKWStart(mff,bpar,tsmz);
864 fgfirstfctt=false;
865 }
866 if (savesig_)
867 header[fib] << hex << paq.HDRMarker() << " " << paq.TRLMarker() << " "
868 << paq.TimeTag2()<< " "<< paq.TimeTag1()<< " "
869 << paq.FrameCounter() << " " << paq.PaqLen() << endl;
870 if (savesig_) // Reza - Ecriture conditionnel fichier fits signal
871 mff[fib].WriteB(paq.Header(),paq.PaquetSize()); // ecriture tout le paquet (modif Mai 2009)
872 } // Fin de la boucle sur les paquets
873 } // Fin de la boucle sur les fibres
874 memgr.FreeMemZone(mid, MemZS_Saved);
875 } // Boucle sur les blocs dans un meme fichier
876 ts.SetNow();
877 filog << ts << " : OK data files " << endl;
878 cout << " MultiDataSaver::run() " << ts << " : OK data files " << endl;
879 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
880 if (savesig_) {
881 if (fgnulldev) strcpy(fname,"/dev/null");
882 else sprintf(fname,"%ssignal%d.fits",bpar.OutputDirectoryFib(fib).c_str(),(int)fnum-1);
883 }
884 else sprintf(fname,"MemDataBloc[%d]-NoDataFile",(int)fnum-1);
885 string pcsum = pcheck[fib].Summary();
886 filog << " Fib " << fib << " -> " << fname << " Stat:" << pcsum << endl;
887 cout << " Fib " << fib << " -> " << fname << " Stat:" << pcsum << endl;
888 }
889 if (savesig_) {
890 // Ajout mots-cle additionnels a tous les fichiers FITS
891 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++)
892 framecnt_last_[fib] = pcheck[fib].LastFrameNum();
893
894 AddFitsKWEnd(mff,bpar,tsmz);
895 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
896 header[fib].close();
897 mff[fib].Close();
898 }
899 }
900
901 } // Fin de boucle sur les fichiers
902 cout << " -------------------- MultiDataSaver::run() -------------------- " << endl;
903 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
904 cout << " MultiDataSaver/Summary Fib " << fib << endl;
905 pcheck[fib].Print(cout);
906 filog << " MultiDataSaver/Summary Fib " << fib << endl;
907 pcheck[fib].Print(filog);
908 }
909 cout << " ---------------------------------------------------------- " << endl;
910 ts.SetNow();
911 filog << " MultiDataSaver::run() - End of processing/run() " << ts << endl;
912
913 }
914 catch (MiniFITSException& exc) {
915 cout << " MultiDataSaver::run()/catched MiniFITSException " << exc.Msg() << endl;
916 setRC(3);
917 return;
918 }
919 catch(...) {
920 cout << " MultiDataSaver::run()/catched unknown ... exception " << endl;
921 setRC(4);
922 return;
923 }
924 setRC(0);
925 return;
926}
927
928
929/* --Methode-- */
930int MultiDataSaver::AddFitsKWStart(MiniFITSFile* mff, BRAcqConfig& acpar, TimeStamp& ts)
931{
932 string cdtu=ts.ToString();
933 string& skysrc=acpar.SkySource();
934 bool hassrc=false;
935 if (skysrc.length()>0) hassrc=true;
936 bool fgredpsz = acpar.GetParams().fgreducpsize;
937 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
938 mff[fib].AddKeyS("DATEOBS", cdtu.c_str(), " Observation Time (YYYY-MM-DDThh:mm:ss UT) ");
939 mff[fib].AddKeyS("TMSTART", cdtu.c_str(), " File Acqu. Start Time/Date ");
940 mff[fib].AddKeyD("ACQVER", acpar.AcqVersion(), " BAORadio Acq Software version ") ;
941 mff[fib].AddKeyS("ACQMODE", acpar.GetParams().AcqMode, " BAORadio Acq run mode" );
942 mff[fib].AddKeyS("BRPAQCFMT", BRPaquet::FmtConvToString(acpar.GetParams().GetDataConvFg()),
943 " BAORadio BRPaquet DataFormatConversion" );
944 mff[fib].AddKeyI("FIBERNUM", acpar.FiberNum(fib), " Fiber number") ;
945 mff[fib].AddKeyI("FIBERID",memgr.FiberId(fib), " Fiber identifier (absolute id)");
946 if (hassrc)
947 mff[fib].AddKeyS("SKYSOURC", skysrc, " Source identification" );
948 if (fgredpsz) {
949 mff[fib].AddKeyS("REDPSZMOD", BRPaquet::ReducActionToString(acpar.GetParams().pqreducmode),
950 "PaquetSize Reduction Mode") ;
951 mff[fib].AddKeyI("REDPSZOF", acpar.GetParams().reducoffset," PaquetSize Reduction Offset") ;
952 }
953 }
954 return 0;
955}
956
957/* --Methode-- */
958int MultiDataSaver::AddFitsKWEnd(MiniFITSFile* mff, BRAcqConfig& acpar, TimeStamp& ts)
959{
960 string cdtu=ts.ToString();
961 for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
962 mff[fib].AddKeyS("TMEND", cdtu.c_str(), " File Acqu. End Time/Date ");
963 mff[fib].AddKeyI("FCFIRST", framecnt_first_[fib], " First valid frame counter in file") ;
964 mff[fib].AddKeyI("FCLAST", framecnt_last_[fib], " Last valid frame counter in file") ;
965 mff[fib].AddKeyI("TTFIRST", timetag_first_[fib], " First valid timetag in file") ;
966 }
967 return 0;
968}
969
Note: See TracBrowser for help on using the repository browser.