Changeset 234 in flair-src for trunk/tools/FlairGCS/src/UdtSocket.cpp

Timestamp:

04/10/18 17:05:27 (6 years ago)

Author:

Sanahuja Guillaume

Message:

create file oscket only when necessary

File:

• trunk/tools/FlairGCS/src/UdtSocket.cpp (modified) (4 diffs)

trunk/tools/FlairGCS/src/UdtSocket.cpp

@@ -4 +4 @@
 // %flair:license}
 #include "UdtSocket.h"
+#include "ConnectionLayout.h"
 #include <stdio.h>
 #include <stdlib.h>
@@ -10 +11 @@
 #include <QDir>
 #include <QDate>
-#include "file_ui.h"
 #include "communication.h"
+#include <zlib.h>
+#include <assert.h>
+#include <QThread>
+#define COMPRESS_CHUNK 1024*100
 
 #ifndef WIN32
@@ -20 +24 @@
 #endif
 
+#define COMPUTE_UDT_STATS_TIMER 2000
+#define HEARTBEAT_TIMER 200
+
 using namespace std;
 
-UdtSocket::UdtSocket(UDTSOCKET file_socket, UDTSOCKET com_socket, QString name)
-    : QObject() {
-  this->file_socket = file_socket;
-  this->com_socket = com_socket;
-  this->name = name;
+UdtSocket::UdtSocket(UDTSOCKET socket,QString name): QObject() {
+  this->socket = socket;
+  this->name=name;
   stop = false;
-  file_dialog = new file_ui();
-
-  bool blocking = true;
-  UDT::setsockopt(file_socket, 0, UDT_RCVSYN, &blocking, sizeof(bool));
-
+  destroySocket=true;
+  socketType=unknown;
+  total_received=0;
+ 
+  bool blocking = false;
+  if (UDT::setsockopt(socket, 0, UDT_RCVSYN, &blocking, sizeof(bool)) != 0)
+    printf("UDT::setsockopt error (UDT_RCVSYN)\n");
+    
   heartbeat_timer = new QTimer(this);
   connect(heartbeat_timer, SIGNAL(timeout()), this, SLOT(heartbeat()));
-  heartbeat_timer->start(200);
+  heartbeat_timer->start(HEARTBEAT_TIMER);
+  
+  udtstats_timer = new QTimer(this);
+  connect(udtstats_timer, SIGNAL(timeout()), this, SLOT(getUTDStats()));
+  udtstats_timer->start(COMPUTE_UDT_STATS_TIMER);
 }
 
 UdtSocket::~UdtSocket() {
   heartbeat_timer->stop();
-
-  UDT::close(file_socket);
-  UDT::close(com_socket);
-
-  file_dialog->deleteLater();
+  udtstats_timer->stop();
+  if(destroySocket) UDT::close(socket);         
+}
+
+void UdtSocket::setName(QString name) {
+  printf("  %s is %s\n",this->name.toLocal8Bit().constData(),name.toLocal8Bit().constData());
+  this->name=name;
 }
 
@@ -52 +66 @@
 void UdtSocket::heartbeat(void) {
   char data = WATCHDOG_HEADER;
-  write(&data, 1);
+  quint64 sent=write(&data, 1,HEARTBEAT_TIMER,false);
+  if (sent != 1 && UDT::getlasterror().getErrorCode() == 2001) {
+    heartbeat_timer->stop();
+  }
+  
+}
+
+void UdtSocket::getUTDStats(void) {
+  float rxRate=((float)total_received/(COMPUTE_UDT_STATS_TIMER/1000))/1000;//in Ko/s
+  total_received=0;
+  
+  UDT::TRACEINFO perf;
+  if (UDT::ERROR == UDT::perfmon(socket, &perf))   {
+    printf("perfmon: %s\n",UDT::getlasterror().getErrorMessage());
+  }/* else {
+    printf("%s socket stats:\n",name.toLocal8Bit().constData());
+    printf("total number of sent packets, including retransmissions: %i\n",perf.pktSentTotal);
+    printf("total number of received packets: %i\n",perf.pktRecvTotal);
+    printf("total number of lost packets, measured in the sending side: %i\n",perf.pktSndLossTotal);
+    printf("total number of lost packets, measured in the receiving side: %i\n",perf.pktRcvLossTotal);
+    printf("total number of retransmitted packets, measured in the sending side: %i\n",perf.pktRetransTotal);
+    printf("total number of sent ACK packets: %i\n",perf.pktSentACKTotal);
+    printf("total number of received ACK packets: %i\n",perf.pktRecvACKTotal);
+    printf("total number of sent NAK packets: %i\n",perf.pktSentNAKTotal);
+    printf("total number of received NAK packets: %i\n",perf.pktRecvNAKTotal);
+    printf("round trip time: %fms\n",perf.msRTT);
+      
+  }*/
+  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);
+  UDTStats(stats);
+  
+}
+
+QString UdtSocket::getUDTStats() {
+  return stats;
 }
 
 void UdtSocket::kill(void) {
-  printf("disconnected\n");
   stop = true;
   deleteLater();
 }
 
