- experiment driver propagates barrier crossed notification

5 files changed, 204 insertions, 50 deletions

diff --git a/src/testbed/gnunet-service-testbed.c b/src/testbed/gnunet-service-testbed.c
index 0373f4196..7d66b6fec 100644
--- a/src/testbed/gnunet-service-testbed.c
+++ b/src/testbed/gnunet-service-testbed.c
@@ -903,6 +903,8 @@ testbed_run (void *cls, struct GNUNET_SERVER_Handle *server,
      GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_INIT, 0},
     {&GST_handle_barrier_cancel, NULL, 
      GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_CANCEL, 0},
+    {&GST_handle_barrier_status, NULL,
+     GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS, 0},
     {NULL, NULL, 0, 0}
   };
   char *logfile;
diff --git a/src/testbed/gnunet-service-testbed_barriers.c b/src/testbed/gnunet-service-testbed_barriers.c
index 663c090b6..68011065d 100644
--- a/src/testbed/gnunet-service-testbed_barriers.c
+++ b/src/testbed/gnunet-service-testbed_barriers.c
@@ -26,6 +26,7 @@
 
 #include "gnunet-service-testbed.h"
 #include "gnunet-service-testbed_barriers.h"
+#include "testbed_api_barriers.h"
 
 
 /**
@@ -328,7 +329,8 @@ cleanup_clientctx (struct ClientCtx *ctx)
 {
   struct MessageQueue *mq;
   
-  GNUNET_SERVER_client_drop (ctx->client);
+  if (NULL != ctx->client)
+    GNUNET_SERVER_client_drop (ctx->client);
   if (NULL != ctx->tx)
     GNUNET_SERVER_notify_transmit_ready_cancel (ctx->tx);
   if (NULL != (mq = ctx->mq_head))
@@ -358,7 +360,13 @@ remove_barrier (struct Barrier *barrier)
   while (NULL != (ctx = barrier->head))
   {
     GNUNET_CONTAINER_DLL_remove (barrier->head, barrier->tail, ctx);
-    cleanup_clientctx (ctx);
+    GNUNET_SERVER_client_drop (ctx->client);
+    ctx->client = NULL;
+    if (NULL != ctx->tx)
+    {
+      GNUNET_SERVER_notify_transmit_ready_cancel (ctx->tx);
+      ctx->tx = NULL;
+    }
   }
   GNUNET_free (barrier->name);
   GNUNET_SERVER_client_drop (barrier->mc);
@@ -436,41 +444,6 @@ send_barrier_status_msg (struct Barrier *barrier, const char *emsg)
 }
 
 
-
-/**
- * Task for sending barrier crossed notifications to waiting client
- *
- * @param cls the barrier which is crossed
- * @param tc scheduler task context
- */
-static void
-notify_task_cb (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
-  struct Barrier *barrier = cls;
-  struct ClientCtx *client_ctx;
-  struct GNUNET_TESTBED_BarrierStatusMsg *msg;
-  struct GNUNET_MessageHeader *dup_msg;
-  uint16_t name_len;
-  uint16_t msize;
-
-  name_len = strlen (barrier->name) + 1;
-  msize = sizeof (struct GNUNET_TESTBED_BarrierStatusMsg) + name_len;  
-  msg = GNUNET_malloc (msize);
-  msg->header.size = htons (msize);
-  msg->header.type = htons (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS);
-  msg->status = htons (BARRIER_STATUS_CROSSED);
-  msg->name_len = htons (name_len);
-  (void) memcpy (msg->data, barrier->name, name_len);
-  msg->data[name_len] = '\0';
-  while (NULL != (client_ctx = barrier->head))
-  {
-    dup_msg = GNUNET_copy_message (&msg->header);
-    queue_message (client_ctx, dup_msg);
-    GNUNET_CONTAINER_DLL_remove (barrier->head, barrier->tail, client_ctx);
-  }
-}
-
-
 /**
  * Message handler for GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_WAIT messages.  This
  * message should come from peers or a shared helper service using the
@@ -538,7 +511,6 @@ handle_barrier_wait (void *cls, struct GNUNET_SERVER_Client *client,
   {
     barrier->status = BARRIER_STATUS_CROSSED;
     send_barrier_status_msg (barrier, NULL);
-    notify_task_cb (barrier, NULL);
   }
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
@@ -560,9 +532,6 @@ disconnect_cb (void *cls, struct GNUNET_SERVER_Client *client)
   if (NULL == client)
     return;
   client_ctx = GNUNET_SERVER_client_get_user_context (client, struct ClientCtx);
-  if (NULL == client_ctx)
-    return;                     /* We only set user context for locally
-                                   connected clients */
   cleanup_clientctx (client_ctx);
 }
 
