X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/6ff34272a5b3ded0fcbbb93d039b960d043e7520..b0b60e5b550acd203efd832a918aa0f3a43b4a8f:/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 7826584779..ed64bc3244 100644 --- a/src/gras/Transport/transport_plugin_buf.c +++ b/src/gras/Transport/transport_plugin_buf.c @@ -2,45 +2,47 @@ /* 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 "xbt/ex.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)"); + +static gras_trp_plugin_t _buf_super; + /*** *** 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 */ + +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); -gras_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); -gras_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); +void gras_trp_buf_chunk_recv(gras_socket_t sd, + char *data, + unsigned long int size); +void gras_trp_buf_flush(gras_socket_t sock); /*** @@ -48,7 +50,7 @@ gras_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; /*** @@ -56,7 +58,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; @@ -67,41 +69,36 @@ struct gras_trp_bufdata_{ int buffsize; }; -gras_error_t gras_trp_buf_init_sock(gras_socket_t *sock) { - gras_trp_bufdata_t *data=gras_new(gras_trp_bufdata_t,1); +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; - data->in.size = 0; - // data->buffsize = 32 * 1024 - 4; /* default socket buffsize (32k) - headers */ + XBT_IN; data->buffsize = 100 * 1024 ; /* 100k */ - if (!(data->in.data = (char*)gras_malloc(data->buffsize))) - RAISE_MALLOC; + data->in.size = 0; + data->in.data = xbt_malloc(data->buffsize); data->in.pos = 0; /* useless, indeed, since size==pos */ - data->out.size = 0; - if (!(data->out.data = (char*)gras_malloc(data->buffsize))) - RAISE_MALLOC; - data->out.pos = 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 = data->out.size; + sock->bufdata = data; - return no_error; } - /*** *** Code ***/ -gras_error_t -gras_trp_buf_setup(gras_trp_plugin_t *plug) { - gras_error_t errcode; - gras_trp_buf_plug_data_t *data =gras_new(gras_trp_buf_plug_data_t,1); - if (!data) - RAISE_MALLOC; - - GRAS_IN; - TRY(gras_trp_plugin_get_by_name(gras_if_RL() ? "tcp" : "sg", - &(data->super))); +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; + _buf_super = gras_trp_plugin_get_by_name(gras_if_RL() ? "tcp" : "sg"); + DEBUG1("Derivate a buffer plugin from %s",gras_if_RL() ? "tcp" : "sg"); plug->socket_client = gras_trp_buf_socket_client; @@ -116,23 +113,15 @@ gras_trp_buf_setup(gras_trp_plugin_t *plug) { plug->data = (void*)data; plug->exit = NULL; - - 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; - 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){ - GRAS_IN; - TRY(super->socket_client(super,host,port,sock)); + XBT_IN; + _buf_super->socket_client(_buf_super,sock); sock->plugin = self; - TRY(gras_trp_buf_init_sock(sock)); - - return no_error; + gras_trp_buf_init_sock(sock); } /** @@ -140,65 +129,67 @@ 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; - gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)self->data)->super; - - GRAS_IN; - TRY(super->socket_server(super,port,sock)); +void gras_trp_buf_socket_server(gras_trp_plugin_t self, + /* OUT */ gras_socket_t sock){ + + XBT_IN; + _buf_super->socket_server(_buf_super,sock); sock->plugin = self; - TRY(gras_trp_buf_init_sock(sock)); - return no_error; + gras_trp_buf_init_sock(sock); } -gras_error_t -gras_trp_buf_socket_accept(gras_socket_t *sock, - gras_socket_t **dst) { - gras_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; - GRAS_IN; - TRY(super->socket_accept(sock,dst)); - (*dst)->plugin = sock->plugin; - TRY(gras_trp_buf_init_sock(*dst)); - return no_error; + XBT_IN; + 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; +void gras_trp_buf_socket_close(gras_socket_t sock){ 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); + _buf_super->socket_close(sock); } /** * gras_trp_buf_chunk_send: * - * Send data on a TCP socket + * Send data on a buffered socket */ -gras_error_t -gras_trp_buf_chunk_send(gras_socket_t *sock, +void +gras_trp_buf_chunk_send(gras_socket_t sock, const char *chunk, - long int size) { + unsigned long int size) { - gras_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 */ @@ -216,49 +207,68 @@ 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); } - GRAS_OUT; - return no_error; + XBT_OUT; } /** * gras_trp_buf_chunk_recv: * - * Receive data on a TCP socket. + * Receive data on a buffered socket. */ -gras_error_t -gras_trp_buf_chunk_recv(gras_socket_t *sock, +void +gras_trp_buf_chunk_recv(gras_socket_t sock, char *chunk, - long int size) { + unsigned long int size) { - gras_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 */ - 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); - TRY(super->chunk_recv(sock, data->in.data, data->in.size)); - data->in.pos=0; + int nextsize; + if (gras_if_RL()) { + DEBUG0("Recv the size"); + TRY { + _buf_super->chunk_recv(sock,(char*)&nextsize, 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); + } else { + data->in.size = -1; + } + + _buf_super->chunk_recv(sock, data->in.data, data->in.size); + + 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",(unsigned char *) 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,8 +281,7 @@ gras_trp_buf_chunk_recv(gras_socket_t *sock, data->in.pos,size - chunck_pos,size,(int)chunck_pos,chunk); } - GRAS_OUT; - return no_error; + XBT_OUT; } /** @@ -280,21 +289,36 @@ 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; - gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)sock->plugin->data)->super; +void +gras_trp_buf_flush(gras_socket_t sock) { + 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) == (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; + } + + 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); + _buf_super->chunk_send(sock,(char*) &size, 4); + } else { + memcpy(data->out.data, &size, 4); + } + - GRAS_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); - data->out.size = 0; - return no_error; + 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 = gras_if_RL()?0:4; }