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 payl_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] msgt_got: where to write the descriptor of the message we got
 252  * @param[out] expe_got: where to create a socket to answer the incomming message
 253  * @param[out] payl_got: where to write the payload of the incomming message
 254  *
 255  * Every message of another type received before the one waited will be queued
 256  * and used by subsequent call to this function or gras_msg_handle().
 257  */
 258
 259 void
 260 gras_msg_wait_ext(double           timeout,
 261
 262                   gras_msgtype_t   msgt_want,
 263                   gras_socket_t    expe_want,
 264                   gras_msg_filter_t filter,
 265                   void             *filter_ctx,
 266
 267                   gras_msg_t       msg_got) {
 268
 269   s_gras_msg_t msg;
 270   double start, now;
 271   gras_msg_procdata_t pd=(gras_msg_procdata_t)gras_libdata_by_id(gras_msg_libdata_id);
 272   int cpt;
 273
 274   xbt_assert0(msgt_want,"Cannot wait for the NULL message");
 275   xbt_assert0(msg_got,"msg_got is an output parameter");
 276
 277   VERB1("Waiting for message '%s'",msgt_want->name);
 278
 279   start = now = gras_os_time();
 280
 281   xbt_dynar_foreach(pd->msg_queue,cpt,msg){
 282     if ( (   !msgt_want || (msg.type->code == msgt_want->code))
 283          && (!expe_want || (!strcmp( gras_socket_peer_name(msg.expe),
 284                                      gras_socket_peer_name(expe_want))))
 285          && (!filter || filter(&msg,filter_ctx))) {
 286
 287       memcpy(msg_got,&msg,sizeof(s_gras_msg_t));
 288       xbt_dynar_cursor_rm(pd->msg_queue, &cpt);
 289       VERB0("The waited message was queued");
 290       return;
 291     }
 292   }
 293
 294   while (1) {
 295     memset(&msg,sizeof(msg),0);
 296
 297     msg.expe = gras_trp_select(timeout - now + start);
 298     gras_msg_recv(msg.expe, &msg);
 299     DEBUG0("Here");
 300
 301     if ( (   !msgt_want || (msg.type->code == msgt_want->code))
 302          && (!expe_want || (!strcmp( gras_socket_peer_name(msg.expe),
 303                                      gras_socket_peer_name(expe_want))))
 304          && (!filter || filter(&msg,filter_ctx))) {
 305
 306       memcpy(msg_got,&msg,sizeof(s_gras_msg_t));
 307       return;
 308     }
 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   return gras_msg_wait_ext(timeout,
 341                            msgt_want, NULL,      NULL, NULL,
 342                            &msg);
 343   memcpy(payload,msg.payl,msg.payl_size);
 344   free(msg.payl);
 345   *expeditor = msg.expe;
 346 }
 347
 348
 349 /** \brief Send the data pointed by \a payload as a message of type
 350  * \a msgtype to the peer \a sock */
 351 void
 352 gras_msg_send(gras_socket_t   sock,
 353               gras_msgtype_t  msgtype,
 354               void           *payload) {
 355
 356   gras_msg_send_ext(sock, e_gras_msg_kind_oneway,0, msgtype, payload);
 357 }
 358
 359 /** @brief Handle an incomming message or timer (or wait up to \a timeOut seconds)
 360  *
 361  * @param timeOut: How long to wait for incoming messages (in seconds)
 362  * @return the error code (or no_error).
 363  *
 364  * Messages are passed to the callbacks.
 365  */
 366 void
 367 gras_msg_handle(double timeOut) {
 368
 369   double          untiltimer;
 370
 371   int             cpt;
 372
 373   s_gras_msg_t    msg;
 374
 375   gras_msg_procdata_t pd=(gras_msg_procdata_t)gras_libdata_by_id(gras_msg_libdata_id);
 376   gras_cblist_t  *list=NULL;
 377   gras_msg_cb_t       cb;
 378   s_gras_msg_cb_ctx_t ctx;
 379
 380   int timerexpected, timeouted;
 381   xbt_ex_t e;
 382
 383   VERB1("Handling message within the next %.2fs",timeOut);
 384
 385   untiltimer = gras_msg_timer_handle();
 386   DEBUG2("[%.0f] Next timer in %f sec", gras_os_time(), untiltimer);
 387   if (untiltimer == 0.0) {
 388      /* A timer was already elapsed and handled */
 389      return;
 390   }
 391   if (untiltimer != -1.0) {
 392      timerexpected = 1;
 393      timeOut = MIN(timeOut, untiltimer);
 394   } else {
 395      timerexpected = 0;
 396   }
 397
 398   /* get a message (from the queue or from the net) */
 399   timeouted = 0;
 400   if (xbt_dynar_length(pd->msg_queue)) {
 401     DEBUG0("Get a message from the queue");
 402     xbt_dynar_shift(pd->msg_queue,&msg);
 403   } else {
 404     TRY {
 405       msg.expe = gras_trp_select(timeOut);
 406     } CATCH(e) {
 407       if (e.category != timeout_error)
 408         RETHROW;
 409       xbt_ex_free(e);
 410       timeouted = 1;
 411     }
 412
 413     if (!timeouted) {
 414       TRY {
 415         /* FIXME: if not the right kind, queue it and recall ourself or goto >:-) */
 416         gras_msg_recv(msg.expe, &msg);
 417         DEBUG0("Here");
 418
 419       } CATCH(e) {
 420         RETHROW1("Error caught while receiving a message on select()ed socket %p: %s",
 421                  msg.expe);
 422       }
 423     }
 424   }
 425
 426   if (timeouted) {
 427      if (timerexpected) {
 428
 429         /* A timer elapsed before the arrival of any message even if we select()ed a bit */
 430         untiltimer = gras_msg_timer_handle();
 431         if (untiltimer == 0.0) {
 432           /* we served a timer, we're done */
 433           return;
 434         } else {
 435            xbt_assert1(untiltimer>0, "Negative timer (%f). I'm 'puzzeled'", untiltimer);
 436            WARN1("No timer elapsed, in contrary to expectations (next in %f sec)",
 437                   untiltimer);
 438            THROW1(timeout_error,0,
 439                   "No timer elapsed, in contrary to expectations (next in %f sec)",
 440                   untiltimer);
 441         }
 442
 443      } else {
 444         /* select timeouted, and no timer elapsed. Nothing to do */
 445        THROW0(timeout_error, 0, "No new message or timer");
 446      }
 447
 448   }
 449
 450   /* A message was already there or arrived in the meanwhile. handle it */
 451   xbt_dynar_foreach(pd->cbl_list,cpt,list) {
 452     if (list->id == msg.type->code) {
 453       break;
 454     } else {
 455       list=NULL;
 456     }
 457   }
 458   if (!list) {
 459     INFO1("No callback for the incomming '%s' message. Discarded.",
 460           msg.type->name);
 461     WARN0("FIXME: gras_datadesc_free not implemented => leaking the payload");
 462     return;
 463   }
 464
 465   ctx.expeditor = msg.expe;
 466   ctx.ID = msg.ID;
 467   ctx.msgtype = msg.type;
 468
 469   switch (msg.type->kind) {
 470   case e_gras_msg_kind_oneway:
 471   case e_gras_msg_kind_rpccall:
 472     TRY {
 473       xbt_dynar_foreach(list->cbs,cpt,cb) {
 474         VERB3("Use the callback #%d (@%p) for incomming msg %s",
 475               cpt+1,cb,msg.type->name);
 476         if ((*cb)(&ctx,msg.payl)) {
 477           /* cb handled the message */
 478           free(msg.payl);
 479           return;
 480         }
 481       }
 482     } CATCH(e) {
 483       if (msg.type->kind == e_gras_msg_kind_rpccall) {
 484         /* The callback raised an exception, propagate it on the network */
 485         e.host = (char*)gras_os_myname();
 486 #ifdef HAVE_EXECINFO_H
 487         e.bt_strings = backtrace_symbols (e.bt, e.used);
 488 #endif
 489         gras_msg_send_ext(msg.expe, e_gras_msg_kind_rpcerror , msg.ID, msg.type, &e);
 490         e.host = NULL;
 491         INFO2("RPC callback raised an exception, which were propagated back to %s:%d",
 492               gras_socket_peer_name(msg.expe),  gras_socket_peer_port(msg.expe));
 493         xbt_ex_free(e);
 494         return;
 495       }
 496       RETHROW;
 497     }
 498     /* FIXME: gras_datadesc_free not implemented => leaking the payload */
 499     THROW1(mismatch_error,0,
 500            "Message '%s' refused by all registered callbacks", msg.type->name);
 501     break;
 502
 503
 504   case e_gras_msg_kind_rpcanswer:
 505     INFO1("Unexpected RPC answer discarded (type: %s)", msg.type->name);
 506     WARN0("FIXME: gras_datadesc_free not implemented => leaking the payload");
 507     return;
 508
 509   case e_gras_msg_kind_rpcerror:
 510     INFO1("Unexpected RPC error discarded (type: %s)", msg.type->name);
 511     WARN0("FIXME: gras_datadesc_free not implemented => leaking the payload");
 512     return;
 513
 514   default:
 515     THROW1(unknown_error,0,
 516            "Cannot handle messages of kind %d yet",msg.type->kind);
 517   }
 518
 519 }
 520
 521 void
 522 gras_cbl_free(void *data){
 523   gras_cblist_t *list=*(void**)data;
 524   if (list) {
 525     xbt_dynar_free(&( list->cbs ));
 526     free(list);
 527   }
 528 }
 529
 530 /** \brief Bind the given callback to the given message type
 531  *
 532  * Several callbacks can be attached to a given message type. The lastly added one will get the message first, and
 533  * if it returns false, the message will be passed to the second one.
 534  * And so on until one of the callbacks accepts the message.
 535  */
 536 void
 537 gras_cb_register(gras_msgtype_t msgtype,
 538                  gras_msg_cb_t cb) {
 539   gras_msg_procdata_t pd=(gras_msg_procdata_t)gras_libdata_by_id(gras_msg_libdata_id);
 540   gras_cblist_t *list=NULL;
 541   int cpt;
 542
 543   DEBUG2("Register %p as callback to '%s'",cb,msgtype->name);
 544
 545   /* search the list of cb for this message on this host (creating if NULL) */
 546   xbt_dynar_foreach(pd->cbl_list,cpt,list) {
 547     if (list->id == msgtype->code) {
 548       break;
 549     } else {
 550       list=NULL;
 551     }
 552   }
 553   if (!list) {
 554     /* First cb? Create room */
 555     list = xbt_new(gras_cblist_t,1);
 556     list->id = msgtype->code;
 557     list->cbs = xbt_dynar_new(sizeof(gras_msg_cb_t), NULL);
 558     xbt_dynar_push(pd->cbl_list,&list);
 559   }
 560
 561   /* Insert the new one into the set */
 562   xbt_dynar_insert_at(list->cbs,0,&cb);
 563 }
 564
 565 /** \brief Unbind the given callback from the given message type */
 566 void
 567 gras_cb_unregister(gras_msgtype_t msgtype,
 568                    gras_msg_cb_t cb) {
 569
 570   gras_msg_procdata_t pd=(gras_msg_procdata_t)gras_libdata_by_id(gras_msg_libdata_id);
 571   gras_cblist_t *list;
 572   gras_msg_cb_t cb_cpt;
 573   int cpt;
 574   int found = 0;
 575
 576   /* search the list of cb for this message on this host */
 577   xbt_dynar_foreach(pd->cbl_list,cpt,list) {
 578     if (list->id == msgtype->code) {
 579       break;
 580     } else {
 581       list=NULL;
 582     }
 583   }
 584
 585   /* Remove it from the set */
 586   if (list) {
 587     xbt_dynar_foreach(list->cbs,cpt,cb_cpt) {
 588       if (cb == cb_cpt) {
 589         xbt_dynar_cursor_rm(list->cbs, &cpt);
 590         found = 1;
 591       }
 592     }
 593   }
 594   if (!found)
 595     VERB1("Ignoring removal of unexisting callback to msg id %d",
 596           msgtype->code);
 597 }
 598
 599 /** \brief Retrieve the expeditor of the message */
 600 gras_socket_t gras_msg_cb_ctx_from(gras_msg_cb_ctx_t ctx) {
 601   return ctx->expeditor;
 602 }