Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
fead6dfa6f28ec966b5ed6f12b8eac0dd9131a9b
[simgrid.git] / src / gras / Msg / rl_msg.c
1 /* messaging - Function related to messaging code specific to RL            */
2
3 /* Copyright (c) 2005, 2006, 2007, 2009, 2010. The SimGrid Team.
4  * All rights reserved.                                                     */
5
6 /* This program is free software; you can redistribute it and/or modify it
7  * under the terms of the license (GNU LGPL) which comes with this package. */
8
9 #include "xbt/ex.h"
10 #include "gras/Msg/msg_private.h"
11
12 #include "gras/DataDesc/datadesc_interface.h"
13 #include "gras/Transport/transport_interface.h" /* gras_trp_send/recv */
14
15 XBT_LOG_EXTERNAL_CATEGORY(gras_msg);
16 XBT_LOG_DEFAULT_CATEGORY(gras_msg);
17
18 void gras_msg_recv(gras_socket_t sock, gras_msg_t msg);
19
20 gras_msg_t gras_msg_recv_any(void)
21 {
22   gras_msg_t msg = xbt_new0(s_gras_msg_t, 1);
23   msg->expe = gras_trp_select(-1);
24   XBT_DEBUG("Select returned something");
25   gras_msg_recv(msg->expe, msg);
26   return msg;
27 }
28
29 void gras_msg_send_ext(gras_socket_t sock,
30                        e_gras_msg_kind_t kind,
31                        unsigned long int ID,
32                        gras_msgtype_t msgtype, void *payload)
33 {
34
35   static gras_datadesc_type_t string_type = NULL;
36   static gras_datadesc_type_t ulong_type = NULL;
37   char c_kind = (char) kind;
38
39   xbt_assert0(msgtype, "Cannot send the NULL message");
40
41   if (!string_type) {
42     string_type = gras_datadesc_by_name("string");
43     xbt_assert(string_type);
44   }
45   if (!ulong_type) {
46     ulong_type = gras_datadesc_by_name("unsigned long int");
47     xbt_assert(ulong_type);
48   }
49
50   XBT_DEBUG("send '%s' to %s:%d", msgtype->name,
51          gras_socket_peer_name(sock), gras_socket_peer_port(sock));
52   gras_trp_send(sock, _GRAS_header, 6, 1 /* stable */ );
53   gras_trp_send(sock, &c_kind, 1, 1 /* stable */ );
54   switch (kind) {
55   case e_gras_msg_kind_oneway:
56     break;
57
58   case e_gras_msg_kind_rpccall:
59   case e_gras_msg_kind_rpcanswer:
60   case e_gras_msg_kind_rpcerror:
61     gras_datadesc_send(sock, ulong_type, &ID);
62     break;
63
64   default:
65     THROWF(unknown_error, 0, "Unknown msg kind %d", kind);
66   }
67
68   gras_datadesc_send(sock, string_type, &msgtype->name);
69   if (kind == e_gras_msg_kind_rpcerror) {
70     /* error on remote host, carfull, payload is an exception */
71     gras_datadesc_send(sock, gras_datadesc_by_name("ex_t"), payload);
72   } else if (kind == e_gras_msg_kind_rpcanswer) {
73     if (msgtype->answer_type)
74       gras_datadesc_send(sock, msgtype->answer_type, payload);
75   } else {
76     /* regular message */
77     if (msgtype->ctn_type)
78       gras_datadesc_send(sock, msgtype->ctn_type, payload);
79   }
80
81   gras_trp_flush(sock);
82 }
83
84 const char *hexa_str(unsigned char *data, int size, int downside);
85
86
87 /*
88  * receive the next message on the given socket.
89  */
90 void gras_msg_recv(gras_socket_t sock, gras_msg_t msg)
91 {
92
93   xbt_ex_t e;
94   static gras_datadesc_type_t string_type = NULL;
95   static gras_datadesc_type_t ulong_type = NULL;
96   char header[6];
97   int cpt;
98   int r_arch;
99   char *msg_name = NULL;
100   char c_kind;
101
102   xbt_assert1(!gras_socket_is_meas(sock),
103               "Asked to receive a message on the measurement socket %p",
104               sock);
105   if (!string_type) {
106     string_type = gras_datadesc_by_name("string");
107     xbt_assert(string_type);
108   }
109   if (!ulong_type) {
110     ulong_type = gras_datadesc_by_name("unsigned long int");
111     xbt_assert(ulong_type);
112   }
113
114
115   TRY {
116     gras_trp_recv(sock, header, 6);
117     gras_trp_recv(sock, &c_kind, 1);
118     msg->kind = (e_gras_msg_kind_t) c_kind;
119   }
120   CATCH(e) {
121     RETHROWF
122         ("Exception caught while trying to get the message header: %s");
123   }
124
125   for (cpt = 0; cpt < 4; cpt++)
126     if (header[cpt] != _GRAS_header[cpt])
127       THROWF(mismatch_error, 0,
128              "Incoming bytes do not look like a GRAS message (header='%s'  not '%.4s')",
129              hexa_str((unsigned char *) header, 4, 0), _GRAS_header);
130   if (header[4] != _GRAS_header[4])
131     THROWF(mismatch_error, 0, "GRAS protocol mismatch (got %d, use %d)",
132            (int) header[4], (int) _GRAS_header[4]);
133   r_arch = (int) header[5];
134
135   switch (msg->kind) {
136   case e_gras_msg_kind_oneway:
137     break;
138
139   case e_gras_msg_kind_rpccall:
140   case e_gras_msg_kind_rpcanswer:
141   case e_gras_msg_kind_rpcerror:
142     gras_datadesc_recv(sock, ulong_type, r_arch, &msg->ID);
143     break;
144
145   default:
146     THROW_IMPOSSIBLE;
147   }
148
149   gras_datadesc_recv(sock, string_type, r_arch, &msg_name);
150   XBT_DEBUG
151       ("Handle an incoming message '%s' (%s) using protocol %d (remote is %s)",
152        msg_name, e_gras_msg_kind_names[msg->kind], (int) header[4],
153        gras_datadesc_arch_name(r_arch));
154
155   TRY {
156     msg->type =
157         (gras_msgtype_t) xbt_set_get_by_name(_gras_msgtype_set, msg_name);
158   } CATCH(e) {
159     /* FIXME: Survive unknown messages */
160     if (e.category == not_found_error) {
161       xbt_ex_free(e);
162       THROWF(not_found_error, 0,
163              "Received an unknown message: %s (FIXME: should survive to these)",
164              msg_name);
165     } else
166       RETHROWF
167           ("Exception caught while retrieving the type associated to messages '%s' : %s",
168            msg_name);
169   }
170   free(msg_name);
171
172   if (msg->kind == e_gras_msg_kind_rpcerror) {
173     /* error on remote host. Carfull with that exception, eugene */
174     msg->payl_size = gras_datadesc_size(gras_datadesc_by_name("ex_t"));
175     msg->payl = xbt_malloc(msg->payl_size);
176     gras_datadesc_recv(sock, gras_datadesc_by_name("ex_t"), r_arch,
177                        msg->payl);
178
179   } else if (msg->kind == e_gras_msg_kind_rpcanswer) {
180     /* answer to RPC */
181     if (msg->type->answer_type) {
182       msg->payl_size = gras_datadesc_size(msg->type->answer_type);
183       xbt_assert2(msg->payl_size > 0,
184                   "%s %s",
185                   "Dynamic array as payload is forbided for now (FIXME?).",
186                   "Reference to dynamic array is allowed.");
187       msg->payl = xbt_malloc(msg->payl_size);
188       gras_datadesc_recv(sock, msg->type->answer_type, r_arch, msg->payl);
189     } else {
190       msg->payl = NULL;
191       msg->payl_size = 0;
192     }
193   } else {
194     /* regular message */
195     if (msg->type->ctn_type) {
196       msg->payl_size = gras_datadesc_size(msg->type->ctn_type);
197       xbt_assert2(msg->payl_size > 0,
198                   "%s %s",
199                   "Dynamic array as payload is forbided for now (FIXME?).",
200                   "Reference to dynamic array is allowed.");
201       msg->payl = xbt_malloc(msg->payl_size);
202       gras_datadesc_recv(sock, msg->type->ctn_type, r_arch, msg->payl);
203     } else {
204       msg->payl = NULL;
205       msg->payl_size = 0;
206     }
207   }
208 }