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

#include "mfacq.h"

//---------------------------------------------------------------
// Programme d'acquisition BAORadio multi-fibres/multi-threads 
//  LAL -   2009 - 2010
//     R. Ansari, M.Taurigna
//---------------------------------------------------------------

static RAcqMemZoneMgr* pMmgr=NULL;
static  PCIEMultiReader* pPcierThr=NULL;
static  PCIEToEthernet* pPcie2Eth=NULL;
static  EthernetReader* pEthRdr=NULL;

void Stop(int s)
{ 
  if (s == 9765) cout << " Stop after exception ..." << endl;
  else printf("............. MAIN ... receive signal %d \n",s);
  if (pMmgr != NULL) pMmgr->Stop();
  if (pPcie2Eth !=NULL) pPcie2Eth->Stop();
  if (pPcierThr !=NULL) pPcierThr->Stop();
  if (pEthRdr !=NULL) pEthRdr->Stop();
}


//-----------------------------------------------------------------------
//-------------------- le programme principal ---------------------------
//-----------------------------------------------------------------------
int main(int narg, char* arg[])
{ 
  
  int rc = 0;
  cout << " ============= BAORadio / Acquisition : mfacq  =================" << endl;
  cout << " ===============================================================" << endl;
  cout << " =========  " <<BAOR_ACQ_VER_STR <<BAOR_ACQ_VER << " ===========" << endl;
  cout << " ===============================================================" << endl;
  
 
  InitTim();
 
  // Initialisation 
  TArrayInitiator _arri;
  FitsIOServerInitiator _fiosi;


  if ((narg > 1)&&(strcmp(arg[1],"-h"))==0) {
    Usage(false);
    return 1;
  }
  if (narg < 3) {
    Usage(true);
    return 3;
  }
  
  const char* desact[4] = {"PCIE_To_Ethernet", "Ethernet_To_Disk", "Ethernet_To_Visibilities","PCIE_DMA_To_Disk"};
  try {
    // Creation/initialisation parametres Acq
    int act = DecodeArgs(narg,arg);
    if ((act < 0)||(act>3))  { 
      cout << "mfacq/ERROR decoding arguments act=" << act << " -> exit -5" << endl;
      return -5;
    }
    cout << " mfacq/INFO action: " << desact[act] << endl; 
    switch (act) {
    case 0:
      rc = PCIEToEthernetTransfer();
      break;
    case 1:    // ethernet --> disk 
      rc = EthernetToMemoryAcq(false);
      break;
    case 2:   // ethernet --> visibility calculation 
      rc = EthernetToMemoryAcq(true);
      break;
    case 3:
      rc = MultiFibreAcq();
      break;
    }    
  }
  catch (MiniFITSException& exc) {
    cerr << " mfacq.cc catched MiniFITSException " << exc.Msg() << endl;
    rc = 77;
  } 
  catch (PCIEWException& exc) {
    cerr << " mfacq.cc catched PCIEWException " << exc.Msg() << endl;
    rc = 75;
  }  
  catch (PThrowable& exc) {
    cerr << " mfacq.cc catched Exception " << exc.Msg() << endl;
    rc = 76;
  }  
  catch (std::exception& sex) {
    cerr << "\n mfacq.cc std::exception :" 
         << (string)typeid(sex).name() << "\n msg= " 
         << sex.what() << endl;
    rc = 78;
  }
  catch (...) {
    cerr << "  mfacq.cc : Catched ... exception " << endl;
    rc = 79;
  }
  cout << "mfacq[9] ----- END --- stopping acq program " << endl;
  return rc;
}


/* --Nouvelle-Fonction-- */
void Usage(bool fgshort) 
{
  cout << " Usage: mfacq Action DataCardFileName [-dc DataCard -dc DataCard] [TargetMachines]" << endl;
  if (fgshort) return;
  cout << " o Action = -pci2disk , -pci2eth , -eth2disk , -eth2visib \n " 
       << "   DataCardFile : File name for parameters list (DataCards) : \n"
       << "     fibres outpathname skysource paqsize dmasizekb nbfiles \n" 
       << "     nblocperfile acqmode memmgr monitor reducpaqsz ... "<< endl;
  cout << " o -dc DataCard : datacard appended for DataCardFileName" << endl;
  cout << " o TargetMachines : List of target machines for PCIe2Ethernet (-pci2eth) \n " << endl;
  return;
}

