source: Sophya/trunk/AddOn/TAcq/racqueth.h@ 3912

#ifndef  RACQUETH_H_SEEN
#define  RACQUETH_H_SEEN

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

#include "racqumem.h"
#include <string>
#include <vector>
#include <iostream>
#include "sopnamsp.h"
#include "zthread.h"
#include "pciewrap.h"
#include "swrapsock.h"
#include "brtypes.h"
#include "brpaqu.h"
#include "brparam.h"
#include "minifits.h"

using namespace std;

// Quelques definitions globales
// Nombre maximum de fibres geres par les classes PCIEToEthernet ...
//  ATTENTION cette definition (MAXANAFIB) se trouve dans plusieurs fichiers 
#ifndef MAXNBFIB
#define MAXNBFIB  32  
#endif
//  ATTENTION cette definition (MAXANAFIB) se trouve dans plusieurs fichiers 
#ifndef MAXANAFIB 
#define MAXANAFIB  32
#endif

#ifndef BRSPORTID
#define BRSPORTID 6912
#endif
#ifndef BRTCPMSGLEN
#define BRTCPMSGLEN 128
#endif
#ifndef BRTCPMSGLEN2
#define BRTCPMSGLEN2 64
#endif


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

class PCIEToEthernet : public ZThread {
public:
  PCIEToEthernet(vector<PCIEWrapperInterface*> vec_pciw , vector<string>& destname, BRParList const& par, int portid=BRSPORTID);
  virtual ~PCIEToEthernet();
  virtual void run(); 
   void Stop();
  inline void STOP() { stop_ = true; }  

  inline int SetPrintLevel(int lev=0, uint_8 prtmodulo=50000) 
    { prtlev_=lev;   prtmodulo_=prtmodulo;  return 0; }

protected: 
  virtual size_t SendToTargets(int fib, Byte* data, size_t len, bool fgfin=false);
  virtual void CleanUpEndSendAllLinks();

  BRParList par_;   // Parametres divers d'acquisition 

  uint_8 nmaxpaq_;  // Nombre maxi de paquets traites - DOIT etre mutiple de mmgr.NbPaquets()
  BRDataFmtConv swapall_;  // select data swap/format conversion for BRPaquet
  bool stop_;  
  unsigned int sizeFr_ ;
  vector<PCIEWrapperInterface *> vec_pciw_;
  vector<string> destname_; 
  int tcpportid_;
  vector< vector<ClientSocket> > vvec_skt_;
  vector< vector<uint_8> > vvec_errorcnt_;
  vector< uint_8 > vec_cntpaq_; 
  vector< bool > vfgmsgfin_;   // true -> message fin envoye pour la fibre correspondante  
  uint_4 packSize_;
  uint_4 packSizeInMgr_;
  uint_4 nbDma_;
  uint_8 totrdsnd_;
  bool fgdirectsend_;  //  true -> direct transfer of DMA data to ethernet

  string stopreason_;   // message descriptif de la raison d'arret 

  int prtlev_;    // print level 
  uint_8 prtmodulo_;   // print periodicity (modulo) 
};

//------------------------------------------------------------------
// Classe thread de lecture sur interface reseau (Ethernet) 
//------------------------------------------------------------------
class EthernetReader : public ZThread {
public:
  EthernetReader(RAcqMemZoneMgr& mem, BRParList const& par, int portid=BRSPORTID);
  virtual ~EthernetReader(); 

  // Configuration du mode de lecture des paquets 
  // Si force_samefc=true, on essaye de lire des paquets avec meme FrameCounter, sans depasser une 
  // difference maximum de maxdiff_paqnum entre nombre de paquets lus sur chaque fibre
  //  - force_samefc=false : lecture des paquet sans alignement des FrameCounter 
  //  - force_samefc=true : lecture paquets avec alignement des FrameCounter
  //     + mode par defaut : maxdiff_paqnum=0, maxresync=0 -> on cherche a aligner FrameCounter sur les 
  //     differents liens ethernet 
  //     + maxdiff_paqnum>0 : difference maximum de nombre de paquets entre les differents liens
  //     + maxresync>0 : Mode de lecture avec resynchronisation du nombre de paquets , nombre 
  //       maxi de resynchronisation - 
  //  Modes conseilles : false   -  true,0,0  - true,10,0        
  inline void SetReadMode(bool force_samefc=false, uint_4 maxdiff_paqnum=0, uint_4 maxresync=0)
  { rdsamefc_=force_samefc; sfc_maxdpc_=maxdiff_paqnum; sfc_maxresync_=maxresync; }
  //  En activant le flag, le reader ne se termine qu'avec la reception des message END sur chaque lien 
  inline void WaitENDMsg4Terminate(bool waitendmsg=false) 
  { waitendmsg_=waitendmsg; }

