Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
cosmetics, plus tracked down a source of problem with measurement sockets
[simgrid.git] / src / gras / Transport / transport.c
index d88a401..9d66b21 100644 (file)
@@ -36,7 +36,7 @@ gras_trp_plugin_new(const char *name, gras_trp_setup_t setup) {
       free(plug->name);
       free(plug);
       plug=NULL;
-      xbt_ex_free(e);
+      xbt_ex_free(&e);
     } else {
       RETHROW;
     }
@@ -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,7 +91,7 @@ void gras_trp_init(void){
 
 void
 gras_trp_exit(void){
-  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;
 
@@ -171,21 +168,24 @@ void gras_trp_socket_new(int incoming,
   
   *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. 
  */
 gras_socket_t
 gras_socket_server_ext(unsigned short port,
                       
-                      unsigned long int bufSize,
+                      unsigned long int buf_size,
                       int measurement) {
  
   xbt_ex_t e;
@@ -195,13 +195,13 @@ gras_socket_server_ext(unsigned short port,
   DEBUG2("Create a server socket from plugin %s on port %d",
         gras_if_RL() ? "tcp" : "sg",
         port);
-  trp = gras_trp_plugin_get_by_name(gras_if_SG() ? "sg":(measurement?"tcp":"buf"));
+  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->buf_size = buf_size>0 ? buf_size : 32*1024;
   sock->meas = measurement;
 
   /* Call plugin socket creation function */
@@ -221,23 +221,26 @@ gras_socket_server_ext(unsigned short port,
 }
    
 /**
- * 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. 
  */
 gras_socket_t
 gras_socket_client_ext(const char *host,
                       unsigned short port,
                       
-                      unsigned long int bufSize,
+                      unsigned long int buf_size,
                       int measurement) {
  
   xbt_ex_t e;
   gras_trp_plugin_t trp;
   gras_socket_t sock;
 
-  trp = gras_trp_plugin_get_by_name(gras_if_SG() ? "sg":(measurement?"tcp":"buf"));
+  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 */
@@ -245,7 +248,7 @@ 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->buf_size = buf_size>0 ? buf_size: 32*1024;
   sock->meas = measurement;
 
   /* plugin-specific */
@@ -288,7 +291,7 @@ gras_socket_client(const char *host,
 
 
 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;
 
@@ -315,21 +318,15 @@ void gras_socket_close(gras_socket_t sock) {
 }
 
 /**
- * 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)
  */
 void
-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);
-  (*sd->plugin->chunk_send)(sd,data,size);
+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:
@@ -337,15 +334,9 @@ gras_trp_chunk_send(gras_socket_t sd,
  * Receive a bunch of bytes from a socket
  */
 void
-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);
-  (sd->plugin->chunk_recv)(sd,data,size,size);
+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);
 }
 
 /**
@@ -355,7 +346,8 @@ gras_trp_chunk_recv(gras_socket_t sd,
  */
 void
 gras_trp_flush(gras_socket_t sd) {
-  (sd->plugin->flush)(sd);
+  if (sd->plugin->flush)
+    (sd->plugin->flush)(sd);
 }
 
 gras_trp_plugin_t
@@ -363,7 +355,7 @@ 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) {
@@ -396,24 +388,29 @@ void gras_socket_meas_send(gras_socket_t peer,
                           unsigned int timeout,
                           unsigned long int exp_size, 
                           unsigned long int msg_size) {
-  char *chunk = xbt_malloc0(msg_size);
+  char *chunk=NULL;
   unsigned long int exp_sofar;
    
   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(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));
-     gras_trp_chunk_send(peer,chunk,msg_size);
+     (*peer->plugin->raw_send)(peer,chunk,msg_size);
   }
   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));
             
-  free(chunk);
+  if (gras_if_RL()) 
+    free(chunk);
 
   XBT_OUT;
 }
@@ -428,24 +425,30 @@ void gras_socket_meas_recv(gras_socket_t peer,
                           unsigned long int exp_size, 
                           unsigned long int msg_size){
   
-  char *chunk = xbt_malloc(msg_size);
+  char *chunk=NULL;
   unsigned long int exp_sofar;
 
   XBT_IN;
 
-  xbt_assert0(peer->meas,"Asked to receive measurement data on a regular socket\n");
+  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");
 
   for (exp_sofar=0; exp_sofar < exp_size; exp_sofar += msg_size) {
      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));
-     gras_trp_chunk_recv(peer,chunk,msg_size);
+     (peer->plugin->raw_recv)(peer,chunk,msg_size);
   }
   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));
 
-  free(chunk);
+  if (gras_if_RL()) 
+    free(chunk);
   XBT_OUT;
 }
 
@@ -468,7 +471,8 @@ gras_socket_t gras_socket_meas_accept(gras_socket_t peer){
              "No need to accept on non-measurement sockets (it's automatic)");
 
   if (!peer->accepting) {
-    /* nothing to accept here */
+    /* nothing to accept here (must be in SG) */
+    /* FIXME: this is BAD! since it makes tricky to free the accepted socket*/
     return peer;
   }
 
@@ -486,6 +490,8 @@ gras_socket_t gras_socket_meas_accept(gras_socket_t peer){
 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;
@@ -498,19 +504,15 @@ 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_procdata_add("gras_trp",gras_trp_procdata_new, gras_trp_procdata_free);
+   gras_trp_libdata_id = gras_procdata_add("gras_trp",gras_trp_procdata_new, gras_trp_procdata_free);
 }
 
-
-xbt_dynar_t 
-gras_socketset_get(void) {
-   /* FIXME: KILLME */
-   return ((gras_trp_procdata_t) gras_libdata_get("gras_trp"))->sockets;
-}