streamers / net_helper-ml.c @ c3de57f0
History | View | Annotate | Download (17.2 KB)
1 |
/*
|
---|---|
2 |
* Copyright (c) 2009 Marco Biazzini
|
3 |
*
|
4 |
* This is free software; see lgpl-2.1.txt
|
5 |
*/
|
6 |
|
7 |
#include <event2/event.h> |
8 |
#include <arpa/inet.h> |
9 |
#include <unistd.h> |
10 |
#include <stdlib.h> |
11 |
#include <stdio.h> |
12 |
#include <stdbool.h> |
13 |
#include <string.h> |
14 |
#include <assert.h> |
15 |
|
16 |
|
17 |
#include "net_helper.h" |
18 |
#include "ml.h" |
19 |
#include "config.h" |
20 |
|
21 |
#include "grapes_msg_types.h" |
22 |
|
23 |
|
24 |
#ifdef MONL
|
25 |
#include "mon.h" |
26 |
#include "grapes_log.h" |
27 |
#include "repoclient.h" |
28 |
#include "grapes.h" |
29 |
#endif
|
30 |
|
31 |
|
32 |
/**
|
33 |
* libevent pointer
|
34 |
*/
|
35 |
struct event_base *base;
|
36 |
|
37 |
#define NH_BUFFER_SIZE 1000 |
38 |
#define NH_LOOKUP_SIZE 1000 |
39 |
#define NH_PACKET_TIMEOUT {0, 500*1000} |
40 |
#define NH_ML_INIT_TIMEOUT {1, 0} |
41 |
|
42 |
#define FDSSIZE 16 |
43 |
|
44 |
static int sIdx = 0; |
45 |
static int rIdxML = 0; //reveive from ML to this buffer position |
46 |
static int rIdxUp = 0; //hand up to layer above at this buffer position |
47 |
|
48 |
typedef struct nodeID { |
49 |
socketID_handle addr; |
50 |
int connID; // connection associated to this node, -1 if myself |
51 |
int refcnt;
|
52 |
#ifdef MONL
|
53 |
//n quick and dirty static vector for measures TODO: make it dinamic
|
54 |
MonHandler mhs[20];
|
55 |
int n_mhs;
|
56 |
#endif
|
57 |
// int addrSize;
|
58 |
// int addrStringSize;
|
59 |
} nodeID; |
60 |
|
61 |
typedef struct msgData_cb { |
62 |
int bIdx; // index of the message in the proper buffer |
63 |
unsigned char msgType; // message type |
64 |
int mSize; // message size |
65 |
bool conn_cb_called;
|
66 |
bool cancelled;
|
67 |
} msgData_cb; |
68 |
|
69 |
static struct nodeID **lookup_array; |
70 |
static int lookup_max = NH_LOOKUP_SIZE; |
71 |
static int lookup_curr = 0; |
72 |
|
73 |
static nodeID *me; //TODO: is it possible to get rid of this (notwithstanding ml callback)?? |
74 |
static int timeoutFired = 0; |
75 |
static bool fdTriggered = false; |
76 |
|
77 |
// pointers to the msgs to be send
|
78 |
static uint8_t *sendingBuffer[NH_BUFFER_SIZE];
|
79 |
// pointers to the received msgs + sender nodeID
|
80 |
struct receivedB {
|
81 |
struct nodeID *id;
|
82 |
int len;
|
83 |
uint8_t *data; |
84 |
}; |
85 |
static struct receivedB receivedBuffer[NH_BUFFER_SIZE]; |
86 |
/**/ static int recv_counter =0; static int snd_counter =0; |
87 |
|
88 |
|
89 |
static void connReady_cb (int connectionID, void *arg); |
90 |
static struct nodeID *new_node(socketID_handle peer) { |
91 |
send_params params = {0,0,0,0}; |
92 |
struct nodeID *res = malloc(sizeof(struct nodeID)); |
93 |
if (!res) {
|
94 |
fprintf(stderr, "Net-helper : memory error\n");
|
95 |
return NULL; |
96 |
} |
97 |
memset(res, 0, sizeof(struct nodeID)); |
98 |
|
99 |
res->addr = malloc(SOCKETID_SIZE); |
100 |
if (! res->addr) {
|
101 |
free (res); |
102 |
fprintf(stderr, "Net-helper : memory error while creating a new nodeID \n");
|
103 |
return NULL; |
104 |
} |
105 |
memset(res->addr, 0, SOCKETID_SIZE);
|
106 |
memcpy(res->addr, peer ,SOCKETID_SIZE); |
107 |
|
108 |
res->refcnt = 1;
|
109 |
|
110 |
res->connID = mlOpenConnection(peer, &connReady_cb, NULL, params);
|
111 |
|
112 |
return res;
|
113 |
} |
114 |
|
115 |
|
116 |
static struct nodeID **id_lookup(socketID_handle target) { |
117 |
|
118 |
int i,here=-1; |
119 |
for (i=0;i<lookup_curr;i++) { |
120 |
if (lookup_array[i] == NULL) { |
121 |
if (here < 0) { |
122 |
here = i; |
123 |
} |
124 |
} else if (!mlCompareSocketIDs(lookup_array[i]->addr,target)) { |
125 |
return &lookup_array[i];
|
126 |
} |
127 |
} |
128 |
|
129 |
if (here == -1) { |
130 |
here = lookup_curr++; |
131 |
} |
132 |
|
133 |
if (lookup_curr > lookup_max) {
|
134 |
lookup_max *= 2;
|
135 |
lookup_array = realloc(lookup_array,lookup_max*sizeof(struct nodeID*)); |
136 |
} |
137 |
|
138 |
lookup_array[here] = new_node(target); |
139 |
return &lookup_array[here];
|
140 |
} |
141 |
|
142 |
static struct nodeID *id_lookup_dup(socketID_handle target) { |
143 |
return nodeid_dup(*id_lookup(target));
|
144 |
} |
145 |
|
146 |
|
147 |
/**
|
148 |
* Look for a free slot in the received buffer and allocates it for immediate use
|
149 |
* @return the index of a free slot in the received msgs buffer, -1 if no free slot available.
|
150 |
*/
|
151 |
static int next_R() { |
152 |
if (receivedBuffer[rIdxML].data==NULL) { |
153 |
int ret = rIdxML;
|
154 |
rIdxML = (rIdxML+1)%NH_BUFFER_SIZE;
|
155 |
return ret;
|
156 |
} else {
|
157 |
//TODO: handle receive overload situation!
|
158 |
return -1; |
159 |
} |
160 |
} |
161 |
|
162 |
/**
|
163 |
* Look for a free slot in the sending buffer and allocates it for immediate use
|
164 |
* @return the index of a free slot in the sending msgs buffer, -1 if no free slot available.
|
165 |
*/
|
166 |
static int next_S() { |
167 |
if (sendingBuffer[sIdx]) {
|
168 |
int count;
|
169 |
for (count=0;count<NH_BUFFER_SIZE;count++) { |
170 |
sIdx = (sIdx+1)%NH_BUFFER_SIZE;
|
171 |
if (sendingBuffer[sIdx]==NULL) |
172 |
break;
|
173 |
} |
174 |
if (count==NH_BUFFER_SIZE) {
|
175 |
return -1; |
176 |
} |
177 |
} |
178 |
return sIdx;
|
179 |
} |
180 |
|
181 |
|
182 |
/**
|
183 |
* Callback used by ml to confirm its initialization. Create a valid self nodeID and register to receive data from remote peers.
|
184 |
* @param local_socketID
|
185 |
* @param errorstatus
|
186 |
*/
|
187 |
static void init_myNodeID_cb (socketID_handle local_socketID,int errorstatus) { |
188 |
switch (errorstatus) {
|
189 |
case 0: |
190 |
me->addr = malloc(SOCKETID_SIZE); |
191 |
if (! me->addr) {
|
192 |
fprintf(stderr, "Net-helper : memory error while creating my new nodeID \n");
|
193 |
return;
|
194 |
} |
195 |
|
196 |
memcpy(me->addr,local_socketID,SOCKETID_SIZE); |
197 |
// me->addrSize = SOCKETID_SIZE;
|
198 |
// me->addrStringSize = SOCKETID_STRING_SIZE;
|
199 |
me->connID = -1;
|
200 |
me->refcnt = 1;
|
201 |
// fprintf(stderr,"Net-helper init : received my own socket: %s.\n",node_addr(me));
|
202 |
break;
|
203 |
case -1: |
204 |
//
|
205 |
fprintf(stderr,"Net-helper init : socket error occurred in ml while creating socket\n");
|
206 |
exit(1);
|
207 |
break;
|
208 |
case 1: |
209 |
//
|
210 |
fprintf(stderr,"Net-helper init : NAT traversal failed while creating socket\n");
|
211 |
exit(1);
|
212 |
break;
|
213 |
case 2: |
214 |
fprintf(stderr,"Net-helper init : NAT traversal timeout while creating socket\n");
|
215 |
fprintf(stderr,"Net-helper init : Retrying without STUN\n");
|
216 |
mlSetStunServer(0,NULL); |
217 |
break;
|
218 |
default : // should never happen |
219 |
//
|
220 |
fprintf(stderr,"Net-helper init : Unknown error in ml while creating socket\n");
|
221 |
} |
222 |
|
223 |
} |
224 |
|
225 |
/**
|
226 |
* Timeout callback to be set in the eventlib loop as needed
|
227 |
* @param socket
|
228 |
* @param flag
|
229 |
* @param arg
|
230 |
*/
|
231 |
static void t_out_cb (int socket, short flag, void* arg) { |
232 |
|
233 |
timeoutFired = 1;
|
234 |
// fprintf(stderr,"TIMEOUT!!!\n");
|
235 |
// event_base_loopbreak(base);
|
236 |
} |
237 |
|
238 |
/**
|
239 |
* File descriptor readable callback to be set in the eventlib loop as needed
|
240 |
*/
|
241 |
static void fd_cb (int fd, short flag, void* arg) |
242 |
{ |
243 |
//fprintf(stderr, "\twait4data: fd %d triggered\n", fd);
|
244 |
fdTriggered = true;
|
245 |
*((bool*)arg) = true; |
246 |
} |
247 |
|
248 |
/**
|
249 |
* Callback called by ml when a remote node ask for a connection
|
250 |
* @param connectionID
|
251 |
* @param arg
|
252 |
*/
|
253 |
static void receive_conn_cb(int connectionID, void *arg) { |
254 |
// fprintf(stderr, "Net-helper : remote peer opened the connection %d with arg = %d\n", connectionID,(int)arg);
|
255 |
|
256 |
} |
257 |
|
258 |
void free_sending_buffer(int i) |
259 |
{ |
260 |
free(sendingBuffer[i]); |
261 |
sendingBuffer[i] = NULL;
|
262 |
} |
263 |
|
264 |
/**
|
265 |
* Callback called by the ml when a connection is ready to be used to send data to a remote peer
|
266 |
* @param connectionID
|
267 |
* @param arg
|
268 |
*/
|
269 |
static void connReady_cb (int connectionID, void *arg) { |
270 |
char mt;
|
271 |
msgData_cb *p; |
272 |
p = (msgData_cb *)arg; |
273 |
if (p == NULL) return; |
274 |
if (p->cancelled) {
|
275 |
free(p); |
276 |
return;
|
277 |
} |
278 |
mlSendData(connectionID,(char *)(sendingBuffer[p->bIdx]),p->mSize,p->msgType,NULL); |
279 |
/**/ mt = ((char*)sendingBuffer[p->bIdx])[0]; ++snd_counter; |
280 |
if (mt!=MSG_TYPE_TOPOLOGY &&
|
281 |
mt!=MSG_TYPE_CHUNK && mt!=MSG_TYPE_SIGNALLING) { |
282 |
fprintf(stderr,"Net-helper ERROR! Sent message # %d of type %c and size %d\n",
|
283 |
snd_counter,mt+'0', p->mSize);}
|
284 |
free_sending_buffer(p->bIdx); |
285 |
// fprintf(stderr,"Net-helper: Message # %d for connection %d sent!\n ", p->bIdx,connectionID);
|
286 |
// event_base_loopbreak(base);
|
287 |
p->conn_cb_called = true;
|
288 |
} |
289 |
|
290 |
/**
|
291 |
* Callback called by ml when a connection error occurs
|
292 |
* @param connectionID
|
293 |
* @param arg
|
294 |
*/
|
295 |
static void connError_cb (int connectionID, void *arg) { |
296 |
// simply get rid of the msg in the buffer....
|
297 |
msgData_cb *p; |
298 |
p = (msgData_cb *)arg; |
299 |
if (p != NULL) { |
300 |
fprintf(stderr,"Net-helper: Connection %d could not be established to send msg %d.\n ", connectionID,p->bIdx);
|
301 |
if (p->cancelled) {
|
302 |
free(p);//p->mSize = -1;
|
303 |
} else {
|
304 |
p->conn_cb_called = true;
|
305 |
} |
306 |
} |
307 |
// event_base_loopbreak(base);
|
308 |
} |
309 |
|
310 |
|
311 |
/**
|
312 |
* Callback to receive data from ml
|
313 |
* @param buffer
|
314 |
* @param buflen
|
315 |
* @param msgtype
|
316 |
* @param arg
|
317 |
*/
|
318 |
static void recv_data_cb(char *buffer, int buflen, unsigned char msgtype, recv_params *arg) { |
319 |
// TODO: lacks a void* arg... moreover: recv_params has a msgtype, but there is also a msgtype explicit argument...
|
320 |
char str[SOCKETID_STRING_SIZE];
|
321 |
// fprintf(stderr, "Net-helper : called back with some news...\n");
|
322 |
/**/ ++recv_counter;
|
323 |
if (arg->remote_socketID != NULL) |
324 |
mlSocketIDToString(arg->remote_socketID,str,SOCKETID_STRING_SIZE); |
325 |
else
|
326 |
sprintf(str,"!Unknown!");
|
327 |
if (arg->nrMissingBytes || !arg->firstPacketArrived) {
|
328 |
fprintf(stderr, "Net-helper : corrupted message arrived from %s\n",str);
|
329 |
/**/ fprintf(stderr, "\tMessage # %d -- Message type: %hhd -- Missing # %d bytes%s\n", |
330 |
recv_counter, buffer[0],arg->nrMissingBytes, arg->firstPacketArrived?"":", Missing first!"); |
331 |
} |
332 |
else {
|
333 |
// fprintf(stderr, "Net-helper : message arrived from %s\n",str);
|
334 |
// buffering the received message only if possible, otherwise ignore it...
|
335 |
int index = next_R();
|
336 |
if (index<0) { |
337 |
fprintf(stderr,"Net-helper: receive buffer full\n ");
|
338 |
return;
|
339 |
} else {
|
340 |
receivedBuffer[index].data = malloc(buflen); |
341 |
if (receivedBuffer[index].data == NULL) { |
342 |
fprintf(stderr,"Net-helper: memory full, can't receive!\n ");
|
343 |
return;
|
344 |
} |
345 |
receivedBuffer[index].len = buflen; |
346 |
memcpy(receivedBuffer[index].data,buffer,buflen); |
347 |
// save the socketID of the sender
|
348 |
receivedBuffer[index].id = id_lookup_dup(arg->remote_socketID); |
349 |
} |
350 |
} |
351 |
// event_base_loopbreak(base);
|
352 |
} |
353 |
|
354 |
|
355 |
struct nodeID *net_helper_init(const char *IPaddr, int port, const char *config) { |
356 |
|
357 |
struct timeval tout = NH_ML_INIT_TIMEOUT;
|
358 |
int s, i;
|
359 |
struct tag *cfg_tags;
|
360 |
const char *res; |
361 |
const char *stun_server = "stun.ekiga.net"; |
362 |
int stun_port = 3478; |
363 |
const char *repo_address = "79.120.193.115:9832"; |
364 |
int publish_interval = 60; |
365 |
|
366 |
base = event_base_new(); |
367 |
lookup_array = calloc(lookup_max,sizeof(struct nodeID *)); |
368 |
|
369 |
cfg_tags = config_parse(config); |
370 |
|
371 |
res = config_value_str(cfg_tags, "stun_server");
|
372 |
if (res) {
|
373 |
stun_server = res; |
374 |
} |
375 |
config_value_int(cfg_tags, "stun_port", &stun_port);
|
376 |
|
377 |
res = config_value_str(cfg_tags, "repo_address");
|
378 |
if (res) {
|
379 |
repo_address = res; |
380 |
} |
381 |
|
382 |
config_value_int(cfg_tags, "publish_interval", &publish_interval);
|
383 |
|
384 |
me = malloc(sizeof(nodeID));
|
385 |
if (me == NULL) { |
386 |
return NULL; |
387 |
} |
388 |
memset(me,0,sizeof(nodeID)); |
389 |
me->connID = -10; // dirty trick to spot later if the ml has called back ... |
390 |
me->refcnt = 1;
|
391 |
|
392 |
for (i=0;i<NH_BUFFER_SIZE;i++) { |
393 |
sendingBuffer[i] = NULL;
|
394 |
receivedBuffer[i].data = NULL;
|
395 |
} |
396 |
|
397 |
mlRegisterErrorConnectionCb(&connError_cb); |
398 |
mlRegisterRecvConnectionCb(&receive_conn_cb); |
399 |
s = mlInit(1, tout, port, IPaddr, stun_port, stun_server, &init_myNodeID_cb, base);
|
400 |
if (s < 0) { |
401 |
fprintf(stderr, "Net-helper : error initializing ML!\n");
|
402 |
free(me); |
403 |
return NULL; |
404 |
} |
405 |
|
406 |
#ifdef MONL
|
407 |
{ |
408 |
void *repoclient;
|
409 |
eventbase = base; |
410 |
|
411 |
// Initialize logging
|
412 |
grapesInitLog(DCLOG_WARNING, NULL, NULL); |
413 |
|
414 |
repInit("");
|
415 |
repoclient = repOpen(repo_address, publish_interval); //repository.napa-wine.eu
|
416 |
if (repoclient == NULL) fatal("Unable to initialize repoclient"); |
417 |
monInit(base, repoclient); |
418 |
} |
419 |
#endif
|
420 |
free(cfg_tags); |
421 |
|
422 |
while (me->connID<-1) { |
423 |
// event_base_once(base,-1, EV_TIMEOUT, &t_out_cb, NULL, &tout);
|
424 |
event_base_loop(base,EVLOOP_ONCE); |
425 |
} |
426 |
timeoutFired = 0;
|
427 |
// fprintf(stderr,"Net-helper init : back from init!\n");
|
428 |
|
429 |
return me;
|
430 |
} |
431 |
|
432 |
|
433 |
void bind_msg_type (unsigned char msgtype) { |
434 |
|
435 |
mlRegisterRecvDataCb(&recv_data_cb,msgtype); |
436 |
} |
437 |
|
438 |
|
439 |
void send_to_peer_cb(int fd, short event, void *arg) |
440 |
{ |
441 |
msgData_cb *p = (msgData_cb *) arg; |
442 |
if (p->conn_cb_called) {
|
443 |
free(p); |
444 |
} |
445 |
else { //don't send it anymore |
446 |
free_sending_buffer(p->bIdx); |
447 |
p->cancelled = true;
|
448 |
// don't free p, the other timeout will do it
|
449 |
} |
450 |
} |
451 |
|
452 |
/**
|
453 |
* Called by the application to send data to a remote peer
|
454 |
* @param from
|
455 |
* @param to
|
456 |
* @param buffer_ptr
|
457 |
* @param buffer_size
|
458 |
* @return The dimension of the buffer or -1 if a connection error occurred.
|
459 |
*/
|
460 |
int send_to_peer(const struct nodeID *from, struct nodeID *to, const uint8_t *buffer_ptr, int buffer_size) |
461 |
{ |
462 |
msgData_cb *p; |
463 |
int current;
|
464 |
send_params params = {0,0,0,0}; |
465 |
|
466 |
if (buffer_size <= 0) { |
467 |
fprintf(stderr,"Net-helper: message size problematic: %d\n", buffer_size);
|
468 |
return buffer_size;
|
469 |
} |
470 |
|
471 |
// if buffer is full, discard the message and return an error flag
|
472 |
int index = next_S();
|
473 |
if (index<0) { |
474 |
// free(buffer_ptr);
|
475 |
fprintf(stderr,"Net-helper: send buffer full\n ");
|
476 |
return -1; |
477 |
} |
478 |
sendingBuffer[index] = malloc(buffer_size); |
479 |
if (! sendingBuffer[index]){
|
480 |
fprintf(stderr,"Net-helper: memory full, can't send!\n ");
|
481 |
return -1; |
482 |
} |
483 |
memset(sendingBuffer[index],0,buffer_size);
|
484 |
memcpy(sendingBuffer[index],buffer_ptr,buffer_size); |
485 |
// free(buffer_ptr);
|
486 |
p = malloc(sizeof(msgData_cb));
|
487 |
p->bIdx = index; p->mSize = buffer_size; p->msgType = (unsigned char)buffer_ptr[0]; p->conn_cb_called = false; p->cancelled = false; |
488 |
current = p->bIdx; |
489 |
|
490 |
to->connID = mlOpenConnection(to->addr,&connReady_cb,p, params); |
491 |
if (to->connID<0) { |
492 |
free_sending_buffer(current); |
493 |
fprintf(stderr,"Net-helper: Couldn't get a connection ID to send msg %d.\n ", p->bIdx);
|
494 |
free(p); |
495 |
return -1; |
496 |
} |
497 |
else {
|
498 |
struct timeval timeout = NH_PACKET_TIMEOUT;
|
499 |
event_base_once(base, -1, EV_TIMEOUT, send_to_peer_cb, (void *) p, &timeout); |
500 |
return buffer_size; //p->mSize; |
501 |
} |
502 |
|
503 |
} |
504 |
|
505 |
|
506 |
/**
|
507 |
* Called by an application to receive data from remote peers
|
508 |
* @param local
|
509 |
* @param remote
|
510 |
* @param buffer_ptr
|
511 |
* @param buffer_size
|
512 |
* @return The number of received bytes or -1 if some error occurred.
|
513 |
*/
|
514 |
int recv_from_peer(const struct nodeID *local, struct nodeID **remote, uint8_t *buffer_ptr, int buffer_size) |
515 |
{ |
516 |
int size;
|
517 |
if (receivedBuffer[rIdxUp].data==NULL) { //block till first message arrives |
518 |
wait4data(local, NULL, NULL); |
519 |
} |
520 |
|
521 |
assert(receivedBuffer[rIdxUp].data && receivedBuffer[rIdxUp].id); |
522 |
|
523 |
(*remote) = receivedBuffer[rIdxUp].id; |
524 |
// retrieve a msg from the buffer
|
525 |
size = receivedBuffer[rIdxUp].len; |
526 |
if (size>buffer_size) {
|
527 |
fprintf(stderr, "Net-helper : recv_from_peer: buffer too small (size:%d > buffer_size: %d)!\n",size,buffer_size);
|
528 |
return -1; |
529 |
} |
530 |
memcpy(buffer_ptr, receivedBuffer[rIdxUp].data, size); |
531 |
free(receivedBuffer[rIdxUp].data); |
532 |
receivedBuffer[rIdxUp].data = NULL;
|
533 |
receivedBuffer[rIdxUp].id = NULL;
|
534 |
|
535 |
rIdxUp = (rIdxUp+1)%NH_BUFFER_SIZE;
|
536 |
|
537 |
// fprintf(stderr, "Net-helper : I've got mail!!!\n");
|
538 |
|
539 |
return size;
|
540 |
} |
541 |
|
542 |
|
543 |
int wait4data(const struct nodeID *n, struct timeval *tout, int *fds) { |
544 |
|
545 |
struct event *timeout_ev = NULL; |
546 |
struct event *fd_ev[FDSSIZE];
|
547 |
bool fd_triggered[FDSSIZE] = { false }; |
548 |
int i;
|
549 |
|
550 |
// fprintf(stderr,"Net-helper : Waiting for data to come...\n");
|
551 |
if (tout) { //if tout==NULL, loop wait infinitely |
552 |
timeout_ev = event_new(base, -1, EV_TIMEOUT, &t_out_cb, NULL); |
553 |
event_add(timeout_ev, tout); |
554 |
} |
555 |
for (i = 0; fds && fds[i] != -1; i ++) { |
556 |
if (i >= FDSSIZE) {
|
557 |
fprintf(stderr, "Can't listen on more than %d file descriptors!\n", FDSSIZE);
|
558 |
break;
|
559 |
} |
560 |
fd_ev[i] = event_new(base, fds[i], EV_READ, &fd_cb, &fd_triggered[i]); |
561 |
event_add(fd_ev[i], NULL);
|
562 |
} |
563 |
|
564 |
while(receivedBuffer[rIdxUp].data==NULL && timeoutFired==0 && fdTriggered==0) { |
565 |
// event_base_dispatch(base);
|
566 |
event_base_loop(base,EVLOOP_ONCE); |
567 |
} |
568 |
|
569 |
//delete one-time events
|
570 |
if (timeout_ev) {
|
571 |
if (!timeoutFired) event_del(timeout_ev);
|
572 |
event_free(timeout_ev); |
573 |
} |
574 |
for (i = 0; fds && fds[i] != -1; i ++) { |
575 |
if (! fd_triggered[i]) {
|
576 |
fds[i] = -2;
|
577 |
event_del(fd_ev[i]); |
578 |
//} else {
|
579 |
//fprintf(stderr, "\twait4data: fd %d triggered\n", fds[i]);
|
580 |
} |
581 |
event_free(fd_ev[i]); |
582 |
} |
583 |
|
584 |
if (fdTriggered) {
|
585 |
fdTriggered = false;
|
586 |
//fprintf(stderr, "\twait4data: fd event\n");
|
587 |
return 2; |
588 |
} else if (timeoutFired) { |
589 |
timeoutFired = 0;
|
590 |
//fprintf(stderr, "\twait4data: timed out\n");
|
591 |
return 0; |
592 |
} else if (receivedBuffer[rIdxUp].data!=NULL) { |
593 |
//fprintf(stderr, "\twait4data: ML receive\n");
|
594 |
return 1; |
595 |
} else {
|
596 |
fprintf(stderr, "BUG in wait4data\n");
|
597 |
exit(EXIT_FAILURE); |
598 |
} |
599 |
} |
600 |
|
601 |
socketID_handle getRemoteSocketID(const char *ip, int port) { |
602 |
char str[SOCKETID_STRING_SIZE];
|
603 |
socketID_handle h; |
604 |
|
605 |
snprintf(str, SOCKETID_STRING_SIZE, "%s:%d-%s:%d", ip, port, ip, port);
|
606 |
h = malloc(SOCKETID_SIZE); |
607 |
mlStringToSocketID(str, h); |
608 |
|
609 |
return h;
|
610 |
} |
611 |
|
612 |
struct nodeID *create_node(const char *rem_IP, int rem_port) { |
613 |
socketID_handle s; |
614 |
struct nodeID *remote;
|
615 |
|
616 |
s = getRemoteSocketID(rem_IP, rem_port); |
617 |
remote = id_lookup_dup(s); |
618 |
free(s); |
619 |
|
620 |
return remote;
|
621 |
} |
622 |
|
623 |
// TODO: check why closing the connection is annoying for the ML
|
624 |
void nodeid_free(struct nodeID *n) { |
625 |
if (n && (--(n->refcnt) == 1)) { |
626 |
/*
|
627 |
struct nodeID **npos;
|
628 |
// mlCloseConnection(n->connID);
|
629 |
npos = id_lookup(n->addr);
|
630 |
*npos = NULL;
|
631 |
mlCloseSocket(n->addr);
|
632 |
free(n);
|
633 |
*/
|
634 |
} |
635 |
} |
636 |
|
637 |
|
638 |
const char *node_addr(const struct nodeID *s) |
639 |
{ |
640 |
static char addr[256]; |
641 |
// TODO: mlSocketIDToString always return 0 !!!
|
642 |
int r = mlSocketIDToString(s->addr,addr,256); |
643 |
if (!r)
|
644 |
return addr;
|
645 |
else
|
646 |
return ""; |
647 |
} |
648 |
|
649 |
struct nodeID *nodeid_dup(struct nodeID *s) |
650 |
{ |
651 |
s->refcnt++; |
652 |
return s;
|
653 |
} |
654 |
|
655 |
int nodeid_equal(const struct nodeID *s1, const struct nodeID *s2) |
656 |
{ |
657 |
return (mlCompareSocketIDs(s1->addr,s2->addr) == 0); |
658 |
} |
659 |
|
660 |
int nodeid_dump(uint8_t *b, const struct nodeID *s, size_t max_write_size) |
661 |
{ |
662 |
if (max_write_size < SOCKETID_STRING_SIZE) return -1; |
663 |
|
664 |
mlSocketIDToString(s->addr,(char *)b,SOCKETID_STRING_SIZE);
|
665 |
//fprintf(stderr,"Dumping nodeID : ho scritto %s (%d bytes)\n",b, strlen((char *)b));
|
666 |
|
667 |
// memcpy(b, s->addr,SOCKETID_SIZE);//sizeof(struct sockaddr_in6)*2
|
668 |
// return SOCKETID_SIZE;//sizeof(struct sockaddr_in6)*2;
|
669 |
|
670 |
return 1 + strlen((char *)b); //terminating \0 IS included in the size |
671 |
} |
672 |
|
673 |
struct nodeID *nodeid_undump(const uint8_t *b, int *len) |
674 |
{ |
675 |
uint8_t sid[SOCKETID_SIZE]; |
676 |
socketID_handle h = (socketID_handle) sid; |
677 |
mlStringToSocketID((char *)b,h);
|
678 |
*len = strlen((char*)b) + 1; |
679 |
return id_lookup_dup(h);
|
680 |
} |