source: Sophya/trunk/AddOn/TAcq/tmtacq.cc@ 3633

#include <iostream>

#include "tmtacq.h"
#include "racqumem.h"
#include "racqurw.h"
#ifndef NOPCIECARD
#include "racquwbin.h"
#include "dmamgrv3.h"
#endif
#include "racquproc.h"


#include "pciewrap.h"
#include "brpaqu.h"
#include "minifits.h"

#include "resusage.h"
#include "ctimer.h"
#include "timing.h"

#include "tarrinit.h"
#include "fiosinit.h"

using namespace std;
using namespace SOPHYA;


//--------------------------------------------------------
// Programme test acquisition BAORadio multi-thread 
//  LAL -   R. Ansari Juillet -  2008
//          M.Taurigna    2009
//--------------------------------------------------------


/*   Fonction d'arret par Ctl<C> et ses pointeurs globaux */    
static DataSaver *pDs = NULL;
static DataSaver *pDs2 = NULL;
static PCIEReaderChecker *pPcierc = NULL;
static PCIEReaderChecker *pPcierc2 = NULL;
static PCIEReader *pPcier = NULL;
static PCIEReader *pPcier2 = NULL;
static DataProc2C *pPr = NULL;
static DataProc2C *pPr2 = NULL;
static DataProcFFT2C *pPrfft = NULL;
static DataProcFFT2C *pPrfft2 = NULL;
static RAcqMemZoneMgr *pMmgr = NULL;
static RAcqMemZoneMgr *pMmgr2 = NULL;

void Stop(int s)
{
  if (s == 9765) cout << " Stop after exception ..." << endl;
  else printf("............. MAIN ... receive signal %d \n",s);
  if (pPcierc != NULL) pPcierc->Stop();
  if (pPcierc2 != NULL) pPcierc2->Stop();
  if (pPcier != NULL) pPcier->Stop();
  if (pPcier2 != NULL) pPcier2->Stop();
  if (pPr !=NULL) pPr->Stop();
  if (pPr2 !=NULL) pPr2->Stop();
  if (pPrfft !=NULL) pPrfft->Stop();
  if (pPrfft2 !=NULL) pPrfft2->Stop();
  if (pDs != NULL) pDs->Stop();
  if (pDs2 != NULL) pDs2->Stop();
  if (pMmgr !=NULL) pMmgr->Stop();
  if (pMmgr2 !=NULL) pMmgr2->Stop();
  
}
/* -------------------------------------------------------- */

// Fonctions appelees par le main 
void Usage();
int DecodeArgs(int narg, char* arg[]);
void DecodeAcqMode();

int MonoCardAcq();
int MultiCardAcq();

//------------------ Parametres de controle ----------------------------
static string acqmode; 
static  bool savesigfits = true;
static  BRDataFmtConv swapall = BR_Copy;
static  bool monothr = false;
static string OutPathName;
static bool singlecard = true;
static int cardlist[4] = {1,1,1,1};
static int card = 1;
static  unsigned int sizeFrame = 4096;
static  unsigned int nbFrameDMA = 32;
static  int NbFiles=1; 
static  int NBlocPerFile=10;
static  int NMaxProc = 0;  // Nombre de blocs traites par le thread de calcul
static  uint_4 nZones = 4;  // Nombre de zones memoires
static  uint_4 nPaqZone = 128;  // 128 Paquets / zone memoire - valeur par defaut
static  bool fg_hard_ctrlc = false;
static  uint_4 PaqSZ = 4096;  // Taille de paquets 
  // uint_4 dmaSize  = nbFrameDMA*PaqSZ ;  // plantage 
static  uint_4 dmaSize  = 32*1024 ; 
static  uint_4 patternSZ=0x400;  // pas utilise avec la fibre 
static  uint_4 NMaxBloc = 10; 
//---------------------------- FIN parametres de controle -----------------

