Revision 336ad402
| net_helper-ml.c | ||
|---|---|---|
| 10 | 10 |
#include <stdlib.h> |
| 11 | 11 |
#include <stdio.h> |
| 12 | 12 |
#include <string.h> |
| 13 |
#include <assert.h> |
|
| 13 | 14 |
|
| 14 | 15 |
|
| 15 | 16 |
#include "net_helper.h" |
| ... | ... | |
| 37 | 38 |
#define NH_ML_INIT_TIMEOUT {1, 0}
|
| 38 | 39 |
|
| 39 | 40 |
static int sIdx = 0; |
| 40 |
static int rIdx = 0; |
|
| 41 |
static int rIdxML = 0; //reveive from ML to this buffer position |
|
| 42 |
static int rIdxUp = 0; //hand up to layer above at this buffer position |
|
| 41 | 43 |
|
| 42 | 44 |
typedef struct nodeID {
|
| 43 | 45 |
socketID_handle addr; |
| ... | ... | |
| 142 | 144 |
* @return the index of a free slot in the received msgs buffer, -1 if no free slot available. |
| 143 | 145 |
*/ |
| 144 | 146 |
static int next_R() {
|
| 145 |
const int size = 1024; |
|
| 146 |
if (receivedBuffer[rIdx].data==NULL) {
|
|
| 147 |
receivedBuffer[rIdx].data = malloc(size); |
|
| 148 |
} |
|
| 149 |
else {
|
|
| 150 |
int count; |
|
| 151 |
for (count=0;count<NH_BUFFER_SIZE;count++) {
|
|
| 152 |
rIdx = (++rIdx)%NH_BUFFER_SIZE; |
|
| 153 |
if (receivedBuffer[rIdx].data==NULL) |
|
| 154 |
break; |
|
| 155 |
} |
|
| 156 |
if (count==NH_BUFFER_SIZE) |
|
| 157 |
return -1; |
|
| 158 |
else {
|
|
| 159 |
receivedBuffer[rIdx].data = malloc(size); |
|
| 160 |
} |
|
| 147 |
if (receivedBuffer[rIdxML].data==NULL) {
|
|
| 148 |
int ret = rIdxML; |
|
| 149 |
rIdxML = (rIdxML+1)%NH_BUFFER_SIZE; |
|
| 150 |
return ret; |
|
| 151 |
} else {
|
|
| 152 |
//TODO: handle receive overload situation! |
|
| 153 |
return -1; |
|
| 161 | 154 |
} |
| 162 |
memset(receivedBuffer[rIdx].data,0,size); |
|
| 163 |
return rIdx; |
|
| 164 | 155 |
} |
| 165 | 156 |
|
| 166 | 157 |
/** |
| ... | ... | |
| 333 | 324 |
// fprintf(stderr, "Net-helper : message arrived from %s\n",str); |
| 334 | 325 |
// buffering the received message only if possible, otherwise ignore it... |
| 335 | 326 |
int index = next_R(); |
| 336 |
if (index >=0) {
|
|
| 337 |
// receivedBuffer[index][0] = malloc(buflen); |
|
| 327 |
if (index<0) {
|
|
| 328 |
fprintf(stderr,"Net-helper: receive buffer full\n "); |
|
| 329 |
return; |
|
| 330 |
} else {
|
|
| 331 |
receivedBuffer[index].data = malloc(buflen); |
|
| 338 | 332 |
if (receivedBuffer[index].data == NULL) {
|
| 339 |
fprintf(stderr, "Net-helper : memory error while creating a new message buffer \n");
|
|
| 333 |
fprintf(stderr,"Net-helper: memory full, can't receive!\n ");
|
|
| 340 | 334 |
return; |
| 341 | 335 |
} |
| 342 |
// creating a new sender nodedID |
|
| 336 |
receivedBuffer[index].len = buflen; |
|
| 337 |
memcpy(receivedBuffer[index].data,buffer,buflen); |
|
| 338 |
// save the socketID of the sender |
|
| 343 | 339 |
receivedBuffer[index].id = id_lookup_dup(arg->remote_socketID); |
| 344 |
receivedBuffer[index].data = realloc(receivedBuffer[index].data,buflen); |
|
| 345 |
memset(receivedBuffer[index].data,0,buflen); |
|
| 346 |
receivedBuffer[index].len = buflen; |
|
| 347 |
//*(receivedBuffer[index][0]) = buflen; |
|
| 348 |
memcpy(receivedBuffer[index].data,buffer,buflen); |
|
| 349 |
// get the socketID of the sender |
|
| 350 | 340 |
} |
| 351 | 341 |
} |
| 352 | 342 |
// event_base_loopbreak(base); |
| ... | ... | |
| 481 | 471 |
int recv_from_peer(const struct nodeID *local, struct nodeID **remote, uint8_t *buffer_ptr, int buffer_size) |
| 482 | 472 |
{
|
| 483 | 473 |
int size; |
| 484 |
if (receivedBuffer[rIdx].id==NULL) { //block till first message arrives
|
|
| 474 |
if (receivedBuffer[rIdxUp].data==NULL) { //block till first message arrives
|
|
| 485 | 475 |
wait4data(local, NULL, NULL); |
| 486 | 476 |
} |
| 487 | 477 |
|
| 488 |
(*remote) = receivedBuffer[rIdx].id; |
|
| 478 |
assert(receivedBuffer[rIdxUp].data && receivedBuffer[rIdxUp].id); |
|
| 479 |
|
|
| 480 |
(*remote) = receivedBuffer[rIdxUp].id; |
|
| 489 | 481 |
// retrieve a msg from the buffer |
| 490 |
size = receivedBuffer[rIdx].len; |
|
| 482 |
size = receivedBuffer[rIdxUp].len;
|
|
| 491 | 483 |
if (size>buffer_size) {
|
| 492 | 484 |
fprintf(stderr, "Net-helper : recv_from_peer: buffer too small (size:%d > buffer_size: %d)!\n",size,buffer_size); |
| 493 | 485 |
return -1; |
| 494 | 486 |
} |
| 495 |
memcpy(buffer_ptr, receivedBuffer[rIdx].data, size); |
|
| 496 |
free(receivedBuffer[rIdx].data); |
|
| 497 |
receivedBuffer[rIdx].data = NULL; |
|
| 498 |
receivedBuffer[rIdx].id = NULL; |
|
| 487 |
memcpy(buffer_ptr, receivedBuffer[rIdxUp].data, size); |
|
| 488 |
free(receivedBuffer[rIdxUp].data); |
|
| 489 |
receivedBuffer[rIdxUp].data = NULL; |
|
| 490 |
receivedBuffer[rIdxUp].id = NULL; |
|
| 491 |
|
|
| 492 |
rIdxUp = (rIdxUp+1)%NH_BUFFER_SIZE; |
|
| 499 | 493 |
|
| 500 | 494 |
// fprintf(stderr, "Net-helper : I've got mail!!!\n"); |
| 501 | 495 |
|
| ... | ... | |
| 509 | 503 |
if (tout) { //if tout==NULL, loop wait infinitely
|
| 510 | 504 |
event_base_once(base,-1, EV_TIMEOUT, &t_out_cb, NULL, tout); |
| 511 | 505 |
} |
| 512 |
while(receivedBuffer[rIdx].data==NULL && timeoutFired==0) {
|
|
| 506 |
while(receivedBuffer[rIdxUp].data==NULL && timeoutFired==0) {
|
|
| 513 | 507 |
// event_base_dispatch(base); |
| 514 | 508 |
event_base_loop(base,EVLOOP_ONCE); |
| 515 | 509 |
} |
| 516 | 510 |
timeoutFired = 0; |
| 517 | 511 |
// fprintf(stderr,"Back from eventlib loop.\n"); |
| 518 | 512 |
|
| 519 |
if (receivedBuffer[rIdx].data!=NULL) |
|
| 513 |
if (receivedBuffer[rIdxUp].data!=NULL)
|
|
| 520 | 514 |
return 1; |
| 521 | 515 |
else |
| 522 | 516 |
return 0; |
Also available in: Unified diff