source: flair-src/trunk/lib/FlairCore/src/ui_com.cpp

Last change on this file was 456, checked in by Sanahuja Guillaume, 8 weeks ago

refactor nb_buffering

File size: 20.3 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//  created:    2011/05/01
6//  filename:   ui_com.cpp
7//
8//  author:     Guillaume Sanahuja
9//              Copyright Heudiasyc UMR UTC/CNRS 7253
10//
11//  version:    $Id: $
12//
13//  purpose:    classe permettant la lecture et l'ecriture sur socket UDT
14//
15//
16/*********************************************************************/
17
18#include "ui_com.h"
19#include <cstdlib>
20#include <string.h>
21#include <unistd.h>
22#include "Mutex.h"
23#include "SendData.h"
24#include "communication.h"
25#include "FrameworkManager.h"
26#include "zlib.h"
27#include <assert.h>
28#include "config.h"
29
30#ifdef __XENO__
31#include <pthread.h>
32#include <native/task.h>
33#include <native/task.h>
34#include <fcntl.h>
35#endif
36
37using std::string;
38using namespace flair::core;
39using namespace flair::gui;
40
41ui_com::ui_com(const Object *parent, UDTSOCKET sock) : Thread(parent, "send", 2,16384*2) {
42  // mutex
43  send_mutex = new Mutex(this, ObjectName());
44
45  socket_fd = sock;
46  connection_lost = false;
47
48#ifdef __XENO__
49 Err("multi buffering is not well implemented in RT\n");
50#endif
51
52#ifdef __XENO__
53  int status;
54  string tmp_name;
55
56  is_running = true;
57
58  // pipe
59  tmp_name = getFrameworkManager()->ObjectName() + "-" + ObjectName() + "-pipe";
60  // xenomai limitation
61  if (tmp_name.size() > 31)
62    Err("rt_pipe_create error (%s is too long)\n", tmp_name.c_str());
63#ifdef RT_PIPE_SIZE
64  status = rt_pipe_create(&pipe, tmp_name.c_str(), P_MINOR_AUTO, RT_PIPE_SIZE);
65#else
66  status = rt_pipe_create(&pipe, tmp_name.c_str(), P_MINOR_AUTO, 0);
67#endif
68
69  if (status != 0) {
70                char errorMsg[256];
71    Err("rt_pipe_create error (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
72    // return -1;
73  }
74
75// start user side thread
76#ifdef NRT_STACK_SIZE
77  // Initialize thread creation attributes
78  pthread_attr_t attr;
79  if (pthread_attr_init(&attr) != 0) {
80    Err("pthread_attr_init error\n");
81  }
82
83  if (pthread_attr_setstacksize(&attr, NRT_STACK_SIZE) != 0) {
84    Err("pthread_attr_setstacksize error\n");
85  }
86
87  if (pthread_create(&thread, &attr, user_thread, (void *)this) < 0)
88#else  // NRT_STACK_SIZE
89  if (pthread_create(&thread, NULL, user_thread, (void *)this) < 0)
90#endif // NRT_STACK_SIZE
91  {
92    Err("pthread_create error\n");
93    // return -1;
94  }
95#ifdef NRT_STACK_SIZE
96  if (pthread_attr_destroy(&attr) != 0) {
97    Err("pthread_attr_destroy error\n");
98  }
99#endif
100
101#endif //__XENO__
102  int timeout = 100;
103  if (UDT::setsockopt(socket_fd, 0, UDT_RCVTIMEO, &timeout, sizeof(int)) != 0)
104    Err("UDT::setsockopt error (UDT_RCVTIMEO)\n");
105    /*
106  timeout=-1;
107  if (UDT::setsockopt(socket_fd, 0, UDT_SNDTIMEO, &timeout, sizeof(int)) != 0)
108    Err("UDT::setsockopt error (UDT_SNDTIMEO)\n");
109*/
110  bool blocking = true;
111  if (UDT::setsockopt(socket_fd, 0, UDT_SNDSYN, &blocking, sizeof(bool)) != 0)
112    Err("UDT::setsockopt error (UDT_SNDSYN)\n");
113
114  if (UDT::setsockopt(socket_fd, 0, UDT_RCVSYN, &blocking, sizeof(bool)) != 0)
115    Err("UDT::setsockopt error (UDT_RCVSYN)\n");
116  //#endif //__XENO__
117
118  Start();
119}
120
121ui_com::~ui_com() {
122//Printf("destruction ui_com\n");
123
124#ifdef __XENO__
125  is_running = false;
126
127  pthread_join(thread, NULL);
128
129  int status = rt_pipe_delete(&pipe);
130  if (status != 0) {
131                char errorMsg[256];
132    Err("rt_pipe_delete error (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
133        }
134#endif
135
136  SafeStop();
137
138  if (IsSuspended() == true)
139    Resume();
140
141  Join();
142 
143  char buf=CLOSING_CONNECTION;
144  Send(&buf,1);
145 
146    for (size_t i = 0; i < pushedDatas.size(); i++) {
147      free(pushedDatas.at(i).buf);
148    }
149  //Printf("destruction ui_com ok\n");
150}
151
152//send, public part; dispatch to SendNRT (nrt) or rt_pipe (rt)
153//rt_pipe is read by user thread which calls SendNRT
154void ui_com::Send(char *buf, ssize_t size,int ttl) {
155  if (connection_lost == true)
156    return;
157
158  char *tosend = buf;
159  if (buf[0] == XML_HEADER) {
160    // cut xml header
161    tosend = strstr(buf, "<root");
162    size -= tosend - buf;
163  }
164
165#ifdef __XENO__
166  //data buf
167  ssize_t nb_write = rt_pipe_write(&pipe, tosend, size, P_NORMAL);
168  if (nb_write < 0) {
169                char errorMsg[256];
170    Err("rt_pipe_write error (%s)\n", strerror_r(-nb_write, errorMsg, sizeof(errorMsg)));
171  } else if (nb_write != size) {
172    Err("rt_pipe_write error %i/%i\n", nb_write, size);
173  }
174  //ttl
175  nb_write = rt_pipe_write(&pipe, &ttl, sizeof(ttl), P_NORMAL);
176  if (nb_write < 0) {
177                char errorMsg[256];
178    Err("rt_pipe_write error (%s)\n", strerror_r(-nb_write, errorMsg, sizeof(errorMsg)));
179  } else if (nb_write != sizeof(ttl)) {
180    Err("rt_pipe_write error %i/%i\n", nb_write, size);
181  }
182#else //__XENO__
183  SendNRT(tosend,size,ttl);
184#endif //__XENO__
185}
186
187//send, private part; called to effectively send to udt
188//this is always in nrt
189void ui_com::SendNRT(char *buf, ssize_t size,int ttl) {
190  ssize_t nb_write;
191  bool sendItCompressed = false;
192 
193#ifdef COMPRESS_FRAMES
194  if (buf[0] == XML_HEADER) {
195    sendItCompressed = true;
196  }
197#endif // COMPRESS_FRAMES
198
199  if (sendItCompressed) {
200    char *out;
201    ssize_t out_size;
202    if (compressBuffer(buf, size, &out, &out_size, 9) == Z_OK) {
203      size = out_size;
204      nb_write = UDT::sendmsg(socket_fd, out, size,ttl, true);
205      free(out);
206    } else {
207      Warn("Compress error, sending it uncompressed\n");
208      sendItCompressed = false;
209    }
210  }
211
212  if (!sendItCompressed) {
213    nb_write = UDT::sendmsg(socket_fd, buf, size, ttl, true);
214  }
215
216   //Printf("write %i %i\n",nb_write,size);
217  if (nb_write < 0) {
218    Err("UDT::sendmsg error (%s)\n", UDT::getlasterror().getErrorMessage());
219    if (UDT::getlasterror().getErrorCode() == CUDTException::ECONNLOST ||
220        UDT::getlasterror().getErrorCode() == CUDTException::EINVSOCK) {
221      connection_lost = true;
222    }
223  } else if (nb_write != size) {
224    Err("%s, code %i (%ld/%ld)\n", UDT::getlasterror().getErrorMessage(),
225        UDT::getlasterror().getErrorCode(), nb_write, size);
226  }
227}
228
229ssize_t ui_com::Receive(char *buf, ssize_t buf_size) {
230  ssize_t bytesRead = UDT::recvmsg(socket_fd, buf, buf_size);
231
232  if (bytesRead < 0) {
233    if (UDT::getlasterror().getErrorCode() == CUDTException::ECONNLOST) {
234      Err("UDT::recvmsg error (%s)\n", UDT::getlasterror().getErrorMessage());
235      connection_lost = true;
236    }
237  }
238
239  return bytesRead;
240}
241
242void ui_com::CheckConnection(void) {
243  int32_t val,len;
244 
245  int result=UDT::getsockopt(socket_fd,0,UDT_STATE,&val,&len); 
246   if (result < 0) {
247    Printf("UDT::getsockopt error (%s)\n", UDT::getlasterror().getErrorMessage());
248   }
249 // printf("opt: %i %i %i\n",result,val,len);
250}
251
252//todo: le run (rt) push les données une par une
253//le nrt recupere et decide quand envoyer :
254// -quand toutes les données d'une meme periode sont recues
255// -quand on a atteint le nb buffering
256// -le nrt alloue dynamiquement le buffer d'envoi
257void ui_com::Run(void) {
258#ifdef __XENO__
259  WarnUponSwitches(true);
260  printf("\n"); // a revoir pourquoi??
261// sans ce printf, dans le simu_roll, le suspend ne fonctionne pas...
262#endif
263  // on attend d'avoir des choses à faire
264  Suspend();
265
266  while (!ToBeStopped()) {
267    size_t resume_id;
268    Time min = 0xffffffffffffffffULL;
269
270    // on recpuere l'id de la prochaine execution
271    send_mutex->GetMutex();
272    resume_id = resumeTimes.size();
273    for (size_t i = 0; i < resumeTimes.size(); i++) {
274      if (resumeTimes.at(i) < min && datasToSend.at(i)->IsEnabled() == true) {
275        min = resumeTimes.at(i);
276        resume_id = i;
277      }
278    }
279
280    // attente
281    if (resume_id < resumeTimes.size()) {
282      Time time = resumeTimes.at(resume_id);
283      uint16_t resume_period = datasToSend.at(resume_id)->SendPeriod();
284      send_mutex->ReleaseMutex();
285      // on dort jusqu'a la prochaine execution
286      SleepUntil(time);
287
288      // envoi des donnees
289      send_mutex->GetMutex();
290
291      //multi buffering
292      for (size_t i = 0; i < datasToSend.size(); i++) {
293        if (datasToSend.at(i)->SendPeriod() == resume_period && datasToSend.at(i)->IsEnabled() == true) {
294          PushDatasToSend(datasToSend.at(i));
295        }
296      }
297     
298      // on planifie la prochaine execution
299      for (size_t i = 0; i < datasToSend.size(); i++) {
300        if (datasToSend.at(i)->SendPeriod() == resume_period) {
301          resumeTimes.at(i) += datasToSend.at(i)->SendPeriod() * 1000000;
302        }
303      }
304      send_mutex->ReleaseMutex();
305      //Printf("%i %lld\n",resume_period,GetTime()/1000000);
306    } else {
307      send_mutex->ReleaseMutex();
308      // rien a faire, suspend
309      //Printf("rien a faire suspend\n");
310      Suspend();
311      //Printf("wake\n");
312      // on planifie la prochaine execution
313      Time time = GetTime();
314      send_mutex->GetMutex();
315      for (size_t i = 0; i < datasToSend.size(); i++) {
316        resumeTimes.at(i) =
317            time + (Time)datasToSend.at(i)->SendPeriod() * 1000000;
318      }
319      send_mutex->ReleaseMutex();
320    }
321  }
322}
323
324
325//called with send_mutex locked
326//in nrt
327//TODO: RT part has to put datas in pipe, then user thread call this
328void ui_com::PushDatasToSend(const SendData *dataToSend) {
329  PushedData_t* ptr=GetCorrespondingPushedData(dataToSend);
330 
331  if(ptr==NULL) {
332      Warn("no match in buffers\n");
333      return;
334  }
335 
336  if(ptr->buf==NULL) {
337      Warn("buffer not allocated\n");
338      return;
339  }
340 
341  if(ptr->actual_size==0) {
342      if(ptr->nb_buffering>1) {
343          if (IsBigEndian()) {
344            ptr->buf[0] = MULTIPLE_DATA_BIG_ENDIAN;
345          } else {
346            ptr->buf[0]  = MULTIPLE_DATA_LITTLE_ENDIAN;
347          }
348          ptr->actual_size+=sizeof(char);
349          memcpy(ptr->buf+ptr->actual_size,&(ptr->period),sizeof(uint16_t));
350          ptr->actual_size+=sizeof(uint16_t);
351          memcpy(ptr->buf+ptr->actual_size,&(ptr->nb_buffering),sizeof(uint16_t));
352          ptr->actual_size+=sizeof(uint16_t);
353      } else {
354          if (IsBigEndian()) {
355            ptr->buf[0] = DATA_BIG_ENDIAN;
356          } else {
357            ptr->buf[0]  = DATA_LITTLE_ENDIAN;
358          }
359          ptr->actual_size+=sizeof(char);
360          memcpy(ptr->buf+ptr->actual_size,&(ptr->period),sizeof(uint16_t));
361          ptr->actual_size+=sizeof(uint16_t);
362      }
363  }
364 
365  if(ptr->actual_size+dataToSend->SendSize()>ptr->final_size) {
366      Warn("out of buf %i/%i\n",ptr->actual_size+dataToSend->SendSize(),ptr->final_size);
367      return;
368  }
369 
370  dataToSend->CopyDatas(ptr->buf+ptr->actual_size);
371  ptr->actual_size+=dataToSend->SendSize();
372 
373  //Printf("nb buffered %i/%i\n",ptr->actual_size,ptr->final_size);
374  //Printf("pushed size %i period %i nb buffering %i\n",dataToSend->SendSize(),dataToSend->SendPeriod(),dataToSend->NbBuffering());
375  if(ptr->actual_size>=ptr->final_size) {//par securité on test aussi le cas ou supérieur
376      //Printf("ready to send\n");
377      SendNRT(ptr->buf, ptr->actual_size,ptr->period);
378      //clean
379      ptr->actual_size=0;
380  }
381}
382
383
384#ifdef __XENO__
385void *ui_com::user_thread(void *arg) {
386  int pipe_fd = -1;
387  string devname;
388  char *buf = NULL;
389  ui_com *caller = (ui_com *)arg;
390  int rv;
391  fd_set set;
392  struct timeval timeout;
393  ssize_t nb_read;
394  int ttl;
395
396  buf = (char *)malloc(NRT_PIPE_SIZE);
397  if (buf == NULL) {
398    caller->Err("malloc error\n");
399  }
400
401  devname = NRT_PIPE_PATH + getFrameworkManager()->ObjectName() + "-" +
402            caller->ObjectName() + "-pipe";
403  while (pipe_fd < 0) {
404    pipe_fd = open(devname.c_str(), O_RDWR);
405    if (pipe_fd < 0 && errno != ENOENT) {
406                        char errorMsg[256];
407      caller->Err("open pipe_fd error (%s)\n", strerror_r(errno, errorMsg, sizeof(errorMsg)));
408                }
409    usleep(1000);
410  }
411
412  while (1) {
413    FD_ZERO(&set);         // clear the set
414    FD_SET(pipe_fd, &set); // add our file descriptor to the set
415
416    timeout.tv_sec = 0;
417    timeout.tv_usec = SELECT_TIMEOUT_MS * 1000;
418
419    rv = select(FD_SETSIZE, &set, NULL, NULL, &timeout);
420
421    if (rv == -1) {
422      if (caller->is_running == false && UDT::getlasterror().getErrorCode() == CUDTException::ETIMEOUT)
423        break; // timeout
424      if (UDT::getlasterror().getErrorCode() != CUDTException::ETIMEOUT)
425        caller->Err("epoll_wait, %s, code %i\n",
426                    UDT::getlasterror().getErrorMessage(), UDT::getlasterror().getErrorCode());
427    } else if (rv == 0) {
428      // printf("timeout\n"); // a timeout occured
429      if (caller->is_running == false)
430        break;
431
432    } else {
433      //read data buf and ttl
434      //RT part has made 2 rt_pipe_write; they should be available in nrt
435      //(select ensure at least one data is ready)
436      nb_read = read(pipe_fd, buf, NRT_PIPE_SIZE);
437      buf[nb_read] = 0;
438      read(pipe_fd, &ttl, sizeof(ttl));
439// printf("envoi\n%s\n",buf);
440
441      caller->SendNRT(buf, nb_read,ttl);
442    }
443  }
444
445  close(pipe_fd);
446  if (buf != NULL)
447    free(buf);
448  pthread_exit(0);
449}
450#endif
451
452int ui_com::compressBuffer(char *in, ssize_t in_size, char **out,
453                           ssize_t *out_size, int level) {
454  int ret, flush;
455  unsigned have;
456  z_stream strm;
457
458  /* allocate deflate state */
459  strm.zalloc = Z_NULL;
460  strm.zfree = Z_NULL;
461  strm.opaque = Z_NULL;
462  ret = deflateInit(&strm, level);
463  if (ret != Z_OK)
464    return ret;
465
466  *out = (char *)malloc(COMPRESS_CHUNK);
467  if (!(*out))
468    return Z_BUF_ERROR;
469
470  strm.next_in = (unsigned char *)in;
471  strm.avail_out = COMPRESS_CHUNK;
472  strm.next_out = (unsigned char *)*out;
473  strm.avail_in = in_size;
474  flush = Z_FINISH;
475
476  ret = deflate(&strm, flush); /* no bad return value */
477  if (ret == Z_STREAM_ERROR) {
478    free(*out);
479    return ret;
480  }
481
482  have = COMPRESS_CHUNK - strm.avail_out;
483  *out_size = have;
484  // printf("%i -> %i\n",in_size,have);
485  /* clean up and return */
486  (void)deflateEnd(&strm);
487
488  if (strm.avail_out != 0) {
489    return Z_OK;
490  } else {
491    return Z_STREAM_ERROR;
492  }
493}
494
495int ui_com::uncompressBuffer(unsigned char *in, ssize_t in_size,
496                             unsigned char **out, ssize_t *out_size) {
497  int ret;
498  // unsigned have;
499  z_stream strm;
500
501  /* allocate inflate state */
502  strm.zalloc = Z_NULL;
503  strm.zfree = Z_NULL;
504  strm.opaque = Z_NULL;
505  strm.avail_in = 0;
506  strm.next_in = Z_NULL;
507  ret = inflateInit(&strm);
508  if (ret != Z_OK)
509    return ret;
510
511  *out = (unsigned char *)malloc(COMPRESS_CHUNK);
512  if (!(*out))
513    return Z_BUF_ERROR;
514
515  strm.avail_in = in_size;
516  strm.next_in = in;
517  strm.avail_out = COMPRESS_CHUNK;
518  strm.next_out = *out;
519
520  ret = inflate(&strm, Z_NO_FLUSH);
521  assert(ret != Z_STREAM_ERROR); /* state not clobbered */
522  switch (ret) {
523  case Z_NEED_DICT:
524    ret = Z_DATA_ERROR; /* and fall through */
525  case Z_DATA_ERROR:
526  case Z_MEM_ERROR:
527    (void)inflateEnd(&strm);
528    return ret;
529  }
530  // have = COMPRESS_CHUNK - strm.avail_out;
531
532  /* clean up and return */
533  (void)inflateEnd(&strm);
534  return ret == Z_STREAM_END ? Z_OK : Z_DATA_ERROR;
535}
536
537bool ui_com::ConnectionLost() { return connection_lost; }
538
539void ui_com::AddSendData(const SendData *obj) {
540    send_mutex->GetMutex();
541
542    resumeTimes.push_back(0);//time will be updated in UpdateResumeTime
543    datasToSend.push_back(obj);
544    UpdateResumeTime(obj);
545   
546    send_mutex->ReleaseMutex();
547}
548
549//must be called with mutex locked
550void ui_com::UpdateResumeTime(const SendData *obj) {
551    size_t id, i;
552
553  // on recupere l'id
554  for (i = 0; i < datasToSend.size(); i++) {
555    if (datasToSend.at(i) == obj) {
556      id = i;
557      break;
558    }
559  }
560
561  // on resume en meme temps tous ceux qui ont la meme periode
562    // si quelqu'un a la meme periode on prend son resume time
563  for (i = 0; i < datasToSend.size(); i++) {
564    if (i == id) continue;
565    if (datasToSend.at(i)->IsEnabled() == true && datasToSend.at(i)->SendPeriod() == obj->SendPeriod()) {
566      resumeTimes.at(id) = resumeTimes.at(i);
567      break;
568    }
569  }
570
571  // si aucun match, on planifie l'execution
572  if (i == datasToSend.size())
573    resumeTimes.at(id) = GetTime() + (Time)obj->SendPeriod() * 1000000;
574   
575}
576
577ui_com::PushedData_t* ui_com::GetCorrespondingPushedData(const SendData *obj) {
578    PushedData_t* ptr=NULL;
579    //get buffer for this couple period/nb_buffering
580    for (size_t i = 0; i < pushedDatas.size(); i++) {
581        if (pushedDatas.at(i).period==obj->SendPeriod() && pushedDatas.at(i).nb_buffering==obj->NbBuffering()) {
582            ptr=&pushedDatas.at(i);
583            break;
584        }
585    }
586    return ptr;
587}
588 
589//TODO: check if it is RT compatible
590//must be called with mutex locked
591//called at senddata creation, or when period/nb_buffering has changed on gcs
592//todo: not ralloc when change in buf size, but delete and create a PushedData_t? permet de factoriser avec les autres methodes?
593void ui_com::UpdateSendData(const SendData *obj) {
594   // Printf("UpdateSendData %s\n",obj->ObjectName().c_str());
595  // le mutex est deja pris par l'appellant
596 
597    UpdateResumeTime(obj);
598
599    if (IsSuspended() == true) Resume();
600
601    //update all buffers if necessary, discard datas
602    std::vector<PushedData_t>::iterator pushedDatasIterator = pushedDatas.begin();
603    while(pushedDatasIterator != pushedDatas.end()) {
604        size_t bufSize=CalcBufferSize(&(*pushedDatasIterator));
605        if(bufSize!=(*pushedDatasIterator).final_size && bufSize!=0) {
606            //Printf("change buf size %i->%i\n",(*pushedDatasIterator).final_size,bufSize);
607            (*pushedDatasIterator).actual_size=0;
608            (*pushedDatasIterator).final_size=bufSize;
609            free((*pushedDatasIterator).buf);
610            (*pushedDatasIterator).buf=(char*)malloc(bufSize);
611        }
612        if(bufSize==0) {
613            //Printf("delete buf\n");
614            free((*pushedDatasIterator).buf);
615            pushedDatasIterator = pushedDatas.erase(pushedDatasIterator);
616        } else {
617            ++pushedDatasIterator;
618        }
619    }
620 
621    //check if we need a new buffer for this couple period/nb_buffering
622    if(obj->IsEnabled()) {
623        PushedData_t *ptr=GetCorrespondingPushedData(obj);
624        //create new one
625        if(ptr==NULL) AddPushedData(obj);
626    }
627    //Printf("nb buf %i\n",pushedDatas.size());
628    return;
629}
630
631//called for exemple when we add a curve to a graph
632void ui_com::UpdateDataToSendBufferSize(const SendData *obj) {
633    if(!obj->IsEnabled()) return;
634    //Printf("UpdateDataToSendBufferSize %s\n",obj->ObjectName().c_str());
635    send_mutex->GetMutex();
636
637    PushedData_t* ptr=GetCorrespondingPushedData(obj);
638 
639    //create new one
640    if(ptr==NULL) {
641        AddPushedData(obj);
642    } else {
643        ptr->final_size=CalcBufferSize(ptr);
644        if(ptr->final_size!=0) {
645            free(ptr->buf);
646            ptr->buf=(char*)malloc(ptr->final_size);
647            if(ptr->buf==NULL) {
648                Err("malloc failed for couple %i %i\n",obj->SendPeriod(),obj->NbBuffering());
649            }
650        }
651    }
652    send_mutex->ReleaseMutex();
653}
654
655void ui_com::AddPushedData(const SendData *obj) {
656    //Printf("no corresponding match for couple %i %i\n",obj->SendPeriod(),obj->NbBuffering());
657    PushedData_t tmp;
658    tmp.period=obj->SendPeriod();
659    tmp.nb_buffering=obj->NbBuffering();
660    tmp.actual_size=0;
661    tmp.final_size=CalcBufferSize(&tmp);
662    if(tmp.final_size!=0) {
663        tmp.buf=(char*)malloc(tmp.final_size);
664        if(tmp.buf!=NULL) {
665            pushedDatas.push_back(tmp);
666        } else {
667            Warn("could not allocate buffer of size %i\n",tmp.final_size);
668        }
669    }
670}
671
672
673//must be called with mutex locked
674size_t ui_com::CalcBufferSize(const PushedData_t* ptr) const {
675    size_t final_size=0;
676    //Printf("actual %i\n",ptr->final_size);
677    for (size_t i = 0; i < datasToSend.size(); i++) {
678        //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());
679        if (datasToSend.at(i)->IsEnabled() && datasToSend.at(i)->SendPeriod() == ptr->period && datasToSend.at(i)->NbBuffering()==ptr->nb_buffering) {
680            final_size+=datasToSend.at(i)->SendSize();
681            //Printf("add %i\n",datasToSend.at(i)->SendSize());
682        }
683    }
684    final_size*=ptr->nb_buffering;
685    if(final_size==0) {
686        return 0;
687    }
688    final_size+=sizeof(char)+sizeof(uint16_t);//header+period
689    if(ptr->nb_buffering>1) final_size+=sizeof(uint16_t);//+nb_buffering
690   
691    return final_size;
692}
693
694void ui_com::RemoveSendData(const SendData *obj) {
695  // printf("remove_data_to_send %i\n",data_to_send.size());
696
697  send_mutex->GetMutex();
698  // on recupere l'id
699  for (size_t i = 0; i < datasToSend.size(); i++) {
700    if (datasToSend.at(i) == obj) {
701      datasToSend.erase(datasToSend.begin() + i);
702      resumeTimes.erase(resumeTimes.begin() + i);
703      // printf("remove_data_to_send %i ok\n",data_to_send.size());
704      break;
705    }
706  }
707  send_mutex->ReleaseMutex();
708
709  return;
710}
711
712void ui_com::Block(void) { send_mutex->GetMutex(); }
713
714void ui_com::UnBlock(void) { send_mutex->ReleaseMutex(); }
Note: See TracBrowser for help on using the repository browser.