src/gras/RL/gras_rl.c

   1 /* $Id$ */
   2
   3 /* gras_rl - implementation of GRAS on real life                            */
   4
   5 /* Authors: Martin Quinson                                                  */
   6 /* Copyright (C) 2003 the OURAGAN project.                                  */
   7
   8 /* This program is free software; you can redistribute it and/or modify it
   9    under the terms of the license (GNU LGPL) which comes with this package. */
  10
  11 #include "gras_rl.h"
  12
  13 #include <stdlib.h>
  14 #include <string.h>
  15
  16 #include <unistd.h> /* sleep() */
  17
  18 #include <sys/types.h>
  19 #include <sys/socket.h>
  20 #include <netinet/in.h> /* struct in_addr */
  21 #include <signal.h>
  22
  23
  24 /* NWS headers */
  25 #include "osutil.h"
  26 #include "timeouts.h"
  27 #include "protocol.h"
  28
  29 XBT_LOG_NEW_DEFAULT_CATEGORY(rl,"Real Life");
  30
  31 /* globals */
  32 static grasProcessData_t *_grasProcessData;
  33
  34 /* Prototypes of internal functions */
  35 static int grasConversionRequired(const DataDescriptor *description, size_t howMany);
  36 static xbt_error_t
  37 _gras_rawsock_exchange(gras_rawsock_t *sd, int sender, unsigned int timeout,
  38                        unsigned int expSize, unsigned int msgSize);
  39
  40
  41 xbt_error_t gras_process_init() {
  42   if (!(_grasProcessData=(grasProcessData_t *)malloc(sizeof(grasProcessData_t)))) {
  43     fprintf(stderr,"gras_process_init: cannot malloc %d bytes\n",sizeof(grasProcessData_t));
  44     return malloc_error;
  45   }
  46   _grasProcessData->grasMsgQueueLen=0;
  47   _grasProcessData->grasMsgQueue = NULL;
  48
  49   _grasProcessData->grasCblListLen = 0;
  50   _grasProcessData->grasCblList = NULL;
  51
  52   _grasProcessData->userdata = NULL;
  53   return no_error;
  54 }
  55 xbt_error_t gras_process_finalize() {
  56   fprintf(stderr,"FIXME: %s not implemented (=> leaking on exit :)\n",__FUNCTION__);
  57   return no_error;
  58 }
  59
  60 /* **************************************************************************
  61  * Openning/Maintaining/Closing connexions
  62  * **************************************************************************/
  63 xbt_error_t
  64 gras_sock_client_open(const char *host, short port,
  65                       /* OUT */ gras_sock_t **sock) {
  66
  67   int addrCount;
  68   IPAddress addresses[10];
  69   int i;
  70   int sd;
  71
  72   if (!(*sock=malloc(sizeof(gras_sock_t)))) {
  73     fprintf(stderr,"Malloc error\n");
  74     return malloc_error;
  75   }
  76   (*sock)->peer_addr=NULL;
  77
  78   if (!(addrCount = IPAddressValues(host, addresses, 10))) {
  79     fprintf(stderr,"grasOpenClientSocket: address retrieval of '%s' failed\n",host);
  80     return system_error;
  81   }
  82
  83   for(i = 0; i < addrCount && i<10 ; i++) {
  84     if(CallAddr(addresses[i], port, &sd, -1)) {
  85       (*sock)->sock = sd;
  86       (*sock)->port = port;
  87       return no_error;
  88     }
  89   }
  90   free(*sock);
  91   fprintf(stderr,"grasOpenClientSocket: something wicked happenned while connecting to %s:%d",
  92           host,port);
  93   return system_error;
  94 }
  95
  96
  97 xbt_error_t
  98 gras_sock_server_open(unsigned short startingPort, unsigned short endingPort,
  99                       /* OUT */ gras_sock_t **sock) {
 100
 101   unsigned short port;
 102
 103   if (!(*sock=malloc(sizeof(gras_sock_t)))) {
 104     fprintf(stderr,"Malloc error\n");
 105     return malloc_error;
 106   }
 107
 108   if (!EstablishAnEar(startingPort,endingPort,&((*sock)->sock),&port)) {
 109     free(*sock);
 110     return unknown_error;
 111   }
 112   (*sock)->peer_addr=NULL;
 113   (*sock)->port=port;
 114
 115   return no_error;
 116 }
 117
 118 xbt_error_t gras_sock_close(gras_sock_t *sock) {
 119   if (sock) {
 120     DROP_SOCKET(&(sock->sock));
 121     if (sock->peer_addr) free(sock->peer_addr);
 122     free(sock);
 123   }
 124   return no_error;
 125 }
 126
 127 unsigned short
 128 gras_sock_get_my_port(gras_sock_t *sd) {
 129   if (!sd) return -1;
 130   return sd->port;
 131 }
 132
 133 unsigned short
 134 gras_sock_get_peer_port(gras_sock_t *sd) {
 135   if (!sd) return -1;
 136   return PeerNamePort(sd->sock);
 137 }
 138
 139 char *
 140 gras_sock_get_peer_name(gras_sock_t *sd) {
 141   char *tmp;
 142
 143   if (!sd) return NULL;
 144   tmp=PeerName_r(sd->sock);
 145   if (tmp) {
 146     strcpy(sd->peer_name,tmp);
 147     free(tmp);
 148   } else {
 149     strcpy(sd->peer_name,"unknown");
 150   }
 151
 152   return sd->peer_name;
 153 }
 154
 155
 156 /* **************************************************************************
 157  * format handling
 158  * **************************************************************************/
 159
 160 /*
 161  * Returns 1 or 0 depending on whether or not format conversion is required for
 162  * data with the format described by the #howMany#-long array #description#.
 163  */
 164 int grasConversionRequired(const DataDescriptor *description, size_t howMany){
 165   int i;
 166
 167   if(DataSize(description, howMany, HOST_FORMAT) !=
 168      DataSize(description, howMany, NETWORK_FORMAT)) {
 169     return 1;
 170   }
 171
 172   for(i = 0; i < howMany; i++) {
 173     if(description[i].type == STRUCT_TYPE) {
 174       if(grasConversionRequired(description[i].members, description[i].length)) {
 175         return 1;
 176       }
 177     } else if(DifferentFormat(description[i].type))
 178       return 1;
 179   }
 180
 181   return DifferentOrder();
 182 }
 183
 184 /* **************************************************************************
 185  * Actually exchanging messages
 186  * **************************************************************************/
 187
 188 /*
 189  * Discard data on the socket because of failure on our side
 190  */
 191 void
 192 gras_msg_discard(gras_sock_t *sd, size_t size) {
 193   char garbage[2048];
 194   int s=size;
 195
 196   while(s > 0) {
 197     (void)RecvBytes(sd->sock,
 198                     garbage,
 199                     (s > sizeof(garbage)) ? sizeof(garbage) : s,
 200                     GetTimeOut(RECV, Peer(sd->sock), 2048));
 201     s -= sizeof(garbage);
 202   }
 203 }
 204
 205 int grasDataRecv( gras_sock_t *sd,
 206                   void **data,
 207                   const DataDescriptor *description,
 208                   size_t description_length,
 209                   unsigned int repetition) {
 210
 211   void *converted=NULL;
 212   int convertIt;
 213   void *destination; /* where to receive the data from the net (*data or converted)*/
 214   int recvResult;
 215   size_t netSize,hostSize;
 216   double start; /* for timeouts */
 217   char *net,*host; /* iterators */
 218   int i;
 219
 220   gras_lock();
 221
 222   netSize = DataSize(description, description_length, NETWORK_FORMAT);
 223   hostSize = DataSize(description, description_length, HOST_FORMAT);
 224
 225   if (!(*data=malloc(hostSize*repetition))) {
 226       gras_unlock();
 227       ERROR1("grasDataRecv: memory allocation of %d bytes failed\n", hostSize*repetition);
 228       return 0;
 229   }
 230
 231   convertIt = grasConversionRequired(description, description_length);
 232
 233   if(convertIt) {
 234     if (!(converted = malloc(netSize*repetition))) {
 235       free(*data);
 236       gras_unlock();
 237       ERROR1("RecvData: memory allocation of %d bytes failed\n", netSize*repetition);
 238       return 0;
 239     }
 240     destination = converted;
 241   } else {
 242     destination = *data;
 243   }
 244
 245   /* adaptive timeout */
 246   start = CurrentTime();
 247
 248   recvResult = RecvBytes(sd->sock, destination, netSize*repetition,
 249                          GetTimeOut(RECV, Peer(sd->sock), netSize*repetition));
 250   /* we assume a failure is a timeout ... Shouldn't hurt
 251    * too much getting a bigger timeout anyway */
 252   SetTimeOut(RECV, sd->sock, CurrentTime()-start, netSize*repetition, !recvResult);
 253
 254   fprintf(stderr,"RECV [seqLen=%d;netsize=%d;hostsize=%d] : (",
 255           repetition,netSize,hostSize);
 256   for (i=0; i<netSize * repetition; i++) {
 257     if (i) fputc('.',stderr);
 258     fprintf(stderr,"%02X",((unsigned char*)destination)[i]);
 259   }
 260   fprintf(stderr,") on %p.\n",destination);
 261
 262   if (recvResult != 0) {
 263     if(convertIt) {
 264       for (i=0, net=(char*)converted, host=(char*)*data;
 265            i<repetition;
 266            i++, net += netSize, host += hostSize) {
 267         ConvertData((void*)host, (void*)net, description, description_length, NETWORK_FORMAT);
 268       }
 269     }
 270     if(converted != NULL)
 271       free(converted);
 272   }
 273   if (!gras_unlock()) return 0;
 274
 275   return recvResult;
 276 }
 277
 278 xbt_error_t grasDataSend(gras_sock_t *sd,
 279                          const void *data,
 280                          const DataDescriptor *description,
 281                          size_t description_length,
 282                          int repetition) {
 283
 284   void *converted;
 285   char *net,*host; /* iterators */
 286   int sendResult,i;
 287   const void *source;
 288   size_t netSize = DataSize(description, description_length, NETWORK_FORMAT);
 289   size_t hostSize = DataSize(description, description_length, HOST_FORMAT);
 290   double start; /* for timeouts */
 291
 292   gras_lock();
 293   converted = NULL;
 294
 295   if(grasConversionRequired(description, description_length)) {
 296     converted = malloc(netSize * repetition);
 297     if(converted == NULL) {
 298       gras_unlock();
 299       fprintf(stderr,"grasDataSend: memory allocation of %d bytes failed.\n",netSize * repetition);
 300       return malloc_error;
 301     }
 302
 303     for (i=0, net=(char*)converted, host=(char*)data;
 304          i<repetition;
 305          i++, net += netSize, host += hostSize)
 306       ConvertData((void*)net, (void*)host, description, description_length, HOST_FORMAT);
 307     source = converted;
 308   } else {
 309     source = data;
 310   }
 311
 312   fprintf(stderr,"SEND (");
 313   for (i=0; i<netSize * repetition; i++) {
 314     if (i) fputc('.',stderr);
 315     fprintf(stderr,"%02X",((unsigned char*)source)[i]);
 316   }
 317   fprintf(stderr,") from %p\n",source);
 318   // grasDataDescDump((const DataDescriptor *)description, description_length);
 319
 320   /* adaptive timeout */
 321   start = CurrentTime();
 322   sendResult = SendBytes(sd->sock, source, netSize * repetition,
 323                          GetTimeOut(SEND, Peer(sd->sock),netSize*repetition));
 324   /* we assume a failure is a timeout ... Shouldn't hurt
 325    * too much getting a bigger timeout anyway */
 326   SetTimeOut(SEND, sd->sock, CurrentTime()-start, netSize * repetition, !sendResult);
 327   if(converted != NULL)
 328     free((void *)converted);
 329
 330   gras_unlock();
 331   return no_error;
 332 }
 333
 334 xbt_error_t
 335 grasMsgRecv(gras_msg_t **msg,
 336             double timeOut) {
 337   int dummyldap;
 338   gras_msg_t *res;
 339   size_t recvd; /* num of bytes received */
 340   int i;
 341   gras_sock_t *sd;
 342
 343   if (!(sd = (gras_sock_t*)malloc(sizeof(gras_sock_t)))) {
 344     fprintf(stderr,"grasMsgRecv: Malloc error\n");
 345     return malloc_error;
 346   }
 347
 348   if(!IncomingRequest(timeOut, &(sd->sock), &dummyldap)) {
 349     free(sd);
 350     return timeout_error;
 351   }
 352
 353   if (!gras_lock()) {
 354     free (sd);
 355     return thread_error;
 356   }
 357   if (!(res = malloc(sizeof(gras_msg_t)))) {
 358     gras_unlock();
 359     *msg=NULL;
 360     free(sd);
 361     return malloc_error;
 362   }
 363   *msg=res;
 364   res->sock=sd;
 365   res->freeDirective=free_after_use;
 366
 367   if (!gras_unlock()) return thread_error;
 368
 369   if(!(recvd=grasDataRecv( sd,
 370                            (void **)&(res->header),
 371                            headerDescriptor,headerDescriptorCount,1))) {
 372     fprintf(stderr,"grasMsgRecv: no message received\n");
 373     return network_error;
 374   }
 375   res->header->dataSize -= recvd;
 376
 377   fprintf(stderr,"Received header=ver:'%s' msg:%d size:%d seqCount:%d\n",
 378           res->header->version,  res->header->message,
 379           res->header->dataSize, res->header->seqCount);
 380   if (strncmp(res->header->version,GRASVERSION,strlen(GRASVERSION))) {
 381     /* The other side do not use the same version than us. Let's panic */
 382     char junk[2046];
 383     int junkint;
 384     while ((junkint=recv(sd->sock,junk,2046,0)) == 2046);
 385     fprintf(stderr,"PANIC: Got a message from a peer (%s:%d) using GRAS version '%10s' while this side use version '%s'.\n",
 386             gras_sock_get_peer_name(sd),gras_sock_get_peer_port(sd),
 387             res->header->version,GRASVERSION);
 388     return mismatch_error;
 389   }
 390   if (!(res->entry=grasMsgEntryGet(res->header->message))) {
 391     /* This message is not registered on our side, discard it */
 392     gras_msg_discard(sd,res->header->dataSize);
 393
 394     i= res->header->message;
 395     free(res->header);
 396     free(res);
 397     *msg=NULL;
 398
 399     fprintf(stderr,"grasMsgRecv: unregistered message %d received from %s\n",
 400             i,gras_sock_get_peer_name(sd));
 401     return mismatch_error;
 402   }
 403
 404   if (!(recvd=grasDataRecv( sd,
 405                             (void **)&(res->dataCount),
 406                             countDescriptor,res->entry->seqCount,1))) {
 407     gras_msg_discard(sd,res->header->dataSize);
 408     i = res->header->message;
 409     free(res->header);
 410     free(res);
 411     *msg=NULL;
 412     fprintf(stderr, "grasMsgRecv: Error while getting elemCounts in message %d received from %s\n",
 413           i,gras_sock_get_peer_name(sd));
 414     return network_error;
 415   }
 416   res->header->dataSize -= recvd;
 417
 418   if (!gras_lock()) return thread_error;
 419   if (!(res->data=(void**)malloc(sizeof(void*)*res->entry->seqCount))) {
 420     gras_msg_discard(sd,res->header->dataSize);
 421
 422     i= res->header->message;
 423     free(res->header);
 424     free(res->dataCount);
 425     free(res);
 426     *msg=NULL;
 427
 428     gras_unlock();
 429     fprintf(stderr,"grasMsgRecv: Out of memory while receiving message %d received from %s\n",
 430           i,gras_sock_get_peer_name(sd));
 431     return malloc_error;
 432   }
 433   if (!gras_unlock()) return thread_error;
 434
 435   for (i=0;i<res->entry->seqCount;i++) {
 436
 437     if(!(recvd=grasDataRecv( sd,
 438                              &(res->data[i]),
 439                              (const DataDescriptor*)res->entry->dd[i],
 440                              res->entry->ddCount[i],
 441                              res->dataCount[i]) )) {
 442       gras_msg_discard(sd,res->header->dataSize);
 443       for (i--;i>=0;i--) free(res->data[i]);
 444       free(res->dataCount);
 445       free(res->data);
 446       free(res);
 447       *msg=NULL;
 448       fprintf(stderr,"grasDataRecv: Transmision error while receiving message %d received from %s\n",
 449             i,gras_sock_get_peer_name(sd));
 450       return network_error;
 451     }
 452     res->header->dataSize -= recvd;
 453   }
 454
 455   if (res->header->dataSize) {
 456     fprintf(stderr,"Damnit dataSize = %d != 0 after receiving message %d received from %s\n",
 457             res->header->dataSize,i,gras_sock_get_peer_name(sd));
 458     return unknown_error;
 459   }
 460   return no_error;
 461 }
 462
 463 /*
 464  * Send a message to the network
 465  */
 466
 467 xbt_error_t
 468 gras_msg_send(gras_sock_t *sd,
 469             gras_msg_t *msg,
 470             e_xbt_free_directive_t freeDirective) {
 471
 472   xbt_error_t errcode;
 473   int i;
 474
 475   /* arg validity checks */
 476   xbt_assert0(msg,"Trying to send NULL message");
 477   xbt_assert0(sd, "Trying to send over a NULL socket");
 478
 479
 480   fprintf(stderr,"Header to send=ver:'%s' msg:%d size:%d seqCount:%d\n",
 481           msg->header->version,  msg->header->message,
 482           msg->header->dataSize, msg->header->seqCount);
 483
 484   /* Send the message */
 485   if ((errcode=grasDataSend(sd,
 486                             msg->header,
 487                             headerDescriptor,headerDescriptorCount,1))) {
 488     fprintf(stderr,"gras_msg_send: Error '%s' while sending header of message %s to %s:%d\n",
 489             xbt_error_name(errcode),msg->entry->name,gras_sock_get_peer_name(sd),gras_sock_get_peer_port(sd));
 490     return errcode;
 491   }
 492
 493   if ((errcode=grasDataSend(sd,
 494                             msg->dataCount,countDescriptor,countDescriptorCount,
 495                             msg->entry->seqCount))) {
 496     fprintf(stderr,"gras_msg_send: Error '%s' while sending sequence counts of message %s to %s\n",
 497             xbt_error_name(errcode),msg->entry->name,gras_sock_get_peer_name(sd));
 498     return errcode;
 499   }
 500
 501   for (i=0; i<msg->entry->seqCount; i++) {
 502     if ((errcode=grasDataSend(sd,
 503                               msg->data[i],
 504                               (const DataDescriptor*)msg->entry->dd[i],msg->entry->ddCount[i],
 505                               msg->dataCount[i]))) {
 506       fprintf(stderr,"gras_msg_send: Error '%s' while sending sequence %d of message %s to %s\n",
 507               xbt_error_name(errcode),i,msg->entry->name,gras_sock_get_peer_name(sd));
 508       return errcode;
 509     }
 510   }
 511
 512   if (freeDirective == free_after_use) gras_msg_free(msg);
 513   return no_error;
 514 }
 515
 516 gras_sock_t *gras_sock_new(void) {
 517   return malloc(sizeof(gras_sock_t));
 518 }
 519
 520 void grasSockFree(gras_sock_t *s) {
 521   if (s) free (s);
 522 }
 523
 524 /* **************************************************************************
 525  * Creating/Using raw sockets
 526  * **************************************************************************/
 527 xbt_error_t gras_rawsock_server_open(unsigned short startingPort,
 528                                   unsigned short endingPort,
 529                                   unsigned int bufSize, gras_rawsock_t **sock) {
 530   struct sockaddr_in sockaddr;
 531
 532
 533   if (!(*sock=malloc(sizeof(gras_rawsock_t)))) {
 534     fprintf(stderr,"Malloc error\n");
 535     return malloc_error;
 536   }
 537
 538
 539   for((*sock)->port = startingPort;
 540       (*sock)->port <= endingPort;
 541       (*sock)->port++) {
 542     if(((*sock)->sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
 543       ERROR0("gras_rawsock_server_open: socket creation failed\n");
 544       free(*sock);
 545       return system_error;
 546     }
 547
 548     setsockopt((*sock)->sock, SOL_SOCKET, SO_RCVBUF, (char *)&bufSize, sizeof(bufSize));
 549     setsockopt((*sock)->sock, SOL_SOCKET, SO_SNDBUF, (char *)&bufSize, sizeof(bufSize));
 550
 551     memset(&sockaddr, 0, sizeof(sockaddr));
 552     sockaddr.sin_port = htons((unsigned short)(*sock)->port);
 553     sockaddr.sin_addr.s_addr = INADDR_ANY;
 554     sockaddr.sin_family = AF_INET;
 555
 556     if (bind((*sock)->sock,
 557              (struct sockaddr *)&sockaddr, sizeof(sockaddr)) != -1  &&
 558         listen((*sock)->sock, 1) != -1) {
 559       break;
 560     }
 561     close((*sock)->sock);
 562   }
 563
 564   if((*sock)->port > endingPort) {
 565     fprintf(stderr,"gras_rawsock_server_open: couldn't find a port between %d and %d\n",
 566             startingPort, endingPort);
 567     free(*sock);
 568     return mismatch_error;
 569   }
 570
 571   return no_error;
 572 }
 573
 574 void Dummy(int);
 575 void Dummy(int sig) {
 576   return;
 577 }
 578
 579 xbt_error_t gras_rawsock_client_open(const char *host, short port,
 580                                   unsigned int bufSize, gras_rawsock_t **sock) {
 581   int i,addrCount;
 582   IPAddress addresses[10];
 583   void (*was)(int);
 584   struct sockaddr_in sockaddr;
 585
 586   if (!(*sock=malloc(sizeof(gras_rawsock_t)))) {
 587     fprintf(stderr,"Malloc error\n");
 588     return malloc_error;
 589   }
 590   (*sock)->port=-1;
 591
 592   if (!(addrCount = IPAddressValues(host, addresses, 10))) {
 593     fprintf(stderr,"grasOpenClientSocket: address retrieval of '%s' failed\n",host);
 594     free(*sock);
 595     return system_error;
 596   }
 597   (*sock)->port = port;
 598   memset(&sockaddr, 0, sizeof(sockaddr));
 599   sockaddr.sin_port = htons((unsigned short)(*sock)->port);
 600   sockaddr.sin_family = AF_INET;
 601
 602   for(i = 0; i < addrCount && i<10 ; i++) {
 603     if(((*sock)->sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
 604       fprintf(stderr,"gras_rawsock_client_open: socket creation failed\n");
 605       free(*sock);
 606       return system_error;
 607     }
 608
 609     setsockopt((*sock)->sock, SOL_SOCKET, SO_RCVBUF,
 610                (char *)&bufSize, sizeof(bufSize));
 611     setsockopt((*sock)->sock, SOL_SOCKET, SO_SNDBUF,
 612                (char *)&bufSize, sizeof(bufSize));
 613
 614     sockaddr.sin_addr.s_addr = addresses[i];
 615
 616     was = signal(SIGALRM,Dummy);
 617     SetRealTimer(GetTimeOut(CONN, addresses[i], 0));
 618     if(connect((*sock)->sock,
 619                (struct sockaddr *)&sockaddr, sizeof(sockaddr)) == 0) {
 620       SetRealTimer(0);
 621       signal(SIGALRM,was);
 622       break;
 623
 624     }
 625     SetRealTimer(0);
 626     close((*sock)->sock);
 627     signal(SIGALRM,was);
 628   }
 629
 630   if (i>=10) {
 631     free(*sock);
 632     fprintf(stderr,
 633             "grasOpenClientRawSocket: Unable to contact %s:%d\n",
 634             host,port);
 635     return network_error;
 636   }
 637
 638   return no_error;
 639 }
 640
 641 xbt_error_t gras_rawsock_close(gras_rawsock_t *sd) {
 642   if (sd) {
 643     CloseSocket(&(sd->sock), 0);
 644     free(sd);
 645   }
 646   return no_error;
 647 }
 648
 649 unsigned short gras_rawsock_get_peer_port(gras_rawsock_t *sd) {
 650   if (!sd) return -1;
 651   return sd->port;
 652 }
 653
 654 /* FIXME: RL ignores the provided timeout and compute an appropriate one */
 655 static xbt_error_t
 656 _gras_rawsock_exchange(gras_rawsock_t *sd, int sender, unsigned int timeout,
 657                      unsigned int expSize, unsigned int msgSize){
 658   char *expData;
 659   int bytesThisCall;
 660   int bytesThisMessage;
 661   int bytesTotal;
 662
 663   fd_set rd_set;
 664   int rv;
 665
 666   char *name;
 667   IPAddress addr;
 668
 669   int ltimeout;
 670   struct timeval timeOut;
 671
 672   if((expData = (char *)malloc(msgSize)) == NULL) {
 673     fprintf(stderr,"gras_rawsock_send: malloc %d failed\n", msgSize);
 674     return malloc_error;
 675   }
 676
 677   /* let's get information on the caller (needed later on) */
 678   name = PeerName_r(sd->sock);
 679   IPAddressValue(name, &addr);
 680   free(name);
 681
 682   ltimeout = (int)GetTimeOut((sender ? SEND : RECV), addr, expSize/msgSize +1);
 683
 684   if (sender)
 685     SetRealTimer(ltimeout);
 686
 687
 688   for(bytesTotal = 0;
 689       bytesTotal < expSize;
 690       bytesTotal += bytesThisMessage) {
 691     for(bytesThisMessage = 0;
 692         bytesThisMessage < msgSize;
 693         bytesThisMessage += bytesThisCall) {
 694
 695
 696       if(sender) {
 697         bytesThisCall = send(sd->sock, expData, msgSize - bytesThisMessage, 0);
 698       } else {
 699         bytesThisCall = 0;
 700         FD_ZERO(&rd_set);
 701         FD_SET(sd->sock,&rd_set);
 702         timeOut.tv_sec = ltimeout;
 703         timeOut.tv_usec = 0;
 704         /*
 705          * YUK!  The timeout can get to be REALLY large if the
 706          * amount of data gets big -- fence it at 60 secs
 707          */
 708         if ((ltimeout <= 0) || (ltimeout > 60)) {
 709           ltimeout = 60;
 710         }
 711         rv = select(sd->sock+1,&rd_set,NULL,NULL,&timeOut);
 712         if(rv > 0) {
 713           bytesThisCall = recv(sd->sock, expData, msgSize-bytesThisMessage, 0);
 714         }
 715       }
 716     }
 717   }
 718   free(expData);
 719   return no_error;
 720 }
 721
 722 xbt_error_t
 723 gras_rawsock_recv(gras_rawsock_t *sd, unsigned int expSize, unsigned int msgSize,
 724                 unsigned int timeout) {
 725   return _gras_rawsock_exchange(sd,0,timeout,expSize,msgSize);
 726 }
 727 xbt_error_t
 728 gras_rawsock_send(gras_rawsock_t *sd, unsigned int expSize, unsigned int msgSize){
 729   return _gras_rawsock_exchange(sd,1,0,expSize,msgSize);
 730 }
 731
 732
 733
 734
 735 /* **************************************************************************
 736  * Process data
 737  * **************************************************************************/
 738 grasProcessData_t *grasProcessDataGet() {
 739   return _grasProcessData;
 740 }
 741
 742 /* **************************************************************************
 743  * Wrappers over OS functions
 744  * **************************************************************************/
 745 double gras_time() {
 746   return MicroTime();
 747 }
 748
 749 void gras_sleep(unsigned long sec,unsigned long usec) {
 750   sleep(sec);
 751   if (usec/1000000) sleep(usec/1000000);
 752   (void)usleep(usec % 1000000);
 753 }