  virtual void run();
  inline void Stop() { stop_ = true; }
  inline void STOP() { stop_ = true; }  

  inline int SetPrintLevel(int lev=0, uint_8 prtmodulo=50000) 
    { prtlev_=lev;   prtmodulo_=prtmodulo;  return 0; }

protected:
  bool ReadNextAllFibers();      // Renvoie true si probleme 
  bool ReadNext(int fib);   // Renvoie true si probleme 
  void CleanUpAllSockets();  // pour finir la lecture de toutes les fibres 
  size_t ReceiveFromSocket(int fib, char* data, size_t len);

  bool ReadNextAllFibersWithSync();      // Renvoie true si probleme 
// recale les links en cas d'ecart important entre nb de paquets lu sur chaque lien
  int SkipAndSyncLinks();   // Renvoie 0 si rien fait, 1 si skip fait, 9 si probleme  

  // Permet d'avancer d'un paquet dans la zone - renvoie true si probleme
  inline bool MoveToNextTarget()  {
    if ((mmbuf_ == NULL )||(targ_npaq_ >= max_targ_npaq)) 
      if (MZoneManage()) return true;
    targ_npaq_++;
    return false;
  }
  inline Byte* GetPaquetTarget(int numfib)  {
    if ((mmbufib_[numfib] == NULL )||(targ_npaq_ > max_targ_npaq)||(targ_npaq_ == 0))  return NULL; 
    Byte* rb=mmbufib_[numfib]+(targ_npaq_-1)*packsize_;
    return rb;
  }
  bool MZoneManage(bool clean=false);  // Renvoie true si probleme

  RAcqMemZoneMgr& memgr_;
  BRParList par_;   // Parametres divers d'acquisition 
  int tcpportid_;
  ServerSocket* srv_sokp_;  // Le socket serveur  
  vector< Socket > vsok_;  //  Les sockets de lecture pour chaque lien ethernet (1 lien= 1 fibre a l'envoi) 
  vector< int > vfibid_;   //  Identificateur des fibres transmis par l'emetteur 
  vector< uint_8 > vec_cntpaq_;   // vecteur de compteurs de paquet ( 1 compteur / lien ) 
  vector< uint_4 > vec_fgsokend_;    // vecteur de flag, >0 : fin de reception (ou ereur) sur le lien 
  bool gl_fgsokend;          // true -> au moins l'un de liens a termine 
  bool stop_;
  bool rdsamefc_;  // if true, read paquets with same value of FrameCounter on different fibers/link 
  uint_4 sfc_maxdpc_;   // difference maxi entre nombre de paquets lu sur les liens 
  uint_4 sfc_maxresync_;  // nb maximum de tentatives de resynchronisation 

  uint_8 totnbytesrd_;
  uint_8 totnpaqrd_;
  uint_8 totsamefc_;   // nombre total de paquets avec meme framecounter  

  uint_8 totnbresync_;   // nombre total de saut+resynchronisation 

  vector<BRPaquet> vpaq_; 
  vector<BRPaqChecker> vpchk_;
  vector<uint_8> curfc_;       // Numeros des FrameCounter des paquets courants
  vector<uint_8> totnpqrd_;    // nombre total de paquets lus / lien 
  vector<uint_8> totnpqok_;    // nombre total de paquets OK / lien   

  uint_4 packsize_;
  int mid_;  // Identificateur zone memoire
  uint_4 targ_npaq_;  // Numero de paquet dans une seule zone memoire 
  uint_4 max_targ_npaq; // =  mmgr.NbPaquets() = Max de targ_npaq_
  Byte* mmbuf_;  // Pointeur zone memoire rendu par RAcqMemZoneMgr 
  Byte* mmbufib_[MAXANAFIB];  // Pointeurs zone memoire de chaque fibre rendu par RAcqMemZoneMgr 

  // zone tampon contenant zero si lecture sur fibre est fini 
  char* dummybuff_;
  string stopreason_;   // message descriptif de la raison d'arret 

  bool waitendmsg_;   // si true, attendre le message  END message sur chaque lien avant de s'arreter 
 
  int prtlev_;   // print level 
  uint_8 prtmodulo_;   // print periodicity (modulo) 
};



#endif