X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/0174b734fea5f64c8c28a1b489ff1cd6e824ca35..2e2ca643ddfa22db6d912fe297d319433217a401:/src/gras/Transport/transport.c

diff --git a/src/gras/Transport/transport.c b/src/gras/Transport/transport.c
index 9aec7307e3..9d66b21e8c 100644
--- a/src/gras/Transport/transport.c
+++ b/src/gras/Transport/transport.c
@@ -7,11 +7,12 @@
 /* This program is free software; you can redistribute it and/or modify it
  * under the terms of the license (GNU LGPL) which comes with this package. */
 
+#include "xbt/ex.h"
 #include "portable.h"
 #include "gras/Transport/transport_private.h"
 
 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(transport,gras,"Conveying bytes over the network");
-XBT_LOG_NEW_SUBCATEGORY(raw_trp,transport,"Conveying bytes over the network without formating");
+XBT_LOG_NEW_SUBCATEGORY(trp_meas,transport,"Conveying bytes over the network without formating for perf measurements");
 static short int _gras_trp_started = 0;
 
 static xbt_dict_t _gras_trp_plugins;      /* All registered plugins */
@@ -19,31 +20,30 @@ static void gras_trp_plugin_free(void *p); /* free one of the plugins */
 
 static void
 gras_trp_plugin_new(const char *name, gras_trp_setup_t setup) {
-  xbt_error_t errcode;
+  xbt_ex_t e;
 
-  gras_trp_plugin_t *plug = xbt_new0(gras_trp_plugin_t, 1);
+  gras_trp_plugin_t plug = xbt_new0(s_gras_trp_plugin_t, 1);
   
   DEBUG1("Create plugin %s",name);
 
   plug->name=xbt_strdup(name);
 
-  errcode = setup(plug);
-  switch (errcode) {
-  case mismatch_error:
-    /* SG plugin return mismatch when in RL mode (and vice versa) */
-    xbt_free(plug->name);
-    xbt_free(plug);
-    break;
-
-  case no_error:
-    xbt_dict_set(_gras_trp_plugins,
-		  name, plug, gras_trp_plugin_free);
-    break;
-
-  default:
-    DIE_IMPOSSIBLE;
+  TRY {
+    setup(plug);
+  } CATCH(e) {
+    if (e.category == mismatch_error) {
+      /* SG plugin raise mismatch when in RL mode (and vice versa) */
+      free(plug->name);
+      free(plug);
+      plug=NULL;
+      xbt_ex_free(&e);
+    } else {
+      RETHROW;
+    }
   }
 
+  if (plug)
+    xbt_dict_set(_gras_trp_plugins, name, plug, gras_trp_plugin_free);
 }
 
 void gras_trp_init(void){
@@ -52,7 +52,7 @@ void gras_trp_init(void){
      _gras_trp_plugins=xbt_dict_new();
 
 #ifdef HAVE_WINSOCK2_H
-     /* initialize the windows mecanism */
+     /* initialize the windows mechanism */
      {  
 	WORD wVersionRequested;
 	WSADATA wsaData;
@@ -81,12 +81,9 @@ void gras_trp_init(void){
 #endif
    
      /* Add plugins */
-     gras_trp_plugin_new("tcp", gras_trp_tcp_setup);
      gras_trp_plugin_new("file",gras_trp_file_setup);
      gras_trp_plugin_new("sg",gras_trp_sg_setup);
-
-     /* buf is composed, so it must come after the others */
-     gras_trp_plugin_new("buf", gras_trp_buf_setup);
+     gras_trp_plugin_new("tcp", gras_trp_tcp_setup);
   }
    
   _gras_trp_started++;
@@ -94,6 +91,10 @@ void gras_trp_init(void){
 
 void
 gras_trp_exit(void){
+  xbt_dynar_t sockets = ((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->sockets;
+  gras_socket_t sock_iter;
+  int cursor;
+
    if (_gras_trp_started == 0) {
       return;
    }
@@ -108,24 +109,32 @@ gras_trp_exit(void){
 	}
 #endif
 
+      /* Close all the sockets */
+      xbt_dynar_foreach(sockets,cursor,sock_iter) {
+	VERB1("Closing the socket %p left open on exit. Maybe a socket leak?",
+	      sock_iter);
+	gras_socket_close(sock_iter);
+      }
+      
+      /* Delete the plugins */
       xbt_dict_free(&_gras_trp_plugins);
    }
 }
 
 
 void gras_trp_plugin_free(void *p) {
-  gras_trp_plugin_t *plug = p;
+  gras_trp_plugin_t plug = p;
 
   if (plug) {
     if (plug->exit) {
       plug->exit(plug);
     } else if (plug->data) {
       DEBUG1("Plugin %s lacks exit(). Free data anyway.",plug->name);
-      xbt_free(plug->data);
+      free(plug->data);
     }
 
-    xbt_free(plug->name);
-    xbt_free(plug);
+    free(plug->name);
+    free(plug);
   }
 }
 
@@ -143,97 +152,95 @@ void gras_trp_socket_new(int incoming,
   DEBUG1("Create a new socket (%p)", (void*)sock);
 
   sock->plugin = NULL;
-  sock->sd     = -1;
-  sock->data   = NULL;
 
   sock->incoming  = incoming ? 1:0;
   sock->outgoing  = incoming ? 0:1;
   sock->accepting = incoming ? 1:0;
+  sock->meas = 0;
 
+  sock->sd     = -1;
   sock->port      = -1;
   sock->peer_port = -1;
   sock->peer_name = NULL;
-  sock->raw = 0;
 
+  sock->data   = NULL;
+  sock->bufdata = NULL;
+  
   *dst = sock;
 
-  xbt_dynar_push(gras_socketset_get(),dst);
+  xbt_dynar_push(((gras_trp_procdata_t) 
+		     gras_libdata_by_id(gras_trp_libdata_id))->sockets,dst);
   XBT_OUT;
 }
 
 
 /**
- * gras_socket_server_ext:
- *
- * Opens a server socket and make it ready to be listened to.
- * In real life, you'll get a TCP socket.
+ * @brief Opens a server socket and makes it ready to be listened to.
+ * @param port: port on which you want to listen
+ * @param buf_size: size of the buffer (in byte) on the socket (for TCP sockets only). If 0, a sain default is used (32k, but may change)
+ * @param measurement: whether this socket is meant to convey measurement (if you don't know, use 0 to exchange regular messages)
+ * 
+ * In real life, you'll get a TCP socket. 
  */
-xbt_error_t
+gras_socket_t
 gras_socket_server_ext(unsigned short port,
 		       
-		       unsigned long int bufSize,
-		       int raw,
-		       
-		       /* OUT */ gras_socket_t *dst) {
+		       unsigned long int buf_size,
+		       int measurement) {
  
-  xbt_error_t errcode;
-  gras_trp_plugin_t *trp;
+  xbt_ex_t e;
+  gras_trp_plugin_t trp;
   gras_socket_t sock;
 
-  *dst = NULL;
-
   DEBUG2("Create a server socket from plugin %s on port %d",
 	 gras_if_RL() ? "tcp" : "sg",
 	 port);
-  TRY(gras_trp_plugin_get_by_name("buf",&trp));
+  trp = gras_trp_plugin_get_by_name(gras_if_SG() ? "sg":"tcp");
 
   /* defaults settings */
   gras_trp_socket_new(1,&sock);
   sock->plugin= trp;
   sock->port=port;
-  sock->bufSize = bufSize;
-  sock->raw = raw;
+  sock->buf_size = buf_size>0 ? buf_size : 32*1024;
+  sock->meas = measurement;
 
   /* Call plugin socket creation function */
   DEBUG1("Prepare socket with plugin (fct=%p)",trp->socket_server);
-  errcode = trp->socket_server(trp, sock);
-  DEBUG3("in=%c out=%c accept=%c",
-	 sock->incoming?'y':'n', 
-	 sock->outgoing?'y':'n',
-	 sock->accepting?'y':'n');
-
-  if (errcode != no_error) {
-    xbt_free(sock);
-    return errcode;
+  TRY {
+    trp->socket_server(trp, sock);
+    DEBUG3("in=%c out=%c accept=%c",
+	   sock->incoming?'y':'n', 
+	   sock->outgoing?'y':'n',
+	   sock->accepting?'y':'n');
+  } CATCH(e) {
+    free(sock);
+    RETHROW;
   }
 
-  *dst = sock;
-
-  return no_error;
+  return sock;
 }
    
 /**
- * gras_socket_client_ext:
- *
- * Opens a client socket to a remote host.
- * In real life, you'll get a TCP socket.
+ * @brief Opens a client socket to a remote host.
+ * @param host: who you want to connect to
+ * @param port: where you want to connect to on this host
+ * @param buf_size: size of the buffer (in bytes) on the socket (for TCP sockets only). If 0, a sain default is used (32k, but may change)
+ * @param measurement: whether this socket is meant to convey measurement (if you don't know, use 0 to exchange regular messages)
+ * 
+ * In real life, you'll get a TCP socket. 
  */
-xbt_error_t
+gras_socket_t
 gras_socket_client_ext(const char *host,
 		       unsigned short port,
 		       
-		       unsigned long int bufSize,
-		       int raw,
-		       
-		       /* OUT */ gras_socket_t *dst) {
+		       unsigned long int buf_size,
+		       int measurement) {
  
-  xbt_error_t errcode;
-  gras_trp_plugin_t *trp;
+  xbt_ex_t e;
+  gras_trp_plugin_t trp;
   gras_socket_t sock;
 
-  *dst = NULL;
-
-  TRY(gras_trp_plugin_get_by_name("buf",&trp));
+  trp = gras_trp_plugin_get_by_name(gras_if_SG() ? "sg":"tcp");
 
   DEBUG1("Create a client socket from plugin %s",gras_if_RL() ? "tcp" : "sg");
   /* defaults settings */
@@ -241,24 +248,22 @@ gras_socket_client_ext(const char *host,
   sock->plugin= trp;
   sock->peer_port = port;
   sock->peer_name = (char*)strdup(host?host:"localhost");
-  sock->bufSize = bufSize;
-  sock->raw = raw;
+  sock->buf_size = buf_size>0 ? buf_size: 32*1024;
+  sock->meas = measurement;
 
   /* plugin-specific */
-  errcode= (*trp->socket_client)(trp, sock);
-  DEBUG3("in=%c out=%c accept=%c",
-	 sock->incoming?'y':'n', 
-	 sock->outgoing?'y':'n',
-	 sock->accepting?'y':'n');
-
-  if (errcode != no_error) {
-    xbt_free(sock);
-    return errcode;
+  TRY {
+    (*trp->socket_client)(trp, sock);
+    DEBUG3("in=%c out=%c accept=%c",
+	   sock->incoming?'y':'n', 
+	   sock->outgoing?'y':'n',
+	   sock->accepting?'y':'n');
+  } CATCH(e) {
+    free(sock);
+    RETHROW;
   }
 
-  *dst = sock;
-
-  return no_error;
+  return sock;
 }
 
 /**
@@ -267,10 +272,9 @@ gras_socket_client_ext(const char *host,
  * Opens a server socket and make it ready to be listened to.
  * In real life, you'll get a TCP socket.
  */
-xbt_error_t
-gras_socket_server(unsigned short port,
-		   /* OUT */ gras_socket_t *dst) {
-   return gras_socket_server_ext(port,32,0,dst);
+gras_socket_t
+gras_socket_server(unsigned short port) {
+   return gras_socket_server_ext(port,32,0);
 }
 
 /**
@@ -279,19 +283,19 @@ gras_socket_server(unsigned short port,
  * Opens a client socket to a remote host.
  * In real life, you'll get a TCP socket.
  */
-xbt_error_t
+gras_socket_t
 gras_socket_client(const char *host,
-		   unsigned short port,
-		   /* OUT */ gras_socket_t *dst) {
-   return gras_socket_client_ext(host,port,32,0,dst);
+		   unsigned short port) {
+   return gras_socket_client_ext(host,port,32,0);
 }
 
 
 void gras_socket_close(gras_socket_t sock) {
-  xbt_dynar_t sockets = gras_socketset_get();
+  xbt_dynar_t sockets = ((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->sockets;
   gras_socket_t sock_iter;
   int cursor;
 
+  XBT_IN;
   /* FIXME: Issue an event when the socket is closed */
   if (sock) {
     xbt_dynar_foreach(sockets,cursor,sock_iter) {
@@ -302,47 +306,37 @@ void gras_socket_close(gras_socket_t sock) {
 
 	/* free the memory */
 	if (sock->peer_name)
-	  xbt_free(sock->peer_name);
-	xbt_free(sock);
+	  free(sock->peer_name);
+	free(sock);
+	XBT_OUT;
 	return;
       }
     }
-    WARN0("Ignoring request to free an unknown socket");
+    WARN1("Ignoring request to free an unknown socket (%p)",sock);
   }
+  XBT_OUT;
 }
 
 /**
- * gras_trp_chunk_send:
+ * gras_trp_send:
  *
  * Send a bunch of bytes from on socket
+ * (stable if we know the storage will keep as is until the next trp_flush)
  */
-xbt_error_t
-gras_trp_chunk_send(gras_socket_t sd,
-		    char *data,
-		    long int size) {
-  xbt_assert1(sd->outgoing,
-	       "Socket not suited for data send (outgoing=%c)",
-	       sd->outgoing?'y':'n');
-  xbt_assert1(sd->plugin->chunk_send,
-	       "No function chunk_send on transport plugin %s",
-	       sd->plugin->name);
-  return (*sd->plugin->chunk_send)(sd,data,size);
+void
+gras_trp_send(gras_socket_t sd, char *data, long int size, int stable) {
+  xbt_assert0(sd->outgoing,"Socket not suited for data send");
+  (*sd->plugin->send)(sd,data,size,stable);
 }
 /**
  * gras_trp_chunk_recv:
  *
  * Receive a bunch of bytes from a socket
  */
-xbt_error_t 
-gras_trp_chunk_recv(gras_socket_t sd,
-		    char *data,
-		    long int size) {
-  xbt_assert0(sd->incoming,
-	       "Socket not suited for data receive");
-  xbt_assert1(sd->plugin->chunk_recv,
-	       "No function chunk_recv on transport plugin %s",
-	       sd->plugin->name);
-  return (sd->plugin->chunk_recv)(sd,data,size);
+void
+gras_trp_recv(gras_socket_t sd, char *data, long int size) {
+  xbt_assert0(sd->incoming,"Socket not suited for data receive");
+  (sd->plugin->recv)(sd,data,size);
 }
 
 /**
@@ -350,19 +344,18 @@ gras_trp_chunk_recv(gras_socket_t sd,
  *
  * Make sure all pending communications are done
  */
-xbt_error_t 
+void
 gras_trp_flush(gras_socket_t sd) {
-  return (sd->plugin->flush)(sd);
+  if (sd->plugin->flush)
+    (sd->plugin->flush)(sd);
 }
 
-xbt_error_t
-gras_trp_plugin_get_by_name(const char *name,
-			    gras_trp_plugin_t **dst){
-
-  return xbt_dict_get(_gras_trp_plugins,name,(void**)dst);
+gras_trp_plugin_t
+gras_trp_plugin_get_by_name(const char *name){
+  return xbt_dict_get(_gras_trp_plugins,name);
 }
 
-int   gras_socket_my_port  (gras_socket_t sock) {
+int gras_socket_my_port  (gras_socket_t sock) {
   return sock->port;
 }
 int   gras_socket_peer_port(gras_socket_t sock) {
@@ -372,49 +365,154 @@ char *gras_socket_peer_name(gras_socket_t sock) {
   return sock->peer_name;
 }
 
-xbt_error_t gras_socket_raw_send(gras_socket_t peer, 
-				  unsigned int timeout,
-				  unsigned long int exp_size, 
-				  unsigned long int msg_size) {
-  xbt_error_t errcode;
-  char *chunk = xbt_malloc(msg_size);
-  int exp_sofar;
+/** \brief Check if the provided socket is a measurement one (or a regular one) */
+int gras_socket_is_meas(gras_socket_t sock) {
+  return sock->meas;
+}
+
+/** \brief Send a chunk of (random) data over a measurement socket 
+ *
+ * @param peer measurement socket to use for the experiment
+ * @param timeout timeout (in seconds)
+ * @param exp_size total amount of data to send (in bytes).
+ * @param msg_size size of each chunk sent over the socket (in bytes).
+ *
+ * Calls to gras_socket_meas_send() and gras_socket_meas_recv() on 
+ * each side of the socket should be paired. 
+ * 
+ * The exchanged data is zeroed to make sure it's initialized, but
+ * there is no way to control what is sent (ie, you cannot use these 
+ * functions to exchange data out of band).
+ */
+void gras_socket_meas_send(gras_socket_t peer, 
+			   unsigned int timeout,
+			   unsigned long int exp_size, 
+			   unsigned long int msg_size) {
+  char *chunk=NULL;
+  unsigned long int exp_sofar;
    
-  xbt_assert0(peer->raw,"Asked to send raw data on a regular socket");
+  XBT_IN;
+
+  if (gras_if_RL()) 
+    chunk=xbt_malloc0(msg_size);
+
+  xbt_assert0(peer->meas,"Asked to send measurement data on a regular socket");
+  xbt_assert0(peer->outgoing,"Socket not suited for data send (was created with gras_socket_server(), not gras_socket_client())");
+
   for (exp_sofar=0; exp_sofar < exp_size; exp_sofar += msg_size) {
-     CDEBUG5(raw_trp,"Sent %d of %lu (msg_size=%ld) to %s:%d",
+     CDEBUG5(trp_meas,"Sent %lu of %lu (msg_size=%ld) to %s:%d",
 	     exp_sofar,exp_size,msg_size,
 	     gras_socket_peer_name(peer), gras_socket_peer_port(peer));
-     TRY(gras_trp_chunk_send(peer,chunk,msg_size));
+     (*peer->plugin->raw_send)(peer,chunk,msg_size);
   }
-  CDEBUG5(raw_trp,"Sent %d of %lu (msg_size=%ld) to %s:%d",
+  CDEBUG5(trp_meas,"Sent %lu of %lu (msg_size=%ld) to %s:%d",
 	  exp_sofar,exp_size,msg_size,
 	  gras_socket_peer_name(peer), gras_socket_peer_port(peer));
 	     
-  xbt_free(chunk);
-  return no_error;/* gras_socket_raw_exchange(peer,1,timeout,expSize,msgSize);    */
+  if (gras_if_RL()) 
+    free(chunk);
+
+  XBT_OUT;
 }
 
-xbt_error_t gras_socket_raw_recv(gras_socket_t peer, 
-				  unsigned int timeout,
-				  unsigned long int exp_size, 
-				  unsigned long int msg_size){
+/** \brief Receive a chunk of data over a measurement socket 
+ *
+ * Calls to gras_socket_meas_send() and gras_socket_meas_recv() on 
+ * each side of the socket should be paired. 
+ */
+void gras_socket_meas_recv(gras_socket_t peer, 
+			   unsigned int timeout,
+			   unsigned long int exp_size, 
+			   unsigned long int msg_size){
   
-  xbt_error_t errcode;
-  char *chunk = xbt_malloc(msg_size);
-  int exp_sofar;
+  char *chunk=NULL;
+  unsigned long int exp_sofar;
+
+  XBT_IN;
+
+  if (gras_if_RL()) 
+    chunk = xbt_malloc(msg_size);
+
+  xbt_assert0(peer->meas,
+	      "Asked to receive measurement data on a regular socket");
+  xbt_assert0(peer->incoming,"Socket not suited for data receive");
 
-  xbt_assert0(peer->raw,"Asked to recveive raw data on a regular socket\n");
   for (exp_sofar=0; exp_sofar < exp_size; exp_sofar += msg_size) {
-     CDEBUG5(raw_trp,"Recvd %d of %lu (msg_size=%ld) from %s:%d",
+     CDEBUG5(trp_meas,"Recvd %ld of %lu (msg_size=%ld) from %s:%d",
 	     exp_sofar,exp_size,msg_size,
 	     gras_socket_peer_name(peer), gras_socket_peer_port(peer));
-     TRY(gras_trp_chunk_recv(peer,chunk,msg_size));
+     (peer->plugin->raw_recv)(peer,chunk,msg_size);
   }
-  CDEBUG5(raw_trp,"Recvd %d of %lu (msg_size=%ld) from %s:%d",
+  CDEBUG5(trp_meas,"Recvd %ld of %lu (msg_size=%ld) from %s:%d",
 	  exp_sofar,exp_size,msg_size,
 	  gras_socket_peer_name(peer), gras_socket_peer_port(peer));
 
-  xbt_free(chunk);
-  return no_error;/* gras_socket_raw_exchange(peer,0,timeout,expSize,msgSize);    */
+  if (gras_if_RL()) 
+    free(chunk);
+  XBT_OUT;
+}
+
+/**
+ * \brief Something similar to the good old accept system call. 
+ *
+ * Make sure that there is someone speaking to the provided server socket.
+ * In RL, it does an accept(2) and return the result as last argument. 
+ * In SG, as accepts are useless, it returns the provided argument as result.
+ * You should thus test whether (peer != accepted) before closing both of them.
+ *
+ * You should only call this on measurement sockets. It is automatically 
+ * done for regular sockets, but you usually want more control about 
+ * what's going on with measurement sockets.
+ */
+gras_socket_t gras_socket_meas_accept(gras_socket_t peer){
+  gras_socket_t res;
+
+  xbt_assert0(peer->meas,
+	      "No need to accept on non-measurement sockets (it's automatic)");
+
+  if (!peer->accepting) {
+    /* nothing to accept here (must be in SG) */
+    /* FIXME: this is BAD! since it makes tricky to free the accepted socket*/
+    return peer;
+  }
+
+  res = (peer->plugin->socket_accept)(peer);
+  res->meas = peer->meas;
+  CDEBUG1(trp_meas,"meas_accepted onto %d",res->sd);
+
+  return res;
+} 
+
+
+/*
+ * Creating procdata for this module
+ */
+static void *gras_trp_procdata_new() {
+   gras_trp_procdata_t res = xbt_new(s_gras_trp_procdata_t,1);
+   
+   res->name = xbt_strdup("gras_trp");
+   res->name_len = 0;
+   res->sockets   = xbt_dynar_new(sizeof(gras_socket_t*), NULL);
+   
+   return (void*)res;
+}
+
+/*
+ * Freeing procdata for this module
+ */
+static void gras_trp_procdata_free(void *data) {
+   gras_trp_procdata_t res = (gras_trp_procdata_t)data;
+   
+   xbt_dynar_free(&( res->sockets ));
+   free(res->name);
+   free(res);
 }
+
+/*
+ * Module registration
+ */
+int gras_trp_libdata_id;
+void gras_trp_register() {
+   gras_trp_libdata_id = gras_procdata_add("gras_trp",gras_trp_procdata_new, gras_trp_procdata_free);
+}
+