fix 2893: Move adaptive parallelisation mechanism to operation queues

10 files changed, 418 insertions, 340 deletions

diff --git a/src/testbed/gnunet-service-testbed.c b/src/testbed/gnunet-service-testbed.c
index 3710b82e1..f49e0331f 100644
--- a/src/testbed/gnunet-service-testbed.c
+++ b/src/testbed/gnunet-service-testbed.c
@@ -918,7 +918,8 @@ testbed_run (void *cls, struct GNUNET_SERVER_Handle *server,
   GNUNET_assert (GNUNET_OK ==
                  GNUNET_CONFIGURATION_get_value_number (cfg, "TESTBED",
                                                         "MAX_OPEN_FDS", &num));
-  GST_opq_openfds = GNUNET_TESTBED_operation_queue_create_ ((unsigned int) num);
+  GST_opq_openfds = GNUNET_TESTBED_operation_queue_create_
+      (OPERATION_QUEUE_TYPE_FIXED, (unsigned int) num);
   GNUNET_assert (GNUNET_OK ==
                  GNUNET_CONFIGURATION_get_value_time (cfg, "TESTBED",
                                                       "OPERATION_TIMEOUT",
diff --git a/src/testbed/test_testbed_api_operations.c b/src/testbed/test_testbed_api_operations.c
index 31b3add0a..feca147b7 100644
--- a/src/testbed/test_testbed_api_operations.c
+++ b/src/testbed/test_testbed_api_operations.c
@@ -490,9 +490,9 @@ static void
 run (void *cls, char *const *args, const char *cfgfile,
      const struct GNUNET_CONFIGURATION_Handle *config)
 {
-  q1 = GNUNET_TESTBED_operation_queue_create_ (1);
+  q1 = GNUNET_TESTBED_operation_queue_create_ (OPERATION_QUEUE_TYPE_FIXED, 1);
   GNUNET_assert (NULL != q1);
-  q2 = GNUNET_TESTBED_operation_queue_create_ (2);
+  q2 = GNUNET_TESTBED_operation_queue_create_ (OPERATION_QUEUE_TYPE_FIXED, 2);
   GNUNET_assert (NULL != q2);
   op1 = GNUNET_TESTBED_operation_create_ (&op1, start_cb, release_cb);
   GNUNET_assert (NULL != op1);
diff --git a/src/testbed/testbed_api.c b/src/testbed/testbed_api.c
index 010d51776..48b7fe189 100644
--- a/src/testbed/testbed_api.c
+++ b/src/testbed/testbed_api.c
@@ -860,7 +860,7 @@ handle_op_fail_event (struct GNUNET_TESTBED_Controller *c,
     struct OverlayConnectData *data;
 
     data = opc->data;
-    data->failed = GNUNET_YES;
+    GNUNET_TESTBED_operation_mark_failed (opc->op);
     if (NULL != data->cb)
       data->cb (data->cb_cls, opc->op, emsg);
   }
@@ -1486,13 +1486,15 @@ GNUNET_TESTBED_controller_connect (struct GNUNET_TESTBED_Host *host,
   GNUNET_TESTBED_mark_host_registered_at_ (host, controller);
   controller->host = host;
   controller->opq_parallel_operations =
-      GNUNET_TESTBED_operation_queue_create_ ((unsigned int)
-                                              max_parallel_operations);
+      GNUNET_TESTBED_operation_queue_create_ (OPERATION_QUEUE_TYPE_FIXED,
+                                              (unsigned int) max_parallel_operations);
   controller->opq_parallel_service_connections =
-      GNUNET_TESTBED_operation_queue_create_ ((unsigned int)
+      GNUNET_TESTBED_operation_queue_create_ (OPERATION_QUEUE_TYPE_FIXED,
+                                              (unsigned int)
                                               max_parallel_service_connections);
   controller->opq_parallel_topology_config_operations =
-      GNUNET_TESTBED_operation_queue_create_ ((unsigned int)
+      GNUNET_TESTBED_operation_queue_create_ (OPERATION_QUEUE_TYPE_FIXED,
+                                              (unsigned int)
                                               max_parallel_topology_config_operations);
   controller_hostname = GNUNET_TESTBED_host_get_hostname (host);
   if (NULL == controller_hostname)
@@ -1856,13 +1858,25 @@ GNUNET_TESTBED_create_helper_init_msg_ (const char *trusted_ip,
 
 
 /**
- * Signal that the information from an operation has been fully
- * processed.  This function MUST be called for each event
- * of type 'operation_finished' to fully remove the operation
- * from the operation queue.  After calling this function, the
- * 'op_result' becomes invalid (!).
+ * This function is used to signal that the event information (struct
+ * GNUNET_TESTBED_EventInformation) from an operation has been fully processed
+ * i.e. if the event callback is ever called for this operation. If the event
+ * callback for this operation has not yet been called, calling this function
+ * cancels the operation, frees its resources and ensures the no event is
+ * generated with respect to this operation. Note that however cancelling an
+ * operation does NOT guarantee that the operation will be fully undone (or that
+ * nothing ever happened). 
  *
- * @param operation operation to signal completion for
+ * This function MUST be called for every operation to fully remove the
+ * operation from the operation queue.  After calling this function, if
+ * operation is completed and its event information is of type
+ * GNUNET_TESTBED_ET_OPERATION_FINISHED, the 'op_result' becomes invalid (!).
+
+ * If the operation is generated from GNUNET_TESTBED_service_connect() then
+ * calling this function on such as operation calls the disconnect adapter if
+ * the connect adapter was ever called.
+ *
+ * @param operation operation to signal completion or cancellation
  */
 void
 GNUNET_TESTBED_operation_done (struct GNUNET_TESTBED_Operation *operation)
diff --git a/src/testbed/testbed_api_hosts.c b/src/testbed/testbed_api_hosts.c
index cd805dca1..b7c824bc2 100644
--- a/src/testbed/testbed_api_hosts.c
+++ b/src/testbed/testbed_api_hosts.c
@@ -35,7 +35,6 @@
 #include "testbed_api_hosts.h"
 #include "testbed_helper.h"
 #include "testbed_api_operations.h"
-#include "testbed_api_sd.h"
 
 #include <zlib.h>
 
@@ -97,28 +96,6 @@ struct RegisteredController
 
 
 /**
- * A slot to record time taken by an overlay connect operation
- */
-struct TimeSlot
-{
-  /**
-   * A key to identify this timeslot
-   */
-  void *key;
-
-  /**
-   * Time
-   */
-  struct GNUNET_TIME_Relative time;
-
-  /**
-   * Number of timing values accumulated
-   */
-  unsigned int nvals;
-};
-
-
-/**
  * Opaque handle to a host running experiments managed by the testing framework.
  * The master process must be able to SSH to this host without password (via
  * ssh-agent).
@@ -161,28 +138,6 @@ struct GNUNET_TESTBED_Host
   struct OperationQueue *opq_parallel_overlay_connect_operations;
 
   /**
-   * An array of timing slots; size should be equal to the current number of parallel
-   * overlay connects
-   */
-  struct TimeSlot *tslots;
-
-  /**
-   * Handle for SD calculations amount parallel overlay connect operation finish
-   * times
-   */
-  struct SDHandle *poc_sd;  
-
-  /**
-   * The number of parallel overlay connects we do currently
-   */
-  unsigned int num_parallel_connects;
-
-  /**
-   * Counter to indicate when all the available time slots are filled
-   */
-  unsigned int tslots_filled;
-
-  /**
    * Is a controller started on this host? FIXME: Is this needed?
    */
   int controller_started;
@@ -382,9 +337,8 @@ GNUNET_TESTBED_host_create_with_id (uint32_t id, const char *hostname,
   host->port = (0 == port) ? 22 : port;
   host->cfg = GNUNET_CONFIGURATION_dup (cfg);
   host->opq_parallel_overlay_connect_operations =
-      GNUNET_TESTBED_operation_queue_create_ (0);
-  GNUNET_TESTBED_set_num_parallel_overlay_connects_ (host, 1);
-  host->poc_sd = GNUNET_TESTBED_SD_init_ (10);
+      GNUNET_TESTBED_operation_queue_create_ (OPERATION_QUEUE_TYPE_ADAPTIVE, 
+                                              UINT_MAX);
   new_size = host_list_size;
   while (id >= new_size)
     new_size += HOST_LIST_GROW_STEP;
@@ -740,8 +694,6 @@ GNUNET_TESTBED_host_destroy (struct GNUNET_TESTBED_Host *host)
   GNUNET_free_non_null ((char *) host->hostname);
   GNUNET_TESTBED_operation_queue_destroy_
       (host->opq_parallel_overlay_connect_operations);
-  GNUNET_TESTBED_SD_destroy_ (host->poc_sd);
-  GNUNET_free_non_null (host->tslots);
   GNUNET_CONFIGURATION_destroy (host->cfg);
   GNUNET_free (host);
   while (host_list_size >= HOST_LIST_GROW_STEP)
@@ -1624,192 +1576,6 @@ GNUNET_TESTBED_cancel_registration (struct GNUNET_TESTBED_HostRegistrationHandle
 
 
 /**
- * Initializes the operation queue for parallel overlay connects
- *
- * @param h the host handle
- * @param npoc the number of parallel overlay connects - the queue size
- */
-void
-GNUNET_TESTBED_set_num_parallel_overlay_connects_ (struct
-                                                   GNUNET_TESTBED_Host *h,
-                                                   unsigned int npoc)
-{
-  //fprintf (stderr, "%d", npoc);
-  GNUNET_free_non_null (h->tslots);
-  h->tslots_filled = 0;
-  h->num_parallel_connects = npoc;
-  h->tslots = GNUNET_malloc (npoc * sizeof (struct TimeSlot));
-  GNUNET_TESTBED_operation_queue_reset_max_active_
-      (h->opq_parallel_overlay_connect_operations, npoc);
-}
-
-
-/**
- * Returns a timing slot which will be exclusively locked
- *
- * @param h the host handle
- * @param key a pointer which is associated to the returned slot; should not be
- *          NULL. It serves as a key to determine the correct owner of the slot
- * @return the time slot index in the array of time slots in the controller
- *           handle
- */
-unsigned int
-GNUNET_TESTBED_get_tslot_ (struct GNUNET_TESTBED_Host *h, void *key)
-{
-  unsigned int slot;
-
-  GNUNET_assert (NULL != h->tslots);
-  GNUNET_assert (NULL != key);
-  for (slot = 0; slot < h->num_parallel_connects; slot++)
-    if (NULL == h->tslots[slot].key)
-    {
-      h->tslots[slot].key = key;
-      return slot;
-    }
-  GNUNET_assert (0);            /* We should always find a free tslot */
-}
-
-
-/**
- * Decides whether any change in the number of parallel overlay connects is
- * necessary to adapt to the load on the system
- *
- * @param h the host handle
- */
-static void
-decide_npoc (struct GNUNET_TESTBED_Host *h)
-{
-  struct GNUNET_TIME_Relative avg;
-  int sd;
-  unsigned int slot;
-  unsigned int nvals;
-
-  if (h->tslots_filled != h->num_parallel_connects)
-    return;
-  avg = GNUNET_TIME_UNIT_ZERO;
-  nvals = 0;
-  for (slot = 0; slot < h->num_parallel_connects; slot++)
-  {
-    avg = GNUNET_TIME_relative_add (avg, h->tslots[slot].time);
-    nvals += h->tslots[slot].nvals;
-  }
-  GNUNET_assert (nvals >= h->num_parallel_connects);
-  avg = GNUNET_TIME_relative_divide (avg, nvals);
-  GNUNET_assert (GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us != avg.rel_value_us);
-  sd = GNUNET_TESTBED_SD_deviation_factor_ (h->poc_sd, (unsigned int) avg.rel_value_us);
-  if ( (sd <= 5) ||
-       (0 == GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
-                                       h->num_parallel_connects)) )
-    GNUNET_TESTBED_SD_add_data_ (h->poc_sd, (unsigned int) avg.rel_value_us);
-  if (GNUNET_SYSERR == sd)
-  {
-    GNUNET_TESTBED_set_num_parallel_overlay_connects_ (h,
-                                                       h->num_parallel_connects);
-    return;
-  }
-  GNUNET_assert (0 <= sd);
-  if (0 == sd)
-  {
-    GNUNET_TESTBED_set_num_parallel_overlay_connects_ (h,
-                                                       h->num_parallel_connects
-                                                       * 2);
-    return;
-  }
-  if (1 == sd)
-  {
-    GNUNET_TESTBED_set_num_parallel_overlay_connects_ (h,
-                                                       h->num_parallel_connects
-                                                       + 1);
-    return;
-  }
-  if (1 == h->num_parallel_connects)
-  {
-    GNUNET_TESTBED_set_num_parallel_overlay_connects_ (h, 1);
-    return;
-  }
-  if (2 == sd)
-  {
-    GNUNET_TESTBED_set_num_parallel_overlay_connects_ (h,
-                                                       h->num_parallel_connects
-                                                       - 1);
-    return;
-  }
-  GNUNET_TESTBED_set_num_parallel_overlay_connects_ (h,
-                                                     h->num_parallel_connects /
-                                                     2);
-}
-
-
-/**
- * Releases a time slot thus making it available for be used again
- *
- * @param h the host handle
- * @param index the index of the the time slot
- * @param key the key to prove ownership of the timeslot
- * @return GNUNET_YES if the time slot is successfully removed; GNUNET_NO if the
- *           time slot cannot be removed - this could be because of the index
- *           greater than existing number of time slots or `key' being different
- */
-int
-GNUNET_TESTBED_release_time_slot_ (struct GNUNET_TESTBED_Host *h,
-                                   unsigned int index, void *key)
-{
-  struct TimeSlot *slot;
-
-  GNUNET_assert (NULL != key);
-  if (index >= h->num_parallel_connects)
-    return GNUNET_NO;
-  slot = &h->tslots[index];
-  if (key != slot->key)
-    return GNUNET_NO;
-  slot->key = NULL;
-  return GNUNET_YES;
-}
-
-
-/**
- * Function to update a time slot
- *
- * @param h the host handle
- * @param index the index of the time slot to update
- * @param key the key to identify ownership of the slot
- * @param time the new time
- * @param failed should this reading be treated as coming from a fail event
- */
-void
-GNUNET_TESTBED_update_time_slot_ (struct GNUNET_TESTBED_Host *h,
-                                  unsigned int index, void *key,
-                                  struct GNUNET_TIME_Relative time, int failed)
-{
-  struct TimeSlot *slot;
-
-  if (GNUNET_YES == failed)
-  {
-    if (1 == h->num_parallel_connects)
-    {
-      GNUNET_TESTBED_set_num_parallel_overlay_connects_ (h, 1);
-      return;
-    }
-    GNUNET_TESTBED_set_num_parallel_overlay_connects_ (h,
-                                                       h->num_parallel_connects
-                                                       - 1);
-  }
-  if (GNUNET_NO == GNUNET_TESTBED_release_time_slot_ (h, index, key))
-    return;
-  slot = &h->tslots[index];
-  slot->nvals++;
-  if (GNUNET_TIME_UNIT_ZERO.rel_value_us == slot->time.rel_value_us)
-  {
-    slot->time = time;
-    h->tslots_filled++;
-    decide_npoc (h);
-    return;
-  }
-  slot->time = GNUNET_TIME_relative_add (slot->time, time);
-}
-
-
-/**
  * Queues the given operation in the queue for parallel overlay connects of the
  * given host
  *
diff --git a/src/testbed/testbed_api_hosts.h b/src/testbed/testbed_api_hosts.h
index 6c305c340..a1cbb5b88 100644
--- a/src/testbed/testbed_api_hosts.h
+++ b/src/testbed/testbed_api_hosts.h
@@ -151,61 +151,6 @@ GNUNET_TESTBED_is_host_registered_ (const struct GNUNET_TESTBED_Host *host,
 
 
 /**
- * (re)sets the operation queue for parallel overlay connects
- *
- * @param h the host handle
- * @param npoc the number of parallel overlay connects - the queue size
- */
-void
-GNUNET_TESTBED_set_num_parallel_overlay_connects_ (struct
-                                                   GNUNET_TESTBED_Host *h,
-                                                   unsigned int npoc);
-
-
-/**
- * Releases a time slot thus making it available for be used again
- *
- * @param h the host handle
- * @param index the index of the the time slot
- * @param key the key to prove ownership of the timeslot
- * @return GNUNET_YES if the time slot is successfully removed; GNUNET_NO if the
- *           time slot cannot be removed - this could be because of the index
- *           greater than existing number of time slots or `key' being different
- */
-int
-GNUNET_TESTBED_release_time_slot_ (struct GNUNET_TESTBED_Host *h,
-                                   unsigned int index, void *key);
-
-
-/**
- * Function to update a time slot
- *
- * @param h the host handle
- * @param index the index of the time slot to update
- * @param key the key to identify ownership of the slot
- * @param time the new time
- * @param failed should this reading be treated as coming from a fail event
- */
-void
-GNUNET_TESTBED_update_time_slot_ (struct GNUNET_TESTBED_Host *h,
-                                  unsigned int index, void *key,
-                                  struct GNUNET_TIME_Relative time, int failed);
-
-
-/**
- * Returns a timing slot which will be exclusively locked
- *
- * @param h the host handle
- * @param key a pointer which is associated to the returned slot; should not be
- *          NULL. It serves as a key to determine the correct owner of the slot
- * @return the time slot index in the array of time slots in the controller
- *           handle
- */
-unsigned int
-GNUNET_TESTBED_get_tslot_ (struct GNUNET_TESTBED_Host *h, void *key);
-
-
-/**
  * Queues the given operation in the queue for parallel overlay connects of the
  * given host
  *
diff --git a/src/testbed/testbed_api_operations.c b/src/testbed/testbed_api_operations.c
index 521645b71..e1034a18c 100644
--- a/src/testbed/testbed_api_operations.c
+++ b/src/testbed/testbed_api_operations.c
@@ -27,6 +27,7 @@
 
 #include "platform.h"
 #include "testbed_api_operations.h"
+#include "testbed_api_sd.h"
 
 
 /**
@@ -60,6 +61,89 @@ struct QueueEntry
  * Queue of operations where we can only support a certain
  * number of concurrent operations of a particular type.
  */
+struct OperationQueue;
+
+
+/**
+ * A slot to record time taken by an operation
+ */
+struct TimeSlot
+{
+  /**
+   * DLL next pointer
+   */
+  struct TimeSlot *next;
+
+  /**
+   * DLL prev pointer
+   */
+  struct TimeSlot *prev;
+
+  /**
+   * This operation queue to which this time slot belongs to
+   */
+  struct OperationQueue *queue;
+
+  /**
+   * The operation to which this timeslot is currently allocated to
+   */
+  struct GNUNET_TESTBED_Operation *op;
+
+  /**
+   * Accumulated time
+   */
+  struct GNUNET_TIME_Relative tsum;
+
+  /**
+   * Number of timing values accumulated
+   */
+  unsigned int nvals;
+};
+
+
+/**
+ * Context for operation queues of type OPERATION_QUEUE_TYPE_ADAPTIVE
+ */
+struct FeedbackCtx
+{
+  /**
+   * Handle for calculating standard deviation
+   */
+  struct SDHandle *sd;
+
+  /**
+   * Head for DLL of time slots which are free to be allocated to operations
+   */
+  struct TimeSlot *alloc_head;
+    
+  /**
+   * Tail for DLL of time slots which are free to be allocated to operations
+   */
+  struct TimeSlot *alloc_tail;
+
+  /**
+   * Pointer to the chunk of time slots.  Free all time slots at a time using
+   * this pointer.
+   */
+  struct TimeSlot *tslots_freeptr;
+
+  /**
+   * Number of time slots filled so far
+   */
+  unsigned int tslots_filled;
+  
+  /**
+   * Bound on the maximum number of operations which can be active
+   */
+  unsigned int max_active_bound;
+
+};
+
+
+/**
+ * Queue of operations where we can only support a certain
+ * number of concurrent operations of a particular type.
+ */
 struct OperationQueue
 {
   /**
@@ -108,12 +192,26 @@ struct OperationQueue
   struct QueueEntry *nq_tail;
 
   /**
+   * Feedback context; only relevant for adaptive operation queues.  NULL for
+   * fixed operation queues
+   */
+  struct FeedbackCtx *fctx;
+
+  /**
+   * The type of this opeartion queue
+   */
+  enum OperationQueueType type;
+
+  /**
    * Number of operations that are currently active in this queue.
    */
   unsigned int active;
 
   /**
-   * Max number of operations which can be active at any time in this queue
+   * Max number of operations which can be active at any time in this queue.
+   * This value can be changed either by calling
+   * GNUNET_TESTBED_operation_queue_reset_max_active_() or by the adaptive
+   * algorithm if this operation queue is of type OPERATION_QUEUE_TYPE_ADAPTIVE
    */
   unsigned int max_active;
 
@@ -222,6 +320,21 @@ struct GNUNET_TESTBED_Operation
   struct ReadyQueueEntry *rq_entry;
 
   /**
+   * Head pointer for DLL of tslots allocated to this operation
+   */
+  struct TimeSlot *tslots_head;
+
+  /**
+   * Tail pointer for DLL of tslots allocated to this operation
+   */
+  struct TimeSlot *tslots_tail;
+
+  /**
+   * The time at which the operation is started
+   */
+  struct GNUNET_TIME_Absolute tstart;
+
+  /**
    * Number of queues in the operation queues array
    */
   unsigned int nqueues;
@@ -231,6 +344,11 @@ struct GNUNET_TESTBED_Operation
    */
   enum OperationState state;
 
+  /**
+   * Is this a failed operation?
+   */
+  int failed;
+
 };
 
 /**
@@ -250,6 +368,29 @@ GNUNET_SCHEDULER_TaskIdentifier process_rq_task_id;
 
 
 /**
+ * Assigns the given operation a time slot from the given operation queue
+ *
+ * @param op the operation
+ * @param queue the operation queue
+ * @return the timeslot
+ */
+static void
+assign_timeslot (struct GNUNET_TESTBED_Operation *op,
+                 struct OperationQueue *queue)
+{
+  struct FeedbackCtx *fctx = queue->fctx;
+  struct TimeSlot *tslot;
+
+  GNUNET_assert (OPERATION_QUEUE_TYPE_ADAPTIVE == queue->type);
+  tslot = fctx->alloc_head;
+  GNUNET_assert (NULL != tslot);
+  GNUNET_CONTAINER_DLL_remove (fctx->alloc_head, fctx->alloc_tail, tslot);
+  GNUNET_CONTAINER_DLL_insert_tail (op->tslots_head, op->tslots_tail, tslot);
+  tslot->op = op;
+}
+
+
+/**
  * Removes a queue entry of an operation from one of the operation queues' lists
  * depending on the state of the operation
  *
@@ -378,6 +519,8 @@ static void
 process_rq_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   struct GNUNET_TESTBED_Operation *op;
+  struct OperationQueue *queue;
+  unsigned int cnt;
 
   process_rq_task_id = GNUNET_SCHEDULER_NO_TASK;
   GNUNET_assert (NULL != rq_head);
@@ -386,8 +529,15 @@ process_rq_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
   if (NULL != rq_head)
     process_rq_task_id = GNUNET_SCHEDULER_add_now (&process_rq_task, NULL);
   change_state (op, OP_STATE_ACTIVE);
+  for (cnt = 0; cnt < op->nqueues; cnt++)
+  {
+    queue = op->queues[cnt];
+    if (OPERATION_QUEUE_TYPE_ADAPTIVE == queue->type)
+      assign_timeslot (op, queue);
+  }
+  op->tstart = GNUNET_TIME_absolute_get ();
   if (NULL != op->start)
-    op->start (op->cb_cls);  
+    op->start (op->cb_cls);
 }
 
 
@@ -582,7 +732,7 @@ check_readiness (struct GNUNET_TESTBED_Operation *op)
   if (NULL != evict_ops)
   {
     for (i = 0; i < n_evict_ops; i++)
-      GNUNET_TESTBED_operation_release_ (evict_ops[i]);
+      GNUNET_TESTBED_operation_release_ (evict_ops[i]); 
     GNUNET_free (evict_ops);
     evict_ops = NULL;
     /* Evicting the operations should schedule this operation */
@@ -619,6 +769,162 @@ defer (struct GNUNET_TESTBED_Operation *op)
 
 
 /**
+ * Cleanups the array of timeslots of an operation queue.  For each time slot in
+ * the array, if it is allocated to an operation, it will be deallocated from
+ * the operation
+ *
+ * @param queue the operation queue
+ */
+static void
+cleanup_tslots (struct OperationQueue *queue)
+{
+  struct FeedbackCtx *fctx = queue->fctx;
+  struct TimeSlot *tslot;
+  struct GNUNET_TESTBED_Operation *op;
+  unsigned int cnt;
+
+  GNUNET_assert (NULL != fctx);
+  for (cnt = 0; cnt < queue->max_active; cnt++)
+  {
+    tslot = &fctx->tslots_freeptr[cnt];
+    op = tslot->op;
+    if (NULL == op)
+      continue;
+    GNUNET_CONTAINER_DLL_remove (op->tslots_head, op->tslots_tail, tslot);
+  }
+  GNUNET_free_non_null (fctx->tslots_freeptr);
+  fctx->tslots_freeptr = NULL;
+  fctx->alloc_head = NULL;
+  fctx->alloc_tail = NULL;
+  fctx->tslots_filled = 0;
+}
+
+
+/**
+ * Initializes the operation queue for parallel overlay connects
+ *
+ * @param h the host handle
+ * @param npoc the number of parallel overlay connects - the queue size
+ */
+static void
+adaptive_queue_set_max_active (struct OperationQueue *queue, unsigned int n)
+{
+  struct FeedbackCtx *fctx = queue->fctx;
+  struct TimeSlot *tslot;
+  unsigned int cnt;
+  
+  cleanup_tslots (queue);
+  n = GNUNET_MIN (n ,fctx->max_active_bound);
+  fctx->tslots_freeptr = GNUNET_malloc (n * sizeof (struct TimeSlot));
+  for (cnt = 0; cnt < n; cnt++)
+  {
+    tslot = &fctx->tslots_freeptr[cnt];
+    tslot->queue = queue;
+    GNUNET_CONTAINER_DLL_insert_tail (fctx->alloc_head, fctx->alloc_tail, tslot);
+  }
+  GNUNET_TESTBED_operation_queue_reset_max_active_ (queue, n);
+}
+
+
+/**
+ * Adapts parallelism in an adaptive queue by using the statistical data from
+ * the feedback context.
+ *
+ * @param queue the queue
+ * @param fail GNUNET_YES if the last operation failed; GNUNET_NO if not;
+ */
+static void
+adapt_parallelism (struct OperationQueue *queue, int fail)
+{
+  struct GNUNET_TIME_Relative avg;
+  struct FeedbackCtx *fctx;
+  struct TimeSlot *tslot;
+  int sd;
+  unsigned int nvals;
+  unsigned int cnt;
+
+  avg = GNUNET_TIME_UNIT_ZERO;
+  nvals = 0;
+  fctx = queue->fctx;
+  for (cnt = 0; cnt < queue->max_active; cnt++)
+  {
+    tslot = &fctx->tslots_freeptr[cnt];
+    avg = GNUNET_TIME_relative_add (avg, tslot->tsum);
+    nvals += tslot->nvals;
+  }
+  GNUNET_assert (nvals >= queue->max_active);
+  avg = GNUNET_TIME_relative_divide (avg, nvals);
+  sd = GNUNET_TESTBED_SD_deviation_factor_ (fctx->sd, (unsigned int)
+                                            avg.rel_value_us);
+  if ( (sd <= 5) ||
+       (0 == GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
+                                       queue->max_active)) )
+    GNUNET_TESTBED_SD_add_data_ (fctx->sd, (unsigned int) avg.rel_value_us);
+  if (GNUNET_SYSERR == sd)
+  {
+    adaptive_queue_set_max_active (queue, queue->max_active); /* no change */
+    return;
+  }
+  GNUNET_assert (0 <= sd);
+  if ((0 == sd) && (! fail))
+  {
+    adaptive_queue_set_max_active (queue, queue->max_active * 2);
+    return;
+  }
+  if ((1 == sd) && (! fail))
+  {
+    adaptive_queue_set_max_active (queue, queue->max_active + 1);
+    return;
+  }
+  if (1 == queue->max_active)
+  {
+    adaptive_queue_set_max_active (queue, 1);
+    return;
+  }
+  if (((sd < 2) && (fail)) || (2 == sd))
+  {
+    adaptive_queue_set_max_active (queue, queue->max_active - 1);
+    return;
+  }
+  adaptive_queue_set_max_active (queue, queue->max_active / 2);
+}
+
+
+/**
+ * update tslots with the operation's completion time.  Additionally, if
+ * updating a timeslot makes all timeslots filled in an adaptive operation
+ * queue, call adapt_parallelism() for that queue.
+ *
+ * @param op the operation
+ */
+static void
+update_tslots (struct GNUNET_TESTBED_Operation *op)
+{
+  struct OperationQueue *queue;
+  struct GNUNET_TIME_Relative t;
+  struct TimeSlot *tslot;
+  struct FeedbackCtx *fctx;
+  
+  t = GNUNET_TIME_absolute_get_duration (op->tstart);
+  while (NULL != (tslot = op->tslots_head)) /* update time slots */
+  {
+    queue = tslot->queue;
+    fctx = queue->fctx;
+    tslot->tsum = GNUNET_TIME_relative_add (tslot->tsum, t);
+    GNUNET_CONTAINER_DLL_remove (op->tslots_head, op->tslots_tail, tslot);
+    tslot->op = NULL;    
+    GNUNET_CONTAINER_DLL_insert_tail (fctx->alloc_head, fctx->alloc_tail,
+  tslot);
+    if (0 != tslot->nvals++)
+      continue;
+    fctx->tslots_filled++;
+    if (queue->max_active == fctx->tslots_filled)
+      adapt_parallelism (queue, op->failed);
+  }
+}
+
+
+/**
  * Create an 'operation' to be performed.
  *
  * @param cls closure for the callbacks
@@ -644,17 +950,32 @@ GNUNET_TESTBED_operation_create_ (void *cls, OperationStart start,
 /**
  * Create an operation queue.
  *
+ * @param type the type of operation queue
  * @param max_active maximum number of operations in this
  *        queue that can be active in parallel at the same time
  * @return handle to the queue
  */
 struct OperationQueue *
-GNUNET_TESTBED_operation_queue_create_ (unsigned int max_active)
+GNUNET_TESTBED_operation_queue_create_ (enum OperationQueueType type,
+                                        unsigned int max_active)
 {
   struct OperationQueue *queue;
+  struct FeedbackCtx *fctx;
 
   queue = GNUNET_malloc (sizeof (struct OperationQueue));
-  queue->max_active = max_active;
+  queue->type = type;
+  if (OPERATION_QUEUE_TYPE_FIXED == type)
+  {
+    queue->max_active = max_active;
+  }
+  else
+  {
+    fctx = GNUNET_malloc (sizeof (struct FeedbackCtx));
+    fctx->max_active_bound = max_active;
+    fctx->sd = GNUNET_TESTBED_SD_init_ (10); /* FIXME: Why 10? */
+    queue->fctx = fctx;
+    adaptive_queue_set_max_active (queue, 1); /* start with 1 */
+  }
   return queue;
 }
 
@@ -668,7 +989,16 @@ GNUNET_TESTBED_operation_queue_create_ (unsigned int max_active)
 void
 GNUNET_TESTBED_operation_queue_destroy_ (struct OperationQueue *queue)
 {
+  struct FeedbackCtx *fctx;
+  
   GNUNET_break (GNUNET_YES == is_queue_empty (queue));
+  if (OPERATION_QUEUE_TYPE_ADAPTIVE == queue->type)
+  {
+    cleanup_tslots (queue);
+    fctx = queue->fctx;
+    GNUNET_TESTBED_SD_destroy_ (fctx->sd);
+    GNUNET_free (fctx);
+  }
   GNUNET_free (queue);
 }
 
@@ -867,8 +1197,10 @@ GNUNET_TESTBED_operation_release_ (struct GNUNET_TESTBED_Operation *op)
     rq_remove (op);
   if (OP_STATE_INACTIVE == op->state) /* Activate the operation if inactive */
     GNUNET_TESTBED_operation_activate_ (op);
+  if (OP_STATE_ACTIVE == op->state)
+    update_tslots (op);
   GNUNET_assert (NULL != op->queues);
-  GNUNET_assert (NULL != op->qentries);
+  GNUNET_assert (NULL != op->qentries);  
   for (i = 0; i < op->nqueues; i++)
   {
     entry = op->qentries[i];
@@ -882,8 +1214,8 @@ GNUNET_TESTBED_operation_release_ (struct GNUNET_TESTBED_Operation *op)
       break;
     case OP_STATE_WAITING:      
       break;
-    case OP_STATE_READY:
     case OP_STATE_ACTIVE:
+    case OP_STATE_READY:
       GNUNET_assert (0 != opq->active);
       GNUNET_assert (opq->active >= entry->nres);
       opq->active -= entry->nres;
@@ -901,4 +1233,16 @@ GNUNET_TESTBED_operation_release_ (struct GNUNET_TESTBED_Operation *op)
 }
 
 
+/**
+ * Marks an operation as failed
+ *
+ * @param op the operation to be marked as failed
+ */
+void
+GNUNET_TESTBED_operation_mark_failed (struct GNUNET_TESTBED_Operation *op)
+{
+  op->failed = GNUNET_YES;
+}
+
+
 /* end of testbed_api_operations.c */
diff --git a/src/testbed/testbed_api_operations.h b/src/testbed/testbed_api_operations.h
index b6885e69a..c44325307 100644
--- a/src/testbed/testbed_api_operations.h
+++ b/src/testbed/testbed_api_operations.h
@@ -38,14 +38,35 @@ struct OperationQueue;
 
 
 /**
+ * The type of operation queue
+ */
+enum OperationQueueType
+{
+  /**
+   * Operation queue which permits a fixed maximum number of operations to be
+   * active at any time
+   */
+  OPERATION_QUEUE_TYPE_FIXED,
+
+  /**
+   * Operation queue which adapts the number of operations to be active based on
+   * the operation completion times of previously executed operation in it
+   */
+  OPERATION_QUEUE_TYPE_ADAPTIVE
+};
+
+
+/**
  * Create an operation queue.
  *
- * @param max_active maximum number of operations in this
- *        queue that can be active in parallel at the same time
+ * @param type the type of operation queue
+ * @param max_active maximum number of operations in this queue that can be
+ *   active in parallel at the same time.
  * @return handle to the queue
  */
 struct OperationQueue *
-GNUNET_TESTBED_operation_queue_create_ (unsigned int max_active);
+GNUNET_TESTBED_operation_queue_create_ (enum OperationQueueType type,
+                                        unsigned int max_active);
 
 
 /**
@@ -199,5 +220,14 @@ void
 GNUNET_TESTBED_operation_activate_ (struct GNUNET_TESTBED_Operation *op);
 
 
+/**
+ * Marks an operation as failed
+ *
+ * @param op the operation to be marked as failed
+ */
+void
+GNUNET_TESTBED_operation_mark_failed (struct GNUNET_TESTBED_Operation *op);
+
+
 #endif
 /* end of testbed_api_operations.h */
diff --git a/src/testbed/testbed_api_peers.c b/src/testbed/testbed_api_peers.c
index 56e28a830..fd1da85ef 100644
--- a/src/testbed/testbed_api_peers.c
+++ b/src/testbed/testbed_api_peers.c
@@ -410,8 +410,6 @@ opstart_overlay_connect (void *cls)
   opc->state = OPC_STATE_STARTED;
   data = opc->data;
   GNUNET_assert (NULL != data);
-  data->tslot_index = GNUNET_TESTBED_get_tslot_ (data->p1->host, data);
-  data->tstart = GNUNET_TIME_absolute_get ();
   msg = GNUNET_malloc (sizeof (struct GNUNET_TESTBED_OverlayConnectMessage));
   msg->header.size =
       htons (sizeof (struct GNUNET_TESTBED_OverlayConnectMessage));
@@ -434,7 +432,6 @@ static void
 oprelease_overlay_connect (void *cls)
 {
   struct OperationContext *opc = cls;
-  struct GNUNET_TIME_Relative duration;
   struct OverlayConnectData *data;
 
   data = opc->data;
@@ -443,14 +440,10 @@ oprelease_overlay_connect (void *cls)
   case OPC_STATE_INIT:
     break;
   case OPC_STATE_STARTED:
-    (void) GNUNET_TESTBED_release_time_slot_ (data->p1->host, data->tslot_index,
-                                              data);
     GNUNET_TESTBED_remove_opc_ (opc->c, opc);
     break;
   case OPC_STATE_FINISHED:
-    duration = GNUNET_TIME_absolute_get_duration (data->tstart);
-    GNUNET_TESTBED_update_time_slot_ (data->p1->host, data->tslot_index, data,
-                                      duration, data->failed);
+    break;
   }
   GNUNET_free (data);
   GNUNET_free (opc);
diff --git a/src/testbed/testbed_api_peers.h b/src/testbed/testbed_api_peers.h
index 50fb29bf6..0a87bf6ee 100644
--- a/src/testbed/testbed_api_peers.h
+++ b/src/testbed/testbed_api_peers.h
@@ -250,21 +250,6 @@ struct OverlayConnectData
    */
   struct OperationContext *sub_opc;
 
-  /**
-   * The starting time of this operation
-   */
-  struct GNUNET_TIME_Absolute tstart;
-
-  /**
-   * Has this operation failed
-   */
-  int failed;
-
-  /**
-   * The timing slot index for this operation
-   */
-  unsigned int tslot_index;
-
 };
 
 
diff --git a/src/testbed/testbed_api_statistics.c b/src/testbed/testbed_api_statistics.c
index 1e12a7c64..99a79b866 100644
--- a/src/testbed/testbed_api_statistics.c
+++ b/src/testbed/testbed_api_statistics.c
@@ -415,8 +415,8 @@ GNUNET_TESTBED_get_statistics (unsigned int num_peers,
   GNUNET_assert (NULL != proc);
   GNUNET_assert (NULL != cont);
   if (NULL == no_wait_queue)
-    no_wait_queue =
-        GNUNET_TESTBED_operation_queue_create_ (UINT_MAX);
+    no_wait_queue = GNUNET_TESTBED_operation_queue_create_ 
+        (OPERATION_QUEUE_TYPE_FIXED, UINT_MAX);
   sc = GNUNET_malloc (sizeof (struct GetStatsContext));
   sc->peers = peers;
   sc->subsystem = (NULL == subsystem) ? NULL : GNUNET_strdup (subsystem);