X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/8c354c48ec90c997cc7213ce96ca97d882934166..b48a2fd4e7621cbc0f96e39bf7db3a0263232908:/src/gras/Transport/transport_plugin_buf.c

diff --git a/src/gras/Transport/transport_plugin_buf.c b/src/gras/Transport/transport_plugin_buf.c
index 9b0cc6a100..aca9d0a812 100644
--- a/src/gras/Transport/transport_plugin_buf.c
+++ b/src/gras/Transport/transport_plugin_buf.c
@@ -7,36 +7,43 @@
 /* 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 <netinet/in.h>   /* htonl/ntohl */
 #include <stdlib.h>
 #include <string.h>       /* memset */
 
+#include "portable.h"
 #include "xbt/misc.h"
+#include "xbt/sysdep.h"
+#include "xbt/ex.h"
 #include "transport_private.h"
 
 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(trp_buf,transport,
-      "Generic buffered transport (works on top of TCP or SG)");
+      "Generic buffered transport (works on top of TCP or Files, but not SG)");
+
+
+static gras_trp_plugin_t _buf_super;
 
 /***
  *** Prototypes 
  ***/
-xbt_error_t gras_trp_buf_socket_client(gras_trp_plugin_t *self,
-					gras_socket_t sock);
-xbt_error_t gras_trp_buf_socket_server(gras_trp_plugin_t *self,
-					gras_socket_t sock);
-xbt_error_t gras_trp_buf_socket_accept(gras_socket_t sock,
-					gras_socket_t *dst);
+void hexa_print(const char*name, unsigned char *data, int size);   /* in gras.c */
+   
+void gras_trp_buf_socket_client(gras_trp_plugin_t self,
+				gras_socket_t sock);
+void gras_trp_buf_socket_server(gras_trp_plugin_t self,
+				gras_socket_t sock);
+gras_socket_t gras_trp_buf_socket_accept(gras_socket_t sock);
 
 void         gras_trp_buf_socket_close(gras_socket_t sd);
   
-xbt_error_t gras_trp_buf_chunk_send(gras_socket_t sd,
-				     const char *data,
-				     long int size);
+void gras_trp_buf_chunk_send(gras_socket_t sd,
+			     const char *data,
+			     unsigned long int size);
 
-xbt_error_t gras_trp_buf_chunk_recv(gras_socket_t sd,
-				     char *data,
-				     long int size);
-xbt_error_t gras_trp_buf_flush(gras_socket_t sock);
+void gras_trp_buf_chunk_recv(gras_socket_t sd,
+			     char *data,
+			     unsigned long int size,
+			     unsigned long int bufsize);
+void gras_trp_buf_flush(gras_socket_t sock);
 
 
 /***
@@ -44,7 +51,7 @@ xbt_error_t gras_trp_buf_flush(gras_socket_t sock);
  ***/
 
 typedef struct {
-  gras_trp_plugin_t *super;
+  int junk;
 } gras_trp_buf_plug_data_t;
 
 /***
@@ -52,7 +59,7 @@ typedef struct {
  ***/
 
 typedef struct {
-  uint32_t size;
+  int size;
   char *data;
   int pos; /* for receive; not exchanged over the net */
 } gras_trp_buf_t;
@@ -73,26 +80,27 @@ void gras_trp_buf_init_sock(gras_socket_t sock) {
   data->in.data  = xbt_malloc(data->buffsize);
   data->in.pos   = 0; /* useless, indeed, since size==pos */
    
+   /* In SG, the 4 first bytes are for the chunk size as htonl'ed, so that we can send it in one shoot.
+    * This is mandatory in SG because all emissions go to the same channel, so if we split them,
+    * they can get mixed. */
   data->out.size = 0;
   data->out.data = xbt_malloc(data->buffsize);
-  data->out.pos  = 0;
+  data->out.pos  = data->out.size;
    
   sock->bufdata = data;
 }
 
-
 /***
  *** Code
  ***/