/* --Nouvelle-Fonction-- */
int DecodeArgs(int narg, char* arg[])
{
  string action=arg[1];
  int act = 0;
  if ((action != "-pci2eth")&&(action != "-pci2disk")&&(action != "-eth2disk")&&(action != "-eth2visib")) {
    cout << " mfacq/Error , Bad action argument : " << action << endl;
    return -2;
  }
  if (action == "-pci2eth")  act=0;
  else if (action == "-eth2disk") act=1;  
  else if (action == "-eth2visib") act=2; 
  else if (action == "-pci2disk")  act=3;

  BRAcqConfig acpar;
  char tmpdcname[128];
  sprintf(tmpdcname,"%s/dc_XXXXXX",acpar.TmpDirectory().c_str());
  mktemp(tmpdcname);
  string dcardfile=tmpdcname;
  dcardfile+=".d";
  string pardcfile=arg[2];
  char cmd[512];
  sprintf(cmd,"cp %s %s",pardcfile.c_str(),dcardfile.c_str());
  cout << " mfacq/DecodeArgs() executing command: " << cmd << endl;
  int rc;
  rc=system(cmd);
  if (rc!=0) { 
    cout << " mfacq/ERROR system(cmd) , Rc=" << rc << endl;
    return -3;
  }
  
  vector<string> oargs;
  int aoff=3;
  {
    ofstream dcf( dcardfile.c_str(), ios_base::out|ios_base::app);
    dcf << "#### Appended datacards from command line " << endl;
    while (aoff<(narg-1)) {
      if (strcmp(arg[aoff],"-dc") != 0) break;
      dcf << arg[aoff+1] << endl;
      aoff+=2;
    }
    if ((act==0)&&(aoff<narg)) {  // default ethernet target destinations 
      dcf << "@ethrtargets  ";
      for(int jj=aoff; jj<narg; jj++)  dcf << arg[jj] << " ";
      dcf << endl;
    }
  }

  acpar.GetParams().fgdoVisiC=false;
  if (act == 2)  acpar.GetParams().fgdoVisiC=true;
  
  acpar.ReadParamFile(dcardfile);
  acpar.CreateOutputDirectories();

  sprintf(cmd,"cp %s %s/acqparm.d",dcardfile.c_str(),acpar.OutputDirectory().c_str());
  cout << " mfacq/DecodeArgs() executing command: " << cmd << endl;
  rc=system(cmd);
  if (rc!=0) { 
    cout << " mfacq/ERROR system(cmd) , Rc=" << rc << endl;
    return -3;
  }

  acpar.Print(cout);
  string pflnm=acpar.OutputDirectory();
  pflnm+="/params.log";
  ofstream pfile(pflnm.c_str());
  acpar.Print(pfile);

  struct sigaction siact;  
  if (!acpar.GetParams().fg_hard_ctrlc) {
    siact.sa_handler=Stop;
    sigaction(SIGINT,&siact,NULL);        
  }

  return act;
}
 
