//----------------------------------------------------------------
// ---- classes de threads pour lecture (transfert DMA) 
//      et ecriture disque pour acquisition   BAORadio       -----
// LAL -      R. Ansari - Juin/Juillet 2008 
//----------------------------------------------------------------

#include "racqueth.h"

#include <stdlib.h>
#include <unistd.h>
#include <fstream>
#include <signal.h>
#include "pexceptions.h"
#include "ctimer.h"
#include "timestamp.h"

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

#include "resusage.h" // Pour mesure temps elapsed/CPU ...
#include <sys/time.h>  // pour gettimeofday

////////////////////////////////////////////////////////////////////////////////////////////////////////
//----------------------------------------------------------------------------------------------------------
// Classe thread de lecture PCI-Express et recopie sur interface reseau (Ethernet) 
//----------------------------------------------------------------------------------------------------------


/* --Methode-- */
PCIEToEthernet::PCIEToEthernet(vector<PCIEWrapperInterface*> vec_pciw, vector<string>& destname, BRParList const& par, int portid)
  :   par_(par), vec_pciw_ (vec_pciw), destname_(destname), tcpportid_(portid)
{
  nmaxpaq_ = par_.MaxNbPaquets();	
  swapall_ = par_.GetDataConvFg();  	// select data swap/format conversion for BRPaquet
  stop_ = false;
  packSize_ = par_.RecvPaquetSize();
  packSizeInMgr_=par_.MMgrPaquetSize();
  sizeFr_=par_.DMASizeBytes();
  if (vec_pciw.size() > MAXNBFIB) 
    throw ParmError("PCIEToEthernet:ERROR/  vec_pciw.size() > MAXNBFIB ");
  nbDma_= vec_pciw.size();

  char msg[BRTCPMSGLEN];
  for(size_t i=0; i<vec_pciw_.size(); i++) {
    vector<ClientSocket> vskt;
    for(size_t j=0; j<vskt.size(); j++) {
      ClientSocket sok(destname_[j], tcpportid_);
      for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0';
      sprintf(msg,"BAORadio-PCIEToEthernet %d %d",0,i);
      sok.Send(msg,BRTCPMSGLEN);
      sok.Receive(msg,BRTCPMSGLEN);
      if (strncmp(msg,"BAORadio-EthernetReader-OK",26)!=0) 
	throw SocketException("PCIEToEthernet:ERROR/  Connection to EthernetReader not established ");
      cout << " PCIEToEthernet: Ethernet connection established for DMA/fiber" << i << " with " << destname_[j] << endl;
      vskt.push_back(sok);    
    }
    vvec_skt_.push_back(vskt);    
  }
}

/* --Methode-- */
PCIEToEthernet::~PCIEToEthernet()
{
  for(size_t i=0; i<vec_pciw_.size(); i++) {
    vector<ClientSocket>& vskt = vvec_skt_[i];
    for(size_t j=0; j<destname_.size(); j++) vskt[j].Close();
  }
}

