Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Let's shild the user against himself: check that he provided a payload variable when...
[simgrid.git] / src / gras / Msg / msg.c
index 6a4c329..a867919 100644 (file)
@@ -8,6 +8,7 @@
  * under the terms of the license (GNU LGPL) which comes with this package. */
 
 #include "xbt/ex.h"
+#include "xbt/ex_interface.h"
 #include "gras/Msg/msg_private.h"
 #include "gras/Virtu/virtu_interface.h"
 #include "gras/DataDesc/datadesc_interface.h"
@@ -311,7 +312,7 @@ gras_msg_wait_ext(double           timeout,
     xbt_dynar_push(pd->msg_queue,&msg);
     
     now=gras_os_time();
-    if (now - start + 0.001 < timeout) {
+    if (now - start + 0.001 > timeout) {
       THROW1(timeout_error,  now-start+0.001-timeout,
             "Timeout while waiting for msg %s",msgt_want->name);
     }
@@ -341,6 +342,16 @@ gras_msg_wait(double           timeout,
                    msgt_want, NULL,      NULL, NULL,
                    &msg);
 
+  if (msgt_want->ctn_type) {
+    xbt_assert1(payload,
+               "Message type '%s' convey a payload you must accept",
+               msgt_want->name);
+  } else {
+    xbt_assert1(!payload,
+               "No payload was declared for message type '%s'",
+               msgt_want->name);
+  }
+
   if (payload) {
     memcpy(payload,msg.payl,msg.payl_size);
     free(msg.payl);
@@ -358,15 +369,52 @@ gras_msg_send(gras_socket_t   sock,
              gras_msgtype_t  msgtype,
              void           *payload) {
 
+  if (msgtype->ctn_type) {
+    xbt_assert1(payload,
+               "Message type '%s' convey a payload you must provide",
+               msgtype->name);
+  } else {
+    xbt_assert1(!payload,
+               "No payload was declared for message type '%s'",
+               msgtype->name);
+  }
+
   gras_msg_send_ext(sock, e_gras_msg_kind_oneway,0, msgtype, payload);
 }
 
+/** @brief Handle all messages arriving within the given period
+ *
+ * @param timeOut: How long to wait for incoming messages (in seconds)
+ * @return the error code (or no_error).
+ *
+ * Messages are dealed with just like gras_msg_handle() would do. The
+ * difference is that gras_msg_handle() handles at most one message (or wait up
+ * to timeout second when no message arrives) while this function handles any
+ * amount of messages, and lasts the given period in any case.
+ */
+void 
+gras_msg_handleall(double period) {
+  xbt_ex_t e;
+  double begin=gras_os_time();
+  double now;
+
+  do {
+    now=gras_os_time();
+    TRY{
+      gras_msg_handle(period - now + begin);
+    } CATCH(e) {
+      if (e.category != timeout_error) 
+       RETHROW0("Error while waiting for messages: %s");
+      xbt_ex_free(e);
+    }
+  } while (now - begin < period);
+}
 /** @brief Handle an incomming message or timer (or wait up to \a timeOut seconds)
  *
  * @param timeOut: How long to wait for incoming messages (in seconds)
  * @return the error code (or no_error).
  *
- * Messages are passed to the callbacks.
+ * Messages are passed to the callbacks. See also gras_msg_handleall().
  */
 void
 gras_msg_handle(double timeOut) {
@@ -423,8 +471,10 @@ gras_msg_handle(double timeOut) {
               e_gras_msg_kind_names[msg.kind]);
     
       } CATCH(e) {
-       RETHROW1("Error caught while receiving a message on select()ed socket %p: %s",
-                msg.expe);
+       RETHROW4("Error while receiving a message on select()ed socket %p to [%s]%s:%d: %s",
+                msg.expe,
+                gras_socket_peer_proc(msg.expe),gras_socket_peer_name(msg.expe),
+                gras_socket_peer_port(msg.expe));
       }
     }
   }
@@ -448,7 +498,8 @@ gras_msg_handle(double timeOut) {
        
      } else {
        /* select timeouted, and no timer elapsed. Nothing to do */
-       THROW0(timeout_error, 0, "No new message or timer");
+       THROW1(timeout_error, 0, "No new message or timer (delay was %f)",
+             timeOut);
      }
      
   }
@@ -478,13 +529,14 @@ gras_msg_handle(double timeOut) {
     ran_ok=0;
     TRY {
       xbt_dynar_foreach(list->cbs,cpt,cb) { 
-       VERB3("Use the callback #%d (@%p) for incomming msg %s",
-             cpt+1,cb,msg.type->name);
-       if ((*cb)(&ctx,msg.payl)) {
-         /* cb handled the message */
-         free(msg.payl);
-         ran_ok = 1;
-         break;
+       if (!ran_ok) {
+         VERB3("Use the callback #%d (@%p) for incomming msg %s",
+               cpt+1,cb,msg.type->name);
+         if ((*cb)(&ctx,msg.payl)) {
+           /* cb handled the message */
+           free(msg.payl);
+           ran_ok = 1;
+         }
        }
       }
     } CATCH(e) {
@@ -493,21 +545,20 @@ gras_msg_handle(double timeOut) {
        /* The callback raised an exception, propagate it on the network */
        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);
-#endif
+         xbt_ex_setup_backtrace(&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",
+       VERB4("Propagate %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;
+       gras_msg_send_ext(msg.expe, e_gras_msg_kind_rpcerror,
+                         msg.ID, msg.type, &e);
        xbt_ex_free(e);
-      } else
-       RETHROW;
+       ran_ok=1;
+      } else {
+       RETHROW0("Callback raised an exception: %s");
+      }
     }
     if (!ran_ok)
       THROW1(mismatch_error,0,