Changeset 456 in flair-src

Timestamp:

Oct 6, 2021, 11:03:44 AM (3 years ago)

Author:

Sanahuja Guillaume

Message:

refactor nb_buffering

Location:

trunk/lib/FlairCore/src

Files:

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

trunk/lib/FlairCore/src/ui_com.cpp

@@ -253 +253 @@
 //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
+// -quand on a atteint le nb buffering
 // -le nrt alloue dynamiquement le buffer d'envoi
-//-> enelver l'alloc du buffer au startup (UpdateDataToSendSize)
 void ui_com::Run(void) {
 #ifdef __XENO__
@@ -290 +289 @@
       send_mutex->GetMutex();
 
-      //mulit buffering
+      //multi buffering
       for (size_t i = 0; i < datasToSend.size(); i++) {
         if (datasToSend.at(i)->SendPeriod() == resume_period && datasToSend.at(i)->IsEnabled() == true) {
@@ -328 +327 @@
 //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==dataToSend->SendPeriod() && pushedDatas.at(i).nb_buffering==dataToSend->NbBuffering()) {
-          ptr=&pushedDatas.at(i);
-        }
-  }
+  PushedData_t* ptr=GetCorrespondingPushedData(dataToSend);
+  
   if(ptr==NULL) {
       Warn("no match in buffers\n");
+      return;
+  }
+  
+  if(ptr->buf==NULL) {
+      Warn("buffer not allocated\n");
       return;
   }
@@ -539 +538 @@
 
 void ui_com::AddSendData(const SendData *obj) {
-  send_mutex->GetMutex();
-
-  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 < datasToSend.size(); i++) {
-    if (datasToSend.at(i)->SendPeriod() == obj->SendPeriod()) {
-      resumeTimes.at(resumeTimes.size() - 1) = resumeTimes.at(i);
-      break;
-    }
-  }
-
-  datasToSend.push_back(obj);
-
-  send_mutex->ReleaseMutex();
-}
-
-//TODO: check if it is RT compatible
+    send_mutex->GetMutex();
+
+    resumeTimes.push_back(0);//time will be updated in UpdateResumeTime
+    datasToSend.push_back(obj);
+    UpdateResumeTime(obj);
+    
+    send_mutex->ReleaseMutex();
+}
+
 //must be called with mutex locked
