} CATCH(e) {
if (e.category != not_found_error)
RETHROW;
- xbt_ex_free(e);
+ xbt_ex_free(&e);
}
if (found) {
TRY {
res = (gras_msgtype_t)xbt_set_get_by_name(_gras_msgtype_set, namev);
} CATCH(e) {
- xbt_ex_free(e);
+ xbt_ex_free(&e);
THROW1(not_found_error,0,"No registred message of that name: %s",name);
}
if (name != namev)
* @param timeout: How long should we wait for this message.
* @param msgt_want: type of awaited msg (or NULL if I'm enclined to accept any message)
* @param expe_want: awaited expeditot (match on hostname, not port; NULL if not relevant)
- * @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)
+ * @param filter: function returning true or false when passed a payload. Messages for which it returns false are not selected. (NULL if not relevant)
* @param filter_ctx: context passed as second argument of the filter (a pattern to match?)
- * @param[out] msgt_got: where to write the descriptor of the message we got
- * @param[out] expe_got: where to create a socket to answer the incomming message
- * @param[out] payl_got: where to write the payload of the incomming message
+ * @param[out] msg_got: where to write the message we got
*
* Every message of another type received before the one waited will be queued
* and used by subsequent call to this function or gras_msg_handle().
while (1) {
memset(&msg,sizeof(msg),0);
- msg.expe = gras_trp_select(timeout - now + start);
+ msg.expe = gras_trp_select(timeout ? timeout - now + start : 0);
gras_msg_recv(msg.expe, &msg);
- DEBUG0("Here");
+ DEBUG0("Got a message from the socket");
if ( ( !msgt_want || (msg.type->code == msgt_want->code))
&& (!expe_want || (!strcmp( gras_socket_peer_name(msg.expe),
&& (!filter || filter(&msg,filter_ctx))) {
memcpy(msg_got,&msg,sizeof(s_gras_msg_t));
+ DEBUG0("Message matches expectations. Use it.");
return;
}
+ DEBUG0("Message does not match expectations. Queue it.");
/* not expected msg type. Queue it for later */
xbt_dynar_push(pd->msg_queue,&msg);
void *payload) {
s_gras_msg_t msg;
- return gras_msg_wait_ext(timeout,
- msgt_want, NULL, NULL, NULL,
- &msg);
- memcpy(payload,msg.payl,msg.payl_size);
- free(msg.payl);
- *expeditor = msg.expe;
+ gras_msg_wait_ext(timeout,
+ msgt_want, NULL, NULL, NULL,
+ &msg);
+
+ if (payload) {
+ memcpy(payload,msg.payl,msg.payl_size);
+ free(msg.payl);
+ }
+
+ if (expeditor)
+ *expeditor = msg.expe;
}
double untiltimer;
- int cpt;
+ int cpt, ran_ok;
s_gras_msg_t msg;
VERB1("Handling message within the next %.2fs",timeOut);
untiltimer = gras_msg_timer_handle();
- DEBUG2("[%.0f] Next timer in %f sec", gras_os_time(), untiltimer);
+ DEBUG1("Next timer in %f sec", untiltimer);
if (untiltimer == 0.0) {
/* A timer was already elapsed and handled */
return;
} CATCH(e) {
if (e.category != timeout_error)
RETHROW;
- xbt_ex_free(e);
+ xbt_ex_free(&e);
timeouted = 1;
}
TRY {
/* FIXME: if not the right kind, queue it and recall ourself or goto >:-) */
gras_msg_recv(msg.expe, &msg);
- DEBUG0("Here");
+ DEBUG1("Received a msg from the socket kind:%s",
+ e_gras_msg_kind_names[msg.kind]);
} CATCH(e) {
RETHROW1("Error caught while receiving a message on select()ed socket %p: %s",
ctx.ID = msg.ID;
ctx.msgtype = msg.type;
- switch (msg.type->kind) {
+ switch (msg.kind) {
case e_gras_msg_kind_oneway:
case e_gras_msg_kind_rpccall:
+ ran_ok=0;
TRY {
xbt_dynar_foreach(list->cbs,cpt,cb) {
VERB3("Use the callback #%d (@%p) for incomming msg %s",
if ((*cb)(&ctx,msg.payl)) {
/* cb handled the message */
free(msg.payl);
- return;
+ ran_ok = 1;
+ break;
}
}
} CATCH(e) {
+ free(msg.payl);
if (msg.type->kind == e_gras_msg_kind_rpccall) {
/* The callback raised an exception, propagate it on the network */
- e.host = (char*)gras_os_myname();
+ if (!e.remote) { /* the exception is born on this machine */
+ e.host = (char*)gras_os_myname();
#ifdef HAVE_EXECINFO_H
- e.bt_strings = backtrace_symbols (e.bt, e.used);
+ e.bt_strings = backtrace_symbols (e.bt, e.used);
#endif
- gras_msg_send_ext(msg.expe, e_gras_msg_kind_rpcerror , msg.ID, msg.type, &e);
+ }
+ gras_msg_send_ext(msg.expe, e_gras_msg_kind_rpcerror,
+ msg.ID, msg.type, &e);
+ INFO4("Propagated %s exception from '%s' RPC cb back to %s:%d",
+ (e.remote ? "remote" : "local"),
+ msg.type->name,
+ gras_socket_peer_name(msg.expe),
+ gras_socket_peer_port(msg.expe));
e.host = NULL;
- INFO2("RPC callback raised an exception, which were propagated back to %s:%d",
- gras_socket_peer_name(msg.expe), gras_socket_peer_port(msg.expe));
- xbt_ex_free(e);
- return;
- }
- RETHROW;
+ xbt_ex_free(&e);
+ } else
+ RETHROW;
}
+ if (!ran_ok)
+ THROW1(mismatch_error,0,
+ "Message '%s' refused by all registered callbacks", msg.type->name);
/* FIXME: gras_datadesc_free not implemented => leaking the payload */
- THROW1(mismatch_error,0,
- "Message '%s' refused by all registered callbacks", msg.type->name);
break;
default:
THROW1(unknown_error,0,
- "Cannot handle messages of kind %d yet",msg.type->kind);
+ "Cannot handle messages of kind %d yet",msg.type->kind);
}
}
