source: Sophya/trunk/AddOn/TAcq/racqueth.cc@ 3759

//----------------------------------------------------------------
// ---- 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];
  cout << "PCIEToEthernet/Info: Establishing TCP/IP connections ... " << endl;
  for(size_t i=0; i<vec_pciw_.size(); i++) {
    vector<ClientSocket> vskt;
    for(size_t j=0; j<destname_.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);    
  }
  totrdsnd_ = 0;
  SetEthSendBlockSize();
  SetPrintLevel();
}

/* --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);
  
  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_.MMgrPaquetSize();
  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++) {  
    //    cout << " DEBUG*AB paqsz=" << paqsz << " paqszmm=" << paqszmm << endl;
    tampon[i]=nexpaq[i]=NULL;
    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[j].Receive(msg,BRTCPMSGLEN);
      if (strncmp(msg,"BAORadio-EthernetReader-Ready-GO",32)!=0) {
        msg[BRTCPMSGLEN-1]='\0';
        cout << "PCIEToEthernet:ERROR BAD Go for fiber=" << i << " dest=" << j << " msg: " << msg << endl;
        throw SocketException("PCIEToEthernet:ERROR/  BAD Go message from EthernetReader ");
      }
    cout << " PCIEToEthernet::run() Received GO message for Fiber/DMA " << i << " from " << destname_[j] << 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; 
  uint_4 prtmod = par_.BlocPerFile()*par_.MMgrNbPaquet();
  if (prtmod < 500) prtmod=500;

  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 ((prtlev_>0)&&(npaqfaitg%prtmod==0)) 
      cout << " PCIEToEthernet::run()/Info NPaqFait= " << npaqfaitg << endl;
    if (fgdirectsend) {  // Pas de copie / reduction de taille de paquet, on rebalance tel quel ...
      for (int fib=0; fib<(int) nbDma_ ;fib++) {
        //DBG cout << " DEBUG*B fib=" << fib << " Datas[fib]=" << hex << Datas[fib] << dec << endl; 
        SendToTargets(fib, Datas[fib], 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_);
          SendToTargets(fib, nexpaq[fib], 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_);
          SendToTargets(fib, nexpaq[fib], 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;

}

static long cnt_prt=0;
/* --Methode-- */
size_t PCIEToEthernet::SendToTargets(int fib, Byte* data, size_t len)
{
  //  cout << " SendToTargets/DBG" << cnt_prt << " len=" << len << " data=" << hex << (unsigned long)data << dec << endl; cnt_prt++; 
  vector<ClientSocket>& vskt = vvec_skt_[fib];
  size_t rc=0;
  size_t nblk = len/eths_bsz_;
  size_t fblk = len%eths_bsz_;
  size_t off = 0;
  if (nblk>0) {
    for(size_t i=0; i<nblk; i++) {
      for(size_t j=0; j<vskt.size(); j++) {
        rc += vskt[j].Send((const char *)data+off, eths_bsz_);
        totrdsnd_ += eths_bsz_;
      }
      off += eths_bsz_;
    }
  }
  if (fblk>0) {
    for(size_t j=0; j<vskt.size(); j++) {
      rc += vskt[j].Send((const char *)data+off, fblk);
      totrdsnd_ += fblk;
    }
  }
  return rc;
}
//------------------------------------------------------------------
// 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)
{
  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; 
  }

  SetEthRecvBlockSize();
  SetPrintLevel();
}


/* --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++) {
      cout << " EthernetReader/Debug - Sending GO message to sender " << fib << endl;
      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) { 
    ReceiveFromSocket(fib, (char *)vpaq_[fib].Begin(), packsize_);
    totnbytesrd_+=packsize_;
    totnpqrd_[fib]++;
    fggood = vpchk_[fib].Check(vpaq_[fib],curfc_[fib]);
  }
  totnpqok_[fib]++;
  return false;
}

/* --Methode-- */
size_t EthernetReader::ReceiveFromSocket(int fib, char* data, size_t len)
{
  size_t nblk = len/ethr_bsz_;
  size_t fblk = len%ethr_bsz_;
  size_t off = 0;
  size_t rc =0;
  if (nblk>0) {
    for(size_t i=0; i<nblk; i++)  {
      rc += vsok_[fib].Receive(data+off, ethr_bsz_);
      off += ethr_bsz_;  
    }
  }
  if (fblk>0) {
    rc += vsok_[fib].Receive(data+off, fblk);
  }
  return rc;
}

/* --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;
}