Changeset 439 in flair-src for trunk/lib/FlairCore/src/ui_com.cpp

Timestamp:

08/25/21 13:13:36 (3 years ago)

Author:

Sanahuja Guillaume

Message:

refactor

File:

• trunk/lib/FlairCore/src/ui_com.cpp (modified) (12 diffs)

trunk/lib/FlairCore/src/ui_com.cpp

@@ -153 +153 @@
 }
 
+//send, public part; dispatch to SendNRT (nrt) or rt_pipe (rt)
+//rt_pipe is read by user thread which calls SendNRT
 void ui_com::Send(char *buf, ssize_t size,int ttl) {
   if (connection_lost == true)
@@ -158 +160 @@
 
   char *tosend = buf;
-  ssize_t nb_write;
-
   if (buf[0] == XML_HEADER) {
     // cut xml header
@@ -167 +167 @@
 
 #ifdef __XENO__
-  nb_write = rt_pipe_write(&pipe, tosend, size, P_NORMAL);
-
+  //data buf
+  ssize_t nb_write = rt_pipe_write(&pipe, tosend, size, P_NORMAL);
   if (nb_write < 0) {
                 char errorMsg[256];
@@ -175 +175 @@
     Err("rt_pipe_write error %i/%i\n", nb_write, size);
   }
+  //ttl
+  nb_write = rt_pipe_write(&pipe, &ttl, sizeof(ttl), P_NORMAL);
+  if (nb_write < 0) {
+                char errorMsg[256];
+    Err("rt_pipe_write error (%s)\n", strerror_r(-nb_write, errorMsg, sizeof(errorMsg)));
+  } else if (nb_write != sizeof(ttl)) {
+    Err("rt_pipe_write error %i/%i\n", nb_write, size);
+  }
 #else //__XENO__
-
+  SendNRT(tosend,size,ttl);
+#endif //__XENO__
+}
+
+//send, private part; called to effectively send to udt
+//this is always in nrt
+void ui_com::SendNRT(char *buf, ssize_t size,int ttl) {
+  ssize_t nb_write;
+  bool sendItCompressed = false;
+ 
 #ifdef COMPRESS_FRAMES
-  bool sendItCompressed = false;
   if (buf[0] == XML_HEADER) {
     sendItCompressed = true;
   }
+#endif // COMPRESS_FRAMES
 
   if (sendItCompressed) {
     char *out;
     ssize_t out_size;
-    if (compressBuffer(tosend, size, &out, &out_size, 9) == Z_OK) {
+    if (compressBuffer(buf, size, &out, &out_size, 9) == Z_OK) {
       size = out_size;
       nb_write = UDT::sendmsg(socket_fd, out, size,ttl, true);
@@ -197 +214 @@
 
   if (!sendItCompressed) {
-    nb_write = UDT::sendmsg(socket_fd, tosend, size, ttl, true);
-  }
-
-#else // COMPRESS_FRAMES
-  nb_write = UDT::sendmsg(socket_fd, tosend, size, ttl, true);
-#endif // COMPRESS_FRAMES
+    nb_write = UDT::sendmsg(socket_fd, buf, size, ttl, true);
+  }
+
    //Printf("write %i %i\n",nb_write,size);
   if (nb_write < 0) {
@@ -214 +228 @@
         UDT::getlasterror().getErrorCode(), nb_write, size);
   }
-#endif //__XENO__
 }
 
@@ -240 +253 @@
 }
 
+//todo: le run (rt) push les données une par une 
+//le nrt recupere et decide quand envoyer :
+// -quand toutes les données d'une meme periode sont recues
+// -quand on a atteind le nb buffering
+// -le nrt alloue dynamiquement le buffer d'envoi
+//-> enelver l'alloc du buffer au startup (UpdateDataToSendSize)
 void ui_com::Run(void) {
   // check endianness
@@ -287 +306 @@
 
       for (size_t i = 0; i < data_to_send.size(); i++) {
-        if (data_to_send.at(i)->SendPeriod() == resume_period &&
-            data_to_send.at(i)->IsEnabled() == true) {
+        if (data_to_send.at(i)->SendPeriod() == resume_period && data_to_send.at(i)->IsEnabled() == true) {
           data_to_send.at(i)->CopyDatas(send_buffer + offset);
           offset += data_to_send.at(i)->SendSize();
@@ -335 +353 @@
   fd_set set;
   struct timeval timeout;
-  ssize_t nb_read, nb_write;
+  ssize_t nb_read;
+  int ttl;
 
   buf = (char *)malloc(NRT_PIPE_SIZE);
@@ -363 +382 @@
 
     if (rv == -1) {
-      if (caller->is_running == false &&
-          UDT::getlasterror().getErrorCode() == CUDTException::ETIMEOUT)
+      if (caller->is_running == false && UDT::getlasterror().getErrorCode() == CUDTException::ETIMEOUT)
         break; // timeout
       if (UDT::getlasterror().getErrorCode() != CUDTException::ETIMEOUT)
         caller->Err("epoll_wait, %s, code %i\n",
-                    UDT::getlasterror().getErrorMessage(),
-                    UDT::getlasterror().getErrorCode());
+                    UDT::getlasterror().getErrorMessage(), UDT::getlasterror().getErrorCode());
     } else if (rv == 0) {
       // printf("timeout\n"); // a timeout occured
@@ -376 +393 @@
 
     } else {
+      //read data buf and ttl
+      //RT part has made 2 rt_pipe_write; they should be available in nrt
+      //(select ensure at least one data is ready)
       nb_read = read(pipe_fd, buf, NRT_PIPE_SIZE);
       buf[nb_read] = 0;
+      read(pipe_fd, &ttl, sizeof(ttl));
 // printf("envoi\n%s\n",buf);
 
-#ifdef COMPRESS_FRAMES
-      char *out;
-      ssize_t out_size;
-
-      if (buf[0] == XML_HEADER) {
-        if (compressBuffer(buf, nb_read, &out, &out_size, 9) == Z_OK) {
-          nb_read = out_size;
-          nb_write = UDT::sendmsg(caller->socket_fd, out, out_size, -1, true);
-          free(out);
-        } else {
-          caller->Warn("Compress error, sending it uncompressed\n");
-          nb_write = UDT::sendmsg(caller->socket_fd, buf, nb_read, -1, true);
-        }
-      } else {
-        nb_write = UDT::sendmsg(caller->socket_fd, buf, nb_read, -1, true);
-      }
-#else
-      nb_write = UDT::sendmsg(caller->socket_fd, buf, nb_read, -1, true);
-#endif
-
-      if (nb_write < 0) {
-        caller->Err("UDT::sendmsg error (%s)\n",
-                    UDT::getlasterror().getErrorMessage());
-        if (UDT::getlasterror().getErrorCode() == CUDTException::ECONNLOST ||
-            UDT::getlasterror().getErrorCode() == CUDTException::EINVSOCK) {
-          caller->connection_lost = true;
-        }
-      } else if (nb_write != nb_read) {
-        caller->Err("UDT::sendmsg error %i/%i\n", nb_write, nb_read);
-      }
+      caller->SendNRT(buf, nb_read,ttl);
     }
   }
@@ -558 +550 @@
 }
 
+//allocate a buffer at startup for the worst case
+//(every data send with same pedriod)
+//TODO: dynamic size depending on what gcs asks?
 void ui_com::UpdateDataToSendSize(void) {
   send_mutex->GetMutex();