//-----------------------------------------------------------------------
//-------------------- le programme principal ---------------------------
//-----------------------------------------------------------------------
int main(int narg, char* arg[])
{ 
  
  int rc = 0;
  cout << " ============ BAORadio / Acquisition : tmtacq  =================" << endl;
  cout << " ===============================================================" << endl;
  cout << " =========  " <<BAOR_ACQ_VER_STR <<BAOR_ACQ_VER <<"                  ===========" << endl;
  cout << " ===============================================================" << endl;
  
  int rcda = DecodeArgs(narg, arg);
  if (rcda != 0) return 1;
  InitTim();
 
  // Initialisation 
  TArrayInitiator _arri;
  FitsIOServerInitiator _fiosi;
 
  try {
    if (singlecard)  rc = MonoCardAcq();
    else rc = MultiCardAcq();
  }
  catch (MiniFITSException& exc) {
    cerr << " tmtacq.cc catched MiniFITSException " << exc.Msg() << endl;
    Stop(9765);
    rc = 77;
  } 
  catch (PCIEWException& exc) {
    cerr << " tmtacq.cc catched MiniFITSException " << exc.Msg() << endl;
    Stop(9765);
    rc = 75;
  }  
  catch (PThrowable& exc) {
    cerr << " tmtacq.cc catched Exception " << exc.Msg() << endl;
    Stop(9765);
    rc = 76;
  }  
  catch (std::exception& sex) {
    cerr << "\n tmtacq.cc std::exception :" 
         << (string)typeid(sex).name() << "\n msg= " 
         << sex.what() << endl;
    Stop(9765);
    rc = 78;
  }
  catch (...) {
    cerr << "  tmtacq.cc : Catched ... exception " << endl;
    Stop(9765);
    rc = 79;
  }
  cout << "tmtacq[9] ----- END --- stopping acq program " << endl;
  return rc;
}


/* --Nouvelle-Fonction-- */
void Usage() 
{
   cout << "\n Usage: tmtacq CardNum PaquetSize NFrameDMA NFiles NBlocPerFile \n"
        << "                 NMaxProc DataDirPath [AcqMode] [MemZoneMgr][HardCtlC]" << endl;
   cout << " - CardNum : PCI-Express card number (=1 OR 2 OR 1,2)" << endl;
   cout << " - PaquetSize or FrameSize (=DATA+HDR+TRL_Size)  " << endl;
   cout << " - NFrameDMA = DMASize= NFrameDMA*1024 , 4KO<=MaxDMASize<=64 ko " << endl;
   cout << " - NFiles = Number of data files produced " << endl;
   cout << " - NBlocPerFile = Number of memory data bloc in one file " << endl;
   cout << " - NMaxProc = Max number of memory data blocs processed (FFT ...) " << endl;
   cout << "      NMaxProc=0 -> No Processing " << endl;
   cout << " - DataDirPath : Subdirectory of /Raid " << endl;
   cout << "   Pour la version du firmware qui ne swappe pas les paquets" << endl;
   cout << " - AcqMode: = swapall , fft1c , fft2c, swh , nof , mxs , mono , monosw (default=std)" << endl;
   cout << "    swapall -> SwapAll+SaveData , swh->swap header only " << endl;
   cout << "    fft1c , ff2c -> reorder data for 1 channel/2channel FFT " << endl;
   cout << "    fft1cnof , ff2cnof -> reorder FFT data, but DONT write files " << endl;
   cout << endl;
   cout << "   Pour la version du firmware qui swappe partiellement " << endl;
   cout << "  -AcqMode : swap32 ,fft1c32 , fft2c32,fft1cnof32 , fft2cnof32 " <<endl;
   cout << "    fft1c32 , fft2c32 -> reorder data for 1 channel/2channel FFT " << endl;
   cout << "    fft1cnof32 , ff2cnof32 -> reorder FFT data, but DONT write files " << endl << endl;
 cout << endl;
   cout << "   Pour la version du firmware qui ne swappe plus " << endl;
   cout << "  -AcqMode :  nosw,fft1cnosw , fft2cnosw" <<endl;
   cout << "    fft1cnosw , fft2cnosw -> reorder data for 1 channel/2channel FFT " << endl;
   cout << "    fft1cnofnosw , ff2cnofnosw -> reorder FFT data, but DONT write files " << endl << endl;


   cout << "   Option pour debug ou test performance swapp initial (std)" << endl ;
   cout << "    nof->Don't write signal fits files , mxs->swh+nof: swap header only + no fits " << endl;
   cout << "    mono->Test with a single thread PCIEReaderChecker " << endl;
   cout << "   monosw->Test with a single thread PCIEReaderChecker, But swap all packet " << endl;
   cout << endl;
   cout << "   mononsw->Test with a single thread PCIEReaderChecker, But no swap paquet " << endl;
   cout << endl;
   cout << " - MemZoneMgr: nZones,NPaq =Number of Zones and number of paquet in each zone (Default=8,128) "
 << endl;
   cout << " -HardCtlC : Y y (direct interrpution by CtrlC ) default (no) " << endl;
}