-void UdtSocket::handleConnections(void) {
+void UdtSocket::receiveData(void) {
+  int buf_size;
+  int opt_size;
+  UDT::getsockopt(socket, 0, UDT_RCVBUF, &buf_size, &opt_size);
+  char *buf = (char *)malloc(buf_size);
+  if (!buf) {
+    printf("error malloc UdtSocket::receiveData buffer\n");
+    return;
+  }
+  char *uncompressbuf=(char *)malloc(COMPRESS_CHUNK);
+  if (!uncompressbuf) {
+    printf("error malloc UdtSocket::receiveData uncompress buffer\n");
+    free(buf);
+    return;
+  }
+  //fprintf(stderr,"receiveData %x\n",thread());
+  
   while (!stop) {
+    QCoreApplication::processEvents();
+    
+    //we need to use epoll?
+    //tests are showing that UDT::recvmsg is waiting for the entire timeout before returning
+    //note that in flair::core the behaviour of UDT::recvmsg timeout is as expected
+    //is it a difference between client and server??
+    //tests also show that we need to recreate the eid every time, otherwise wait returns immediately
     int eid = UDT::epoll_create();
     if (eid < 0) {
-      printf("epoll_create error: %s\n", UDT::getlasterror().getErrorMessage());
-    }
-
-    if (UDT::epoll_add_usock(eid, file_socket) < 0) {
+      printf("%s: epoll_create error (%s)\n",name.toLocal8Bit().constData(),UDT::getlasterror().getErrorMessage());
+    }
+
+    if (UDT::epoll_add_usock(eid, socket) < 0) {
       if (UDT::getlasterror().getErrorCode() == 5004) {
-        printf("epoll_add_usock error\n");
-        break;
+        printf("disconnected from %s\n",name.toLocal8Bit().constData());
+        heartbeat_timer->stop();
+        deleteLater();
+        stop=true;;
       } else {
-        printf("epoll_add_usock error: %s\n",
-               UDT::getlasterror().getErrorMessage());
+        printf("%s: epoll_add_usock error (%s)\n",name.toLocal8Bit().constData(),UDT::getlasterror().getErrorMessage());
       }
     }
-    if (UDT::epoll_add_usock(eid, com_socket) < 0) {
-      if (UDT::getlasterror().getErrorCode() == 5004) {
-        printf("epoll_add_usock error\n");
-        break;
-      } else {
-        printf("epoll_add_usock error: %s\n",
-               UDT::getlasterror().getErrorMessage());
-      }
-    }
-
-    set<UDTSOCKET> readfds;
-
-    int rv = UDT::epoll_wait(eid, &readfds, NULL, 10);
-
+  
+    int num = 1;
+    UDTSOCKET readfds;
+    int rv = UDT::epoll_wait2(eid, &readfds, &num,NULL, NULL,100);
+  
     if (rv == -1) {
       if (UDT::getlasterror().getErrorCode() != 6003)
         printf("prob %i\n", UDT::getlasterror().getErrorCode());
-      // printf("wait\n");
-    } else if (rv == 0) {
-      printf("timeout\n"); // a timeout occured
+    } else if(readfds==socket && num==1 && rv==1) {
+       
+      int size=UDT::recvmsg(socket, buf, buf_size);
+      if (size > 0) {
+        total_received+=size;
+        
+        switch ((unsigned char)buf[0]) {
+          case ZLIB_HEADER: {
+            ssize_t out_size;
+            uncompressBuffer(buf, size, uncompressbuf, &out_size);
+            if((unsigned char)uncompressbuf[0]==XML_HEADER && socketType==unknown) {
+              socketType=gui;
+              QString remoteName=ConnectionLayout::getDocRootName(uncompressbuf, out_size);
+              setName(remoteName);
+              emit newConnectionLayout(remoteName);//connection is Qt::BlockingQueuedConnection
+            }
+            emit dataReady(uncompressbuf, out_size);//connection is Qt::BlockingQueuedConnection, as we have only one buffer
+            break;
+          }
+          case START_SENDING_FILES: {
+            if((unsigned char)uncompressbuf[0]==XML_HEADER && socketType==unknown) {
+              socketType=log;
+            }
+            setName("log files");
+            heartbeat_timer->stop();
+            emit newFileUI(socket);
+            deleteLater();
+            stop=true;
+            destroySocket=false;
+            break;
+          }
+          case XML_HEADER:
+            if(socketType==unknown) {
+              socketType=gui;
+              QString remoteName=ConnectionLayout::getDocRootName(buf, size );
+              setName(remoteName);
+              emit newConnectionLayout(remoteName);
+            }
+          case DATAS_BIG_ENDIAN:
+          case DATAS_LITTLE_ENDIAN:
+            emit dataReady(buf, size );
+            break;
+        }
+      } else {
+        //if(UDT::getlasterror().getErrorCode()!=6002) 
+        printf("udt socket:%s\n",UDT::getlasterror().getErrorMessage());
+        //UDT::close(socket);//si deconnecté
+        //free(buf);
+        //break;
+      }
     } else {
-      /*
-      if(UDT::getlasterror().getErrorCode()==2001) {
-          UDT::epoll_release(eid);
-      }*/
-      for (set<UDTSOCKET>::iterator i = readfds.begin(); i != readfds.end();
-           i++) {
-        // printf("a\n");
-        if (*i == file_socket)
-          receiveFile();
-        if (*i == com_socket)
-          receiveData();
-      }
-    }
-
-    UDT::epoll_remove_usock(eid, file_socket);
-    UDT::epoll_remove_usock(eid, com_socket);
+      printf("udt socket:%s\n",UDT::getlasterror().getErrorMessage());
+    }
+    UDT::epoll_remove_usock(eid, socket);
     UDT::epoll_release(eid);
-
-    QCoreApplication::processEvents();
-  }
-  kill();
-}
-
-void UdtSocket::receiveFile(void) {
-  char *recv_buf;
-  int bytesRead;
-  static bool flag_new_seq = true;
-  static QString folder_name;
-
-  // receive file info
-  recv_buf = (char *)malloc(1024);
-  bytesRead = UDT::recvmsg(file_socket, recv_buf, 1024);
-
-  if (bytesRead <= 0) {
-    free(recv_buf);
-    return;
-  }
-
-  int size;
-  memcpy(&size, &recv_buf[1], sizeof(int));
-  if (recv_buf[0] == FILE_INFO_BIG_ENDIAN)
-    size = qFromBigEndian(size);
-
-  // printf("file_ui recu %i %x\n",bytesRead,recv_buf[0]);
-  if ((recv_buf[0] == FILE_INFO_LITTLE_ENDIAN ||
-       recv_buf[0] == FILE_INFO_BIG_ENDIAN) &&
-      size > 0) {
-    if (flag_new_seq == true) {
-      // create directory for storage
-      QDateTime dateTime = QDateTime::currentDateTime();
-      folder_name = dateTime.toString("yyyyMMdd_hhmm") + "_" + name;
-      if (QDir().exists(folder_name) == true) {
-        folder_name = dateTime.toString("yyyyMMdd_hhmm_ss") + "_" + name;
-      }
-      QDir().mkdir(folder_name);
-
-      flag_new_seq = false;
-      file_dialog->log("Creating directory " + folder_name);
-    }
-
-    QString file_name =
-        QString::fromAscii((const char *)&recv_buf[5], bytesRead - 5);
-    QString file_path = folder_name + "/" + file_name;
-    file_dialog->log(
-        QString("receiving %1 (%2 bytes)").arg(file_name).arg(size));
-    QFile fichier(file_path);
-
-    if (!fichier.open(QIODevice::WriteOnly)) {
-      file_dialog->log("      could not write to file!");
-    } else {
-      // receive file
-      recv_buf = (char *)realloc((void *)recv_buf, size);
-      bytesRead = UDT::recvmsg(file_socket, recv_buf, size);
-      if (bytesRead != size) {
-        file_dialog->log(QString("      error receiving file! (%1/%2)")
-                             .arg(bytesRead)
-                             .arg(size));
-        free(recv_buf);
-        return;
-      } else {
-        file_dialog->log("      ok");
-      }
-
-      QDataStream stream(&fichier);
-      stream.writeRawData(recv_buf, size);
-      fichier.close();
-
-      file_dialog->addFile(file_path);
-    }
-
-    free(recv_buf);
-  } else if (recv_buf[0] == END) {
-    file_dialog->endOfFiles();
-    flag_new_seq = true;
-    //end ack
-    UDT::sendmsg(file_socket,&recv_buf[0],1);
-  }
-}
-
-void UdtSocket::receiveData(void) {
-  while (1) {
-    int buf_size;
-    int opt_size;
-    UDT::getsockopt(com_socket, 0, UDT_RCVBUF, &buf_size, &opt_size);
-
-    char *buf = (char *)malloc(buf_size);
-    int size;
-    size = UDT::recvmsg(com_socket, buf, buf_size);
-    buf = (char *)realloc(buf, size + 1);
-
-    if (size > 0) {
-      buf[size] = 0;
-      emit dataReady(buf, size + 1);
-    } else {
-      // if(UDT::getlasterror().getErrorCode()!=6002) printf("udt socket:
-      // %s\n",UDT::getlasterror().getErrorMessage());
-      free(buf);
-      break;
-    }
-  }
-}
-
-void UdtSocket::write(const char *buf, qint64 size) {
-  // printf("write\n%s\n",buf);
-  qint64 sent = UDT::sendmsg(com_socket, buf, size, -1, true);
+  }
+  free(uncompressbuf);
+  free(buf);
+  thread()->quit();
+}
+
+qint64 UdtSocket::write(const char *buf, qint64 size,int ttl,bool inOrder) {
+  qint64 sent = UDT::sendmsg(socket, buf, size, ttl, inOrder);
   if (sent != size) {
-    printf("erreur envoi: %s\n", UDT::getlasterror().getErrorMessage());
+    printf("%s, error writting to udt (%s)\n",name.toLocal8Bit().constData(), UDT::getlasterror().getErrorMessage());
     if (UDT::getlasterror().getErrorCode() == 2001) {
       stop = true;
-    }
-  }
-}
+      heartbeat_timer->stop();
+      deleteLater();
+    }
+  }
+  return sent;
+}
+
+int UdtSocket::uncompressBuffer(char *in, ssize_t in_size, char *out,
+                                       ssize_t *out_size) {
+  int ret;
+  unsigned have;
+  z_stream strm;
+
+  // allocate inflate state
+  strm.zalloc = Z_NULL;
+  strm.zfree = Z_NULL;
+  strm.opaque = Z_NULL;
+  strm.avail_in = 0;
+  strm.next_in = Z_NULL;
+  ret = inflateInit(&strm);
+  if (ret != Z_OK)
+    return ret;
+
+  strm.avail_in = in_size;
+  strm.next_in = (unsigned char *)in;
+  strm.avail_out = COMPRESS_CHUNK;
+  strm.next_out = (unsigned char *)out;
+
+  ret = inflate(&strm, Z_NO_FLUSH);
+  assert(ret != Z_STREAM_ERROR); // state not clobbered
+  switch (ret) {
+  case Z_NEED_DICT:
+    ret = Z_DATA_ERROR; // and fall through
+  case Z_DATA_ERROR:
+  case Z_MEM_ERROR:
+    (void)inflateEnd(&strm);
+    return ret;
+  }
+  have = COMPRESS_CHUNK - strm.avail_out;
+  *out_size = have;
+
+  // clean up and return
+  (void)inflateEnd(&strm);
+  return ret == Z_STREAM_END ? Z_OK : Z_DATA_ERROR;
+}