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

#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  16   
#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 size_t SetEthSendBlockSize(size_t bsz=1024)
  { size_t rs=eths_bsz_;  eths_bsz_=bsz; return rs; }
  inline int SetPrintLevel(int lev=0) { prtlev_=lev; return 0;}

protected: 
  virtual size_t SendToTargets(int fib, Byte* data, size_t len);

  BRParList par_;   // Parametres divers d'acquisition 

  uint_8 nmaxpaq_;  // Nombre maxi de paquets traites
  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< uint_4 > vec_cntpaq_; 
  uint_4 packSize_;
  uint_4 packSizeInMgr_;
  uint_4 nbDma_;
  uint_8 totrdsnd_;
  bool fgdirectsend_;  //  true -> direct transfer of DMA data to ethernet
  size_t eths_bsz_;
  int prtlev_;
};

//------------------------------------------------------------------
// Classe thread de lecture sur interface reseau (Ethernet) 
//------------------------------------------------------------------
class EthernetReader : public ZThread {
public:
  EthernetReader(RAcqMemZoneMgr& mem, BRParList const& par, int portid=BRSPORTID, bool rdsamefc=false);
  virtual void run();
  inline void Stop() { stop_ = true; }
  inline void STOP() { stop_ = true; }  

  inline size_t SetEthRecvBlockSize(size_t bsz=1024)
  { size_t rs=ethr_bsz_;  ethr_bsz_=bsz; return rs; }
  inline int SetPrintLevel(int lev=0) { prtlev_=lev; return 0;}

protected:
  bool ReadNextAllFibers();      // Renvoie true si probleme 
  bool ReadNext(int fib);   // Renvoie true si probleme 
  size_t ReceiveFromSocket(int fib, char* data, size_t len);

  // 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_;
  vector< Socket > vsok_;  //  Les sockets de lecture pour chaque fibre/lien 
  vector< uint_4 > vec_cntpaq_; 
  bool stop_;
  bool rdsamefc_;  // if true, read paquets with same value of FrameCounter on different fibers/link 
  uint_8 totnbytesrd_;
  uint_8 totnpaqrd_;
  uint_8 totsamefc_;   // nombre total de paquets avec meme framecounter  

  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 

  size_t ethr_bsz_;
  int prtlev_;
};



#endif