/* --Nouvelle-Fonction-- */
int DecodeArgs(int narg, char* arg[]) 
{
  if (narg < 8) {
    Usage();
    return 1;
  }
  acqmode = "std";
  acqmode = arg[8];
  savesigfits = true;
  // BRDataFmtConv swapall = BR_SwapAll;
  swapall = BR_Copy;
  monothr = false;
  DecodeAcqMode();

  if (index(arg[1],',') != NULL) {
    sscanf(arg[1],"%d,%d", cardlist, cardlist+1);
    cout << " tmtacq[1]/MultiPCICard acquisition ... Cards= "  << cardlist[0] << " ," << cardlist[1] 
         << " FrameSize=PaqSize=" << sizeFrame << endl;
    singlecard = false;  
  } 
  else {
    card = atoi(arg[1]);
    cardlist[0] = cardlist[1] = card;
    singlecard = true;  
    cout << "tmtacq[1]/MonoPCICard acquisition ... Card= " << card <<  " FrameSize=PaqSize=" << sizeFrame << endl;
  }
  sizeFrame = atoi(arg[2]);
  nbFrameDMA = atoi(arg[3]);


  NbFiles = atoi(arg[4]);
  NBlocPerFile = atoi(arg[5]);
  NMaxProc = atoi(arg[6]);  // Nombre de blocs traites par le thread de calcul

#ifdef NOPCIECARD
  string OutPathName = string(arg[7])+"/";
#else  
  string OutPathName =string("/Raid/")+arg[7]+"/";
#endif 
  cout << "  DataDirpath=" << OutPathName << " SwapMode=" << BRPaquet::FmtConvToString(swapall) 
       << "  DataSaveMode=" << ((savesigfits)?"Yes/FitsFile":"NO") << endl; 
  
  char cmd[512];
  if (singlecard) {
    sprintf(cmd,"mkdir %s",OutPathName.c_str());
    if (system(cmd) < 0) {
      cout << "tmtacq/Error: Can not create  subdirectory " << OutPathName << " -> exit" << endl;
      return 2;
    }
    else cout << " tmtacq[1] - Executed command " << cmd << endl;
  }
  else {
    sprintf(cmd,"mkdir %s",OutPathName.c_str());
    if (system(cmd) < 0) {
      cout << "tmtacq/Error: Can not create  subdirectory " << OutPathName << " -> exit" << endl;
      return 2;
    }
    else cout << " tmtacq[1] - Executed command " << cmd << endl;
    sprintf(cmd,"mkdir %s/Card1",OutPathName.c_str());
    if (system(cmd) < 0) {
      cout << "tmtacq/Error: Can not create  subdirectory " << OutPathName << " -> exit" << endl;
      return 2;
    }
    else cout << " tmtacq[1] - Executed command " << cmd << endl;
    sprintf(cmd,"mkdir %s/Card2",OutPathName.c_str());
    if (system(cmd) < 0) {
      cout << "tmtacq/Error: Can not create  subdirectory " << OutPathName << " -> exit" << endl;
      return 2;
    }
    else cout << " tmtacq[1] - Executed command " << cmd << endl;
  }
  nZones = 8;  // Nombre de zones memoires
  nPaqZone = 128;  // 128 Paquets / zone memoire - valeur par defaut
  if (narg > 9)  { 
    int ia1, ia2;
    sscanf(arg[9],"%d,%d", &ia1, &ia2);
    nZones = ia1;  nPaqZone = ia2; 
  }

  fg_hard_ctrlc = false;

  struct sigaction act;
 
  if (narg > 10)    // pour traiter eventuellement un arret brutal par CtlC mettre le 10eme arg a Y
    if ((*arg[10]=='y')||(*arg[10]=='Y'))  fg_hard_ctrlc=true;
  if (!fg_hard_ctrlc) {
    act.sa_handler=Stop;
    sigaction(SIGINT,&act,NULL);        
  }

  PaqSZ =sizeFrame;  // Taille de paquets 
  // uint_4 dmaSize  = nbFrameDMA*PaqSZ ;  // plantage 
  dmaSize  = nbFrameDMA*1024 ; 
  patternSZ=0x400;  // pas utilise avec la fibre 
  NMaxBloc = NbFiles*NBlocPerFile; 
  cout << "tmtacq[2] - PaqSize = " << PaqSZ << " NbPaq/Zone=" << nPaqZone << " NZones=" << nZones << endl; 
  cout << " NbFiles=" << NbFiles << " NBloc/File=" << NBlocPerFile << " -> NMaxBloc=" << NMaxBloc << endl;
  cout << "tmtacq[2] NbFrameDMA=" << nbFrameDMA << "  DMASize=" << dmaSize << " bytes" << endl;
  cout << "tmtacq[2] NMaxProc=" << NMaxProc << endl;

  return 0;
}

