changes for experimentation

6 files changed, 541 insertions, 10 deletions

diff --git a/src/experimentation/Makefile.am b/src/experimentation/Makefile.am
index 479a4e9be..effbbbd00 100644
--- a/src/experimentation/Makefile.am
+++ b/src/experimentation/Makefile.am
@@ -18,8 +18,7 @@ endif
 
 
 if HAVE_EXPERIMENTAL
- TEXT_EXP_CLIQUE = test_experimentation_clique_connect 
-#test_experimentation_clique_run
+ TEXT_EXP_CLIQUE = test_experimentation_clique_connect test_experimentation_clique_run
 endif 
 
 check_PROGRAMS = \
@@ -51,6 +50,13 @@ test_experimentation_clique_connect_LDADD = \
  $(top_builddir)/src/util/libgnunetutil.la \
  $(top_builddir)/src/statistics/libgnunetstatistics.la \
  $(top_builddir)/src/testbed/libgnunettestbed.la
+
+test_experimentation_clique_run_SOURCES = \
+ test_experimentation_clique_run.c
+test_experimentation_clique_run_LDADD = \
+ $(top_builddir)/src/util/libgnunetutil.la \
+ $(top_builddir)/src/statistics/libgnunetstatistics.la \
+ $(top_builddir)/src/testbed/libgnunettestbed.la
  
 #test_experimentation_clique_run_SOURCES = \
 # test_experimentation_clique_run.c
diff --git a/src/experimentation/gnunet-daemon-experimentation.c b/src/experimentation/gnunet-daemon-experimentation.c
index 4ecc03d7d..cd8806985 100644
--- a/src/experimentation/gnunet-daemon-experimentation.c
+++ b/src/experimentation/gnunet-daemon-experimentation.c
@@ -96,9 +96,7 @@ run (void *cls, char *const *args, const char *cfgfile,
         }
 
         GNUNET_EXPERIMENTATION_nodes_start ();
-        /*
   GNUNET_EXPERIMENTATION_scheduler_start ();
-*/
   GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, &shutdown_task,
                                 NULL);
 }
diff --git a/src/experimentation/gnunet-daemon-experimentation.h b/src/experimentation/gnunet-daemon-experimentation.h
index 999c565ad..649017c7f 100644
--- a/src/experimentation/gnunet-daemon-experimentation.h
+++ b/src/experimentation/gnunet-daemon-experimentation.h
@@ -165,6 +165,9 @@ struct Node
          * Array of fssuer ids
          */
         struct GNUNET_PeerIdentity *issuer_id;
+
+        struct ExperimentStartCtx *e_req_head;
+        struct ExperimentStartCtx *e_req_tail;
 };
 
 struct Experimentation_Issuer
@@ -205,8 +208,11 @@ struct Experimentation_Response
         uint32_t issuer_count;
 };
 