-void ui_com::UpdateSendData(const SendData *obj) {
-   // Printf("UpdateSendData %s\n",obj->ObjectName().c_str());
-  // le mutex est deja pris par l'appellant
-  size_t id, i;
+void ui_com::UpdateResumeTime(const SendData *obj) {
+    size_t id, i;
 
   // on recupere l'id
@@ -571 +559 @@
   }
 
-  // on resume en meme temps tout ceux qui ont la meme periode
+  // on resume en meme temps tous ceux qui ont la meme periode
+    // si quelqu'un a la meme periode on prend son resume time 
   for (i = 0; i < datasToSend.size(); i++) {
-    if (i == id)
-      continue;
+    if (i == id) continue;
     if (datasToSend.at(i)->IsEnabled() == true && datasToSend.at(i)->SendPeriod() == obj->SendPeriod()) {
       resumeTimes.at(id) = resumeTimes.at(i);
@@ -584 +572 @@
   if (i == datasToSend.size())
     resumeTimes.at(id) = GetTime() + (Time)obj->SendPeriod() * 1000000;
-
-  if (IsSuspended() == true) Resume();
-
-//update all buffers if necessary, discard datas
+    
+}
+
+ui_com::PushedData_t* ui_com::GetCorrespondingPushedData(const SendData *obj) {
+    PushedData_t* ptr=NULL;
+    //get buffer for this couple period/nb_buffering
+    for (size_t i = 0; i < pushedDatas.size(); i++) {
+        if (pushedDatas.at(i).period==obj->SendPeriod() && pushedDatas.at(i).nb_buffering==obj->NbBuffering()) {
+            ptr=&pushedDatas.at(i);
+            break;
+        }
+    }
+    return ptr;
+}
+  
+//TODO: check if it is RT compatible
+//must be called with mutex locked
+//called at senddata creation, or when period/nb_buffering has changed on gcs
+//todo: not ralloc when change in buf size, but delete and create a PushedData_t? permet de factoriser avec les autres methodes?
+void ui_com::UpdateSendData(const SendData *obj) {
+   // Printf("UpdateSendData %s\n",obj->ObjectName().c_str());
+  // le mutex est deja pris par l'appellant
+  
+    UpdateResumeTime(obj);
+
+    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!=0) {
-            bufSize+=sizeof(char)+sizeof(uint16_t);//header+period
-            if((*pushedDatasIterator).nb_buffering>1) bufSize+=sizeof(uint16_t);//+nb_buffering
-        }   
+        size_t bufSize=CalcBufferSize(&(*pushedDatasIterator));
         if(bufSize!=(*pushedDatasIterator).final_size && bufSize!=0) {
             //Printf("change buf size %i->%i\n",(*pushedDatasIterator).final_size,bufSize);
@@ -617 +619 @@
     }
   
-  //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, size %i\n",i,obj->SendPeriod(),obj->NbBuffering(),pushedDatas.at(i).final_size);
-              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++) {
-            //Printf("data %s %i/%i %i %i\n",datasToSend.at(i)->ObjectName().c_str(),i,datasToSend.size(),datasToSend.at(i)->SendPeriod() ,datasToSend.at(i)->NbBuffering());
-            if (datasToSend.at(i)->IsEnabled() && datasToSend.at(i)->SendPeriod() == obj->SendPeriod() && datasToSend.at(i)->NbBuffering()==obj->NbBuffering()) {
-              tmp.final_size+=datasToSend.at(i)->SendSize();
-              //Printf("add %i\n",datasToSend.at(i)->SendSize());
+    //check if we need a new buffer for this couple period/nb_buffering
+    if(obj->IsEnabled()) {
+        PushedData_t *ptr=GetCorrespondingPushedData(obj);
+        //create new one
+        if(ptr==NULL) AddPushedData(obj);
+    }
+    //Printf("nb buf %i\n",pushedDatas.size());
+    return;
+}
+
+//called for exemple when we add a curve to a graph
+void ui_com::UpdateDataToSendBufferSize(const SendData *obj) {
+    if(!obj->IsEnabled()) return;
+    //Printf("UpdateDataToSendBufferSize %s\n",obj->ObjectName().c_str());
+    send_mutex->GetMutex();
+
+    PushedData_t* ptr=GetCorrespondingPushedData(obj);
+  
+    //create new one
+    if(ptr==NULL) {
+        AddPushedData(obj);
+    } else {
+        ptr->final_size=CalcBufferSize(ptr);
+        if(ptr->final_size!=0) {
+            free(ptr->buf);
+            ptr->buf=(char*)malloc(ptr->final_size);
+            if(ptr->buf==NULL) {
+                Err("malloc failed for couple %i %i\n",obj->SendPeriod(),obj->NbBuffering());
             }
         }
-        tmp.final_size*=obj->NbBuffering();
-        tmp.final_size+=sizeof(char)+sizeof(uint16_t);//header+period
-        if(tmp.nb_buffering>1) tmp.final_size+=sizeof(uint16_t);//+nb_buffering
-        //printf("final size %i\n",tmp.final_size);
+    }
+    send_mutex->ReleaseMutex();
+}
+
+void ui_com::AddPushedData(const SendData *obj) {
+    //Printf("no corresponding match for couple %i %i\n",obj->SendPeriod(),obj->NbBuffering());
+    PushedData_t tmp;
+    tmp.period=obj->SendPeriod();
+    tmp.nb_buffering=obj->NbBuffering();
+    tmp.actual_size=0;
+    tmp.final_size=CalcBufferSize(&tmp);
+    if(tmp.final_size!=0) {
         tmp.buf=(char*)malloc(tmp.final_size);
         if(tmp.buf!=NULL) {
@@ -651 +667 @@
             Warn("could not allocate buffer of size %i\n",tmp.final_size);
         }
-      }
-  }
-  //Printf("nb buf %i\n",pushedDatas.size());
-  return;
-}
-
-void ui_com::UpdateDataToSendBufferSize(const SendData *obj) {
-    if(!obj->IsEnabled()) return;
-    //Printf("UpdateDataToSendBufferSize %s\n",obj->ObjectName().c_str());
-    send_mutex->GetMutex();
-    
-    PushedData_t* ptr=NULL;
-    //get buffer for this couple period/nb_buffering
-    for (size_t i = 0; i < pushedDatas.size(); i++) {
-      if (pushedDatas.at(i).period==obj->SendPeriod() && pushedDatas.at(i).nb_buffering==obj->NbBuffering()) {
-          ptr=&pushedDatas.at(i);
-        }
-    }
-    if(ptr==NULL) {
-        Err("no corresponding match for couple %i %i\n",obj->SendPeriod(),obj->NbBuffering());
-        return;
-    }
-    
-    ptr->final_size=0;
+    }
+}
+
+
+//must be called with mutex locked
+size_t ui_com::CalcBufferSize(const PushedData_t* ptr) const {
+    size_t final_size=0;
     //Printf("actual %i\n",ptr->final_size);
     for (size_t i = 0; i < datasToSend.size(); i++) {
         //Printf("data %s %i/%i %i %i\n",datasToSend.at(i)->ObjectName().c_str(),i,datasToSend.size(),datasToSend.at(i)->SendPeriod() ,datasToSend.at(i)->NbBuffering());
-        if (datasToSend.at(i)->IsEnabled() && datasToSend.at(i)->SendPeriod() == obj->SendPeriod() && datasToSend.at(i)->NbBuffering()==obj->NbBuffering()) {
-          ptr->final_size+=datasToSend.at(i)->SendSize();
-          //Printf("add %i\n",datasToSend.at(i)->SendSize());
+        if (datasToSend.at(i)->IsEnabled() && datasToSend.at(i)->SendPeriod() == ptr->period && datasToSend.at(i)->NbBuffering()==ptr->nb_buffering) {
+            final_size+=datasToSend.at(i)->SendSize();
+            //Printf("add %i\n",datasToSend.at(i)->SendSize());
         }
     }
-    ptr->final_size*=obj->NbBuffering();
-    ptr->final_size+=sizeof(char)+sizeof(uint16_t);//header+period
-    if(ptr->nb_buffering>1) ptr->final_size+=sizeof(uint16_t);//+nb_buffering
-        
-    //Printf("final %i\n",ptr->final_size);
-    free(ptr->buf);
-    ptr->buf=(char*)malloc(ptr->final_size);
+    final_size*=ptr->nb_buffering;
+    if(final_size==0) {
+        return 0;
+    }
+    final_size+=sizeof(char)+sizeof(uint16_t);//header+period
+    if(ptr->nb_buffering>1) final_size+=sizeof(uint16_t);//+nb_buffering
     
-    send_mutex->ReleaseMutex();
+    return final_size;
 }
 

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

@@ -51 +51 @@
   void CheckConnection(void);
 private:
+    //datasToSend and resumeTimes are pushed/pop together
   std::vector<const flair::gui::SendData *> datasToSend;
   std::vector<flair::core::Time> resumeTimes;
@@ -61 +62 @@
   void SendNRT(char *buf, ssize_t size,int ttl);
   void PushDatasToSend(const flair::gui::SendData *dataToSend);
+  void UpdateResumeTime(const flair::gui::SendData *obj);//must be called with mutex locked
   typedef struct {
     char* buf;
@@ -69 +71 @@
   } PushedData_t;
   std::vector<PushedData_t> pushedDatas;
+  size_t CalcBufferSize(const PushedData_t* ptr) const;//must be called with mutex locked
+  PushedData_t* GetCorrespondingPushedData(const flair::gui::SendData *dataToSend);//returns null if no corresponding buffer
+  void AddPushedData(const flair::gui::SendData *obj);
+
 
   static int compressBuffer(char *in, ssize_t in_size, char **out,