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

Last change on this file since 3671 was 3671, checked in by ansari, 16 years ago
  • Nettoyage complet des programmes d'acquisition
  • Ajout methode BRPaquet::CopyFrom() pour reduction de paquet
  • Introduction des classes BRParList , BRConfList , BRAcqConfig
  • Introduction du nouveau programme acquisition mfacq.cc (MultiFibresAcq)

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