/* --Nouvelle-Fonction-- */
void DecodeAcqMode() 
{
  if (acqmode == "swapall")  swapall = BR_SwapAll;
  if (acqmode == "fft1c")  swapall = BR_FFTOneChan;
  if (acqmode == "fft2c")  swapall = BR_FFTTwoChan;
  if (acqmode == "fft1cnof")  { swapall = BR_FFTOneChan;  savesigfits = false; }
  if (acqmode == "fft2cnof")  { swapall = BR_FFTTwoChan;  savesigfits = false; }
  if ((acqmode == "swh") || (acqmode == "mxs") || (acqmode == "mono") )  swapall = BR_SwapHDR;
  if ((acqmode == "nof") || (acqmode == "mxs") || (acqmode == "mono"))  savesigfits = false;
  if (acqmode == "mono")  { monothr = true;   swapall = BR_SwapHDR;; }
  if (acqmode == "monosw")  { monothr = true;   swapall = BR_SwapAll; }
  if (acqmode == "mononsw")  { monothr = true;   swapall = BR_Copy; }

  if (acqmode == "swap32")   swapall = BR_Swap32 ;
  if (acqmode == "fft1c32")  swapall = BR_FFTOneChan32;
  if (acqmode == "fft2c32")  swapall = BR_FFTTwoChan32;
  if (acqmode == "fft1cnof32")  { swapall = BR_FFTOneChan32;  savesigfits = false; }
  if (acqmode == "fft2cnof32")  { swapall = BR_FFTTwoChan32;  savesigfits = false; }

  if (acqmode == "nosw")   swapall = BR_Copy ;
  if (acqmode == "fft1cnosw")  swapall = BR_FFTOneChanNoSwap;
  if (acqmode == "fft2cnosw")  swapall = BR_FFTTwoChanNoSwap;
  if (acqmode == "fft1cnofnosw")  { swapall = BR_FFTOneChanNoSwap;  savesigfits = false; }
  if (acqmode == "fft2cnofnosw")  { swapall = BR_FFTTwoChanNoSwap;  savesigfits = false; }
}


