Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
cosmetics
[simgrid.git] / src / gras / Transport / transport_plugin_tcp.c
index ef3f63a..ca497c0 100644 (file)
@@ -14,7 +14,7 @@
 #include "xbt/misc.h"
 #include "xbt/sysdep.h"
 #include "xbt/ex.h"
-#include "transport_private.h"
+#include "gras/Transport/transport_private.h"
 
 /* FIXME maybe READV is sometime a good thing? */
 #undef HAVE_READV
@@ -30,7 +30,6 @@
 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gras_trp_tcp,gras_trp,
       "TCP buffered transport");
 
-int _gras_master_socket_port = -1;
 /***
  *** Specific socket part
  ***/
@@ -64,7 +63,7 @@ struct gras_trp_bufdata_{
 /*****************************/
 /* we exchange port number on client side on socket creation,
    so we need to be able to talk right now. */
-static inline void gras_trp_tcp_send(gras_socket_t sock, const char *data,
+static XBT_INLINE void gras_trp_tcp_send(gras_socket_t sock, const char *data,
                                     unsigned long int size);
 static int gras_trp_tcp_recv(gras_socket_t sock, char *data,
                             unsigned long int size);
@@ -72,32 +71,32 @@ static int gras_trp_tcp_recv(gras_socket_t sock, char *data,
 
 static int _gras_tcp_proto_number(void);
 
-static inline void gras_trp_sock_socket_client(gras_trp_plugin_t ignored,
+static XBT_INLINE void gras_trp_sock_socket_client(gras_trp_plugin_t ignored,
                                               gras_socket_t sock){
   
   struct sockaddr_in addr;
   struct hostent *he;
   struct in_addr *haddr;
   int size = sock->buf_size; 
-  uint32_t myport = htonl(_gras_master_socket_port);
+  uint32_t myport = htonl(((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->myport);
 
   sock->incoming = 1; /* TCP sockets are duplex'ed */
 
   sock->sd = socket (AF_INET, SOCK_STREAM, 0);
   
   if (sock->sd < 0) {
-    THROW1(system_error,0, "Failed to create socket: %s", sock_errstr);
+    THROW1(system_error,0, "Failed to create socket: %s", sock_errstr(sock_errno));
   }
 
   if (setsockopt(sock->sd, SOL_SOCKET, SO_RCVBUF, (char *)&size, sizeof(size)) ||
       setsockopt(sock->sd, SOL_SOCKET, SO_SNDBUF, (char *)&size, sizeof(size))) {
-     WARN1("setsockopt failed, cannot set buffer size: %s",sock_errstr);
+     WARN1("setsockopt failed, cannot set buffer size: %s",sock_errstr(sock_errno));
   }
   
   he = gethostbyname (sock->peer_name);
   if (he == NULL) {
     THROW2(system_error,0, "Failed to lookup hostname %s: %s",
-          sock->peer_name, sock_errstr);
+          sock->peer_name, sock_errstr(sock_errno));
   }
   
   haddr = ((struct in_addr *) (he->h_addr_list)[0]);
@@ -111,7 +110,7 @@ static inline void gras_trp_sock_socket_client(gras_trp_plugin_t ignored,
     tcp_close(sock->sd);
     THROW3(system_error,0,
           "Failed to connect socket to %s:%d (%s)",
-          sock->peer_name, sock->peer_port, sock_errstr);
+          sock->peer_name, sock->peer_port, sock_errstr(sock_errno));
   }
 
   gras_trp_tcp_send(sock,(char*)&myport,sizeof(uint32_t));
@@ -126,7 +125,7 @@ static inline void gras_trp_sock_socket_client(gras_trp_plugin_t ignored,
  *
  * Open a socket used to receive messages.
  */
-static inline void gras_trp_sock_socket_server(gras_trp_plugin_t ignored,
+static XBT_INLINE void gras_trp_sock_socket_server(gras_trp_plugin_t ignored,
                                               gras_socket_t sock){
   int size = sock->buf_size; 
   int on = 1;
@@ -134,32 +133,38 @@ static inline void gras_trp_sock_socket_server(gras_trp_plugin_t ignored,
 
   sock->outgoing  = 1; /* TCP => duplex mode */
 
-  _gras_master_socket_port = sock->port;
   server.sin_port = htons((u_short)sock->port);
   server.sin_addr.s_addr = INADDR_ANY;
   server.sin_family = AF_INET;
   if((sock->sd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
-    THROW1(system_error,0,"Socket allocation failed: %s", sock_errstr);
+    THROW1(system_error,0,"Socket allocation failed: %s", sock_errstr(sock_errno));
 
   if (setsockopt(sock->sd, SOL_SOCKET, SO_REUSEADDR, (char *)&on, sizeof(on)))
-     THROW1(system_error,0,"setsockopt failed, cannot condition the socket: %s",
-           sock_errstr);
+     THROW1(system_error,0,
+           "setsockopt failed, cannot condition the socket: %s",
+           sock_errstr(sock_errno));
    
-  if (setsockopt(sock->sd, SOL_SOCKET, SO_RCVBUF, (char *)&size, sizeof(size)) ||
-      setsockopt(sock->sd, SOL_SOCKET, SO_SNDBUF, (char *)&size, sizeof(size))) {
+  if (   setsockopt(sock->sd, SOL_SOCKET, SO_RCVBUF,
+                   (char *)&size, sizeof(size)) 
+      || setsockopt(sock->sd, SOL_SOCKET, SO_SNDBUF, 
+                   (char *)&size, sizeof(size))) {
      WARN1("setsockopt failed, cannot set buffer size: %s",
-          sock_errstr);
+          sock_errstr(sock_errno));
   }
        
   if (bind(sock->sd, (struct sockaddr *)&server, sizeof(server)) == -1) {
     tcp_close(sock->sd);
-    THROW2(system_error,0,"Cannot bind to port %d: %s",sock->port, sock_errstr);
+    THROW2(system_error,0,
+          "Cannot bind to port %d: %s",
+          sock->port, sock_errstr(sock_errno));
   }
 
   DEBUG2("Listen on port %d (sd=%d)",sock->port, sock->sd);
   if (listen(sock->sd, 5) < 0) {
     tcp_close(sock->sd);
-    THROW2(system_error,0,"Cannot listen on port %d: %s",sock->port,sock_errstr);
+    THROW2(system_error,0,
+          "Cannot listen on port %d: %s",
+          sock->port,sock_errstr(sock_errno));
   }
 
   VERB2("Openned a server socket on port %d (sd=%d)",sock->port,sock->sd);
@@ -184,24 +189,24 @@ static gras_socket_t gras_trp_sock_socket_accept(gras_socket_t sock) {
   gras_trp_socket_new(1,&res);
 
   sd = accept(sock->sd, (struct sockaddr *)&peer_in, &peer_in_len);
-  tmp_errno = errno;
+  tmp_errno = sock_errno;
 
   if (sd == -1) {
     gras_socket_close(sock);
     THROW1(system_error,0,
-          "Accept failed (%s). Droping server socket.", sock_errstr);
+          "Accept failed (%s). Droping server socket.", sock_errstr(tmp_errno));
   }
   
   if (setsockopt(sd, SOL_SOCKET, SO_KEEPALIVE, (char *)&i, s) 
       || setsockopt(sd, _gras_tcp_proto_number(), TCP_NODELAY, (char *)&i, s))
     THROW1(system_error,0,"setsockopt failed, cannot condition the socket: %s",
-          sock_errstr);
+          sock_errstr(tmp_errno));
 
   res->buf_size = sock->buf_size;
   size = sock->buf_size;
   if (setsockopt(sd, SOL_SOCKET, SO_RCVBUF, (char *)&size, sizeof(size))
       || setsockopt(sd, SOL_SOCKET, SO_SNDBUF, (char *)&size, sizeof(size)))
-    WARN1("setsockopt failed, cannot set buffer size: %s", sock_errstr);
+    WARN1("setsockopt failed, cannot set buffer size: %s", sock_errstr(tmp_errno));
      
   res->plugin    = sock->plugin;
   res->incoming  = sock->incoming;
@@ -261,7 +266,7 @@ static void gras_trp_sock_socket_close(gras_socket_t sock){
   /* close the socket */
   if(tcp_close(sock->sd) < 0) {
     WARN3("error while closing tcp socket %d: %d (%s)\n", 
-            sock->sd, sock_errno, sock_errstr);
+            sock->sd, sock_errno, sock_errstr(sock_errno));
   }
 
 }
@@ -275,7 +280,7 @@ static void gras_trp_sock_socket_close(gras_socket_t sock){
 /************************************/
 /* Temptation to merge this with file data exchange is great, 
    but doesn't work on BillWare (see tcp_write() in portable.h) */
-static inline void gras_trp_tcp_send(gras_socket_t sock,
+static XBT_INLINE void gras_trp_tcp_send(gras_socket_t sock,
                                     const char *data,
                                     unsigned long int size) {
   
@@ -286,9 +291,16 @@ static inline void gras_trp_tcp_send(gras_socket_t sock,
     DEBUG3("write(%d, %p, %ld);", sock->sd, data, size);
     
     if (status < 0) {
+#ifdef EWOULDBLOCK
+       if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)
+#else
+       if (errno == EINTR || errno == EAGAIN)
+#endif
+        continue;
+       
       THROW4(system_error,0,"write(%d,%p,%ld) failed: %s",
             sock->sd, data, size,
-            sock_errstr);
+            sock_errstr(sock_errno));
     }
     
     if (status) {
@@ -296,11 +308,11 @@ static inline void gras_trp_tcp_send(gras_socket_t sock,
       data  += status;
     } else {
       THROW1(system_error,0,"file descriptor closed (%s)",
-             sock_errstr);
+             sock_errstr(sock_errno));
     }
   }
 }
-static inline int 
+static XBT_INLINE int 
 gras_trp_tcp_recv_withbuffer(gras_socket_t sock,
                                   char *data,
                                   unsigned long int size,
@@ -313,17 +325,17 @@ gras_trp_tcp_recv_withbuffer(gras_socket_t sock,
     
     DEBUG5("read(%d, %p, %ld) got %d so far (%s)",
          sock->sd, data+got, bufsize, got,
-         hexa_str((unsigned char*)data,got));
+         hexa_str((unsigned char*)data,got,0));
     status = tcp_read(sock->sd, data+got, (size_t)bufsize);
     
     if (status < 0) {
       THROW7(system_error,0,"read(%d,%p,%d) from %s:%d failed: %s; got %d so far",
             sock->sd, data+got, (int)size,
             gras_socket_peer_name(sock),gras_socket_peer_port(sock),
-            sock_errstr,
+            sock_errstr(sock_errno),
             got);
     }
-    DEBUG2("Got %d more bytes (%s)",status,hexa_str((unsigned char*)data+got,status));
+    DEBUG2("Got %d more bytes (%s)",status,hexa_str((unsigned char*)data+got,status,0));
     
     if (status) {
       bufsize -= status;
@@ -333,6 +345,7 @@ gras_trp_tcp_recv_withbuffer(gras_socket_t sock,
             got);
     }
   }
+
   return got;
 }
 
@@ -415,14 +428,14 @@ gras_trp_buf_send(gras_socket_t sock,
           (int)data->out_buf.size,
           ((int)data->out_buf.size) + thissize -1,
           size,
-          hexa_str((unsigned char*)chunk,thissize));
+          hexa_str((unsigned char*)chunk,thissize,0));
 
     memcpy(data->out_buf.data + data->out_buf.size, chunk + chunk_pos, thissize);
 
     data->out_buf.size += thissize;
     chunk_pos      += thissize;
     DEBUG4("New pos = %d; Still to send = %ld of %ld; ctn sofar=(%s)",
-          data->out_buf.size,size-chunk_pos,size,hexa_str((unsigned char*)chunk,chunk_pos));
+          data->out_buf.size,size-chunk_pos,size,hexa_str((unsigned char*)chunk,chunk_pos,0));
 
     if (data->out_buf.size == data->buffsize) /* out of space. Flush it */
       gras_trp_bufiov_flush(sock);
@@ -466,9 +479,12 @@ gras_trp_buf_recv(gras_socket_t sock,
     data->in_buf.pos += thissize;
     chunk_pos        += thissize;
     DEBUG4("New pos = %d; Still to receive = %ld of %ld. Ctn so far=(%s)",
-          data->in_buf.pos,size - chunk_pos,size,hexa_str((unsigned char*)chunk,chunk_pos));
+          data->in_buf.pos,size - chunk_pos,size,hexa_str((unsigned char*)chunk,chunk_pos,0));
   }
-
+  /* indicate on need to the gras_select function that there is more to read on this socket so that it does not actually select */
+  sock->moredata = (data->in_buf.size > data->in_buf.pos);
+  DEBUG1("There is %smore data",(sock->moredata?"":"no "));
+   
   XBT_OUT;
   return chunk_pos;
 }
@@ -491,7 +507,7 @@ gras_trp_iov_send(gras_socket_t sock,
     
 
   DEBUG1("Buffer one chunk to be sent later (%s)",
-       hexa_str((char*)chunk,size));
+       hexa_str((char*)chunk,size,0));
 
   elm.iov_len = (size_t)size;