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 #include "portable.h" /* execinfo when available to propagate exceptions */
  16
  17 #ifndef MIN
  18 #define MIN(a,b) ((a) < (b) ? (a) : (b))
  19 #endif
  20
  21 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gras_msg,gras,"High level messaging");
  22
  23 xbt_set_t _gras_msgtype_set = NULL;
  24 static char *make_namev(const char *name, short int ver);
  25 char _GRAS_header[6];
  26 const char *e_gras_msg_kind_names[e_gras_msg_kind_count]=
  27   {"UNKNOWN","ONEWAY","RPC call","RPC answer","RPC error"};
  28
  29 /*
  30  * Creating procdata for this module
  31  */
  32 static void *gras_msg_procdata_new() {
  33    gras_msg_procdata_t res = xbt_new(s_gras_msg_procdata_t,1);
  34
  35    res->name = xbt_strdup("gras_msg");
  36    res->name_len = 0;
  37    res->msg_queue = xbt_dynar_new(sizeof(s_gras_msg_t),   NULL);
  38    res->cbl_list  = xbt_dynar_new(sizeof(gras_cblist_t *),gras_cbl_free);
  39    res->timers    = xbt_dynar_new(sizeof(s_gras_timer_t), NULL);
  40
  41    return (void*)res;
  42 }
  43
  44 /*
  45  * Freeing procdata for this module
  46  */
  47 static void gras_msg_procdata_free(void *data) {
  48    gras_msg_procdata_t res = (gras_msg_procdata_t)data;
  49
  50    xbt_dynar_free(&( res->msg_queue ));
  51    xbt_dynar_free(&( res->cbl_list ));
  52    xbt_dynar_free(&( res->timers ));
  53
  54    free(res->name);
  55    free(res);
  56 }
  57
  58 /*
  59  * Module registration
  60  */
  61 int gras_msg_libdata_id;
  62 void gras_msg_register() {
  63    gras_msg_libdata_id = gras_procdata_add("gras_msg",gras_msg_procdata_new, gras_msg_procdata_free);
  64 }
  65
  66 /*
  67  * Initialize this submodule.
  68  */
  69 void gras_msg_init(void) {
  70   /* only initialize once */
  71   if (_gras_msgtype_set != NULL)
  72     return;
  73
  74   VERB0("Initializing Msg");
  75
  76   _gras_msgtype_set = xbt_set_new();
  77
  78   memcpy(_GRAS_header,"GRAS", 4);
  79   _GRAS_header[4]=GRAS_PROTOCOL_VERSION;
  80   _GRAS_header[5]=(char)GRAS_THISARCH;
  81 }
  82
  83 /*
  84  * Finalize the msg module
  85  */
  86 void
  87 gras_msg_exit(void) {
  88   VERB0("Exiting Msg");
  89   xbt_set_free(&_gras_msgtype_set);
  90 }
  91
  92 /*
  93  * Reclamed memory
  94  */
  95 void gras_msgtype_free(void *t) {
  96   gras_msgtype_t msgtype=(gras_msgtype_t)t;
  97   if (msgtype) {
  98     free(msgtype->name);
  99     free(msgtype);
 100   }
 101 }
 102
 103 /**
 104  * make_namev:
 105  *
 106  * Returns the versionned name of the message. If the version is 0, that's
 107  * the name unchanged. Pay attention to this before free'ing the result.
 108  */
 109 static char *make_namev(const char *name, short int ver) {
 110   char *namev;
 111
 112   if (!ver)
 113     return (char *)name;
 114
 115   namev = (char*)xbt_malloc(strlen(name)+2+3+1);
 116
 117   if (namev)
 118       sprintf(namev,"%s_v%d",name,ver);
 119
 120   return namev;
 121 }
 122
 123 /* Internal function doing the crude work of registering messages */
 124 void
 125 gras_msgtype_declare_ext(const char           *name,
 126                          short int             version,
 127                          e_gras_msg_kind_t     kind,
 128                          gras_datadesc_type_t  payload_request,
 129                          gras_datadesc_type_t  payload_answer) {
 130
 131   gras_msgtype_t msgtype=NULL;
 132   char *namev=make_namev(name,version);
 133   volatile int found = 0;
 134   xbt_ex_t e;
 135
 136   TRY {
 137     msgtype = (gras_msgtype_t)xbt_set_get_by_name(_gras_msgtype_set,namev);
 138     found = 1;
 139   } CATCH(e) {
 140     if (e.category != not_found_error)
 141       RETHROW;
 142     xbt_ex_free(e);
 143   }
 144
 145   if (found) {
 146     VERB2("Re-register version %d of message '%s' (same kind & payload, ignored).",
 147           version, name);
 148     xbt_assert3(msgtype->kind == kind,
 149                 "Message %s re-registered as a %s (it was known as a %s)",
 150                 namev,e_gras_msg_kind_names[kind],e_gras_msg_kind_names[msgtype->kind]);
 151     xbt_assert3(!gras_datadesc_type_cmp(msgtype->ctn_type, payload_request),
 152                  "Message %s re-registred with another payload (%s was %s)",
 153                  namev,gras_datadesc_get_name(payload_request),
 154                  gras_datadesc_get_name(msgtype->ctn_type));
 155
 156     xbt_assert3(!gras_datadesc_type_cmp(msgtype->answer_type, payload_answer),
 157              "Message %s re-registred with another answer payload (%s was %s)",
 158                  namev,gras_datadesc_get_name(payload_answer),
 159                  gras_datadesc_get_name(msgtype->answer_type));
 160
 161     return ; /* do really ignore it */
 162
 163   }
 164
 165   VERB4("Register version %d of message '%s' "
 166         "(payload: %s; answer payload: %s).",
 167         version, name, gras_datadesc_get_name(payload_request),
 168         gras_datadesc_get_name(payload_answer));
 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->kind = kind;
 175   msgtype->ctn_type = payload_request;
 176   msgtype->answer_type = payload_answer;
 177
 178   xbt_set_add(_gras_msgtype_set, (xbt_set_elm_t)msgtype,
 179                &gras_msgtype_free);
 180 }
 181
 182
 183 /** @brief declare a new message type of the given name. It only accepts the given datadesc as payload
 184  *
 185  * @param name: name as it should be used for logging messages (must be uniq)
 186  * @param payload: datadescription of the payload
 187  */
 188 void gras_msgtype_declare(const char           *name,
 189                           gras_datadesc_type_t  payload) {
 190    gras_msgtype_declare_ext(name, 0, e_gras_msg_kind_oneway, payload, NULL);
 191 }
 192
 193
 194
 195 /** @brief declare a new versionned message type of the given name and payload
 196  *
 197  * @param name: name as it should be used for logging messages (must be uniq)
 198  * @param version: something like versionning symbol
 199  * @param payload: datadescription of the payload
 200  *
 201  * Registers a message to the GRAS mechanism. Use this version instead of
 202  * gras_msgtype_declare when you change the semantic or syntax of a message and
 203  * want your programs to be able to deal with both versions. Internally, each
 204  * will be handled as an independent message type, so you can register
 205  * differents for each of them.
 206  */
 207 void
 208 gras_msgtype_declare_v(const char           *name,
 209                        short int             version,
 210                        gras_datadesc_type_t  payload) {
 211
 212    gras_msgtype_declare_ext(name, version,
 213                             e_gras_msg_kind_oneway, payload, NULL);
 214 }
 215
 216 /** @brief retrive an existing message type from its name. */
 217 gras_msgtype_t gras_msgtype_by_name (const char *name) {
 218   return gras_msgtype_by_namev(name,0);
 219 }
 220
 221 /** @brief retrive an existing message type from its name and version. */
 222 gras_msgtype_t gras_msgtype_by_namev(const char      *name,
 223                                      short int        version) {
 224   gras_msgtype_t res = NULL;
 225   char *namev = make_namev(name,version);
 226   xbt_ex_t e;
 227
 228   TRY {
 229     res = (gras_msgtype_t)xbt_set_get_by_name(_gras_msgtype_set, namev);
 230   } CATCH(e) {
 231     xbt_ex_free(e);
 232     THROW1(not_found_error,0,"No registred message of that name: %s",name);
 233   }
 234   if (name != namev)
 235     free(namev);
 236
 237   return res;
 238 }
 239 /** @brief retrive an existing message type from its name and version. */
 240 gras_msgtype_t gras_msgtype_by_id(int id) {
 241   return (gras_msgtype_t)xbt_set_get_by_id(_gras_msgtype_set, id);
 242 }
 243
 244 /** \brief Waits for a message to come in over a given socket.
 245  *
 246  * @param timeout: How long should we wait for this message.
 247  * @param msgt_want: type of awaited msg (or NULL if I'm enclined to accept any message)
 248  * @param expe_want: awaited expeditot (match on hostname, not port; NULL if not relevant)
 249  * @param filter: function returning true or false when passed a payload. Messages for which it returns false are not selected. (NULL if not relevant)
 250  * @param filter_ctx: context passed as second argument of the filter (a pattern to match?)
 251  * @param[out] msg_got: where to write the message we got
 252  *
 253  * Every message of another type received before the one waited will be queued
 254  * and used by subsequent call to this function or gras_msg_handle().
 255  */
 256
 257 void
 258 gras_msg_wait_ext(double           timeout,
 259
 260                   gras_msgtype_t   msgt_want,
 261                   gras_socket_t    expe_want,
 262                   gras_msg_filter_t filter,
 263                   void             *filter_ctx,
 264
 265                   gras_msg_t       msg_got) {
 266
 267   s_gras_msg_t msg;
 268   double start, now;
 269   gras_msg_procdata_t pd=(gras_msg_procdata_t)gras_libdata_by_id(gras_msg_libdata_id);
 270   int cpt;
 271
 272   xbt_assert0(msgt_want,"Cannot wait for the NULL message");
 273   xbt_assert0(msg_got,"msg_got is an output parameter");
 274
 275   VERB1("Waiting for message '%s'",msgt_want->name);
 276
 277   start = now = gras_os_time();
 278
 279   xbt_dynar_foreach(pd->msg_queue,cpt,msg){
 280     if ( (   !msgt_want || (msg.type->code == msgt_want->code))
 281          && (!expe_want || (!strcmp( gras_socket_peer_name(msg.expe),
 282                                      gras_socket_peer_name(expe_want))))
 283          && (!filter || filter(&msg,filter_ctx))) {
 284
 285       memcpy(msg_got,&msg,sizeof(s_gras_msg_t));
 286       xbt_dynar_cursor_rm(pd->msg_queue, &cpt);
 287       VERB0("The waited message was queued");
 288       return;
 289     }
 290   }
 291
 292   while (1) {
 293     memset(&msg,sizeof(msg),0);
 294
 295     msg.expe = gras_trp_select(timeout ? timeout - now + start : 0);
 296     gras_msg_recv(msg.expe, &msg);
 297     DEBUG0("Got a message from the socket");
 298
 299     if ( (   !msgt_want || (msg.type->code == msgt_want->code))
 300          && (!expe_want || (!strcmp( gras_socket_peer_name(msg.expe),
 301                                      gras_socket_peer_name(expe_want))))
 302          && (!filter || filter(&msg,filter_ctx))) {
 303
 304       memcpy(msg_got,&msg,sizeof(s_gras_msg_t));
 305       DEBUG0("Message matches expectations. Use it.");
 306       return;
 307     }
 308     DEBUG0("Message does not match expectations. Queue it.");
 309
 310     /* not expected msg type. Queue it for later */
 311     xbt_dynar_push(pd->msg_queue,&msg);
 312
 313     now=gras_os_time();
 314     if (now - start + 0.001 < timeout) {
 315       THROW1(timeout_error,  now-start+0.001-timeout,
 316              "Timeout while waiting for msg %s",msgt_want->name);
 317     }
 318   }
 319
 320   THROW_IMPOSSIBLE;
 321 }
 322 /** \brief Waits for a message to come in over a given socket.
 323  *
 324  * @param timeout: How long should we wait for this message.
 325  * @param msgt_want: type of awaited msg
 326  * @param[out] expeditor: where to create a socket to answer the incomming message
 327  * @param[out] payload: where to write the payload of the incomming message
 328  * @return the error code (or no_error).
 329  *
 330  * Every message of another type received before the one waited will be queued
 331  * and used by subsequent call to this function or gras_msg_handle().
 332  */
 333 void
 334 gras_msg_wait(double           timeout,
 335               gras_msgtype_t   msgt_want,
 336               gras_socket_t   *expeditor,
 337               void            *payload) {
 338   s_gras_msg_t msg;
 339
 340   gras_msg_wait_ext(timeout,
 341                     msgt_want, NULL,      NULL, NULL,
 342                     &msg);
 343
 344   if (payload) {
 345     memcpy(payload,msg.payl,msg.payl_size);
 346     free(msg.payl);
 347   }
 348
 349   if (expeditor)
 350     *expeditor = msg.expe;
 351 }
 352
 353
 354 /** \brief Send the data pointed by \a payload as a message of type
 355  * \a msgtype to the peer \a sock */
 356 void
 357 gras_msg_send(gras_socket_t   sock,
 358               gras_msgtype_t  msgtype,
 359               void           *payload) {
 360
 361   gras_msg_send_ext(sock, e_gras_msg_kind_oneway,0, msgtype, payload);
 362 }
 363
 364 /** @brief Handle an incomming message or timer (or wait up to \a timeOut seconds)
 365  *
 366  * @param timeOut: How long to wait for incoming messages (in seconds)
 367  * @return the error code (or no_error).
 368  *
 369  * Messages are passed to the callbacks.
 370  */
 371 void
 372 gras_msg_handle(double timeOut) {
 373
 374   double          untiltimer;
 375
 376   int             cpt;
 377
 378   s_gras_msg_t    msg;
 379
 380   gras_msg_procdata_t pd=(gras_msg_procdata_t)gras_libdata_by_id(gras_msg_libdata_id);
 381   gras_cblist_t  *list=NULL;
 382   gras_msg_cb_t       cb;
 383   s_gras_msg_cb_ctx_t ctx;
 384
 385   int timerexpected, timeouted;
 386   xbt_ex_t e;
 387
 388   VERB1("Handling message within the next %.2fs",timeOut);
 389
 390   untiltimer = gras_msg_timer_handle();
 391   DEBUG2("[%.0f] Next timer in %f sec", gras_os_time(), untiltimer);
 392   if (untiltimer == 0.0) {
 393      /* A timer was already elapsed and handled */
 394      return;
 395   }
 396   if (untiltimer != -1.0) {
 397      timerexpected = 1;
 398      timeOut = MIN(timeOut, untiltimer);
 399   } else {
 400      timerexpected = 0;
 401   }
 402
 403   /* get a message (from the queue or from the net) */
 404   timeouted = 0;
 405   if (xbt_dynar_length(pd->msg_queue)) {
 406     DEBUG0("Get a message from the queue");
 407     xbt_dynar_shift(pd->msg_queue,&msg);
 408   } else {
 409     TRY {
 410       msg.expe = gras_trp_select(timeOut);
 411     } CATCH(e) {
 412       if (e.category != timeout_error)
 413         RETHROW;
 414       xbt_ex_free(e);
 415       timeouted = 1;
 416     }
 417
 418     if (!timeouted) {
 419       TRY {
 420         /* FIXME: if not the right kind, queue it and recall ourself or goto >:-) */
 421         gras_msg_recv(msg.expe, &msg);
 422         DEBUG0("Received a msg from the socket");
 423
 424       } CATCH(e) {
 425         RETHROW1("Error caught while receiving a message on select()ed socket %p: %s",
 426                  msg.expe);
 427       }
 428     }
 429   }
 430
 431   if (timeouted) {
 432      if (timerexpected) {
 433
 434         /* A timer elapsed before the arrival of any message even if we select()ed a bit */
 435         untiltimer = gras_msg_timer_handle();
 436         if (untiltimer == 0.0) {
 437           /* we served a timer, we're done */
 438           return;
 439         } else {
 440            xbt_assert1(untiltimer>0, "Negative timer (%f). I'm 'puzzeled'", untiltimer);
 441            WARN1("No timer elapsed, in contrary to expectations (next in %f sec)",
 442                   untiltimer);
 443            THROW1(timeout_error,0,
 444                   "No timer elapsed, in contrary to expectations (next in %f sec)",
 445                   untiltimer);
 446         }
 447
 448      } else {
 449         /* select timeouted, and no timer elapsed. Nothing to do */
 450        THROW0(timeout_error, 0, "No new message or timer");
 451      }
 452
 453   }
 454
 455   /* A message was already there or arrived in the meanwhile. handle it */
 456   xbt_dynar_foreach(pd->cbl_list,cpt,list) {
 457     if (list->id == msg.type->code) {
 458       break;
 459     } else {
 460       list=NULL;
 461     }
 462   }
 463   if (!list) {
 464     INFO1("No callback for the incomming '%s' message. Discarded.",
 465           msg.type->name);
 466     WARN0("FIXME: gras_datadesc_free not implemented => leaking the payload");
 467     return;
 468   }
 469
 470   ctx.expeditor = msg.expe;
 471   ctx.ID = msg.ID;
 472   ctx.msgtype = msg.type;
 473
 474   switch (msg.type->kind) {
 475   case e_gras_msg_kind_oneway:
 476   case e_gras_msg_kind_rpccall:
 477     TRY {
 478       xbt_dynar_foreach(list->cbs,cpt,cb) {
 479         VERB3("Use the callback #%d (@%p) for incomming msg %s",
 480               cpt+1,cb,msg.type->name);
 481         if ((*cb)(&ctx,msg.payl)) {
 482           /* cb handled the message */
 483           free(msg.payl);
 484           return;
 485         }
 486       }
 487     } CATCH(e) {
 488       free(msg.payl);
 489       if (msg.type->kind == e_gras_msg_kind_rpccall) {
 490         /* The callback raised an exception, propagate it on the network */
 491         e.host = (char*)gras_os_myname();
 492 #ifdef HAVE_EXECINFO_H
 493         e.bt_strings = backtrace_symbols (e.bt, e.used);
 494 #endif
 495         gras_msg_send_ext(msg.expe, e_gras_msg_kind_rpcerror , msg.ID, msg.type, &e);
 496         e.host = NULL;
 497         INFO2("RPC callback raised an exception, which were propagated back to %s:%d",
 498               gras_socket_peer_name(msg.expe),  gras_socket_peer_port(msg.expe));
 499         xbt_ex_free(e);
 500         return;
 501       }
 502       RETHROW;
 503     }
 504     /* FIXME: gras_datadesc_free not implemented => leaking the payload */
 505     THROW1(mismatch_error,0,
 506            "Message '%s' refused by all registered callbacks", msg.type->name);
 507     break;
 508
 509
 510   case e_gras_msg_kind_rpcanswer:
 511     INFO1("Unexpected RPC answer discarded (type: %s)", msg.type->name);
 512     WARN0("FIXME: gras_datadesc_free not implemented => leaking the payload");
 513     return;
 514
 515   case e_gras_msg_kind_rpcerror:
 516     INFO1("Unexpected RPC error discarded (type: %s)", msg.type->name);
 517     WARN0("FIXME: gras_datadesc_free not implemented => leaking the payload");
 518     return;
 519
 520   default:
 521     THROW1(unknown_error,0,
 522            "Cannot handle messages of kind %d yet",msg.type->kind);
 523   }
 524
 525 }
 526
 527 void
 528 gras_cbl_free(void *data){
 529   gras_cblist_t *list=*(void**)data;
 530   if (list) {
 531     xbt_dynar_free(&( list->cbs ));
 532     free(list);
 533   }
 534 }
 535
 536 /** \brief Bind the given callback to the given message type
 537  *
 538  * Several callbacks can be attached to a given message type. The lastly added one will get the message first, and
 539  * if it returns false, the message will be passed to the second one.
 540  * And so on until one of the callbacks accepts the message.
 541  */
 542 void
 543 gras_cb_register(gras_msgtype_t msgtype,
 544                  gras_msg_cb_t cb) {
 545   gras_msg_procdata_t pd=(gras_msg_procdata_t)gras_libdata_by_id(gras_msg_libdata_id);
 546   gras_cblist_t *list=NULL;
 547   int cpt;
 548
 549   DEBUG2("Register %p as callback to '%s'",cb,msgtype->name);
 550
 551   /* search the list of cb for this message on this host (creating if NULL) */
 552   xbt_dynar_foreach(pd->cbl_list,cpt,list) {
 553     if (list->id == msgtype->code) {
 554       break;
 555     } else {
 556       list=NULL;
 557     }
 558   }
 559   if (!list) {
 560     /* First cb? Create room */
 561     list = xbt_new(gras_cblist_t,1);
 562     list->id = msgtype->code;
 563     list->cbs = xbt_dynar_new(sizeof(gras_msg_cb_t), NULL);
 564     xbt_dynar_push(pd->cbl_list,&list);
 565   }
 566
 567   /* Insert the new one into the set */
 568   xbt_dynar_insert_at(list->cbs,0,&cb);
 569 }
 570
 571 /** \brief Unbind the given callback from the given message type */
 572 void
 573 gras_cb_unregister(gras_msgtype_t msgtype,
 574                    gras_msg_cb_t cb) {
 575
 576   gras_msg_procdata_t pd=(gras_msg_procdata_t)gras_libdata_by_id(gras_msg_libdata_id);
 577   gras_cblist_t *list;
 578   gras_msg_cb_t cb_cpt;
 579   int cpt;
 580   int found = 0;
 581
 582   /* search the list of cb for this message on this host */
 583   xbt_dynar_foreach(pd->cbl_list,cpt,list) {
 584     if (list->id == msgtype->code) {
 585       break;
 586     } else {
 587       list=NULL;
 588     }
 589   }
 590
 591   /* Remove it from the set */
 592   if (list) {
 593     xbt_dynar_foreach(list->cbs,cpt,cb_cpt) {
 594       if (cb == cb_cpt) {
 595         xbt_dynar_cursor_rm(list->cbs, &cpt);
 596         found = 1;
 597       }
 598     }
 599   }
 600   if (!found)
 601     VERB1("Ignoring removal of unexisting callback to msg id %d",
 602           msgtype->code);
 603 }
 604
 605 /** \brief Retrieve the expeditor of the message */
 606 gras_socket_t gras_msg_cb_ctx_from(gras_msg_cb_ctx_t ctx) {
 607   return ctx->expeditor;
 608 }