| [3757] | 1 | //---------------------------------------------------------------- | 
|---|
|  | 2 | // ---- classes de threads pour lecture (transfert DMA) | 
|---|
|  | 3 | //      et ecriture disque pour acquisition   BAORadio       ----- | 
|---|
|  | 4 | // LAL -      R. Ansari - Juin/Juillet 2008 | 
|---|
|  | 5 | //---------------------------------------------------------------- | 
|---|
|  | 6 |  | 
|---|
|  | 7 | #include "racqueth.h" | 
|---|
|  | 8 |  | 
|---|
|  | 9 | #include <stdlib.h> | 
|---|
|  | 10 | #include <unistd.h> | 
|---|
|  | 11 | #include <fstream> | 
|---|
|  | 12 | #include <signal.h> | 
|---|
|  | 13 | #include "pexceptions.h" | 
|---|
|  | 14 | #include "ctimer.h" | 
|---|
|  | 15 | #include "timestamp.h" | 
|---|
|  | 16 |  | 
|---|
|  | 17 | #include "pciewrap.h" | 
|---|
|  | 18 | #include "brpaqu.h" | 
|---|
|  | 19 |  | 
|---|
|  | 20 | #include "resusage.h" // Pour mesure temps elapsed/CPU ... | 
|---|
|  | 21 | #include <sys/time.h>  // pour gettimeofday | 
|---|
|  | 22 |  | 
|---|
|  | 23 | //////////////////////////////////////////////////////////////////////////////////////////////////////// | 
|---|
|  | 24 | //---------------------------------------------------------------------------------------------------------- | 
|---|
|  | 25 | // Classe thread de lecture PCI-Express et recopie sur interface reseau (Ethernet) | 
|---|
|  | 26 | //---------------------------------------------------------------------------------------------------------- | 
|---|
|  | 27 |  | 
|---|
|  | 28 |  | 
|---|
|  | 29 | /* --Methode-- */ | 
|---|
|  | 30 | PCIEToEthernet::PCIEToEthernet(vector<PCIEWrapperInterface*> vec_pciw, vector<string>& destname, BRParList const& par, int portid) | 
|---|
|  | 31 | :   par_(par), vec_pciw_ (vec_pciw), destname_(destname), tcpportid_(portid) | 
|---|
|  | 32 | { | 
|---|
|  | 33 | nmaxpaq_ = par_.MaxNbPaquets(); | 
|---|
|  | 34 | swapall_ = par_.GetDataConvFg();      // select data swap/format conversion for BRPaquet | 
|---|
|  | 35 | stop_ = false; | 
|---|
|  | 36 | packSize_ = par_.RecvPaquetSize(); | 
|---|
|  | 37 | packSizeInMgr_=par_.MMgrPaquetSize(); | 
|---|
|  | 38 | sizeFr_=par_.DMASizeBytes(); | 
|---|
|  | 39 | if (vec_pciw.size() > MAXNBFIB) | 
|---|
|  | 40 | throw ParmError("PCIEToEthernet:ERROR/  vec_pciw.size() > MAXNBFIB "); | 
|---|
|  | 41 | nbDma_= vec_pciw.size(); | 
|---|
|  | 42 |  | 
|---|
| [3762] | 43 | //  true -> direct transfer of data to ethernet | 
|---|
|  | 44 | fgdirectsend_ = (par_.pci2eth_fgdirect && (swapall_==BR_Copy) ) ? true : false; | 
|---|
| [3757] | 45 | char msg[BRTCPMSGLEN]; | 
|---|
| [3758] | 46 | cout << "PCIEToEthernet/Info: Establishing TCP/IP connections ... " << endl; | 
|---|
| [3757] | 47 | for(size_t i=0; i<vec_pciw_.size(); i++) { | 
|---|
|  | 48 | vector<ClientSocket> vskt; | 
|---|
| [3758] | 49 | for(size_t j=0; j<destname_.size(); j++) { | 
|---|
| [3757] | 50 | ClientSocket sok(destname_[j], tcpportid_); | 
|---|
|  | 51 | for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0'; | 
|---|
| [3762] | 52 | uint_4 dmasz = (fgdirectsend_)  ? vec_pciw_[0]->TransferSize() : 0; | 
|---|
|  | 53 | sprintf(msg,"BAORadio-PCIEToEthernet  %d %d %d %d %d", par_.MMgrNbPaquet(), par_.MMgrPaquetSize(), dmasz, 0,i); | 
|---|
|  | 54 | //           123456789012345678901234567890 | 
|---|
| [3763] | 55 | sok.SendAll(msg,BRTCPMSGLEN); | 
|---|
|  | 56 | sok.ReceiveAll(msg,BRTCPMSGLEN); | 
|---|
| [3757] | 57 | if (strncmp(msg,"BAORadio-EthernetReader-OK",26)!=0) | 
|---|
|  | 58 | throw SocketException("PCIEToEthernet:ERROR/  Connection to EthernetReader not established "); | 
|---|
|  | 59 | cout << " PCIEToEthernet: Ethernet connection established for DMA/fiber" << i << " with " << destname_[j] << endl; | 
|---|
|  | 60 | vskt.push_back(sok); | 
|---|
|  | 61 | } | 
|---|
| [3762] | 62 | vvec_skt_.push_back(vskt); | 
|---|
|  | 63 | vec_cntpaq_.push_back(0); | 
|---|
| [3757] | 64 | } | 
|---|
| [3758] | 65 | totrdsnd_ = 0; | 
|---|
| [3759] | 66 | SetPrintLevel(); | 
|---|
| [3757] | 67 | } | 
|---|
|  | 68 |  | 
|---|
|  | 69 | /* --Methode-- */ | 
|---|
|  | 70 | PCIEToEthernet::~PCIEToEthernet() | 
|---|
|  | 71 | { | 
|---|
|  | 72 | for(size_t i=0; i<vec_pciw_.size(); i++) { | 
|---|
|  | 73 | vector<ClientSocket>& vskt = vvec_skt_[i]; | 
|---|
|  | 74 | for(size_t j=0; j<destname_.size(); j++) vskt[j].Close(); | 
|---|
|  | 75 | } | 
|---|
|  | 76 | } | 
|---|
|  | 77 |  | 
|---|
|  | 78 | /* --Methode-- */ | 
|---|
|  | 79 | void PCIEToEthernet::run() | 
|---|
|  | 80 | { | 
|---|
|  | 81 |  | 
|---|
|  | 82 | struct timeval tv1,tv2; | 
|---|
|  | 83 | gettimeofday(&tv1, NULL); | 
|---|
|  | 84 |  | 
|---|
|  | 85 | cout << " PCIEToEthernet::run() - Starting , NMaxMemZones=" <<  par_.MaxNbBlocs() | 
|---|
|  | 86 | << " MMgrNbPaquet=" << par_.MMgrNbPaquet() << " MMgr.PaqSiz=" << par_.MMgrPaquetSize() | 
|---|
|  | 87 | << " DMA-Paqsize " << packSize_ << "  " << BRPaquet::FmtConvToString(swapall_) << endl; | 
|---|
|  | 88 | setRC(1); | 
|---|
|  | 89 |  | 
|---|
|  | 90 | // sigaddset(&act.sa_mask,SIGINT);  // pour proteger le transfert DMA | 
|---|
|  | 91 | //sigaction(SIGINT,&act,NULL); | 
|---|
| [3758] | 92 | uint_4 paqszmm = par_.MMgrPaquetSize(); | 
|---|
| [3757] | 93 | uint_4 paqsz =  packSize_; | 
|---|
|  | 94 | uint_4 dmasz = vec_pciw_[0]->TransferSize(); | 
|---|
|  | 95 | //DEL  vec_pciw_[0]->StartTransfers(); | 
|---|
|  | 96 |  | 
|---|
|  | 97 | BRPaqChecker pcheck[MAXNBFIB];  // Verification/comptage des paquets | 
|---|
|  | 98 | Byte* Datas[MAXNBFIB]; | 
|---|
|  | 99 | Byte* tampon[MAXNBFIB] ; | 
|---|
|  | 100 | Byte* nexpaq[MAXNBFIB] ;  // Pour recevevoir les paquets apres reduction de taille | 
|---|
|  | 101 |  | 
|---|
|  | 102 | Byte* predtampon=NULL;  // tampon de recopie pour la reduction des tailles de paquets | 
|---|
|  | 103 | Byte* nextpaq=NULL; | 
|---|
|  | 104 | uint_4 off_acheval=0; | 
|---|
|  | 105 |  | 
|---|
|  | 106 | int nerrdma = 0; | 
|---|
|  | 107 | int maxerrdma = 10; | 
|---|
|  | 108 | bool fgarret = false; | 
|---|
|  | 109 |  | 
|---|
|  | 110 | // Initialisation des tampons pour recopie des paquets a cheval pour chaque DMA | 
|---|
|  | 111 | for (int i=0;i< (int)nbDma_ ;i++) { | 
|---|
| [3758] | 112 | //    cout << " DEBUG*AB paqsz=" << paqsz << " paqszmm=" << paqszmm << endl; | 
|---|
|  | 113 | tampon[i]=nexpaq[i]=NULL; | 
|---|
| [3757] | 114 | tampon[i] = new Byte[paqsz]; | 
|---|
|  | 115 | nexpaq[i] = new Byte[paqszmm]; | 
|---|
|  | 116 | } | 
|---|
|  | 117 |  | 
|---|
| [3762] | 118 | if (fgdirectsend_) | 
|---|
| [3757] | 119 | cout << " PCIEToEthernet::run() Direct transfer mode DMA to Ethernet ... " << endl; | 
|---|
|  | 120 | bool fgredpaq=par_.fgreducpsize; | 
|---|
|  | 121 | if (fgredpaq) { | 
|---|
|  | 122 | cout << " PCIEToEthernet::run() - PaquetSizeReduction - RedSize=" << par_.redpqsize | 
|---|
|  | 123 | << " Offset=" << par_.reducoffset << " " << ((par_.reducneedcopy)?"NeedCopy":"NOCopy") | 
|---|
|  | 124 | << "  " << BRPaquet::ReducActionToString(par_.pqreducmode) << endl; | 
|---|
|  | 125 | predtampon = new Byte[paqsz]; | 
|---|
|  | 126 | } | 
|---|
|  | 127 |  | 
|---|
|  | 128 |  | 
|---|
|  | 129 | uint_8 trnbytes[MAXNBFIB]; | 
|---|
|  | 130 | uint_4 npaqfait[MAXNBFIB]; | 
|---|
|  | 131 | for (int i=0;i< (int)nbDma_ ;i++) { | 
|---|
|  | 132 | npaqfait[i]=0; trnbytes[i]=0; | 
|---|
|  | 133 | } | 
|---|
|  | 134 | // Attente du message GO des taches lecteurs | 
|---|
|  | 135 | cout << " PCIEToEthernet::run() Waiting for GO message from EthernetReader's ..." << endl; | 
|---|
|  | 136 | char msg[BRTCPMSGLEN]; | 
|---|
|  | 137 | for(size_t i=0; i<vec_pciw_.size(); i++) { | 
|---|
|  | 138 | vector<ClientSocket>& vskt=vvec_skt_[i]; | 
|---|
|  | 139 | for(size_t j=0; j<vskt.size(); j++) { | 
|---|
|  | 140 | for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0'; | 
|---|
| [3763] | 141 | vskt[j].ReceiveAll(msg,BRTCPMSGLEN); | 
|---|
| [3758] | 142 | if (strncmp(msg,"BAORadio-EthernetReader-Ready-GO",32)!=0) { | 
|---|
|  | 143 | msg[BRTCPMSGLEN-1]='\0'; | 
|---|
|  | 144 | cout << "PCIEToEthernet:ERROR BAD Go for fiber=" << i << " dest=" << j << " msg: " << msg << endl; | 
|---|
| [3757] | 145 | throw SocketException("PCIEToEthernet:ERROR/  BAD Go message from EthernetReader "); | 
|---|
| [3758] | 146 | } | 
|---|
|  | 147 | cout << " PCIEToEthernet::run() Received GO message for Fiber/DMA " << i << " from " << destname_[j] << endl; | 
|---|
| [3757] | 148 | } | 
|---|
|  | 149 | } | 
|---|
|  | 150 |  | 
|---|
|  | 151 | cout << " PCIEToEthernet::run()  Starting DMA to Ethernet transfers - NbFibers=NbDMA=" << vec_pciw_.size() | 
|---|
|  | 152 | << " x NbEthDestinations=" << destname_.size() << endl; | 
|---|
|  | 153 |  | 
|---|
|  | 154 | // Byte* nextdma = locdata+((kmz%memgr.NbZones())*(paqsz*memgr.NbPaquets())); | 
|---|
|  | 155 | uint_4 npaqfaitg = 0; | 
|---|
| [3759] | 156 | uint_4 prtmod = par_.BlocPerFile()*par_.MMgrNbPaquet(); | 
|---|
|  | 157 | if (prtmod < 500) prtmod=500; | 
|---|
|  | 158 |  | 
|---|
| [3757] | 159 | while (npaqfaitg < nmaxpaq_) {  // Boucle global G | 
|---|
|  | 160 | if (fgarret) break; | 
|---|
|  | 161 | if (stop_) break; | 
|---|
|  | 162 |  | 
|---|
|  | 163 | // Lancement des DMA | 
|---|
|  | 164 | for (int dma=0; dma < (int)nbDma_ ;dma++) vec_pciw_[dma]->StartTransfers(); | 
|---|
|  | 165 |  | 
|---|
|  | 166 | // On pointe vers le debut de la zone a remplir aver le prochain DMA | 
|---|
|  | 167 | //-- Zone memoire locale Byte* nextdma = buff+i*paqsz; | 
|---|
|  | 168 |  | 
|---|
|  | 169 | bool fgbaddma=false; | 
|---|
|  | 170 | // On boucle sur les nbDma_ en attente de leurs terminaison | 
|---|
|  | 171 | for (int dma=0; dma <(int) nbDma_ ;dma++)  { | 
|---|
|  | 172 | Datas[dma]=vec_pciw_[dma]->GetData(); | 
|---|
|  | 173 | if (Datas[dma] == NULL) { // No data Read in DMA | 
|---|
|  | 174 | nerrdma ++;   fgbaddma=true; | 
|---|
|  | 175 | cout << "PCIEToEthernetChecker/Erreur Waiting for datas ..." << endl; | 
|---|
|  | 176 | vec_pciw_[dma]->PrintStatus(cout); | 
|---|
|  | 177 | if (nerrdma>=maxerrdma) { fgarret = true; break; } | 
|---|
|  | 178 | } | 
|---|
|  | 179 | } | 
|---|
|  | 180 | if (fgbaddma) continue; | 
|---|
| [3759] | 181 | if ((prtlev_>0)&&(npaqfaitg%prtmod==0)) | 
|---|
|  | 182 | cout << " PCIEToEthernet::run()/Info NPaqFait= " << npaqfaitg << endl; | 
|---|
| [3762] | 183 | if (fgdirectsend_) {  // Pas de copie / reduction de taille de paquet, on rebalance tel quel ... | 
|---|
| [3757] | 184 | for (int fib=0; fib<(int) nbDma_ ;fib++) { | 
|---|
| [3758] | 185 | //DBG cout << " DEBUG*B fib=" << fib << " Datas[fib]=" << hex << Datas[fib] << dec << endl; | 
|---|
|  | 186 | SendToTargets(fib, Datas[fib], dmasz); | 
|---|
| [3757] | 187 | trnbytes[fib] += (uint_8)dmasz; | 
|---|
|  | 188 | npaqfait[fib] += (trnbytes[fib]/(uint_8)paqszmm); | 
|---|
|  | 189 | if (fib==nbDma_-1) npaqfaitg += (trnbytes[fib]/(uint_8)paqszmm); | 
|---|
|  | 190 | trnbytes[fib] = trnbytes[fib]%(uint_8)paqszmm; | 
|---|
|  | 191 | } | 
|---|
|  | 192 | continue;  // On bypasse le reste | 
|---|
|  | 193 | } | 
|---|
| [3758] | 194 |  | 
|---|
| [3757] | 195 | // Si on arrive ici, c'est qu'il faut copier / reduire les paquets .... | 
|---|
|  | 196 | uint_4 curoff=0; | 
|---|
|  | 197 | //1- On traite le paquet a cheval, rempli partiellement avec le DMA d'avant si necessaire pour les n fibres | 
|---|
|  | 198 | if (off_acheval  > 0) {  // IF Numero B | 
|---|
|  | 199 | if ((paqsz-off_acheval)< dmasz) {  // IF Numero A | 
|---|
|  | 200 | for(uint_4 fib=0; fib<nbDma_; fib++) | 
|---|
|  | 201 | memcpy((void *)((tampon[fib])+off_acheval), (void *)Datas[fib], paqsz-off_acheval); | 
|---|
|  | 202 | curoff = paqsz-off_acheval;  off_acheval = 0; | 
|---|
|  | 203 | for(uint_4 fib=0; fib<nbDma_; fib++) { | 
|---|
|  | 204 | // CHECK  S'il faut faire une reduction de taille de paquet | 
|---|
|  | 205 | if (fgredpaq) { // reduction taille de paquet | 
|---|
|  | 206 | if (par_.reducneedcopy) { | 
|---|
|  | 207 | BRPaquet paqc1(tampon[fib], predtampon, paqsz, swapall_); | 
|---|
|  | 208 | BRPaquet paqc2(nexpaq[fib], par_.redpqsize); | 
|---|
|  | 209 | paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset); | 
|---|
|  | 210 | } | 
|---|
|  | 211 | else { | 
|---|
|  | 212 | BRPaquet paqc1(tampon[fib], paqsz); | 
|---|
|  | 213 | BRPaquet paqc2(nexpaq[fib], par_.redpqsize); | 
|---|
|  | 214 | paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset); | 
|---|
|  | 215 | } | 
|---|
|  | 216 | } | 
|---|
|  | 217 | else  { | 
|---|
|  | 218 | BRPaquet paqc(tampon[fib], nexpaq[fib], paqsz, swapall_); | 
|---|
|  | 219 | } | 
|---|
|  | 220 | BRPaquet paq(nexpaq[fib], packSizeInMgr_); | 
|---|
| [3758] | 221 | SendToTargets(fib, nexpaq[fib], packSizeInMgr_); | 
|---|
| [3757] | 222 | npaqfait[fib]++; | 
|---|
|  | 223 | if (fib==nbDma_-1) npaqfaitg++;  // Ne pas oublier le compteur de paquets faits | 
|---|
|  | 224 | pcheck[fib].Check(paq);   // Verification du paquet / FrameCounter | 
|---|
|  | 225 | } | 
|---|
|  | 226 | } | 
|---|
|  | 227 | else {  // se rapporte au IF numero A | 
|---|
|  | 228 | for(uint_4 fib=0; fib<nbDma_; fib++) | 
|---|
|  | 229 | memcpy((void *)(tampon[fib]+off_acheval), (void *)Datas[fib], dmasz); | 
|---|
|  | 230 | curoff =dmasz; | 
|---|
|  | 231 | off_acheval = (dmasz+off_acheval); | 
|---|
|  | 232 | } | 
|---|
|  | 233 | }  // Fin IF Numero B | 
|---|
|  | 234 |  | 
|---|
|  | 235 | //2- On traite les paquets complets qui se trouvent dans la zone du DMA | 
|---|
|  | 236 | while ((curoff+paqsz)<=dmasz) {  // while numero C | 
|---|
|  | 237 | //          if ((dma==nbDma_-1)&&(npaqfait >= nmax_* memgr.NbPaquets())) break; | 
|---|
|  | 238 | // CHECK  S'il faut faire une reduction de taille de paquet | 
|---|
|  | 239 | for(uint_4 fib=0; fib<nbDma_; fib++) { | 
|---|
|  | 240 | if (fgredpaq) { // reduction taille de paquet | 
|---|
|  | 241 | if (par_.reducneedcopy) { | 
|---|
|  | 242 | BRPaquet paqc1(Datas[fib]+curoff, predtampon, paqsz, swapall_); | 
|---|
|  | 243 | BRPaquet paqc2(nexpaq[fib], par_.redpqsize); | 
|---|
|  | 244 | paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset); | 
|---|
|  | 245 | } | 
|---|
|  | 246 | else { | 
|---|
|  | 247 | BRPaquet paqc1(Datas[fib]+curoff, paqsz); | 
|---|
|  | 248 | BRPaquet paqc2(nexpaq[fib], par_.redpqsize); | 
|---|
|  | 249 | paqc2.CopyFrom(paqc1, par_.pqreducmode, par_.reducoffset); | 
|---|
|  | 250 | } | 
|---|
|  | 251 | } | 
|---|
|  | 252 | else { | 
|---|
|  | 253 | BRPaquet paqc(Datas[fib]+curoff, nexpaq[fib], paqsz, swapall_); | 
|---|
|  | 254 | } | 
|---|
|  | 255 | BRPaquet paq(nexpaq[fib], packSizeInMgr_); | 
|---|
| [3758] | 256 | SendToTargets(fib, nexpaq[fib], packSizeInMgr_); | 
|---|
| [3757] | 257 | npaqfait[fib]++; | 
|---|
|  | 258 | if (fib==nbDma_-1) npaqfaitg++;  // Ne pas oublier le compteur de paquets faits | 
|---|
|  | 259 | pcheck[fib].Check(paq);   // Verification du paquet / FrameCounter | 
|---|
|  | 260 | } | 
|---|
|  | 261 | curoff += paqsz;  // On avance l'index dans le buffer du DMA | 
|---|
|  | 262 | } // -- FIN traitement des paquets complets ds un DMA - FIN du while numero C | 
|---|
|  | 263 | //3- On copie si besoin la fin du DMA dans la zone tampon | 
|---|
|  | 264 | if (curoff < dmasz) {  // IF numero D | 
|---|
|  | 265 | off_acheval = dmasz-curoff; | 
|---|
|  | 266 | for(uint_4 fib=0; fib<nbDma_; fib++) | 
|---|
|  | 267 | memcpy(tampon[fib], (void*)(Datas[fib]+curoff), off_acheval); | 
|---|
|  | 268 | // ne sert a rien         curoff += off_acheval; | 
|---|
|  | 269 | } // FIN du if numero D | 
|---|
|  | 270 | }  //   FIN  Boucle global G | 
|---|
|  | 271 |  | 
|---|
|  | 272 | setRC(0); | 
|---|
|  | 273 | gettimeofday(&tv2, NULL); | 
|---|
|  | 274 | double tmelaps2 = (tv2.tv_sec-tv1.tv_sec)*1000.+(tv2.tv_usec-tv1.tv_usec)/1000.; | 
|---|
|  | 275 | if (tmelaps2<0.1) tmelaps2=0.1; | 
|---|
|  | 276 | cout << " ---------- PCIEToEthernet::run()-End summary NPaqFait=" << npaqfaitg | 
|---|
| [3758] | 277 | << " TotSend (kb)=" << totrdsnd_/1024 << "------ " << endl; | 
|---|
| [3757] | 278 | for (int dma=0; dma < (int)nbDma_ ;dma++)   { | 
|---|
|  | 279 | cout << " --Fib=" << dma << " NPaqFait=" << npaqfait[dma] <<  " TotDMATransfer=" | 
|---|
|  | 280 | << vec_pciw_[dma]->TotTransferBytes()/1024 | 
|---|
|  | 281 | << " ElapsTime=" << tmelaps2 << " ms ->" | 
|---|
|  | 282 | << (double)vec_pciw_[dma]->TotTransferBytes()/tmelaps2 << " kb/s" << endl; | 
|---|
|  | 283 |  | 
|---|
|  | 284 | vector<ClientSocket>& vskt = vvec_skt_[dma]; | 
|---|
|  | 285 | cout << " TotEthTransfer="; | 
|---|
|  | 286 | for(size_t j=0; j<vskt.size(); j++) | 
|---|
|  | 287 | cout << vskt[j].NBytesSent()/1024 << " , "; | 
|---|
|  | 288 | cout << " kb" << endl; | 
|---|
|  | 289 |  | 
|---|
| [3762] | 290 | if (!fgdirectsend_) pcheck[dma].Print(cout); | 
|---|
| [3757] | 291 | } | 
|---|
|  | 292 | cout << " --------------------------------------------------------------------" << endl; | 
|---|
|  | 293 |  | 
|---|
|  | 294 | for (int i=0;i< (int)nbDma_ ;i++) { | 
|---|
|  | 295 | delete[] tampon[i]; | 
|---|
|  | 296 | delete[] nexpaq[i]; | 
|---|
|  | 297 | } | 
|---|
|  | 298 | if ((fgredpaq)&&predtampon) delete[] predtampon; | 
|---|
|  | 299 |  | 
|---|
|  | 300 | //DBG  cout << " fin thread ========================" <<endl; | 
|---|
|  | 301 | return; | 
|---|
|  | 302 | } | 
|---|
|  | 303 |  | 
|---|
|  | 304 |  | 
|---|
|  | 305 | /* --Methode-- */ | 
|---|
|  | 306 | void PCIEToEthernet::Stop() | 
|---|
|  | 307 | { | 
|---|
|  | 308 | // cout << " PCIEReaderChecker::stop()  ........ STOP" <<endl; | 
|---|
|  | 309 | stop_ = true; | 
|---|
|  | 310 |  | 
|---|
|  | 311 | } | 
|---|
|  | 312 |  | 
|---|
| [3758] | 313 | static long cnt_prt=0; | 
|---|
|  | 314 | /* --Methode-- */ | 
|---|
|  | 315 | size_t PCIEToEthernet::SendToTargets(int fib, Byte* data, size_t len) | 
|---|
|  | 316 | { | 
|---|
| [3763] | 317 | //DBG    cout << " SendToTargets/DBG" << cnt_prt << " len=" << len << " data=" << hex << (unsigned long)data << dec << endl; cnt_prt++; | 
|---|
| [3758] | 318 | vector<ClientSocket>& vskt = vvec_skt_[fib]; | 
|---|
|  | 319 | size_t rc=0; | 
|---|
| [3762] | 320 | if (vec_cntpaq_[fib]%par_.MMgrNbPaquet() == 0) { | 
|---|
|  | 321 | char msg[BRTCPMSGLEN2]; | 
|---|
|  | 322 | sprintf(msg,"PCIEToEthernet-SendCnt %d",vec_cntpaq_[fib]); | 
|---|
|  | 323 | for(size_t j=0; j<vskt.size(); j++) { | 
|---|
| [3763] | 324 | vskt[j].SendAll(msg, BRTCPMSGLEN2); | 
|---|
| [3762] | 325 | } | 
|---|
|  | 326 | if (vec_cntpaq_[fib]>0) { | 
|---|
|  | 327 | for(size_t j=0; j<vskt.size(); j++) { | 
|---|
| [3763] | 328 | vskt[j].ReceiveAll(msg, BRTCPMSGLEN2); | 
|---|
| [3762] | 329 | msg[BRTCPMSGLEN2-1]='\0'; | 
|---|
|  | 330 | if (strncmp(msg,"EthernetReader-RecvCnt",22) != 0) | 
|---|
|  | 331 | cout << " PCIEToEthernet::SendToTargets/ BAD HandShake message : " << msg << endl; | 
|---|
|  | 332 | } | 
|---|
|  | 333 | } | 
|---|
|  | 334 | } | 
|---|
| [3763] | 335 | for(size_t j=0; j<vskt.size(); j++) { | 
|---|
|  | 336 | rc += vskt[j].SendAll((const char *)data, len); | 
|---|
| [3764] | 337 | totrdsnd_ += len; | 
|---|
| [3763] | 338 | } | 
|---|
|  | 339 | vec_cntpaq_[fib]++; | 
|---|
|  | 340 | return rc; | 
|---|
| [3758] | 341 | } | 
|---|
| [3757] | 342 | //------------------------------------------------------------------ | 
|---|
|  | 343 | // Classe thread de lecture sur interface reseau (Ethernet) | 
|---|
|  | 344 | //------------------------------------------------------------------ | 
|---|
|  | 345 |  | 
|---|
|  | 346 | /* --Methode-- */ | 
|---|
|  | 347 | EthernetReader::EthernetReader(RAcqMemZoneMgr& mem, BRParList const& par, int portid, bool rdsamefc) | 
|---|
|  | 348 | :  memgr_(mem), par_(par), stop_(false), rdsamefc_(rdsamefc), tcpportid_(portid) | 
|---|
|  | 349 | { | 
|---|
|  | 350 | totnbytesrd_ = 0; | 
|---|
|  | 351 | totnpaqrd_ = 0; | 
|---|
|  | 352 | totsamefc_ = 0; | 
|---|
|  | 353 | if (memgr_.NbFibres() > MAXANAFIB) | 
|---|
|  | 354 | throw BAORadioException("EthernetReader::EthernetReader/ NbFibres>MAXANAFIB "); | 
|---|
|  | 355 | if (par_.ethr_nlink != memgr_.NbFibres()) | 
|---|
|  | 356 | throw BAORadioException("EthernetReader::EthernetReader/ NbFibres != ethr_nlink"); | 
|---|
|  | 357 |  | 
|---|
|  | 358 | packsize_=memgr_.PaqSize(); | 
|---|
|  | 359 | mid_=-2; | 
|---|
|  | 360 | mmbuf_=NULL; | 
|---|
|  | 361 | max_targ_npaq = memgr_.NbPaquets(); | 
|---|
|  | 362 | for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) mmbufib_[fib]=NULL; | 
|---|
|  | 363 |  | 
|---|
|  | 364 | for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) { | 
|---|
|  | 365 | vpaq_.push_back(BRPaquet(NULL,memgr_.PaqSize())); | 
|---|
|  | 366 | vpchk_.push_back(BRPaqChecker(true,0)); | 
|---|
|  | 367 | curfc_.push_back(0); | 
|---|
|  | 368 | totnpqrd_.push_back(0); | 
|---|
|  | 369 | totnpqok_.push_back(0); | 
|---|
|  | 370 | } | 
|---|
|  | 371 | ServerSocket srv(tcpportid_, par_.ethr_nlink); | 
|---|
|  | 372 | char msg[BRTCPMSGLEN]; | 
|---|
|  | 373 | for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) { | 
|---|
|  | 374 | Socket sok=srv.WaitClientConnection(); | 
|---|
|  | 375 | for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0'; | 
|---|
| [3763] | 376 | sok.ReceiveAll(msg,BRTCPMSGLEN); | 
|---|
| [3757] | 377 | if (strncmp(msg,"BAORadio-PCIEToEthernet",23)!=0) | 
|---|
|  | 378 | throw SocketException("EthernetReader:ERROR/  Bad message from PCIEToEthernet client !"); | 
|---|
| [3762] | 379 | int ia,ib,ic,id,ie; | 
|---|
|  | 380 | sscanf(msg+25,"%d %d %d %d %d",&ia,&ib,&ic,&id,&ie); | 
|---|
|  | 381 | if (((uint_4)ia!=par_.MMgrNbPaquet()) || ((uint_4)ib!=par_.MMgrPaquetSize())) { | 
|---|
|  | 382 | strcpy(msg,"BAORadio-EthernetReader-BAD MMgrNbPaquet/MMgrPaquetSize()"); | 
|---|
| [3763] | 383 | sok.SendAll(msg,BRTCPMSGLEN); | 
|---|
| [3762] | 384 | throw SocketException("EthernetReader:ERROR/  Bad MMgrNbPaquet/MMgrPaquetSize() from PCIEToEthernet client !"); | 
|---|
|  | 385 | } | 
|---|
| [3757] | 386 | for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0'; | 
|---|
|  | 387 | strcpy(msg,"BAORadio-EthernetReader-OK"); | 
|---|
| [3763] | 388 | sok.SendAll(msg,BRTCPMSGLEN); | 
|---|
| [3757] | 389 | vsok_.push_back(sok); | 
|---|
|  | 390 | cout << " EthernetReader/Info connection/link" << fib << " established." << endl; | 
|---|
| [3762] | 391 | vec_cntpaq_.push_back(0); | 
|---|
| [3757] | 392 | } | 
|---|
| [3759] | 393 |  | 
|---|
|  | 394 | SetPrintLevel(); | 
|---|
| [3757] | 395 | } | 
|---|
|  | 396 |  | 
|---|
|  | 397 |  | 
|---|
|  | 398 | /* --Methode-- */ | 
|---|
|  | 399 | void EthernetReader::run() | 
|---|
|  | 400 | { | 
|---|
|  | 401 | setRC(1); | 
|---|
|  | 402 | try { | 
|---|
|  | 403 | TimeStamp ts; | 
|---|
|  | 404 | Timer tm("EthernetReader", false); | 
|---|
|  | 405 | cout << " EthernetReader::run() - Starting " << ts << " NbFibres()=" << memgr_.NbFibres() | 
|---|
|  | 406 | << "  PaqSize() = " << memgr_.PaqSize() << endl; | 
|---|
|  | 407 | cout << " ...ReadMode: " << ((rdsamefc_)?"Paquets With SameFrameCounter":"All OK paquets") << endl; | 
|---|
|  | 408 |  | 
|---|
|  | 409 | cout << " Sending GO message to all PCIEToEthernet clients ... " << endl; | 
|---|
|  | 410 | char msg[BRTCPMSGLEN]; | 
|---|
|  | 411 | for(int ii=0; ii<BRTCPMSGLEN; ii++) msg[ii]='\0'; | 
|---|
|  | 412 | strcpy(msg,"BAORadio-EthernetReader-Ready-GO"); | 
|---|
|  | 413 | for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) { | 
|---|
| [3758] | 414 | cout << " EthernetReader/Debug - Sending GO message to sender " << fib << endl; | 
|---|
| [3763] | 415 | vsok_[fib].SendAll(msg,BRTCPMSGLEN); | 
|---|
| [3757] | 416 | } | 
|---|
|  | 417 | Byte* nextpaq=NULL; | 
|---|
|  | 418 | bool fgok=true; | 
|---|
|  | 419 | while (fgok&&(totnpaqrd_<par_.MaxNbPaquets())) { | 
|---|
|  | 420 | if (stop_) break; | 
|---|
|  | 421 | if ( MoveToNextTarget() ) { | 
|---|
|  | 422 | cout << "EthernetReader::run()/Error-A- MoveToNextTarget() returned true ->STOP 9" << endl; | 
|---|
|  | 423 | setRC(7);  fgok=false;  break; | 
|---|
|  | 424 | } | 
|---|
|  | 425 | for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) { | 
|---|
|  | 426 | nextpaq=GetPaquetTarget(fib); | 
|---|
|  | 427 | if (nextpaq == NULL) { // Cela ne devrait pas arriver | 
|---|
|  | 428 | cout << "EthernetReader::run()/Error-A2- GetPaquetTarget(fib)  returned NULL ->STOP 9" << endl; | 
|---|
|  | 429 | setRC(9);  fgok=false;  break; | 
|---|
|  | 430 | } | 
|---|
|  | 431 | vpaq_[fib].Set(nextpaq); | 
|---|
|  | 432 | } | 
|---|
|  | 433 | if (ReadNextAllFibers()) { fgok=false;  break; } | 
|---|
|  | 434 | totnpaqrd_++; | 
|---|
|  | 435 | } | 
|---|
|  | 436 |  | 
|---|
|  | 437 | MoveToNextTarget();  // Pour faire traiter le dernier paquet si plein | 
|---|
|  | 438 | MZoneManage(true);   // Nettoyage final | 
|---|
|  | 439 | usleep(50000);       // Attente de traitement du dernier paquet | 
|---|
|  | 440 | memgr_.Stop();        // Arret | 
|---|
|  | 441 |  | 
|---|
|  | 442 | cout << " ------------------  EthernetReader::run() END ----------------- " << endl; | 
|---|
|  | 443 | ts.SetNow(); | 
|---|
|  | 444 | tm.SplitQ(); | 
|---|
|  | 445 | cout << "  END reading TotNPaq/Link=" << totnpaqrd_ << " :"  << ts ; | 
|---|
|  | 446 | if (rdsamefc_) cout << " NSameFC=" << totsamefc_ << endl; | 
|---|
|  | 447 | else cout << endl; | 
|---|
|  | 448 | for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)  { | 
|---|
|  | 449 | int perc=0; | 
|---|
|  | 450 | if (totnpqrd_[fib]>0) perc=100*totsamefc_/totnpqrd_[fib]; | 
|---|
|  | 451 | cout << " Fiber" << fib << " TotNPaqRd=" << totnpqrd_[fib] << " TotNPaqOK=" << totnpqok_[fib] | 
|---|
|  | 452 | << " FracSameFC=" << perc << " %" << endl; | 
|---|
|  | 453 | if (prtlev_ > 0)  vpchk_[fib].Print(cout); | 
|---|
|  | 454 | } | 
|---|
|  | 455 | cout << "  TotalEthernetRead= " << totnbytesrd_/(1024*1024) << " MBytes Ethernet-Read rate= " | 
|---|
|  | 456 | << (double)(totnbytesrd_)/1024./tm.PartialElapsedTimems() << " MB/s" << endl; | 
|---|
|  | 457 | cout << " EthernetReader::run()/Timing: \n"; | 
|---|
|  | 458 | tm.Print(); | 
|---|
|  | 459 | cout << " ---------------------------------------------------------- " << endl; | 
|---|
|  | 460 |  | 
|---|
|  | 461 | }  // Fin du bloc try | 
|---|
|  | 462 | catch (std::exception& exc) { | 
|---|
|  | 463 | cout << " EthernetReader::run()/catched execption msg=  " << exc.what() << endl; | 
|---|
|  | 464 | setRC(3); | 
|---|
|  | 465 | return; | 
|---|
|  | 466 | } | 
|---|
|  | 467 | catch(...) { | 
|---|
|  | 468 | cout << " EthernetReader::run()/catched unknown ... exception " << endl; | 
|---|
|  | 469 | setRC(4); | 
|---|
|  | 470 | return; | 
|---|
|  | 471 | } | 
|---|
|  | 472 | setRC(0); | 
|---|
|  | 473 | return; | 
|---|
|  | 474 | } | 
|---|
|  | 475 |  | 
|---|
|  | 476 | /* --Methode-- */ | 
|---|
|  | 477 | bool EthernetReader::ReadNextAllFibers() | 
|---|
|  | 478 | { | 
|---|
|  | 479 | for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)  { | 
|---|
|  | 480 | if (ReadNext(fib)) return true;  // probleme | 
|---|
|  | 481 | } | 
|---|
|  | 482 | if (!rdsamefc_ || (memgr_.NbFibres()<2))  { | 
|---|
|  | 483 | totsamefc_++;  return false;  // c'est OK | 
|---|
|  | 484 | } | 
|---|
|  | 485 | uint_8 cfc=curfc_[0]; | 
|---|
|  | 486 | bool fgsamefc=true; | 
|---|
|  | 487 | for(size_t fib=1; fib<memgr_.NbFibres(); fib++) { | 
|---|
|  | 488 | if (curfc_[fib]!=cfc) { | 
|---|
|  | 489 | fgsamefc=false; | 
|---|
|  | 490 | if (curfc_[fib] > cfc)  cfc=curfc_[fib]; | 
|---|
|  | 491 | } | 
|---|
|  | 492 | } | 
|---|
|  | 493 | if (fgsamefc) { | 
|---|
|  | 494 | totsamefc_++;  return false;  // c'est OK  , same framecounter | 
|---|
|  | 495 | } | 
|---|
|  | 496 | else {  // else !fgsame | 
|---|
|  | 497 | for(uint_4 fib=0; fib<memgr_.NbFibres(); fib++) { | 
|---|
|  | 498 | while (curfc_[fib]<cfc) { | 
|---|
|  | 499 | if (ReadNext(fib)) return true;  // probleme | 
|---|
|  | 500 | } | 
|---|
|  | 501 | } | 
|---|
|  | 502 | }   // fin de  else !fgsame | 
|---|
|  | 503 | totsamefc_++; | 
|---|
|  | 504 | return false; // c'est OK | 
|---|
|  | 505 | } | 
|---|
|  | 506 |  | 
|---|
|  | 507 | /* --Methode-- */ | 
|---|
|  | 508 | bool EthernetReader::ReadNext(int fib) | 
|---|
|  | 509 | { | 
|---|
|  | 510 | bool fggood=false; | 
|---|
|  | 511 | while(!fggood) { | 
|---|
| [3759] | 512 | ReceiveFromSocket(fib, (char *)vpaq_[fib].Begin(), packsize_); | 
|---|
| [3757] | 513 | totnbytesrd_+=packsize_; | 
|---|
|  | 514 | totnpqrd_[fib]++; | 
|---|
|  | 515 | fggood = vpchk_[fib].Check(vpaq_[fib],curfc_[fib]); | 
|---|
|  | 516 | } | 
|---|
|  | 517 | totnpqok_[fib]++; | 
|---|
|  | 518 | return false; | 
|---|
|  | 519 | } | 
|---|
|  | 520 |  | 
|---|
|  | 521 | /* --Methode-- */ | 
|---|
| [3759] | 522 | size_t EthernetReader::ReceiveFromSocket(int fib, char* data, size_t len) | 
|---|
|  | 523 | { | 
|---|
|  | 524 | size_t rc =0; | 
|---|
| [3762] | 525 | if (vec_cntpaq_[fib]%par_.MMgrNbPaquet() == 0) { | 
|---|
|  | 526 | char msg[BRTCPMSGLEN2]; | 
|---|
|  | 527 | msg[0]='\0'; | 
|---|
| [3763] | 528 | vsok_[fib].ReceiveAll(msg, BRTCPMSGLEN2); | 
|---|
| [3762] | 529 | if (strncmp(msg,"PCIEToEthernet-SendCnt",22) != 0) | 
|---|
|  | 530 | cout << " EthernetReader::ReceiveFromSocket/ BAD HandShake message : " << msg << endl; | 
|---|
|  | 531 | } | 
|---|
| [3763] | 532 | rc += vsok_[fib].ReceiveAll(data, len); | 
|---|
|  | 533 | vec_cntpaq_[fib]++; | 
|---|
| [3762] | 534 |  | 
|---|
| [3764] | 535 | /* | 
|---|
|  | 536 | size_t nblk = len/ethr_bsz_; | 
|---|
|  | 537 | size_t fblk = len%ethr_bsz_; | 
|---|
|  | 538 | size_t off = 0; | 
|---|
|  | 539 | if (nblk>0) { | 
|---|
|  | 540 | for(size_t i=0; i<nblk; i++)  { | 
|---|
|  | 541 | rc += vsok_[fib].ReceiveAll(data+off, ethr_bsz_); | 
|---|
|  | 542 | off += ethr_bsz_; | 
|---|
| [3759] | 543 | } | 
|---|
|  | 544 | } | 
|---|
| [3764] | 545 | if (fblk>0) { | 
|---|
|  | 546 | rc += vsok_[fib].ReceiveAll(data+off, fblk); | 
|---|
|  | 547 | } | 
|---|
| [3762] | 548 | vec_cntpaq_[fib]++; | 
|---|
| [3764] | 549 | */ | 
|---|
| [3762] | 550 |  | 
|---|
|  | 551 | if (vec_cntpaq_[fib]%par_.MMgrNbPaquet() == 0) { | 
|---|
|  | 552 | char msg[BRTCPMSGLEN2]; | 
|---|
|  | 553 | sprintf(msg,"EthernetReader-RecvCnt %d",vec_cntpaq_[fib]); | 
|---|
| [3763] | 554 | vsok_[fib].SendAll(msg, BRTCPMSGLEN2); | 
|---|
| [3762] | 555 | } | 
|---|
| [3759] | 556 | return rc; | 
|---|
|  | 557 | } | 
|---|
|  | 558 |  | 
|---|
|  | 559 | /* --Methode-- */ | 
|---|
| [3757] | 560 | bool EthernetReader::MZoneManage(bool fgclean)    // Retourne true si probleme | 
|---|
|  | 561 | { | 
|---|
|  | 562 | /* Pour debug | 
|---|
|  | 563 | cout << " EthernetReader::MZoneManage() mid_=" << mid_ << " arg_npaq_= " << targ_npaq_ | 
|---|
|  | 564 | << " max_targ_npaq=" << max_targ_npaq << endl; | 
|---|
|  | 565 | */ | 
|---|
|  | 566 | if (mid_ >= 0)  { | 
|---|
|  | 567 | if (fgclean) memgr_.FreeMemZone(mid_, MemZS_Free); | 
|---|
|  | 568 | else memgr_.FreeMemZone(mid_, MemZS_Filled); | 
|---|
|  | 569 | } | 
|---|
|  | 570 | mmbuf_ = NULL;  targ_npaq_ = 0;  mid_ = -2; | 
|---|
|  | 571 | for (int fib=0;fib<(int)memgr_.NbFibres() ;fib++) mmbufib_[fib]=NULL; | 
|---|
|  | 572 | if (fgclean)  return false; | 
|---|
|  | 573 | mid_ = memgr_.FindMemZoneId(MemZA_Fill); | 
|---|
|  | 574 | mmbuf_ = memgr_.GetMemZone(mid_); | 
|---|
|  | 575 | if (mmbuf_==NULL)   return true; | 
|---|
|  | 576 | for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) | 
|---|
|  | 577 | mmbufib_[fib]=memgr_.GetMemZone(mid_,fib); | 
|---|
|  | 578 | return false; | 
|---|
|  | 579 | } | 
|---|