From: mquinson Date: Mon, 5 Sep 2005 21:53:11 +0000 (+0000) Subject: Move msg_send/recv to a RL specific file, so that we can provide a SG specific versio... X-Git-Tag: v3.3~3675 X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/commitdiff_plain/710e2ad7303dce00b7a72feebdc2cec6823806b6 Move msg_send/recv to a RL specific file, so that we can provide a SG specific version (and do so) git-svn-id: svn+ssh://scm.gforge.inria.fr/svn/simgrid/simgrid/trunk@1682 48e7efb5-ca39-0410-a469-dd3cf9ba447f --- diff --git a/src/gras/Msg/msg.c b/src/gras/Msg/msg.c index 6ec3e4d56e..91c9937a9f 100644 --- a/src/gras/Msg/msg.c +++ b/src/gras/Msg/msg.c @@ -9,17 +9,17 @@ #include "xbt/ex.h" #include "gras/Msg/msg_private.h" -#include "gras/DataDesc/datadesc_interface.h" -#include "gras/Transport/transport_interface.h" /* gras_trp_chunk_send/recv */ #include "gras/Virtu/virtu_interface.h" +#include "gras/DataDesc/datadesc_interface.h" +#include "gras/Transport/transport_interface.h" /* gras_select */ #define MIN(a,b) ((a) < (b) ? (a) : (b)) XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gras_msg,gras,"High level messaging"); xbt_set_t _gras_msgtype_set = NULL; -static char GRAS_header[6]; static char *make_namev(const char *name, short int ver); +char _GRAS_header[6]; /* * Creating procdata for this module @@ -66,9 +66,9 @@ void gras_msg_init(void) { _gras_msgtype_set = xbt_set_new(); - memcpy(GRAS_header,"GRAS", 4); - GRAS_header[4]=GRAS_PROTOCOL_VERSION; - GRAS_header[5]=(char)GRAS_THISARCH; + memcpy(_GRAS_header,"GRAS", 4); + _GRAS_header[4]=GRAS_PROTOCOL_VERSION; + _GRAS_header[5]=(char)GRAS_THISARCH; } /* @@ -194,97 +194,9 @@ gras_msgtype_t gras_msgtype_by_namev(const char *name, return res; } - -/** \brief Send the data pointed by \a payload as a message of type - * \a msgtype to the peer \a sock */ -void -gras_msg_send(gras_socket_t sock, - gras_msgtype_t msgtype, - void *payload) { - - static gras_datadesc_type_t string_type=NULL; - - if (!msgtype) - THROW0(arg_error,0, - "Cannot send the NULL message (did msgtype_by_name fail?)"); - - if (!string_type) { - string_type = gras_datadesc_by_name("string"); - xbt_assert(string_type); - } - - DEBUG3("send '%s' to %s:%d", msgtype->name, - gras_socket_peer_name(sock),gras_socket_peer_port(sock)); - gras_trp_chunk_send(sock, GRAS_header, 6); - - gras_datadesc_send(sock, string_type, &msgtype->name); - if (msgtype->ctn_type) - gras_datadesc_send(sock, msgtype->ctn_type, payload); - gras_trp_flush(sock); -} -/* - * receive the next message on the given socket. - */ -void -gras_msg_recv(gras_socket_t sock, - gras_msgtype_t *msgtype, - void **payload, - int *payload_size) { - - xbt_ex_t e; - static gras_datadesc_type_t string_type=NULL; - char header[6]; - int cpt; - int r_arch; - char *msg_name=NULL; - - xbt_assert1(!gras_socket_is_meas(sock), - "Asked to receive a message on the measurement socket %p", sock); - if (!string_type) { - string_type=gras_datadesc_by_name("string"); - xbt_assert(string_type); - } - - TRY { - gras_trp_chunk_recv(sock, header, 6); - } CATCH(e) { - RETHROW1("Exception caught while trying to get the mesage header on socket %p : %s", - sock); - } - - for (cpt=0; cpt<4; cpt++) - if (header[cpt] != GRAS_header[cpt]) - THROW2(mismatch_error,0, - "Incoming bytes do not look like a GRAS message (header='%.4s' not '%.4s')",header,GRAS_header); - if (header[4] != GRAS_header[4]) - THROW2(mismatch_error,0,"GRAS protocol mismatch (got %d, use %d)", - (int)header[4], (int)GRAS_header[4]); - r_arch = (int)header[5]; - DEBUG2("Handle an incoming message using protocol %d (remote is %s)", - (int)header[4],gras_datadesc_arch_name(r_arch)); - - gras_datadesc_recv(sock, string_type, r_arch, &msg_name); - TRY { - *msgtype = (gras_msgtype_t)xbt_set_get_by_name(_gras_msgtype_set,msg_name); - } CATCH(e) { - /* FIXME: Survive unknown messages */ - RETHROW1("Exception caught while retrieving the type associated to messages '%s' : %s", - msg_name); - } - free(msg_name); - - if ((*msgtype)->ctn_type) { - *payload_size=gras_datadesc_size((*msgtype)->ctn_type); - xbt_assert2(*payload_size > 0, - "%s %s", - "Dynamic array as payload is forbided for now (FIXME?).", - "Reference to dynamic array is allowed."); - *payload = xbt_malloc(*payload_size); - gras_datadesc_recv(sock, (*msgtype)->ctn_type, r_arch, *payload); - } else { - *payload = NULL; - *payload_size = 0; - } +/** @brief retrive an existing message type from its name and version. */ +gras_msgtype_t gras_msgtype_by_id(int id) { + return (gras_msgtype_t)xbt_set_get_by_id(_gras_msgtype_set, id); } /** \brief Waits for a message to come in over a given socket. diff --git a/src/gras/Msg/msg_private.h b/src/gras/Msg/msg_private.h index 7923f49504..6e8b0ad84e 100644 --- a/src/gras/Msg/msg_private.h +++ b/src/gras/Msg/msg_private.h @@ -28,7 +28,8 @@ #include "gras/Msg/msg_interface.h" - +extern char _GRAS_header[6]; + /** @brief Message instance */ typedef struct { gras_socket_t expeditor; diff --git a/src/gras/Msg/rl_msg.c b/src/gras/Msg/rl_msg.c new file mode 100644 index 0000000000..b39e7db01e --- /dev/null +++ b/src/gras/Msg/rl_msg.c @@ -0,0 +1,109 @@ +/* $Id$ */ + +/* messaging - Function related to messaging code specific to RL */ + +/* Copyright (c) 2003-2005 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. */ + +#include "xbt/ex.h" +#include "gras/Msg/msg_private.h" + +#include "gras/DataDesc/datadesc_interface.h" +#include "gras/Transport/transport_interface.h" /* gras_trp_chunk_send/recv */ + +XBT_LOG_EXTERNAL_CATEGORY(gras_msg); +XBT_LOG_DEFAULT_CATEGORY(gras_msg); + +/** \brief Send the data pointed by \a payload as a message of type + * \a msgtype to the peer \a sock */ +void +gras_msg_send(gras_socket_t sock, + gras_msgtype_t msgtype, + void *payload) { + + static gras_datadesc_type_t string_type=NULL; + + if (!msgtype) + THROW0(arg_error,0, + "Cannot send the NULL message (did msgtype_by_name fail?)"); + + if (!string_type) { + string_type = gras_datadesc_by_name("string"); + xbt_assert(string_type); + } + + DEBUG3("send '%s' to %s:%d", msgtype->name, + gras_socket_peer_name(sock),gras_socket_peer_port(sock)); + gras_trp_chunk_send(sock, _GRAS_header, 6); + + gras_datadesc_send(sock, string_type, &msgtype->name); + if (msgtype->ctn_type) + gras_datadesc_send(sock, msgtype->ctn_type, payload); + gras_trp_flush(sock); +} +/* + * receive the next message on the given socket. + */ +void +gras_msg_recv(gras_socket_t sock, + gras_msgtype_t *msgtype, + void **payload, + int *payload_size) { + + xbt_ex_t e; + static gras_datadesc_type_t string_type=NULL; + char header[6]; + int cpt; + int r_arch; + char *msg_name=NULL; + + xbt_assert1(!gras_socket_is_meas(sock), + "Asked to receive a message on the measurement socket %p", sock); + if (!string_type) { + string_type=gras_datadesc_by_name("string"); + xbt_assert(string_type); + } + + TRY { + gras_trp_chunk_recv(sock, header, 6); + } CATCH(e) { + RETHROW1("Exception caught while trying to get the mesage header on socket %p: %s", + sock); + } + + for (cpt=0; cpt<4; cpt++) + if (header[cpt] != _GRAS_header[cpt]) + THROW2(mismatch_error,0, + "Incoming bytes do not look like a GRAS message (header='%.4s' not '%.4s')",header,_GRAS_header); + if (header[4] != _GRAS_header[4]) + THROW2(mismatch_error,0,"GRAS protocol mismatch (got %d, use %d)", + (int)header[4], (int)_GRAS_header[4]); + r_arch = (int)header[5]; + DEBUG2("Handle an incoming message using protocol %d (remote is %s)", + (int)header[4],gras_datadesc_arch_name(r_arch)); + + gras_datadesc_recv(sock, string_type, r_arch, &msg_name); + TRY { + *msgtype = (gras_msgtype_t)xbt_set_get_by_name(_gras_msgtype_set,msg_name); + } CATCH(e) { + /* FIXME: Survive unknown messages */ + RETHROW1("Exception caught while retrieving the type associated to messages '%s' : %s", + msg_name); + } + free(msg_name); + + if ((*msgtype)->ctn_type) { + *payload_size=gras_datadesc_size((*msgtype)->ctn_type); + xbt_assert2(*payload_size > 0, + "%s %s", + "Dynamic array as payload is forbided for now (FIXME?).", + "Reference to dynamic array is allowed."); + *payload = xbt_malloc(*payload_size); + gras_datadesc_recv(sock, (*msgtype)->ctn_type, r_arch, *payload); + } else { + *payload = NULL; + *payload_size = 0; + } +} diff --git a/src/gras/Msg/sg_msg.c b/src/gras/Msg/sg_msg.c new file mode 100644 index 0000000000..94d57aa7d2 --- /dev/null +++ b/src/gras/Msg/sg_msg.c @@ -0,0 +1,138 @@ +/* $Id$ */ + +/* messaging - Function related to messaging code specific to SG */ + +/* Copyright (c) 2003-2005 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. */ + +#include "xbt/ex.h" + +//#include "msg/msg.h" +#include "gras/Virtu/virtu_sg.h" + +#include "gras/Msg/msg_private.h" + +#include "gras/DataDesc/datadesc_interface.h" +#include "gras/Transport/transport_interface.h" /* gras_trp_chunk_send/recv */ +#include "gras/Transport/transport_private.h" /* sock->data */ + +XBT_LOG_EXTERNAL_CATEGORY(gras_msg); +XBT_LOG_DEFAULT_CATEGORY(gras_msg); + +/** \brief Send the data pointed by \a payload as a message of type + * \a msgtype to the peer \a sock */ +void +gras_msg_send(gras_socket_t sock, + gras_msgtype_t msgtype, + void *payload) { + + m_task_t task=NULL; + gras_trp_sg_sock_data_t *sock_data = (gras_trp_sg_sock_data_t *)sock->data; + gras_msg_t msg; + + xbt_assert1(!gras_socket_is_meas(sock), + "Asked to send a message on the measurement socket %p", sock); + + msg=xbt_new0(s_gras_msg_t,1); + msg->type=msgtype; + + msg->payload=xbt_malloc(gras_datadesc_size(msgtype->ctn_type)); + if (msgtype->ctn_type) + msg->payload_size=gras_datadesc_copy(msgtype->ctn_type,payload,msg->payload); + + task=MSG_task_create(msgtype->name,0, + ((double)msg->payload_size)/(1024.0*1024.0),msg); + + if (MSG_task_put(task, sock_data->to_host,sock_data->to_chan) != MSG_OK) + THROW0(system_error,0,"Problem during the MSG_task_put"); + +} +/* + * receive the next message on the given socket. + */ +void +gras_msg_recv(gras_socket_t sock, + gras_msgtype_t *msgtype, + void **payload, + int *payload_size) { + + m_task_t task=NULL; + /* gras_trp_sg_sock_data_t *sock_data = (gras_trp_sg_sock_data_t *)sock->data;*/ + gras_msg_t msg; + gras_trp_procdata_t pd=(gras_trp_procdata_t)gras_libdata_get("gras_trp"); + + xbt_assert1(!gras_socket_is_meas(sock), + "Asked to receive a message on the measurement socket %p", sock); + + if (MSG_task_get(&task, pd->chan) != MSG_OK) + THROW0(system_error,0,"Error in MSG_task_get()"); + + msg = MSG_task_get_data(task); + *msgtype = gras_msgtype_by_id(msg->type->code); + *payload = msg->payload; + *payload_size = msg->payload_size; + + free(msg); + if (MSG_task_destroy(task) != MSG_OK) + THROW0(system_error,0,"Error in MSG_task_destroy()"); + + + /* + THROW_UNIMPLEMENTED; + + xbt_ex_t e; + static gras_datadesc_type_t string_type=NULL; + char header[6]; + int cpt; + int r_arch; + char *msg_name=NULL; + + if (!string_type) { + string_type=gras_datadesc_by_name("string"); + xbt_assert(string_type); + } + + TRY { + gras_trp_chunk_recv(sock, header, 6); + } CATCH(e) { + RETHROW1("Exception caught while trying to get the mesage header on socket %p: %s", + sock); + } + + for (cpt=0; cpt<4; cpt++) + if (header[cpt] != _GRAS_header[cpt]) + THROW2(mismatch_error,0, + "Incoming bytes do not look like a GRAS message (header='%.4s' not '%.4s')",header,_GRAS_header); + if (header[4] != _GRAS_header[4]) + THROW2(mismatch_error,0,"GRAS protocol mismatch (got %d, use %d)", + (int)header[4], (int)_GRAS_header[4]); + r_arch = (int)header[5]; + DEBUG2("Handle an incoming message using protocol %d (remote is %s)", + (int)header[4],gras_datadesc_arch_name(r_arch)); + + gras_datadesc_recv(sock, string_type, r_arch, &msg_name); + TRY { + *msgtype = (gras_msgtype_t)xbt_set_get_by_name(_gras_msgtype_set,msg_name); + } CATCH(e) { + / * FIXME: Survive unknown messages * / + RETHROW1("Exception caught while retrieving the type associated to messages '%s' : %s", + msg_name); + } + free(msg_name); + + if ((*msgtype)->ctn_type) { + *payload_size=gras_datadesc_size((*msgtype)->ctn_type); + xbt_assert2(*payload_size > 0, + "%s %s", + "Dynamic array as payload is forbided for now (FIXME?).", + "Reference to dynamic array is allowed."); + *payload = xbt_malloc(*payload_size); + gras_datadesc_recv(sock, (*msgtype)->ctn_type, r_arch, *payload); + } else { + *payload = NULL; + *payload_size = 0; + } +*/ +}