source: Sophya/trunk/AddOn/TAcq/mfacq.cc@ 3921

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

Correction bug ds MonitorProc et ajout controle remplissage DataTable des visibiltes par BRVisibilityCalculator , cmv+Reza 02/12/2010
File size: 16.9 KB
Line 
1#include "mfacq.h"
2
3//---------------------------------------------------------------
4// Programme d'acquisition BAORadio multi-fibres/multi-threads
5// LAL - 2009 - 2010
6// R. Ansari, M.Taurigna
7//---------------------------------------------------------------
8
9static RAcqMemZoneMgr* pMmgr=NULL;
10static PCIEMultiReader* pPcierThr=NULL;
11static PCIEToEthernet* pPcie2Eth=NULL;
12static EthernetReader* pEthRdr=NULL;
13
14void Stop(int s)
15{
16 if (s == 9765) cout << " Stop after exception ..." << endl;
17 else printf("............. MAIN ... receive signal %d \n",s);
18 if (pMmgr != NULL) pMmgr->Stop();
19 if (pPcie2Eth !=NULL) pPcie2Eth->Stop();
20 if (pPcierThr !=NULL) pPcierThr->Stop();
21 if (pEthRdr !=NULL) pEthRdr->Stop();
22}
23
24
25//-----------------------------------------------------------------------
26//-------------------- le programme principal ---------------------------
27//-----------------------------------------------------------------------
28int main(int narg, char* arg[])
29{
30
31 int rc = 0;
32 cout << " ============= BAORadio / Acquisition : mfacq =================" << endl;
33 cout << " ===============================================================" << endl;
34 cout << " ========= " <<BAOR_ACQ_VER_STR <<BAOR_ACQ_VER << " ===========" << endl;
35 cout << " ===============================================================" << endl;
36
37
38 InitTim();
39
40 // Initialisation
41 TArrayInitiator _arri;
42 FitsIOServerInitiator _fiosi;
43
44
45 if ((narg > 1)&&(strcmp(arg[1],"-h"))==0) {
46 Usage(false);
47 return 1;
48 }
49 if (narg < 3) {
50 Usage(true);
51 return 3;
52 }
53
54 const char* desact[4] = {"PCIE_To_Ethernet", "Ethernet_To_Disk", "Ethernet_To_Visibilities","PCIE_DMA_To_Disk"};
55 try {
56 // Creation/initialisation parametres Acq
57 int act = DecodeArgs(narg,arg);
58 if ((act < 0)||(act>3)) {
59 cout << "mfacq/ERROR decoding arguments act=" << act << " -> exit -5" << endl;
60 return -5;
61 }
62 cout << " mfacq/INFO action: " << desact[act] << endl;
63 switch (act) {
64 case 0:
65 rc = PCIEToEthernetTransfer();
66 break;
67 case 1: // ethernet --> disk
68 rc = EthernetToMemoryAcq(false);
69 break;
70 case 2: // ethernet --> visibility calculation
71 rc = EthernetToMemoryAcq(true);
72 break;
73 case 3:
74 rc = MultiFibreAcq();
75 break;
76 }
77 }
78 catch (MiniFITSException& exc) {
79 cerr << " mfacq.cc catched MiniFITSException " << exc.Msg() << endl;
80 rc = 77;
81 }
82 catch (PCIEWException& exc) {
83 cerr << " mfacq.cc catched PCIEWException " << exc.Msg() << endl;
84 rc = 75;
85 }
86 catch (PThrowable& exc) {
87 cerr << " mfacq.cc catched Exception " << exc.Msg() << endl;
88 rc = 76;
89 }
90 catch (std::exception& sex) {
91 cerr << "\n mfacq.cc std::exception :"
92 << (string)typeid(sex).name() << "\n msg= "
93 << sex.what() << endl;
94 rc = 78;
95 }
96 catch (...) {
97 cerr << " mfacq.cc : Catched ... exception " << endl;
98 rc = 79;
99 }
100 cout << "mfacq[9] ----- END --- stopping acq program " << endl;
101 return rc;
102}
103
104
105/* --Nouvelle-Fonction-- */
106void Usage(bool fgshort)
107{
108 cout << " Usage: mfacq Action DataCardFileName [-dc DataCard -dc DataCard] [TargetMachines]" << endl;
109 if (fgshort) return;
110 cout << " o Action = -pci2disk , -pci2eth , -eth2disk , -eth2visib \n "
111 << " DataCardFile : File name for parameters list (DataCards) : \n"
112 << " fibres outpathname skysource paqsize dmasizekb nbfiles \n"
113 << " nblocperfile acqmode memmgr monitor reducpaqsz ... "<< endl;
114 cout << " o -dc DataCard : datacard appended for DataCardFileName" << endl;
115 cout << " o TargetMachines : List of target machines for PCIe2Ethernet (-pci2eth) \n " << endl;
116 return;
117}
118
119/* --Nouvelle-Fonction-- */
120int DecodeArgs(int narg, char* arg[])
121{
122 string action=arg[1];
123 int act = 0;
124 if ((action != "-pci2eth")&&(action != "-pci2disk")&&(action != "-eth2disk")&&(action != "-eth2visib")) {
125 cout << " mfacq/Error , Bad action argument : " << action << endl;
126 return -2;
127 }
128 if (action == "-pci2eth") act=0;
129 else if (action == "-eth2disk") act=1;
130 else if (action == "-eth2visib") act=2;
131 else if (action == "-pci2disk") act=3;
132
133 BRAcqConfig acpar;
134 char tmpdcname[128];
135 sprintf(tmpdcname,"%s/dc_XXXXXX",acpar.TmpDirectory().c_str());
136 mktemp(tmpdcname);
137 string dcardfile=tmpdcname;
138 dcardfile+=".d";
139 string pardcfile=arg[2];
140 char cmd[512];
141 sprintf(cmd,"cp %s %s",pardcfile.c_str(),dcardfile.c_str());
142 cout << " mfacq/DecodeArgs() executing command: " << cmd << endl;
143 int rc;
144 rc=system(cmd);
145 if (rc!=0) {
146 cout << " mfacq/ERROR system(cmd) , Rc=" << rc << endl;
147 return -3;
148 }
149
150 vector<string> oargs;
151 int aoff=3;
152 {
153 ofstream dcf( dcardfile.c_str(), ios_base::out|ios_base::app);
154 dcf << "#### Appended datacards from command line " << endl;
155 while (aoff<(narg-1)) {
156 if (strcmp(arg[aoff],"-dc") != 0) break;
157 dcf << arg[aoff+1] << endl;
158 aoff+=2;
159 }
160 if ((act==0)&&(aoff<narg)) { // default ethernet target destinations
161 dcf << "@ethrtargets ";
162 for(int jj=aoff; jj<narg; jj++) dcf << arg[jj] << " ";
163 dcf << endl;
164 }
165 }
166
167 acpar.GetParams().fgdoVisiC=false;
168 if (act == 2) acpar.GetParams().fgdoVisiC=true;
169
170 acpar.ReadParamFile(dcardfile);
171 acpar.CreateOutputDirectories();
172
173 sprintf(cmd,"cp %s %s/acqparm.d",dcardfile.c_str(),acpar.OutputDirectory().c_str());
174 cout << " mfacq/DecodeArgs() executing command: " << cmd << endl;
175 rc=system(cmd);
176 if (rc!=0) {
177 cout << " mfacq/ERROR system(cmd) , Rc=" << rc << endl;
178 return -3;
179 }
180
181 acpar.Print(cout);
182 string pflnm=acpar.OutputDirectory();
183 pflnm+="/params.log";
184 ofstream pfile(pflnm.c_str());
185 acpar.Print(pfile);
186
187 struct sigaction siact;
188 if (!acpar.GetParams().fg_hard_ctrlc) {
189 siact.sa_handler=Stop;
190 sigaction(SIGINT,&siact,NULL);
191 }
192
193 return act;
194}
195
196/* --Nouvelle-Fonction-- */
197int MultiFibreAcq()
198{
199 Timer tm("mfacq/MultiFibre");
200 cout << " ---------- mfacq/ MultiFibreAcq() ------------- " << endl;
201 PCIEWrapperInterface* pciwp[16]={NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,
202 NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL};
203
204 BRAcqConfig bpar;
205 BRParList& acpar=bpar.GetParams();
206
207 cout << " mfacq[1]/Info: Creating RAcqMemZoneMgr for" << acpar.NbFibers() << " fibers , nZones="
208 << acpar.nZones << " NbPaquet/Zone=" << acpar.nPaqZone
209 << " MmgrPaqSize=" << acpar.MMgrPaquetSize() << endl;
210 RAcqMemZoneMgr mmgr(acpar.nZones, bpar.NFibers(), acpar.nPaqZone, acpar.MMgrPaquetSize());
211 pMmgr=&mmgr;
212 if (acpar.fgdoProc && (acpar.stepProc>0)) {
213 // Dans ce cas, toutes les zones doivent passer ds le thread de monitoring
214 // pour etre au moins marque comme traite - seul 1/acpar.stepProc sont effectivement traite
215 mmgr.SetFinalizedMask(((uint_4)MemZS_Saved) | ((uint_4)MemZS_Proc));
216 cout << " mfacq[1.b]/Info: Mmgr.SetFinalizedMask( MemZS_Saved | MemZS_Proc )" << endl;
217 //mmgr.SetFinalizedMask( (uint_4)MemZS_Saved );
218 //cout << " mfacq[1.b]/Info: Mmgr.SetFinalizedMask( MemZS_Saved )" << endl;
219 }
220
221#ifndef NOPCIECARD
222 for (size_t i=0 ;i <acpar.NFibers() ;i++) {
223 UINT32 card=(acpar.FiberNum(i)-1)/2+1;
224 UINT32 cardfiber=(acpar.FiberNum(i)-1)%2;
225 cout <<"mfacq[2] CreatePCIEWrapperV6- indice " << i << "INIT card " << card
226 << " fibre " << cardfiber << endl;
227 pciwp[i] = CreatePCIEWrapperV6(card,acpar.PatternSize(),acpar.DMASizeBytes(),acpar.activate_pattern,cardfiber);
228 }
229#else
230 TestPCIWrapperNODMA pciw1(bpar.RecvPaquetSize(), acpar.nopciLossRate);
231 TestPCIWrapperNODMA pciw2(bpar.RecvPaquetSize(), acpar.nopciLossRate);
232 TestPCIWrapperNODMA pciw3(bpar.RecvPaquetSize(), acpar.nopciLossRate);
233 TestPCIWrapperNODMA pciw4(bpar.RecvPaquetSize(), acpar.nopciLossRate);
234 TestPCIWrapperNODMA pciw5(bpar.RecvPaquetSize(), acpar.nopciLossRate);
235 TestPCIWrapperNODMA pciw6(bpar.RecvPaquetSize(), acpar.nopciLossRate);
236 TestPCIWrapperNODMA pciw7(bpar.RecvPaquetSize(), acpar.nopciLossRate);
237 TestPCIWrapperNODMA pciw8(bpar.RecvPaquetSize(), acpar.nopciLossRate);
238 pciwp[0] = &pciw1;
239 pciwp[1] = &pciw2;
240 pciwp[2] = &pciw3;
241 pciwp[3] = &pciw4;
242 pciwp[4] = &pciw5;
243 pciwp[5] = &pciw6;
244 pciwp[6] = &pciw7;
245 pciwp[7] = &pciw8;
246#endif
247
248 cout <<"mfacq[2] Creating MultiDataSaver and MonitorProc thread objects ... " << endl;
249 MultiDataSaver DsThr(mmgr); // Utilise les parametres globaux BRAcqConfig
250 string ppath=bpar.OutputDirectory();
251 MonitorProc PrThr(mmgr);
252 cout << "mfacq[3] Creating PCIEMultiReader thread object " << endl;
253 vector<PCIEWrapperInterface*> vec_pciw;
254 for (size_t i=0 ;i<bpar.NFibers();i++) {
255 pciwp[i]->SetFiberNumId(bpar.FiberNum(i), bpar.FiberId(i));
256 vec_pciw.push_back( pciwp[i]);
257 // cout << " mfacq[3.b]/Debug - pciwp[" << i << "] " << hex << pciwp[i] << dec << endl;
258 }
259 PCIEMultiReader PcierThr(vec_pciw, mmgr, bpar.GetParams());
260 // usleep(200); attente au cas ou ...
261 pPcierThr=&PcierThr;
262 tm.Split("Threads created");
263 if (acpar.fgdoProc>0)
264 cout << "mfacq[4] - starting three threads: PCIEMultiReader, MultiDataSaver, MonitorProc ... " << endl;
265 else
266 cout << "mfacq[4] - starting two threads: PCIEMultiReader, MultiDataSaver ... " << endl;
267
268 PcierThr.start();
269 DsThr.start();
270 if (acpar.fgdoProc>0) { // On ne demarre que si au moins NMaxProc>0
271 PrThr.start();
272 }
273
274 // On attend avant de declencher la terminaison des threads
275 usleep(200000);
276
277 cout << "mfacq[5] - Waiting for threads to finish ... " << endl;
278 PcierThr.join();
279 DsThr.join();
280 mmgr.Stop();
281 if (acpar.fgdoProc) { // On n'attend la fin que si le thread a ete demarre (NMaxProc>0)
282 PrThr.join();
283 }
284 pMmgr=NULL;
285 cout << "mfacq[6] ---------- threads finished ---------------- " << endl;
286 tm.Split("Threads Finished");
287
288 mmgr.Print(cout);
289#ifndef NOPCIECARD
290 for (size_t i=0; i<acpar.NbFibers(); i++) {
291 UINT32 card=(acpar.FiberNum(i)-1)/2+1;
292 UINT32 cardfiber=(acpar.FiberNum(i)-1)%2;
293 DeletePCIEWrapperV6(card,cardfiber);
294 }
295#endif
296return 0;
297
298}
299
300/* --Nouvelle-Fonction-- */
301int PCIEToEthernetTransfer()
302{
303 Timer tm("mfacq/PCIEToEthernetTransfer");
304 cout << " ---------- mfacq/ PCIEToEthernetTransfer() ------------- " << endl;
305 PCIEWrapperInterface* pciwp[16]={NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,
306 NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL};
307
308 pMmgr=NULL;
309 BRAcqConfig bpar;
310 BRParList& acpar=bpar.GetParams();
311
312 cout << " mfacq[1]/Info: PCIEToEthernet for" << acpar.NbFibers() << " fibers , nZones="
313 << acpar.nZones << " NbPaquet/Zone=" << acpar.nPaqZone
314 << " MmgrPaqSize=" << acpar.MMgrPaquetSize() << endl;
315#ifndef NOPCIECARD
316 for (size_t i=0 ;i <acpar.NFibers() ;i++) {
317 UINT32 card=(acpar.FiberNum(i)-1)/2+1;
318 UINT32 cardfiber=(acpar.FiberNum(i)-1)%2;
319 cout <<"mfacq[2] CreatePCIEWrapperV6- indice " << i << "INIT card " << card
320 << " fibre " << cardfiber << endl;
321 pciwp[i] = CreatePCIEWrapperV6(card,acpar.PatternSize(),acpar.DMASizeBytes(),acpar.activate_pattern,cardfiber);
322 }
323#else
324 TestPCIWrapperNODMA pciw1(bpar.RecvPaquetSize(), acpar.nopciLossRate);
325 TestPCIWrapperNODMA pciw2(bpar.RecvPaquetSize(), acpar.nopciLossRate);
326 TestPCIWrapperNODMA pciw3(bpar.RecvPaquetSize(), acpar.nopciLossRate);
327 TestPCIWrapperNODMA pciw4(bpar.RecvPaquetSize(), acpar.nopciLossRate);
328 TestPCIWrapperNODMA pciw5(bpar.RecvPaquetSize(), acpar.nopciLossRate);
329 TestPCIWrapperNODMA pciw6(bpar.RecvPaquetSize(), acpar.nopciLossRate);
330 TestPCIWrapperNODMA pciw7(bpar.RecvPaquetSize(), acpar.nopciLossRate);
331 TestPCIWrapperNODMA pciw8(bpar.RecvPaquetSize(), acpar.nopciLossRate);
332 pciwp[0] = &pciw1;
333 pciwp[1] = &pciw2;
334 pciwp[2] = &pciw3;
335 pciwp[3] = &pciw4;
336 pciwp[4] = &pciw5;
337 pciwp[5] = &pciw6;
338 pciwp[6] = &pciw7;
339 pciwp[7] = &pciw8;
340#endif
341
342 vector<PCIEWrapperInterface*> vec_pciw;
343 for (size_t i=0 ;i<bpar.NFibers();i++) {
344 pciwp[i]->SetFiberNumId(bpar.FiberNum(i), bpar.FiberId(i));
345 vec_pciw.push_back( pciwp[i]);
346 }
347 cout <<"mfacq[2] Creating PCIEToEthernet thread object " << endl;
348 PCIEToEthernet pci2eth(vec_pciw, bpar.GetParams().GetEthTargets(), bpar.GetParams(), bpar.GetParams().tcpportid);
349 pci2eth.SetPrintLevel(acpar.prtlevel_, acpar.prtmodulo_);
350 // usleep(200); attente au cas ou ...
351 pPcie2Eth=&pci2eth;
352 tm.Split("Threads created");
353 cout << "mfacq[3] - starting one threads: PCIEToEthernet... " << endl;
354 pci2eth.start();
355
356 // On attend avant de declencher la terminaison des threads
357 usleep(200000);
358
359 cout << "mfacq[5] - Waiting for threads to finish ... " << endl;
360 pci2eth.join();
361 cout << "mfacq[6] ---------- threads finished ---------------- " << endl;
362 tm.Split("Threads Finished");
363
364#ifndef NOPCIECARD
365 for (size_t i=0; i<acpar.NbFibers(); i++) {
366 UINT32 card=(acpar.FiberNum(i)-1)/2+1;
367 UINT32 cardfiber=(acpar.FiberNum(i)-1)%2;
368 DeletePCIEWrapperV6(card,cardfiber);
369 }
370#endif
371return 0;
372}
373
374
375/* --Nouvelle-Fonction-- */
376int EthernetToMemoryAcq(bool fgviscal)
377{
378 Timer tm("mfacq/EthernetToMemoryAcq");
379 cout << " ---------- mfacq/ EthernetToMemoryAcq() ------------- " << endl;
380
381 BRAcqConfig bpar;
382 BRParList& acpar=bpar.GetParams();
383
384 if (fgviscal) cout << "mfacq[0]/EthernetToMemoryAcq() : On the fly Visibility Calculation " << endl;
385 else cout << "mfacq[0]/EthernetToMemoryAcq() : data dumped to disk " << endl;
386 cout << " mfacq[1]/Info: Creating RAcqMemZoneMgr for" << acpar.NbEthLinks() << " Eth-links , nZones="
387 << acpar.nZones << " NbPaquet/Zone=" << acpar.nPaqZone
388 << " MmgrPaqSize=" << acpar.MMgrPaquetSize() << endl;
389 RAcqMemZoneMgr mmgr(acpar.nZones, acpar.NbEthLinks(), acpar.nPaqZone, acpar.MMgrPaquetSize());
390 pMmgr=&mmgr;
391
392
393 string strfmask;
394 uint_4 fmask=0;
395 MemZStatus mskmon=MemZS_Proc;
396 if (fgviscal) { // gestion de finalized_mask lors du calcul de visibilites
397 if (acpar.nbcalgrpVisiC<1) acpar.nbcalgrpVisiC=1;
398 if (acpar.nbcalgrpVisiC>4) acpar.nbcalgrpVisiC=4;
399 cout << " mfacq[1.b]/Info: NbCalGrpVisiC = " << acpar.nbcalgrpVisiC << " (1<=NbCalGrpVisiC<=4)" << endl;
400 fmask=(uint_4)MemZS_ProcA; strfmask="MemZS_ProcA"; mskmon=MemZS_ProcB;
401 if (acpar.nbcalgrpVisiC>1)
402 { fmask |= (uint_4)MemZS_ProcB; strfmask+=" | MemZS_ProcB"; mskmon=MemZS_ProcC; }
403 if (acpar.nbcalgrpVisiC>2)
404 { fmask |= (uint_4)MemZS_ProcC; strfmask+=" | MemZS_ProcC"; mskmon=MemZS_ProcD; }
405 if (acpar.nbcalgrpVisiC>3)
406 { fmask |= (uint_4)MemZS_ProcD; strfmask+=" | MemZS_ProcD"; mskmon=MemZS_ProcE; }
407 }
408 else {
409 fmask=(uint_4)MemZS_Saved; strfmask="MemZS_Saved"; mskmon=MemZS_Proc;
410 }
411 if (acpar.fgdoProc)
412 { fmask |= (uint_4)mskmon; strfmask+=" | MemZS_Proc[Monotoring]"; }
413 // Dans ce cas, toutes les zones doivent passer ds le thread de monitoring
414 // pour etre au moins marque comme traite - seul 1/acpar.stepProc sont effectivement traite
415
416 mmgr.SetFinalizedMask(fmask);
417 cout << " mfacq[1.b]/Info: Mmgr.SetFinalizedMask( " << strfmask << " )" << endl;
418
419 if (fgviscal)
420 cout <<"mfacq[2] Creating BRVisCalcGroup, NbVisibCalculator=" << acpar.nbcalgrpVisiC
421 << " and MonitorProc " << endl;
422 else
423 cout <<"mfacq[2] Creating MultiDataSaver and MonitorProc thread objects ... " << endl;
424 MultiDataSaver DsThr(mmgr); // Utilise les parametres globaux BRAcqConfig
425 string ppath=bpar.OutputDirectory();
426 MonitorProc PrThr(mmgr);
427 PrThr.SetMemZAction( RAcqMemZoneMgr::Convert_Status2Action( mskmon ) );
428 BRVisCalcGroup VCGThr(acpar.nbcalgrpVisiC, mmgr, bpar.OutputDirectory(), acpar.nmeanVisiC,
429 acpar.firstpairVisiC, acpar.nbpairsVisiC, acpar.fgpimpVisiC, acpar.nthrVisiC);
430 VCGThr.SelectFreqBinning(acpar.freqminVisiC, acpar.freqmaxVisiC, acpar.nbinfreqVisiC);
431 VCGThr.ActivateVisDTable(acpar.fgfdtVisiC);
432
433 cout << "mfacq[3] Creating EthernetReader thread object " << endl;
434 EthernetReader ethrdr(mmgr, bpar.GetParams(), bpar.GetParams().tcpportid);
435 ethrdr.SetReadMode(acpar.ethr_forcesamefc_, acpar.ethr_sfc_maxdpc_,acpar.ethr_sfc_maxresync_);
436 ethrdr.WaitENDMsg4Terminate(acpar.ethr_waitendmsg_);
437 ethrdr.SetPrintLevel(acpar.prtlevel_, acpar.prtmodulo_);
438
439 // usleep(200); attente au cas ou ...
440 pEthRdr=&ethrdr;
441 tm.Split("Threads created");
442 cout << "mfacq[4] - starting EthernetReader thread object ..." << endl;
443 ethrdr.start();
444 if (fgviscal) {
445 cout << "mfacq[4.b] - starting Visibility calculator threads " << endl;
446 VCGThr.start();
447 }
448 else {
449 cout << "mfacq[4.b] - starting MultiDataSaver " << endl;
450 DsThr.start();
451 }
452 if (acpar.fgdoProc) { // demarrage (optionnel) du thread de monitoring
453 cout << " mfacq[4.c] - starting MonitorProc thread object " << endl;
454 PrThr.start();
455 }
456
457 // On attend avant de declencher la terminaison des threads
458 usleep(200000);
459
460 cout << "mfacq[5] - Waiting for threads to finish ... " << endl;
461 ethrdr.join();
462 if (fgviscal) VCGThr.join();
463 else DsThr.join();
464 mmgr.Stop();
465 if (acpar.fgdoProc) { // On n'attend la fin que si le thread a ete demarre (NMaxProc>0)
466 PrThr.join();
467 }
468 pMmgr=NULL;
469 cout << "mfacq[6] ---------- threads finished ---------------- " << endl;
470 tm.Split("Threads Finished");
471
472 mmgr.Print(cout);
473
474return 0;
475
476}
477
478
Note: See TracBrowser for help on using the repository browser.