X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/e0d198d9da38e08cede43a169bde62366bf0fbe9..778f65057da68465382593cd036b6ee59ada54e9:/src/gras/Msg/sg_msg.c diff --git a/src/gras/Msg/sg_msg.c b/src/gras/Msg/sg_msg.c index 2aeed71112..67b2d8b32b 100644 --- a/src/gras/Msg/sg_msg.c +++ b/src/gras/Msg/sg_msg.c @@ -20,118 +20,105 @@ 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) { +void gras_msg_send_ext(gras_socket_t sock, + e_gras_msg_kind_t kind, + unsigned long int ID, + 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; + int whole_payload_size=0; /* msg->payload_size is used to memcpy the payload. + This is used to report the load onto the simulator. It also counts the size of pointed stuff */ + char *name; 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->ID = ID; + + if (kind == e_gras_msg_kind_rpcerror) { + /* error on remote host, carfull, payload is an exception */ + msg->payl_size=gras_datadesc_size(gras_datadesc_by_name("ex_t")); + msg->payl=xbt_malloc(msg->payl_size); + whole_payload_size = gras_datadesc_copy(gras_datadesc_by_name("ex_t"), + payload,msg->payl); + } else if (kind == e_gras_msg_kind_rpcanswer) { + msg->payl_size=gras_datadesc_size(msgtype->answer_type); + msg->payl=xbt_malloc(msg->payl_size); + if (msgtype->answer_type) + whole_payload_size = gras_datadesc_copy(msgtype->answer_type, + payload, msg->payl); + } else { + msg->payl_size=gras_datadesc_size(msgtype->ctn_type); + msg->payl=msg->payl_size?xbt_malloc(msg->payl_size):NULL; + if (msgtype->ctn_type) + whole_payload_size = gras_datadesc_copy(msgtype->ctn_type, + payload, msg->payl); + } - 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"); + msg->kind = kind; + if (XBT_LOG_ISENABLED(gras_msg,xbt_log_priority_verbose)) { + asprintf(&name,"type:'%s';kind:'%s';ID %lu from %s:%d to %s:%d", + msg->type->name, e_gras_msg_kind_names[msg->kind], msg->ID, + gras_os_myname(),gras_os_myport(), + gras_socket_peer_name(sock), gras_socket_peer_port(sock)); + task=MSG_task_create(name,0, + ((double)whole_payload_size),msg); + free(name); + } else { + task=MSG_task_create(msg->type->name,0, + ((double)whole_payload_size),msg); + } + + + DEBUG1("Prepare to send a message to %s", + MSG_host_get_name (sock_data->to_host)); + if (MSG_task_put_with_timeout(task, sock_data->to_host,sock_data->to_chan,60.0) != MSG_OK) + THROW0(system_error,0,"Problem during the MSG_task_put with timeout 60"); + + VERB5("Sent to %s(%d) a message type '%s' kind '%s' ID %lu", + MSG_host_get_name(sock_data->to_host),sock_data->to_PID, + msg->type->name, + e_gras_msg_kind_names[msg->kind], + msg->ID); } /* * 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) { + gras_msg_t msg) { 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"); + gras_msg_t msg_got; + gras_trp_procdata_t pd=(gras_trp_procdata_t)gras_libdata_by_name("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()"); + xbt_assert0(msg,"msg is an out parameter of gras_msg_recv..."); - 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()"); + if (MSG_task_get(&task, pd->chan) != MSG_OK) + THROW0(system_error,0,"Error in MSG_task_get()"); + msg_got=MSG_task_get_data(task); - /* - 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; + msg_got->expe= msg->expe; + memcpy(msg,msg_got,sizeof(s_gras_msg_t)); - 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); - } + free(msg_got); + if (MSG_task_destroy(task) != MSG_OK) + THROW0(system_error,0,"Error in MSG_task_destroy()"); - 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; - } -*/ + VERB3("Received a message type '%s' kind '%s' ID %lu",// from %s", + msg->type->name, + e_gras_msg_kind_names[msg->kind], + msg->ID); }