X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/0eaef9d438cbed19d62f9bd57b70659d5cac96b8..2479b96521c45f40637d8e4482c301b677266443:/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 6045943bbb..d640ab7562 100644 --- a/src/gras/Transport/transport_plugin_buf.c +++ b/src/gras/Transport/transport_plugin_buf.c @@ -2,44 +2,44 @@ /* 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 "gras_private.h" +#include "portable.h" +#include "xbt/misc.h" +#include "xbt/sysdep.h" #include "transport_private.h" -GRAS_LOG_NEW_DEFAULT_SUBCATEGORY(trp_buf,transport); +XBT_LOG_NEW_DEFAULT_SUBCATEGORY(trp_buf,transport, + "Generic buffered transport (works on top of TCP or SG)"); /*** *** Prototypes ***/ -gras_error_t gras_trp_buf_socket_client(gras_trp_plugin_t *self, - const char *host, - unsigned short port, - /* OUT */ gras_socket_t *sock); -gras_error_t gras_trp_buf_socket_server(gras_trp_plugin_t *self, - unsigned short port, - /* OUT */ gras_socket_t *sock); -gras_error_t gras_trp_buf_socket_accept(gras_socket_t *sock, - gras_socket_t **dst); - -void gras_trp_buf_socket_close(gras_socket_t *sd); +void 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, + gras_socket_t sock); +xbt_error_t gras_trp_buf_socket_accept(gras_socket_t sock, + gras_socket_t *dst); + +void gras_trp_buf_socket_close(gras_socket_t sd); -gras_error_t gras_trp_buf_chunk_send(gras_socket_t *sd, +xbt_error_t gras_trp_buf_chunk_send(gras_socket_t sd, const char *data, long int size); -gras_error_t gras_trp_buf_chunk_recv(gras_socket_t *sd, +xbt_error_t gras_trp_buf_chunk_recv(gras_socket_t sd, char *data, long int size); -gras_error_t gras_trp_buf_flush(gras_socket_t *sock); +xbt_error_t gras_trp_buf_flush(gras_socket_t sock); /*** @@ -55,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; @@ -66,38 +66,36 @@ struct gras_trp_bufdata_{ int buffsize; }; -gras_error_t gras_trp_buf_init_sock(gras_socket_t *sock) { - gras_trp_bufdata_t *data=malloc(sizeof(gras_trp_bufdata_t)); +void gras_trp_buf_init_sock(gras_socket_t sock) { + gras_trp_bufdata_t *data=xbt_new(gras_trp_bufdata_t,1); - GRAS_IN; - if (!data) - RAISE_MALLOC; + XBT_IN; + data->buffsize = 100 * 1024 ; /* 100k */ + data->in.size = 0; - if (!(data->in.data = malloc(data->buffsize))) - RAISE_MALLOC; + data->in.data = xbt_malloc(data->buffsize); data->in.pos = 0; /* useless, indeed, since size==pos */ - data->out.size = 0; - if (!(data->out.data = malloc(data->buffsize))) - RAISE_MALLOC; - data->out.pos = 0; - // data->buffsize = 32 * 1024 - 4; /* default socket buffsize (32k) - headers */ - data->buffsize = 100 * 1024 ; /* 100k */ + + /* 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 = data->out.size; + sock->bufdata = data; - return no_error; } /*** *** Code ***/ -gras_error_t +xbt_error_t gras_trp_buf_setup(gras_trp_plugin_t *plug) { - gras_error_t errcode; - gras_trp_buf_plug_data_t *data =malloc(sizeof(gras_trp_buf_plug_data_t)); - if (!data) - RAISE_MALLOC; + xbt_error_t errcode; + gras_trp_buf_plug_data_t *data =xbt_new(gras_trp_buf_plug_data_t,1); - GRAS_IN; + 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"); @@ -118,17 +116,15 @@ gras_trp_buf_setup(gras_trp_plugin_t *plug) { return no_error; } -gras_error_t gras_trp_buf_socket_client(gras_trp_plugin_t *self, - const char *host, - unsigned short port, - /* OUT */ gras_socket_t *sock){ - gras_error_t errcode; +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; - GRAS_IN; - TRY(super->socket_client(super,host,port,sock)); + XBT_IN; + TRY(super->socket_client(super,sock)); sock->plugin = self; - TRY(gras_trp_buf_init_sock(sock)); + gras_trp_buf_init_sock(sock); return no_error; } @@ -138,46 +134,54 @@ gras_error_t gras_trp_buf_socket_client(gras_trp_plugin_t *self, * * Open a socket used to receive messages. */ -gras_error_t gras_trp_buf_socket_server(gras_trp_plugin_t *self, - unsigned short port, - /* OUT */ gras_socket_t *sock){ - gras_error_t errcode; +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; - GRAS_IN; - TRY(super->socket_server(super,port,sock)); + XBT_IN; + TRY(super->socket_server(super,sock)); sock->plugin = self; - TRY(gras_trp_buf_init_sock(sock)); + gras_trp_buf_init_sock(sock); return no_error; } -gras_error_t -gras_trp_buf_socket_accept(gras_socket_t *sock, - gras_socket_t **dst) { - gras_error_t errcode; +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_IN; + XBT_IN; TRY(super->socket_accept(sock,dst)); (*dst)->plugin = sock->plugin; - TRY(gras_trp_buf_init_sock(*dst)); + gras_trp_buf_init_sock(*dst); + XBT_OUT; return no_error; } -void gras_trp_buf_socket_close(gras_socket_t *sock){ +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; - GRAS_IN; - if (data->in.size || data->out.size) + XBT_IN; + 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) free(data->in.data); if (data->out.data) free(data->out.data); free(data); - return super->socket_close(sock); + super->socket_close(sock); } /** @@ -185,18 +189,18 @@ void gras_trp_buf_socket_close(gras_socket_t *sock){ * * Send data on a TCP socket */ -gras_error_t -gras_trp_buf_chunk_send(gras_socket_t *sock, +xbt_error_t +gras_trp_buf_chunk_send(gras_socket_t sock, const char *chunk, long int size) { - gras_error_t errcode; + xbt_error_t errcode; gras_trp_bufdata_t *data=(gras_trp_bufdata_t*)sock->bufdata; int chunk_pos=0; - GRAS_IN; + XBT_IN; /* Let underneath plugin check for direction, we work even in duplex */ - gras_assert0(size >= 0, "Cannot send a negative amount of data"); + xbt_assert0(size >= 0, "Cannot send a negative amount of data"); while (chunk_pos < size) { /* size of the chunck to receive in that shot */ @@ -217,7 +221,7 @@ gras_trp_buf_chunk_send(gras_socket_t *sock, TRY(gras_trp_buf_flush(sock)); } - GRAS_OUT; + XBT_OUT; return no_error; } @@ -226,37 +230,53 @@ gras_trp_buf_chunk_send(gras_socket_t *sock, * * Receive data on a TCP socket. */ -gras_error_t -gras_trp_buf_chunk_recv(gras_socket_t *sock, +xbt_error_t +gras_trp_buf_chunk_recv(gras_socket_t sock, char *chunk, long int size) { - gras_error_t errcode; + xbt_error_t errcode; gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)sock->plugin->data)->super; gras_trp_bufdata_t *data=sock->bufdata; long int chunck_pos = 0; /* Let underneath plugin check for direction, we work even in duplex */ - gras_assert0(sock, "Cannot recv on an NULL socket"); - gras_assert0(size >= 0, "Cannot receive a negative amount of data"); + xbt_assert0(sock, "Cannot recv on an NULL socket"); + xbt_assert0(size >= 0, "Cannot receive a negative amount of data"); - GRAS_IN; + XBT_IN; while (chunck_pos < size) { /* size of the chunck to receive in that shot */ 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 = 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, @@ -269,7 +289,7 @@ gras_trp_buf_chunk_recv(gras_socket_t *sock, data->in.pos,size - chunck_pos,size,(int)chunck_pos,chunk); } - GRAS_OUT; + XBT_OUT; return no_error; } @@ -278,21 +298,39 @@ gras_trp_buf_chunk_recv(gras_socket_t *sock, * * Make sure the data is sent */ -gras_error_t -gras_trp_buf_flush(gras_socket_t *sock) { - gras_error_t errcode; - uint32_t size; +xbt_error_t +gras_trp_buf_flush(gras_socket_t sock) { + xbt_error_t errcode; + int size; gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)sock->plugin->data)->super; gras_trp_bufdata_t *data=sock->bufdata; - GRAS_IN; - size = htonl(data->out.size); - DEBUG1("Send the size (=%d)",data->out.size); - TRY(super->chunk_send(sock,(char*) &size, 4)); + XBT_IN; + 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)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()) { + size = (int)htonl(size); + 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); - data->out.size = 0; + if (XBT_LOG_ISENABLED(trp_buf,xbt_log_priority_debug)) + hexa_print("chunck sent ",data->out.data,data->out.size); + data->out.size = gras_if_RL()?0:4; return no_error; }