/* --Methode-- */
void PCIEToEthernet::run()
{
  
  struct timeval tv1,tv2;
  gettimeofday(&tv1, NULL);
  uint_8 totrdsnd = 0;
  
  cout << " PCIEToEthernet::run() - Starting , NMaxMemZones=" <<  par_.MaxNbBlocs()
       << " MMgrNbPaquet=" << par_.MMgrNbPaquet() << " MMgr.PaqSiz=" << par_.MMgrPaquetSize()
       << " DMA-Paqsize " << packSize_ << "  " << BRPaquet::FmtConvToString(swapall_) << endl;	
  setRC(1);	

  // sigaddset(&act.sa_mask,SIGINT);  // pour proteger le transfert DMA
  //sigaction(SIGINT,&act,NULL);  	
  uint_4 paqszmm = par_.MMgrNbPaquet();
  uint_4 paqsz =  packSize_;
  uint_4 dmasz = vec_pciw_[0]->TransferSize();
  //DEL  vec_pciw_[0]->StartTransfers();

  BRPaqChecker pcheck[MAXNBFIB];  // Verification/comptage des paquets  
  Byte* Datas[MAXNBFIB];
  Byte* tampon[MAXNBFIB] ; 
  Byte* nexpaq[MAXNBFIB] ;  // Pour recevevoir les paquets apres reduction de taille 

  Byte* predtampon=NULL;  // tampon de recopie pour la reduction des tailles de paquets 
  Byte* nextpaq=NULL;
  uint_4 off_acheval=0;
  
  int nerrdma = 0;
  int maxerrdma = 10;
  bool fgarret = false;
  
  // Initialisation des tampons pour recopie des paquets a cheval pour chaque DMA
  for (int i=0;i< (int)nbDma_ ;i++) {  
    tampon[i] = new Byte[paqsz];
    nexpaq[i] = new Byte[paqszmm];
  }

  bool fgdirectsend = (par_.pci2eth_fgdirect && (swapall_==BR_Copy) ) ? true : false;   //  true -> direct transfer of data to ethernet
  if (fgdirectsend) 
    cout << " PCIEToEthernet::run() Direct transfer mode DMA to Ethernet ... " << endl;
  bool fgredpaq=par_.fgreducpsize;
  if (fgredpaq) {
    cout << " PCIEToEthernet::run() - PaquetSizeReduction - RedSize=" << par_.redpqsize 
	 << " Offset=" << par_.reducoffset << " " << ((par_.reducneedcopy)?"NeedCopy":"NOCopy") 
	 << "  " << BRPaquet::ReducActionToString(par_.pqreducmode) << endl;
    predtampon = new Byte[paqsz];
  }
  
  
  uint_8 trnbytes[MAXNBFIB];
  uint_4 npaqfait[MAXNBFIB];
  for (int i=0;i< (int)nbDma_ ;i++) {
    npaqfait[i]=0; trnbytes[i]=0; 
  }
  // Attente du message GO des taches lecteurs
  cout << " PCIEToEthernet::run() Waiting for GO message from EthernetReader's ..." << endl;
  char msg[BRTCPMSGLEN];
  for(size_t i=0; i<vec_pciw_.size(); i++) {
    vector<ClientSocket>& vskt=vvec_skt_[i];
    for(size_t j=0; j<vskt.size(); j++) {
      for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0';
      vskt[i].Receive(msg,BRTCPMSGLEN);
      if (strncmp(msg,"BAORadio-EthernetReader-Ready-GO",32)!=0)
	throw SocketException("PCIEToEthernet:ERROR/  BAD Go message from EthernetReader ");
    }
    cout << " PCIEToEthernet::run() Received GO message for Fiber/DMA " << i << " from " << destname_[i] << endl;
  }

  cout << " PCIEToEthernet::run()  Starting DMA to Ethernet transfers - NbFibers=NbDMA=" << vec_pciw_.size() 
       << " x NbEthDestinations=" << destname_.size() << endl;

    // Byte* nextdma = locdata+((kmz%memgr.NbZones())*(paqsz*memgr.NbPaquets()));
  uint_4 npaqfaitg = 0; 
  while (npaqfaitg < nmaxpaq_) {  // Boucle global G 
    if (fgarret) break;
    if (stop_) break;
    
    // Lancement des DMA 
    for (int dma=0; dma < (int)nbDma_ ;dma++) vec_pciw_[dma]->StartTransfers();
    
    // On pointe vers le debut de la zone a remplir aver le prochain DMA
    //-- Zone memoire locale Byte* nextdma = buff+i*paqsz;
    
    bool fgbaddma=false;
    // On boucle sur les nbDma_ en attente de leurs terminaison    
    for (int dma=0; dma <(int) nbDma_ ;dma++)  {
      Datas[dma]=vec_pciw_[dma]->GetData();
      if (Datas[dma] == NULL) { // No data Read in DMA 
	nerrdma ++;   fgbaddma=true;
	cout << "PCIEToEthernetChecker/Erreur Waiting for datas ..." << endl;
	vec_pciw_[dma]->PrintStatus(cout);
	if (nerrdma>=maxerrdma) { fgarret = true; break; }
      }
    }
    if (fgbaddma) continue;  
    if (fgdirectsend) {  // Pas de copie / reduction de taille de paquet, on rebalance tel quel ...
      for (int fib=0; fib<(int) nbDma_ ;fib++) {
	vector<ClientSocket>& vskt = vvec_skt_[fib];
	for(size_t j=0; j<vskt.size(); j++) {
	  vskt[j].Send((const char *)Datas[fib], dmasz);
	  totrdsnd += dmasz;
	}
	trnbytes[fib] += (uint_8)dmasz;
	npaqfait[fib] += (trnbytes[fib]/(uint_8)paqszmm);
	if (fib==nbDma_-1) npaqfaitg += (trnbytes[fib]/(uint_8)paqszmm);
	trnbytes[fib] = trnbytes[fib]%(uint_8)paqszmm;
      }
      continue;  // On bypasse le reste 
    }
    // Si on arrive ici, c'est qu'il faut copier / reduire les paquets ....
    uint_4 curoff=0;
    //1- On traite le paquet a cheval, rempli partiellement avec le DMA d'avant si necessaire pour les n fibres
    if (off_acheval  > 0) {  // IF Numero B
      if ((paqsz-off_acheval)< dmasz) {  // IF Numero A
	for(uint_4 fib=0; fib<nbDma_; fib++) 
	  memcpy((void *)((tampon[fib])+off_acheval), (void *)Datas[fib], paqsz-off_acheval);
	curoff = paqsz-off_acheval;  off_acheval = 0;
	for(uint_4 fib=0; fib<nbDma_; fib++) {
	  // CHECK  S'il faut faire une reduction de taille de paquet
	  if (fgredpaq) { // reduction taille de paquet
	    if (par_.reducneedcopy) {
	      BRPaquet paqc1(tampon[fib], predtampon, paqsz, swapall_);
	      BRPaquet paqc2(nexpaq[fib], par_.redpqsize);
	      paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
	    }
	    else {
	      BRPaquet paqc1(tampon[fib], paqsz);
	      BRPaquet paqc2(nexpaq[fib], par_.redpqsize);
	      paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
	    }
	  }
	  else  { 
	    BRPaquet paqc(tampon[fib], nexpaq[fib], paqsz, swapall_);
	  }
	  BRPaquet paq(nexpaq[fib], packSizeInMgr_);
	  vector<ClientSocket>& vskt = vvec_skt_[fib];
	  for(size_t j=0; j<vskt.size(); j++) {
	    vskt[j].Send((const char *)nexpaq[fib], packSizeInMgr_);
	    totrdsnd += packSizeInMgr_;
	  }
	  npaqfait[fib]++;
	  if (fib==nbDma_-1) npaqfaitg++;  // Ne pas oublier le compteur de paquets faits 
	  pcheck[fib].Check(paq);   // Verification du paquet / FrameCounter 
	}
      }
      else {  // se rapporte au IF numero A
	for(uint_4 fib=0; fib<nbDma_; fib++) 
	  memcpy((void *)(tampon[fib]+off_acheval), (void *)Datas[fib], dmasz);
	curoff =dmasz;
	off_acheval = (dmasz+off_acheval);
      } 
    }  // Fin IF Numero B

    //2- On traite les paquets complets qui se trouvent dans la zone du DMA 
    while ((curoff+paqsz)<=dmasz) {  // while numero C 
      //	  if ((dma==nbDma_-1)&&(npaqfait >= nmax_* memgr.NbPaquets())) break;
      // CHECK  S'il faut faire une reduction de taille de paquet
      for(uint_4 fib=0; fib<nbDma_; fib++) {
	  if (fgredpaq) { // reduction taille de paquet
	    if (par_.reducneedcopy) {
	      BRPaquet paqc1(Datas[fib]+curoff, predtampon, paqsz, swapall_);
	      BRPaquet paqc2(nexpaq[fib], par_.redpqsize);
	      paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
	    }
	    else {
	      BRPaquet paqc1(Datas[fib]+curoff, paqsz);
	      BRPaquet paqc2(nexpaq[fib], par_.redpqsize);
	      paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset);
	    }
	  }
	  else { 
	    BRPaquet paqc(Datas[fib]+curoff, nexpaq[fib], paqsz, swapall_);
	  }
	  BRPaquet paq(nexpaq[fib], packSizeInMgr_);
	  vector<ClientSocket>& vskt = vvec_skt_[fib];
	  for(size_t j=0; j<vskt.size(); j++) {
	    vskt[j].Send((const char *)nexpaq[fib], packSizeInMgr_);
	    totrdsnd += packSizeInMgr_;
	  }
	  npaqfait[fib]++;
	  if (fib==nbDma_-1) npaqfaitg++;  // Ne pas oublier le compteur de paquets faits 
	  pcheck[fib].Check(paq);   // Verification du paquet / FrameCounter 
	}
      curoff += paqsz;  // On avance l'index dans le buffer du DMA
    } // -- FIN traitement des paquets complets ds un DMA - FIN du while numero C 
      //3- On copie si besoin la fin du DMA dans la zone tampon
    if (curoff < dmasz) {  // IF numero D 
      off_acheval = dmasz-curoff;
      for(uint_4 fib=0; fib<nbDma_; fib++) 
	memcpy(tampon[fib], (void*)(Datas[fib]+curoff), off_acheval);
      // ne sert a rien 	curoff += off_acheval; 
    } // FIN du if numero D 
  }  //   FIN  Boucle global G 
 
  setRC(0);
  gettimeofday(&tv2, NULL);
  double tmelaps2 = (tv2.tv_sec-tv1.tv_sec)*1000.+(tv2.tv_usec-tv1.tv_usec)/1000.;
  if (tmelaps2<0.1) tmelaps2=0.1; 
  cout << " ---------- PCIEToEthernet::run()-End summary NPaqFait=" << npaqfaitg 
       << " TotSend (kb)=" << totrdsnd/1024 << "------ " << endl;
  for (int dma=0; dma < (int)nbDma_ ;dma++)   {
    cout << " --Fib=" << dma << " NPaqFait=" << npaqfait[dma] <<  " TotDMATransfer=" 
	 << vec_pciw_[dma]->TotTransferBytes()/1024 
	 << " ElapsTime=" << tmelaps2 << " ms ->" 
	 << (double)vec_pciw_[dma]->TotTransferBytes()/tmelaps2 << " kb/s" << endl;

	vector<ClientSocket>& vskt = vvec_skt_[dma];
	cout << " TotEthTransfer="; 
	for(size_t j=0; j<vskt.size(); j++)
	  cout << vskt[j].NBytesSent()/1024 << " , ";
	cout << " kb" << endl;
	
    if (!fgdirectsend) pcheck[dma].Print(cout);
  }
  cout << " --------------------------------------------------------------------" << endl;

  for (int i=0;i< (int)nbDma_ ;i++) { 
    delete[] tampon[i]; 
    delete[] nexpaq[i];
  }
  if ((fgredpaq)&&predtampon) delete[] predtampon;

  //DBG  cout << " fin thread ========================" <<endl;  
  return;
}


