- towards testbed barriers

11 files changed, 935 insertions, 11 deletions

diff --git a/src/include/gnunet_protocols.h b/src/include/gnunet_protocols.h
index 04745c5f9..a52727f4d 100644
--- a/src/include/gnunet_protocols.h
+++ b/src/include/gnunet_protocols.h
@@ -1566,10 +1566,27 @@ extern "C"
 #define GNUNET_MESSAGE_TYPE_TESTBED_MANAGE_PEER_SERVICE 483
 
 /**
+ * Message to initialise a barrier.  Messages of these type are flooded to all
+ * sub-controllers
+ */
+#define GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_INIT 484
+
+/**
+ * Message for signalling status of a barrier
+ */
+#define GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS 485
+
+/**
+ * Message sent by a peer when it has reached a barrier and is waiting for it to
+ * be crossed
+ */
+#define GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_WAIT 486
+
+/**
  * Not really a message, but for careful checks on the testbed messages; Should
  * always be the maximum and never be used to send messages with this type
  */
-#define GNUNET_MESSAGE_TYPE_TESTBED_MAX 484
+#define GNUNET_MESSAGE_TYPE_TESTBED_MAX 487
 
 /**
  * The initialization message towards gnunet-testbed-helper
diff --git a/src/include/gnunet_testbed_service.h b/src/include/gnunet_testbed_service.h
index 6249707d9..173aa76b3 100644
--- a/src/include/gnunet_testbed_service.h
+++ b/src/include/gnunet_testbed_service.h
@@ -1471,6 +1471,63 @@ GNUNET_TESTBED_test_run (const char *testname,
                          void *test_master_cls);
 
 
+/**
+ * Opaque handle for barrier
+ */
+struct GNUNET_TESTBED_Barrier;
+
+
+/**
+ * Functions of this type are to be given as callback argument to
+ * GNUNET_TESTBED_barrier_init().  The callback will be called when status
+ * information is available for the barrier.
+ *
+ * @param cls the closure given to GNUNET_TESTBED_barrier_init()
+ * @param name the name of the barrier
+ * @param barrier the barrier handle
+ * @param status status of the barrier; GNUNET_OK if the barrier is crossed;
+ *   GNUNET_SYSERR upon error
+ * @param emsg if the status were to be GNUNET_SYSERR, this parameter has the
+ *   error messsage
+ */
+typedef void (*GNUNET_TESTBED_barrier_status_cb) (void *cls,
+                                                  const char *name,
+                                                  struct GNUNET_TESTBED_Barrier
+                                                  *barrier,
+                                                  int status,
+                                                  const char *emsg);
+
+
+/**
+ * Initialise a barrier and call the given callback when the required percentage
+ * of peers (quorum) reach the barrier.  
+ *
+ * @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
+ */
+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);
+
+
+/**
+ * Cancel a barrier.
+ *
+ * @param barrier the barrier handle
+ */
+void
+GNUNET_TESTBED_barrier_cancel (struct GNUNET_TESTBED_Barrier *barrier);
+
+
 #if 0                           /* keep Emacsens' auto-indent happy */
 {
 #endif
diff --git a/src/testbed/Makefile.am b/src/testbed/Makefile.am
index 3aa88453b..f07639b8a 100644
--- a/src/testbed/Makefile.am
+++ b/src/testbed/Makefile.am
@@ -39,7 +39,8 @@ gnunet_service_testbed_SOURCES = \
   gnunet-service-testbed_oc.c \
   gnunet-service-testbed_cpustatus.c \
   gnunet-service-testbed_meminfo.c \
-  gnunet-service-testbed_meminfo.h
+  gnunet-service-testbed_meminfo.h \
+  gnunet-service-testbed_barriers.c
 gnunet_service_testbed_LDADD = $(XLIB) \
  $(top_builddir)/src/util/libgnunetutil.la \
  $(top_builddir)/src/core/libgnunetcore.la \
@@ -98,7 +99,8 @@ libgnunettestbed_la_SOURCES = \
   testbed_api_testbed.c \
   testbed_api_test.c \
   testbed_api_topology.c testbed_api_topology.h \
-  testbed_api_sd.c testbed_api_sd.h
+  testbed_api_sd.c testbed_api_sd.h \
+  testbed_api_barriers.c 
 libgnunettestbed_la_LIBADD = $(XLIB) \
  $(top_builddir)/src/core/libgnunetcore.la \
  $(top_builddir)/src/statistics/libgnunetstatistics.la \
diff --git a/src/testbed/gnunet-service-testbed.c b/src/testbed/gnunet-service-testbed.c
index de038eeae..3b01bc504 100644
--- a/src/testbed/gnunet-service-testbed.c
+++ b/src/testbed/gnunet-service-testbed.c
@@ -34,7 +34,7 @@
 /**
  * Our configuration
  */
-struct GNUNET_CONFIGURATION_Handle *our_config;
+struct GNUNET_CONFIGURATION_Handle *GST_config;
 
 /**
  * The master context; generated with the first INIT message
@@ -516,11 +516,11 @@ handle_init (void *cls, struct GNUNET_SERVER_Client *client,
   }
   ss_str = NULL;
   ss = NULL;
-  if (GNUNET_OK == GNUNET_CONFIGURATION_get_value_string (our_config, "TESTBED",
+  if (GNUNET_OK == GNUNET_CONFIGURATION_get_value_string (GST_config, "TESTBED",
                                                           "SHARED_SERVICES",
                                                           &ss_str))
   {
-    ss = parse_shared_services (ss_str, our_config);
+    ss = parse_shared_services (ss_str, GST_config);
     GNUNET_free (ss_str);
     ss_str = NULL;
   }
@@ -546,7 +546,7 @@ handle_init (void *cls, struct GNUNET_SERVER_Client *client,
   host =
       GNUNET_TESTBED_host_create_with_id (GST_context->host_id,
                                           GST_context->master_ip, NULL,
-                                          our_config, 0);
+                                          GST_config, 0);
   host_list_add (host);
   LOG_DEBUG ("Created master context with host ID: %u\n", GST_context->host_id);
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
@@ -824,7 +824,7 @@ shutdown_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
     GNUNET_free (mq_entry);
   }
   GNUNET_free_non_null (hostname);
-  GNUNET_CONFIGURATION_destroy (our_config);
+  GNUNET_CONFIGURATION_destroy (GST_config);
   /* Free hello cache */
   GST_cache_clear ();
   GNUNET_TESTBED_operation_queue_destroy_ (GST_opq_openfds);
@@ -929,7 +929,7 @@ testbed_run (void *cls, struct GNUNET_SERVER_Handle *server,
   GNUNET_assert (GNUNET_OK ==
                  GNUNET_CONFIGURATION_get_value_string (cfg, "testbed",
                                                         "HOSTNAME", &hostname));
-  our_config = GNUNET_CONFIGURATION_dup (cfg);
+  GST_config = GNUNET_CONFIGURATION_dup (cfg);
   GNUNET_SERVER_add_handlers (server, message_handlers);
   GNUNET_SERVER_disconnect_notify (server, &client_disconnect_cb, NULL);
   shutdown_task_id =
@@ -937,7 +937,7 @@ testbed_run (void *cls, struct GNUNET_SERVER_Handle *server,
                                                   GNUNET_SCHEDULER_PRIORITY_IDLE,
                                                   &shutdown_task, NULL);
   LOG_DEBUG ("Testbed startup complete\n");
-  GST_stats_init (our_config);
+  GST_stats_init (GST_config);
 }
 
 
diff --git a/src/testbed/gnunet-service-testbed.h b/src/testbed/gnunet-service-testbed.h
index 38d1ae80e..55861faba 100644
--- a/src/testbed/gnunet-service-testbed.h
+++ b/src/testbed/gnunet-service-testbed.h
@@ -412,7 +412,7 @@ struct HandlerContext_ShutdownPeers
 /**
  * Our configuration
  */
-extern struct GNUNET_CONFIGURATION_Handle *our_config;
+extern struct GNUNET_CONFIGURATION_Handle *GST_config;
 
 /**
  * The master context; generated with the first INIT message
@@ -455,6 +455,11 @@ const extern struct GNUNET_TIME_Relative GST_timeout;
 extern unsigned int GST_peer_list_size;
 
 /**
+ * The current number of peers running locally under this controller
+ */
+extern unsigned int GST_num_local_peers;
+
+/**
  * The size of the host list
  */
 extern unsigned int GST_host_list_size;
diff --git a/src/testbed/gnunet-service-testbed_barriers.c b/src/testbed/gnunet-service-testbed_barriers.c
new file mode 100644
index 000000000..079096d86
--- /dev/null
+++ b/src/testbed/gnunet-service-testbed_barriers.c
@@ -0,0 +1,512 @@
+/*
+  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/gnunet-service-testbed_barriers.c
+ * @brief barrier handling at the testbed controller
+ * @author Sree Harsha Totakura <sreeharsha@totakura.in> 
+ */
+
+#include "gnunet-service-testbed.h"
+
+/**
+ * timeout for outgoing message transmissions in seconds
+ */
+#define MESSAGE_SEND_TIMEOUT(s) \
+  GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, s)
+
+
+/**
+ * Barrier
+ */
+struct Barrier;
+
+
+/**
+ * Message queue for transmitting messages
+ */
+struct MessageQueue
+{
+  /**
+   * next pointer for DLL
+   */
+  struct MessageQueue *next;
+
+  /**
+   * prev pointer for DLL
+   */
+  struct MessageQueue *prev;
+
+  /**
+   * The message to be sent
+   */
+  struct GNUNET_MessageHeader *msg;
+};
+
+/**
+ * Context to be associated with each client
+ */
+struct ClientCtx
+{
+  /**
+   * The barrier this client is waiting for
+   */
+  struct Barrier *barrier;
+
+  /**
+   * DLL next ptr
+   */
+  struct ClientCtx *next;
+
+  /**
+   * DLL prev ptr
+   */
+  struct ClientCtx *prev;
+
+  /**
+   * The client handle
+   */
+  struct GNUNET_SERVER_Client *client;
+
+  /**
+   * the transmission handle
+   */
+  struct GNUNET_SERVER_TransmitHandle *tx;
+
+  /**
+   * message queue head
+   */
+  struct MessageQueue *mq_head;
+
+  /**
+   * message queue tail
+   */
+  struct MessageQueue *mq_tail;
+};
+
+
+/**
+ * Barrier
+ */
+struct Barrier
+{
+  /**
+   * The hashcode of the barrier name
+   */
+  struct GNUNET_HashCode hash;
+
+  /**
+   * The name of the barrier
+   */
+  char *name;
+
+  /**
+   * DLL head for the list of clients waiting for this barrier
+   */
+  struct ClientCtx *head;
+
+  /**
+   * DLL tail for the list of clients waiting for this barrier
+   */
+  struct ClientCtx *tail;
+
+  /**
+   * Number of peers which have reached this barrier
+   */
+  unsigned int nreached;
+
+  /**
+   * Number of slaves we have initialised this barrier
+   */
+  unsigned int nslaves;
+
+  /**
+   * Quorum percentage to be reached
+   */
+  uint8_t quorum;
+  
+  /**
+   * Was there a timeout while propagating initialisation
+   */
+  uint8_t timedout;
+};
+
+
+/**
+ * Hashtable handle for storing initialised barriers
+ */
+static struct GNUNET_CONTAINER_MultiHashMap *barrier_map;
+
+/**
+ * Service context
+ */
+static struct GNUNET_SERVICE_Context *ctx;
+
+
+/**
+ * Function called to notify a client about the connection
+ * begin ready to queue more data.  "buf" will be
+ * NULL and "size" zero if the connection was closed for
+ * writing in the meantime.
+ *
+ * @param cls client context
+ * @param size number of bytes available in buf
+ * @param buf where the callee should write the message
+ * @return number of bytes written to buf
+ */
+static size_t 
+transmit_ready_cb (void *cls, size_t size, void *buf)
+{
+  struct ClientCtx *ctx = cls;
+  struct GNUNET_SERVER_Client *client = ctx->client;
+  struct MessageQueue *mq;
+  struct GNUNET_MessageHeader *msg;
+  size_t wrote;
+
+  ctx->tx = NULL;
+  wrote = 0;
+  if ((0 == size) || (NULL == buf))
+  {
+    GNUNET_assert (NULL != ctx->client);
+    GNUNET_SERVER_client_drop (ctx->client);
+    ctx->client = NULL;    
+    return 0;
+  }
+  mq = ctx->mq_head;
+  msg = mq->msg;
+  wrote = ntohs (msg->size);
+  GNUNET_assert (size >= wrote);
+  (void) memcpy (buf, msg, wrote);
+  GNUNET_CONTAINER_DLL_remove (ctx->mq_head, ctx->mq_tail, mq);
+  GNUNET_free (mq->msg);
+  GNUNET_free (mq);
+  if (NULL != (mq = ctx->mq_head))
+    ctx->tx = GNUNET_SERVER_notify_transmit_ready (client, ntohs (msg->size),
+                                                  MESSAGE_SEND_TIMEOUT (30),
+                                                  &transmit_ready_cb, ctx);
+  return wrote;
+}
+
+
+/**
+ * Queue a message into a clients message queue
+ *
+ * @param ctx the context associated with the client
+ * @param msg the message to queue.  Will be consumed
+ * @param suspended is the client suspended at the time of calling queue_message
+ */
+static void
+queue_message (struct ClientCtx *ctx, struct GNUNET_MessageHeader *msg)
+{
+  struct MessageQueue *mq;
+  struct GNUNET_SERVER_Client *client = ctx->client;
+  
+  mq = GNUNET_malloc (sizeof (struct MessageQueue));
+  mq->msg = msg;
+  GNUNET_CONTAINER_DLL_insert_tail (ctx->mq_head, ctx->mq_tail, mq);
+  if (NULL == ctx->tx)
+   ctx->tx = GNUNET_SERVER_notify_transmit_ready (client, ntohs (msg->size),
+                                                  MESSAGE_SEND_TIMEOUT (30),
+                                                  &transmit_ready_cb, ctx);
+}
+
+
+#if 0
+/**
+ * Function to remove a barrier from the barrier map and cleanup resources
+ * occupied by a barrier
+ *
+ * @param barrier the barrier handle
+ */
+static void
+remove_barrier (struct Barrier *barrier)
+{
+  GNUNET_assert (GNUNET_YES == GNUNET_CONTAINER_multihashmap_remove (barrier_map,
+                                                                    &barrier->hash,
+                                                                    barrier));
+  GNUNET_free (barrier->name);
+  GNUNET_free (barrier);
+}
+
+
+/**
+ * Function called upon timeout while waiting for a response from the
+ * subcontrollers to barrier init message
+ *
+ * @param 
+ * @return 
+ */
+static void
+fwd_tout_barrier_init (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct ForwardedOperationContext *foctx = cls;
+  struct Barrier *barrier = foctx->cls;
+  
+  barrier->nslaves--;
+  barrier->timedout = GNUNET_YES;
+  if (0 == barrier->nslaves)
+  {
+    GST_send_operation_fail_msg (foctx->client, foctx->operation_id,
+                                 "Timeout while contacting a slave controller");
+    remove_barrier (barrier);
+  }
+}
+#endif
+
+/**
+ * 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_BarrierStatus *msg;
+  struct GNUNET_MessageHeader *dup_msg;
+  uint16_t name_len;
+  uint16_t msize;
+
+  name_len = strlen (barrier->name) + 1;
+  msize = sizeof (struct GNUNET_TESTBED_BarrierStatus) + name_len;  
+  msg = GNUNET_malloc (msize);
+  msg->header.size = htons (msize);
+  msg->header.type = htons (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS);
+  msg->status = 0;
+  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);
+    GNUNET_SERVER_client_set_user_context_ (client_ctx->client, NULL, 0);
+    GNUNET_free (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
+ * testbed-barrier client API (@see gnunet_testbed_barrier_service.h)
+ *
+ * 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
+ */
+static void
+handle_barrier_wait (void *cls, struct GNUNET_SERVER_Client *client,
+                     const struct GNUNET_MessageHeader *message)
+{
+  const struct GNUNET_TESTBED_BarrierWait *msg;
+  struct Barrier *barrier;
+  char *name;
+  struct ClientCtx *client_ctx;
+  struct GNUNET_HashCode key;
+  size_t name_len;
+  uint16_t msize;
+  
+  msize = ntohs (message->size);
+  if (msize <= sizeof (struct GNUNET_TESTBED_BarrierWait))
+  {
+    GNUNET_break_op (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+  if (NULL == barrier_map)
+  {
+    GNUNET_break (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+  msg = (const struct GNUNET_TESTBED_BarrierWait *) message;
+  name_len = msize - sizeof (struct GNUNET_TESTBED_BarrierWait);
+  name = GNUNET_malloc (name_len + 1);
+  name[name_len] = '\0';
+  (void) memcpy (name, msg->name, name_len);
+  GNUNET_CRYPTO_hash (name, name_len - 1, &key);
+  if (NULL == (barrier = GNUNET_CONTAINER_multihashmap_get (barrier_map, &key)))
+  {
+    GNUNET_break (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    GNUNET_free (name);
+    return;
+  }
+  client_ctx = GNUNET_SERVER_client_get_user_context (client, struct ClientCtx);
+  if (NULL == client_ctx)
+  {
+    client_ctx = GNUNET_malloc (sizeof (struct ClientCtx));
+    client_ctx->client = client;
+    GNUNET_SERVER_client_keep (client);
+    client_ctx->barrier = barrier;
+    GNUNET_CONTAINER_DLL_insert_tail (barrier->head, barrier->tail, client_ctx);
+    barrier->nreached++;
+    if ((barrier->quorum * GST_num_local_peers) <= (barrier->nreached * 100))
+      notify_task_cb (barrier, NULL);
+  }
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+}
+
+
+/**
+ * Functions with this signature are called whenever a client
+ * is disconnected on the network level.
+ *
+ * @param cls closure
+ * @param client identification of the client; NULL
+ *        for the last call when the server is destroyed
+ */
+static void
+disconnect_cb (void *cls, struct GNUNET_SERVER_Client *client)
+{
+  struct ClientCtx *client_ctx;
+  struct Barrier *barrier;
+  
+  client_ctx = GNUNET_SERVER_client_get_user_context (client, struct ClientCtx);
+  if (NULL == client_ctx)
+    return;
+  barrier = client_ctx->barrier;
+  GNUNET_CONTAINER_DLL_remove (barrier->head, barrier->tail, client_ctx);
+  if (NULL != client_ctx->tx)
+    GNUNET_SERVER_notify_transmit_ready_cancel (client_ctx->tx);
+  
+}
+
+
+/**
+ * Function to initialise barrriers component
+ */
+void
+GST_barriers_init (struct GNUNET_CONFIGURATION_Handle *cfg)
+{
+  static const struct GNUNET_SERVER_MessageHandler message_handlers[] = {
+    {&handle_barrier_wait, NULL, GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_WAIT, 0},
+    {NULL, NULL, 0, 0}
+  };
+  struct GNUNET_SERVER_Handle *srv;
+
+  barrier_map = GNUNET_CONTAINER_multihashmap_create (3, GNUNET_YES);
+  ctx = GNUNET_SERVICE_start ("testbed-barrier", cfg,
+                              GNUNET_SERVICE_OPTION_MANUAL_SHUTDOWN);
+  srv = GNUNET_SERVICE_get_server (ctx);
+  GNUNET_SERVER_add_handlers (srv, message_handlers);
+  GNUNET_SERVER_disconnect_notify (srv, &disconnect_cb, NULL);  
+}
+
+
+/**
+ * Function to stop the barrier service
+ */
+void
+GST_barriers_stop ()
+{
+  GNUNET_assert (NULL != ctx);
+  GNUNET_SERVICE_stop (ctx);
+}
+
+
+/**
+ * Message handler for GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_INIT messages.  This
+ * message should always come from a parent controller or the testbed API if we
+ * are the root controller.
+ *
+ * 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_init (void *cls, struct GNUNET_SERVER_Client *client,
+                         const struct GNUNET_MessageHeader *message)
+{
+  const struct GNUNET_TESTBED_BarrierInit *msg;
+  const char *name;
+  struct Barrier *barrier;
+  struct Slave *slave;
+  struct GNUNET_HashCode hash;
+  size_t name_len;
+  uint64_t op_id;
+  unsigned int cnt;
+  uint16_t msize;
+  
+  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_BarrierInit))
+  {
+    GNUNET_break_op (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+  msg = (const struct GNUNET_TESTBED_BarrierInit *) message;
+  op_id = GNUNET_ntohll (msg->op_id);
+  name = msg->name;
+  name_len = (size_t) msize - sizeof (struct GNUNET_TESTBED_BarrierInit);
+  GNUNET_CRYPTO_hash (name, name_len, &hash);
+  if (GNUNET_YES == GNUNET_CONTAINER_multihashmap_contains (barrier_map, &hash))
+  {
+    GST_send_operation_fail_msg (client, op_id, "Barrier already initialised");
+    GNUNET_SERVER_receive_done (client, GNUNET_OK);
+    return;
+  }
+  barrier = GNUNET_malloc (sizeof (struct Barrier));
+  (void) memcpy (&barrier->hash, &hash, sizeof (struct GNUNET_HashCode));
+  barrier->quorum = msg->quorum;
+  barrier->name = GNUNET_malloc (name_len + 1);
+  barrier->name[name_len] = '\0';
+  (void) memcpy (barrier->name, name, name_len);
+  GNUNET_assert (GNUNET_OK ==
+                 GNUNET_CONTAINER_multihashmap_put (barrier_map,
+                                                    &barrier->hash,
+                                                    barrier,
+                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+  /* Propagate barrier init to subcontrollers */
+  for (cnt = 0; cnt < GST_slave_list_size; cnt++)
+  {
+    if (NULL == (slave = GST_slave_list[cnt]))
+      continue;
+    if (NULL == slave->controller)
+    {
+      GNUNET_break (0);/* May happen when we are connecting to the controller */
+      continue;
+    }    
+    GNUNET_break (0);           /* FIXME */
+  }
+}
diff --git a/src/testbed/gnunet-service-testbed_peers.c b/src/testbed/gnunet-service-testbed_peers.c
index ece0e83e9..1792373a3 100644
--- a/src/testbed/gnunet-service-testbed_peers.c
+++ b/src/testbed/gnunet-service-testbed_peers.c
@@ -35,6 +35,11 @@
  */
 struct Peer **GST_peer_list;
 
+/**
+ * The current number of peers running locally under this controller
+ */
+unsigned int GST_num_local_peers;
+
 
 /**
  * Context information to manage peers' services
@@ -162,6 +167,8 @@ peer_list_add (struct Peer *peer)
     GST_array_grow_large_enough (GST_peer_list, GST_peer_list_size, peer->id);
   GNUNET_assert (NULL == GST_peer_list[peer->id]);
   GST_peer_list[peer->id] = peer;
+  if (GNUNET_NO == peer->is_remote)
+    GST_num_local_peers++;
 }
 
 
@@ -176,6 +183,8 @@ peer_list_remove (struct Peer *peer)
   unsigned int orig_size;
   uint32_t id;
 
+  if (GNUNET_NO == peer->is_remote)
+    GST_num_local_peers--;
   GST_peer_list[peer->id] = NULL;
   orig_size = GST_peer_list_size;
   while (GST_peer_list_size >= LIST_GROW_STEP)
diff --git a/src/testbed/testbed.h b/src/testbed/testbed.h
index 66f8ddb17..3b159591d 100644
--- a/src/testbed/testbed.h
+++ b/src/testbed/testbed.h
@@ -768,6 +768,84 @@ struct GNUNET_TESTBED_ManagePeerServiceMessage
 };
 
 
+/**
+ * Message to initialise a barrier
+ */
+struct GNUNET_TESTBED_BarrierInit
+{
+  /**
+   * Type is GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_INIT
+   */
+  struct GNUNET_MessageHeader header;
+
+  /**
+   * Unused.  Only of alignment.
+   */
+  uint32_t unused;
+  
+  /**
+   * The operation id
+   */
+  uint64_t op_id;
+  
+  /**
+   * The quorum percentage needed for crossing the barrier
+   */
+  uint8_t quorum;
+
+  /**
+   * name of the barrier.  Non NULL-terminated.
+   */
+  char name[0];
+};
+
+
+/**
+ * Message for signalling status changes of a barrier
+ */
+struct GNUNET_TESTBED_BarrierStatus
+{
+  /**
+   * Type is GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS
+   */
+  struct GNUNET_MessageHeader header;
+  
+  /**
+   * status.  0 to signal success (barrier is crossed).  1 for error.
+   */
+  uint16_t status;
+
+  /**
+   * strlen of the barrier name
+   */
+  uint16_t name_len;
+  
+  /**
+   * the barrier name (NULL terminated) concatenated with an error message (NULL
+   * terminated) if the status were to indicate an error
+   */
+  char data[0];
+};
+
+
+/**
+ * Message sent from peers to the testbed-barrier service to indicate that they
+ * have reached a barrier and are waiting for it to be crossed
+ */
+struct GNUNET_TESTBED_BarrierWait
+{
+  /**
+   * Type is GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_WAIT
+   */
+  struct GNUNET_MessageHeader header;
+
+  /**
+   * The name of the barrier they have reached.  Non-NULL terminated.
+   */
+  char name[0];
+};
+
+
 GNUNET_NETWORK_STRUCT_END
 #endif
 /* end of testbed.h */
diff --git a/src/testbed/testbed_api.c b/src/testbed/testbed_api.c
index 48b7fe189..0a109aa7f 100644
--- a/src/testbed/testbed_api.c
+++ b/src/testbed/testbed_api.c
@@ -1159,6 +1159,13 @@ message_handler (void *cls, const struct GNUNET_MessageHeader *msg)
                                          GNUNET_TESTBED_ControllerLinkResponse
                                          *) msg);
     break;
+  case GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS:
+    status =
+        GNUNET_TESTBED_handle_barrier_status_ (c,
+                                               (const struct
+                                                GNUNET_TESTBED_BarrierStatus *)
+                                               msg);
+    break;
   default:
     GNUNET_assert (0);
   }
diff --git a/src/testbed/testbed_api.h b/src/testbed/testbed_api.h
index dbeea9243..f658c08dd 100644
--- a/src/testbed/testbed_api.h
+++ b/src/testbed/testbed_api.h
@@ -496,6 +496,22 @@ GNUNET_TESTBED_controller_link_ (void *op_cls,
                                  *slave_cfg, int is_subordinate);
 
 
+/**
+ * Handler for GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS messages.  This
+ * function is defined in @file testbed_api_barriers.c
+ *
+ * @param c the controller handle to determine the connection this message
+ *   belongs to
+ * @param msg the barrier status message
+ * @return GNUNET_OK to keep the connection active; GNUNET_SYSERR to tear it
+ *   down signalling an error
+ */
+int
+GNUNET_TESTBED_handle_barrier_status_ (struct GNUNET_TESTBED_Controller *c,
+                                       const struct GNUNET_TESTBED_BarrierStatus
+                                       *msg);
+
+
 
 
 #endif
diff --git a/src/testbed/testbed_api_barriers.c b/src/testbed/testbed_api_barriers.c
new file mode 100644
index 000000000..231e00600
--- /dev/null
+++ b/src/testbed/testbed_api_barriers.c
@@ -0,0 +1,221 @@
+/*
+  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.c
+ * @brief API implementation for testbed barriers
+ * @author Sree Harsha Totakura <sreeharsha@totakura.in> 
+ */
+
+#include "platform.h"
+#include "gnunet_testbed_service.h"
+#include "testbed_api.h"
+
+/**
+ * Handle for barrier
+ */
+struct GNUNET_TESTBED_Barrier
+{
+  /**
+   * hashcode identifying this barrier in the hashmap
+   */
+  struct GNUNET_HashCode key;
+
+  /**
+   * The controller handle given while initiliasing this barrier
+   */
+  struct GNUNET_TESTBED_Controller *c;
+  
+  /**
+   * The name of the barrier
+   */
+  char *name;
+
+  /**
+   * The continuation callback to call when we have a status update on this
+   */
+  GNUNET_TESTBED_barrier_status_cb cb;
+
+  /**
+   * the closure for the above callback
+   */
+  void *cls;
+ 
+};
+
+
+/**
+ * handle for hashtable of barrier handles
+ */
+static struct GNUNET_CONTAINER_MultiHashMap *barrier_map;
+
+
+/**
+ * Remove a barrier and it was the last one in the barrier hash map, destroy the
+ * hash map
+ *
+ * @param barrier the barrier to remove
+ */
+static void
+barrier_remove (struct GNUNET_TESTBED_Barrier *barrier)
+{
+  GNUNET_assert (NULL != barrier_map); /* No barriers present */
+  GNUNET_assert (GNUNET_OK ==
+                 GNUNET_CONTAINER_multihashmap_remove (barrier_map,
+                                                       &barrier->key,
+                                                       barrier));
+  GNUNET_free (barrier->name);
+  GNUNET_free (barrier);
+  if (0 == GNUNET_CONTAINER_multihashmap_size (barrier_map))
+  {
+    GNUNET_CONTAINER_multihashmap_destroy (barrier_map);
+    barrier_map = NULL;
+  }
+}
+
+
+/**
+ * Handler for GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS messages
+ *
+ * @param c the controller handle to determine the connection this message
+ *   belongs to
+ * @param msg the barrier status message
+ * @return GNUNET_OK to keep the connection active; GNUNET_SYSERR to tear it
+ *   down signalling an error
+ */
+int
+GNUNET_TESTBED_handle_barrier_status_ (struct GNUNET_TESTBED_Controller *c,
+                                       const struct GNUNET_TESTBED_BarrierStatus
+                                       *msg)
+{
+  struct GNUNET_TESTBED_Barrier *barrier;
+  char *emsg;
+  const char *name;
+  struct GNUNET_HashCode key;  
+  size_t emsg_len;
+  int status;
+  uint16_t msize;
+  uint16_t name_len;
+  
+  emsg = NULL;
+  barrier = NULL;
+  msize = ntohs (msg->header.size);  
+  name = msg->data;
+  name_len = ntohs (msg->name_len);
+  if (  (sizeof (struct GNUNET_TESTBED_BarrierStatus) + name_len + 1 > msize)
+        || ('\0' != name[name_len])  )
+  {
+    GNUNET_break_op (0);
+    return GNUNET_SYSERR;
+  }
+  status = ntohs (msg->status);
+  if (0 != status)
+  {
+    status = -1;
+    emsg_len = msize - (sizeof (struct GNUNET_TESTBED_BarrierStatus) + name_len
+                        + 1);
+    if (0 == emsg_len)
+    {
+      GNUNET_break_op (0);
+      return GNUNET_SYSERR;
+    }
+    emsg_len++;
+    emsg = GNUNET_malloc (emsg_len);
+    emsg_len--;
+    emsg[emsg_len] = '\0';
+    (void) memcpy (emsg, msg->data + name_len + 1, emsg_len);
+  }
+  if (NULL == barrier_map)
+    goto cleanup;
+  GNUNET_CRYPTO_hash (name, name_len, &key);
+  barrier = GNUNET_CONTAINER_multihashmap_get (barrier_map, &key);
+  if (NULL == barrier)
+    goto cleanup;
+  GNUNET_assert (NULL != barrier->cb);
+  barrier->cb (barrier->cls, name, barrier, status, emsg);
+
+ cleanup:
+  GNUNET_free_non_null (emsg);
+  if (NULL != barrier)
+    barrier_remove (barrier);
+  return GNUNET_OK;
+}
+
+
+/**
+ * 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)
+{
+  struct GNUNET_TESTBED_Barrier *barrier;
+  struct GNUNET_HashCode key;
+  size_t name_len;
+  
+  GNUNET_assert (quorum <= 100);
+  GNUNET_assert (NULL != cb);
+  name_len = strlen (name);
+  GNUNET_assert (0 < name_len);
+  GNUNET_CRYPTO_hash (name, name_len, &key);
+  if (NULL == barrier_map)
+    barrier_map = GNUNET_CONTAINER_multihashmap_create (3, GNUNET_YES);
+  if (GNUNET_YES ==
+      GNUNET_CONTAINER_multihashmap_contains (barrier_map, &key))
+  {
+    GNUNET_break (0);
+    return NULL;
+  }
+  barrier = GNUNET_malloc (sizeof (struct GNUNET_TESTBED_Barrier));
+  barrier->name = GNUNET_strdup (name);
+  barrier->cb = cb;
+  barrier->cls = cls;
+  (void) memcpy (&barrier->key, &key, sizeof (struct GNUNET_HashCode));
+  GNUNET_assert (GNUNET_OK ==
+                 GNUNET_CONTAINER_multihashmap_put (barrier_map, &barrier->key,
+                                                    barrier,
+                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
+  return barrier;
+}
+
+
+/**
+ * Cancel a barrier.
+ *
+ * @param barrier the barrier handle
+ */
+void
+GNUNET_TESTBED_barrier_cancel (struct GNUNET_TESTBED_Barrier *barrier)
+{
+  barrier_remove (barrier);
+}