/* --Nouvelle-Fonction-- */
int MultiFibreAcq() 
{
  Timer tm("mfacq/MultiFibre");
  cout << " ---------- mfacq/ MultiFibreAcq()  ------------- " << endl;
  PCIEWrapperInterface* pciwp[16]={NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,
                                   NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL};
    
  BRAcqConfig bpar;
  BRParList& acpar=bpar.GetParams();
  
  cout << " mfacq[1]/Info: Creating RAcqMemZoneMgr for" << acpar.NbFibers() << " fibers , nZones=" 
       << acpar.nZones << " NbPaquet/Zone=" << acpar.nPaqZone 
       << " MmgrPaqSize=" << acpar.MMgrPaquetSize() << endl;
  RAcqMemZoneMgr mmgr(acpar.nZones, bpar.NFibers(), acpar.nPaqZone, acpar.MMgrPaquetSize());
  pMmgr=&mmgr;  
  if (acpar.fgdoProc && (acpar.stepProc>0)) {
    // Dans ce cas, toutes les zones doivent passer ds le thread de monitoring 
    // pour etre au moins marque comme traite - seul 1/acpar.stepProc sont effectivement traite 
    mmgr.SetFinalizedMask(((uint_4)MemZS_Saved) | ((uint_4)MemZS_Proc));
    cout << " mfacq[1.b]/Info: Mmgr.SetFinalizedMask( MemZS_Saved | MemZS_Proc )" << endl;
    //mmgr.SetFinalizedMask( (uint_4)MemZS_Saved );
    //cout << " mfacq[1.b]/Info: Mmgr.SetFinalizedMask( MemZS_Saved  )" << endl;
  }
  
#ifndef NOPCIECARD  
  for (size_t i=0 ;i <acpar.NFibers() ;i++) {
    UINT32 card=(acpar.FiberNum(i)-1)/2+1;
    UINT32 cardfiber=(acpar.FiberNum(i)-1)%2;
    cout <<"mfacq[2] CreatePCIEWrapperV6- indice " << i <<  "INIT card " << card 
         << " fibre " << cardfiber << endl;
    pciwp[i] =  CreatePCIEWrapperV6(card,acpar.PatternSize(),acpar.DMASizeBytes(),acpar.activate_pattern,cardfiber);
  }
#else 
  TestPCIWrapperNODMA pciw1(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  TestPCIWrapperNODMA pciw2(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  TestPCIWrapperNODMA pciw3(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  TestPCIWrapperNODMA pciw4(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  TestPCIWrapperNODMA pciw5(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  TestPCIWrapperNODMA pciw6(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  TestPCIWrapperNODMA pciw7(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  TestPCIWrapperNODMA pciw8(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  pciwp[0] = &pciw1;
  pciwp[1] = &pciw2;
  pciwp[2] = &pciw3;
  pciwp[3] = &pciw4;
  pciwp[4] = &pciw5;
  pciwp[5] = &pciw6;  
  pciwp[6] = &pciw7;
  pciwp[7] = &pciw8;  
#endif 

  cout <<"mfacq[2] Creating MultiDataSaver and  MonitorProc thread objects ... " << endl;
  MultiDataSaver DsThr(mmgr);   //   Utilise les parametres globaux BRAcqConfig
  string ppath=bpar.OutputDirectory();
  MonitorProc PrThr(mmgr);
  cout << "mfacq[3] Creating   PCIEMultiReader thread object     " << endl;
  vector<PCIEWrapperInterface*> vec_pciw;
  for  (size_t i=0 ;i<bpar.NFibers();i++)  {
    pciwp[i]->SetFiberNumId(bpar.FiberNum(i), bpar.FiberId(i));
    vec_pciw.push_back( pciwp[i]);
    //    cout << " mfacq[3.b]/Debug - pciwp[" << i << "] " << hex << pciwp[i] << dec << endl;
  }
  PCIEMultiReader PcierThr(vec_pciw, mmgr, bpar.GetParams());
  //  usleep(200);  attente au cas ou ...
  pPcierThr=&PcierThr;
  tm.Split("Threads created");  
  if (acpar.fgdoProc>0) 
    cout << "mfacq[4] - starting three threads: PCIEMultiReader, MultiDataSaver, MonitorProc ... " << endl;
  else 
    cout << "mfacq[4] - starting two threads: PCIEMultiReader, MultiDataSaver  ... " << endl;

  PcierThr.start();
  DsThr.start();
  if (acpar.fgdoProc>0) { // On ne demarre que si au moins NMaxProc>0
    PrThr.start();
  }
  
  // On attend avant de declencher la terminaison des threads
  usleep(200000);
  
  cout << "mfacq[5] -  Waiting for threads to finish ... " << endl;  
  PcierThr.join();
  DsThr.join();
  mmgr.Stop();
  if (acpar.fgdoProc) { // On n'attend la fin que si le thread a ete demarre (NMaxProc>0)
    PrThr.join();
  }
  pMmgr=NULL;  
  cout << "mfacq[6] ---------- threads finished ---------------- " << endl;
  tm.Split("Threads Finished");  
 
  mmgr.Print(cout);
#ifndef NOPCIECARD
  for (size_t i=0; i<acpar.NbFibers(); i++) {
    UINT32 card=(acpar.FiberNum(i)-1)/2+1;
    UINT32 cardfiber=(acpar.FiberNum(i)-1)%2;
    DeletePCIEWrapperV6(card,cardfiber);
  }
#endif    
return 0;

}

/* --Nouvelle-Fonction-- */
int PCIEToEthernetTransfer()
{
  Timer tm("mfacq/PCIEToEthernetTransfer");
  cout << " ---------- mfacq/ PCIEToEthernetTransfer()  ------------- " << endl;
  PCIEWrapperInterface* pciwp[16]={NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,
                                   NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL};

  pMmgr=NULL;    
  BRAcqConfig bpar;
  BRParList& acpar=bpar.GetParams();
  
  cout << " mfacq[1]/Info: PCIEToEthernet for" << acpar.NbFibers() << " fibers , nZones=" 
       << acpar.nZones << " NbPaquet/Zone=" << acpar.nPaqZone 
       << " MmgrPaqSize=" << acpar.MMgrPaquetSize() << endl;
#ifndef NOPCIECARD  
  for (size_t i=0 ;i <acpar.NFibers() ;i++) {
    UINT32 card=(acpar.FiberNum(i)-1)/2+1;
    UINT32 cardfiber=(acpar.FiberNum(i)-1)%2;
    cout <<"mfacq[2] CreatePCIEWrapperV6- indice " << i <<  "INIT card " << card 
         << " fibre " << cardfiber << endl;
    pciwp[i] =  CreatePCIEWrapperV6(card,acpar.PatternSize(),acpar.DMASizeBytes(),acpar.activate_pattern,cardfiber);
  }
#else 
  TestPCIWrapperNODMA pciw1(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  TestPCIWrapperNODMA pciw2(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  TestPCIWrapperNODMA pciw3(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  TestPCIWrapperNODMA pciw4(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  TestPCIWrapperNODMA pciw5(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  TestPCIWrapperNODMA pciw6(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  TestPCIWrapperNODMA pciw7(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  TestPCIWrapperNODMA pciw8(bpar.RecvPaquetSize(), acpar.nopciLossRate);
  pciwp[0] = &pciw1;
  pciwp[1] = &pciw2;
  pciwp[2] = &pciw3;
  pciwp[3] = &pciw4;
  pciwp[4] = &pciw5;
  pciwp[5] = &pciw6;  
  pciwp[6] = &pciw7;
  pciwp[7] = &pciw8;  
#endif 

  vector<PCIEWrapperInterface*> vec_pciw;
  for  (size_t i=0 ;i<bpar.NFibers();i++)  { 
    pciwp[i]->SetFiberNumId(bpar.FiberNum(i), bpar.FiberId(i));
    vec_pciw.push_back( pciwp[i]);
  }
  cout <<"mfacq[2] Creating PCIEToEthernet  thread object " << endl;
  PCIEToEthernet pci2eth(vec_pciw, bpar.GetParams().GetEthTargets(), bpar.GetParams(), bpar.GetParams().tcpportid);
  pci2eth.SetPrintLevel(acpar.prtlevel_, acpar.prtmodulo_); 
  //  usleep(200);  attente au cas ou ...
  pPcie2Eth=&pci2eth;
  tm.Split("Threads created");  
  cout << "mfacq[3] - starting one threads: PCIEToEthernet... " << endl;
  pci2eth.start();
  
  // On attend avant de declencher la terminaison des threads
  usleep(200000);
  
  cout << "mfacq[5] -  Waiting for threads to finish ... " << endl;  
  pci2eth.join();
  cout << "mfacq[6] ---------- threads finished ---------------- " << endl;
  tm.Split("Threads Finished");  
 
#ifndef NOPCIECARD
  for (size_t i=0; i<acpar.NbFibers(); i++) {
    UINT32 card=(acpar.FiberNum(i)-1)/2+1;
    UINT32 cardfiber=(acpar.FiberNum(i)-1)%2;
    DeletePCIEWrapperV6(card,cardfiber);
  }
#endif    
return 0;
}


/* --Nouvelle-Fonction-- */
int EthernetToMemoryAcq(bool fgviscal) 
{
  Timer tm("mfacq/EthernetToMemoryAcq");
  cout << " ---------- mfacq/ EthernetToMemoryAcq()  ------------- " << endl;
    
  BRAcqConfig bpar;
  BRParList& acpar=bpar.GetParams();

  if (fgviscal)  cout << "mfacq[0]/EthernetToMemoryAcq() : On the fly Visibility Calculation " << endl;
  else cout << "mfacq[0]/EthernetToMemoryAcq() : data dumped to disk  " << endl;  
  cout << " mfacq[1]/Info: Creating RAcqMemZoneMgr for" << acpar.NbEthLinks() << " Eth-links , nZones=" 
       << acpar.nZones << " NbPaquet/Zone=" << acpar.nPaqZone 
       << " MmgrPaqSize=" << acpar.MMgrPaquetSize() << endl;
  RAcqMemZoneMgr mmgr(acpar.nZones, acpar.NbEthLinks(), acpar.nPaqZone, acpar.MMgrPaquetSize());
  pMmgr=&mmgr;  


  string strfmask;
  uint_4 fmask=0;
  MemZStatus mskmon=MemZS_Proc;
  if (fgviscal) {  // gestion de finalized_mask lors du calcul de visibilites 
    if (acpar.nbcalgrpVisiC<1)  acpar.nbcalgrpVisiC=1;
    if (acpar.nbcalgrpVisiC>4)  acpar.nbcalgrpVisiC=4;
    cout << " mfacq[1.b]/Info: NbCalGrpVisiC = " << acpar.nbcalgrpVisiC << " (1<=NbCalGrpVisiC<=4)" << endl;
    fmask=(uint_4)MemZS_ProcA;   strfmask="MemZS_ProcA";  mskmon=MemZS_ProcB;
    if (acpar.nbcalgrpVisiC>1) 
      { fmask |= (uint_4)MemZS_ProcB;  strfmask+=" | MemZS_ProcB";  mskmon=MemZS_ProcC; }
    if (acpar.nbcalgrpVisiC>2) 
      { fmask |= (uint_4)MemZS_ProcC;  strfmask+=" | MemZS_ProcC";  mskmon=MemZS_ProcD; }
    if (acpar.nbcalgrpVisiC>3) 
      { fmask |= (uint_4)MemZS_ProcD;  strfmask+=" | MemZS_ProcD";  mskmon=MemZS_ProcE; }
  }
  else { 
    fmask=(uint_4)MemZS_Saved;  strfmask="MemZS_Saved";  mskmon=MemZS_Proc;
  }
  if (acpar.fgdoProc) 
    { fmask |= (uint_4)mskmon; strfmask+=" | MemZS_Proc[Monotoring]"; }
    // Dans ce cas, toutes les zones doivent passer ds le thread de monitoring 
    // pour etre au moins marque comme traite - seul 1/acpar.stepProc sont effectivement traite 

  mmgr.SetFinalizedMask(fmask);
  cout << " mfacq[1.b]/Info: Mmgr.SetFinalizedMask( " <<  strfmask << " )" << endl;
  
  if (fgviscal) 
    cout <<"mfacq[2] Creating BRVisCalcGroup, NbVisibCalculator=" << acpar.nbcalgrpVisiC 
         << " and MonitorProc " << endl;
  else 
    cout <<"mfacq[2] Creating MultiDataSaver and  MonitorProc thread objects ... " << endl;
  MultiDataSaver DsThr(mmgr);   //   Utilise les parametres globaux BRAcqConfig
  string ppath=bpar.OutputDirectory();
  MonitorProc PrThr(mmgr);
  PrThr.SetMemZAction( RAcqMemZoneMgr::Convert_Status2Action( mskmon ) );
  BRVisCalcGroup VCGThr(acpar.nbcalgrpVisiC, mmgr, bpar.OutputDirectory(), acpar.nmeanVisiC, acpar.nthrVisiC);
  VCGThr.SelectFreqBinning(acpar.freqminVisiC, acpar.freqmaxVisiC, acpar.nbinfreqVisiC);


  cout << "mfacq[3] Creating   EthernetReader thread object     " << endl;
  EthernetReader ethrdr(mmgr, bpar.GetParams(), bpar.GetParams().tcpportid);
  ethrdr.SetReadMode(acpar.ethr_forcesamefc_, acpar.ethr_sfc_maxdpc_,acpar.ethr_sfc_maxresync_);
  ethrdr.WaitENDMsg4Terminate(acpar.ethr_waitendmsg_);
  ethrdr.SetPrintLevel(acpar.prtlevel_, acpar.prtmodulo_); 

  //  usleep(200);  attente au cas ou ...
  pEthRdr=&ethrdr;
  tm.Split("Threads created"); 
  cout << "mfacq[4] - starting EthernetReader thread object ..." << endl;
  ethrdr.start();
  if (fgviscal) {
    cout << "mfacq[4.b] - starting  Visibility calculator threads " << endl;
    VCGThr.start();
  }
  else {
    cout << "mfacq[4.b] - starting  MultiDataSaver  " << endl;
    DsThr.start();
  }
  if (acpar.fgdoProc) { // demarrage (optionnel) du thread de monitoring
    cout << " mfacq[4.c] - starting MonitorProc thread object " << endl;
    PrThr.start();
  }
  
  // On attend avant de declencher la terminaison des threads
  usleep(200000);
  
  cout << "mfacq[5] -  Waiting for threads to finish ... " << endl;  
  ethrdr.join();
  if (fgviscal)  VCGThr.join();
  else DsThr.join();
  mmgr.Stop();
  if (acpar.fgdoProc) { // On n'attend la fin que si le thread a ete demarre (NMaxProc>0)
    PrThr.join();
  }
  pMmgr=NULL;  
  cout << "mfacq[6] ---------- threads finished ---------------- " << endl;
  tm.Split("Threads Finished");  
 
  mmgr.Print(cout);

return 0;

}