convert testbed_api to MQ (not perfect, but working)

6 files changed, 345 insertions, 426 deletions

diff --git a/src/testbed/gnunet-service-testbed_oc.c b/src/testbed/gnunet-service-testbed_oc.c
index 55dd5cad94..44f408764e 100644
--- a/src/testbed/gnunet-service-testbed_oc.c
+++ b/src/testbed/gnunet-service-testbed_oc.c
@@ -1716,12 +1716,6 @@ rocc_hello_sent_cb (void *cls)
                                   rocc);
     return;
   }
-  if (GNUNET_SCHEDULER_REASON_READ_READY != tc->reason)
-  {
-    GNUNET_break (0);
-    return;
-  }
-
   rocc->tcc.cgh_p2_ats =
     GST_connection_pool_get_handle (rocc->peer->id,
                                     rocc->peer->details.local.cfg,
@@ -1752,7 +1746,8 @@ attempt_connect_task (void *cls)
   rocc->ohh =
       GNUNET_TRANSPORT_offer_hello (rocc->tcc.th_,
                                     rocc->hello,
-                                    rocc_hello_sent_cb, rocc);
+                                    &rocc_hello_sent_cb,
+                                    rocc);
   if (NULL == rocc->ohh)
     rocc->attempt_connect_task_id =
         GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
diff --git a/src/testbed/test_testbed_api_controllerlink.c b/src/testbed/test_testbed_api_controllerlink.c
index fd876a07f2..699a182157 100644
--- a/src/testbed/test_testbed_api_controllerlink.c
+++ b/src/testbed/test_testbed_api_controllerlink.c
@@ -244,7 +244,7 @@ static struct GNUNET_CONFIGURATION_Handle *cfg3;
 /**
  * Abort task
  */
-static struct GNUNET_SCHEDULER_Task * abort_task;
+static struct GNUNET_SCHEDULER_Task *abort_task;
 
 /**
  * Operation handle for linking controllers
@@ -274,17 +274,17 @@ static struct GNUNET_TESTBED_Peer *master_peer;
 /**
  * The handle for whether a host is habitable or not
  */
-struct GNUNET_TESTBED_HostHabitableCheckHandle *hc_handle;
+static struct GNUNET_TESTBED_HostHabitableCheckHandle *hc_handle;
 
 /**
  * The task handle for the delay task
  */
-struct GNUNET_SCHEDULER_Task * delay_task_id;
+static struct GNUNET_SCHEDULER_Task *delay_task_id;
 
 /**
  * Event mask
  */
-uint64_t event_mask;
+static uint64_t event_mask;
 
 /**
  * Global testing status
@@ -300,7 +300,7 @@ static enum Stage result;
       if (NULL != abort_task)               \
         GNUNET_SCHEDULER_cancel (abort_task);                   \
       abort_task = NULL;                    \
-      GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);             \
+      GNUNET_SCHEDULER_shutdown ();              \
       return;                                                   \
     }                                                          \
   } while (0)
@@ -323,6 +323,11 @@ do_shutdown (void *cls)
   }
   if (NULL != hc_handle)
     GNUNET_TESTBED_is_host_habitable_cancel (hc_handle);
+  if (NULL != op)
+  {
+    GNUNET_TESTBED_operation_done (op);
+    op = NULL;
+  }
   if (NULL != mc)
     GNUNET_TESTBED_controller_disconnect (mc);
   if (NULL != cp)
@@ -352,9 +357,11 @@ do_shutdown (void *cls)
 static void
 do_abort (void *cls)
 {
-  LOG (GNUNET_ERROR_TYPE_WARNING, "Aborting\n");
+  LOG (GNUNET_ERROR_TYPE_WARNING,
+       "Aborting in stage %d\n",
+       result);
   abort_task = NULL;
-  do_shutdown (cls);
+  GNUNET_SCHEDULER_shutdown ();
 }
 
 
@@ -423,7 +430,9 @@ delay_task (void *cls)
  * @param emsg NULL if peer is not NULL; else MAY contain the error description
  */
 static void
-peer_create_cb (void *cls, struct GNUNET_TESTBED_Peer *peer, const char *emsg)
+peer_create_cb (void *cls,
+                struct GNUNET_TESTBED_Peer *peer,
+                const char *emsg)
 {
   FAIL_TEST (NULL != peer);
   FAIL_TEST (NULL == emsg);
@@ -445,12 +454,13 @@ peer_create_cb (void *cls, struct GNUNET_TESTBED_Peer *peer, const char *emsg)
     result = SLAVE2_PEER_CREATE_SUCCESS;
     slave2_peer = peer;
     GNUNET_TESTBED_operation_done (op);
+    op = NULL;
     delay_task_id =
         GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                       (GNUNET_TIME_UNIT_SECONDS, 1),
                                       &delay_task,
                                       NULL);
-    break;
+    return;
   case SLAVE3_STARTED:
     result = SLAVE3_PEER_CREATE_SUCCESS;
     slave3_peer = peer;
@@ -490,7 +500,8 @@ check_operation_success (const struct GNUNET_TESTBED_EventInformation *event)
  * @param event information about the event
  */
 static void