/* --Methode-- */
void PCIEToEthernet::Stop()
{ 
  // cout << " PCIEReaderChecker::stop()  ........ STOP" <<endl; 
  stop_ = true;

}


//------------------------------------------------------------------
// Classe thread de lecture sur interface reseau (Ethernet) 
//------------------------------------------------------------------

/* --Methode-- */
EthernetReader::EthernetReader(RAcqMemZoneMgr& mem, BRParList const& par, int portid, bool rdsamefc)
  :  memgr_(mem), par_(par), stop_(false), rdsamefc_(rdsamefc), tcpportid_(portid)
{
  SetPrintLevel();
  totnbytesrd_ = 0;
  totnpaqrd_ = 0;
  totsamefc_ = 0;
  if (memgr_.NbFibres() > MAXANAFIB) 
    throw BAORadioException("EthernetReader::EthernetReader/ NbFibres>MAXANAFIB "); 
  if (par_.ethr_nlink != memgr_.NbFibres()) 
    throw BAORadioException("EthernetReader::EthernetReader/ NbFibres != ethr_nlink"); 

  packsize_=memgr_.PaqSize();
  mid_=-2; 
  mmbuf_=NULL;
  max_targ_npaq = memgr_.NbPaquets();
  for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) mmbufib_[fib]=NULL;

  for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
    vpaq_.push_back(BRPaquet(NULL,memgr_.PaqSize()));
    vpchk_.push_back(BRPaqChecker(true,0)); 
    curfc_.push_back(0);
    totnpqrd_.push_back(0);
    totnpqok_.push_back(0);    
  }
  ServerSocket srv(tcpportid_, par_.ethr_nlink);
  char msg[BRTCPMSGLEN];
  for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
    Socket sok=srv.WaitClientConnection();
    for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0';
    sok.Receive(msg,BRTCPMSGLEN);
    if (strncmp(msg,"BAORadio-PCIEToEthernet",23)!=0) 
      throw SocketException("EthernetReader:ERROR/  Bad message from PCIEToEthernet client !");
    for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0';
    strcpy(msg,"BAORadio-EthernetReader-OK");
    sok.Send(msg,BRTCPMSGLEN);
    vsok_.push_back(sok);
    cout << " EthernetReader/Info connection/link" << fib << " established." << endl; 
  }
}


