source: flair-src/trunk/tools/FlairGCS/src/UdtSocket.cpp @ 244

Last change on this file since 244 was 244, checked in by Sanahuja Guillaume, 3 years ago

modifs segfault when closing connection

File size: 8.8 KB
Line 
1// %flair:license{
2// This file is part of the Flair framework distributed under the
3// CECILL-C License, Version 1.0.
4// %flair:license}
5#include "UdtSocket.h"
6#include "ConnectionLayout.h"
7#include <stdio.h>
8#include <stdlib.h>
9#include <QApplication>
10#include <QtEndian>
11#include <QDir>
12#include <QDate>
13#include "communication.h"
14#include <zlib.h>
15#include <assert.h>
16#include <QThread>
17#define COMPRESS_CHUNK 1024*100
18
19#ifndef WIN32
20#include <arpa/inet.h>
21#else
22#include <winsock2.h>
23#include <ws2tcpip.h>
24#endif
25
26#define COMPUTE_UDT_STATS_TIMER 2000
27#define HEARTBEAT_TIMER 200
28
29using namespace std;
30
31UdtSocket::UdtSocket(UDTSOCKET socket,QString name): QObject() {
32  this->socket = socket;
33  this->name=name;
34  stop = false;
35  destroySocket=true;
36  socketType=unknown;
37  total_received=0;
38 
39  bool blocking = false;
40  if (UDT::setsockopt(socket, 0, UDT_RCVSYN, &blocking, sizeof(bool)) != 0)
41    fprintf(stderr,"UDT::setsockopt error (UDT_RCVSYN)\n");
42   
43  heartbeat_timer = new QTimer(this);
44  connect(heartbeat_timer, SIGNAL(timeout()), this, SLOT(heartbeat()));
45  heartbeat_timer->start(HEARTBEAT_TIMER);
46 
47  udtstats_timer = new QTimer(this);
48  connect(udtstats_timer, SIGNAL(timeout()), this, SLOT(getUTDStats()));
49  udtstats_timer->start(COMPUTE_UDT_STATS_TIMER);
50}
51
52UdtSocket::~UdtSocket() {
53  heartbeat_timer->stop();
54  udtstats_timer->stop();
55  if(destroySocket) UDT::close(socket);         
56}
57
58void UdtSocket::setName(QString name) {
59  fprintf(stderr,"  %s is %s\n",this->name.toLocal8Bit().constData(),name.toLocal8Bit().constData());
60  this->name=name;
61}
62
63// send signal to uav, to check connectivity through a watchdog
64// this is necessary because we use udt (udp based), and we cannot check
65// disconnection of ground station
66void UdtSocket::heartbeat(void) {
67  char data = WATCHDOG_HEADER;
68  quint64 sent=write(&data, 1,HEARTBEAT_TIMER,false);
69  if (sent != 1 && UDT::getlasterror().getErrorCode() == 2001) {
70    heartbeat_timer->stop();
71  }
72 
73}
74
75void UdtSocket::getUTDStats(void) {
76  float rxRate=((float)total_received/(COMPUTE_UDT_STATS_TIMER/1000))/1000;//in Ko/s
77  total_received=0;
78 
79  UDT::TRACEINFO perf;
80  if (UDT::ERROR == UDT::perfmon(socket, &perf))   {
81    fprintf(stderr,"perfmon: %s\n",UDT::getlasterror().getErrorMessage());
82  }/* else {
83    fprintf(stderr,"%s socket stats:\n",name.toLocal8Bit().constData());
84    fprintf(stderr,"total number of sent packets, including retransmissions: %i\n",perf.pktSentTotal);
85    fprintf(stderr,"total number of received packets: %i\n",perf.pktRecvTotal);
86    fprintf(stderr,"total number of lost packets, measured in the sending side: %i\n",perf.pktSndLossTotal);
87    fprintf(stderr,"total number of lost packets, measured in the receiving side: %i\n",perf.pktRcvLossTotal);
88    fprintf(stderr,"total number of retransmitted packets, measured in the sending side: %i\n",perf.pktRetransTotal);
89    fprintf(stderr,"total number of sent ACK packets: %i\n",perf.pktSentACKTotal);
90    fprintf(stderr,"total number of received ACK packets: %i\n",perf.pktRecvACKTotal);
91    fprintf(stderr,"total number of sent NAK packets: %i\n",perf.pktSentNAKTotal);
92    fprintf(stderr,"total number of received NAK packets: %i\n",perf.pktRecvNAKTotal);
93    fprintf(stderr,"round trip time: %fms\n",perf.msRTT);
94     
95  }*/
96  stats=QString("rx rate %1kB/s, round trip %2ms, lost packets %3").arg(rxRate,0,'f',3).arg(perf.msRTT,0,'f',3).arg(perf.pktSndLossTotal);
97  UDTStats(stats);
98 
99}
100
101QString UdtSocket::getUDTStats() {
102  return stats;
103}
104
105void UdtSocket::kill(void) {
106  stop = true;
107  deleteLater();
108}
109
110void UdtSocket::receiveData(void) {
111  int buf_size;
112  int opt_size;
113  UDT::getsockopt(socket, 0, UDT_RCVBUF, &buf_size, &opt_size);
114  char *buf = (char *)malloc(buf_size);
115  if (!buf) {
116    fprintf(stderr,"error malloc UdtSocket::receiveData buffer\n");
117    return;
118  }
119  char *uncompressbuf=(char *)malloc(COMPRESS_CHUNK);
120  if (!uncompressbuf) {
121    fprintf(stderr,"error malloc UdtSocket::receiveData uncompress buffer\n");
122    free(buf);
123    return;
124  }
125  //ffprintf(stderr,stderr,"receiveData %x\n",thread());
126 
127  while (!stop) {
128    QCoreApplication::processEvents();
129   
130    //we need to use epoll?
131    //tests are showing that UDT::recvmsg is waiting for the entire timeout before returning
132    //note that in flair::core the behaviour of UDT::recvmsg timeout is as expected
133    //is it a difference between client and server??
134    //tests also show that we need to recreate the eid every time, otherwise wait returns immediately
135    int eid = UDT::epoll_create();
136    if (eid < 0) {
137      fprintf(stderr,"%s: epoll_create error (%s)\n",name.toLocal8Bit().constData(),UDT::getlasterror().getErrorMessage());
138    }
139
140    if (UDT::epoll_add_usock(eid, socket) < 0) {
141      if (UDT::getlasterror().getErrorCode() == 5004) {
142        fprintf(stderr,"disconnected from %s\n",name.toLocal8Bit().constData());
143        heartbeat_timer->stop();
144        deleteLater();
145        stop=true;;
146      } else {
147        fprintf(stderr,"%s: epoll_add_usock error (%s)\n",name.toLocal8Bit().constData(),UDT::getlasterror().getErrorMessage());
148      }
149    }
150 
151    int num = 1;
152    UDTSOCKET readfds;
153    int rv = UDT::epoll_wait2(eid, &readfds, &num,NULL, NULL,100);
154 
155    if (rv == -1) {
156      if (UDT::getlasterror().getErrorCode() != 6003)
157        fprintf(stderr,"prob %i\n", UDT::getlasterror().getErrorCode());
158    } else if(readfds==socket && num==1 && rv==1) {
159       
160      int size=UDT::recvmsg(socket, buf, buf_size);
161      if (size > 0) {
162        total_received+=size;
163       
164        switch ((unsigned char)buf[0]) {
165          case ZLIB_HEADER: {
166            ssize_t out_size;
167            uncompressBuffer(buf, size, uncompressbuf, &out_size);
168            if((unsigned char)uncompressbuf[0]==XML_HEADER && socketType==unknown) {
169              socketType=gui;
170              QString remoteName=ConnectionLayout::getDocRootName(uncompressbuf, out_size);
171              setName(remoteName);
172              emit newConnectionLayout(remoteName);//connection is Qt::BlockingQueuedConnection
173            }
174            emit dataReady(uncompressbuf, out_size);//connection is Qt::BlockingQueuedConnection, as we have only one buffer
175            break;
176          }
177          case START_SENDING_FILES: {
178            if((unsigned char)uncompressbuf[0]==XML_HEADER && socketType==unknown) {
179              socketType=log;
180            }
181            setName("log files");
182            heartbeat_timer->stop();
183            emit newFileUI(socket);
184            deleteLater();
185            stop=true;
186            destroySocket=false;
187            break;
188          }
189          case XML_HEADER:
190            if(socketType==unknown) {
191              socketType=gui;
192              QString remoteName=ConnectionLayout::getDocRootName(buf, size );
193              setName(remoteName);
194              emit newConnectionLayout(remoteName);
195            }
196          case DATAS_BIG_ENDIAN:
197          case DATAS_LITTLE_ENDIAN:
198            emit dataReady(buf, size );
199            break;
200        }
201      } else {
202        if(UDT::getlasterror().getErrorCode()!=6002) 
203          fprintf(stderr,"udt socket: %s %i\n",UDT::getlasterror().getErrorMessage(),size);
204        //UDT::close(socket);//si deconnecté
205        //free(buf);
206        //break;
207      }
208    } else {
209      fprintf(stderr,"udt socket: %s\n",UDT::getlasterror().getErrorMessage());
210    }
211    UDT::epoll_remove_usock(eid, socket);
212    UDT::epoll_release(eid);
213  }
214  free(uncompressbuf);
215  free(buf);
216  thread()->quit();
217}
218
219qint64 UdtSocket::write(const char *buf, qint64 size,int ttl,bool inOrder) {
220  qint64 sent = UDT::sendmsg(socket, buf, size, ttl, inOrder);
221  if (sent != size) {
222    fprintf(stderr,"%s, error writting to udt (%s)\n",name.toLocal8Bit().constData(), UDT::getlasterror().getErrorMessage());
223    if (UDT::getlasterror().getErrorCode() == 2001) {
224      fprintf(stderr,"%s, closing connection\n",name.toLocal8Bit().constData());
225      stop = true;
226      heartbeat_timer->stop();
227      deleteLater();
228    }
229  }
230  return sent;
231}
232
233int UdtSocket::uncompressBuffer(char *in, ssize_t in_size, char *out,
234                                       ssize_t *out_size) {
235  int ret;
236  unsigned have;
237  z_stream strm;
238
239  // allocate inflate state
240  strm.zalloc = Z_NULL;
241  strm.zfree = Z_NULL;
242  strm.opaque = Z_NULL;
243  strm.avail_in = 0;
244  strm.next_in = Z_NULL;
245  ret = inflateInit(&strm);
246  if (ret != Z_OK)
247    return ret;
248
249  strm.avail_in = in_size;
250  strm.next_in = (unsigned char *)in;
251  strm.avail_out = COMPRESS_CHUNK;
252  strm.next_out = (unsigned char *)out;
253
254  ret = inflate(&strm, Z_NO_FLUSH);
255  assert(ret != Z_STREAM_ERROR); // state not clobbered
256  switch (ret) {
257  case Z_NEED_DICT:
258    ret = Z_DATA_ERROR; // and fall through
259  case Z_DATA_ERROR:
260  case Z_MEM_ERROR:
261    (void)inflateEnd(&strm);
262    return ret;
263  }
264  have = COMPRESS_CHUNK - strm.avail_out;
265  *out_size = have;
266
267  // clean up and return
268  (void)inflateEnd(&strm);
269  return ret == Z_STREAM_END ? Z_OK : Z_DATA_ERROR;
270}
Note: See TracBrowser for help on using the repository browser.