-controller_cb (void *cls, const struct GNUNET_TESTBED_EventInformation *event)
+controller_cb (void *cls,
+               const struct GNUNET_TESTBED_EventInformation *event)
 {
   switch (result)
   {
@@ -516,6 +527,7 @@ controller_cb (void *cls, const struct GNUNET_TESTBED_EventInformation *event)
     FAIL_TEST (event->details.peer_start.host == host);
     FAIL_TEST (event->details.peer_start.peer == master_peer);
     GNUNET_TESTBED_operation_done (op);
+    op = NULL;
     result = MASTER_PEER_START_SUCCESS;
     slave = GNUNET_TESTBED_host_create_with_id (1, "127.0.0.1", NULL, cfg, 0);
     FAIL_TEST (NULL != slave);
@@ -622,7 +634,6 @@ controller_cb (void *cls, const struct GNUNET_TESTBED_EventInformation *event)
   case SLAVE3_REGISTERED:
     check_operation_success (event);
     GNUNET_TESTBED_operation_done (op);
-    op = NULL;
     result = SLAVE3_STARTED;
     op = GNUNET_TESTBED_peer_create (mc, slave3, cfg, peer_create_cb, NULL);
     FAIL_TEST (NULL != op);
@@ -642,9 +653,7 @@ controller_cb (void *cls, const struct GNUNET_TESTBED_EventInformation *event)
     result = SUCCESS;
     GNUNET_TESTBED_operation_done (op);
     op = NULL;
-    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
-                                  &do_shutdown,
-                                  NULL);
+    GNUNET_SCHEDULER_shutdown ();
     break;
   default:
     FAIL_TEST (0);
@@ -740,10 +749,11 @@ status_cb (void *cls, const struct GNUNET_CONFIGURATION_Handle *config,
  * @param cls NULL
  * @param host the host whose status is being reported; will be NULL if the host
  *          given to GNUNET_TESTBED_is_host_habitable() is NULL
- * @param status GNUNET_YES if it is habitable; GNUNET_NO if not
+ * @param status #GNUNET_YES if it is habitable; #GNUNET_NO if not
  */
 static void
-host_habitable_cb (void *cls, const struct GNUNET_TESTBED_Host *_host,
+host_habitable_cb (void *cls,
+                   const struct GNUNET_TESTBED_Host *_host,
                    int status)
 {
   hc_handle = NULL;
@@ -755,7 +765,7 @@ host_habitable_cb (void *cls, const struct GNUNET_TESTBED_Host *_host,
                    "Skipping test\n");
     GNUNET_SCHEDULER_cancel (abort_task);
     abort_task = NULL;
-    (void) GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
+    GNUNET_SCHEDULER_shutdown ();
     result = SKIP;
     return;
   }
@@ -799,6 +809,8 @@ run (void *cls, char *const *args, const char *cfgfile,
       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                     (GNUNET_TIME_UNIT_MINUTES, 5), &do_abort,
                                     NULL);
+  GNUNET_SCHEDULER_add_shutdown (&do_shutdown,
+                                 NULL);
 }
 
 
diff --git a/src/testbed/testbed_api.c b/src/testbed/testbed_api.c
index 9b80664124..aad5055ef2 100644
--- a/src/testbed/testbed_api.c
+++ b/src/testbed/testbed_api.c
@@ -67,28 +67,6 @@
 
 
 /**
- * The message queue for sending messages to the controller service
- */
-struct MessageQueue
-{
-  /**
-   * The message to be sent
-   */
-  struct GNUNET_MessageHeader *msg;
-
-  /**
-   * next pointer for DLL
-   */
-  struct MessageQueue *next;
-
-  /**
-   * prev pointer for DLL
-   */
-  struct MessageQueue *prev;
-};
-
-
-/**
  * Context data for forwarded Operation
  */
 struct ForwardedOperationData
@@ -272,12 +250,13 @@ struct SearchContext
  * @param cls the serach context
  * @param key current key code
  * @param value value in the hash map
- * @return GNUNET_YES if we should continue to
- *         iterate,
- *         GNUNET_NO if not.
+ * @return #GNUNET_YES if we should continue to iterate,
+ *         #GNUNET_NO if not.
  */
 static int
-opc_search_iterator (void *cls, uint32_t key, void *value)
+opc_search_iterator (void *cls,
+                     uint32_t key,
+                     void *value)
 {
   struct SearchContext *sc = cls;
   struct OperationContext *opc = value;
@@ -358,6 +337,85 @@ GNUNET_TESTBED_remove_opc_ (const struct GNUNET_TESTBED_Controller *c,
 }
 
 
+
+/**
+ * Check #GNUNET_MESSAGE_TYPE_TESTBED_ADDHOSTCONFIRM message is well-formed.
+ *
+ * @param cls the controller handler
+ * @param msg message received
+ * @return #GNUNET_OK if message is well-formed
+ */
+static int
+check_add_host_confirm (void *cls,
+                        const struct GNUNET_TESTBED_HostConfirmedMessage *msg)
+{
+  const char *emsg;
+  uint16_t msg_size;
+
+  msg_size = ntohs (msg->header.size) - sizeof (*msg);
+  if (0 == msg_size)
+    return GNUNET_OK;
+  /* We have an error message */
+  emsg = (const char *) &msg[1];
+  if ('\0' != emsg[msg_size - 1])
+  {
+    GNUNET_break (0);
+    return GNUNET_SYSERR;
+  }
+  return GNUNET_OK;
+}
+
+
+/**
+ * Handler for #GNUNET_MESSAGE_TYPE_TESTBED_ADDHOSTCONFIRM message from
+ * controller (testbed service)
+ *
+ * @param cls the controller handler
+ * @param msg message received
+ */
+static void
+handle_add_host_confirm (void *cls,
+                         const struct GNUNET_TESTBED_HostConfirmedMessage *msg)
+{
+  struct GNUNET_TESTBED_Controller *c = cls;
+  struct GNUNET_TESTBED_HostRegistrationHandle *rh = c->rh;
+  const char *emsg;
+  uint16_t msg_size;
+
+  if (NULL == rh)
+    return;
+  if (GNUNET_TESTBED_host_get_id_ (rh->host) != ntohl (msg->host_id))
+  {
+    LOG_DEBUG ("Mismatch in host id's %u, %u of host confirm msg\n",
+               GNUNET_TESTBED_host_get_id_ (rh->host),
+               ntohl (msg->host_id));
+    return;
+  }
+  c->rh = NULL;
+  msg_size = ntohs (msg->header.size) - sizeof (*msg);
+  if (0 == msg_size)
+  {
+    LOG_DEBUG ("Host %u successfully registered\n",
+               ntohl (msg->host_id));
+    GNUNET_TESTBED_mark_host_registered_at_ (rh->host,
+                                             c);
+    rh->cc (rh->cc_cls,
+            NULL);
+    GNUNET_free (rh);
+    return;
+  }
+  /* We have an error message */
+  emsg = (const char *) &msg[1];
+  LOG (GNUNET_ERROR_TYPE_ERROR,
+       _("Adding host %u failed with error: %s\n"),
+       ntohl (msg->host_id),
+       emsg);
+  rh->cc (rh->cc_cls,
+          emsg);
+  GNUNET_free (rh);
+}
+
+
 /**
  * Handler for forwarded operations
  *
@@ -690,6 +748,22 @@ handle_peer_conevent (void *cls,
 
 
 /**
+ * Validate #GNUNET_MESSAGE_TYPE_TESTBED_PEER_INFORMATION message from
+ * controller (testbed service)
+ *
+ * @param c the controller handler
+ * @param msg message received
+ */
+static int
+check_peer_config (void *cls,
+                   const struct GNUNET_TESTBED_PeerConfigurationInformationMessage *msg)
+{
+  /* anything goes? */
+  return GNUNET_OK;
+}
+
+
+/**
  * Handler for #GNUNET_MESSAGE_TYPE_TESTBED_PEER_INFORMATION message from
  * controller (testbed service)
  *
@@ -698,8 +772,7 @@ handle_peer_conevent (void *cls,
  */
 static void
 handle_peer_config (void *cls,
-                    const struct
-                    GNUNET_TESTBED_PeerConfigurationInformationMessage *msg)
+                    const struct GNUNET_TESTBED_PeerConfigurationInformationMessage *msg)
 {
   struct GNUNET_TESTBED_Controller *c = cls;
   struct OperationContext *opc;
@@ -718,8 +791,9 @@ handle_peer_config (void *cls,
   }
   if (OP_FORWARDED == opc->type)
   {
-    handle_forwarded_operation_msg (c, opc,
-                                    (const struct GNUNET_MessageHeader *) msg);
+    handle_forwarded_operation_msg (c,
+                                    opc,
+                                    &msg->header);
     return;
   }
   data = opc->data;
@@ -761,6 +835,23 @@ handle_peer_config (void *cls,
 
 
 /**
+ * Validate #GNUNET_MESSAGE_TYPE_TESTBED_OPERATION_FAIL_EVENT message from
+ * controller (testbed service)
+ *
+ * @param c the controller handler
+ * @param msg message received
+ * @return #GNUNET_OK if message is well-formed
+ */
+static int
+check_op_fail_event (void *cls,
+                     const struct GNUNET_TESTBED_OperationFailureEventMessage *msg)
+{
+  /* we accept anything as a valid error message */
+  return GNUNET_OK;
+}
+
+
+/**
  * Handler for #GNUNET_MESSAGE_TYPE_TESTBED_OPERATION_FAIL_EVENT message from
  * controller (testbed service)
  *
@@ -917,6 +1008,23 @@ GNUNET_TESTBED_generate_slavegetconfig_msg_ (uint64_t op_id, uint32_t slave_id)
 }
 
 
+
+/**
+ * Validate #GNUNET_MESSAGE_TYPE_TESTBED_SLAVE_INFORMATION message from
+ * controller (testbed service)
+ *
+ * @param c the controller handler
+ * @param msg message received
+ */
+static int
+check_slave_config (void *cls,
+                    const struct GNUNET_TESTBED_SlaveConfiguration *msg)
+{
+  /* anything goes? */
+  return GNUNET_OK;
+}
+
+
 /**
  * Handler for #GNUNET_MESSAGE_TYPE_TESTBED_SLAVE_CONFIGURATION message from controller
  * (testbed service)
@@ -963,6 +1071,23 @@ handle_slave_config (void *cls,
 
 
 /**
+ * Check #GNUNET_MESSAGE_TYPE_TESTBED_LINK_CONTROLLERS_RESULT message from controller
+ * (testbed service)
+ *
+ * @param c the controller handler
+ * @param msg message received
+ * @return #GNUNET_OK if @a msg is well-formed
+ */
+static int
+check_link_controllers_result (void *cls,
+                                const struct GNUNET_TESTBED_ControllerLinkResponse *msg)
+{
+  /* actual check to be implemented */
+  return GNUNET_OK;
+}
+
+
+/**
  * Handler for #GNUNET_MESSAGE_TYPE_TESTBED_LINK_CONTROLLERS_RESULT message from controller
  * (testbed service)
  *
@@ -971,8 +1096,7 @@ handle_slave_config (void *cls,
  */
 static void
 handle_link_controllers_result (void *cls,
-                                const struct
-                                GNUNET_TESTBED_ControllerLinkResponse *msg)
+                                const struct GNUNET_TESTBED_ControllerLinkResponse *msg)
 {
   struct GNUNET_TESTBED_Controller *c = cls;
   struct OperationContext *opc;
@@ -1056,8 +1180,8 @@ handle_link_controllers_result (void *cls,
  *   down signalling an error (message malformed)
  */
 static int
-check_barrier_status_ (struct GNUNET_TESTBED_Controller *c,
-                       const struct GNUNET_TESTBED_BarrierStatusMsg *msg)
+check_barrier_status (void *cls,
+                      const struct GNUNET_TESTBED_BarrierStatusMsg *msg)
 {
   uint16_t msize;
   uint16_t name_len;
@@ -1097,14 +1221,15 @@ check_barrier_status_ (struct GNUNET_TESTBED_Controller *c,
 /**
  * Handler for #GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS messages
  *
- * @param c the controller handle to determine the connection this message
+ * @param cls the controller handle to determine the connection this message
  *   belongs to
  * @param msg the barrier status message
  */
 static void
-handle_barrier_status_ (struct GNUNET_TESTBED_Controller *c,
-                        const struct GNUNET_TESTBED_BarrierStatusMsg *msg)
+handle_barrier_status (void *cls,
+                       const struct GNUNET_TESTBED_BarrierStatusMsg *msg)
 {
+  struct GNUNET_TESTBED_Controller *c = cls;
   struct GNUNET_TESTBED_Barrier *barrier;
   char *emsg;
   const char *name;
@@ -1159,191 +1284,18 @@ handle_barrier_status_ (struct GNUNET_TESTBED_Controller *c,
 
 
 /**
- * Handler for messages from controller (testbed service)
- *
- * @param cls the controller handler
- * @param msg message received, NULL on timeout or fatal error
- */
-static void
-message_handler (void *cls, const struct GNUNET_MessageHeader *msg)
-{
-  struct GNUNET_TESTBED_Controller *c = cls;
-  int status;
-  uint16_t msize;
-
-  c->in_receive = GNUNET_NO;
-  /* FIXME: Add checks for message integrity */
-  if (NULL == msg)
-  {
-    LOG_DEBUG ("Receive timed out or connection to service dropped\n");
-    return;
-  }
-  msize = ntohs (msg->size);
-  switch (ntohs (msg->type))
-  {
-  case GNUNET_MESSAGE_TYPE_TESTBED_ADD_HOST_SUCCESS:
-    GNUNET_assert (msize >=
-                   sizeof (struct GNUNET_TESTBED_HostConfirmedMessage));
-    status =
-        GNUNET_TESTBED_host_handle_addhostconfirm_
-        (c, (const struct GNUNET_TESTBED_HostConfirmedMessage*) msg);
-    break;
-  case GNUNET_MESSAGE_TYPE_TESTBED_GENERIC_OPERATION_SUCCESS:
-    GNUNET_assert (msize ==
-                   sizeof (struct
-                           GNUNET_TESTBED_GenericOperationSuccessEventMessage));
-    handle_opsuccess (c,
-                      (const struct
-                       GNUNET_TESTBED_GenericOperationSuccessEventMessage *)
-                      msg);
-    status = GNUNET_YES;
-    break;
-  case GNUNET_MESSAGE_TYPE_TESTBED_OPERATION_FAIL_EVENT:
-    GNUNET_assert (msize >=
-                   sizeof (struct GNUNET_TESTBED_OperationFailureEventMessage));
-    handle_op_fail_event (c,
-                          (const struct
-                           GNUNET_TESTBED_OperationFailureEventMessage *)
-                          msg);
-    status = GNUNET_YES;
-    break;
-  case GNUNET_MESSAGE_TYPE_TESTBED_CREATE_PEER_SUCCESS:
-    GNUNET_assert (msize ==
-                   sizeof (struct
-                           GNUNET_TESTBED_PeerCreateSuccessEventMessage));
-    handle_peer_create_success (c,
-                                (const struct
-                                 GNUNET_TESTBED_PeerCreateSuccessEventMessage
-                                 *) msg);
-    status = GNUNET_YES;
-    break;
-  case GNUNET_MESSAGE_TYPE_TESTBED_PEER_EVENT:
-    GNUNET_assert (msize == sizeof (struct GNUNET_TESTBED_PeerEventMessage));
-    handle_peer_event (c,
-                       (const struct GNUNET_TESTBED_PeerEventMessage *)
-                       msg);
-
-    status = GNUNET_YES;
-    break;
-  case GNUNET_MESSAGE_TYPE_TESTBED_PEER_INFORMATION:
-    GNUNET_assert (msize >=
-                   sizeof (struct
-                           GNUNET_TESTBED_PeerConfigurationInformationMessage));
-    handle_peer_config (c,
-                        (const struct
-                         GNUNET_TESTBED_PeerConfigurationInformationMessage
-                         *) msg);
-    status = GNUNET_YES;
-
-    break;
-  case GNUNET_MESSAGE_TYPE_TESTBED_PEER_CONNECT_EVENT:
-    GNUNET_assert (msize ==
-                   sizeof (struct GNUNET_TESTBED_ConnectionEventMessage));
-    handle_peer_conevent (c,
-                          (const struct
-                           GNUNET_TESTBED_ConnectionEventMessage *) msg);
-    status = GNUNET_YES;
-    break;
-  case GNUNET_MESSAGE_TYPE_TESTBED_SLAVE_CONFIGURATION:
-    GNUNET_assert (msize > sizeof (struct GNUNET_TESTBED_SlaveConfiguration));
-    handle_slave_config (c,
-                         (const struct GNUNET_TESTBED_SlaveConfiguration *)
-                         msg);
-    status = GNUNET_YES;
-    break;
-  case GNUNET_MESSAGE_TYPE_TESTBED_LINK_CONTROLLERS_RESULT:
-    handle_link_controllers_result (c,
-                                    (const struct
-                                     GNUNET_TESTBED_ControllerLinkResponse
-                                     *) msg);
-    status = GNUNET_YES;
-    break;
-  case GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS:
-    status = check_barrier_status_ (c,
-                                    (const struct GNUNET_TESTBED_BarrierStatusMsg *) msg);
-    if (GNUNET_OK == status)
-      handle_barrier_status_ (c,
-                              (const struct
-                               GNUNET_TESTBED_BarrierStatusMsg *)
-                              msg);
-    break;
-  default:
-    GNUNET_assert (0);
-  }
-  if ((GNUNET_OK == status) && (GNUNET_NO == c->in_receive))
-  {
-    c->in_receive = GNUNET_YES;
-    GNUNET_CLIENT_receive (c->client, &message_handler, c,
-                           GNUNET_TIME_UNIT_FOREVER_REL);
-  }
-}
-
-
-/**
- * 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 closure
- * @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_notify (void *cls, size_t size, void *buf)
-{
-  struct GNUNET_TESTBED_Controller *c = cls;
-  struct MessageQueue *mq_entry;
-
-  c->th = NULL;
-  mq_entry = c->mq_head;
-  GNUNET_assert (NULL != mq_entry);
-  if ((0 == size) && (NULL == buf))     /* Timeout */
-  {
-    LOG_DEBUG ("Message sending timed out -- retrying\n");
-    c->th =
-        GNUNET_CLIENT_notify_transmit_ready (c->client,
-                                             ntohs (mq_entry->msg->size),
-                                             TIMEOUT_REL, GNUNET_YES,
-                                             &transmit_ready_notify, c);
-    return 0;
-  }
-  GNUNET_assert (ntohs (mq_entry->msg->size) <= size);
-  size = ntohs (mq_entry->msg->size);
-  memcpy (buf, mq_entry->msg, size);
-  LOG_DEBUG ("Message of type: %u and size: %u sent\n",
-             ntohs (mq_entry->msg->type), size);
-  GNUNET_free (mq_entry->msg);
-  GNUNET_CONTAINER_DLL_remove (c->mq_head, c->mq_tail, mq_entry);
-  GNUNET_free (mq_entry);
-  mq_entry = c->mq_head;
-  if (NULL != mq_entry)
-    c->th =
-        GNUNET_CLIENT_notify_transmit_ready (c->client,
-                                             ntohs (mq_entry->msg->size),
-                                             TIMEOUT_REL, GNUNET_YES,
-                                             &transmit_ready_notify, c);
-  if (GNUNET_NO == c->in_receive)
-  {
-    c->in_receive = GNUNET_YES;
-    GNUNET_CLIENT_receive (c->client, &message_handler, c,
-                           GNUNET_TIME_UNIT_FOREVER_REL);
-  }
-  return size;
-}
-
-
-/**
  * Queues a message in send queue for sending to the service
  *
  * @param controller the handle to the controller
  * @param msg the message to queue
+ * @deprecated
  */
 void
 GNUNET_TESTBED_queue_message_ (struct GNUNET_TESTBED_Controller *controller,
                                struct GNUNET_MessageHeader *msg)
 {
-  struct MessageQueue *mq_entry;
+  struct GNUNET_MQ_Envelope *env;
+  struct GNUNET_MessageHeader *m2;
   uint16_t type;
   uint16_t size;
 
@@ -1351,19 +1303,13 @@ GNUNET_TESTBED_queue_message_ (struct GNUNET_TESTBED_Controller *controller,
   size = ntohs (msg->size);
   GNUNET_assert ((GNUNET_MESSAGE_TYPE_TESTBED_INIT <= type) &&
                  (GNUNET_MESSAGE_TYPE_TESTBED_MAX > type));
-  mq_entry = GNUNET_new (struct MessageQueue);
-  mq_entry->msg = msg;
-  LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Queueing message of type %u, size %u for sending\n", type,
-       ntohs (msg->size));
-  GNUNET_CONTAINER_DLL_insert_tail (controller->mq_head, controller->mq_tail,
-                                    mq_entry);
-  if (NULL == controller->th)
-    controller->th =
-        GNUNET_CLIENT_notify_transmit_ready (controller->client, size,
-                                             TIMEOUT_REL, GNUNET_YES,
-                                             &transmit_ready_notify,
-                                             controller);
+  env = GNUNET_MQ_msg_extra (m2,
+                             size - sizeof (*m2),
+                             type);
+  memcpy (m2, msg, size);
+  GNUNET_free (msg);
+  GNUNET_MQ_send (controller->mq,
+                  env);
 }
 
 
@@ -1527,6 +1473,28 @@ oprelease_get_slave_config (void *cls)
 
 
 /**
+ * Generic error handler, called with the appropriate error code and
+ * the same closure specified at the creation of the message queue.
+ * Not every message queue implementation supports an error handler.
+ *
+ * @param cls closure, a `struct GNUNET_TESTBED_Controller *`
+ * @param error error code
+ */
+static void
+mq_error_handler (void *cls,
+                  enum GNUNET_MQ_Error error)
+{
+  /* struct GNUNET_TESTBED_Controller *c = cls; */
+
+  GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+              "Encountered MQ error: %d\n",
+              error);
+  /* now what? */
+  GNUNET_SCHEDULER_shutdown (); /* seems most reasonable */
+}
+
+
+/**
  * Start a controller process using the given configuration at the
  * given host.
  *
@@ -1547,7 +1515,51 @@ GNUNET_TESTBED_controller_connect (struct GNUNET_TESTBED_Host *host,
                                    GNUNET_TESTBED_ControllerCallback cc,
                                    void *cc_cls)
 {
-  struct GNUNET_TESTBED_Controller *controller;
+  GNUNET_MQ_hd_var_size (add_host_confirm,
+                         GNUNET_MESSAGE_TYPE_TESTBED_ADD_HOST_SUCCESS,
+                         struct GNUNET_TESTBED_HostConfirmedMessage);
+  GNUNET_MQ_hd_fixed_size (peer_conevent,
+                           GNUNET_MESSAGE_TYPE_TESTBED_PEER_CONNECT_EVENT,
+                           struct GNUNET_TESTBED_ConnectionEventMessage);
+  GNUNET_MQ_hd_fixed_size (opsuccess,
+                           GNUNET_MESSAGE_TYPE_TESTBED_GENERIC_OPERATION_SUCCESS,
+                           struct GNUNET_TESTBED_GenericOperationSuccessEventMessage);
+  GNUNET_MQ_hd_var_size (op_fail_event,
+                         GNUNET_MESSAGE_TYPE_TESTBED_OPERATION_FAIL_EVENT,
+                         struct GNUNET_TESTBED_OperationFailureEventMessage);
+  GNUNET_MQ_hd_fixed_size (peer_create_success,
+                           GNUNET_MESSAGE_TYPE_TESTBED_CREATE_PEER_SUCCESS,
+                           struct GNUNET_TESTBED_PeerCreateSuccessEventMessage);
+  GNUNET_MQ_hd_fixed_size (peer_event,
+                           GNUNET_MESSAGE_TYPE_TESTBED_PEER_EVENT,
+                           struct GNUNET_TESTBED_PeerEventMessage);
+  GNUNET_MQ_hd_var_size (peer_config,
+                         GNUNET_MESSAGE_TYPE_TESTBED_PEER_INFORMATION,
+                         struct GNUNET_TESTBED_PeerConfigurationInformationMessage);
+  GNUNET_MQ_hd_var_size (slave_config,
+                         GNUNET_MESSAGE_TYPE_TESTBED_SLAVE_CONFIGURATION,
+                         struct GNUNET_TESTBED_SlaveConfiguration);
+  GNUNET_MQ_hd_var_size (link_controllers_result,
+                         GNUNET_MESSAGE_TYPE_TESTBED_LINK_CONTROLLERS_RESULT,
+                         struct GNUNET_TESTBED_ControllerLinkResponse);
+  GNUNET_MQ_hd_var_size (barrier_status,
+                         GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS,
+                         const struct GNUNET_TESTBED_BarrierStatusMsg);
+   struct GNUNET_TESTBED_Controller *controller
+     = GNUNET_new (struct GNUNET_TESTBED_Controller);
+   struct GNUNET_MQ_MessageHandler handlers[] = {
+     make_add_host_confirm_handler (controller),
+     make_peer_conevent_handler (controller),
+     make_opsuccess_handler (controller),
+     make_op_fail_event_handler (controller),
+     make_peer_create_success_handler (controller),
+     make_peer_event_handler (controller),
+     make_peer_config_handler (controller),
+     make_slave_config_handler (controller),
+     make_link_controllers_result_handler (controller),
+     make_barrier_status_handler (controller),
+     GNUNET_MQ_handler_end ()
+  };
   struct GNUNET_TESTBED_InitMessage *msg;
   const struct GNUNET_CONFIGURATION_Handle *cfg;
   const char *controller_hostname;
@@ -1562,6 +1574,7 @@ GNUNET_TESTBED_controller_connect (struct GNUNET_TESTBED_Host *host,
                                              &max_parallel_operations))
   {
     GNUNET_break (0);
+    GNUNET_free (controller);
     return NULL;
   }
   if (GNUNET_OK !=
@@ -1570,6 +1583,7 @@ GNUNET_TESTBED_controller_connect (struct GNUNET_TESTBED_Host *host,
                                              &max_parallel_service_connections))
   {
     GNUNET_break (0);
+    GNUNET_free (controller);
     return NULL;
   }
   if (GNUNET_OK !=
@@ -1578,16 +1592,21 @@ GNUNET_TESTBED_controller_connect (struct GNUNET_TESTBED_Host *host,
                                              &max_parallel_topology_config_operations))
   {
     GNUNET_break (0);
+    GNUNET_free (controller);
     return NULL;
   }
-  controller = GNUNET_new (struct GNUNET_TESTBED_Controller);
   controller->cc = cc;
   controller->cc_cls = cc_cls;
   controller->event_mask = event_mask;
   controller->cfg = GNUNET_CONFIGURATION_dup (cfg);
-  controller->client = GNUNET_CLIENT_connect ("testbed", controller->cfg);
-  if (NULL == controller->client)
+  controller->mq = GNUNET_CLIENT_connecT (controller->cfg,
+                                          "testbed",
+                                          handlers,
+                                          &mq_error_handler,
+                                          controller);
+  if (NULL == controller->mq)
   {
+    GNUNET_break (0);
     GNUNET_TESTBED_controller_disconnect (controller);
     return NULL;
   }
@@ -1629,9 +1648,8 @@ GNUNET_TESTBED_controller_connect (struct GNUNET_TESTBED_Host *host,
  * @param cls closure
  * @param key current key code
  * @param value value in the hash map
- * @return GNUNET_YES if we should continue to
- *         iterate,
- *         GNUNET_NO if not.
+ * @return #GNUNET_YES if we should continue to iterate,
+ *         #GNUNET_NO if not.
  */
 static int
 opc_free_iterator (void *cls, uint32_t key, void *value)
@@ -1658,20 +1676,11 @@ opc_free_iterator (void *cls, uint32_t key, void *value)
 void
 GNUNET_TESTBED_controller_disconnect (struct GNUNET_TESTBED_Controller *c)
 {
-  struct MessageQueue *mq_entry;
-
-  if (NULL != c->th)
-    GNUNET_CLIENT_notify_transmit_ready_cancel (c->th);
-  /* Clear the message queue */
-  while (NULL != (mq_entry = c->mq_head))
+  if (NULL != c->mq)
   {
-    GNUNET_CONTAINER_DLL_remove (c->mq_head, c->mq_tail,
-                                 mq_entry);
-    GNUNET_free (mq_entry->msg);
-    GNUNET_free (mq_entry);
+    GNUNET_MQ_destroy (c->mq);
+    c->mq = NULL;
   }
-  if (NULL != c->client)
-    GNUNET_CLIENT_disconnect (c->client);
   if (NULL != c->host)
     GNUNET_TESTBED_deregister_host_at_ (c->host, c);
   GNUNET_CONFIGURATION_destroy (c->cfg);
@@ -1703,7 +1712,8 @@ GNUNET_TESTBED_controller_disconnect (struct GNUNET_TESTBED_Controller *c)
  * @return the size of the xconfig
  */
 size_t
-GNUNET_TESTBED_compress_config_ (const char *config, size_t size,
+GNUNET_TESTBED_compress_config_ (const char *config,
+                                 size_t size,
                                  char **xconfig)
 {
   size_t xsize;
@@ -2002,6 +2012,8 @@ GNUNET_TESTBED_operation_done (struct GNUNET_TESTBED_Operation *operation)
  * #GNUNET_MESSAGE_TYPE_TESTBED_LINK_CONTROLLERS,
  * #GNUNET_MESSAGE_TYPE_TESTBED_LINK_CONTROLLERS_RESULT,
  *
+ * FIXME: This API is incredibly ugly.
+ *
  * @param msg the message containing compressed configuration
  * @return handle to the parsed configuration; NULL upon error while parsing the message
  */
@@ -2407,7 +2419,4 @@ GNUNET_TESTBED_barrier_cancel (struct GNUNET_TESTBED_Barrier *barrier)
 }
 
 
-
-
-
 /* end of testbed_api.c */
diff --git a/src/testbed/testbed_api.h b/src/testbed/testbed_api.h
index cd3358c9a8..82ff5918fa 100644
--- a/src/testbed/testbed_api.h
+++ b/src/testbed/testbed_api.h
@@ -105,33 +105,28 @@ enum OperationType
 };
 
 
-/**
- * The message queue for sending messages to the controller service
- */
-struct MessageQueue;
-
 
 /**
  * Enumeration of states of OperationContext
  */
 enum OperationContextState
 {
-    /**
-     * The initial state where the associated operation has just been created
-     * and is waiting in the operation queues to be started
-     */
+  /**
+   * The initial state where the associated operation has just been created
+   * and is waiting in the operation queues to be started
+   */
   OPC_STATE_INIT = 0,
 
-    /**
-     * The operation has been started. It may occupy some resources which are to
-     * be freed if cancelled.
-     */
+  /**
+   * The operation has been started. It may occupy some resources which are to
+   * be freed if cancelled.
+   */
   OPC_STATE_STARTED,
 
-    /**
-     * The operation has finished. The end results of this operation may occupy
-     * some resources which are to be freed by operation_done
-     */
+  /**
+   * The operation has finished. The end results of this operation may occupy
+   * some resources which are to be freed by operation_done
+   */
   OPC_STATE_FINISHED
 };
 
@@ -219,24 +214,9 @@ struct GNUNET_TESTBED_Controller
   struct GNUNET_CONFIGURATION_Handle *cfg;
 
   /**
-   * The client connection handle to the controller service
-   */
-  struct GNUNET_CLIENT_Connection *client;
-
-  /**
-   * The head of the message queue
-   */
-  struct MessageQueue *mq_head;
-
-  /**
-   * The tail of the message queue
+   * The message queue to the controller service
    */
-  struct MessageQueue *mq_tail;
-
-  /**
-   * The client transmit handle
-   */
-  struct GNUNET_CLIENT_TransmitHandle *th;
+  struct GNUNET_MQ_Handle *mq;
 
   /**
    * The host registration handle; NULL if no current registration requests are
@@ -286,11 +266,6 @@ struct GNUNET_TESTBED_Controller
   uint64_t event_mask;
 
   /**
-   * Did we start the receive loop yet?
-   */
-  int in_receive;
-
-  /**
    * The operation id counter. use current value and increment
    */
   uint32_t operation_counter;
@@ -341,6 +316,7 @@ struct GNUNET_TESTBED_Barrier
  *
  * @param controller the handle to the controller
  * @param msg the message to queue
+ * @deprecated
  */
 void
 GNUNET_TESTBED_queue_message_ (struct GNUNET_TESTBED_Controller *controller,
@@ -382,7 +358,8 @@ GNUNET_TESTBED_remove_opc_ (const struct GNUNET_TESTBED_Controller *c,
  * @return the size of the xconfig
  */
 size_t
-GNUNET_TESTBED_compress_config_ (const char *config, size_t size,
+GNUNET_TESTBED_compress_config_ (const char *config,
+                                 size_t size,
                                  char **xconfig);
 
 
@@ -397,7 +374,8 @@ GNUNET_TESTBED_compress_config_ (const char *config, size_t size,
  */
 char *
 GNUNET_TESTBED_compress_cfg_ (const struct GNUNET_CONFIGURATION_Handle *cfg,
-                              size_t *size, size_t *xsize);
+                              size_t *size,
+                              size_t *xsize);
 
 
 /**
@@ -414,9 +392,9 @@ GNUNET_TESTBED_compress_cfg_ (const struct GNUNET_CONFIGURATION_Handle *cfg,
  * @return the initialization message
  */
 struct GNUNET_TESTBED_HelperInit *
-GNUNET_TESTBED_create_helper_init_msg_ (const char *cname, const char *hostname,
-                                        const struct GNUNET_CONFIGURATION_Handle
-                                        *cfg);
+GNUNET_TESTBED_create_helper_init_msg_ (const char *cname,
+                                        const char *hostname,
+                                        const struct GNUNET_CONFIGURATION_Handle *cfg);
 
 
 /**
@@ -434,8 +412,8 @@ GNUNET_TESTBED_create_helper_init_msg_ (const char *cname, const char *hostname,
  *           operation
  */
 struct OperationContext *
-GNUNET_TESTBED_forward_operation_msg_ (struct GNUNET_TESTBED_Controller
-                                       *controller, uint64_t operation_id,
+GNUNET_TESTBED_forward_operation_msg_ (struct GNUNET_TESTBED_Controller *controller,
+                                       uint64_t operation_id,
                                        const struct GNUNET_MessageHeader *msg,
                                        GNUNET_CLIENT_MessageHandler cc,
                                        void *cc_cls);
@@ -453,8 +431,8 @@ GNUNET_TESTBED_forward_operation_msg_cancel_ (struct OperationContext *opc);
 /**
  * Generates configuration by uncompressing configuration in given message. The
  * given message should be of the following types:
- * GNUNET_MESSAGE_TYPE_TESTBED_PEERCONFIG,
- * GNUNET_MESSAGE_TYPE_TESTBED_SLAVECONFIG
+ * #GNUNET_MESSAGE_TYPE_TESTBED_PEERCONFIG,
+ * #GNUNET_MESSAGE_TYPE_TESTBED_SLAVECONFIG
  *
  * @param msg the message containing compressed configuration
  * @return handle to the parsed configuration
@@ -471,9 +449,7 @@ GNUNET_TESTBED_extract_config_ (const struct GNUNET_MessageHeader *msg);
  * @return the error message
  */
 const char *
-GNUNET_TESTBED_parse_error_string_ (const struct
-                                    GNUNET_TESTBED_OperationFailureEventMessage
-                                    *msg);
+GNUNET_TESTBED_parse_error_string_ (const struct GNUNET_TESTBED_OperationFailureEventMessage *msg);
 
 
 /**
diff --git a/src/testbed/testbed_api_hosts.c b/src/testbed/testbed_api_hosts.c
index 5b1df615ec..4e14a4d4d0 100644
--- a/src/testbed/testbed_api_hosts.c
+++ b/src/testbed/testbed_api_hosts.c
@@ -1385,33 +1385,6 @@ GNUNET_TESTBED_is_host_habitable_cancel (struct
 
 
 /**
- * handle for host registration
- */
-struct GNUNET_TESTBED_HostRegistrationHandle
-{
-  /**
-   * The host being registered
-   */
-  struct GNUNET_TESTBED_Host *host;
-
-  /**
-   * The controller at which this host is being registered
-   */
-  struct GNUNET_TESTBED_Controller *c;
-
-  /**
-   * The Registartion completion callback
-   */
-  GNUNET_TESTBED_HostRegistrationCompletion cc;
-
-  /**
-   * The closure for above callback
-   */
-  void *cc_cls;
-};
-
-
-/**
  * Register a host with the controller
  *
  * @param controller the controller handle
@@ -1537,63 +1510,6 @@ GNUNET_TESTBED_host_queue_oc_ (struct GNUNET_TESTBED_Host *h,
 
 
 /**
- * Handler for GNUNET_MESSAGE_TYPE_TESTBED_ADDHOSTCONFIRM message from
- * controller (testbed service)
- *
- * @param c the controller handler
- * @param msg message received
- * @return GNUNET_YES if we can continue receiving from service; GNUNET_NO if
- *           not
- */
-int
-GNUNET_TESTBED_host_handle_addhostconfirm_ (struct GNUNET_TESTBED_Controller *c,
-                                            const struct
-                                            GNUNET_TESTBED_HostConfirmedMessage
-                                            *msg)
-{
-  struct GNUNET_TESTBED_HostRegistrationHandle *rh;
-  char *emsg;
-  uint16_t msg_size;
-
-  rh = c->rh;
-  if (NULL == rh)
-  {
-    return GNUNET_OK;
-  }
-  if (GNUNET_TESTBED_host_get_id_ (rh->host) != ntohl (msg->host_id))
-  {
-    LOG_DEBUG ("Mismatch in host id's %u, %u of host confirm msg\n",
-               GNUNET_TESTBED_host_get_id_ (rh->host), ntohl (msg->host_id));
-    return GNUNET_OK;
-  }
-  c->rh = NULL;
-  msg_size = ntohs (msg->header.size);
-  if (sizeof (struct GNUNET_TESTBED_HostConfirmedMessage) == msg_size)
-  {
-    LOG_DEBUG ("Host %u successfully registered\n", ntohl (msg->host_id));
-    GNUNET_TESTBED_mark_host_registered_at_ (rh->host, c);
-    rh->cc (rh->cc_cls, NULL);
-    GNUNET_free (rh);
-    return GNUNET_OK;
-  }
-  /* We have an error message */
-  emsg = (char *) &msg[1];
-  if ('\0' !=
-      emsg[msg_size - sizeof (struct GNUNET_TESTBED_HostConfirmedMessage)])
-  {
-    GNUNET_break (0);
-    GNUNET_free (rh);
-    return GNUNET_NO;
-  }
-  LOG (GNUNET_ERROR_TYPE_ERROR, _("Adding host %u failed with error: %s\n"),
-       ntohl (msg->host_id), emsg);
-  rh->cc (rh->cc_cls, emsg);
-  GNUNET_free (rh);
-  return GNUNET_OK;
-}
-
-
-/**
  * Resolves the hostname of the host to an ip address
  *
  * @param host the host whose hostname is to be resolved
diff --git a/src/testbed/testbed_api_hosts.h b/src/testbed/testbed_api_hosts.h
index e58da1961d..2e53e1c649 100644
--- a/src/testbed/testbed_api_hosts.h
+++ b/src/testbed/testbed_api_hosts.h
@@ -24,8 +24,8 @@
  * @author Christian Grothoff
  */
 
-#ifndef NEW_TESTING_API_HOSTS_H
-#define NEW_TESTING_API_HOSTS_H
+#ifndef TESTBED_API_HOSTS_H
+#define TESTBED_API_HOSTS_H
 
 //#include "gnunet_testbed_service.h"
 //#include "testbed_helper.h"
@@ -33,6 +33,33 @@
 
 
 /**
+ * handle for host registration
+ */
+struct GNUNET_TESTBED_HostRegistrationHandle
+{
+  /**
+   * The host being registered
+   */
+  struct GNUNET_TESTBED_Host *host;
+
+  /**
+   * The controller at which this host is being registered
+   */
+  struct GNUNET_TESTBED_Controller *c;
+
+  /**
+   * The Registartion completion callback
+   */
+  GNUNET_TESTBED_HostRegistrationCompletion cc;
+
+  /**
+   * The closure for above callback
+   */
+  void *cc_cls;
+};
+
+
+/**
  * Lookup a host by ID.
  *
  * @param id global host ID assigned to the host; 0 is
@@ -164,22 +191,6 @@ GNUNET_TESTBED_host_queue_oc_ (struct GNUNET_TESTBED_Host *h,
 
 
 /**
- * Handler for GNUNET_MESSAGE_TYPE_TESTBED_ADDHOSTCONFIRM message from
- * controller (testbed service)
- *
- * @param c the controller handler
- * @param msg message received
- * @return GNUNET_YES if we can continue receiving from service; GNUNET_NO if
- *           not
- */
-int
-GNUNET_TESTBED_host_handle_addhostconfirm_ (struct GNUNET_TESTBED_Controller *c,
-                                            const struct
-                                            GNUNET_TESTBED_HostConfirmedMessage
-                                            *msg);
-
-
-/**
  * Sends termination signal to the controller's helper process
  *
  * @param cproc the handle to the controller's helper process