/* --Nouvelle-Fonction-- */
int MonoCardAcq() 
{
  Timer tm("tmtacq/MonoCard");

  cout << " ---- tmtacq/ MonoCardAcq()  ------------- " << endl;
#ifdef NOPCIECARD
  TestPCIWrapperNODMA pciw(PaqSZ);
#else
  DMAMgr dma1(card,patternSZ,dmaSize ); 
  if (! dma1.StatusFibre() ) {
    cout << " tmtacq[3] - fibre non accrochee -> exit " << endl;
    throw PCIEWException(" Fibre non accrochee ");
  }
  else cout << " tmtacq[3] - fibre accrochee OK " << endl;
  PCIEWrapper pciw(dma1); 
#endif    
  RAcqMemZoneMgr mmgr(nZones, nPaqZone, PaqSZ);
  pMmgr =&mmgr;

  if (monothr) {
    cout << "tmtacq[4] single thread PCIE test PCIEReaderChecker ... ";
    PCIEReaderChecker pcirc(pciw,dmaSize,PaqSZ ,mmgr, NMaxBloc, swapall);
    pPcierc=&pcirc;
    pcirc.start();
    sleep(1);
    pcirc.join();
    cout << "tmtacq[5] - single thread PCIEReaderChecker finished " << endl;
    tm.Split("Single Thread Finished");  
    mmgr.Print(cout);
    return 0;
  }
  PCIEReader pcir(pciw,dmaSize,PaqSZ ,mmgr, NMaxBloc, swapall);
  pPcier= &pcir;
  DataSaver ds(mmgr, OutPathName, NbFiles, NBlocPerFile, savesigfits);
  pDs= &ds;
  //DataBinSaver ds(mmgr, OutPathName, NbFiles ,NBlocPerFile);
  //     DataProc1C pr(mmgr, OutPathName, nmean, stepproc, 100);     Pour processer un canal
  int stepproc = 2;
  int nmean = nPaqZone*NBlocPerFile/stepproc;
  DataProc2C pr(mmgr, OutPathName, nmean, stepproc, NMaxProc);
  DataProcFFT2C prfft(mmgr, OutPathName, nmean, stepproc, NMaxProc);
  tm.Split("Threads created");  
  cout << "tmtacq[4] - starting 3 threads pcir, ds, pr ... " << endl;
  /*
  char test[100];
  cout << " <CR> to continue, x to exit ..." << endl;
  gets(test); if (test[0] == 'x')  return 9;
  */
  pcir.start();
  ds.start();
  if (NMaxProc>0) { // On ne demarre que si au moins NMaxProc>0
    if (acqmode.substr(0,5)=="fft2c")   { prfft.start(); pPrfft=&prfft;}
    else { pr.start();   pPr=&pr;}
  }
  cout << "tmtacq[5] - waiting for threads to finish ... " << endl;
  sleep(1);
  pcir.join();
  ds.join();
  sleep(1);
  mmgr.Stop();
  if (NMaxProc>0) if ( pPr != NULL) pr.join();
  if (NMaxProc>0) if ( pPrfft != NULL) prfft.join();
  cout << "tmtacq[6] - threads finished " << endl;
  tm.Split("Threads Finished");  
  mmgr.Print(cout);
  return 0;
}


