src/xbt/xbt_trp_plugin_tcp.c

   1 /* buffered transport using the TCP plugin or a custom 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 "xbt/xbt_socket_private.h"
  17
  18 #include "gras/Transport/transport_interface.h" /* FIXME */
  19 #include "gras/Virtu/virtu_interface.h" /* FIXME */
  20 #include "gras/Msg/msg_interface.h" /* FIXME */
  21 extern int gras_trp_libdata_id; /* FIXME */
  22
  23 /* FIXME maybe READV is sometime a good thing? */
  24 #undef HAVE_READV
  25
  26 #ifdef HAVE_READV
  27 #include <sys/uio.h>
  28 #endif
  29
  30 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(xbt_trp_tcp, xbt_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_xbt_trp_tcp_sock_data_t, *xbt_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 } xbt_trp_buf_t;
  51
  52 struct s_xbt_trp_bufdata {
  53   int buffsize;
  54   xbt_trp_buf_t in_buf;
  55   xbt_trp_buf_t out_buf;
  56
  57 #ifdef HAVE_READV
  58   xbt_dynar_t in_buf_v;
  59   xbt_dynar_t out_buf_v;
  60 #endif
  61
  62   buffering_kind in;
  63   buffering_kind out;
  64 };
  65
  66 /*****************************/
  67 /****[ SOCKET MANAGEMENT ]****/
  68 /*****************************/
  69
  70 /* we exchange port number on client side on socket creation,
  71    so we need to be able to talk right now. */
  72 static XBT_INLINE void xbt_trp_tcp_send(xbt_socket_t sock,
  73                                         const char *data,
  74                                         unsigned long int size);
  75 static int xbt_trp_tcp_recv(xbt_socket_t sock, char *data,
  76                             unsigned long int size);
  77
  78 static int xbt_trp_tcp_proto_number(void);
  79 static void xbt_trp_sock_socket_close(xbt_socket_t sock);
  80
  81 static XBT_INLINE
  82 void xbt_trp_sock_socket_client(xbt_trp_plugin_t ignored,
  83                                 const char *host,
  84                                 int port,
  85                                 /* OUT */ xbt_socket_t sock)
  86 {
  87   xbt_trp_tcp_sock_data_t sockdata = xbt_new(s_xbt_trp_tcp_sock_data_t,1);
  88   sockdata->port = port;
  89   sockdata->peer_proc = NULL;
  90   sockdata->peer_port = port;
  91   sockdata->peer_name = (char *) xbt_strdup(host ? host : "localhost");
  92   sock->data = sockdata;
  93
  94   struct sockaddr_in addr;
  95   struct hostent *he;
  96   struct in_addr *haddr;
  97   int size = sock->buf_size;
  98   uint32_t myport = htonl(((gras_trp_procdata_t)
  99                            gras_libdata_by_id
 100                            (gras_trp_libdata_id))->myport);
 101
 102   sock->incoming = 1;           /* TCP sockets are duplex'ed */
 103
 104   sock->sd = socket(AF_INET, SOCK_STREAM, 0);
 105
 106   if (sock->sd < 0) {
 107     THROWF(system_error, 0, "Failed to create socket: %s",
 108            sock_errstr(sock_errno));
 109   }
 110
 111   if (setsockopt
 112       (sock->sd, SOL_SOCKET, SO_RCVBUF, (char *) &size, sizeof(size))
 113       || setsockopt(sock->sd, SOL_SOCKET, SO_SNDBUF, (char *) &size,
 114                     sizeof(size))) {
 115     XBT_VERB("setsockopt failed, cannot set buffer size: %s",
 116           sock_errstr(sock_errno));
 117   }
 118
 119   he = gethostbyname(sockdata->peer_name);
 120   if (he == NULL) {
 121     THROWF(system_error, 0, "Failed to lookup hostname %s: %s",
 122         sockdata->peer_name, sock_errstr(sock_errno));
 123   }
 124
 125   haddr = ((struct in_addr *) (he->h_addr_list)[0]);
 126
 127   memset(&addr, 0, sizeof(struct sockaddr_in));
 128   memcpy(&addr.sin_addr, haddr, sizeof(struct in_addr));
 129   addr.sin_family = AF_INET;
 130   addr.sin_port = htons(sockdata->peer_port);
 131
 132   if (connect(sock->sd, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
 133     tcp_close(sock->sd);
 134     THROWF(system_error, 0,
 135            "Failed to connect socket to %s:%d (%s)",
 136            sockdata->peer_name, sockdata->peer_port, sock_errstr(sock_errno));
 137   }
 138
 139   xbt_trp_tcp_send(sock, (char *) &myport, sizeof(uint32_t));
 140   XBT_DEBUG("peerport sent to %d", sockdata->peer_port);
 141
 142   XBT_VERB("Connect to %s:%d (sd=%d, port %d here)",
 143         sockdata->peer_name, sockdata->peer_port, sock->sd, sockdata->port);
 144 }
 145
 146 /**
 147  * gras_trp_sock_socket_server:
 148  *
 149  * Open a socket used to receive messages.
 150  */
 151 static XBT_INLINE
 152 void xbt_trp_sock_socket_server(xbt_trp_plugin_t ignored,
 153                                 int port,
 154                                 xbt_socket_t sock)
 155 {
 156   int size = sock->buf_size;
 157   int on = 1;
 158   struct sockaddr_in server;
 159
 160   xbt_trp_tcp_sock_data_t sockdata = xbt_new(s_xbt_trp_tcp_sock_data_t,1);
 161   sockdata->port = port;
 162   sockdata->peer_port = -1;
 163   sockdata->peer_name = NULL;
 164   sockdata->peer_proc = NULL;
 165   sock->data = sockdata;
 166
 167   sock->outgoing = 1;           /* TCP => duplex mode */
 168
 169   server.sin_port = htons((u_short) sockdata->port);
 170   server.sin_addr.s_addr = INADDR_ANY;
 171   server.sin_family = AF_INET;
 172   if ((sock->sd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
 173     THROWF(system_error, 0, "Socket allocation failed: %s",
 174            sock_errstr(sock_errno));
 175
 176   if (setsockopt
 177       (sock->sd, SOL_SOCKET, SO_REUSEADDR, (char *) &on, sizeof(on)))
 178     THROWF(system_error, 0,
 179            "setsockopt failed, cannot condition the socket: %s",
 180            sock_errstr(sock_errno));
 181
 182   if (setsockopt(sock->sd, SOL_SOCKET, SO_RCVBUF,
 183                  (char *) &size, sizeof(size))
 184       || setsockopt(sock->sd, SOL_SOCKET, SO_SNDBUF,
 185                     (char *) &size, sizeof(size))) {
 186     XBT_VERB("setsockopt failed, cannot set buffer size: %s",
 187           sock_errstr(sock_errno));
 188   }
 189
 190   if (bind(sock->sd, (struct sockaddr *) &server, sizeof(server)) == -1) {
 191     tcp_close(sock->sd);
 192     THROWF(system_error, 0,
 193            "Cannot bind to port %d: %s", sockdata->port,
 194            sock_errstr(sock_errno));
 195   }
 196
 197   XBT_DEBUG("Listen on port %d (sd=%d)", sockdata->port, sock->sd);
 198   if (listen(sock->sd, 5) < 0) {
 199     tcp_close(sock->sd);
 200     THROWF(system_error, 0,
 201            "Cannot listen on port %d: %s",
 202            sockdata->port, sock_errstr(sock_errno));
 203   }
 204
 205   XBT_VERB("Openned a server socket on port %d (sd=%d)", sockdata->port,
 206         sock->sd);
 207 }
 208
 209 static xbt_socket_t xbt_trp_sock_socket_accept(xbt_socket_t sock)
 210 {
 211   xbt_socket_t res;
 212
 213   struct sockaddr_in peer_in;
 214   socklen_t peer_in_len = sizeof(peer_in);
 215
 216   int sd;
 217   int tmp_errno;
 218   int size;
 219
 220   int i = 1;
 221   socklen_t s = sizeof(int);
 222
 223   uint32_t hisport;
 224
 225   int failed = 0;
 226
 227   XBT_IN();
 228   xbt_socket_new_ext(1, &res, sock->plugin, sock->buf_size, 0);
 229
 230   sd = accept(sock->sd, (struct sockaddr *) &peer_in, &peer_in_len);
 231   tmp_errno = sock_errno;
 232
 233   if (sd == -1) {
 234     xbt_trp_sock_socket_close(sock);
 235     THROWF(system_error, 0,
 236            "Accept failed (%s). Droping server socket.",
 237            sock_errstr(tmp_errno));
 238   }
 239
 240   if (xbt_trp_tcp_proto_number() != -1)
 241     if (setsockopt(sd, xbt_trp_tcp_proto_number(), TCP_NODELAY, (char *) &i,s))
 242       failed = 1;
 243
 244   if (setsockopt(sd, SOL_SOCKET, SO_KEEPALIVE, (char *) &i, s))
 245     failed = 1;
 246
 247   if (failed)
 248     THROWF(system_error, 0,
 249            "setsockopt failed, cannot condition the socket: %s",
 250            sock_errstr(tmp_errno));
 251
 252   size = sock->buf_size;
 253   if (setsockopt(sd, SOL_SOCKET, SO_RCVBUF, (char *) &size, sizeof(size))
 254       || setsockopt(sd, SOL_SOCKET, SO_SNDBUF, (char *) &size,
 255                     sizeof(size)))
 256     XBT_VERB("setsockopt failed, cannot set buffer size: %s",
 257           sock_errstr(tmp_errno));
 258
 259   res->plugin = sock->plugin;
 260   res->incoming = sock->incoming;
 261   res->outgoing = sock->outgoing;
 262   res->accepting = 0;
 263   res->sd = sd;
 264   xbt_trp_tcp_sock_data_t sockdata = xbt_new(s_xbt_trp_tcp_sock_data_t, 1);
 265   sockdata->port = -1;
 266   res->data = sockdata;
 267
 268   xbt_trp_tcp_recv(res, (char *) &hisport, sizeof(hisport));
 269   sockdata->peer_port = ntohl(hisport);
 270   XBT_DEBUG("peerport %d received", sockdata->peer_port);
 271
 272   /* FIXME: Lock to protect inet_ntoa */
 273   if (((struct sockaddr *) &peer_in)->sa_family != AF_INET) {
 274     sockdata->peer_name = (char *) strdup("unknown");
 275   } else {
 276     struct in_addr addrAsInAddr;
 277     char *tmp;
 278
 279     addrAsInAddr.s_addr = peer_in.sin_addr.s_addr;
 280
 281     tmp = inet_ntoa(addrAsInAddr);
 282     if (tmp != NULL) {
 283       sockdata->peer_name = (char *) strdup(tmp);
 284     } else {
 285       sockdata->peer_name = (char *) strdup("unknown");
 286     }
 287   }
 288
 289   XBT_VERB("Accepted from %s:%d (sd=%d)", sockdata->peer_name, sockdata->peer_port, sd);
 290   xbt_dynar_push(((gras_trp_procdata_t)
 291                   gras_libdata_by_id(gras_trp_libdata_id))->sockets, &res);
 292   /* FIXME we depend on GRAS here */
 293
 294   XBT_OUT();
 295   return res;
 296 }
 297
 298 static void xbt_trp_sock_socket_close(xbt_socket_t sock)
 299 {
 300   if (!sock)
 301     return;                     /* close only once */
 302
 303   free(((xbt_trp_tcp_sock_data_t)sock->data)->peer_name);
 304   free(sock->data);
 305
 306   XBT_VERB("close tcp connection %d", sock->sd);
 307
 308   gras_msg_listener_close_socket(sock->sd); /* FIXME */
 309 }
 310
 311 /************************************/
 312 /****[ end of SOCKET MANAGEMENT ]****/
 313 /************************************/
 314
 315
 316 /************************************/
 317 /****[ UNBUFFERED DATA EXCHANGE ]****/
 318 /************************************/
 319 /* Temptation to merge this with file data exchange is great,
 320    but doesn't work on BillWare (see tcp_write() in portable.h) */
 321 static XBT_INLINE void xbt_trp_tcp_send(xbt_socket_t sock,
 322                                         const char *data,
 323                                         unsigned long int size)
 324 {
 325   while (size) {
 326     int status = 0;
 327
 328     status = tcp_write(sock->sd, data, (size_t) size);
 329     XBT_DEBUG("write(%d, %p, %lu);", sock->sd, data, size);
 330
 331     if (status < 0) {
 332 #ifdef EWOULDBLOCK
 333       if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)
 334 #else
 335       if (errno == EINTR || errno == EAGAIN)
 336 #endif
 337         continue;
 338
 339       THROWF(system_error, 0, "write(%d,%p,%lu) failed: %s",
 340              sock->sd, data, size, sock_errstr(sock_errno));
 341     }
 342
 343     if (status) {
 344       size -= status;
 345       data += status;
 346     } else {
 347       THROWF(system_error, 0, "file descriptor closed (%s)",
 348              sock_errstr(sock_errno));
 349     }
 350   }
 351 }
 352
 353 static XBT_INLINE int
 354 xbt_trp_tcp_recv_withbuffer(xbt_socket_t sock,
 355                             char *data,
 356                             unsigned long int size,
 357                             unsigned long int bufsize)
 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     XBT_DEBUG("read(%d, %p, %lu) 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       THROWF(system_error, 0,
 378              "read(%d,%p,%lu) from %s:%d failed: %s; got %d so far",
 379              sock->sd, data + got, size, xbt_socket_peer_name(sock),
 380              xbt_socket_peer_port(sock), sock_errstr(sock_errno), got);
 381     }
 382     XBT_DEBUG("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       THROWF(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 xbt_trp_tcp_recv(xbt_socket_t sock,
 399                             char *data,
 400                             unsigned long int size)
 401 {
 402   return xbt_trp_tcp_recv_withbuffer(sock, data, size, size);
 403
 404 }
 405
 406 /*******************************************/
 407 /****[ end of UNBUFFERED DATA EXCHANGE ]****/
 408 /*******************************************/
 409
 410 /**********************************/
 411 /****[ BUFFERED DATA EXCHANGE ]****/
 412 /**********************************/
 413
 414 /* Make sure the data is sent */
 415 static void xbt_trp_bufiov_flush(xbt_socket_t sock)
 416 {
 417 #ifdef HAVE_READV
 418   xbt_dynar_t vect;
 419   int size;
 420 #endif
 421   xbt_trp_bufdata_t *data = sock->bufdata;
 422   XBT_IN();
 423
 424   XBT_DEBUG("Flush");
 425   if (data->out == buffering_buf) {
 426     if (XBT_LOG_ISENABLED(xbt_trp_tcp, xbt_log_priority_debug))
 427       hexa_print("chunk to send ",
 428                  (unsigned char *) data->out_buf.data, data->out_buf.size);
 429     if ((data->out_buf.size - data->out_buf.pos) != 0) {
 430       XBT_DEBUG("Send the chunk (size=%d) to %s:%d", data->out_buf.size,
 431              xbt_socket_peer_name(sock), xbt_socket_peer_port(sock));
 432       xbt_trp_tcp_send(sock, data->out_buf.data, data->out_buf.size);
 433       XBT_VERB("Chunk sent (size=%d)", data->out_buf.size);
 434       data->out_buf.size = 0;
 435     }
 436   }
 437 #ifdef HAVE_READV
 438   if (data->out == buffering_iov) {
 439     XBT_DEBUG("Flush out iov");
 440     vect = sock->bufdata->out_buf_v;
 441     if ((size = xbt_dynar_length(vect))) {
 442       XBT_DEBUG("Flush %d chunks out of this socket", size);
 443       writev(sock->sd, xbt_dynar_get_ptr(vect, 0), size);
 444       xbt_dynar_reset(vect);
 445     }
 446     data->out_buf.size = 0;     /* reset the buffer containing non-stable data */
 447   }
 448
 449   if (data->in == buffering_iov) {
 450     XBT_DEBUG("Flush in iov");
 451     vect = sock->bufdata->in_buf_v;
 452     if ((size = xbt_dynar_length(vect))) {
 453       XBT_DEBUG("Get %d chunks from of this socket", size);
 454       readv(sock->sd, xbt_dynar_get_ptr(vect, 0), size);
 455       xbt_dynar_reset(vect);
 456     }
 457   }
 458 #endif
 459 }
 460
 461 static void
 462 xbt_trp_buf_send(xbt_socket_t sock,
 463                  const char *chunk,
 464                  unsigned long int size,
 465                  int stable_ignored)
 466 {
 467
 468   xbt_trp_bufdata_t *data = (xbt_trp_bufdata_t *) sock->bufdata;
 469   int chunk_pos = 0;
 470
 471   XBT_IN();
 472
 473   while (chunk_pos < size) {
 474     /* size of the chunk to receive in that shot */
 475     long int thissize =
 476         min(size - chunk_pos, data->buffsize - data->out_buf.size);
 477     XBT_DEBUG("Set the chars %d..%ld into the buffer; size=%lu, ctn=(%s)",
 478            data->out_buf.size,
 479            data->out_buf.size + thissize - 1, size,
 480            hexa_str((unsigned char *) chunk, thissize, 0));
 481
 482     memcpy(data->out_buf.data + data->out_buf.size, chunk + chunk_pos,
 483            thissize);
 484
 485     data->out_buf.size += thissize;
 486     chunk_pos += thissize;
 487     XBT_DEBUG("New pos = %d; Still to send = %lu of %lu; ctn sofar=(%s)",
 488            data->out_buf.size, size - chunk_pos, size,
 489            hexa_str((unsigned char *) chunk, chunk_pos, 0));
 490
 491     if (data->out_buf.size == data->buffsize)   /* out of space. Flush it */
 492       xbt_trp_bufiov_flush(sock);
 493   }
 494
 495   XBT_OUT();
 496 }
 497
 498 static int
 499 xbt_trp_buf_recv(xbt_socket_t sock, char *chunk, unsigned long int size)
 500 {
 501
 502   xbt_trp_bufdata_t *data = sock->bufdata;
 503   long int chunk_pos = 0;
 504
 505   XBT_IN();
 506
 507   while (chunk_pos < size) {
 508     /* size of the chunk to receive in that shot */
 509     long int thissize;
 510
 511     if (data->in_buf.size == data->in_buf.pos) {        /* out of data. Get more */
 512
 513       XBT_DEBUG("Get more data (size=%d,bufsize=%d)",
 514              (int) MIN(size - chunk_pos, data->buffsize),
 515              (int) data->buffsize);
 516
 517       data->in_buf.size =
 518           xbt_trp_tcp_recv_withbuffer(sock, data->in_buf.data,
 519                                       MIN(size - chunk_pos, 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     XBT_DEBUG("New pos = %d; Still to receive = %lu of %lu. 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   XBT_DEBUG("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 xbt_trp_iov_send(xbt_socket_t sock,
 553                  const char *chunk, unsigned long int size, int stable)
 554 {
 555   struct iovec elm;
 556   xbt_trp_bufdata_t *data = (xbt_trp_bufdata_t *) sock->bufdata;
 557
 558
 559   XBT_DEBUG("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       xbt_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 xbt_trp_iov_recv(gras_socket_t sock, char *chunk, unsigned long int size)
 595 {
 596   struct iovec elm;
 597
 598   XBT_DEBUG("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 static xbt_socket_t xbt_trp_buf_init_sock(xbt_socket_t sock);
 616 static void xbt_trp_buf_socket_client(xbt_trp_plugin_t self,
 617                                const char *host,
 618                                int port,
 619                                xbt_socket_t sock);
 620 static void xbt_trp_buf_socket_server(xbt_trp_plugin_t self,
 621                                 int port,
 622                                 xbt_socket_t sock);
 623 static xbt_socket_t xbt_trp_buf_socket_accept(xbt_socket_t sock);
 624
 625 static void xbt_trp_buf_socket_close(xbt_socket_t sd);
 626
 627
 628 static xbt_socket_t xbt_trp_buf_init_sock(xbt_socket_t sock)
 629 {
 630   xbt_trp_bufdata_t *data = xbt_new(xbt_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 xbt_trp_tcp_my_port(xbt_socket_t s) {
 661   xbt_trp_tcp_sock_data_t sockdata = s->data;
 662   return sockdata->port;
 663 }
 664
 665 static int xbt_trp_tcp_peer_port(xbt_socket_t s) {
 666   xbt_trp_tcp_sock_data_t sockdata = s->data;
 667   return sockdata->peer_port;
 668 }
 669
 670 static const char* xbt_trp_tcp_peer_name(xbt_socket_t s) {
 671   xbt_trp_tcp_sock_data_t sockdata = s->data;
 672   return sockdata->peer_name;
 673 }
 674
 675 static const char* xbt_trp_tcp_peer_proc(xbt_socket_t s) {
 676   xbt_trp_tcp_sock_data_t sockdata = s->data;
 677   return sockdata->peer_proc;
 678 }
 679
 680 static void xbt_trp_tcp_peer_proc_set(xbt_socket_t s,char *name) {
 681   xbt_trp_tcp_sock_data_t sockdata = s->data;
 682   sockdata->peer_proc = xbt_strdup(name);
 683 }
 684
 685 /***
 686  *** Code
 687  ***/
 688 void xbt_trp_tcp_setup(xbt_trp_plugin_t plug)
 689 {
 690   plug->my_port = xbt_trp_tcp_my_port;
 691   plug->peer_port = xbt_trp_tcp_peer_port;
 692   plug->peer_name = xbt_trp_tcp_peer_name;
 693   plug->peer_proc = xbt_trp_tcp_peer_proc;
 694   plug->peer_proc_set = xbt_trp_tcp_peer_proc_set;
 695
 696
 697   plug->socket_client = xbt_trp_buf_socket_client;
 698   plug->socket_server = xbt_trp_buf_socket_server;
 699   plug->socket_accept = xbt_trp_buf_socket_accept;
 700   plug->socket_close = xbt_trp_buf_socket_close;
 701
 702 #ifdef HAVE_READV
 703   plug->send = xbt_trp_iov_send;
 704 #else
 705   plug->send = xbt_trp_buf_send;
 706 #endif
 707   plug->recv = xbt_trp_buf_recv;
 708
 709   plug->raw_send = xbt_trp_tcp_send;
 710   plug->raw_recv = xbt_trp_tcp_recv;
 711
 712   plug->flush = xbt_trp_bufiov_flush;
 713
 714   plug->data = NULL;
 715   plug->exit = NULL;
 716 }
 717
 718 static void xbt_trp_buf_socket_client(xbt_trp_plugin_t self,
 719                                const char *host,
 720                                int port,
 721                                /* OUT */ xbt_socket_t sock)
 722 {
 723   xbt_trp_sock_socket_client(NULL, host, port, sock);
 724   xbt_trp_buf_init_sock(sock);
 725 }
 726
 727 /**
 728  * xbt_trp_buf_socket_server:
 729  *
 730  * Open a socket used to receive messages.
 731  */
 732 static void xbt_trp_buf_socket_server(xbt_trp_plugin_t self,
 733                                int port,
 734                                /* OUT */ xbt_socket_t sock)
 735 {
 736   xbt_trp_sock_socket_server(NULL, port, sock);
 737   xbt_trp_buf_init_sock(sock);
 738 }
 739
 740 static xbt_socket_t xbt_trp_buf_socket_accept(xbt_socket_t sock)
 741 {
 742   return xbt_trp_buf_init_sock(xbt_trp_sock_socket_accept(sock));
 743 }
 744
 745 static void xbt_trp_buf_socket_close(xbt_socket_t sock)
 746 {
 747   xbt_trp_bufdata_t *data = sock->bufdata;
 748
 749   if (data->in_buf.size != data->in_buf.pos) {
 750     XBT_WARN("Socket closed, but %d bytes were unread (size=%d,pos=%d)",
 751         data->in_buf.size - data->in_buf.pos,
 752         data->in_buf.size, data->in_buf.pos);
 753   }
 754   free(data->in_buf.data);
 755
 756   if (data->out_buf.size != data->out_buf.pos) {
 757     XBT_DEBUG("Flush the socket before closing (in=%d,out=%d)",
 758            data->in_buf.size, data->out_buf.size);
 759     xbt_trp_bufiov_flush(sock);
 760   }
 761   free(data->out_buf.data);
 762
 763 #ifdef HAVE_READV
 764   if (data->in_buf_v) {
 765     if (!xbt_dynar_is_empty(data->in_buf_v))
 766       XBT_WARN("Socket closed, but some bytes were unread");
 767     xbt_dynar_free(&data->in_buf_v);
 768   }
 769   if (data->out_buf_v) {
 770     if (!xbt_dynar_is_empty(data->out_buf_v)) {
 771       XBT_DEBUG("Flush the socket before closing");
 772       xbt_trp_bufiov_flush(sock);
 773     }
 774     xbt_dynar_free(&data->out_buf_v);
 775   }
 776 #endif
 777
 778   free(data);
 779   xbt_trp_sock_socket_close(sock);
 780 }
 781
 782 /****************************/
 783 /****[ HELPER FUNCTIONS ]****/
 784 /****************************/
 785
 786 /*
 787  * Returns the tcp protocol number from the network protocol data base, or -1 if not found
 788  *
 789  * getprotobyname() is not thread safe. We need to lock it.
 790  */
 791 static int xbt_trp_tcp_proto_number(void)
 792 {
 793   struct protoent *fetchedEntry;
 794   static int returnValue = 0;
 795
 796   if (returnValue == 0) {
 797     fetchedEntry = getprotobyname("tcp");
 798     if (fetchedEntry == NULL) {
 799       XBT_VERB("getprotobyname(tcp) gave NULL");
 800       returnValue = -1;
 801     } else {
 802       returnValue = fetchedEntry->p_proto;
 803     }
 804   }
 805
 806   return returnValue;
 807 }
 808
 809 #ifdef HAVE_WINSOCK_H
 810 #define RETSTR( x ) case x: return #x
 811
 812 const char *xbt_wsa_err2string(int err)
 813 {
 814   switch (err) {
 815     RETSTR(WSAEINTR);
 816     RETSTR(WSAEBADF);
 817     RETSTR(WSAEACCES);
 818     RETSTR(WSAEFAULT);
 819     RETSTR(WSAEINVAL);
 820     RETSTR(WSAEMFILE);
 821     RETSTR(WSAEWOULDBLOCK);
 822     RETSTR(WSAEINPROGRESS);
 823     RETSTR(WSAEALREADY);
 824     RETSTR(WSAENOTSOCK);
 825     RETSTR(WSAEDESTADDRREQ);
 826     RETSTR(WSAEMSGSIZE);
 827     RETSTR(WSAEPROTOTYPE);
 828     RETSTR(WSAENOPROTOOPT);
 829     RETSTR(WSAEPROTONOSUPPORT);
 830     RETSTR(WSAESOCKTNOSUPPORT);
 831     RETSTR(WSAEOPNOTSUPP);
 832     RETSTR(WSAEPFNOSUPPORT);
 833     RETSTR(WSAEAFNOSUPPORT);
 834     RETSTR(WSAEADDRINUSE);
 835     RETSTR(WSAEADDRNOTAVAIL);
 836     RETSTR(WSAENETDOWN);
 837     RETSTR(WSAENETUNREACH);
 838     RETSTR(WSAENETRESET);
 839     RETSTR(WSAECONNABORTED);
 840     RETSTR(WSAECONNRESET);
 841     RETSTR(WSAENOBUFS);
 842     RETSTR(WSAEISCONN);
 843     RETSTR(WSAENOTCONN);
 844     RETSTR(WSAESHUTDOWN);
 845     RETSTR(WSAETOOMANYREFS);
 846     RETSTR(WSAETIMEDOUT);
 847     RETSTR(WSAECONNREFUSED);
 848     RETSTR(WSAELOOP);
 849     RETSTR(WSAENAMETOOLONG);
 850     RETSTR(WSAEHOSTDOWN);
 851     RETSTR(WSAEHOSTUNREACH);
 852     RETSTR(WSAENOTEMPTY);
 853     RETSTR(WSAEPROCLIM);
 854     RETSTR(WSAEUSERS);
 855     RETSTR(WSAEDQUOT);
 856     RETSTR(WSAESTALE);
 857     RETSTR(WSAEREMOTE);
 858     RETSTR(WSASYSNOTREADY);
 859     RETSTR(WSAVERNOTSUPPORTED);
 860     RETSTR(WSANOTINITIALISED);
 861     RETSTR(WSAEDISCON);
 862
 863 #ifdef HAVE_WINSOCK2_H
 864     RETSTR(WSAENOMORE);
 865     RETSTR(WSAECANCELLED);
 866     RETSTR(WSAEINVALIDPROCTABLE);
 867     RETSTR(WSAEINVALIDPROVIDER);
 868     RETSTR(WSASYSCALLFAILURE);
 869     RETSTR(WSASERVICE_NOT_FOUND);
 870     RETSTR(WSATYPE_NOT_FOUND);
 871     RETSTR(WSA_E_NO_MORE);
 872     RETSTR(WSA_E_CANCELLED);
 873     RETSTR(WSAEREFUSED);
 874 #endif                          /* HAVE_WINSOCK2_H */
 875
 876     RETSTR(WSAHOST_NOT_FOUND);
 877     RETSTR(WSATRY_AGAIN);
 878     RETSTR(WSANO_RECOVERY);
 879     RETSTR(WSANO_DATA);
 880   }
 881   return "unknown WSA error";
 882 }
 883 #endif                          /* HAVE_WINSOCK_H */
 884
 885 /***********************************/
 886 /****[ end of HELPER FUNCTIONS ]****/
 887 /***********************************/