X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/2539fff313cbd67c45b8490f7961e45e358d9ba2..a6a92abfb622b4e667e6853a5f39f6f003007588:/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 e19f2eaa77..7439836096 100644 --- a/src/gras/Transport/transport_plugin_buf.c +++ b/src/gras/Transport/transport_plugin_buf.c @@ -2,17 +2,17 @@ /* buf trp (transport) - buffered transport using the TCP one */ -/* Authors: Martin Quinson */ -/* Copyright (C) 2004 Martin Quinson. */ +/* Copyright (c) 2004 Martin Quinson. All rights reserved. */ /* 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. */ + * under the terms of the license (GNU LGPL) which comes with this package. */ -#include /* htonl/ntohl */ #include #include /* memset */ +#include "portable.h" #include "xbt/misc.h" +#include "xbt/sysdep.h" #include "transport_private.h" XBT_LOG_NEW_DEFAULT_SUBCATEGORY(trp_buf,transport, @@ -21,6 +21,8 @@ XBT_LOG_NEW_DEFAULT_SUBCATEGORY(trp_buf,transport, /*** *** Prototypes ***/ +hexa_print(const char*name, unsigned char *data, int size); /* in gras.c */ + 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, @@ -53,7 +55,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; @@ -74,9 +76,12 @@ 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 */ - data->out.size = 0; + /* 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 = gras_if_RL()?0:4; data->out.data = xbt_malloc(data->buffsize); - data->out.pos = 0; + data->out.pos = data->out.size; sock->bufdata = data; } @@ -151,6 +156,7 @@ gras_trp_buf_socket_accept(gras_socket_t sock, TRY(super->socket_accept(sock,dst)); (*dst)->plugin = sock->plugin; gras_trp_buf_init_sock(*dst); + XBT_OUT; return no_error; } @@ -159,13 +165,20 @@ void gras_trp_buf_socket_close(gras_socket_t sock){ gras_trp_bufdata_t *data=sock->bufdata; XBT_IN; - if (data->in.size || data->out.size) + if (data->in.size!=data->in.pos) { + WARN1("Socket closed, but %d bytes were unread",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); } @@ -237,16 +250,32 @@ 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; - DEBUG0("Recv the size"); - TRY(super->chunk_recv(sock,(char*)&nextsize, 4)); - data->in.size = ntohl(nextsize); - - VERB1("Recv the chunk (size=%d)",data->in.size); + int nextsize; + if (gras_if_RL()) { + DEBUG0("Recv the size"); + TRY(super->chunk_recv(sock,(char*)&nextsize, 4)); + data->in.size = (int)ntohl(nextsize); + VERB1("Recv the chunk (size=%d)",data->in.size); + } else { + data->in.size = -1; + } + TRY(super->chunk_recv(sock, data->in.data, data->in.size)); - data->in.pos=0; + + if (gras_if_RL()) { + data->in.pos=0; + } else { + memcpy((char*)&nextsize,data->in.data,4); + data->in.size = (int)ntohl(nextsize)+4; + data->in.pos=4; + VERB3("Got the chunk (size=%d+4 for the size ifself)='%.*s'",data->in.size-4, + data->in.size,data->in.data); + if (XBT_LOG_ISENABLED(trp_buf,xbt_log_priority_debug)) + hexa_print("chunck received",data->in.data,data->in.size); + } + } - + thissize = min(size-chunck_pos , data->in.size - data->in.pos); DEBUG2("Get the chars %d..%ld out of the buffer", data->in.pos, @@ -271,18 +300,35 @@ gras_trp_buf_chunk_recv(gras_socket_t sock, xbt_error_t gras_trp_buf_flush(gras_socket_t sock) { xbt_error_t errcode; - uint32_t size; + int size; gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)sock->plugin->data)->super; gras_trp_bufdata_t *data=sock->bufdata; XBT_IN; - size = htonl(data->out.size); - DEBUG1("Send the size (=%d)",data->out.size); - TRY(super->chunk_send(sock,(char*) &size, 4)); + DEBUG0("Flush"); + if (XBT_LOG_ISENABLED(trp_buf,xbt_log_priority_debug)) + hexa_print("chunck to send",data->out.data,data->out.size); + if ((data->out.size - data->out.pos) == (gras_if_RL()?0:4) ) { /* 4 first bytes=size in SG mode*/ + DEBUG2("Nothing to flush (size=%d; pos=%d)",data->out.size,data->out.pos); + return no_error; + } + + size = (int)htonl(data->out.size - data->out.pos); + DEBUG4("%s the size (=%d) to %s:%d",(gras_if_RL()?"Send":"Embeed"),data->out.size-data->out.pos, + gras_socket_peer_name(sock),gras_socket_peer_port(sock)); + if (gras_if_RL()) { + TRY(super->chunk_send(sock,(char*) &size, 4)); + } else { + memcpy(data->out.data, &size, 4); + } + - DEBUG1("Send the chunk (size=%d)",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)); TRY(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",data->out.data,data->out.size); data->out.size = 0; return no_error; }