-xbt_error_t
-gras_trp_buf_setup(gras_trp_plugin_t *plug) {
-  xbt_error_t errcode;
+void
+gras_trp_buf_setup(gras_trp_plugin_t plug) {
   gras_trp_buf_plug_data_t *data =xbt_new(gras_trp_buf_plug_data_t,1);
 
   XBT_IN;
-  TRY(gras_trp_plugin_get_by_name(gras_if_RL() ? "tcp" : "sg",
-				  &(data->super)));
-  DEBUG1("Derivate a buffer plugin from %s",gras_if_RL() ? "tcp" : "sg");
+  _buf_super = gras_trp_plugin_get_by_name("tcp");
+
+  DEBUG1("Derivate a buffer plugin from %s","tcp");
 
   plug->socket_client = gras_trp_buf_socket_client;
   plug->socket_server = gras_trp_buf_socket_server;
@@ -106,21 +114,15 @@ gras_trp_buf_setup(gras_trp_plugin_t *plug) {
 
   plug->data = (void*)data;
   plug->exit = NULL;
-  
-  return no_error;
 }
 
-xbt_error_t gras_trp_buf_socket_client(gras_trp_plugin_t *self,
-					/* OUT */ gras_socket_t sock){
-  xbt_error_t errcode;
-  gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)self->data)->super;
+void gras_trp_buf_socket_client(gras_trp_plugin_t self,
+				/* OUT */ gras_socket_t sock){
 
   XBT_IN;
-  TRY(super->socket_client(super,sock));
+  _buf_super->socket_client(_buf_super,sock);
   sock->plugin = self;
   gras_trp_buf_init_sock(sock);
-    
-  return no_error;
 }
 
 /**
@@ -128,58 +130,61 @@ xbt_error_t gras_trp_buf_socket_client(gras_trp_plugin_t *self,
  *
  * Open a socket used to receive messages.
  */
-xbt_error_t gras_trp_buf_socket_server(gras_trp_plugin_t *self,
-					/* OUT */ gras_socket_t sock){
-  xbt_error_t errcode;
-  gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)self->data)->super;
+void gras_trp_buf_socket_server(gras_trp_plugin_t self,
+				/* OUT */ gras_socket_t sock){
 
   XBT_IN;
-  TRY(super->socket_server(super,sock));
+  _buf_super->socket_server(_buf_super,sock);
   sock->plugin = self;
   gras_trp_buf_init_sock(sock);
-  return no_error;
 }
 
-xbt_error_t
-gras_trp_buf_socket_accept(gras_socket_t  sock,
-			   gras_socket_t *dst) {
-  xbt_error_t errcode;
-  gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)sock->plugin->data)->super;
+gras_socket_t
+gras_trp_buf_socket_accept(gras_socket_t  sock) {
+
+  gras_socket_t res;
       
   XBT_IN;
-  TRY(super->socket_accept(sock,dst));
-  (*dst)->plugin = sock->plugin;
-  gras_trp_buf_init_sock(*dst);
-  return no_error;
+  res = _buf_super->socket_accept(sock);
+  res->plugin = sock->plugin;
+  gras_trp_buf_init_sock(res);
+  XBT_OUT;
+  return res;
 }
 
 void gras_trp_buf_socket_close(gras_socket_t sock){
-  gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)sock->plugin->data)->super;
   gras_trp_bufdata_t *data=sock->bufdata;
 
   XBT_IN;
-  if (data->in.size || data->out.size)
+  if (data->in.size!=data->in.pos) {
+     WARN3("Socket closed, but %d bytes were unread (size=%d,pos=%d)",
+	   data->in.size - data->in.pos,data->in.size, data->in.pos);
+  }
+   
+  if (data->out.size!=data->out.pos) {
+    DEBUG2("Flush the socket before closing (in=%d,out=%d)",data->in.size, data->out.size);
     gras_trp_buf_flush(sock);
+  }
+   
   if (data->in.data)
-    xbt_free(data->in.data);
+    free(data->in.data);
   if (data->out.data)
-    xbt_free(data->out.data);
-  xbt_free(data);
+    free(data->out.data);
+  free(data);
 
-  super->socket_close(sock);
+  _buf_super->socket_close(sock);
 }
 
 /**
  * gras_trp_buf_chunk_send:
  *
- * Send data on a TCP socket
+ * Send data on a buffered socket
  */
-xbt_error_t 
+void
 gras_trp_buf_chunk_send(gras_socket_t sock,
 			const char *chunk,
-			long int size) {
+			unsigned long int size) {
 
-  xbt_error_t errcode;
   gras_trp_bufdata_t *data=(gras_trp_bufdata_t*)sock->bufdata;
   int chunk_pos=0;
 
@@ -203,28 +208,27 @@ gras_trp_buf_chunk_send(gras_socket_t sock,
 	   data->out.size,size-chunk_pos,size,(int)chunk_pos,chunk);
 
     if (data->out.size == data->buffsize) /* out of space. Flush it */
-      TRY(gras_trp_buf_flush(sock));
+      gras_trp_buf_flush(sock);
   }
 
   XBT_OUT;
