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

Last change on this file since 441 was 441, checked in by Sanahuja Guillaume, 3 months ago

update buffering

File size: 19.4 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)
42    : Thread(parent, "send", 2,16384*2) {
43  // buffer envoi
44  send_buffer = (char *)malloc(3);
45  send_size = 3; // id(1)+period(2)
46
47  // mutex
48  send_mutex = NULL;
49  send_mutex = new Mutex(this, ObjectName());
50
51  socket_fd = sock;
52  connection_lost = false;
53
54#ifdef __XENO__
55  int status;
56  string tmp_name;
57
58  is_running = true;
59
60  // pipe
61  tmp_name = getFrameworkManager()->ObjectName() + "-" + ObjectName() + "-pipe";
62  // xenomai limitation
63  if (tmp_name.size() > 31)
64    Err("rt_pipe_create error (%s is too long)\n", tmp_name.c_str());
65#ifdef RT_PIPE_SIZE
66  status = rt_pipe_create(&pipe, tmp_name.c_str(), P_MINOR_AUTO, RT_PIPE_SIZE);
67#else
68  status = rt_pipe_create(&pipe, tmp_name.c_str(), P_MINOR_AUTO, 0);
69#endif
70
71  if (status != 0) {
72                char errorMsg[256];
73    Err("rt_pipe_create error (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
74    // return -1;
75  }
76
77// start user side thread
78#ifdef NRT_STACK_SIZE
79  // Initialize thread creation attributes
80  pthread_attr_t attr;
81  if (pthread_attr_init(&attr) != 0) {
82    Err("pthread_attr_init error\n");
83  }
84
85  if (pthread_attr_setstacksize(&attr, NRT_STACK_SIZE) != 0) {
86    Err("pthread_attr_setstacksize error\n");
87  }
88
89  if (pthread_create(&thread, &attr, user_thread, (void *)this) < 0)
90#else  // NRT_STACK_SIZE
91  if (pthread_create(&thread, NULL, user_thread, (void *)this) < 0)
92#endif // NRT_STACK_SIZE
93  {
94    Err("pthread_create error\n");
95    // return -1;
96  }
97#ifdef NRT_STACK_SIZE
98  if (pthread_attr_destroy(&attr) != 0) {
99    Err("pthread_attr_destroy error\n");
100  }
101#endif
102
103#endif //__XENO__
104  int timeout = 100;
105  if (UDT::setsockopt(socket_fd, 0, UDT_RCVTIMEO, &timeout, sizeof(int)) != 0)
106    Err("UDT::setsockopt error (UDT_RCVTIMEO)\n");
107    /*
108  timeout=-1;
109  if (UDT::setsockopt(socket_fd, 0, UDT_SNDTIMEO, &timeout, sizeof(int)) != 0)
110    Err("UDT::setsockopt error (UDT_SNDTIMEO)\n");
111*/
112  bool blocking = true;
113  if (UDT::setsockopt(socket_fd, 0, UDT_SNDSYN, &blocking, sizeof(bool)) != 0)
114    Err("UDT::setsockopt error (UDT_SNDSYN)\n");
115
116  if (UDT::setsockopt(socket_fd, 0, UDT_RCVSYN, &blocking, sizeof(bool)) != 0)
117    Err("UDT::setsockopt error (UDT_RCVSYN)\n");
118  //#endif //__XENO__
119
120  Start();
121}
122
123ui_com::~ui_com() {
124//Printf("destruction ui_com\n");
125
126#ifdef __XENO__
127  is_running = false;
128
129  pthread_join(thread, NULL);
130
131  int status = rt_pipe_delete(&pipe);
132  if (status != 0) {
133                char errorMsg[256];
134    Err("rt_pipe_delete error (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
135        }
136#endif
137
138  SafeStop();
139
140  if (IsSuspended() == true)
141    Resume();
142
143  Join();
144 
145  char buf=CLOSING_CONNECTION;
146  Send(&buf,1);
147 
148  if (send_buffer != NULL)
149    free(send_buffer);
150  send_buffer = NULL;
151
152  //Printf("destruction ui_com ok\n");
153}
154
155//send, public part; dispatch to SendNRT (nrt) or rt_pipe (rt)
156//rt_pipe is read by user thread which calls SendNRT
157void ui_com::Send(char *buf, ssize_t size,int ttl) {
158  if (connection_lost == true)
159    return;
160
161  char *tosend = buf;
162  if (buf[0] == XML_HEADER) {
163    // cut xml header
164    tosend = strstr(buf, "<root");
165    size -= tosend - buf;
166  }
167
168#ifdef __XENO__
169  //data buf
170  ssize_t nb_write = rt_pipe_write(&pipe, tosend, size, P_NORMAL);
171  if (nb_write < 0) {
172                char errorMsg[256];
173    Err("rt_pipe_write error (%s)\n", strerror_r(-nb_write, errorMsg, sizeof(errorMsg)));
174  } else if (nb_write != size) {
175    Err("rt_pipe_write error %i/%i\n", nb_write, size);
176  }
177  //ttl
178  nb_write = rt_pipe_write(&pipe, &ttl, sizeof(ttl), P_NORMAL);
179  if (nb_write < 0) {
180                char errorMsg[256];
181    Err("rt_pipe_write error (%s)\n", strerror_r(-nb_write, errorMsg, sizeof(errorMsg)));
182  } else if (nb_write != sizeof(ttl)) {
183    Err("rt_pipe_write error %i/%i\n", nb_write, size);
184  }
185#else //__XENO__
186  SendNRT(tosend,size,ttl);
187#endif //__XENO__
188}
189
190//send, private part; called to effectively send to udt
191//this is always in nrt
192void ui_com::SendNRT(char *buf, ssize_t size,int ttl) {
193  ssize_t nb_write;
194  bool sendItCompressed = false;
195 
196#ifdef COMPRESS_FRAMES
197  if (buf[0] == XML_HEADER) {
198    sendItCompressed = true;
199  }
200#endif // COMPRESS_FRAMES
201
202  if (sendItCompressed) {
203    char *out;
204    ssize_t out_size;
205    if (compressBuffer(buf, size, &out, &out_size, 9) == Z_OK) {
206      size = out_size;
207      nb_write = UDT::sendmsg(socket_fd, out, size,ttl, true);
208      free(out);
209    } else {
210      Warn("Compress error, sending it uncompressed\n");
211      sendItCompressed = false;
212    }
213  }
214
215  if (!sendItCompressed) {
216    nb_write = UDT::sendmsg(socket_fd, buf, size, ttl, true);
217  }
218
219   //Printf("write %i %i\n",nb_write,size);
220  if (nb_write < 0) {
221    Err("UDT::sendmsg error (%s)\n", UDT::getlasterror().getErrorMessage());
222    if (UDT::getlasterror().getErrorCode() == CUDTException::ECONNLOST ||
223        UDT::getlasterror().getErrorCode() == CUDTException::EINVSOCK) {
224      connection_lost = true;
225    }
226  } else if (nb_write != size) {
227    Err("%s, code %i (%ld/%ld)\n", UDT::getlasterror().getErrorMessage(),
228        UDT::getlasterror().getErrorCode(), nb_write, size);
229  }
230}
231
232ssize_t ui_com::Receive(char *buf, ssize_t buf_size) {
233  ssize_t bytesRead = UDT::recvmsg(socket_fd, buf, buf_size);
234
235  if (bytesRead < 0) {
236    if (UDT::getlasterror().getErrorCode() == CUDTException::ECONNLOST) {
237      Err("UDT::recvmsg error (%s)\n", UDT::getlasterror().getErrorMessage());
238      connection_lost = true;
239    }
240  }
241
242  return bytesRead;
243}
244
245void ui_com::CheckConnection(void) {
246  int32_t val,len;
247 
248  int result=UDT::getsockopt(socket_fd,0,UDT_STATE,&val,&len); 
249   if (result < 0) {
250    Printf("UDT::getsockopt error (%s)\n", UDT::getlasterror().getErrorMessage());
251   }
252 // printf("opt: %i %i %i\n",result,val,len);
253}
254
255//todo: le run (rt) push les données une par une
256//le nrt recupere et decide quand envoyer :
257// -quand toutes les données d'une meme periode sont recues
258// -quand on a atteind le nb buffering
259// -le nrt alloue dynamiquement le buffer d'envoi
260//-> enelver l'alloc du buffer au startup (UpdateDataToSendSize)
261void ui_com::Run(void) {
262  // check endianness
263  char header;
264  if (IsBigEndian()) {
265    header = DATA_BIG_ENDIAN;
266    Printf("System is big endian\n");
267  } else {
268    header = DATA_LITTLE_ENDIAN;
269    Printf("System is little endian\n");
270  }
271
272#ifdef __XENO__
273  WarnUponSwitches(true);
274  printf("\n"); // a revoir pourquoi??
275// sans ce printf, dans le simu_roll, le suspend ne fonctionne pas...
276#endif
277  // on attend d'avoir des choses à faire
278  Suspend();
279
280  while (!ToBeStopped()) {
281    size_t resume_id;
282    Time min = 0xffffffffffffffffULL;
283
284    // on recpuere l'id de la prochaine execution
285    send_mutex->GetMutex();
286    resume_id = resumeTimes.size();
287    for (size_t i = 0; i < resumeTimes.size(); i++) {
288      if (resumeTimes.at(i) < min && datasToSend.at(i)->IsEnabled() == true) {
289        min = resumeTimes.at(i);
290        resume_id = i;
291      }
292    }
293
294    // attente
295    if (resume_id < resumeTimes.size()) {
296      Time time = resumeTimes.at(resume_id);
297      uint16_t resume_period = datasToSend.at(resume_id)->SendPeriod();
298      send_mutex->ReleaseMutex();
299      // on dort jusqu'a la prochaine execution
300      SleepUntil(time);
301
302      // envoi des donnees
303      int offset = 3;
304      send_buffer[0] = header;
305      send_mutex->GetMutex();
306
307      for (size_t i = 0; i < datasToSend.size(); i++) {
308        if (datasToSend.at(i)->SendPeriod() == resume_period && datasToSend.at(i)->IsEnabled() == true) {
309          datasToSend.at(i)->CopyDatas(send_buffer + offset);
310          offset += datasToSend.at(i)->SendSize();
311        }
312      }
313      if (offset != 3) {
314        memcpy(&send_buffer[1], &resume_period, sizeof(uint16_t));
315        //printf("send %i %i %i %x %x\n",resume_period,offset,sizeof(uint16_t),send_buffer,&send_buffer[1]);
316        // for(int i=0;i<offset;i++) printf("%x ",send_buffer[i]);
317        // printf("\n");
318        Send(send_buffer, offset,resume_period);
319      }
320
321      //test to push datas
322      for (size_t i = 0; i < datasToSend.size(); i++) {
323        if (datasToSend.at(i)->SendPeriod() == resume_period && datasToSend.at(i)->IsEnabled() == true) {
324          PushDatasToSend(datasToSend.at(i));
325        }
326      }
327      //end test
328     
329      // on planifie la prochaine execution
330      for (size_t i = 0; i < datasToSend.size(); i++) {
331        if (datasToSend.at(i)->SendPeriod() == resume_period) {
332          resumeTimes.at(i) += datasToSend.at(i)->SendPeriod() * 1000000;
333        }
334      }
335      send_mutex->ReleaseMutex();
336      //Printf("%i %lld\n",resume_period,GetTime()/1000000);
337    } else {
338      send_mutex->ReleaseMutex();
339      // rien a faire, suspend
340      //Printf("rien a faire suspend\n");
341      Suspend();
342      //Printf("wake\n");
343      // on planifie la prochaine execution
344      Time time = GetTime();
345      send_mutex->GetMutex();
346      for (size_t i = 0; i < datasToSend.size(); i++) {
347        resumeTimes.at(i) =
348            time + (Time)datasToSend.at(i)->SendPeriod() * 1000000;
349      }
350      send_mutex->ReleaseMutex();
351    }
352  }
353}
354
355
356//called with send_mutex locked
357//in nrt
358//TODO: RT part has to put datas in pipe, then user thread call this
359void ui_com::PushDatasToSend(const SendData *dataToSend) {
360  PushedData_t* ptr=NULL;
361  //check if we already have one buffer for this couple period/nb_buffering
362  for (size_t i = 0; i < pushedDatas.size(); i++) {
363      if (pushedDatas.at(i).period==dataToSend->SendPeriod() && pushedDatas.at(i).nb_buffering==dataToSend->NbBuffering()) {
364          ptr=&pushedDatas.at(i);
365        }
366  }
367  if(ptr==NULL) {
368      Warn("no match in buffers\n");
369      return;
370  }
371 
372  dataToSend->CopyDatas(ptr->buf+ptr->actual_size);
373  ptr->actual_size+=dataToSend->SendSize();
374  Printf("nb buffered %i\n",ptr->actual_size);
375  Printf("pushed size %i period %i nb buffering %i\n",dataToSend->SendSize(),dataToSend->SendPeriod(),dataToSend->NbBuffering());
376  if(ptr->actual_size==ptr->final_size) {
377      Printf("ready to send\n");
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(GetTime() + (Time)obj->SendPeriod() * 1000000);
543
544  // on resume en meme temps tout ceux qui ont la meme periode
545  for (size_t i = 0; i < datasToSend.size(); i++) {
546    if (datasToSend.at(i)->SendPeriod() == obj->SendPeriod()) {
547      resumeTimes.at(resumeTimes.size() - 1) = resumeTimes.at(i);
548      break;
549    }
550  }
551
552  datasToSend.push_back(obj);
553
554  send_mutex->ReleaseMutex();
555}
556
557//TODO: check if it is RT compatible
558void ui_com::UpdateSendData(const SendData *obj) {
559  // le mutex est deja pris par l'appellant
560  size_t id, i;
561
562  // on recupere l'id
563  for (i = 0; i < datasToSend.size(); i++) {
564    if (datasToSend.at(i) == obj) {
565      id = i;
566      break;
567    }
568  }
569
570  // on resume en meme temps tout ceux qui ont la meme periode
571  for (i = 0; i < datasToSend.size(); i++) {
572    if (i == id)
573      continue;
574    if (datasToSend.at(i)->IsEnabled() == true && datasToSend.at(i)->SendPeriod() == obj->SendPeriod()) {
575      resumeTimes.at(id) = resumeTimes.at(i);
576      break;
577    }
578  }
579
580  // si aucun match, on planifie l'execution
581  if (i == datasToSend.size())
582    resumeTimes.at(id) = GetTime() + (Time)obj->SendPeriod() * 1000000;
583
584  if (IsSuspended() == true) Resume();
585
586//update all buffers if necessary, discard datas
587    std::vector<PushedData_t>::iterator pushedDatasIterator = pushedDatas.begin();
588    while(pushedDatasIterator != pushedDatas.end()) {
589        size_t bufSize=0;
590        for (size_t i = 0; i < datasToSend.size(); i++) {
591            if (datasToSend.at(i)->IsEnabled() && datasToSend.at(i)->SendPeriod() == (*pushedDatasIterator).period && datasToSend.at(i)->NbBuffering()==(*pushedDatasIterator).nb_buffering) {
592                bufSize+=datasToSend.at(i)->SendSize();
593            }
594        }
595        bufSize*=(*pushedDatasIterator).nb_buffering;
596        if(bufSize!=(*pushedDatasIterator).final_size && bufSize!=0) {
597            Printf("change buf size %i->%i\n",(*pushedDatasIterator).final_size,bufSize);
598            (*pushedDatasIterator).actual_size=0;
599            (*pushedDatasIterator).final_size=bufSize;
600            free((*pushedDatasIterator).buf);
601            (*pushedDatasIterator).buf=(char*)malloc(bufSize);
602        }
603        if(bufSize==0) {
604            Printf("delete buf\n");
605            free((*pushedDatasIterator).buf);
606            pushedDatasIterator = pushedDatas.erase(pushedDatasIterator);
607        } else {
608            ++pushedDatasIterator;
609        }
610    }
611 
612  //check if we need a new buffer for this couple period/nb_buffering
613  if(obj->IsEnabled()) {
614      bool match=false;
615      for (size_t i = 0; i < pushedDatas.size(); i++) {
616          if (pushedDatas.at(i).period==obj->SendPeriod() && pushedDatas.at(i).nb_buffering==obj->NbBuffering()) {
617              Printf("match item %i, period %i nb buff %i\n",i,obj->SendPeriod(),obj->NbBuffering());
618              match=true;
619            }
620      }
621      //create new one
622      if(!match) {
623        printf("no match\n");
624        PushedData_t tmp;
625        tmp.period=obj->SendPeriod();
626        tmp.nb_buffering=obj->NbBuffering();
627        tmp.actual_size=0;
628        tmp.final_size=0;
629        for (size_t i = 0; i < datasToSend.size(); i++) {
630            if (datasToSend.at(i)->IsEnabled() && datasToSend.at(i)->SendPeriod() == obj->SendPeriod() && datasToSend.at(i)->NbBuffering()==obj->NbBuffering()) {
631              tmp.final_size+=datasToSend.at(i)->SendSize();
632            }
633        }
634        tmp.final_size*=obj->NbBuffering();
635        printf("final size %i\n",tmp.final_size);
636        tmp.buf=(char*)malloc(tmp.final_size);
637        if(tmp.buf!=NULL) {
638            pushedDatas.push_back(tmp);
639        } else {
640            Warn("could not allocate buffer of size %i\n",tmp.final_size);
641        }
642      }
643  }
644  Printf("nb buf %i\n",pushedDatas.size());
645  return;
646}
647
648//allocate a buffer at startup for the worst case
649//(every data send with same pedriod)
650//TODO: dynamic size depending on what gcs asks?
651void ui_com::UpdateDataToSendSize(void) {
652  send_mutex->GetMutex();
653  send_size = 3; // id(1)+period(2)
654  for (size_t i = 0; i < datasToSend.size(); i++) {
655    if (datasToSend[i] != NULL)
656      send_size += datasToSend[i]->SendSize();
657  }
658
659  // send_buffer=(char*)realloc((void*)send_buffer,send_size*sizeof(char));
660  if (send_buffer != NULL)
661    free(send_buffer);
662  send_buffer = (char *)malloc(send_size * sizeof(char));
663  send_mutex->ReleaseMutex();
664}
665
666void ui_com::RemoveSendData(const SendData *obj) {
667  // printf("remove_data_to_send %i\n",data_to_send.size());
668
669  send_mutex->GetMutex();
670  // on recupere l'id
671  for (size_t i = 0; i < datasToSend.size(); i++) {
672    if (datasToSend.at(i) == obj) {
673      datasToSend.erase(datasToSend.begin() + i);
674      resumeTimes.erase(resumeTimes.begin() + i);
675      // printf("remove_data_to_send %i ok\n",data_to_send.size());
676      break;
677    }
678  }
679  send_mutex->ReleaseMutex();
680
681  return;
682}
683
684void ui_com::Block(void) { send_mutex->GetMutex(); }
685
686void ui_com::UnBlock(void) { send_mutex->ReleaseMutex(); }
Note: See TracBrowser for help on using the repository browser.