src/gras/Msg/msg.c

   1 /* $Id$ */
   2
   3 /* messaging - Function related to messaging (code shared between RL and SG)*/
   4
   5 /* Copyright (c) 2003, 2004 Martin Quinson. All rights reserved.            */
   6
   7 /* This program is free software; you can redistribute it and/or modify it
   8  * under the terms of the license (GNU LGPL) which comes with this package. */
   9
  10 #include "xbt/ex.h"
  11 #include "gras/Msg/msg_private.h"
  12 #include "gras/Virtu/virtu_interface.h"
  13 #include "gras/DataDesc/datadesc_interface.h"
  14 #include "gras/Transport/transport_interface.h" /* gras_select */
  15
  16 #define MIN(a,b) ((a) < (b) ? (a) : (b))
  17
  18 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gras_msg,gras,"High level messaging");
  19
  20 xbt_set_t _gras_msgtype_set = NULL;
  21 static char *make_namev(const char *name, short int ver);
  22 char _GRAS_header[6];
  23
  24 /*
  25  * Creating procdata for this module
  26  */
  27 static void *gras_msg_procdata_new() {
  28    gras_msg_procdata_t res = xbt_new(s_gras_msg_procdata_t,1);
  29
  30    res->msg_queue = xbt_dynar_new(sizeof(s_gras_msg_t),   NULL);
  31    res->cbl_list  = xbt_dynar_new(sizeof(gras_cblist_t *),gras_cbl_free);
  32    res->timers    = xbt_dynar_new(sizeof(s_gras_timer_t), NULL);
  33
  34    return (void*)res;
  35 }
  36
  37 /*
  38  * Freeing procdata for this module
  39  */
  40 static void gras_msg_procdata_free(void *data) {
  41    gras_msg_procdata_t res = (gras_msg_procdata_t)data;
  42
  43    xbt_dynar_free(&( res->msg_queue ));
  44    xbt_dynar_free(&( res->cbl_list ));
  45    xbt_dynar_free(&( res->timers ));
  46
  47    free(res);
  48 }
  49
  50 /*
  51  * Module registration
  52  */
  53 void gras_msg_register() {
  54    gras_procdata_add("gras_msg",gras_msg_procdata_new, gras_msg_procdata_free);
  55 }
  56
  57 /*
  58  * Initialize this submodule.
  59  */
  60 void gras_msg_init(void) {
  61   /* only initialize once */
  62   if (_gras_msgtype_set != NULL)
  63     return;
  64
  65   VERB0("Initializing Msg");
  66
  67   _gras_msgtype_set = xbt_set_new();
  68
  69   memcpy(_GRAS_header,"GRAS", 4);
  70   _GRAS_header[4]=GRAS_PROTOCOL_VERSION;
  71   _GRAS_header[5]=(char)GRAS_THISARCH;
  72 }
  73
  74 /*
  75  * Finalize the msg module
  76  */
  77 void
  78 gras_msg_exit(void) {
  79   VERB0("Exiting Msg");
  80   xbt_set_free(&_gras_msgtype_set);
  81 }
  82
  83 /*
  84  * Reclamed memory
  85  */
  86 void gras_msgtype_free(void *t) {
  87   gras_msgtype_t msgtype=(gras_msgtype_t)t;
  88   if (msgtype) {
  89     free(msgtype->name);
  90     free(msgtype);
  91   }
  92 }
  93
  94 /**
  95  * make_namev:
  96  *
  97  * Returns the versionned name of the message. If the version is 0, that's
  98  * the name unchanged. Pay attention to this before free'ing the result.
  99  */
 100 static char *make_namev(const char *name, short int ver) {
 101   char *namev;
 102
 103   if (!ver)
 104     return (char *)name;
 105
 106   namev = (char*)xbt_malloc(strlen(name)+2+3+1);
 107
 108   if (namev)
 109       sprintf(namev,"%s_v%d",name,ver);
 110
 111   return namev;
 112 }
 113
 114 /** @brief declare a new message type of the given name. It only accepts the given datadesc as payload
 115  *
 116  * @param name: name as it should be used for logging messages (must be uniq)
 117  * @param payload: datadescription of the payload
 118  */
 119 void gras_msgtype_declare(const char           *name,
 120                           gras_datadesc_type_t  payload) {
 121    gras_msgtype_declare_v(name, 0, payload);
 122 }
 123
 124 /** @brief declare a new versionned message type of the given name and payload
 125  *
 126  * @param name: name as it should be used for logging messages (must be uniq)
 127  * @param version: something like versionning symbol
 128  * @param payload: datadescription of the payload
 129  *
 130  * Registers a message to the GRAS mechanism. Use this version instead of
 131  * gras_msgtype_declare when you change the semantic or syntax of a message and
 132  * want your programs to be able to deal with both versions. Internally, each
 133  * will be handled as an independent message type, so you can register
 134  * differents for each of them.
 135  */
 136 void
 137 gras_msgtype_declare_v(const char           *name,
 138                        short int             version,
 139                        gras_datadesc_type_t  payload) {
 140
 141   gras_msgtype_t msgtype=NULL;
 142   char *namev=make_namev(name,version);
 143   volatile int found = 0;
 144   xbt_ex_t e;
 145
 146   TRY {
 147     msgtype = (gras_msgtype_t)xbt_set_get_by_name(_gras_msgtype_set,namev);
 148     found = 1;
 149   } CATCH(e) {
 150     if (e.category != not_found_error)
 151       RETHROW;
 152     xbt_ex_free(e);
 153   }
 154
 155   if (found) {
 156     VERB2("Re-register version %d of message '%s' (same payload, ignored).",
 157           version, name);
 158     xbt_assert3(!gras_datadesc_type_cmp(msgtype->ctn_type, payload),
 159                  "Message %s re-registred with another payload (%s was %s)",
 160                  namev,gras_datadesc_get_name(payload),
 161                  gras_datadesc_get_name(msgtype->ctn_type));
 162
 163     return ; /* do really ignore it */
 164
 165   }
 166
 167   VERB3("Register version %d of message '%s' (payload: %s).",
 168         version, name, gras_datadesc_get_name(payload));
 169
 170   msgtype = xbt_new(s_gras_msgtype_t,1);
 171   msgtype->name = (namev == name ? strdup(name) : namev);
 172   msgtype->name_len = strlen(namev);
 173   msgtype->version = version;
 174   msgtype->ctn_type = payload;
 175
 176   xbt_set_add(_gras_msgtype_set, (xbt_set_elm_t)msgtype,
 177                &gras_msgtype_free);
 178 }
 179
 180 /** @brief retrive an existing message type from its name. */
 181 gras_msgtype_t gras_msgtype_by_name (const char *name) {
 182   return gras_msgtype_by_namev(name,0);
 183 }
 184
 185 /** @brief retrive an existing message type from its name and version. */
 186 gras_msgtype_t gras_msgtype_by_namev(const char      *name,
 187                                      short int        version) {
 188   gras_msgtype_t res;
 189   char *namev = make_namev(name,version);
 190
 191   res = (gras_msgtype_t)xbt_set_get_by_name(_gras_msgtype_set, namev);
 192   if (name != namev)
 193     free(namev);
 194
 195   return res;
 196 }
 197 /** @brief retrive an existing message type from its name and version. */
 198 gras_msgtype_t gras_msgtype_by_id(int id) {
 199   return (gras_msgtype_t)xbt_set_get_by_id(_gras_msgtype_set, id);
 200 }
 201
 202 /** \brief Waits for a message to come in over a given socket.
 203  *
 204  * @param timeout: How long should we wait for this message.
 205  * @param msgt_want: type of awaited msg
 206  * @param[out] expeditor: where to create a socket to answer the incomming message
 207  * @param[out] payload: where to write the payload of the incomming message
 208  * @return the error code (or no_error).
 209  *
 210  * Every message of another type received before the one waited will be queued
 211  * and used by subsequent call to this function or gras_msg_handle().
 212  */
 213 void
 214 gras_msg_wait(double           timeout,
 215               gras_msgtype_t   msgt_want,
 216               gras_socket_t   *expeditor,
 217               void            *payload) {
 218
 219   gras_msgtype_t msgt_got;
 220   void *payload_got;
 221   int payload_size_got;
 222   double start, now;
 223   gras_msg_procdata_t pd=(gras_msg_procdata_t)gras_libdata_get("gras_msg");
 224   int cpt;
 225   s_gras_msg_t msg;
 226   gras_socket_t expeditor_res = NULL;
 227
 228   payload_got = NULL;
 229
 230   xbt_assert0(msgt_want,"Cannot wait for the NULL message");
 231
 232   VERB1("Waiting for message '%s'",msgt_want->name);
 233
 234   start = now = gras_os_time();
 235
 236   xbt_dynar_foreach(pd->msg_queue,cpt,msg){
 237     if (msg.type->code == msgt_want->code) {
 238       if (expeditor)
 239         *expeditor = msg.expeditor;
 240       memcpy(payload, msg.payload, msg.payload_size);
 241       free(msg.payload);
 242       xbt_dynar_cursor_rm(pd->msg_queue, &cpt);
 243       VERB0("The waited message was queued");
 244       return;
 245     }
 246   }
 247
 248   while (1) {
 249     expeditor_res = gras_trp_select(timeout - now + start);
 250     gras_msg_recv(expeditor_res, &msgt_got, &payload_got, &payload_size_got);
 251     if (msgt_got->code == msgt_want->code) {
 252       if (expeditor)
 253         *expeditor=expeditor_res;
 254       memcpy(payload, payload_got, payload_size_got);
 255       free(payload_got);
 256       VERB0("Got waited message");
 257       return;
 258     }
 259
 260     /* not expected msg type. Queue it for later */
 261     msg.expeditor = expeditor_res;
 262     msg.type      = msgt_got;
 263     msg.payload   = payload;
 264     msg.payload_size = payload_size_got;
 265     xbt_dynar_push(pd->msg_queue,&msg);
 266
 267     now=gras_os_time();
 268     if (now - start + 0.001 < timeout) {
 269       THROW1(timeout_error,  now-start+0.001-timeout,
 270              "Timeout while waiting for msg %s",msgt_want->name);
 271     }
 272   }
 273
 274   THROW_IMPOSSIBLE;
 275 }
 276
 277 /** @brief Handle an incomming message or timer (or wait up to \a timeOut seconds)
 278  *
 279  * @param timeOut: How long to wait for incoming messages (in seconds)
 280  * @return the error code (or no_error).
 281  *
 282  * Messages are passed to the callbacks.
 283  */
 284 void
 285 gras_msg_handle(double timeOut) {
 286
 287   double          untiltimer;
 288
 289   int             cpt;
 290
 291   s_gras_msg_t    msg;
 292   gras_socket_t   expeditor=NULL;
 293   void           *payload=NULL;
 294   int             payload_size;
 295   gras_msgtype_t  msgtype;
 296
 297   gras_msg_procdata_t pd=(gras_msg_procdata_t)gras_libdata_get("gras_msg");
 298   gras_cblist_t  *list=NULL;
 299   gras_msg_cb_t       cb;
 300
 301   int timerexpected, timeouted;
 302   xbt_ex_t e;
 303
 304   VERB1("Handling message within the next %.2fs",timeOut);
 305
 306   untiltimer = gras_msg_timer_handle();
 307   DEBUG2("[%.0f] Next timer in %f sec", gras_os_time(), untiltimer);
 308   if (untiltimer == 0.0) {
 309      /* A timer was already elapsed and handled */
 310      return;
 311   }
 312   if (untiltimer != -1.0) {
 313      timerexpected = 1;
 314      timeOut = MIN(timeOut, untiltimer);
 315   } else {
 316      timerexpected = 0;
 317   }
 318
 319   /* get a message (from the queue or from the net) */
 320   timeouted = 0;
 321   if (xbt_dynar_length(pd->msg_queue)) {
 322     DEBUG0("Get a message from the queue");
 323     xbt_dynar_shift(pd->msg_queue,&msg);
 324     expeditor = msg.expeditor;
 325     msgtype   = msg.type;
 326     payload   = msg.payload;
 327   } else {
 328     TRY {
 329       expeditor = gras_trp_select(timeOut);
 330     } CATCH(e) {
 331       if (e.category != timeout_error)
 332         RETHROW;
 333       xbt_ex_free(e);
 334       timeouted = 1;
 335     }
 336
 337     if (!timeouted) {
 338       TRY {
 339         gras_msg_recv(expeditor, &msgtype, &payload, &payload_size);
 340       } CATCH(e) {
 341         RETHROW1("Error caught  while receiving a message on select()ed socket %p: %s",
 342                  expeditor);
 343       }
 344     }
 345   }
 346
 347   if (timeouted) {
 348      if (timerexpected) {
 349
 350         /* A timer elapsed before the arrival of any message even if we select()ed a bit */
 351         untiltimer = gras_msg_timer_handle();
 352         if (untiltimer == 0.0) {
 353           /* we served a timer, we're done */
 354           return;
 355         } else {
 356            xbt_assert1(untiltimer>0, "Negative timer (%f). I'm 'puzzeled'", untiltimer);
 357            WARN1("No timer elapsed, in contrary to expectations (next in %f sec)",
 358                   untiltimer);
 359            THROW1(timeout_error,0,
 360                   "No timer elapsed, in contrary to expectations (next in %f sec)",
 361                   untiltimer);
 362         }
 363
 364      } else {
 365         /* select timeouted, and no timer elapsed. Nothing to do */
 366        THROW0(timeout_error, 0, "No new message or timer");
 367      }
 368
 369   }
 370
 371   /* A message was already there or arrived in the meanwhile. handle it */
 372   xbt_dynar_foreach(pd->cbl_list,cpt,list) {
 373     if (list->id == msgtype->code) {
 374       break;
 375     } else {
 376       list=NULL;
 377     }
 378   }
 379   if (!list) {
 380     INFO1("No callback for the incomming '%s' message. Discarded.",
 381           msgtype->name);
 382     WARN0("FIXME: gras_datadesc_free not implemented => leaking the payload");
 383     return;
 384   }
 385
 386   xbt_dynar_foreach(list->cbs,cpt,cb) {
 387     VERB3("Use the callback #%d (@%p) for incomming msg %s",
 388           cpt+1,cb,msgtype->name);
 389     if ((*cb)(expeditor,payload)) {
 390       /* cb handled the message */
 391       free(payload);
 392       return;
 393     }
 394   }
 395
 396   /* FIXME: gras_datadesc_free not implemented => leaking the payload */
 397   THROW1(mismatch_error,0,
 398          "Message '%s' refused by all registered callbacks", msgtype->name);
 399 }
 400
 401 void
 402 gras_cbl_free(void *data){
 403   gras_cblist_t *list=*(void**)data;
 404   if (list) {
 405     xbt_dynar_free(&( list->cbs ));
 406     free(list);
 407   }
 408 }
 409
 410 /** \brief Bind the given callback to the given message type
 411  *
 412  * Several callbacks can be attached to a given message type. The lastly added one will get the message first, and
 413  * if it returns false, the message will be passed to the second one.
 414  * And so on until one of the callbacks accepts the message.
 415  */
 416 void
 417 gras_cb_register(gras_msgtype_t msgtype,
 418                  gras_msg_cb_t cb) {
 419   gras_msg_procdata_t pd=(gras_msg_procdata_t)gras_libdata_get("gras_msg");
 420   gras_cblist_t *list=NULL;
 421   int cpt;
 422
 423   DEBUG2("Register %p as callback to '%s'",cb,msgtype->name);
 424
 425   /* search the list of cb for this message on this host (creating if NULL) */
 426   xbt_dynar_foreach(pd->cbl_list,cpt,list) {
 427     if (list->id == msgtype->code) {
 428       break;
 429     } else {
 430       list=NULL;
 431     }
 432   }
 433   if (!list) {
 434     /* First cb? Create room */
 435     list = xbt_new(gras_cblist_t,1);
 436     list->id = msgtype->code;
 437     list->cbs = xbt_dynar_new(sizeof(gras_msg_cb_t), NULL);
 438     xbt_dynar_push(pd->cbl_list,&list);
 439   }
 440
 441   /* Insert the new one into the set */
 442   xbt_dynar_insert_at(list->cbs,0,&cb);
 443 }
 444
 445 /** \brief Unbind the given callback from the given message type */
 446 void
 447 gras_cb_unregister(gras_msgtype_t msgtype,
 448                    gras_msg_cb_t cb) {
 449
 450   gras_msg_procdata_t pd=(gras_msg_procdata_t)gras_libdata_get("gras_msg");
 451   gras_cblist_t *list;
 452   gras_msg_cb_t cb_cpt;
 453   int cpt;
 454   int found = 0;
 455
 456   /* search the list of cb for this message on this host */
 457   xbt_dynar_foreach(pd->cbl_list,cpt,list) {
 458     if (list->id == msgtype->code) {
 459       break;
 460     } else {
 461       list=NULL;
 462     }
 463   }
 464
 465   /* Remove it from the set */
 466   if (list) {
 467     xbt_dynar_foreach(list->cbs,cpt,cb_cpt) {
 468       if (cb == cb_cpt) {
 469         xbt_dynar_cursor_rm(list->cbs, &cpt);
 470         found = 1;
 471       }
 472     }
 473   }
 474   if (!found)
 475     VERB1("Ignoring removal of unexisting callback to msg id %d",
 476           msgtype->code);
 477 }