Changeset 441 in flair-src

Timestamp:

Aug 31, 2021, 4:15:03 PM (3 years ago)

Author:

Sanahuja Guillaume

Message:

update buffering

Location:

trunk/lib/FlairCore/src

Files:

• ui_com.cpp (modified) (13 diffs)
• unexported/ui_com.h (modified) (3 diffs)

trunk/lib/FlairCore/src/ui_com.cpp

@@ -284 +284 @@
     // on recpuere l'id de la prochaine execution
     send_mutex->GetMutex();
-    resume_id = resume_time.size();
-    for (size_t i = 0; i < resume_time.size(); i++) {
-      if (resume_time.at(i) < min && data_to_send.at(i)->IsEnabled() == true) {
-        min = resume_time.at(i);
+    resume_id = resumeTimes.size();
+    for (size_t i = 0; i < resumeTimes.size(); i++) {
+      if (resumeTimes.at(i) < min && datasToSend.at(i)->IsEnabled() == true) {
+        min = resumeTimes.at(i);
         resume_id = i;
       }
@@ -293 +293 @@
 
     // attente
-    if (resume_id < resume_time.size()) {
-      Time time = resume_time.at(resume_id);
-      uint16_t resume_period = data_to_send.at(resume_id)->SendPeriod();
+    if (resume_id < resumeTimes.size()) {
+      Time time = resumeTimes.at(resume_id);
+      uint16_t resume_period = datasToSend.at(resume_id)->SendPeriod();
       send_mutex->ReleaseMutex();
       // on dort jusqu'a la prochaine execution
@@ -305 +305 @@
       send_mutex->GetMutex();
 
-      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) {
-          data_to_send.at(i)->CopyDatas(send_buffer + offset);
-          offset += data_to_send.at(i)->SendSize();
+      for (size_t i = 0; i < datasToSend.size(); i++) {
+        if (datasToSend.at(i)->SendPeriod() == resume_period && datasToSend.at(i)->IsEnabled() == true) {
+          datasToSend.at(i)->CopyDatas(send_buffer + offset);
+          offset += datasToSend.at(i)->SendSize();
         }
       }
@@ -320 +320 @@
 
       //test to push datas
-      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) {
-          PushDatasToSend(data_to_send.at(i));
+      for (size_t i = 0; i < datasToSend.size(); i++) {
+        if (datasToSend.at(i)->SendPeriod() == resume_period && datasToSend.at(i)->IsEnabled() == true) {
+          PushDatasToSend(datasToSend.at(i));
         }
       }
@@ -328 +328 @@
       
       // on planifie la prochaine execution
-      for (size_t i = 0; i < data_to_send.size(); i++) {
-        if (data_to_send.at(i)->SendPeriod() == resume_period) {
-          resume_time.at(i) += data_to_send.at(i)->SendPeriod() * 1000000;
+      for (size_t i = 0; i < datasToSend.size(); i++) {
+        if (datasToSend.at(i)->SendPeriod() == resume_period) {
+          resumeTimes.at(i) += datasToSend.at(i)->SendPeriod() * 1000000;
         }
       }
@@ -344 +344 @@
       Time time = GetTime();
       send_mutex->GetMutex();
-      for (size_t i = 0; i < data_to_send.size(); i++) {
-        resume_time.at(i) =
-            time + (Time)data_to_send.at(i)->SendPeriod() * 1000000;
+      for (size_t i = 0; i < datasToSend.size(); i++) {
+        resumeTimes.at(i) =
+            time + (Time)datasToSend.at(i)->SendPeriod() * 1000000;
       }
       send_mutex->ReleaseMutex();
@@ -356 +356 @@
 //called with send_mutex locked
 //in nrt
-//TODO: RT part has to put datas in pipe, then useer thread call this
-void ui_com::PushDatasToSend(const SendData *data_to_send) {
+//TODO: RT part has to put datas in pipe, then user thread call this
+void ui_com::PushDatasToSend(const SendData *dataToSend) {
   PushedData_t* ptr=NULL;
   //check if we already have one buffer for this couple period/nb_buffering
-  for (size_t i = 0; i < PushedDatas.size(); i++) {
-      if (PushedDatas.at(i).period==data_to_send->SendPeriod() && PushedDatas.at(i).nb_buffering==data_to_send->NbBuffering()) {
-          Printf("match item %i, period %i nb buff %i\n",i,data_to_send->SendPeriod(),data_to_send->NbBuffering());
-          ptr=&PushedDatas.at(i);
-        }
-  }
-  //otherwire create new one
+  for (size_t i = 0; i < pushedDatas.size(); i++) {
+      if (pushedDatas.at(i).period==dataToSend->SendPeriod() && pushedDatas.at(i).nb_buffering==dataToSend->NbBuffering()) {
+          ptr=&pushedDatas.at(i);
+        }
+  }
   if(ptr==NULL) {
-    printf("no match\n");
-    PushedData_t tmp;
-    tmp.period=data_to_send->SendPeriod();
-    tmp.nb_buffering=data_to_send->NbBuffering();
-    tmp.size=0;
-    tmp.buf=NULL;
-    PushedDatas.push_back(tmp);
-    ptr=&PushedDatas.back();
-  }
-   
-  ptr->buf=(char*)realloc(ptr->buf,ptr->size+data_to_send->SendSize());
+      Warn("no match in buffers\n");
+      return;
+  }
   
-  data_to_send->CopyDatas(ptr->buf+ptr->size);
-  ptr->size+=data_to_send->SendSize();
-  Printf("nb buffered %i\n",ptr->size);
-//TODO: savoir quand a fait tous les nb buffers
-  Printf("pushed size %i period %i nb buffering %i\n",data_to_send->SendSize(),data_to_send->SendPeriod(),data_to_send->NbBuffering());
+  dataToSend->CopyDatas(ptr->buf+ptr->actual_size);
+  ptr->actual_size+=dataToSend->SendSize();
+  Printf("nb buffered %i\n",ptr->actual_size);
+  Printf("pushed size %i period %i nb buffering %i\n",dataToSend->SendSize(),dataToSend->SendPeriod(),dataToSend->NbBuffering());
+  if(ptr->actual_size==ptr->final_size) {
+      Printf("ready to send\n");
+      //clean
+      ptr->actual_size=0;
+  }
 }
 
@@ -546 +540 @@
   send_mutex->GetMutex();
 
-  resume_time.push_back(GetTime() + (Time)obj->SendPeriod() * 1000000);
+  resumeTimes.push_back(GetTime() + (Time)obj->SendPeriod() * 1000000);
 
   // on resume en meme temps tout ceux qui ont la meme periode
-  for (size_t i = 0; i < data_to_send.size(); i++) {
-    if (data_to_send.at(i)->SendPeriod() == obj->SendPeriod()) {
-      resume_time.at(resume_time.size() - 1) = resume_time.at(i);
+  for (size_t i = 0; i < datasToSend.size(); i++) {
+    if (datasToSend.at(i)->SendPeriod() == obj->SendPeriod()) {
+      resumeTimes.at(resumeTimes.size() - 1) = resumeTimes.at(i);
       break;
     }
   }
 
-  data_to_send.push_back(obj);
+  datasToSend.push_back(obj);
 
   send_mutex->ReleaseMutex();
 }
 
+//TODO: check if it is RT compatible
 void ui_com::UpdateSendData(const SendData *obj) {
   // le mutex est deja pris par l'appellant
@@ -566 +561 @@
 
   // on recupere l'id
-  for (i = 0; i < data_to_send.size(); i++) {
-    if (data_to_send.at(i) == obj) {
+  for (i = 0; i < datasToSend.size(); i++) {
+    if (datasToSend.at(i) == obj) {
       id = i;
       break;
@@ -574 +569 @@
 
   // on resume en meme temps tout ceux qui ont la meme periode
-  for (i = 0; i < data_to_send.size(); i++) {
+  for (i = 0; i < datasToSend.size(); i++) {
     if (i == id)
       continue;
-    if (data_to_send.at(i)->IsEnabled() == true &&
-        data_to_send.at(i)->SendPeriod() == obj->SendPeriod()) {
-      resume_time.at(id) = resume_time.at(i);
+    if (datasToSend.at(i)->IsEnabled() == true && datasToSend.at(i)->SendPeriod() == obj->SendPeriod()) {
+      resumeTimes.at(id) = resumeTimes.at(i);
       break;
     }
@@ -585 +579 @@
 
   // si aucun match, on planifie l'execution
-  if (i == data_to_send.size())
-    resume_time.at(id) = GetTime() + (Time)obj->SendPeriod() * 1000000;
-
-  if (IsSuspended() == true)
-    Resume();
-
+  if (i == datasToSend.size())
+    resumeTimes.at(id) = GetTime() + (Time)obj->SendPeriod() * 1000000;
+
+  if (IsSuspended() == true) Resume();
+
+//update all buffers if necessary, discard datas
+    std::vector<PushedData_t>::iterator pushedDatasIterator = pushedDatas.begin();
+    while(pushedDatasIterator != pushedDatas.end()) {
+        size_t bufSize=0;
+        for (size_t i = 0; i < datasToSend.size(); i++) {
+            if (datasToSend.at(i)->IsEnabled() && datasToSend.at(i)->SendPeriod() == (*pushedDatasIterator).period && datasToSend.at(i)->NbBuffering()==(*pushedDatasIterator).nb_buffering) {
+                bufSize+=datasToSend.at(i)->SendSize();
+            }
+        }
+        bufSize*=(*pushedDatasIterator).nb_buffering;
+        if(bufSize!=(*pushedDatasIterator).final_size && bufSize!=0) {
+            Printf("change buf size %i->%i\n",(*pushedDatasIterator).final_size,bufSize);
+            (*pushedDatasIterator).actual_size=0;
+            (*pushedDatasIterator).final_size=bufSize;
+            free((*pushedDatasIterator).buf);
+            (*pushedDatasIterator).buf=(char*)malloc(bufSize);
+        }
+        if(bufSize==0) {
+            Printf("delete buf\n");
+            free((*pushedDatasIterator).buf);
+            pushedDatasIterator = pushedDatas.erase(pushedDatasIterator);
+        } else {
+            ++pushedDatasIterator;
+        }
+    }
+  
+  //check if we need a new buffer for this couple period/nb_buffering
+  if(obj->IsEnabled()) {
+      bool match=false;
+      for (size_t i = 0; i < pushedDatas.size(); i++) {
+          if (pushedDatas.at(i).period==obj->SendPeriod() && pushedDatas.at(i).nb_buffering==obj->NbBuffering()) {
+              Printf("match item %i, period %i nb buff %i\n",i,obj->SendPeriod(),obj->NbBuffering());
+              match=true;
+            }
+      }
+      //create new one
+      if(!match) {
+        printf("no match\n");
+        PushedData_t tmp;
+        tmp.period=obj->SendPeriod();
+        tmp.nb_buffering=obj->NbBuffering();
+        tmp.actual_size=0;
+        tmp.final_size=0;
+        for (size_t i = 0; i < datasToSend.size(); i++) {
+            if (datasToSend.at(i)->IsEnabled() && datasToSend.at(i)->SendPeriod() == obj->SendPeriod() && datasToSend.at(i)->NbBuffering()==obj->NbBuffering()) {
+              tmp.final_size+=datasToSend.at(i)->SendSize();
+            }
+        }
+        tmp.final_size*=obj->NbBuffering();
+        printf("final size %i\n",tmp.final_size);
+        tmp.buf=(char*)malloc(tmp.final_size);
+        if(tmp.buf!=NULL) {
+            pushedDatas.push_back(tmp);
+        } else {
+            Warn("could not allocate buffer of size %i\n",tmp.final_size);
+        }
+      }
+  }
+  Printf("nb buf %i\n",pushedDatas.size());
   return;
 }
@@ -600 +652 @@
   send_mutex->GetMutex();
   send_size = 3; // id(1)+period(2)
-  for (size_t i = 0; i < data_to_send.size(); i++) {
-    if (data_to_send[i] != NULL)
-      send_size += data_to_send[i]->SendSize();
+  for (size_t i = 0; i < datasToSend.size(); i++) {
+    if (datasToSend[i] != NULL)
+      send_size += datasToSend[i]->SendSize();
   }
 
@@ -617 +669 @@
   send_mutex->GetMutex();
   // on recupere l'id
-  for (size_t i = 0; i < data_to_send.size(); i++) {
-    if (data_to_send.at(i) == obj) {
-      data_to_send.erase(data_to_send.begin() + i);
-      resume_time.erase(resume_time.begin() + i);
+  for (size_t i = 0; i < datasToSend.size(); i++) {
+    if (datasToSend.at(i) == obj) {
+      datasToSend.erase(datasToSend.begin() + i);
+      resumeTimes.erase(resumeTimes.begin() + i);
       // printf("remove_data_to_send %i ok\n",data_to_send.size());
       break;

trunk/lib/FlairCore/src/unexported/ui_com.h

@@ -43 +43 @@
   ssize_t Receive(char *buf, ssize_t buf_size);
   void AddSendData(const flair::gui::SendData *obj);
-  void UpdateSendData(const flair::gui::SendData *obj);
+  void UpdateSendData(const flair::gui::SendData *obj);//must be called with mutex locked
   void RemoveSendData(const flair::gui::SendData *obj);
   void UpdateDataToSendSize(void);
@@ -53 +53 @@
   ssize_t send_size;
   char *send_buffer;
-  std::vector<const flair::gui::SendData *> data_to_send;
-  std::vector<flair::core::Time> resume_time;
+  std::vector<const flair::gui::SendData *> datasToSend;
+  std::vector<flair::core::Time> resumeTimes;
   flair::core::Mutex *send_mutex;
   UDTSOCKET socket_fd;
@@ -62 +62 @@
   //private part, called to effectively send to udt
   void SendNRT(char *buf, ssize_t size,int ttl);
-  void PushDatasToSend(const flair::gui::SendData *data_to_send);
+  void PushDatasToSend(const flair::gui::SendData *dataToSend);
   typedef struct {
     char* buf;
-    ssize_t size;
+    size_t actual_size;
+    size_t final_size;
     uint16_t period;
     uint16_t nb_buffering;
   } PushedData_t;
-  std::vector<PushedData_t> PushedDatas;
+  std::vector<PushedData_t> pushedDatas;
 
   static int compressBuffer(char *in, ssize_t in_size, char **out,