src/gras/Msg/sg_msg.c

   1 /* messaging - Function related to messaging code specific to SG            */
   2
   3 /* Copyright (c) 2003-2005 Martin Quinson. All rights reserved.             */
   4
   5 /* This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the license (GNU LGPL) which comes with this package. */
   7
   8 #include "xbt/ex.h"
   9
  10 #include "gras/Virtu/virtu_sg.h"
  11
  12 #include "gras/Msg/msg_private.h"
  13
  14 #include "gras/DataDesc/datadesc_interface.h"
  15 #include "gras/Transport/transport_interface.h" /* gras_trp_chunk_send/recv */
  16 #include "gras/Transport/transport_private.h"   /* sock->data */
  17
  18 XBT_LOG_EXTERNAL_CATEGORY(gras_msg);
  19 XBT_LOG_DEFAULT_CATEGORY(gras_msg);
  20
  21 typedef void *gras_trp_bufdata_;
  22
  23 void gras_msg_send_ext(gras_socket_t sock,
  24                        e_gras_msg_kind_t kind,
  25                        unsigned long int ID,
  26                        gras_msgtype_t msgtype, void *payload)
  27 {
  28
  29   smx_action_t act;             /* simix action */
  30   gras_trp_sg_sock_data_t *sock_data;
  31   gras_hostdata_t *hd;
  32   gras_trp_procdata_t trp_remote_proc;
  33   gras_msg_procdata_t msg_remote_proc;
  34   gras_msg_t msg;               /* message to send */
  35   int whole_payload_size = 0;   /* msg->payload_size is used to memcpy the payload.
  36                                    This is used to report the load onto the simulator. It also counts the size of pointed stuff */
  37
  38   sock_data = (gras_trp_sg_sock_data_t *) sock->data;
  39
  40   hd = (gras_hostdata_t *) SIMIX_host_get_data(SIMIX_host_self());
  41
  42   xbt_assert1(!gras_socket_is_meas(sock),
  43               "Asked to send a message on the measurement socket %p", sock);
  44
  45   /*initialize gras message */
  46   msg = xbt_new(s_gras_msg_t, 1);
  47   msg->expe = sock;
  48   msg->kind = kind;
  49   msg->type = msgtype;
  50   msg->ID = ID;
  51   if (kind == e_gras_msg_kind_rpcerror) {
  52     /* error on remote host, carfull, payload is an exception */
  53     msg->payl_size = gras_datadesc_size(gras_datadesc_by_name("ex_t"));
  54     msg->payl = xbt_malloc(msg->payl_size);
  55     whole_payload_size = gras_datadesc_memcpy(gras_datadesc_by_name("ex_t"),
  56                                               payload, msg->payl);
  57   } else if (kind == e_gras_msg_kind_rpcanswer) {
  58     msg->payl_size = gras_datadesc_size(msgtype->answer_type);
  59     if (msg->payl_size)
  60       msg->payl = xbt_malloc(msg->payl_size);
  61     else
  62       msg->payl = NULL;
  63
  64     if (msgtype->answer_type)
  65       whole_payload_size = gras_datadesc_memcpy(msgtype->answer_type,
  66                                                 payload, msg->payl);
  67   } else {
  68     msg->payl_size = gras_datadesc_size(msgtype->ctn_type);
  69     msg->payl = msg->payl_size ? xbt_malloc(msg->payl_size) : NULL;
  70     if (msgtype->ctn_type)
  71       whole_payload_size = gras_datadesc_memcpy(msgtype->ctn_type,
  72                                                 payload, msg->payl);
  73   }
  74
  75   /* put the selectable socket on the queue */
  76   trp_remote_proc = (gras_trp_procdata_t)
  77     gras_libdata_by_name_from_remote("gras_trp", sock_data->to_process);
  78
  79   xbt_queue_push(trp_remote_proc->msg_selectable_sockets, &sock);
  80
  81   /* put message on msg_queue */
  82   msg_remote_proc = (gras_msg_procdata_t)
  83     gras_libdata_by_name_from_remote("gras_msg", sock_data->to_process);
  84   xbt_fifo_push(msg_remote_proc->msg_to_receive_queue, msg);
  85
  86   /* wait for the receiver */
  87   SIMIX_cond_wait(sock_data->cond, sock_data->mutex);
  88
  89   /* creates simix action and waits its ends, waits in the sender host
  90      condition */
  91   act = SIMIX_action_communicate(SIMIX_host_self(),
  92                                  sock_data->to_host, msgtype->name,
  93                                  (double) whole_payload_size, -1);
  94   SIMIX_register_action_to_condition(act, sock_data->cond);
  95
  96   VERB5("Sending to %s(%s) a message type '%s' kind '%s' ID %lu",
  97         SIMIX_host_get_name(sock_data->to_host),
  98         SIMIX_process_get_name(sock_data->to_process),
  99         msg->type->name, e_gras_msg_kind_names[msg->kind], msg->ID);
 100
 101   SIMIX_cond_wait(sock_data->cond, sock_data->mutex);
 102   SIMIX_unregister_action_to_condition(act, sock_data->cond);
 103   /* error treatmeant (FIXME) */
 104
 105   /* cleanup structures */
 106   SIMIX_action_destroy(act);
 107   SIMIX_mutex_unlock(sock_data->mutex);
 108
 109   VERB0("Message sent");
 110
 111 }
 112
 113 /*
 114  * receive the next message on the given socket.
 115  */
 116 void gras_msg_recv(gras_socket_t sock, gras_msg_t msg)
 117 {
 118
 119   gras_trp_sg_sock_data_t *sock_data;
 120   gras_trp_sg_sock_data_t *remote_sock_data;
 121   gras_hostdata_t *remote_hd;
 122   gras_msg_t msg_got;
 123   gras_msg_procdata_t msg_procdata =
 124     (gras_msg_procdata_t) gras_libdata_by_name("gras_msg");
 125
 126   xbt_assert1(!gras_socket_is_meas(sock),
 127               "Asked to receive a message on the measurement socket %p",
 128               sock);
 129
 130   xbt_assert0(msg, "msg is an out parameter of gras_msg_recv...");
 131
 132   sock_data = (gras_trp_sg_sock_data_t *) sock->data;
 133   remote_sock_data =
 134     ((gras_trp_sg_sock_data_t *) sock->data)->to_socket->data;
 135   DEBUG3("Remote host %s, Remote Port: %d Local port %d",
 136          SIMIX_host_get_name(sock_data->to_host), sock->peer_port,
 137          sock->port);
 138   remote_hd = (gras_hostdata_t *) SIMIX_host_get_data(sock_data->to_host);
 139
 140   if (xbt_fifo_size(msg_procdata->msg_to_receive_queue) == 0) {
 141     THROW_IMPOSSIBLE;
 142   }
 143   DEBUG1("Size msg_to_receive buffer: %d",
 144          xbt_fifo_size(msg_procdata->msg_to_receive_queue));
 145   msg_got = xbt_fifo_shift(msg_procdata->msg_to_receive_queue);
 146
 147   SIMIX_mutex_lock(remote_sock_data->mutex);
 148   /* ok, I'm here, you can continuate the communication */
 149   SIMIX_cond_signal(remote_sock_data->cond);
 150
 151   /* wait for communication end */
 152   SIMIX_cond_wait(remote_sock_data->cond, remote_sock_data->mutex);
 153
 154   msg_got->expe = msg->expe;
 155   memcpy(msg, msg_got, sizeof(s_gras_msg_t));
 156   xbt_free(msg_got);
 157   SIMIX_mutex_unlock(remote_sock_data->mutex);
 158
 159   VERB3("Received a message type '%s' kind '%s' ID %lu",        // from %s",
 160         msg->type->name, e_gras_msg_kind_names[msg->kind], msg->ID);
 161 }