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

Last change on this file since 254 was 254, checked in by Sanahuja Guillaume, 6 years ago

change watchdog timeout with gcs
remove unnecessary messages on bldc
File size: 9.9 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 pktSndLossTotal=0;
39
40 bool blocking = false;
41 if (UDT::setsockopt(socket, 0, UDT_RCVSYN, &blocking, sizeof(bool)) != 0)
42 fprintf(stderr,"UDT::setsockopt error (UDT_RCVSYN)\n");
43
44 heartbeat_timer = new QTimer(this);
45 connect(heartbeat_timer, SIGNAL(timeout()), this, SLOT(heartbeat()));
46 heartbeat_timer->start(HEARTBEAT_TIMER);
47 //heartbeat();//send directly the first one, but conflicts if this socket is for logs... TODO: delay start of watchdog timer on uav side
48 udtstats_timer = new QTimer(this);
49 connect(udtstats_timer, SIGNAL(timeout()), this, SLOT(getUTDStats()));
50 udtstats_timer->start(COMPUTE_UDT_STATS_TIMER);
51}
52
53UdtSocket::~UdtSocket() {
54 heartbeat_timer->stop();
55 udtstats_timer->stop();
56 if(destroySocket) UDT::close(socket);
57}
58
59void UdtSocket::setName(QString name) {
60 fprintf(stderr," %s is %s\n",this->name.toLocal8Bit().constData(),name.toLocal8Bit().constData());
61 this->name=name;
62}
63
64// send signal to uav, to check connectivity through a watchdog
65// this is necessary because we use udt (udp based), and we cannot check
66// disconnection of ground station
67void UdtSocket::heartbeat(void) {
68 char data = WATCHDOG_HEADER;
69 quint64 sent=write(&data, 1,HEARTBEAT_TIMER,true);
70 if (sent != 1 && UDT::getlasterror().getErrorCode() == 2001) {
71 heartbeat_timer->stop();
72 }
73
74}
75
76void UdtSocket::getUTDStats(void) {
77 float rxRate=((float)total_received/(COMPUTE_UDT_STATS_TIMER/1000))/1000;//in Ko/s
78 total_received=0;
79
80 UDT::TRACEINFO perf;
81 if (UDT::ERROR == UDT::perfmon(socket, &perf)) {
82 fprintf(stderr,"perfmon: %s\n",UDT::getlasterror().getErrorMessage());
83 }/* else {
84 fprintf(stderr,"%s socket stats:\n",name.toLocal8Bit().constData());
85 fprintf(stderr,"total number of sent packets, including retransmissions: %i\n",perf.pktSentTotal);
86 fprintf(stderr,"total number of received packets: %i\n",perf.pktRecvTotal);
87 fprintf(stderr,"total number of lost packets, measured in the sending side: %i\n",perf.pktSndLossTotal);
88 fprintf(stderr,"total number of lost packets, measured in the receiving side: %i\n",perf.pktRcvLossTotal);
89 fprintf(stderr,"total number of retransmitted packets, measured in the sending side: %i\n",perf.pktRetransTotal);
90 fprintf(stderr,"total number of sent ACK packets: %i\n",perf.pktSentACKTotal);
91 fprintf(stderr,"total number of received ACK packets: %i\n",perf.pktRecvACKTotal);
92 fprintf(stderr,"total number of sent NAK packets: %i\n",perf.pktSentNAKTotal);
93 fprintf(stderr,"total number of received NAK packets: %i\n",perf.pktRecvNAKTotal);
94 fprintf(stderr,"round trip time: %fms\n",perf.msRTT);
95
96 }*/
97 bool loosingPackets=false;
98 if(perf.pktSndLossTotal>pktSndLossTotal) loosingPackets=true;
99 pktSndLossTotal=perf.pktSndLossTotal;
100 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);
101 UDTStats(stats,loosingPackets);
102}
103
104QString UdtSocket::getUDTStats() {
105 return stats;
106}
107
108void UdtSocket::kill(void) {
109 stop = true;
110 deleteLater();
111}
112
113void UdtSocket::receiveData(void) {
114 int buf_size;
115 int opt_size;
116 UDT::getsockopt(socket, 0, UDT_RCVBUF, &buf_size, &opt_size);
117 char *buf = (char *)malloc(buf_size);
118 if (!buf) {
119 fprintf(stderr,"error malloc UdtSocket::receiveData buffer\n");
120 return;
121 }
122 char *uncompressbuf=(char *)malloc(COMPRESS_CHUNK);
123 if (!uncompressbuf) {
124 fprintf(stderr,"error malloc UdtSocket::receiveData uncompress buffer\n");
125 free(buf);
126 return;
127 }
128 //fprintf(stderr,stderr,"receiveData %x\n",thread());
129
130 while (!stop) {
131 QCoreApplication::processEvents();
132
133 //we need to use epoll?
134 //tests are showing that UDT::recvmsg is waiting for the entire timeout before returning
135 //note that in flair::core the behaviour of UDT::recvmsg timeout is as expected
136 //is it a difference between client and server??
137 //tests also show that we need to recreate the eid every time, otherwise wait returns immediately
138 int eid = UDT::epoll_create();
139 if (eid < 0) {
140 fprintf(stderr,"%s: epoll_create error (%s)\n",name.toLocal8Bit().constData(),UDT::getlasterror().getErrorMessage());
141 }
142
143 if (UDT::epoll_add_usock(eid, socket) < 0) {
144 if (UDT::getlasterror().getErrorCode() == 5004) {
145 fprintf(stderr,"disconnected from %s\n",name.toLocal8Bit().constData());
146 heartbeat_timer->stop();
147 deleteLater();
148 stop=true;;
149 } else {
150 fprintf(stderr,"%s: epoll_add_usock error (%s)\n",name.toLocal8Bit().constData(),UDT::getlasterror().getErrorMessage());
151 }
152 }
153
154 int num = 1;
155 UDTSOCKET readfds;
156
157 int rv = UDT::epoll_wait2(eid, &readfds, &num,NULL, NULL,100);
158 if (rv == -1) {
159 if (UDT::getlasterror().getErrorCode() != 6003)
160 fprintf(stderr,"prob %i\n", UDT::getlasterror().getErrorCode());
161 } else if(readfds==socket && num==1 && rv==1) {
162
163 int size;
164 do {
165 QCoreApplication::processEvents();//let watchdog timer send heartbeat
166 size=UDT::recvmsg(socket, buf, buf_size);
167
168 //fprintf(stderr,"recu %i %x\n",size,buf[0]);
169 if (size > 0) {
170 total_received+=size;
171
172 switch ((unsigned char)buf[0]) {
173 case ZLIB_HEADER: {
174 ssize_t out_size;
175 uncompressBuffer(buf, size, uncompressbuf, &out_size);
176 if((unsigned char)uncompressbuf[0]==XML_HEADER && socketType==unknown) {
177 socketType=gui;
178 QString remoteName=ConnectionLayout::getDocRootName(uncompressbuf, out_size);
179 setName(remoteName);
180 emit newConnectionLayout(remoteName);//connection is Qt::BlockingQueuedConnection
181 }
182 emit dataReady(uncompressbuf, out_size);//connection is Qt::BlockingQueuedConnection, as we have only one buffer
183 break;
184 }
185 case START_SENDING_FILES: {
186 printf("log\n");
187 if((unsigned char)uncompressbuf[0]==XML_HEADER && socketType==unknown) {
188 socketType=log;
189 }
190 setName("log files");
191 heartbeat_timer->stop();
192 emit newFileUI(socket);
193 deleteLater();
194 stop=true;
195 size=-1;//exit from do while loop
196 destroySocket=false;
197 break;
198 }
199 case XML_HEADER:
200 if(socketType==unknown) {
201 socketType=gui;
202 QString remoteName=ConnectionLayout::getDocRootName(buf, size );
203 setName(remoteName);
204 emit newConnectionLayout(remoteName);
205 }
206 case DATA_BIG_ENDIAN:
207 case DATA_LITTLE_ENDIAN:
208 emit dataReady(buf, size );
209 break;
210 case CLOSING_CONNECTION:
211 fprintf(stderr,"%s, connection closed\n",name.toLocal8Bit().constData());
212 emit dataReady(buf, size );
213 stop = true;
214 heartbeat_timer->stop();
215 deleteLater();
216 break;
217 default:
218 fprintf(stderr,"udt trame non supportée %x\n", buf[0]);
219 }
220 } else {
221 //not necessary to check, watchdog (heartbeat_timer) can do it
222 //if(UDT::getlasterror().getErrorCode()!=6002 && !stop)
223 //fprintf(stderr,"udt socket: %s\n",UDT::getlasterror().getErrorMessage());
224 //UDT::close(socket);//si deconnecté
225 //free(buf);
226 //break;
227 }
228 } while(size>0);
229 } else {
230 //not necessary to check, watchdog (heartbeat_timer) can do it
231 //fprintf(stderr,"udt socket: %s\n",UDT::getlasterror().getErrorMessage());
232 }
233 UDT::epoll_remove_usock(eid, socket);
234 UDT::epoll_release(eid);
235 }
236
237 free(uncompressbuf);
238 free(buf);
239 thread()->quit();
240}
241
242qint64 UdtSocket::write(const char *buf, qint64 size,int ttl,bool inOrder) {
243 qint64 sent = UDT::sendmsg(socket, buf, size, ttl, inOrder);
244 if (sent != size) {
245 fprintf(stderr,"%s, error writting to udt (%s)\n",name.toLocal8Bit().constData(), UDT::getlasterror().getErrorMessage());
246 if (UDT::getlasterror().getErrorCode() == 2001) {
247 fprintf(stderr,"%s, closing connection\n",name.toLocal8Bit().constData());
248 stop = true;
249 heartbeat_timer->stop();
250 deleteLater();
251 }
252 }
253 return sent;
254}
255
256int UdtSocket::uncompressBuffer(char *in, ssize_t in_size, char *out,
257 ssize_t *out_size) {
258 int ret;
259 unsigned have;
260 z_stream strm;
261
262 // allocate inflate state
263 strm.zalloc = Z_NULL;
264 strm.zfree = Z_NULL;
265 strm.opaque = Z_NULL;
266 strm.avail_in = 0;
267 strm.next_in = Z_NULL;
268 ret = inflateInit(&strm);
269 if (ret != Z_OK)
270 return ret;
271
272 strm.avail_in = in_size;
273 strm.next_in = (unsigned char *)in;
274 strm.avail_out = COMPRESS_CHUNK;
275 strm.next_out = (unsigned char *)out;
276
277 ret = inflate(&strm, Z_NO_FLUSH);
278 assert(ret != Z_STREAM_ERROR); // state not clobbered
279 switch (ret) {
280 case Z_NEED_DICT:
281 ret = Z_DATA_ERROR; // and fall through
282 case Z_DATA_ERROR:
283 case Z_MEM_ERROR:
284 (void)inflateEnd(&strm);
285 return ret;
286 }
287 have = COMPRESS_CHUNK - strm.avail_out;
288 *out_size = have;
289
290 // clean up and return
291 (void)inflateEnd(&strm);
292 return ret == Z_STREAM_END ? Z_OK : Z_DATA_ERROR;
293}
Note: See TracBrowser for help on using the repository browser.