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

Last change on this file since 465 was 456, checked in by Sanahuja Guillaume, 3 years ago

refactor nb_buffering

File size: 20.3 KB
RevLine 
[2]1// %flair:license{
[15]2// This file is part of the Flair framework distributed under the
3// CECILL-C License, Version 1.0.
[2]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
[445]41ui_com::ui_com(const Object *parent, UDTSOCKET sock) : Thread(parent, "send", 2,16384*2) {
[15]42 // mutex
43 send_mutex = new Mutex(this, ObjectName());
[2]44
[15]45 socket_fd = sock;
46 connection_lost = false;
[2]47
48#ifdef __XENO__
[442]49 Err("multi buffering is not well implemented in RT\n");
50#endif
51
52#ifdef __XENO__
[15]53 int status;
54 string tmp_name;
[2]55
[15]56 is_running = true;
[2]57
[15]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());
[2]63#ifdef RT_PIPE_SIZE
[15]64 status = rt_pipe_create(&pipe, tmp_name.c_str(), P_MINOR_AUTO, RT_PIPE_SIZE);
[2]65#else
[15]66 status = rt_pipe_create(&pipe, tmp_name.c_str(), P_MINOR_AUTO, 0);
[2]67#endif
68
[15]69 if (status != 0) {
[133]70 char errorMsg[256];
71 Err("rt_pipe_create error (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
[15]72 // return -1;
73 }
[2]74
[15]75// start user side thread
[2]76#ifdef NRT_STACK_SIZE
[15]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 }
[2]82
[15]83 if (pthread_attr_setstacksize(&attr, NRT_STACK_SIZE) != 0) {
84 Err("pthread_attr_setstacksize error\n");
85 }
[2]86
[15]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 }
[2]95#ifdef NRT_STACK_SIZE
[15]96 if (pthread_attr_destroy(&attr) != 0) {
97 Err("pthread_attr_destroy error\n");
98 }
[2]99#endif
100
[15]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");
[247]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*/
[15]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");
[2]113
[15]114 if (UDT::setsockopt(socket_fd, 0, UDT_RCVSYN, &blocking, sizeof(bool)) != 0)
115 Err("UDT::setsockopt error (UDT_RCVSYN)\n");
116 //#endif //__XENO__
[2]117
[15]118 Start();
[2]119}
120
[15]121ui_com::~ui_com() {
[243]122//Printf("destruction ui_com\n");
[2]123
124#ifdef __XENO__
[15]125 is_running = false;
[2]126
[15]127 pthread_join(thread, NULL);
[2]128
[15]129 int status = rt_pipe_delete(&pipe);
[133]130 if (status != 0) {
131 char errorMsg[256];
132 Err("rt_pipe_delete error (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
133 }
[2]134#endif
135
[15]136 SafeStop();
[2]137
[15]138 if (IsSuspended() == true)
139 Resume();
[2]140
[15]141 Join();
[247]142
143 char buf=CLOSING_CONNECTION;
144 Send(&buf,1);
145
[445]146 for (size_t i = 0; i < pushedDatas.size(); i++) {
147 free(pushedDatas.at(i).buf);
148 }
[243]149 //Printf("destruction ui_com ok\n");
[2]150}
151
[439]152//send, public part; dispatch to SendNRT (nrt) or rt_pipe (rt)
153//rt_pipe is read by user thread which calls SendNRT
[252]154void ui_com::Send(char *buf, ssize_t size,int ttl) {
[15]155 if (connection_lost == true)
156 return;
[2]157
[15]158 char *tosend = buf;
159 if (buf[0] == XML_HEADER) {
160 // cut xml header
161 tosend = strstr(buf, "<root");
162 size -= tosend - buf;
163 }
[2]164
165#ifdef __XENO__
[439]166 //data buf
167 ssize_t nb_write = rt_pipe_write(&pipe, tosend, size, P_NORMAL);
[15]168 if (nb_write < 0) {
[133]169 char errorMsg[256];
170 Err("rt_pipe_write error (%s)\n", strerror_r(-nb_write, errorMsg, sizeof(errorMsg)));
[15]171 } else if (nb_write != size) {
172 Err("rt_pipe_write error %i/%i\n", nb_write, size);
173 }
[439]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 }
[2]182#else //__XENO__
[439]183 SendNRT(tosend,size,ttl);
184#endif //__XENO__
185}
[2]186
[439]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
[2]193#ifdef COMPRESS_FRAMES
[15]194 if (buf[0] == XML_HEADER) {
195 sendItCompressed = true;
196 }
[439]197#endif // COMPRESS_FRAMES
[2]198
[15]199 if (sendItCompressed) {
200 char *out;
201 ssize_t out_size;
[439]202 if (compressBuffer(buf, size, &out, &out_size, 9) == Z_OK) {
[15]203 size = out_size;
[252]204 nb_write = UDT::sendmsg(socket_fd, out, size,ttl, true);
[15]205 free(out);
206 } else {
207 Warn("Compress error, sending it uncompressed\n");
208 sendItCompressed = false;
[2]209 }
[15]210 }
[2]211
[15]212 if (!sendItCompressed) {
[439]213 nb_write = UDT::sendmsg(socket_fd, buf, size, ttl, true);
[15]214 }
[2]215
[247]216 //Printf("write %i %i\n",nb_write,size);
[15]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;
[2]222 }
[15]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 }
[2]227}
228
[15]229ssize_t ui_com::Receive(char *buf, ssize_t buf_size) {
230 ssize_t bytesRead = UDT::recvmsg(socket_fd, buf, buf_size);
[2]231
[15]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;
[2]236 }
[15]237 }
[2]238
[15]239 return bytesRead;
[2]240}
241
[234]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
[439]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
[456]255// -quand on a atteint le nb buffering
[439]256// -le nrt alloue dynamiquement le buffer d'envoi
[15]257void ui_com::Run(void) {
[2]258#ifdef __XENO__
[15]259 WarnUponSwitches(true);
260 printf("\n"); // a revoir pourquoi??
261// sans ce printf, dans le simu_roll, le suspend ne fonctionne pas...
[2]262#endif
[15]263 // on attend d'avoir des choses à faire
264 Suspend();
[2]265
[15]266 while (!ToBeStopped()) {
267 size_t resume_id;
268 Time min = 0xffffffffffffffffULL;
[2]269
[15]270 // on recpuere l'id de la prochaine execution
271 send_mutex->GetMutex();
[441]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);
[15]276 resume_id = i;
277 }
278 }
[2]279
[15]280 // attente
[441]281 if (resume_id < resumeTimes.size()) {
282 Time time = resumeTimes.at(resume_id);
283 uint16_t resume_period = datasToSend.at(resume_id)->SendPeriod();
[15]284 send_mutex->ReleaseMutex();
285 // on dort jusqu'a la prochaine execution
286 SleepUntil(time);
[2]287
[15]288 // envoi des donnees
289 send_mutex->GetMutex();
[2]290
[456]291 //multi buffering
[441]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));
[440]295 }
296 }
297
[15]298 // on planifie la prochaine execution
[441]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;
[2]302 }
[15]303 }
304 send_mutex->ReleaseMutex();
[157]305 //Printf("%i %lld\n",resume_period,GetTime()/1000000);
[15]306 } else {
307 send_mutex->ReleaseMutex();
308 // rien a faire, suspend
[157]309 //Printf("rien a faire suspend\n");
[15]310 Suspend();
[157]311 //Printf("wake\n");
[15]312 // on planifie la prochaine execution
313 Time time = GetTime();
314 send_mutex->GetMutex();
[441]315 for (size_t i = 0; i < datasToSend.size(); i++) {
316 resumeTimes.at(i) =
317 time + (Time)datasToSend.at(i)->SendPeriod() * 1000000;
[15]318 }
319 send_mutex->ReleaseMutex();
[2]320 }
[15]321 }
[2]322}
[440]323
324
325//called with send_mutex locked
326//in nrt
[441]327//TODO: RT part has to put datas in pipe, then user thread call this
328void ui_com::PushDatasToSend(const SendData *dataToSend) {
[456]329 PushedData_t* ptr=GetCorrespondingPushedData(dataToSend);
330
[440]331 if(ptr==NULL) {
[441]332 Warn("no match in buffers\n");
333 return;
[440]334 }
335
[456]336 if(ptr->buf==NULL) {
337 Warn("buffer not allocated\n");
338 return;
339 }
340
[442]341 if(ptr->actual_size==0) {
[445]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);
[442]353 } else {
[445]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);
[442]362 }
363 }
364
[453]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
[441]370 dataToSend->CopyDatas(ptr->buf+ptr->actual_size);
371 ptr->actual_size+=dataToSend->SendSize();
[453]372
[444]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());
[442]375 if(ptr->actual_size>=ptr->final_size) {//par securité on test aussi le cas ou supérieur
[444]376 //Printf("ready to send\n");
[443]377 SendNRT(ptr->buf, ptr->actual_size,ptr->period);
[441]378 //clean
379 ptr->actual_size=0;
380 }
[440]381}
382
383
[2]384#ifdef __XENO__
[15]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;
[439]393 ssize_t nb_read;
394 int ttl;
[2]395
[15]396 buf = (char *)malloc(NRT_PIPE_SIZE);
397 if (buf == NULL) {
398 caller->Err("malloc error\n");
399 }
[2]400
[15]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);
[133]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 }
[15]409 usleep(1000);
410 }
[2]411
[15]412 while (1) {
413 FD_ZERO(&set); // clear the set
414 FD_SET(pipe_fd, &set); // add our file descriptor to the set
[2]415
[15]416 timeout.tv_sec = 0;
417 timeout.tv_usec = SELECT_TIMEOUT_MS * 1000;
[2]418
[15]419 rv = select(FD_SETSIZE, &set, NULL, NULL, &timeout);
[2]420
[15]421 if (rv == -1) {
[439]422 if (caller->is_running == false && UDT::getlasterror().getErrorCode() == CUDTException::ETIMEOUT)
[15]423 break; // timeout
424 if (UDT::getlasterror().getErrorCode() != CUDTException::ETIMEOUT)
425 caller->Err("epoll_wait, %s, code %i\n",
[439]426 UDT::getlasterror().getErrorMessage(), UDT::getlasterror().getErrorCode());
[15]427 } else if (rv == 0) {
428 // printf("timeout\n"); // a timeout occured
429 if (caller->is_running == false)
430 break;
[2]431
[15]432 } else {
[439]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)
[15]436 nb_read = read(pipe_fd, buf, NRT_PIPE_SIZE);
437 buf[nb_read] = 0;
[439]438 read(pipe_fd, &ttl, sizeof(ttl));
[15]439// printf("envoi\n%s\n",buf);
[2]440
[439]441 caller->SendNRT(buf, nb_read,ttl);
[2]442 }
[15]443 }
[2]444
[15]445 close(pipe_fd);
446 if (buf != NULL)
447 free(buf);
448 pthread_exit(0);
[2]449}
450#endif
451
[15]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;
[2]457
[15]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;
[2]465
[15]466 *out = (char *)malloc(COMPRESS_CHUNK);
467 if (!(*out))
468 return Z_BUF_ERROR;
[2]469
[15]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;
[2]475
[15]476 ret = deflate(&strm, flush); /* no bad return value */
477 if (ret == Z_STREAM_ERROR) {
478 free(*out);
479 return ret;
480 }
[2]481
[15]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);
[2]487
[15]488 if (strm.avail_out != 0) {
489 return Z_OK;
490 } else {
491 return Z_STREAM_ERROR;
492 }
[2]493}
494
[15]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;
[2]500
[15]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;
[2]510
[15]511 *out = (unsigned char *)malloc(COMPRESS_CHUNK);
512 if (!(*out))
513 return Z_BUF_ERROR;
[2]514
[15]515 strm.avail_in = in_size;
516 strm.next_in = in;
517 strm.avail_out = COMPRESS_CHUNK;
518 strm.next_out = *out;
[2]519
[15]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:
[2]527 (void)inflateEnd(&strm);
[15]528 return ret;
529 }
530 // have = COMPRESS_CHUNK - strm.avail_out;
[2]531
[15]532 /* clean up and return */
533 (void)inflateEnd(&strm);
534 return ret == Z_STREAM_END ? Z_OK : Z_DATA_ERROR;
[2]535}
536
[15]537bool ui_com::ConnectionLost() { return connection_lost; }
[2]538
[15]539void ui_com::AddSendData(const SendData *obj) {
[456]540 send_mutex->GetMutex();
[2]541
[456]542 resumeTimes.push_back(0);//time will be updated in UpdateResumeTime
543 datasToSend.push_back(obj);
544 UpdateResumeTime(obj);
545
546 send_mutex->ReleaseMutex();
[2]547}
548
[442]549//must be called with mutex locked
[456]550void ui_com::UpdateResumeTime(const SendData *obj) {
551 size_t id, i;
[2]552
[15]553 // on recupere l'id
[441]554 for (i = 0; i < datasToSend.size(); i++) {
555 if (datasToSend.at(i) == obj) {
[15]556 id = i;
557 break;
[2]558 }
[15]559 }
[2]560
[456]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
[441]563 for (i = 0; i < datasToSend.size(); i++) {
[456]564 if (i == id) continue;
[441]565 if (datasToSend.at(i)->IsEnabled() == true && datasToSend.at(i)->SendPeriod() == obj->SendPeriod()) {
566 resumeTimes.at(id) = resumeTimes.at(i);
[15]567 break;
[2]568 }
[15]569 }
[2]570
[15]571 // si aucun match, on planifie l'execution
[441]572 if (i == datasToSend.size())
573 resumeTimes.at(id) = GetTime() + (Time)obj->SendPeriod() * 1000000;
[456]574
575}
[2]576
[456]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);
[2]598
[456]599 if (IsSuspended() == true) Resume();
600
601 //update all buffers if necessary, discard datas
[441]602 std::vector<PushedData_t>::iterator pushedDatasIterator = pushedDatas.begin();
603 while(pushedDatasIterator != pushedDatas.end()) {
[456]604 size_t bufSize=CalcBufferSize(&(*pushedDatasIterator));
[441]605 if(bufSize!=(*pushedDatasIterator).final_size && bufSize!=0) {
[444]606 //Printf("change buf size %i->%i\n",(*pushedDatasIterator).final_size,bufSize);
[441]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) {
[444]613 //Printf("delete buf\n");
[441]614 free((*pushedDatasIterator).buf);
615 pushedDatasIterator = pushedDatas.erase(pushedDatasIterator);
616 } else {
617 ++pushedDatasIterator;
618 }
619 }
620
[456]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());
[441]649 }
650 }
[456]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) {
[441]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 }
[456]669 }
[2]670}
671
[456]672
673//must be called with mutex locked
674size_t ui_com::CalcBufferSize(const PushedData_t* ptr) const {
675 size_t final_size=0;
[452]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());
[456]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());
[452]682 }
683 }
[456]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
[442]690
[456]691 return final_size;
[442]692}
693
[15]694void ui_com::RemoveSendData(const SendData *obj) {
695 // printf("remove_data_to_send %i\n",data_to_send.size());
[2]696
[15]697 send_mutex->GetMutex();
698 // on recupere l'id
[441]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);
[15]703 // printf("remove_data_to_send %i ok\n",data_to_send.size());
704 break;
[2]705 }
[15]706 }
707 send_mutex->ReleaseMutex();
[2]708
[15]709 return;
[2]710}
711
[15]712void ui_com::Block(void) { send_mutex->GetMutex(); }
[2]713
[15]714void ui_com::UnBlock(void) { send_mutex->ReleaseMutex(); }
Note: See TracBrowser for help on using the repository browser.