/* --Methode-- */
void EthernetReader::run()
{
  setRC(1); 
  try {
    TimeStamp ts; 
    Timer tm("EthernetReader", false);
    cout << " EthernetReader::run() - Starting " << ts << " NbFibres()=" << memgr_.NbFibres()
	 << "  PaqSize() = " << memgr_.PaqSize() << endl;
    cout << " ...ReadMode: " << ((rdsamefc_)?"Paquets With SameFrameCounter":"All OK paquets") << endl;

    cout << " Sending GO message to all PCIEToEthernet clients ... " << endl;
    char msg[BRTCPMSGLEN];
    for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0';
    strcpy(msg,"BAORadio-EthernetReader-Ready-GO");
    for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
      vsok_[fib].Send(msg,BRTCPMSGLEN);
    }
    Byte* nextpaq=NULL;
    bool fgok=true;
    while (fgok&&(totnpaqrd_<par_.MaxNbPaquets())) {
      if (stop_) break;
      if ( MoveToNextTarget() ) {
	cout << "EthernetReader::run()/Error-A- MoveToNextTarget() returned true ->STOP 9" << endl;
	setRC(7);  fgok=false;  break;
      }
      for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
	nextpaq=GetPaquetTarget(fib);
	if (nextpaq == NULL) { // Cela ne devrait pas arriver 
	  cout << "EthernetReader::run()/Error-A2- GetPaquetTarget(fib)  returned NULL ->STOP 9" << endl;
	  setRC(9);  fgok=false;  break;
	}
	vpaq_[fib].Set(nextpaq);
      }
      if (ReadNextAllFibers()) { fgok=false;  break; }
      totnpaqrd_++;
    }

    MoveToNextTarget();  // Pour faire traiter le dernier paquet si plein 
    MZoneManage(true);   // Nettoyage final 
    usleep(50000);       // Attente de traitement du dernier paquet
    memgr_.Stop();        // Arret 

    cout << " ------------------  EthernetReader::run() END ----------------- " << endl;
    ts.SetNow();
    tm.SplitQ();
    cout << "  END reading TotNPaq/Link=" << totnpaqrd_ << " :"  << ts ;
    if (rdsamefc_) cout << " NSameFC=" << totsamefc_ << endl;
    else cout << endl;
    for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)  {
      int perc=0;
      if (totnpqrd_[fib]>0) perc=100*totsamefc_/totnpqrd_[fib];
      cout << " Fiber" << fib << " TotNPaqRd=" << totnpqrd_[fib] << " TotNPaqOK=" << totnpqok_[fib] 
	   << " FracSameFC=" << perc << " %" << endl;
      if (prtlev_ > 0)  vpchk_[fib].Print(cout);
    }
    cout << "  TotalEthernetRead= " << totnbytesrd_/(1024*1024) << " MBytes Ethernet-Read rate= " 
         << (double)(totnbytesrd_)/1024./tm.PartialElapsedTimems() << " MB/s" << endl;    
    cout << " EthernetReader::run()/Timing: \n"; 
    tm.Print();
    cout << " ---------------------------------------------------------- " << endl;

  }  // Fin du bloc try 
  catch (std::exception& exc) {
    cout << " EthernetReader::run()/catched execption msg=  " << exc.what() << endl;
    setRC(3);	
    return; 
  }
  catch(...) {
    cout << " EthernetReader::run()/catched unknown ... exception " << endl;
    setRC(4);	
    return; 
  }
  setRC(0);
  return;
}