-void
-GNUNET_EXPERIMENT_nodes_request_start (struct Node *n, struct Experiment *e);
+int
+GNUNET_EXPERIMENTATION_nodes_rts (struct Node *n);
+
+int
+GNUNET_EXPERIMENTATION_nodes_request_start (struct Node *n, struct Experiment *e);
 
 
 /**
diff --git a/src/experimentation/gnunet-daemon-experimentation_nodes.c b/src/experimentation/gnunet-daemon-experimentation_nodes.c
index a77e841b8..0782e7ff1 100644
--- a/src/experimentation/gnunet-daemon-experimentation_nodes.c
+++ b/src/experimentation/gnunet-daemon-experimentation_nodes.c
@@ -547,6 +547,44 @@ static void handle_response (const struct GNUNET_PeerIdentity *peer,
 }
 
 /**
+ * Handle a response
+ *
+ * @param peer the source
+ * @param message the message
+ */
+static void handle_start (const struct GNUNET_PeerIdentity *peer,
+                                                                                                                 const struct GNUNET_MessageHeader *message)
+{
+fprintf (stderr, "FIXME\n");
+        GNUNET_STATISTICS_update (GSE_stats, "# experiments running",
+                        1, GNUNET_NO);
+}
+
+/**
+ * Handle a response
+ *
+ * @param peer the source
+ * @param message the message
+ */
+static void handle_start_ack (const struct GNUNET_PeerIdentity *peer,
+                                                                                                                 const struct GNUNET_MessageHeader *message)
+{
+
+}
+
+/**
+ * Handle a response
+ *
+ * @param peer the source
+ * @param message the message
+ */
+static void handle_stop (const struct GNUNET_PeerIdentity *peer,
+                                                                                                                 const struct GNUNET_MessageHeader *message)
+{
+
+}
+
+/**
  * Method called whenever a given peer connects.
  *
  * @param cls closure
@@ -619,6 +657,15 @@ core_receive_handler (void *cls,
                 case GNUNET_MESSAGE_TYPE_EXPERIMENTATION_RESPONSE:
                         handle_response (other, message);
                         break;
+                case GNUNET_MESSAGE_TYPE_EXPERIMENTATION_START:
+                        handle_start (other, message);
+                        break;
+                case GNUNET_MESSAGE_TYPE_EXPERIMENTATION_START_ACK:
+                        handle_start_ack (other, message);
+                        break;
+                case GNUNET_MESSAGE_TYPE_EXPERIMENTATION_STOP:
+                        handle_stop (other, message);
+                        break;
                 default:
                         break;
         }
@@ -626,12 +673,88 @@ core_receive_handler (void *cls,
         return GNUNET_OK;
 }
 
+#define FAST_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
 
-void
-GNUNET_EXPERIMENT_nodes_request_start (struct Node *n, struct Experiment *e)
+struct GNUNET_EXPERIMENTATION_start_message
 {
-        GNUNET_log (GNUNET_ERROR_TYPE_INFO, _("Sending start request to peer `%s' for experiment `%s'\n"),
+        struct GNUNET_MessageHeader header;
+};
+
+struct ExperimentStartCtx
+{
+        struct ExperimentStartCtx *prev;
+        struct ExperimentStartCtx *next;
+
+        struct Node *n;
+        struct Experiment *e;
+};
+
+size_t node_experiment_start_cb (void *cls, size_t bufsize, void *buf)
+{
+        struct ExperimentStartCtx *e_ctx = cls;
+        struct GNUNET_EXPERIMENTATION_start_message msg;
+
+        GNUNET_CONTAINER_DLL_remove (e_ctx->n->e_req_head, e_ctx->n->e_req_tail, e_ctx);
+        e_ctx->n->cth = NULL;
+        if (NULL == buf)
+        {
+                GNUNET_free (e_ctx);
+                return 0;
+        }
+
+        size_t size = sizeof (struct GNUNET_EXPERIMENTATION_start_message);
+        msg.header.size = htons (size);
+        msg.header.type = htons (GNUNET_MESSAGE_TYPE_EXPERIMENTATION_START);
+
+        memcpy (buf, &msg, size);
+        GNUNET_free (e_ctx);
+        return size;
+}
+
+int
+GNUNET_EXPERIMENTATION_nodes_rts (struct Node *n)
+{
+        if (NULL == n->cth)
+                return GNUNET_YES;
+        else
+                return GNUNET_NO;
+
+}
+
+/**
+ * Request a experiment to start with a node
+ *
+ * @return GNUNET_NO if core was busy with sending, GNUNET_OK otherwise
+ */
+int
+GNUNET_EXPERIMENTATION_nodes_request_start (struct Node *n, struct Experiment *e)
+{
+        struct ExperimentStartCtx *e_ctx;
+
+        if (NULL != n->cth)
+        {
+                GNUNET_break (0); /* should call rts before */
+                return GNUNET_NO;
+        }
+
+        GNUNET_log (GNUNET_ERROR_TYPE_INFO, _("Sending experiment start request to peer `%s' for experiment `%s'\n"),
                         GNUNET_i2s(&n->id), e->name);
+
+        e_ctx = GNUNET_malloc (sizeof (struct ExperimentStartCtx));
+        e_ctx->n = n;
+        e_ctx->e = e;
+        n->cth = GNUNET_CORE_notify_transmit_ready (ch, GNUNET_NO, 0, FAST_TIMEOUT, &n->id,
+                        sizeof (struct GNUNET_EXPERIMENTATION_start_message),
+                        &node_experiment_start_cb, e_ctx);
+        if (NULL == n->cth)
+        {
+                GNUNET_log (GNUNET_ERROR_TYPE_INFO, _("Cannot send experiment start request to peer `%s' for experiment `%s'\n"),
+                                GNUNET_i2s(&n->id), e->name);
+                GNUNET_free (e_ctx);
+        }
+        GNUNET_CONTAINER_DLL_insert (n->e_req_head, n->e_req_tail, e_ctx);
+
+        return GNUNET_OK;
 }
 
 
