src/gras/Transport/transport_plugin_tcp.c

   1 /* buf trp (transport) - buffered transport using the TCP one               */
   2
   3 /* Copyright (c) 2004, 2005, 2006, 2007, 2009, 2010. The SimGrid Team.
   4  * All rights reserved.                                                     */
   5
   6 /* This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the license (GNU LGPL) which comes with this package. */
   8
   9 #include <stdlib.h>
  10 #include <string.h>             /* memset */
  11
  12 #include "portable.h"
  13 #include "xbt/misc.h"
  14 #include "xbt/sysdep.h"
  15 #include "xbt/ex.h"
  16 #include "gras/Transport/transport_private.h"
  17 #include "gras/Msg/msg_interface.h"     /* listener_close_socket */
  18
  19 /* FIXME maybe READV is sometime a good thing? */
  20 #undef HAVE_READV
  21
  22 #ifdef HAVE_READV
  23 #include <sys/uio.h>
  24 #endif
  25
  26 #ifndef MIN
  27 #define MIN(a,b) ((a)<(b)?(a):(b))
  28 #endif
  29
  30 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gras_trp_tcp, gras_trp,
  31                                 "TCP buffered transport");
  32
  33 /***
  34  *** Specific socket part
  35  ***/
  36
  37 typedef struct {
  38   int port;                     /* port on this side */
  39   int peer_port;                /* port on the other side */
  40   char *peer_name;              /* hostname of the other side */
  41   char *peer_proc;              /* process on the other side */
  42 } s_gras_trp_tcp_sock_data_t, *gras_trp_tcp_sock_data_t;
  43
  44 typedef enum { buffering_buf, buffering_iov } buffering_kind;
  45
  46 typedef struct {
  47   int size;
  48   char *data;
  49   int pos;                      /* for receive; not exchanged over the net */
  50 } gras_trp_buf_t;
  51
  52
  53 struct gras_trp_bufdata_ {
  54   int buffsize;
  55   gras_trp_buf_t in_buf;
  56   gras_trp_buf_t out_buf;
  57
  58 #ifdef HAVE_READV
  59   xbt_dynar_t in_buf_v;
  60   xbt_dynar_t out_buf_v;
  61 #endif
  62
  63   buffering_kind in;
  64   buffering_kind out;
  65 };
  66
  67
  68 /*****************************/
  69 /****[ SOCKET MANAGEMENT ]****/
  70 /*****************************/
  71 /* we exchange port number on client side on socket creation,
  72    so we need to be able to talk right now. */
  73 static XBT_INLINE void gras_trp_tcp_send(gras_socket_t sock,
  74                                          const char *data,
  75                                          unsigned long int size);
  76 static int gras_trp_tcp_recv(gras_socket_t sock, char *data,
  77                              unsigned long int size);
  78
  79
  80 static int _gras_tcp_proto_number(void);
  81
  82 static XBT_INLINE
  83 void gras_trp_sock_socket_client(gras_trp_plugin_t ignored,
  84                                  const char *host,
  85                                  int port,
  86                                  /*OUT*/gras_socket_t sock)
  87 {
  88   gras_trp_tcp_sock_data_t sockdata = xbt_new(s_gras_trp_tcp_sock_data_t,1);
  89   sockdata->port = port;
  90   sockdata->peer_proc = NULL;
  91   sockdata->peer_port = port;
  92   sockdata->peer_name = (char *) strdup(host ? host : "localhost");
  93   sock->data = sockdata;
  94
  95   struct sockaddr_in addr;
  96   struct hostent *he;
  97   struct in_addr *haddr;
  98   int size = sock->buf_size;
  99   uint32_t myport = htonl(((gras_trp_procdata_t)
 100                            gras_libdata_by_id
 101                            (gras_trp_libdata_id))->myport);
 102
 103   sock->incoming = 1;           /* TCP sockets are duplex'ed */
 104
 105   sock->sd = socket(AF_INET, SOCK_STREAM, 0);
 106
 107   if (sock->sd < 0) {
 108     THROW1(system_error, 0, "Failed to create socket: %s",
 109            sock_errstr(sock_errno));
 110   }
 111
 112   if (setsockopt
 113       (sock->sd, SOL_SOCKET, SO_RCVBUF, (char *) &size, sizeof(size))
 114       || setsockopt(sock->sd, SOL_SOCKET, SO_SNDBUF, (char *) &size,
 115                     sizeof(size))) {
 116     VERB1("setsockopt failed, cannot set buffer size: %s",
 117           sock_errstr(sock_errno));
 118   }
 119
 120   he = gethostbyname(sockdata->peer_name);
 121   if (he == NULL) {
 122     THROW2(system_error, 0, "Failed to lookup hostname %s: %s",
 123         sockdata->peer_name, sock_errstr(sock_errno));
 124   }
 125
 126   haddr = ((struct in_addr *) (he->h_addr_list)[0]);
 127
 128   memset(&addr, 0, sizeof(struct sockaddr_in));
 129   memcpy(&addr.sin_addr, haddr, sizeof(struct in_addr));
 130   addr.sin_family = AF_INET;
 131   addr.sin_port = htons(sockdata->peer_port);
 132
 133   if (connect(sock->sd, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
 134     tcp_close(sock->sd);
 135     THROW3(system_error, 0,
 136            "Failed to connect socket to %s:%d (%s)",
 137            sockdata->peer_name, sockdata->peer_port, sock_errstr(sock_errno));
 138   }
 139
 140   gras_trp_tcp_send(sock, (char *) &myport, sizeof(uint32_t));
 141   DEBUG1("peerport sent to %d", sockdata->peer_port);
 142
 143   VERB4("Connect to %s:%d (sd=%d, port %d here)",
 144         sockdata->peer_name, sockdata->peer_port, sock->sd, sockdata->port);
 145 }
 146
 147 /**
 148  * gras_trp_sock_socket_server:
 149  *
 150  * Open a socket used to receive messages.
 151  */
 152 static XBT_INLINE
 153 void gras_trp_sock_socket_server(gras_trp_plugin_t ignored,
 154                                  int port,
 155                                  gras_socket_t sock)
 156 {
 157   int size = sock->buf_size;
 158   int on = 1;
 159   struct sockaddr_in server;
 160
 161   gras_trp_tcp_sock_data_t sockdata = xbt_new(s_gras_trp_tcp_sock_data_t,1);
 162   sockdata->port = port;
 163   sockdata->peer_port = -1;
 164   sockdata->peer_name = NULL;
 165   sockdata->peer_proc = NULL;
 166   sock->data=sockdata;
 167
 168   sock->outgoing = 1;           /* TCP => duplex mode */
 169
 170   server.sin_port = htons((u_short) sockdata->port);
 171   server.sin_addr.s_addr = INADDR_ANY;
 172   server.sin_family = AF_INET;
 173   if ((sock->sd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
 174     THROW1(system_error, 0, "Socket allocation failed: %s",
 175            sock_errstr(sock_errno));
 176
 177   if (setsockopt
 178       (sock->sd, SOL_SOCKET, SO_REUSEADDR, (char *) &on, sizeof(on)))
 179     THROW1(system_error, 0,
 180            "setsockopt failed, cannot condition the socket: %s",
 181            sock_errstr(sock_errno));
 182
 183   if (setsockopt(sock->sd, SOL_SOCKET, SO_RCVBUF,
 184                  (char *) &size, sizeof(size))
 185       || setsockopt(sock->sd, SOL_SOCKET, SO_SNDBUF,
 186                     (char *) &size, sizeof(size))) {
 187     VERB1("setsockopt failed, cannot set buffer size: %s",
 188           sock_errstr(sock_errno));
 189   }
 190
 191   if (bind(sock->sd, (struct sockaddr *) &server, sizeof(server)) == -1) {
 192     tcp_close(sock->sd);
 193     THROW2(system_error, 0,
 194            "Cannot bind to port %d: %s", sockdata->port,
 195            sock_errstr(sock_errno));
 196   }
 197
 198   DEBUG2("Listen on port %d (sd=%d)", sockdata->port, sock->sd);
 199   if (listen(sock->sd, 5) < 0) {
 200     tcp_close(sock->sd);
 201     THROW2(system_error, 0,
 202            "Cannot listen on port %d: %s",
 203            sockdata->port, sock_errstr(sock_errno));
 204   }
 205
 206   VERB2("Openned a server socket on port %d (sd=%d)", sockdata->port,
 207         sock->sd);
 208 }
 209
 210 static gras_socket_t gras_trp_sock_socket_accept(gras_socket_t sock)
 211 {
 212   gras_socket_t res;
 213
 214   struct sockaddr_in peer_in;
 215   socklen_t peer_in_len = sizeof(peer_in);
 216
 217   int sd;
 218   int tmp_errno;
 219   int size;
 220
 221   int i = 1;
 222   socklen_t s = sizeof(int);
 223
 224   uint32_t hisport;
 225
 226   XBT_IN;
 227   gras_trp_socket_new(1, &res);
 228
 229   sd = accept(sock->sd, (struct sockaddr *) &peer_in, &peer_in_len);
 230   tmp_errno = sock_errno;
 231
 232   if (sd == -1) {
 233     gras_socket_close(sock);
 234     THROW1(system_error, 0,
 235            "Accept failed (%s). Droping server socket.",
 236            sock_errstr(tmp_errno));
 237   }
 238
 239   if (setsockopt(sd, SOL_SOCKET, SO_KEEPALIVE, (char *) &i, s)
 240       || setsockopt(sd, _gras_tcp_proto_number(), TCP_NODELAY, (char *) &i,
 241                     s))
 242     THROW1(system_error, 0,
 243            "setsockopt failed, cannot condition the socket: %s",
 244            sock_errstr(tmp_errno));
 245
 246   res->buf_size = sock->buf_size;
 247   size = sock->buf_size;
 248   if (setsockopt(sd, SOL_SOCKET, SO_RCVBUF, (char *) &size, sizeof(size))
 249       || setsockopt(sd, SOL_SOCKET, SO_SNDBUF, (char *) &size,
 250                     sizeof(size)))
 251     VERB1("setsockopt failed, cannot set buffer size: %s",
 252           sock_errstr(tmp_errno));
 253
 254   res->plugin = sock->plugin;
 255   res->incoming = sock->incoming;
 256   res->outgoing = sock->outgoing;
 257   res->accepting = 0;
 258   res->sd = sd;
 259   gras_trp_tcp_sock_data_t sockdata = xbt_new(s_gras_trp_tcp_sock_data_t,1);
 260   sockdata->port = -1;
 261   res->data=sockdata;
 262
 263
 264   gras_trp_tcp_recv(res, (char *) &hisport, sizeof(hisport));
 265   sockdata->peer_port = ntohl(hisport);
 266   DEBUG1("peerport %d received", sockdata->peer_port);
 267
 268   /* FIXME: Lock to protect inet_ntoa */
 269   if (((struct sockaddr *) &peer_in)->sa_family != AF_INET) {
 270     sockdata->peer_name = (char *) strdup("unknown");
 271   } else {
 272     struct in_addr addrAsInAddr;
 273     char *tmp;
 274
 275     addrAsInAddr.s_addr = peer_in.sin_addr.s_addr;
 276
 277     tmp = inet_ntoa(addrAsInAddr);
 278     if (tmp != NULL) {
 279       sockdata->peer_name = (char *) strdup(tmp);
 280     } else {
 281       sockdata->peer_name = (char *) strdup("unknown");
 282     }
 283   }
 284
 285   VERB3("Accepted from %s:%d (sd=%d)", sockdata->peer_name, sockdata->peer_port, sd);
 286   xbt_dynar_push(((gras_trp_procdata_t)
 287                   gras_libdata_by_id(gras_trp_libdata_id))->sockets, &res);
 288
 289   XBT_OUT;
 290   return res;
 291 }
 292
 293 static void gras_trp_sock_socket_close(gras_socket_t sock)
 294 {
 295
 296   if (!sock)
 297     return;                     /* close only once */
 298
 299   if (((gras_trp_tcp_sock_data_t)sock->data)->peer_name)
 300     free(((gras_trp_tcp_sock_data_t)sock->data)->peer_name);
 301   free(sock->data);
 302
 303   VERB1("close tcp connection %d", sock->sd);
 304
 305   /* ask the listener to close the socket */
 306   gras_msg_listener_close_socket(sock->sd);
 307 }
 308
 309 /************************************/
 310 /****[ end of SOCKET MANAGEMENT ]****/
 311 /************************************/
 312
 313
 314 /************************************/
 315 /****[ UNBUFFERED DATA EXCHANGE ]****/
 316 /************************************/
 317 /* Temptation to merge this with file data exchange is great,
 318    but doesn't work on BillWare (see tcp_write() in portable.h) */
 319 static XBT_INLINE void gras_trp_tcp_send(gras_socket_t sock,
 320                                          const char *data,
 321                                          unsigned long int size)
 322 {
 323
 324   while (size) {
 325     int status = 0;
 326
 327     status = tcp_write(sock->sd, data, (size_t) size);
 328     DEBUG3("write(%d, %p, %ld);", sock->sd, data, size);
 329
 330     if (status < 0) {
 331 #ifdef EWOULDBLOCK
 332       if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)
 333 #else
 334       if (errno == EINTR || errno == EAGAIN)
 335 #endif
 336         continue;
 337
 338       THROW4(system_error, 0, "write(%d,%p,%ld) failed: %s",
 339              sock->sd, data, size, sock_errstr(sock_errno));
 340     }
 341
 342     if (status) {
 343       size -= status;
 344       data += status;
 345     } else {
 346       THROW1(system_error, 0, "file descriptor closed (%s)",
 347              sock_errstr(sock_errno));
 348     }
 349   }
 350 }
 351
 352 static XBT_INLINE int
 353 gras_trp_tcp_recv_withbuffer(gras_socket_t sock,
 354                              char *data,
 355                              unsigned long int size,
 356                              unsigned long int bufsize)
 357 {
 358
 359   int got = 0;
 360
 361   if (sock->recvd) {
 362     data[0] = sock->recvd_val;
 363     sock->recvd = 0;
 364     got++;
 365     bufsize--;
 366   }
 367
 368   while (size > got) {
 369     int status = 0;
 370
 371     DEBUG5("read(%d, %p, %ld) got %d so far (%s)",
 372            sock->sd, data + got, bufsize, got,
 373            hexa_str((unsigned char *) data, got, 0));
 374     status = tcp_read(sock->sd, data + got, (size_t) bufsize);
 375
 376     if (status < 0) {
 377       THROW7(system_error, 0,
 378              "read(%d,%p,%d) from %s:%d failed: %s; got %d so far",
 379              sock->sd, data + got, (int) size, gras_socket_peer_name(sock),
 380              gras_socket_peer_port(sock), sock_errstr(sock_errno), got);
 381     }
 382     DEBUG2("Got %d more bytes (%s)", status,
 383            hexa_str((unsigned char *) data + got, status, 0));
 384
 385     if (status) {
 386       bufsize -= status;
 387       got += status;
 388     } else {
 389       THROW1(system_error, errno,
 390              "Socket closed by remote side (got %d bytes before this)",
 391              got);
 392     }
 393   }
 394
 395   return got;
 396 }
 397
 398 static int gras_trp_tcp_recv(gras_socket_t sock,
 399                              char *data, unsigned long int size)
 400 {
 401   return gras_trp_tcp_recv_withbuffer(sock, data, size, size);
 402
 403 }
 404
 405 /*******************************************/
 406 /****[ end of UNBUFFERED DATA EXCHANGE ]****/
 407 /*******************************************/
 408
 409 /**********************************/
 410 /****[ BUFFERED DATA EXCHANGE ]****/
 411 /**********************************/
 412
 413 /* Make sure the data is sent */
 414 static void gras_trp_bufiov_flush(gras_socket_t sock)
 415 {
 416 #ifdef HAVE_READV
 417   xbt_dynar_t vect;
 418   int size;
 419 #endif
 420   gras_trp_bufdata_t *data = sock->bufdata;
 421   XBT_IN;
 422
 423   DEBUG0("Flush");
 424   if (data->out == buffering_buf) {
 425     if (XBT_LOG_ISENABLED(gras_trp_tcp, xbt_log_priority_debug))
 426       hexa_print("chunk to send ",
 427                  (unsigned char *) data->out_buf.data, data->out_buf.size);
 428     if ((data->out_buf.size - data->out_buf.pos) != 0) {
 429       DEBUG3("Send the chunk (size=%d) to %s:%d", data->out_buf.size,
 430              gras_socket_peer_name(sock), gras_socket_peer_port(sock));
 431       gras_trp_tcp_send(sock, data->out_buf.data, data->out_buf.size);
 432       VERB1("Chunk sent (size=%d)", data->out_buf.size);
 433       data->out_buf.size = 0;
 434     }
 435   }
 436 #ifdef HAVE_READV
 437   if (data->out == buffering_iov) {
 438     DEBUG0("Flush out iov");
 439     vect = sock->bufdata->out_buf_v;
 440     if ((size = xbt_dynar_length(vect))) {
 441       DEBUG1("Flush %d chunks out of this socket", size);
 442       writev(sock->sd, xbt_dynar_get_ptr(vect, 0), size);
 443       xbt_dynar_reset(vect);
 444     }
 445     data->out_buf.size = 0;     /* reset the buffer containing non-stable data */
 446   }
 447
 448   if (data->in == buffering_iov) {
 449     DEBUG0("Flush in iov");
 450     vect = sock->bufdata->in_buf_v;
 451     if ((size = xbt_dynar_length(vect))) {
 452       DEBUG1("Get %d chunks from of this socket", size);
 453       readv(sock->sd, xbt_dynar_get_ptr(vect, 0), size);
 454       xbt_dynar_reset(vect);
 455     }
 456   }
 457 #endif
 458 }
 459
 460 static void
 461 gras_trp_buf_send(gras_socket_t sock,
 462                   const char *chunk,
 463                   unsigned long int size, int stable_ignored)
 464 {
 465
 466   gras_trp_bufdata_t *data = (gras_trp_bufdata_t *) sock->bufdata;
 467   int chunk_pos = 0;
 468
 469   XBT_IN;
 470
 471   while (chunk_pos < size) {
 472     /* size of the chunk to receive in that shot */
 473     long int thissize =
 474         min(size - chunk_pos, data->buffsize - data->out_buf.size);
 475     DEBUG4("Set the chars %d..%ld into the buffer; size=%ld, ctn=(%s)",
 476            (int) data->out_buf.size,
 477            ((int) data->out_buf.size) + thissize - 1, size,
 478            hexa_str((unsigned char *) chunk, thissize, 0));
 479
 480     memcpy(data->out_buf.data + data->out_buf.size, chunk + chunk_pos,
 481            thissize);
 482
 483     data->out_buf.size += thissize;
 484     chunk_pos += thissize;
 485     DEBUG4("New pos = %d; Still to send = %ld of %ld; ctn sofar=(%s)",
 486            data->out_buf.size, size - chunk_pos, size,
 487            hexa_str((unsigned char *) chunk, chunk_pos, 0));
 488
 489     if (data->out_buf.size == data->buffsize)   /* out of space. Flush it */
 490       gras_trp_bufiov_flush(sock);
 491   }
 492
 493   XBT_OUT;
 494 }
 495
 496 static int
 497 gras_trp_buf_recv(gras_socket_t sock, char *chunk, unsigned long int size)
 498 {
 499
 500   gras_trp_bufdata_t *data = sock->bufdata;
 501   long int chunk_pos = 0;
 502
 503   XBT_IN;
 504
 505   while (chunk_pos < size) {
 506     /* size of the chunk to receive in that shot */
 507     long int thissize;
 508
 509     if (data->in_buf.size == data->in_buf.pos) {        /* out of data. Get more */
 510
 511       DEBUG2("Get more data (size=%d,bufsize=%d)",
 512              (int) MIN(size - chunk_pos, data->buffsize),
 513              (int) data->buffsize);
 514
 515
 516       data->in_buf.size =
 517           gras_trp_tcp_recv_withbuffer(sock, data->in_buf.data,
 518                                        MIN(size - chunk_pos,
 519                                            data->buffsize),
 520                                        data->buffsize);
 521
 522       data->in_buf.pos = 0;
 523     }
 524
 525     thissize = min(size - chunk_pos, data->in_buf.size - data->in_buf.pos);
 526     memcpy(chunk + chunk_pos, data->in_buf.data + data->in_buf.pos,
 527            thissize);
 528
 529     data->in_buf.pos += thissize;
 530     chunk_pos += thissize;
 531     DEBUG4("New pos = %d; Still to receive = %ld of %ld. Ctn so far=(%s)",
 532            data->in_buf.pos, size - chunk_pos, size,
 533            hexa_str((unsigned char *) chunk, chunk_pos, 0));
 534   }
 535   /* indicate on need to the gras_select function that there is more to read on this socket so that it does not actually select */
 536   sock->moredata = (data->in_buf.size > data->in_buf.pos);
 537   DEBUG1("There is %smore data", (sock->moredata ? "" : "no "));
 538
 539   XBT_OUT;
 540   return chunk_pos;
 541 }
 542
 543 /*****************************************/
 544 /****[ end of BUFFERED DATA EXCHANGE ]****/
 545 /*****************************************/
 546
 547 /********************************/
 548 /****[ VECTOR DATA EXCHANGE ]****/
 549 /********************************/
 550 #ifdef HAVE_READV
 551 static void
 552 gras_trp_iov_send(gras_socket_t sock,
 553                   const char *chunk, unsigned long int size, int stable)
 554 {
 555   struct iovec elm;
 556   gras_trp_bufdata_t *data = (gras_trp_bufdata_t *) sock->bufdata;
 557
 558
 559   DEBUG1("Buffer one chunk to be sent later (%s)",
 560          hexa_str((char *) chunk, size, 0));
 561
 562   elm.iov_len = (size_t) size;
 563
 564   if (!stable) {
 565     /* data storage won't last until flush. Save it in a buffer if we can */
 566
 567     if (size > data->buffsize - data->out_buf.size) {
 568       /* buffer too small:
 569          flush the socket, using data in its actual storage */
 570       elm.iov_base = (void *) chunk;
 571       xbt_dynar_push(data->out_buf_v, &elm);
 572
 573       gras_trp_bufiov_flush(sock);
 574       return;
 575     } else {
 576       /* buffer big enough:
 577          copy data into it, and chain it for upcoming writev */
 578       memcpy(data->out_buf.data + data->out_buf.size, chunk, size);
 579       elm.iov_base = (void *) (data->out_buf.data + data->out_buf.size);
 580       data->out_buf.size += size;
 581
 582       xbt_dynar_push(data->out_buf_v, &elm);
 583     }
 584
 585   } else {
 586     /* data storage stable. Chain it */
 587
 588     elm.iov_base = (void *) chunk;
 589     xbt_dynar_push(data->out_buf_v, &elm);
 590   }
 591 }
 592
 593 static int
 594 gras_trp_iov_recv(gras_socket_t sock, char *chunk, unsigned long int size)
 595 {
 596   struct iovec elm;
 597
 598   DEBUG0("Buffer one chunk to be received later");
 599   elm.iov_base = (void *) chunk;
 600   elm.iov_len = (size_t) size;
 601   xbt_dynar_push(sock->bufdata->in_buf_v, &elm);
 602
 603   return size;
 604 }
 605
 606 #endif
 607 /***************************************/
 608 /****[ end of VECTOR DATA EXCHANGE ]****/
 609 /***************************************/
 610
 611
 612 /***
 613  *** Prototypes of BUFFERED
 614  ***/
 615
 616 void gras_trp_buf_socket_client(gras_trp_plugin_t self,
 617     const char *host,
 618     int port,
 619                                 gras_socket_t sock);
 620 void gras_trp_buf_socket_server(gras_trp_plugin_t self,
 621                                 int port,
 622                                 gras_socket_t sock);
 623 gras_socket_t gras_trp_buf_socket_accept(gras_socket_t sock);
 624
 625 void gras_trp_buf_socket_close(gras_socket_t sd);
 626
 627
 628 gras_socket_t gras_trp_buf_init_sock(gras_socket_t sock)
 629 {
 630   gras_trp_bufdata_t *data = xbt_new(gras_trp_bufdata_t, 1);
 631
 632   data->buffsize = 100 * 1024;  /* 100k */
 633
 634   data->in_buf.size = 0;
 635   data->in_buf.data = xbt_malloc(data->buffsize);
 636   data->in_buf.pos = 0;         /* useless, indeed, since size==pos */
 637
 638   data->out_buf.size = 0;
 639   data->out_buf.data = xbt_malloc(data->buffsize);
 640   data->out_buf.pos = data->out_buf.size;
 641
 642 #ifdef HAVE_READV
 643   data->in_buf_v = data->out_buf_v = NULL;
 644   data->in_buf_v = xbt_dynar_new(sizeof(struct iovec), NULL);
 645   data->out_buf_v = xbt_dynar_new(sizeof(struct iovec), NULL);
 646   data->out = buffering_iov;
 647 #else
 648   data->out = buffering_buf;
 649 #endif
 650
 651   data->in = buffering_buf;
 652
 653   sock->bufdata = data;
 654   return sock;
 655 }
 656
 657 /***
 658  *** Info about who's speaking
 659  ***/
 660 static int gras_trp_tcp_my_port(gras_socket_t s) {
 661   gras_trp_tcp_sock_data_t sockdata = s->data;
 662   return sockdata->port;
 663 }
 664 static int gras_trp_tcp_peer_port(gras_socket_t s) {
 665   gras_trp_tcp_sock_data_t sockdata = s->data;
 666   return sockdata->peer_port;
 667 }
 668 static const char* gras_trp_tcp_peer_name(gras_socket_t s) {
 669   gras_trp_tcp_sock_data_t sockdata = s->data;
 670   return sockdata->peer_name;
 671 }
 672 static const char* gras_trp_tcp_peer_proc(gras_socket_t s) {
 673   gras_trp_tcp_sock_data_t sockdata = s->data;
 674   return sockdata->peer_proc;
 675 }
 676 static void gras_trp_tcp_peer_proc_set(gras_socket_t s,char *name) {
 677   gras_trp_tcp_sock_data_t sockdata = s->data;
 678   sockdata->peer_proc = xbt_strdup(name);
 679 }
 680
 681 /***
 682  *** Code
 683  ***/
 684 void gras_trp_tcp_setup(gras_trp_plugin_t plug)
 685 {
 686
 687   plug->my_port = gras_trp_tcp_my_port;
 688   plug->peer_port = gras_trp_tcp_peer_port;
 689   plug->peer_name = gras_trp_tcp_peer_name;
 690   plug->peer_proc = gras_trp_tcp_peer_proc;
 691   plug->peer_proc_set = gras_trp_tcp_peer_proc_set;
 692
 693
 694   plug->socket_client = gras_trp_buf_socket_client;
 695   plug->socket_server = gras_trp_buf_socket_server;
 696   plug->socket_accept = gras_trp_buf_socket_accept;
 697   plug->socket_close = gras_trp_buf_socket_close;
 698
 699 #ifdef HAVE_READV
 700   plug->send = gras_trp_iov_send;
 701 #else
 702   plug->send = gras_trp_buf_send;
 703 #endif
 704   plug->recv = gras_trp_buf_recv;
 705
 706   plug->raw_send = gras_trp_tcp_send;
 707   plug->raw_recv = gras_trp_tcp_recv;
 708
 709   plug->flush = gras_trp_bufiov_flush;
 710
 711   plug->data = NULL;
 712   plug->exit = NULL;
 713 }
 714
 715 void gras_trp_buf_socket_client(gras_trp_plugin_t self,
 716     const char *host,
 717     int port,
 718                                 /* OUT */ gras_socket_t sock)
 719 {
 720
 721   gras_trp_sock_socket_client(NULL, host,port,sock);
 722   gras_trp_buf_init_sock(sock);
 723 }
 724
 725 /**
 726  * gras_trp_buf_socket_server:
 727  *
 728  * Open a socket used to receive messages.
 729  */
 730 void gras_trp_buf_socket_server(gras_trp_plugin_t self,
 731       int port,
 732                                 /* OUT */ gras_socket_t sock)
 733 {
 734
 735   gras_trp_sock_socket_server(NULL, port, sock);
 736   gras_trp_buf_init_sock(sock);
 737 }
 738
 739 gras_socket_t gras_trp_buf_socket_accept(gras_socket_t sock)
 740 {
 741   return gras_trp_buf_init_sock(gras_trp_sock_socket_accept(sock));
 742 }
 743
 744 void gras_trp_buf_socket_close(gras_socket_t sock)
 745 {
 746   gras_trp_bufdata_t *data = sock->bufdata;
 747
 748   if (data->in_buf.size != data->in_buf.pos) {
 749     WARN3("Socket closed, but %d bytes were unread (size=%d,pos=%d)",
 750           data->in_buf.size - data->in_buf.pos,
 751           data->in_buf.size, data->in_buf.pos);
 752   }
 753   if (data->in_buf.data)
 754     free(data->in_buf.data);
 755
 756   if (data->out_buf.size != data->out_buf.pos) {
 757     DEBUG2("Flush the socket before closing (in=%d,out=%d)",
 758            data->in_buf.size, data->out_buf.size);
 759     gras_trp_bufiov_flush(sock);
 760   }
 761   if (data->out_buf.data)
 762     free(data->out_buf.data);
 763
 764 #ifdef HAVE_READV
 765   if (data->in_buf_v) {
 766     if (xbt_dynar_length(data->in_buf_v))
 767       WARN0("Socket closed, but some bytes were unread");
 768     xbt_dynar_free(&data->in_buf_v);
 769   }
 770   if (data->out_buf_v) {
 771     if (xbt_dynar_length(data->out_buf_v)) {
 772       DEBUG0("Flush the socket before closing");
 773       gras_trp_bufiov_flush(sock);
 774     }
 775     xbt_dynar_free(&data->out_buf_v);
 776   }
 777 #endif
 778
 779   free(data);
 780   gras_trp_sock_socket_close(sock);
 781 }
 782
 783 /****************************/
 784 /****[ HELPER FUNCTIONS ]****/
 785 /****************************/
 786
 787 /*
 788  * Returns the tcp protocol number from the network protocol data base.
 789  *
 790  * getprotobyname() is not thread safe. We need to lock it.
 791  */
 792 static int _gras_tcp_proto_number(void)
 793 {
 794   struct protoent *fetchedEntry;
 795   static int returnValue = 0;
 796
 797   if (returnValue == 0) {
 798     fetchedEntry = getprotobyname("tcp");
 799     xbt_assert0(fetchedEntry, "getprotobyname(tcp) gave NULL");
 800     returnValue = fetchedEntry->p_proto;
 801   }
 802
 803   return returnValue;
 804 }
 805
 806 #ifdef HAVE_WINSOCK_H
 807 #define RETSTR( x ) case x: return #x
 808
 809 const char *gras_wsa_err2string(int err)
 810 {
 811   switch (err) {
 812     RETSTR(WSAEINTR);
 813     RETSTR(WSAEBADF);
 814     RETSTR(WSAEACCES);
 815     RETSTR(WSAEFAULT);
 816     RETSTR(WSAEINVAL);
 817     RETSTR(WSAEMFILE);
 818     RETSTR(WSAEWOULDBLOCK);
 819     RETSTR(WSAEINPROGRESS);
 820     RETSTR(WSAEALREADY);
 821     RETSTR(WSAENOTSOCK);
 822     RETSTR(WSAEDESTADDRREQ);
 823     RETSTR(WSAEMSGSIZE);
 824     RETSTR(WSAEPROTOTYPE);
 825     RETSTR(WSAENOPROTOOPT);
 826     RETSTR(WSAEPROTONOSUPPORT);
 827     RETSTR(WSAESOCKTNOSUPPORT);
 828     RETSTR(WSAEOPNOTSUPP);
 829     RETSTR(WSAEPFNOSUPPORT);
 830     RETSTR(WSAEAFNOSUPPORT);
 831     RETSTR(WSAEADDRINUSE);
 832     RETSTR(WSAEADDRNOTAVAIL);
 833     RETSTR(WSAENETDOWN);
 834     RETSTR(WSAENETUNREACH);
 835     RETSTR(WSAENETRESET);
 836     RETSTR(WSAECONNABORTED);
 837     RETSTR(WSAECONNRESET);
 838     RETSTR(WSAENOBUFS);
 839     RETSTR(WSAEISCONN);
 840     RETSTR(WSAENOTCONN);
 841     RETSTR(WSAESHUTDOWN);
 842     RETSTR(WSAETOOMANYREFS);
 843     RETSTR(WSAETIMEDOUT);
 844     RETSTR(WSAECONNREFUSED);
 845     RETSTR(WSAELOOP);
 846     RETSTR(WSAENAMETOOLONG);
 847     RETSTR(WSAEHOSTDOWN);
 848     RETSTR(WSAEHOSTUNREACH);
 849     RETSTR(WSAENOTEMPTY);
 850     RETSTR(WSAEPROCLIM);
 851     RETSTR(WSAEUSERS);
 852     RETSTR(WSAEDQUOT);
 853     RETSTR(WSAESTALE);
 854     RETSTR(WSAEREMOTE);
 855     RETSTR(WSASYSNOTREADY);
 856     RETSTR(WSAVERNOTSUPPORTED);
 857     RETSTR(WSANOTINITIALISED);
 858     RETSTR(WSAEDISCON);
 859
 860 #ifdef HAVE_WINSOCK2
 861     RETSTR(WSAENOMORE);
 862     RETSTR(WSAECANCELLED);
 863     RETSTR(WSAEINVALIDPROCTABLE);
 864     RETSTR(WSAEINVALIDPROVIDER);
 865     RETSTR(WSASYSCALLFAILURE);
 866     RETSTR(WSASERVICE_NOT_FOUND);
 867     RETSTR(WSATYPE_NOT_FOUND);
 868     RETSTR(WSA_E_NO_MORE);
 869     RETSTR(WSA_E_CANCELLED);
 870     RETSTR(WSAEREFUSED);
 871 #endif                          /* HAVE_WINSOCK2 */
 872
 873     RETSTR(WSAHOST_NOT_FOUND);
 874     RETSTR(WSATRY_AGAIN);
 875     RETSTR(WSANO_RECOVERY);
 876     RETSTR(WSANO_DATA);
 877   }
 878   return "unknown WSA error";
 879 }
 880 #endif                          /* HAVE_WINSOCK_H */
 881
 882 /***********************************/
 883 /****[ end of HELPER FUNCTIONS ]****/
 884 /***********************************/