-  return no_error;
 }
 
 /**
  * gras_trp_buf_chunk_recv:
  *
- * Receive data on a TCP socket.
+ * Receive data on a buffered socket.
  */
-xbt_error_t 
+void
 gras_trp_buf_chunk_recv(gras_socket_t sock,
 			char *chunk,
-			long int size) {
+			unsigned long int size,
+			unsigned long int bufsize) {
 
-  xbt_error_t errcode;
-  gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)sock->plugin->data)->super;
+  xbt_ex_t e;
   gras_trp_bufdata_t *data=sock->bufdata;
   long int chunck_pos = 0;
-
+ 
   /* Let underneath plugin check for direction, we work even in duplex */
   xbt_assert0(sock, "Cannot recv on an NULL socket");
   xbt_assert0(size >= 0, "Cannot receive a negative amount of data");
@@ -236,16 +240,22 @@ gras_trp_buf_chunk_recv(gras_socket_t sock,
     long int thissize;
 
     if (data->in.size == data->in.pos) { /* out of data. Get more */
-      uint32_t nextsize;
+      int nextsize;
       DEBUG0("Recv the size");
-      TRY(super->chunk_recv(sock,(char*)&nextsize, 4));
-      data->in.size = ntohl(nextsize);
-
+      TRY {
+	_buf_super->chunk_recv(sock,(char*)&nextsize, 4,4);
+      } CATCH(e) {
+	RETHROW3("Unable to get the chunk size on %p (peer = %s:%d): %s",
+		 sock,gras_socket_peer_name(sock),gras_socket_peer_port(sock));
+      }
+      data->in.size = (int)ntohl(nextsize);
       VERB1("Recv the chunk (size=%d)",data->in.size);
-      TRY(super->chunk_recv(sock, data->in.data, data->in.size));
+       
+      _buf_super->chunk_recv(sock, data->in.data, data->in.size, data->in.size);
+       
       data->in.pos=0;
     }
-    
+     
     thissize = min(size-chunck_pos ,  data->in.size - data->in.pos);
     DEBUG2("Get the chars %d..%ld out of the buffer",
 	   data->in.pos,
@@ -259,7 +269,6 @@ gras_trp_buf_chunk_recv(gras_socket_t sock,
   }
 
   XBT_OUT;
-  return no_error;
 }
 
 /**
@@ -267,21 +276,32 @@ gras_trp_buf_chunk_recv(gras_socket_t sock,
  *
  * Make sure the data is sent
  */
-xbt_error_t 
+void
 gras_trp_buf_flush(gras_socket_t sock) {
-  xbt_error_t errcode;
-  uint32_t size;
-  gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)sock->plugin->data)->super;
+  int size;
   gras_trp_bufdata_t *data=sock->bufdata;
+  XBT_IN;    
+  
+  DEBUG0("Flush");
+  if (XBT_LOG_ISENABLED(trp_buf,xbt_log_priority_debug))
+     hexa_print("chunck to send ",(unsigned char *) data->out.data,data->out.size);
+  if ((data->out.size - data->out.pos) == 0) { 
+     DEBUG2("Nothing to flush (size=%d; pos=%d)",data->out.size,data->out.pos);
+     return;
+  }
+   
+  size = (int)data->out.size - data->out.pos;
+  DEBUG3("Send the size (=%d) to %s:%d",data->out.size-data->out.pos,
+	 gras_socket_peer_name(sock),gras_socket_peer_port(sock));
+  size = (int)htonl(size);
+  _buf_super->chunk_send(sock,(char*) &size, 4);
+      
 
-  XBT_IN;
-  size = htonl(data->out.size);
-  DEBUG1("Send the size (=%d)",data->out.size);
-  TRY(super->chunk_send(sock,(char*) &size, 4));
-
-  DEBUG1("Send the chunk (size=%d)",data->out.size);
-  TRY(super->chunk_send(sock, data->out.data, data->out.size));
+  DEBUG3("Send the chunk (size=%d) to %s:%d",data->out.size,
+	 gras_socket_peer_name(sock),gras_socket_peer_port(sock));
+  _buf_super->chunk_send(sock, data->out.data, data->out.size);
   VERB1("Chunk sent (size=%d)",data->out.size);
+  if (XBT_LOG_ISENABLED(trp_buf,xbt_log_priority_debug))
+     hexa_print("chunck sent    ",(unsigned char *) data->out.data,data->out.size);
   data->out.size = 0;
-  return no_error;
 }