src/gras/Msg/rpc.c

   1 /* rpc - RPC implementation on top of GRAS messages                         */
   2
   3 /* Copyright (c) 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 "gras/Msg/msg_private.h"
   9
  10 xbt_set_t _gras_rpctype_set = NULL;
  11 xbt_dynar_t _gras_rpc_cancelled = NULL;
  12
  13 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gras_msg_rpc, gras_msg, "RPC mecanism");
  14
  15
  16 /** @brief declare a new versionned RPC type of the given name and payloads
  17  *
  18  * @param name: name as it should be used for logging messages (must be uniq)
  19  * @param payload_request: datatype of request
  20  * @param payload_answer: datatype of answer
  21  *
  22  * Registers a new RPC message to the GRAS mechanism. RPC are constituted of a pair
  23  * of messages.
  24  */
  25 void
  26 gras_msgtype_declare_rpc(const char *name,
  27                          gras_datadesc_type_t payload_request,
  28                          gras_datadesc_type_t payload_answer)
  29 {
  30
  31   gras_msgtype_declare_ext(name, 0,
  32                            e_gras_msg_kind_rpccall,
  33                            payload_request, payload_answer);
  34
  35 }
  36
  37 /** @brief declare a new versionned RPC type of the given name and payloads
  38  *
  39  * @param name: name as it should be used for logging messages (must be uniq)
  40  * @param version: something like versionning symbol
  41  * @param payload_request: datatype of request
  42  * @param payload_answer: datatype of answer
  43  *
  44  * Registers a new RPC message to the GRAS mechanism. RPC are constituted of a pair
  45  * of messages.
  46  *
  47  * Use this version instead of gras_rpctype_declare when you change the
  48  * semantic or syntax of a message and want your programs to be able to deal
  49  * with both versions. Internally, each will be handled as an independent
  50  * message type, so you can register differents for each of them.
  51  */
  52 void
  53 gras_msgtype_declare_rpc_v(const char *name,
  54                            short int version,
  55                            gras_datadesc_type_t payload_request,
  56                            gras_datadesc_type_t payload_answer)
  57 {
  58
  59   gras_msgtype_declare_ext(name, version,
  60                            e_gras_msg_kind_rpccall,
  61                            payload_request, payload_answer);
  62
  63 }
  64
  65 static unsigned long int last_msg_ID = 0;
  66
  67 static int msgfilter_rpcID(gras_msg_t msg, void *ctx)
  68 {
  69   unsigned long int ID = *(unsigned long int *) ctx;
  70   int res = msg->ID == ID &&
  71     (msg->kind == e_gras_msg_kind_rpcanswer
  72      || msg->kind == e_gras_msg_kind_rpcerror);
  73   unsigned int cursor;
  74   gras_msg_cb_ctx_t rpc_ctx;
  75
  76
  77   DEBUG5
  78     ("Filter a message of ID %lu, type '%s' and kind '%s'. Waiting for ID=%lu. %s",
  79      msg->ID, msg->type->name, e_gras_msg_kind_names[msg->kind], ID,
  80      res ? "take it" : "reject");
  81
  82   if (res && !_gras_rpc_cancelled)
  83     return res;
  84
  85   /* Check whether it is an old answer to a message we already canceled */
  86   xbt_dynar_foreach(_gras_rpc_cancelled, cursor, rpc_ctx) {
  87     if (msg->ID == rpc_ctx->ID && msg->kind == e_gras_msg_kind_rpcanswer) {
  88       VERB1
  89         ("Got an answer to the already canceled (timeouted?) RPC %ld. Ignore it (leaking the payload!).",
  90          msg->ID);
  91       xbt_dynar_cursor_rm(_gras_rpc_cancelled, &cursor);
  92       return 1;
  93     }
  94   }
  95
  96   return res;
  97 }
  98
  99 /* Mallocator cruft */
 100 xbt_mallocator_t gras_msg_ctx_mallocator = NULL;
 101 void *gras_msg_ctx_mallocator_new_f(void)
 102 {
 103   return xbt_new0(s_gras_msg_cb_ctx_t, 1);
 104 }
 105
 106 void gras_msg_ctx_mallocator_free_f(void *ctx)
 107 {
 108   xbt_free(ctx);
 109 }
 110
 111 void gras_msg_ctx_mallocator_reset_f(void *ctx)
 112 {
 113   memset(ctx, 0, sizeof(s_gras_msg_cb_ctx_t));
 114 }
 115
 116 /** @brief Launch a RPC call, but do not block for the answer */
 117 gras_msg_cb_ctx_t
 118 gras_msg_rpc_async_call_(gras_socket_t server,
 119                          double timeOut,
 120                          gras_msgtype_t msgtype, void *request)
 121 {
 122   gras_msg_cb_ctx_t ctx = xbt_mallocator_get(gras_msg_ctx_mallocator);
 123
 124   if (msgtype->ctn_type) {
 125     xbt_assert1(request,
 126                 "RPC type '%s' convey a payload you must provide",
 127                 msgtype->name);
 128   } else {
 129     xbt_assert1(!request,
 130                 "No payload was declared for RPC type '%s'", msgtype->name);
 131   }
 132
 133   ctx->ID = last_msg_ID++;
 134   ctx->expeditor = server;
 135   ctx->msgtype = msgtype;
 136   ctx->timeout = timeOut;
 137
 138   VERB4("Send to %s:%d a RPC of type '%s' (ID=%lu)",
 139         gras_socket_peer_name(server),
 140         gras_socket_peer_port(server), msgtype->name, ctx->ID);
 141
 142   gras_msg_send_ext(server, e_gras_msg_kind_rpccall, ctx->ID, msgtype,
 143                     request);
 144
 145   return ctx;
 146 }
 147
 148 /** @brief Wait the answer of a RPC call previously launched asynchronously */
 149 void gras_msg_rpc_async_wait(gras_msg_cb_ctx_t ctx, void *answer)
 150 {
 151   xbt_ex_t e;
 152   s_gras_msg_t received;
 153
 154   if (ctx->msgtype->answer_type) {
 155     xbt_assert1(answer,
 156                 "Answers to RPC '%s' convey a payload you must accept",
 157                 ctx->msgtype->name);
 158   } else {
 159     xbt_assert1(!answer,
 160                 "No payload was declared for answers to RPC '%s'",
 161                 ctx->msgtype->name);
 162   }
 163
 164   TRY {
 165     /* The filter returns 1 when we eat an old RPC answer to something canceled */
 166     do {
 167       gras_msg_wait_ext_(ctx->timeout,
 168                          ctx->msgtype, NULL, msgfilter_rpcID, &ctx->ID,
 169                          &received);
 170     } while (received.ID != ctx->ID);
 171
 172   }
 173   CATCH(e) {
 174     if (!_gras_rpc_cancelled)
 175       _gras_rpc_cancelled = xbt_dynar_new(sizeof(ctx), NULL);
 176     xbt_dynar_push(_gras_rpc_cancelled, &ctx);
 177     INFO5("canceled RPC %ld pushed onto the stack (%s from %s:%d) Reason: %s",
 178           ctx->ID, ctx->msgtype->name,
 179           gras_socket_peer_name(ctx->expeditor),
 180           gras_socket_peer_port(ctx->expeditor), e.msg);
 181     RETHROW;
 182   }
 183
 184   xbt_mallocator_release(gras_msg_ctx_mallocator, ctx);
 185   if (received.kind == e_gras_msg_kind_rpcerror) {
 186     xbt_ex_t e;
 187     memcpy(&e, received.payl, received.payl_size);
 188     free(received.payl);
 189     VERB3("Raise a remote exception cat:%d comming from %s (%s)",
 190           e.category, e.host, e.msg);
 191     __xbt_ex_ctx()->ctx_ex.msg = e.msg;
 192     __xbt_ex_ctx()->ctx_ex.category = e.category;
 193     __xbt_ex_ctx()->ctx_ex.value = e.value;
 194     __xbt_ex_ctx()->ctx_ex.remote = 1;
 195     __xbt_ex_ctx()->ctx_ex.host = e.host;
 196     __xbt_ex_ctx()->ctx_ex.procname = e.procname;
 197     __xbt_ex_ctx()->ctx_ex.pid = e.pid;
 198     __xbt_ex_ctx()->ctx_ex.file = e.file;
 199     __xbt_ex_ctx()->ctx_ex.line = e.line;
 200     __xbt_ex_ctx()->ctx_ex.func = e.func;
 201     __xbt_ex_ctx()->ctx_ex.used = e.used;
 202     __xbt_ex_ctx()->ctx_ex.bt_strings = e.bt_strings;
 203     memset(&__xbt_ex_ctx()->ctx_ex.bt, 0, sizeof(__xbt_ex_ctx()->ctx_ex.bt));
 204     DO_THROW(__xbt_ex_ctx()->ctx_ex);
 205   }
 206   memcpy(answer, received.payl, received.payl_size);
 207   free(received.payl);
 208 }
 209
 210 /** @brief Conduct a RPC call */
 211 void gras_msg_rpccall_(gras_socket_t server,
 212                        double timeout,
 213                        gras_msgtype_t msgtype, void *request, void *answer)
 214 {
 215
 216   gras_msg_cb_ctx_t ctx;
 217
 218   ctx = gras_msg_rpc_async_call_(server, timeout, msgtype, request);
 219   gras_msg_rpc_async_wait(ctx, answer);
 220 }
 221
 222
 223 /** @brief Return the result of a RPC call
 224  *
 225  * It done before the actual return of the callback so that the callback can do
 226  * some cleanups before leaving.
 227  */
 228
 229 void gras_msg_rpcreturn(double timeOut, gras_msg_cb_ctx_t ctx, void *answer)
 230 {
 231   xbt_assert0(ctx->answer_due,
 232               "RPC return not allowed here. Either not a RPC message or already returned a result");
 233   ctx->answer_due = 0;
 234   DEBUG5("Return to RPC '%s' from %s:%d (tOut=%f, payl=%p)",
 235          ctx->msgtype->name,
 236          gras_socket_peer_name(ctx->expeditor),
 237          gras_socket_peer_port(ctx->expeditor), timeOut, answer);
 238   gras_msg_send_ext(ctx->expeditor, e_gras_msg_kind_rpcanswer, ctx->ID,
 239                     ctx->msgtype, answer);
 240 }