Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Eradicate the last bits of the buffered transport which used to live on top of either...
[simgrid.git] / src / gras / Transport / transport.c
index 6417ea3..3545c21 100644 (file)
@@ -35,13 +35,15 @@ gras_trp_plugin_new(const char *name, gras_trp_setup_t setup) {
       /* SG plugin raise mismatch when in RL mode (and vice versa) */
       free(plug->name);
       free(plug);
+      plug=NULL;
       xbt_ex_free(e);
     } else {
       RETHROW;
     }
   }
 
-  xbt_dict_set(_gras_trp_plugins, name, plug, gras_trp_plugin_free);
+  if (plug)
+    xbt_dict_set(_gras_trp_plugins, name, plug, gras_trp_plugin_free);
 }
 
 void gras_trp_init(void){
@@ -79,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++;
@@ -92,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;
 
@@ -169,7 +168,7 @@ 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;
 }
 
@@ -193,8 +192,7 @@ 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((measurement? (gras_if_RL() ? "tcp" : "sg")
-                                               :"buf"));
+  trp = gras_trp_plugin_get_by_name(gras_if_SG() ? "sg":(measurement?"tcp":"buf"));
 
   /* defaults settings */
   gras_trp_socket_new(1,&sock);
@@ -236,8 +234,7 @@ gras_socket_client_ext(const char *host,
   gras_trp_plugin_t trp;
   gras_socket_t sock;
 
-  trp = gras_trp_plugin_get_by_name((measurement? (gras_if_RL() ? "tcp" : "sg")
-                                               : "buf"));
+  trp = gras_trp_plugin_get_by_name(gras_if_SG() ? "sg":(measurement?"tcp":"buf"));
 
   DEBUG1("Create a client socket from plugin %s",gras_if_RL() ? "tcp" : "sg");
   /* defaults settings */
@@ -288,7 +285,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 +312,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 +328,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);
+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 +340,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 +349,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) {
@@ -402,12 +388,13 @@ void gras_socket_meas_send(gras_socket_t peer,
   XBT_IN;
 
   xbt_assert0(peer->meas,"Asked to send measurement data on a regular socket");
+  xbt_assert0(peer->outgoing,"Socket not suited for data send");
 
   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,
@@ -433,13 +420,15 @@ void gras_socket_meas_recv(gras_socket_t peer,
 
   XBT_IN;
 
-  xbt_assert0(peer->meas,"Asked to receive measurement data on a regular socket\n");
+  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,
@@ -486,6 +475,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;
@@ -504,13 +495,8 @@ static void gras_trp_procdata_free(void *data) {
 /*
  * 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;
-}