/* --Methode-- */
bool EthernetReader::ReadNextAllFibers()
{
  for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)  {
    if (ReadNext(fib)) return true;  // probleme 
  }
  if (!rdsamefc_ || (memgr_.NbFibres()<2))  { 
    totsamefc_++;  return false;  // c'est OK 
  }
  uint_8 cfc=curfc_[0];
  bool fgsamefc=true;
  for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
    if (curfc_[fib]!=cfc) {
      fgsamefc=false;
      if (curfc_[fib] > cfc)  cfc=curfc_[fib]; 
    }
  }
  if (fgsamefc) { 
    totsamefc_++;  return false;  // c'est OK  , same framecounter
  }
  else {  // else !fgsame
    for(uint_4 fib=0; fib<memgr_.NbFibres(); fib++) {
      while (curfc_[fib]<cfc) {
	if (ReadNext(fib)) return true;  // probleme 
      }
    }
  }   // fin de  else !fgsame 
  totsamefc_++; 
  return false; // c'est OK
}

/* --Methode-- */
bool EthernetReader::ReadNext(int fib)
{
  bool fggood=false;
  while(!fggood) { 
    vsok_[fib].Receive((char *)vpaq_[fib].Begin(), packsize_);
    totnbytesrd_+=packsize_;
    totnpqrd_[fib]++;
    fggood = vpchk_[fib].Check(vpaq_[fib],curfc_[fib]);
  }
  totnpqok_[fib]++;
  return false;
}

/* --Methode-- */
bool EthernetReader::MZoneManage(bool fgclean)    // Retourne true si probleme 
{
  /* Pour debug 
  cout << " EthernetReader::MZoneManage() mid_=" << mid_ << " arg_npaq_= " << targ_npaq_  
       << " max_targ_npaq=" << max_targ_npaq << endl;
  */
  if (mid_ >= 0)  { 
    if (fgclean) memgr_.FreeMemZone(mid_, MemZS_Free);    
    else memgr_.FreeMemZone(mid_, MemZS_Filled);
  }
  mmbuf_ = NULL;  targ_npaq_ = 0;  mid_ = -2;
  for (int fib=0;fib<(int)memgr_.NbFibres() ;fib++) mmbufib_[fib]=NULL;
  if (fgclean)  return false;
  mid_ = memgr_.FindMemZoneId(MemZA_Fill);
  mmbuf_ = memgr_.GetMemZone(mid_);
  if (mmbuf_==NULL)   return true;
  for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)
    mmbufib_[fib]=memgr_.GetMemZone(mid_,fib);
  return false;
}