diff --git a/src/experimentation/gnunet-daemon-experimentation_scheduler.c b/src/experimentation/gnunet-daemon-experimentation_scheduler.c
index 795c993dd..5d2a51438 100644
--- a/src/experimentation/gnunet-daemon-experimentation_scheduler.c
+++ b/src/experimentation/gnunet-daemon-experimentation_scheduler.c
@@ -34,6 +34,7 @@
 enum ExperimentState
 {
         NOT_RUNNING,
+        BUSY,
         REQUESTED,
         STARTED,
         STOPPED
@@ -78,13 +79,27 @@ static void start_experiment (void *cls,const struct GNUNET_SCHEDULER_TaskContex
 {
         struct ScheduledExperiment *se = cls;
         struct GNUNET_TIME_Relative end;
+        struct GNUNET_TIME_Relative backoff;
+
         se->task = GNUNET_SCHEDULER_NO_TASK;
 
+        if (GNUNET_NO == GNUNET_EXPERIMENTATION_nodes_rts (se->n))
+        {
+                se->state = BUSY;
+                backoff = GNUNET_TIME_UNIT_SECONDS;
+                backoff.rel_value += GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 1000);
+                GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Delaying start request to peer `%s' for `%s' for %llu ms\n",
+                                GNUNET_i2s (&se->n->id), se->e->name, (unsigned long long) backoff.rel_value);
+                se->task = GNUNET_SCHEDULER_add_delayed (backoff, &start_experiment, se);
+                return;
+        }
+        else if (BUSY == se->state)
+                se->state = NOT_RUNNING;
 
         if (NOT_RUNNING == se->state)
         {
                         /* Send start message */
-                        GNUNET_EXPERIMENT_nodes_request_start (se->n, se->e);
+                        GNUNET_EXPERIMENTATION_nodes_request_start (se->n, se->e);
                         se->state = REQUESTED;
                         se->task = GNUNET_SCHEDULER_add_delayed (EXP_RESPONSE_TIMEOUT, &request_timeout, se);
 
@@ -114,6 +129,7 @@ static void start_experiment (void *cls,const struct GNUNET_SCHEDULER_TaskContex
                                 se->state = STOPPED;
                                 return; /* End of experiment is reached */
                         }
+                        /* Reschedule */
                 se->task = GNUNET_SCHEDULER_add_delayed (se->e->frequency, &start_experiment, se);
         }
 
@@ -127,6 +143,25 @@ static void start_experiment (void *cls,const struct GNUNET_SCHEDULER_TaskContex
  * Start the scheduler component
  */
 void
+GNUNET_EXPERIMENTATION_scheduler_handle_start (struct Node *n, struct Experiment *e)
+{
+
+}
+
+
+/**
+ * Start the scheduler component
+ */
+void
+GNUNET_EXPERIMENTATION_scheduler_handle_stop (struct Node *n, struct Experiment *e)
+{
+
+}
+
+/**
+ * Start the scheduler component
+ */
+void
 GNUNET_EXPERIMENTATION_scheduler_add (struct Node *n, struct Experiment *e)
 {
         struct ScheduledExperiment *se;
diff --git a/src/experimentation/test_experimentation_clique_run.c b/src/experimentation/test_experimentation_clique_run.c
new file mode 100644
index 000000000..35f48bce2
--- /dev/null
+++ b/src/experimentation/test_experimentation_clique_run.c
@@ -0,0 +1,363 @@
+/*
+  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 src/experimentation/test_experimentation_clique_run.c
+ * @brief test case to run experiments with experimentation daemons in a clique
+ * @author Sree Harsha Totakura <sreeharsha@totakura.in>
+ * @author Matthias Wachs
+ */
+
+#include "platform.h"
+#include "gnunet_common.h"
+#include "gnunet_testbed_service.h"
+
+
+/**
+ * Number of peers we want to start
+ */
+#define NUM_PEERS 2
+
+#define NUM_ISSUER 1
+
+#define NUM_EXPERIMENTS 2
+
+#define TEST_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, (5 * NUM_PEERS) + 20)
+
+/**
+ * Array of peers
+ */
+static struct GNUNET_TESTBED_Peer **peers;
+
+/**
+ * Operation handle
+ */
+static struct GNUNET_TESTBED_Operation *op;
+
+/**
+ * Shutdown task
+ */
+static GNUNET_SCHEDULER_TaskIdentifier shutdown_task;
+
+/**
+ * Testing result
+ */
+static int result;
+
+/**
+ * Counter for counting overlay connections
+ */
+static unsigned int overlay_connects;
+
+/**
+ * Information we track for a peer in the testbed.
+ */
+struct ExperimentationPeer
+{
+  /**
+   * Handle with testbed.
+   */
+  struct GNUNET_TESTBED_Peer *daemon;
+
+  /**
+   * Testbed operation to connect to statistics service
+   */
+  struct GNUNET_TESTBED_Operation *stat_op;
+
+  /**
+   * Handle to the statistics service
+   */
+  struct GNUNET_STATISTICS_Handle *sh;
+
+  unsigned int active_nodes;
+  unsigned int requested_nodes;
+  unsigned int inactive_nodes;
+  unsigned int issuer;
+  unsigned int experiments_active;
+  unsigned int experiments_running;
+};
+
+
+struct ExperimentationPeer ph[NUM_PEERS];
+
+/**
+ * Shutdown nicely
+ *
+ * @param cls NULL
+ * @param tc the task context
+ */
+static void
+do_shutdown (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  unsigned int peer;
+        shutdown_task = GNUNET_SCHEDULER_NO_TASK;
+
+  for (peer = 0; peer < NUM_PEERS; peer++)
+  {
+        if (NULL != ph[peer].stat_op)
+                GNUNET_TESTBED_operation_done (ph[peer].stat_op);
+        ph[peer].stat_op = NULL;
+  }
+
+  if (NULL != op)
+  {
+    GNUNET_TESTBED_operation_done (op);
+    op = NULL;
+  }
+  GNUNET_SCHEDULER_shutdown ();
+}
+
+/**
+ * Controller event callback
+ *
+ * @param cls NULL
+ * @param event the controller event
+ */
+static void
+controller_event_cb (void *cls,
+                     const struct GNUNET_TESTBED_EventInformation *event)
+{
+  switch (event->type)
+  {
+  case GNUNET_TESTBED_ET_CONNECT:
+    overlay_connects++;
+    if ((NUM_PEERS * (NUM_PEERS - 1)) == overlay_connects)
+    {
+      result = GNUNET_OK;
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO, "All %u peers connected \n", NUM_PEERS);
+      if (GNUNET_SCHEDULER_NO_TASK != shutdown_task)
+      {
+                GNUNET_SCHEDULER_cancel (shutdown_task);
+      }
+      shutdown_task = GNUNET_SCHEDULER_add_delayed (TEST_TIMEOUT, do_shutdown, NULL);
+    }
+    break;
+  case GNUNET_TESTBED_ET_OPERATION_FINISHED:
+    break;
+  default:
+    GNUNET_break (0);
+    result = GNUNET_SYSERR;
+    GNUNET_SCHEDULER_cancel (shutdown_task);
+    shutdown_task = GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
+  }
+}
+
+static void
+check_end ()
+{
+
+        static int last_experiments_value = 0;
+  unsigned int peer;
+  unsigned int t_running_experiments = 0;
+
+        for (peer = 0; peer < NUM_PEERS; peer++)
+        {
+                t_running_experiments += ph[peer].experiments_running;
+        }
+
+        if (last_experiments_value < t_running_experiments)
+                fprintf (stderr, ".");
+        last_experiments_value = t_running_experiments;
+
+        if (t_running_experiments == (NUM_PEERS * NUM_EXPERIMENTS))
+        {
+                        fprintf (stderr, "\n");
+                        GNUNET_log (GNUNET_ERROR_TYPE_INFO, "All %u peers are running experiments\n", NUM_PEERS);
+                        GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
+        }
+}
+
+
+
+/**
+ * Callback function to process statistic values.
+ *
+ * @param cls struct StatsContext
+ * @param subsystem name of subsystem that created the statistic
+ * @param name the name of the datum
+ * @param value the current value
+ * @param is_persistent GNUNET_YES if the value is persistent, GNUNET_NO if not
+ * @return GNUNET_OK to continue, GNUNET_SYSERR to abort iteration
+ */
+static int
+stat_iterator (void *cls, const char *subsystem, const char *name,
+                     uint64_t value, int is_persistent)
+{
+  struct ExperimentationPeer *peer = cls;
+
+        if (0 == strcmp (name, "# experiments active"))
+        {
+                        peer->experiments_active = value;
+        }
+
+        if (0 == strcmp (name, "# experiments running"))
+        {
+                        peer->experiments_running = value;
+        }
+
+        check_end ();
+
+        return GNUNET_OK;
+}
+
+/**
+ * Called after successfully opening a connection to a peer's statistics
+ * service; we register statistics monitoring here.
+ *
+ * @param cls the callback closure from functions generating an operation
+ * @param op the operation that has been finished
+ * @param ca_result the service handle returned from GNUNET_TESTBED_ConnectAdapter()
+ * @param emsg error message in case the operation has failed; will be NULL if
+ *          operation has executed successfully.
+ */
+static void
+stat_comp_cb (void *cls, struct GNUNET_TESTBED_Operation *op,
+              void *ca_result, const char *emsg )
+{
+  //struct GNUNET_STATISTICS_Handle *sh = ca_result;
+  struct ExperimentationPeer *peer = cls;
+
+  if (NULL != emsg)
+  {
+    GNUNET_break (0);
+    return;
+  }
+
+  GNUNET_break (GNUNET_OK == GNUNET_STATISTICS_watch
+                (peer->sh, "experimentation", "# experiments active",
+                 stat_iterator, peer));
+  GNUNET_break (GNUNET_OK == GNUNET_STATISTICS_watch
+                (peer->sh, "experimentation", "# experiments running",
+                 stat_iterator, peer));
+}
+
+/**
+ * Called to open a connection to the peer's statistics
+ *
+ * @param cls peer context
+ * @param cfg configuration of the peer to connect to; will be available until
+ *          GNUNET_TESTBED_operation_done() is called on the operation returned
+ *          from GNUNET_TESTBED_service_connect()
+ * @return service handle to return in 'op_result', NULL on error
+ */
+static void *
+stat_connect_adapter (void *cls,
+                      const struct GNUNET_CONFIGURATION_Handle *cfg)
+{
+  struct ExperimentationPeer *peer = cls;
+  peer->sh = GNUNET_STATISTICS_create ("experimentation", cfg);
+  if (NULL == peer->sh)
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Failed to create statistics \n");
+  return peer->sh;
+}
+
+
+/**
+ * Called to disconnect from peer's statistics service
+ *
+ * @param cls peer context
+ * @param op_result service handle returned from the connect adapter
+ */
+static void
+stat_disconnect_adapter (void *cls, void *op_result)
+{
+  struct ExperimentationPeer *peer = cls;
+  GNUNET_break (GNUNET_OK == GNUNET_STATISTICS_watch_cancel
+                (peer->sh, "experimentation", "# experiments active",
+                 stat_iterator, peer));
+  GNUNET_break (GNUNET_OK == GNUNET_STATISTICS_watch_cancel
+                (peer->sh, "experimentation", "# experiments running",
+                 stat_iterator, peer));
+  GNUNET_STATISTICS_destroy (op_result, GNUNET_NO);
+  peer->sh = NULL;
+}
+
+
+
+/**
+ * Signature of a main function for a testcase.
+ *
+ * @param cls closure
+ * @param num_peers number of peers in 'peers'
+ * @param peers_ handle to peers run in the testbed
+ * @param links_succeeded the number of overlay link connection attempts that
+ *          succeeded
+ * @param links_failed the number of overlay link connection attempts that
+ *          failed
+ */
+static void
+test_master (void *cls, unsigned int num_peers,
+             struct GNUNET_TESTBED_Peer **peers_,
+             unsigned int links_succeeded,
+             unsigned int links_failed)
+{
+  unsigned int peer;
+
+  GNUNET_assert (NULL == cls);
+  GNUNET_assert (NUM_PEERS == num_peers);
+  GNUNET_assert (NULL != peers_);
+  for (peer = 0; peer < num_peers; peer++)
+  {
+    GNUNET_assert (NULL != peers_[peer]);
+    /* Connect to peer's statistic service */
+    ph[peer].stat_op = GNUNET_TESTBED_service_connect (NULL,
+                                                                                                                                peers_[peer], "statistics",
+                                                                                                                                &stat_comp_cb, &ph[peer],
+                                    &stat_connect_adapter,
+                                    &stat_disconnect_adapter,
+                                    &ph[peer]);
+
+  }
+  peers = peers_;
+  overlay_connects = 0;
+  op = GNUNET_TESTBED_overlay_configure_topology (NULL, NUM_PEERS, peers, NULL,
+                                                  NULL,
+                                                  NULL,
+                                                  GNUNET_TESTBED_TOPOLOGY_CLIQUE,
+                                                  /* GNUNET_TESTBED_TOPOLOGY_ERDOS_RENYI, */
+                                                  /* NUM_PEERS, */
+                                                  GNUNET_TESTBED_TOPOLOGY_OPTION_END);
+  GNUNET_assert (NULL != op);
+  shutdown_task = GNUNET_SCHEDULER_add_delayed (TEST_TIMEOUT, do_shutdown, NULL);
+}
+
+
+/**
+ * Main function
+ */
+int
+main (int argc, char **argv)
+{
+  uint64_t event_mask;
+
+  result = GNUNET_SYSERR;
+  event_mask = 0;
+  event_mask |= (1LL << GNUNET_TESTBED_ET_CONNECT);
+  event_mask |= (1LL << GNUNET_TESTBED_ET_OPERATION_FINISHED);
+  (void) GNUNET_TESTBED_test_run ("test_experimentation_clique_run",
+                                  "test_experimentation_clique.conf", NUM_PEERS,
+                                  event_mask, &controller_event_cb, NULL,
+                                  &test_master, NULL);
+  if (GNUNET_OK != result)
+    return 1;
+  return 0;
+}
+
+/* end of test_experimentation_clique_run.c */