@@ -777,11 +746,12 @@ GST_handle_barrier_init (void *cls, struct GNUNET_SERVER_Client *client,
     wrapper = GNUNET_malloc (sizeof (struct WBarrier));
     wrapper->barrier = barrier;
     GNUNET_CONTAINER_DLL_insert_tail (barrier->whead, barrier->wtail, wrapper);
-    wrapper->hbarrier = GNUNET_TESTBED_barrier_init (slave->controller,
-                                                     barrier->name,
-                                                     barrier->quorum,
-                                                     &wbarrier_status_cb,
-                                                     wrapper);    
+    wrapper->hbarrier = GNUNET_TESTBED_barrier_init_ (slave->controller,
+                                                      barrier->name,
+                                                      barrier->quorum,
+                                                      &wbarrier_status_cb,
+                                                      wrapper,
+                                                      GNUNET_NO);
   }
   if (NULL == barrier->whead)   /* No further propagation */
   {
@@ -856,4 +826,85 @@ GST_handle_barrier_cancel (void *cls, struct GNUNET_SERVER_Client *client,
   GNUNET_SERVER_receive_done (client, GNUNET_OK);  
 }
 
+
+/**
+ * Message handler for GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS messages.
+ * This handler is queued in the main service and will handle the messages sent
+ * either from the testbed driver or from a high level controller
+ *
+ * @param cls NULL
+ * @param client identification of the client
+ * @param message the actual message
+ */
+void
+GST_handle_barrier_status (void *cls, struct GNUNET_SERVER_Client *client,
+                           const struct GNUNET_MessageHeader *message)
+{
+  const struct GNUNET_TESTBED_BarrierStatusMsg *msg;
+  struct Barrier *barrier; 
+  struct ClientCtx *client_ctx;
+  const char *name;
+  struct GNUNET_HashCode key;
+  enum GNUNET_TESTBED_BarrierStatus status;
+  uint16_t msize;
+  uint16_t name_len;
+  
+  if (NULL == GST_context)
+  {
+    GNUNET_break_op (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }  
+  if (client != GST_context->client)
+  {
+    GNUNET_break_op (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+  msize = ntohs (message->size);
+  if (msize <= sizeof (struct GNUNET_TESTBED_BarrierStatusMsg))
+  {
+    GNUNET_break_op (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+  msg = (const struct GNUNET_TESTBED_BarrierStatusMsg *) message;
+  status = ntohs (msg->status);
+  if (BARRIER_STATUS_CROSSED != status)
+  {
+    GNUNET_break_op (0);        /* current we only expect BARRIER_CROSSED
+                                   status message this way */
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+  name = msg->data;
+  name_len = ntohs (msg->name_len);
+  if ((sizeof (struct GNUNET_TESTBED_BarrierStatusMsg) + name_len + 1) != msize)
+  {
+    GNUNET_break_op (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+  if ('\0' != name[name_len])
+  {
+    GNUNET_break_op (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+  GNUNET_CRYPTO_hash (name, name_len, &key);
+  barrier = GNUNET_CONTAINER_multihashmap_get (barrier_map, &key);
+  if (NULL == barrier)
+  {
+    GNUNET_break_op (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+  while (NULL != (client_ctx = barrier->head)) /* Notify peers */
+  {
+    queue_message (client_ctx, GNUNET_copy_message (message));
+    GNUNET_CONTAINER_DLL_remove (barrier->head, barrier->tail, client_ctx);
+  }
+}
+
 /* end of gnunet-service-testbed_barriers.c */
diff --git a/src/testbed/gnunet-service-testbed_barriers.h b/src/testbed/gnunet-service-testbed_barriers.h
index ed1b339b0..a96812086 100644
--- a/src/testbed/gnunet-service-testbed_barriers.h
+++ b/src/testbed/gnunet-service-testbed_barriers.h
@@ -76,6 +76,20 @@ void
 GST_handle_barrier_cancel (void *cls, struct GNUNET_SERVER_Client *client,
                            const struct GNUNET_MessageHeader *message);
 
+
+/**
+ * Message handler for GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS messages.
+ * This handler is queued in the main service and will handle the messages sent
+ * either from the testbed driver or from a high level controller
+ *
+ * @param cls NULL
+ * @param client identification of the client
+ * @param message the actual message
+ */
+void
+GST_handle_barrier_status (void *cls, struct GNUNET_SERVER_Client *client,
+                           const struct GNUNET_MessageHeader *message);
+
 #endif  /* GNUNET_SERVER_TESTBED_BARRIERS_H_ */
 
 /* end of gnunet-service-testbed_barriers.h */
diff --git a/src/testbed/testbed_api_barriers.c b/src/testbed/testbed_api_barriers.c
index 96fd3be66..3e4640309 100644
--- a/src/testbed/testbed_api_barriers.c
+++ b/src/testbed/testbed_api_barriers.c
@@ -27,6 +27,7 @@
 #include "platform.h"
 #include "gnunet_testbed_service.h"
 #include "testbed_api.h"
+#include "testbed_api_barriers.h"
 
 /**
  * Logging shorthand
@@ -70,6 +71,10 @@ struct GNUNET_TESTBED_Barrier
    */
   void *cls;
  
+  /**
+   * Should the barrier crossed status message be echoed back to the controller?
+   */
+  int echo;
 };
 
 
@@ -172,6 +177,8 @@ GNUNET_TESTBED_handle_barrier_status_ (struct GNUNET_TESTBED_Controller *c,
     goto cleanup;
   }
   GNUNET_assert (NULL != barrier->cb);
+  if ((GNUNET_YES == barrier->echo) && (BARRIER_STATUS_CROSSED == status))
+    GNUNET_TESTBED_queue_message_ (c, GNUNET_copy_message (&msg->header));
   barrier->cb (barrier->cls, name, barrier, status, emsg);
   if (BARRIER_STATUS_INITIALISED == status)
     return GNUNET_OK;           /* just initialised; skip cleanup */
@@ -196,13 +203,16 @@ GNUNET_TESTBED_handle_barrier_status_ (struct GNUNET_TESTBED_Controller *c,
  * @param cb the callback to call when the barrier is reached or upon error.
  *   Cannot be NULL.
  * @param cls closure for the above callback
+ * @param echo GNUNET_YES to echo the barrier crossed status message back to the
+ *   controller
  * @return barrier handle; NULL upon error
  */
 struct GNUNET_TESTBED_Barrier *
-GNUNET_TESTBED_barrier_init (struct GNUNET_TESTBED_Controller *controller,
-                             const char *name,
-                             unsigned int quorum,
-                             GNUNET_TESTBED_barrier_status_cb cb, void *cls)
+GNUNET_TESTBED_barrier_init_ (struct GNUNET_TESTBED_Controller *controller,
+                              const char *name,
+                              unsigned int quorum,
+                              GNUNET_TESTBED_barrier_status_cb cb, void *cls,
+                              int echo)
 {
   struct GNUNET_TESTBED_BarrierInit *msg;
   struct GNUNET_TESTBED_Barrier *barrier;
@@ -229,6 +239,7 @@ GNUNET_TESTBED_barrier_init (struct GNUNET_TESTBED_Controller *controller,
   barrier->name = GNUNET_strdup (name);
   barrier->cb = cb;
   barrier->cls = cls;
+  barrier->echo = echo;
   (void) memcpy (&barrier->key, &key, sizeof (struct GNUNET_HashCode));
   GNUNET_assert (GNUNET_OK ==
                  GNUNET_CONTAINER_multihashmap_put (barrier_map, &barrier->key,
@@ -246,6 +257,30 @@ GNUNET_TESTBED_barrier_init (struct GNUNET_TESTBED_Controller *controller,
 
 
 /**
+ * Initialise a barrier and call the given callback when the required percentage
+ * of peers (quorum) reach the barrier OR upon error.
+ *
+ * @param controller the handle to the controller
+ * @param name identification name of the barrier
+ * @param quorum the percentage of peers that is required to reach the barrier.
+ *   Peers signal reaching a barrier by calling
+ *   GNUNET_TESTBED_barrier_reached().
+ * @param cb the callback to call when the barrier is reached or upon error.
+ *   Cannot be NULL.
+ * @param cls closure for the above callback
+ * @return barrier handle; NULL upon error
+ */
+struct GNUNET_TESTBED_Barrier *
+GNUNET_TESTBED_barrier_init (struct GNUNET_TESTBED_Controller *controller,
+                             const char *name,
+                             unsigned int quorum,
+                             GNUNET_TESTBED_barrier_status_cb cb, void *cls)
+{
+  return GNUNET_TESTBED_barrier_init_ (controller, name, quorum, cb, cls, GNUNET_YES);
+}
+
+
+/**
  * Cancel a barrier.
  *
  * @param barrier the barrier handle
diff --git a/src/testbed/testbed_api_barriers.h b/src/testbed/testbed_api_barriers.h
new file mode 100644
index 000000000..89aa56ca5
--- /dev/null
+++ b/src/testbed/testbed_api_barriers.h
@@ -0,0 +1,52 @@
+/*
+  This file is part of GNUnet.
+  (C) 2008--2013 Christian Grothoff (and other contributing authors)
+
+  GNUnet is free software; you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published
+  by the Free Software Foundation; either version 3, or (at your
+  option) any later version.
+
+  GNUnet is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+  General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with GNUnet; see the file COPYING.  If not, write to the
+  Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+  Boston, MA 02111-1307, USA.
+*/
+
+/**
+ * @file testbed/testbed_api_barriers.h
+ * @brief interface for barriers API function that are used internally (not
+ *   exposed as user API)
+ * @author Sree Harsha Totakura <sreeharsha@totakura.in> 
+ */
+
+#include "gnunet_testbed_service.h"
+
+
+/**
+ * Initialise a barrier and call the given callback when the required percentage
+ * of peers (quorum) reach the barrier OR upon error.
+ *
+ * @param controller the handle to the controller
+ * @param name identification name of the barrier
+ * @param quorum the percentage of peers that is required to reach the barrier.
+ *   Peers signal reaching a barrier by calling
+ *   GNUNET_TESTBED_barrier_reached().
+ * @param cb the callback to call when the barrier is reached or upon error.
+ *   Cannot be NULL.
+ * @param cls closure for the above callback
+ * @param echo GNUNET_YES to echo the barrier crossed status message back to the
+ *   controller
+ * @return barrier handle; NULL upon error
+ */
+struct GNUNET_TESTBED_Barrier *
+GNUNET_TESTBED_barrier_init_ (struct GNUNET_TESTBED_Controller *controller,
+                              const char *name,
+                              unsigned int quorum,
+                              GNUNET_TESTBED_barrier_status_cb cb, void *cls,
+                              int echo);