/* --Nouvelle-Fonction-- */
int MultiCardAcq() 
{
  Timer tm("tmtacq/MultiCard");

  cout << " ---- tmtacq/ MultiCardAcq()  ------------- " << endl;

#ifdef NOPCIECARD
  TestPCIWrapperNODMA pciw1(PaqSZ);
  TestPCIWrapperNODMA pciw2(PaqSZ);
#else
  DMAMgr dma1(cardlist[0],patternSZ,dmaSize ); 
  if (! dma1.StatusFibre() ) {
    cout << " tmtacq[3] - fibre non accrochee Card" << cardlist[0] << " -> exit " << endl;
    throw PCIEWException(" Fibre non accrochee 1");
  }
  else cout << " tmtacq[3] - fibre accrochee OK Card " << cardlist[0] << endl;

  DMAMgr dma2(cardlist[1],patternSZ,dmaSize ); 
  if (! dma1.StatusFibre() ) {
    cout << " tmtacq[3] - fibre non accrochee Card" << cardlist[1] << " -> exit " << endl;
    throw PCIEWException(" Fibre non accrochee 2");
  }
  else cout << " tmtacq[3] - fibre accrochee OK Card " << cardlist[1] << endl;

  PCIEWrapper pciw1(dma1); 
  PCIEWrapper pciw2(dma2); 
#endif    
  RAcqMemZoneMgr mmgr1(nZones, nPaqZone, PaqSZ);
  RAcqMemZoneMgr mmgr2(nZones, nPaqZone, PaqSZ);
  pMmgr =&mmgr1;
  pMmgr2 =&mmgr2;

  if (monothr) {
    cout << "tmtacq[4] single thread/Card PCIE test PCIEReaderChecker ... ";
    PCIEReaderChecker pcirc1(pciw1,dmaSize,PaqSZ ,mmgr1, NMaxBloc, swapall);
    PCIEReaderChecker pcirc2(pciw1,dmaSize,PaqSZ ,mmgr1, NMaxBloc, swapall);
    pPcierc=&pcirc1;
    pPcierc2=&pcirc2;
    pcirc1.start();

    sleep(1);
    pcirc1.join();
    pcirc2.join();

    cout << "tmtacq[5] - single thread/Card PCIEReaderChecker finished " << endl;
    tm.Split("Single Thread Finished");  
    mmgr1.Print(cout);
    mmgr2.Print(cout);
    return 0;
  }
  PCIEReader pcir1(pciw1,dmaSize,PaqSZ ,mmgr1, NMaxBloc, swapall);
  pPcier= &pcir1;
  PCIEReader pcir2(pciw2,dmaSize,PaqSZ ,mmgr2, NMaxBloc, swapall);
  pPcier2= &pcir2;
  char buff[32];
  sprintf(buff, "Card%d", cardlist[0]); 
  string path1 = OutPathName + buff;
  DataSaver ds1(mmgr1, path1, NbFiles, NBlocPerFile, savesigfits);
  pDs= &ds1;
  sprintf(buff, "Card%d", cardlist[1]); 
  string path2 = OutPathName + buff;
  DataSaver ds2(mmgr2, path2, NbFiles, NBlocPerFile, savesigfits);
  pDs2= &ds2;
  //DataBinSaver ds(mmgr, OutPathName, NbFiles ,NBlocPerFile);
  //     DataProc1C pr(mmgr, OutPathName, nmean, stepproc, 100);     Pour processer un canal
  int stepproc = 2;
  int nmean = nPaqZone*NBlocPerFile/stepproc;
  DataProc2C pr1(mmgr1, path1, nmean, stepproc, NMaxProc);
  DataProcFFT2C prfft1(mmgr1, path1, nmean, stepproc, NMaxProc);
  DataProc2C pr2(mmgr2, path2, nmean, stepproc, NMaxProc);
  DataProcFFT2C prfft2(mmgr2, path2, nmean, stepproc, NMaxProc);

  tm.Split("Threads created");  
  cout << "tmtacq[4] - starting 3 threads pcir, ds, pr ... " << endl;
  /*
  char test[100];
  cout << " <CR> to continue, x to exit ..." << endl;
  gets(test); if (test[0] == 'x')  return 9;
  */
  pcir1.start();
  ds1.start();
  pcir2.start();
  ds2.start();
  if (NMaxProc>0) { // On ne demarre que si au moins NMaxProc>0
    if (acqmode.substr(0,5)=="fft2c")   { 
      prfft1.start(); pPrfft=&prfft1;
      prfft2.start(); pPrfft2=&prfft2;
    }
    else { 
      pr1.start();  pPr=&pr1;
      pr2.start();  pPr2=&pr2;
    }
  }
  cout << "tmtacq[5] - waiting for threads to finish ... " << endl;
  sleep(1);
  pcir1.join();
  ds1.join();
  sleep(1);
  mmgr1.Stop();
  mmgr2.Stop();
  if (NMaxProc>0) { 
    if ( pPr != NULL) pr1.join();
    if ( pPr2 != NULL) pr2.join();
    if ( pPrfft != NULL) prfft1.join();
    if ( pPrfft2 != NULL) prfft2.join();
  }

  cout << "tmtacq[6] - threads finished " << endl;
  tm.Split("Threads Finished");  
  mmgr1.Print(cout);
  mmgr2.Print(cout);
  return 0;
}