refactoring ARM api to use new MQ

1 files changed, 505 insertions, 714 deletions

diff --git a/src/arm/arm_api.c b/src/arm/arm_api.c
index a89d423ec..ed36c61cd 100644
--- a/src/arm/arm_api.c
+++ b/src/arm/arm_api.c
@@ -32,151 +32,131 @@
 
 #define LOG(kind,...) GNUNET_log_from (kind, "arm-api",__VA_ARGS__)
 
+
 /**
- * Handle for interacting with ARM.
+ * Entry in a doubly-linked list of operations awaiting for replies
+ * (in-order) from the ARM service.
  */
-struct GNUNET_ARM_Handle
+struct GNUNET_ARM_Operation
 {
   /**
-   * Our control connection to the ARM service.
-   */
-  struct GNUNET_CLIENT_Connection *client;
-
-  /**
-   * The configuration that we are using.
-   */
-  struct GNUNET_CONFIGURATION_Handle *cfg;
-
-  /**
-   * Handle for our current transmission request.
-   */
-  struct GNUNET_CLIENT_TransmitHandle *cth;
-
-  /**
-   * Head of doubly-linked list of pending requests.
-   */
-  struct ARMControlMessage *control_pending_head;
-
-  /**
-   * Tail of doubly-linked list of pending requests.
-   */
-  struct ARMControlMessage *control_pending_tail;
-
-  /**
-   * Head of doubly-linked list of sent requests.
+   * This is a doubly-linked list.
    */
-  struct ARMControlMessage *control_sent_head;
+  struct GNUNET_ARM_Operation *next;
 
   /**
-   * Tail of doubly-linked list of sent requests.
+   * This is a doubly-linked list.
    */
-  struct ARMControlMessage *control_sent_tail;
+  struct GNUNET_ARM_Operation *prev;
 
   /**
-   * Callback to invoke on connection/disconnection.
+   * ARM handle.
    */
-  GNUNET_ARM_ConnectionStatusCallback conn_status;
+  struct GNUNET_ARM_Handle *h;
 
   /**
-   * Closure for conn_status.
+   * Callback for service state change requests.
    */
-  void *conn_status_cls;
+  GNUNET_ARM_ResultCallback result_cont;
 
   /**
-   * ARM control message for the 'arm_termination_handler'
-   * with the continuation to call once the ARM shutdown is done.
+   * Callback for service list requests.
    */
-  struct ARMControlMessage *thm;
+  GNUNET_ARM_ServiceListCallback list_cont;
 
   /**
-   * ID of the reconnect task (if any).
+   * Closure for @e result_cont or @e list_cont.
    */
-  struct GNUNET_SCHEDULER_Task *reconnect_task;
+  void *cont_cls;
 
   /**
-   * Current delay we use for re-trying to connect to core.
+   * Task for async completion.
    */
-  struct GNUNET_TIME_Relative retry_backoff;
+  struct GNUNET_SCHEDULER_Task *async;
 
   /**
-   * Counter for request identifiers
+   * Unique ID for the request.
    */
-  uint64_t request_id_counter;
+  uint64_t id;
 
   /**
-   * Are we currently disconnected and hence unable to send?
+   * Result of this operation for #notify_starting().
    */
-  unsigned char currently_down;
+  enum GNUNET_ARM_Result starting_ret;
 
   /**
-   * #GNUNET_YES if we're running a service test.
+   * Is this an operation to stop the ARM service?
    */
-  unsigned char service_test_is_active;
+  int is_arm_stop;
 };
 
 
 /**
- * Entry in a doubly-linked list of control messages to be transmitted
- * to the arm service.
- *
- * The actual message is allocated at the end of this struct.
+ * Handle for interacting with ARM.
  */
-struct ARMControlMessage
+struct GNUNET_ARM_Handle
 {
   /**
-   * This is a doubly-linked list.
+   * Our connection to the ARM service.
    */
-  struct ARMControlMessage *next;
+  struct GNUNET_MQ_Handle *mq;
 
   /**
-   * This is a doubly-linked list.
+   * The configuration that we are using.
    */
-  struct ARMControlMessage *prev;
+  const struct GNUNET_CONFIGURATION_Handle *cfg;
 
   /**
-   * ARM handle.
+   * Head of doubly-linked list of pending operations.
    */
-  struct GNUNET_ARM_Handle *h;
+  struct GNUNET_ARM_Operation *operation_pending_head;
 
   /**
-   * Message to send.
+   * Tail of doubly-linked list of pending operations.
    */
-  struct GNUNET_ARM_Message *msg;
+  struct GNUNET_ARM_Operation *operation_pending_tail;
 
   /**
-   * Callback for service state change requests.
+   * Callback to invoke on connection/disconnection.
    */
-  GNUNET_ARM_ResultCallback result_cont;
+  GNUNET_ARM_ConnectionStatusCallback conn_status;
 
   /**
-   * Callback for service list requests.
+   * Closure for @e conn_status.
    */
-  GNUNET_ARM_ServiceListCallback list_cont;
+  void *conn_status_cls;
 
   /**
-   * Closure for @e result_cont or @e list_cont.
+   * ARM operation where the goal is to wait for ARM shutdown to
+   * complete.  This operation is special in that it waits for an
+   * error on the @e mq.  So we complete it by calling the
+   * continuation in the #mq_error_handler().  Note that the operation
+   * is no longer in the @e operation_pending_head DLL once it is
+   * referenced from this field.
    */
-  void *cont_cls;
+  struct GNUNET_ARM_Operation *thm;
 
   /**
-   * Timeout for the operation.
+   * ID of the reconnect task (if any).
    */
-  struct GNUNET_TIME_Absolute timeout;
+  struct GNUNET_SCHEDULER_Task *reconnect_task;
 
   /**
-   * Task to run when request times out.
+   * Current delay we use for re-trying to connect to core.
    */
-  struct GNUNET_SCHEDULER_Task *timeout_task_id;
+  struct GNUNET_TIME_Relative retry_backoff;
 
   /**
-   * Flags for passing std descriptors to ARM (when starting ARM).
+   * Counter for request identifiers.  They are used to match replies
+   * from ARM to operations in the @e operation_pending_head DLL.
    */
-  enum GNUNET_OS_InheritStdioFlags std_inheritance;
+  uint64_t request_id_counter;
 
   /**
-   * Type of the request expressed as a message type (start, stop or list).
+   * Have we detected that ARM is up?
    */
-  uint16_t type;
+  int currently_up;
+
 };
 
 
@@ -191,18 +171,6 @@ reconnect_arm (struct GNUNET_ARM_Handle *h);
 
 
 /**
- * Check the list of pending requests, send the next
- * one to the arm.
- *
- * @param h arm handle
- * @param ignore_currently_down transmit message even if not initialized?
- */
-static void
-trigger_next_request (struct GNUNET_ARM_Handle *h,
-                      int ignore_currently_down);
-
-
-/**
  * Task scheduled to try to re-connect to arm.
  *
  * @param cls the `struct GNUNET_ARM_Handle`
@@ -213,8 +181,6 @@ reconnect_arm_task (void *cls)
   struct GNUNET_ARM_Handle *h = cls;
 
   h->reconnect_task = NULL;
-  LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Connecting to ARM service after delay\n");
   reconnect_arm (h);
 }
 
@@ -228,28 +194,33 @@ reconnect_arm_task (void *cls)
 static void
 reconnect_arm_later (struct GNUNET_ARM_Handle *h)
 {
-  if (GNUNET_NO != h->currently_down)
-    return;
-  if (NULL != h->cth)
-  {
-    GNUNET_CLIENT_notify_transmit_ready_cancel (h->cth);
-    h->cth = NULL;
-  }
-  if (NULL != h->client)
+  struct GNUNET_ARM_Operation *op;
+
+  if (NULL != h->mq)
   {
-    GNUNET_CLIENT_disconnect (h->client);
-    h->client = NULL;
+    GNUNET_MQ_destroy (h->mq);
+    h->mq = NULL;
   }
-  h->currently_down = GNUNET_YES;
+  h->currently_up = GNUNET_NO;
   GNUNET_assert (NULL == h->reconnect_task);
   h->reconnect_task =
       GNUNET_SCHEDULER_add_delayed (h->retry_backoff,
                                     &reconnect_arm_task,
                                     h);
-  /* Don't clear pending messages on disconnection, deliver them later
-  clear_pending_messages (h, GNUNET_ARM_REQUEST_DISCONNECTED);
-  GNUNET_assert (NULL == h->control_pending_head);
-  */
+  while (NULL != (op = h->operation_pending_head))
+  {
+    if (NULL != op->result_cont)
+      op->result_cont (op->cont_cls,
+                       GNUNET_ARM_REQUEST_DISCONNECTED,
+                       0);
+    if (NULL != op->list_cont)
+      op->list_cont (op->cont_cls,
+                     GNUNET_ARM_REQUEST_DISCONNECTED,
+                     0,
+                     NULL);
+    GNUNET_ARM_operation_cancel (op);
+  }
+  GNUNET_assert (NULL == h->operation_pending_head);
   h->retry_backoff = GNUNET_TIME_STD_BACKOFF (h->retry_backoff);
   if (NULL != h->conn_status)
     h->conn_status (h->conn_status_cls,
@@ -264,176 +235,50 @@ reconnect_arm_later (struct GNUNET_ARM_Handle *h)
  * @param id unique message ID to use for the lookup
  * @return NULL if not found
  */
-static struct ARMControlMessage *
-find_cm_by_id (struct GNUNET_ARM_Handle *h,
+static struct GNUNET_ARM_Operation *
+find_op_by_id (struct GNUNET_ARM_Handle *h,
                uint64_t id)
 {
-  struct ARMControlMessage *result;
+  struct GNUNET_ARM_Operation *result;
 
-  for (result = h->control_sent_head; NULL != result; result = result->next)
-    if (id == result->msg->request_id)
+  for (result = h->operation_pending_head; NULL != result; result = result->next)
+    if (id == result->id)
       return result;
   return NULL;
 }
 
 
 /**
- * Handler for ARM 'termination' reply (failure to receive).
- *
- * @param cls our `struct GNUNET_ARM_Handle`
- * @param msg expected to be NULL
- */
-static void
-arm_termination_handler (void *cls,
-                         const struct GNUNET_MessageHeader *msg)
-{
-  struct GNUNET_ARM_Handle *h = cls;
-  struct ARMControlMessage *cm;
-
-  if (NULL != msg)
-  {
-    GNUNET_break (0);
-    GNUNET_CLIENT_receive (h->client,
-                           &arm_termination_handler,
-                           h,
-                           GNUNET_TIME_UNIT_FOREVER_REL);
-    return;
-  }
-  cm = h->thm;
-  h->thm = NULL;
-  h->currently_down = GNUNET_YES;
-  GNUNET_CLIENT_disconnect (h->client);
-  h->client = NULL;
-  if (NULL != cm->result_cont)
-    cm->result_cont (cm->cont_cls,
-                     GNUNET_ARM_REQUEST_SENT_OK,
-                     (const char *) &cm->msg[1],
-                     GNUNET_ARM_RESULT_STOPPED);
-  GNUNET_free (cm->msg);
-  GNUNET_free (cm);
-}
-
-
-/**
  * Handler for ARM replies.
  *
  * @param cls our `struct GNUNET_ARM_Handle`
- * @param msg the message received from the arm service
+ * @param res the message received from the arm service
  */
 static void
-client_notify_handler (void *cls,
-                       const struct GNUNET_MessageHeader *msg)
+handle_arm_result (void *cls,
+                   const struct GNUNET_ARM_ResultMessage *res)
 {
   struct GNUNET_ARM_Handle *h = cls;
-  const struct GNUNET_ARM_Message *arm_msg;
-  const struct GNUNET_ARM_ResultMessage *res;
-  const struct GNUNET_ARM_ListResultMessage *lres;
-  struct ARMControlMessage *cm;
-  const char **list;
-  const char *pos;
+  struct GNUNET_ARM_Operation *op;
   uint64_t id;
   enum GNUNET_ARM_Result result;
-  uint16_t size_check;
-  uint16_t rcount;
-  uint16_t msize;
-  unsigned char fail;
+  GNUNET_ARM_ResultCallback result_cont;
+  void *result_cont_cls;
 
-  list = NULL;
-  rcount = 0;
-  if (NULL == msg)
-  {
-    LOG (GNUNET_ERROR_TYPE_INFO,
-         _("Client was disconnected from arm service, trying to reconnect.\n"));
-    reconnect_arm_later (h);
-    return;
-  }
-  msize = ntohs (msg->size);
-  LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Processing message of type %u and size %u from arm service\n",
-       ntohs (msg->type), msize);
-  if (msize < sizeof (struct GNUNET_ARM_Message))
-  {
-    GNUNET_break (0);
-    reconnect_arm_later (h);
-    return;
-  }
-  arm_msg = (const struct GNUNET_ARM_Message *) msg;
-  GNUNET_break (0 == ntohl (arm_msg->reserved));
-  id = GNUNET_ntohll (arm_msg->request_id);
-  cm = find_cm_by_id (h, id);
-  if (NULL == cm)
+  id = GNUNET_ntohll (res->arm_msg.request_id);
+  op = find_op_by_id (h,
+                      id);
+  if (NULL == op)
   {
     LOG (GNUNET_ERROR_TYPE_DEBUG,
          "Message with unknown id %llu\n",
-         id);
-    return;
-  }
-  fail = GNUNET_NO;
-  switch (ntohs (msg->type))
-  {
-  case GNUNET_MESSAGE_TYPE_ARM_RESULT:
-    if (msize < sizeof (struct GNUNET_ARM_ResultMessage))
-    {
-      GNUNET_assert (0);
-      fail = GNUNET_YES;
-    }
-    break;
-  case GNUNET_MESSAGE_TYPE_ARM_LIST_RESULT:
-    if (msize < sizeof (struct GNUNET_ARM_ListResultMessage))
-    {
-      GNUNET_break (0);
-      fail = GNUNET_YES;
-      break;
-    }
-    size_check = 0;
-    lres = (const struct GNUNET_ARM_ListResultMessage *) msg;
-    rcount = ntohs (lres->count);
-    {
-      unsigned int i;
-
-      list = GNUNET_malloc (sizeof (const char *) * rcount);
-      pos = (const char *)&lres[1];
-      for (i = 0; i < rcount; i++)
-      {
-        const char *end = memchr (pos, 0, msize - size_check);
-        if (NULL == end)
-        {
-          GNUNET_break (0);
-          fail = GNUNET_YES;
-          break;
-        }
-        list[i] = pos;
-        size_check += (end - pos) + 1;
-        pos = end + 1;
-      }
-      if (GNUNET_YES == fail)
-      {
-        GNUNET_free (list);
-        list = NULL;
-      }
-    }
-    break;
-  default:
-    fail = GNUNET_YES;
-    break;
-  }
-  GNUNET_assert (NULL != cm->timeout_task_id);
-  GNUNET_SCHEDULER_cancel (cm->timeout_task_id);
-  GNUNET_CONTAINER_DLL_remove (h->control_sent_head,
-                               h->control_sent_tail,
-                               cm);
-  if (GNUNET_YES == fail)
-  {
-    reconnect_arm_later (h);
-    GNUNET_free (cm->msg);
-    GNUNET_free (cm);
+         (unsigned long long) id);
     return;
   }
-  if ( (GNUNET_MESSAGE_TYPE_ARM_RESULT == ntohs (msg->type)) &&
-       (0 == strcasecmp ((const char *) &cm->msg[1],
-                         "arm")) &&
-       (NULL != (res = (const struct GNUNET_ARM_ResultMessage *) msg)) &&
-       (GNUNET_ARM_RESULT_STOPPING == ntohl (res->result)) )
+
+  result = (enum GNUNET_ARM_Result) ntohl (res->result);
+  if ( (GNUNET_YES == op->is_arm_stop) &&
+       (GNUNET_ARM_RESULT_STOPPING == result) )
   {
     /* special case: if we are stopping 'gnunet-service-arm', we do not just
        wait for the result message, but also wait for the service to close
@@ -443,184 +288,159 @@ client_notify_handler (void *cls,
     if (NULL != h->thm)
     {
       GNUNET_break (0);
-      cm->result_cont (h->thm->cont_cls,
+      op->result_cont (h->thm->cont_cls,
                        GNUNET_ARM_REQUEST_SENT_OK,
-                       (const char *) &h->thm->msg[1],
                        GNUNET_ARM_RESULT_IS_NOT_KNOWN);
-      GNUNET_free (h->thm->msg);
       GNUNET_free (h->thm);
     }
-    h->thm = cm;
-    GNUNET_CLIENT_receive (h->client,
-                           &arm_termination_handler,
-                           h,
-                           GNUNET_TIME_UNIT_FOREVER_REL);
+    GNUNET_CONTAINER_DLL_remove (h->operation_pending_head,
+                                 h->operation_pending_tail,
+                                 op);
+    h->thm = op;
     return;
   }
-  GNUNET_CLIENT_receive (h->client,
-                         &client_notify_handler,
-                         h,
-                         GNUNET_TIME_UNIT_FOREVER_REL);
-  switch (ntohs (msg->type))
-  {
-  case GNUNET_MESSAGE_TYPE_ARM_RESULT:
-    res = (const struct GNUNET_ARM_ResultMessage *) msg;
-    LOG (GNUNET_ERROR_TYPE_DEBUG,
-         "Received response from ARM for service `%s': %u\n",
-         (const char *) &cm->msg[1], ntohs (msg->type));
-    result = (enum GNUNET_ARM_Result) ntohl (res->result);
-    if (NULL != cm->result_cont)
-      cm->result_cont (cm->cont_cls,
-                       GNUNET_ARM_REQUEST_SENT_OK,
-                       (const char *) &cm->msg[1],
-                       result);
-    break;
-  case GNUNET_MESSAGE_TYPE_ARM_LIST_RESULT:
-    if (NULL != cm->list_cont)
-        cm->list_cont (cm->cont_cls,
-                       GNUNET_ARM_REQUEST_SENT_OK,
-                       rcount,
-                       list);
-    GNUNET_free_non_null (list);
-    break;
-  }
-  GNUNET_free (cm->msg);
-  GNUNET_free (cm);
+  result_cont = op->result_cont;
+  result_cont_cls = op->cont_cls;
+  GNUNET_ARM_operation_cancel (op);
+  if (NULL != result_cont)
+    result_cont (result_cont_cls,
+                 GNUNET_ARM_REQUEST_SENT_OK,
+                 result);
 }
 
 
 /**
- * Transmit the next message to the arm service.
+ * Checked that list result message is well-formed.
  *
- * @param cls closure with the `struct GNUNET_ARM_Handle`
- * @param size number of bytes available in @a buf
- * @param buf where the callee should write the message
- * @return number of bytes written to @a buf
+ * @param cls our `struct GNUNET_ARM_Handle`
+ * @param lres the message received from the arm service
+ * @return #GNUNET_OK if message is well-formed
  */
-static size_t
-transmit_arm_message (void *cls,
-                      size_t size,
-                      void *buf)
+static int
+check_arm_list_result (void *cls,
+                       const struct GNUNET_ARM_ListResultMessage *lres)
 {
-  struct GNUNET_ARM_Handle *h = cls;
-  struct ARMControlMessage *cm;
-  struct GNUNET_ARM_Message *arm_msg;
-  uint64_t request_id;
-  int notify_connection;
-  uint16_t msize;
+  const char *pos = (const char *) &lres[1];
+  uint16_t rcount = ntohs (lres->count);
+  uint16_t msize = ntohs (lres->arm_msg.header.size);
+  uint16_t size_check;
 
-  notify_connection = GNUNET_NO;
-  LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "transmit_arm_message is running with %p buffer of size %lu. ARM is known to be %s\n",
-       buf, size, h->currently_down ? "unconnected" : "connected");
-  GNUNET_assert (NULL == h->reconnect_task);
-  h->cth = NULL;
-  if ((GNUNET_YES == h->currently_down) && (NULL != buf))
-  {
-    h->currently_down = GNUNET_NO;
-    notify_connection = GNUNET_YES;
-    h->retry_backoff = GNUNET_TIME_UNIT_MILLISECONDS;
-    GNUNET_CLIENT_receive (h->client, &client_notify_handler, h,
-                           GNUNET_TIME_UNIT_FOREVER_REL);
-  }
-  if (NULL == buf)
+  size_check = 0;
+  for (unsigned int i = 0; i < rcount; i++)
   {
-    LOG (GNUNET_ERROR_TYPE_DEBUG,
-         "Transmission failed, initiating reconnect\n");
-    reconnect_arm_later (h);
-    return 0;
-  }
-  if (NULL == (cm = h->control_pending_head))
-  {
-    LOG (GNUNET_ERROR_TYPE_DEBUG,
-         "Queue is empty, not sending anything\n");
-    msize = 0;
-    goto end;
-  }
-  GNUNET_assert (NULL != cm->msg);
-  msize = ntohs (cm->msg->header.size);
-  if (size < msize)
-  {
-    LOG (GNUNET_ERROR_TYPE_DEBUG,
-        "Request is too big (%u < %u), not sending it\n", size, msize);
-    trigger_next_request (h, GNUNET_NO);
-    msize = 0;
-    goto end;
+    const char *end = memchr (pos, 0, msize - size_check);
+    if (NULL == end)
+    {
+      GNUNET_break (0);
+      return GNUNET_SYSERR;
+    }
+    size_check += (end - pos) + 1;
+    pos = end + 1;
   }
-  arm_msg = cm->msg;
-  if (0 == h->request_id_counter)
-    h->request_id_counter++;
-  request_id = h->request_id_counter++;
-  LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Transmitting control message with %u bytes of type %u to arm with id %llu\n",
-       (unsigned int) msize,
-       (unsigned int) ntohs (cm->msg->header.type),
-       request_id);
-  arm_msg->reserved = htonl (0);
-  arm_msg->request_id = GNUNET_htonll (request_id);
-  memcpy (buf, cm->msg, msize);
-  /* Otherwise we won't be able to find it later! */
-  arm_msg->request_id = request_id;
-  GNUNET_CONTAINER_DLL_remove (h->control_pending_head,
-                               h->control_pending_tail,
-                               cm);
-  GNUNET_CONTAINER_DLL_insert_tail (h->control_sent_head,
-                                    h->control_sent_tail,
-                                    cm);
-  /* Don't free msg, keep it around (kind of wasteful, but then we don't
-   * really have many messages to handle, and it'll be freed when it times
-   * out anyway.
-   */
-  trigger_next_request (h, GNUNET_NO);
-
- end:
-  if ((GNUNET_YES == notify_connection) && (NULL != h->conn_status))
-    h->conn_status (h->conn_status_cls, GNUNET_YES);
-  return msize;
+  return GNUNET_OK;
 }
 
 
 /**
- * Check the list of pending requests, send the next
- * one to the arm.
+ * Handler for ARM list replies.
  *
- * @param h arm handle
- * @param ignore_currently_down transmit message even if not initialized?
+ * @param cls our `struct GNUNET_ARM_Handle`
+ * @param lres the message received from the arm service
  */
 static void
-trigger_next_request (struct GNUNET_ARM_Handle *h,
-                      int ignore_currently_down)
+handle_arm_list_result (void *cls,
+                        const struct GNUNET_ARM_ListResultMessage *lres)
 {
-  uint16_t msize;
+  struct GNUNET_ARM_Handle *h = cls;
+  uint16_t rcount = ntohs (lres->count);
+  const char *list[rcount];
+  const char *pos = (const char *) &lres[1];
+  uint16_t msize = ntohs (lres->arm_msg.header.size);
+  struct GNUNET_ARM_Operation *op;
+  uint16_t size_check;
+  uint64_t id;
 
-  msize = sizeof (struct GNUNET_MessageHeader);
-  if ((GNUNET_YES == h->currently_down) && (ignore_currently_down == GNUNET_NO))
+  id = GNUNET_ntohll (lres->arm_msg.request_id);
+  op = find_op_by_id (h,
+                      id);
+  if (NULL == op)
   {
     LOG (GNUNET_ERROR_TYPE_DEBUG,
-         "ARM connection down, not processing queue\n");
+         "Message with unknown id %llu\n",
+         (unsigned long long) id);
     return;
   }
-  if (NULL != h->cth)
+  size_check = 0;
+  for (unsigned int i = 0; i < rcount; i++)
+  {
+    const char *end = memchr (pos,
+                              0,
+                              msize - size_check);
+
+    /* Assert, as this was already checked in #check_arm_list_result() */
+    GNUNET_assert (NULL != end);
+    list[i] = pos;
+    size_check += (end - pos) + 1;
+    pos = end + 1;
+  }
+  if (NULL != op->list_cont)
+    op->list_cont (op->cont_cls,
+                   GNUNET_ARM_REQUEST_SENT_OK,
+                   rcount,
+                   list);
+  GNUNET_ARM_operation_cancel (op);
+}
+
+
+/**
+ * Receive confirmation from test, ARM service is up.
+ *
+ * @param cls closure with the `struct GNUNET_ARM_Handle`
+ * @param msg message received
+ */
+static void
+handle_confirm (void *cls,
+                const struct GNUNET_MessageHeader *msg)
+{
+  struct GNUNET_ARM_Handle *h = cls;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Got confirmation from ARM that we are up!\n");
+  if (GNUNET_NO == h->currently_up)
   {
-    LOG (GNUNET_ERROR_TYPE_DEBUG,
-         "Request pending, not processing queue\n");
-    return;
+    h->currently_up = GNUNET_YES;
+    if (NULL != h->conn_status)
+      h->conn_status (h->conn_status_cls,
+                      GNUNET_YES);
   }
-  if (NULL != h->control_pending_head)
-    msize =
-        ntohs (h->control_pending_head->msg->header.size);
-  else if (GNUNET_NO == ignore_currently_down)
+}
+
+
+/**
+ * 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 with the `struct GNUNET_ARM_Handle *`
+ * @param error error code
+ */
+static void
+mq_error_handler (void *cls,
+                  enum GNUNET_MQ_Error error)
+{
+  struct GNUNET_ARM_Handle *h = cls;
+  struct GNUNET_ARM_Operation *op;
+
+  h->currently_up = GNUNET_NO;
+  if (NULL != (op = h->thm))
   {
-    LOG (GNUNET_ERROR_TYPE_DEBUG,
-         "Request queue empty, not processing queue\n");
-    return;                     /* no pending message */
+    h->thm = NULL;
+    op->result_cont (op->cont_cls,
+                     GNUNET_ARM_REQUEST_SENT_OK,
+                     GNUNET_ARM_RESULT_STOPPED);
+    GNUNET_free (op);
   }
-  h->cth =
-      GNUNET_CLIENT_notify_transmit_ready (h->client,
-                                           msize,
-                                           GNUNET_TIME_UNIT_FOREVER_REL,
-                                           GNUNET_NO,
-                                           &transmit_arm_message, h);
+  reconnect_arm_later (h);
 }
 
 
@@ -633,22 +453,47 @@ trigger_next_request (struct GNUNET_ARM_Handle *h,
 static int
 reconnect_arm (struct GNUNET_ARM_Handle *h)
 {
-  GNUNET_assert (NULL == h->client);
-  GNUNET_assert (GNUNET_YES == h->currently_down);
-  h->client = GNUNET_CLIENT_connect ("arm", h->cfg);
-  if (NULL == h->client)
+  GNUNET_MQ_hd_fixed_size (arm_result,
+                           GNUNET_MESSAGE_TYPE_ARM_RESULT,
+                           struct GNUNET_ARM_ResultMessage);
+  GNUNET_MQ_hd_var_size (arm_list_result,
+                         GNUNET_MESSAGE_TYPE_ARM_LIST_RESULT,
+                         struct GNUNET_ARM_ListResultMessage);
+  GNUNET_MQ_hd_fixed_size (confirm,
+                           GNUNET_MESSAGE_TYPE_TEST,
+                           struct GNUNET_MessageHeader);
+  struct GNUNET_MQ_MessageHandler handlers[] = {
+    make_arm_result_handler (h),
+    make_arm_list_result_handler (h),
+    make_confirm_handler (h),
+    GNUNET_MQ_handler_end ()
+  };
+  struct GNUNET_MessageHeader *test;
+  struct GNUNET_MQ_Envelope *env;
+
+  if (NULL != h->mq)
+    return GNUNET_OK;
+  GNUNET_assert (GNUNET_NO == h->currently_up);
+  h->mq = GNUNET_CLIENT_connecT (h->cfg,
+                                 "arm",
+                                 handlers,
+                                 &mq_error_handler,
+                                 h);
+  if (NULL == h->mq)
   {
     LOG (GNUNET_ERROR_TYPE_DEBUG,
-           "arm_api, GNUNET_CLIENT_connect returned NULL\n");
+         "GNUNET_CLIENT_connect returned NULL\n");
     if (NULL != h->conn_status)
       h->conn_status (h->conn_status_cls,
                       GNUNET_SYSERR);
     return GNUNET_SYSERR;
   }
   LOG (GNUNET_ERROR_TYPE_DEBUG,
-         "arm_api, GNUNET_CLIENT_connect returned non-NULL\n");
-  trigger_next_request (h,
-                        GNUNET_YES);
+       "Sending TEST message to ARM\n");
+  env = GNUNET_MQ_msg (test,
+                       GNUNET_MESSAGE_TYPE_TEST);
+  GNUNET_MQ_send (h->mq,
+                  env);
   return GNUNET_OK;
 }
 
@@ -661,22 +506,20 @@ reconnect_arm (struct GNUNET_ARM_Handle *h)
  *        the ARM service may internally use a different
  *        configuration to determine how to start the service).
  * @param conn_status will be called when connecting/disconnecting
- * @param cls closure for conn_status
+ * @param conn_status_cls closure for @a conn_status
  * @return context to use for further ARM operations, NULL on error.
  */
 struct GNUNET_ARM_Handle *
 GNUNET_ARM_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
                     GNUNET_ARM_ConnectionStatusCallback conn_status,
-                    void *cls)
+                    void *conn_status_cls)
 {
   struct GNUNET_ARM_Handle *h;
 
   h = GNUNET_new (struct GNUNET_ARM_Handle);
-  h->cfg = GNUNET_CONFIGURATION_dup (cfg);
-  h->currently_down = GNUNET_YES;
-  h->reconnect_task = NULL;
+  h->cfg = cfg;
   h->conn_status = conn_status;
-  h->conn_status_cls = cls;
+  h->conn_status_cls = conn_status_cls;
   if (GNUNET_OK != reconnect_arm (h))
   {
     GNUNET_free (h);
@@ -692,113 +535,60 @@ GNUNET_ARM_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
  * @param h the handle that was being used
  */
 void
-GNUNET_ARM_disconnect_and_free (struct GNUNET_ARM_Handle *h)
+GNUNET_ARM_disconnect (struct GNUNET_ARM_Handle *h)
 {
-  struct ARMControlMessage *cm;
+  struct GNUNET_ARM_Operation *op;
 
   LOG (GNUNET_ERROR_TYPE_DEBUG,
        "Disconnecting from ARM service\n");
-  if (NULL != h->cth)
+  while (NULL != (op = h->operation_pending_head))
   {
-    GNUNET_CLIENT_notify_transmit_ready_cancel (h->cth);
-    h->cth = NULL;
-  }
-  while ((NULL != (cm = h->control_pending_head))
-         || (NULL != (cm = h->control_sent_head)) )
-  {
-    if (NULL != h->control_pending_head)
-      GNUNET_CONTAINER_DLL_remove (h->control_pending_head,
-                                   h->control_pending_tail,
-                                   cm);
-    else
-      GNUNET_CONTAINER_DLL_remove (h->control_sent_head,
-                                   h->control_sent_tail,
-                                   cm);
-    GNUNET_assert (NULL != cm->timeout_task_id);
-    GNUNET_SCHEDULER_cancel (cm->timeout_task_id);
-    if (NULL != cm->result_cont)
-      cm->result_cont (cm->cont_cls,
+    GNUNET_CONTAINER_DLL_remove (h->operation_pending_head,
+                                 h->operation_pending_tail,
+                                 op);
+    if (NULL != op->result_cont)
+      op->result_cont (op->cont_cls,
                        GNUNET_ARM_REQUEST_DISCONNECTED,
-                       NULL,
                        0);
-    /* FIXME: What about list callback? */
-    GNUNET_free_non_null (cm->msg);
-    GNUNET_free (cm);
+    if (NULL != op->list_cont)
+      op->list_cont (op->cont_cls,
+                     GNUNET_ARM_REQUEST_DISCONNECTED,
+                     0,
+                     NULL);
+    if (NULL != op->async)
+    {
+      GNUNET_SCHEDULER_cancel (op->async);
+      op->async = NULL;
+    }
+    GNUNET_free (op);
   }
-  if (NULL != h->client)
+  if (NULL != h->mq)
   {
-    GNUNET_CLIENT_disconnect (h->client);
-    h->client = NULL;
+    GNUNET_MQ_destroy (h->mq);
+    h->mq = NULL;
   }
   if (NULL != h->reconnect_task)
   {
     GNUNET_SCHEDULER_cancel (h->reconnect_task);
     h->reconnect_task = NULL;
   }
-  if (GNUNET_NO == h->service_test_is_active)
-  {
-    GNUNET_CONFIGURATION_destroy (h->cfg);
-    GNUNET_free (h);
-  }
-}
-
-
-/**
- * Message timed out. Remove it from the queue.
- *
- * @param cls the message (struct ARMControlMessage *)
- */
-static void
-control_message_timeout (void *cls)
-{
-  struct ARMControlMessage *cm = cls;
-  struct GNUNET_ARM_Message *arm_msg;
-
-  LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Control message timed out\n");
-  arm_msg = cm->msg;
-  if ((NULL == arm_msg) || (0 == arm_msg->request_id))
-  {
-    GNUNET_CONTAINER_DLL_remove (cm->h->control_pending_head,
-                                 cm->h->control_pending_tail,
-                                 cm);
-  }
-  else
-  {
-    GNUNET_CONTAINER_DLL_remove (cm->h->control_sent_head,
-                                 cm->h->control_sent_tail,
-                                 cm);
-  }
-  if (NULL != cm->result_cont)
-    cm->result_cont (cm->cont_cls,
-                     GNUNET_ARM_REQUEST_TIMEOUT,
-                     NULL, 0);
-  else if (NULL != cm->list_cont)
-    cm->list_cont (cm->cont_cls,
-                   GNUNET_ARM_REQUEST_TIMEOUT,
-                   0, NULL);
-  GNUNET_free_non_null (cm->msg);
-  GNUNET_free (cm);
+  GNUNET_free (h);
 }
 
 
 /**
  * A client specifically requested starting of ARM itself.
- * This function is called with information about whether
- * or not ARM is running; if it is, report success.  If
- * it is not, start the ARM process.
+ * Starts the ARM service.
  *
- * @param cls the context for the request that we will report on (struct ARMControlMessage *)
- * @param result #GNUNET_YES if ARM is running
+ * @param h the handle with configuration details
+ * @param std_inheritance inheritance of std streams
+ * @return operation status code
  */
-static void
-arm_service_report (void *cls,
-                    int result)
+static enum GNUNET_ARM_Result
+start_arm_service (struct GNUNET_ARM_Handle *h,
+                   enum GNUNET_OS_InheritStdioFlags std_inheritance)
 {
-  struct ARMControlMessage *cm = cls;
-  struct GNUNET_ARM_Handle *h;
   struct GNUNET_OS_Process *proc;
-  unsigned char test_is_active;
   char *cbinary;
   char *binary;
   char *quotedbinary;
@@ -806,51 +596,20 @@ arm_service_report (void *cls,
   char *loprefix;
   char *lopostfix;
 
-  test_is_active = cm->h->service_test_is_active;
-  if ((GNUNET_YES == test_is_active) &&
-      (GNUNET_YES == result))
-  {
-    LOG (GNUNET_ERROR_TYPE_DEBUG,
-         "Looks like `%s' is already running.\n",
-         "gnunet-service-arm");
-    /* arm is running! */
-    if (cm->result_cont)
-      cm->result_cont (cm->cont_cls,
-                       GNUNET_ARM_REQUEST_SENT_OK, "arm",
-                       GNUNET_ARM_RESULT_IS_STARTED_ALREADY);
-  }
-  if (GNUNET_NO == test_is_active)
-  {
-    /* User disconnected & destroyed ARM handle in the middle of
-     * the service test, so we kept the handle around until now.
-     */
-    GNUNET_CONFIGURATION_destroy (cm->h->cfg);
-    GNUNET_free (cm->h);
-  }
-  if ((GNUNET_YES == result) ||
-      (GNUNET_NO == test_is_active))
-  {
-    GNUNET_free (cm);
-    return;
-  }
-  cm->h->service_test_is_active = GNUNET_NO;
-  LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Looks like `%s' is not running, will start it.\n",
-       "gnunet-service-arm");
   if (GNUNET_OK !=
-      GNUNET_CONFIGURATION_get_value_string (cm->h->cfg,
+      GNUNET_CONFIGURATION_get_value_string (h->cfg,
                                              "arm",
                                              "PREFIX",
                                              &loprefix))
     loprefix = GNUNET_strdup ("");
   if (GNUNET_OK !=
-      GNUNET_CONFIGURATION_get_value_string (cm->h->cfg,
+      GNUNET_CONFIGURATION_get_value_string (h->cfg,
                                              "arm",
                                              "OPTIONS",
                                              &lopostfix))
     lopostfix = GNUNET_strdup ("");
   if (GNUNET_OK !=
-      GNUNET_CONFIGURATION_get_value_string (cm->h->cfg,
+      GNUNET_CONFIGURATION_get_value_string (h->cfg,
                                              "arm",
                                              "BINARY",
                                              &cbinary))
@@ -858,18 +617,14 @@ arm_service_report (void *cls,
     GNUNET_log_config_missing (GNUNET_ERROR_TYPE_WARNING,
                                "arm",
                                "BINARY");
-    if (cm->result_cont)
-      cm->result_cont (cm->cont_cls,
-                       GNUNET_ARM_REQUEST_SENT_OK, "arm",
-                       GNUNET_ARM_RESULT_IS_NOT_KNOWN);
-    GNUNET_free (cm);
     GNUNET_free (loprefix);
     GNUNET_free (lopostfix);
-    return;
+    return GNUNET_ARM_RESULT_IS_NOT_KNOWN;
   }
   if (GNUNET_OK !=
-      GNUNET_CONFIGURATION_get_value_filename (cm->h->cfg,
-                                               "arm", "CONFIG",
+      GNUNET_CONFIGURATION_get_value_filename (h->cfg,
+                                               "arm",
+                                               "CONFIG",
                                                &config))
     config = NULL;
   binary = GNUNET_OS_get_libexec_binary_path (cbinary);
@@ -878,15 +633,15 @@ arm_service_report (void *cls,
                    binary);
   GNUNET_free (cbinary);
   if ( (GNUNET_YES ==
-        GNUNET_CONFIGURATION_have_value (cm->h->cfg,
+        GNUNET_CONFIGURATION_have_value (h->cfg,
                                          "TESTING",
                                          "WEAKRANDOM")) &&
        (GNUNET_YES ==
-        GNUNET_CONFIGURATION_get_value_yesno (cm->h->cfg,
+        GNUNET_CONFIGURATION_get_value_yesno (h->cfg,
                                               "TESTING",
                                               "WEAKRANDOM")) &&
        (GNUNET_NO ==
-        GNUNET_CONFIGURATION_have_value (cm->h->cfg,
+        GNUNET_CONFIGURATION_have_value (h->cfg,
                                          "TESTING",
                                          "HOSTFILE")))
   {
@@ -894,39 +649,43 @@ arm_service_report (void *cls,
     /* we're clearly running a test, don't daemonize */
     if (NULL == config)
       proc = GNUNET_OS_start_process_s (GNUNET_NO,
-                                        cm->std_inheritance,
+                                        std_inheritance,
                                         NULL,
                                         loprefix,
                                         quotedbinary,
                                         /* no daemonization! */
-                                        lopostfix, NULL);
+                                        lopostfix,
+                                        NULL);
     else
       proc = GNUNET_OS_start_process_s (GNUNET_NO,
-                                        cm->std_inheritance,
+                                        std_inheritance,
                                         NULL,
                                         loprefix,
                                         quotedbinary,
                                         "-c", config,
                                         /* no daemonization! */
-                                        lopostfix, NULL);
+                                        lopostfix,
+                                        NULL);
   }
   else
   {
     if (NULL == config)
       proc = GNUNET_OS_start_process_s (GNUNET_NO,
-                                        cm->std_inheritance,
+                                        std_inheritance,
                                         NULL,
                                         loprefix,
                                         quotedbinary,
-                                        "-d", lopostfix, NULL);
+                                        "-d", /* do daemonize */
+                                        lopostfix, NULL);
     else
       proc = GNUNET_OS_start_process_s (GNUNET_NO,
-                                        cm->std_inheritance,
+                                        std_inheritance,
                                         NULL,
                                         loprefix,
                                         quotedbinary,
                                         "-c", config,
-                                        "-d", lopostfix,
+                                        "-d", /* do daemonize */
+                                        lopostfix,
                                         NULL);
   }
   GNUNET_free (binary);
@@ -935,22 +694,32 @@ arm_service_report (void *cls,
   GNUNET_free (loprefix);
   GNUNET_free (lopostfix);
   if (NULL == proc)
+    return GNUNET_ARM_RESULT_START_FAILED;
+  GNUNET_OS_process_destroy (proc);
+  return GNUNET_ARM_RESULT_STARTING;
+}
+
+
+/**
+ * Abort an operation.  Only prevents the callback from being
+ * called, the operation may still complete.
+ *
+ * @param op operation to cancel
+ */
+void
+GNUNET_ARM_operation_cancel (struct GNUNET_ARM_Operation *op)
+{
+  struct GNUNET_ARM_Handle *h = op->h;
+
+  if (h->thm == op)
   {
-    if (cm->result_cont)
-      cm->result_cont (cm->cont_cls,
-                       GNUNET_ARM_REQUEST_SENT_OK, "arm",
-                       GNUNET_ARM_RESULT_START_FAILED);
-    GNUNET_free (cm);
+    op->result_cont = NULL;
     return;
   }
-  if (cm->result_cont)
-    cm->result_cont (cm->cont_cls,
-                     GNUNET_ARM_REQUEST_SENT_OK, "arm",
-                     GNUNET_ARM_RESULT_STARTING);
-  GNUNET_OS_process_destroy (proc);
-  h = cm->h;
-  GNUNET_free (cm);
-  reconnect_arm (h);
+  GNUNET_CONTAINER_DLL_remove (h->operation_pending_head,
+                               h->operation_pending_tail,
+                               op);
+  GNUNET_free (op);
 }
 
 
@@ -959,21 +728,21 @@ arm_service_report (void *cls,
  *
  * @param h handle to ARM
  * @param service_name name of the service
- * @param timeout how long to wait before failing for good
  * @param cb callback to invoke when service is ready
  * @param cb_cls closure for @a cb
  * @param type type of the request
+ * @return handle to queue, NULL on error
  */
-static void
+static struct GNUNET_ARM_Operation *
 change_service (struct GNUNET_ARM_Handle *h,
                 const char *service_name,
-                struct GNUNET_TIME_Relative timeout,
                 GNUNET_ARM_ResultCallback cb,
                 void *cb_cls,
                 uint16_t type)
 {
-  struct ARMControlMessage *cm;
+  struct GNUNET_ARM_Operation *op;
   size_t slen;
+  struct GNUNET_MQ_Envelope *env;
   struct GNUNET_ARM_Message *msg;
 
   slen = strlen (service_name) + 1;
@@ -981,38 +750,81 @@ change_service (struct GNUNET_ARM_Handle *h,
       GNUNET_SERVER_MAX_MESSAGE_SIZE)
   {
     GNUNET_break (0);
-    if (cb != NULL)
-      cb (cb_cls, GNUNET_ARM_REQUEST_TOO_LONG, NULL, 0);
-    return;
+    return NULL;
   }
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "Requesting %s of service `%s'.\n",
-       (GNUNET_MESSAGE_TYPE_ARM_START == type) ? "start" : "termination",
-       service_name);
-  cm = GNUNET_malloc (sizeof (struct ARMControlMessage) + slen);
-  cm->h = h;
-  cm->result_cont = cb;
-  cm->cont_cls = cb_cls;
-  cm->timeout = GNUNET_TIME_relative_to_absolute (timeout);
-  memcpy (&cm[1], service_name, slen);
-  msg = GNUNET_malloc (sizeof (struct GNUNET_ARM_Message) + slen);
-  msg->header.size = htons (sizeof (struct GNUNET_ARM_Message) + slen);
-  msg->header.type = htons (type);
+  if (0 == h->request_id_counter)
+    h->request_id_counter++;
+  op = GNUNET_new (struct GNUNET_ARM_Operation);
+  op->h = h;
+  op->result_cont = cb;
+  op->cont_cls = cb_cls;
+  op->id = h->request_id_counter++;
+  GNUNET_CONTAINER_DLL_insert_tail (h->operation_pending_head,
+                                    h->operation_pending_tail,
+                                    op);
+  env = GNUNET_MQ_msg_extra (msg,
+                             slen,
+                             type);
   msg->reserved = htonl (0);
-  memcpy (&msg[1], service_name, slen);
-  cm->msg = msg;
+  msg->request_id = GNUNET_htonll (op->id);
+  memcpy (&msg[1],
+          service_name,
+          slen);
+  GNUNET_MQ_send (h->mq,
+                  env);
+  return op;
+}
+
+
+/**
+ * Task run to notify application that ARM is already up.
+ *
+ * @param cls the operation that asked ARM to be started
+ */
+static void
+notify_running (void *cls)
+{
+  struct GNUNET_ARM_Operation *op = cls;
+  struct GNUNET_ARM_Handle *h = op->h;
+
+  op->async = NULL;
+  GNUNET_CONTAINER_DLL_remove (h->operation_pending_head,
+                               h->operation_pending_tail,
+                               op);
+  if (NULL != op->result_cont)
+    op->result_cont (op->cont_cls,
+                     GNUNET_ARM_REQUEST_SENT_OK,
+                     GNUNET_ARM_RESULT_IS_STARTED_ALREADY);
+  if ( (GNUNET_YES == h->currently_up) &&
+       (NULL != h->conn_status) )
+    h->conn_status (h->conn_status_cls,
+                    GNUNET_YES);
+  GNUNET_free (op);
+}
+
+
+/**
+ * Task run to notify application that ARM is being started.
+ *
+ * @param cls the operation that asked ARM to be started
+ */
+static void
+notify_starting (void *cls)
+{
+  struct GNUNET_ARM_Operation *op = cls;
+  struct GNUNET_ARM_Handle *h = op->h;
+
+  op->async = NULL;
   LOG (GNUNET_ERROR_TYPE_DEBUG,
-      "Inserting a control message into the queue. Timeout is %s\n",
-       GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (cm->timeout),
-                                               GNUNET_NO));
-  GNUNET_CONTAINER_DLL_insert_tail (h->control_pending_head,
-                                    h->control_pending_tail,
-                                    cm);
-  cm->timeout_task_id =
-      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining
-                                    (cm->timeout),
-                                    &control_message_timeout,
-                                    cm);
-  trigger_next_request (h, GNUNET_NO);
+       "Notifying client that we started the ARM service\n");
+  GNUNET_CONTAINER_DLL_remove (h->operation_pending_head,
+                               h->operation_pending_tail,
+                               op);
+  if (NULL != op->result_cont)
+    op->result_cont (op->cont_cls,
+                     GNUNET_ARM_REQUEST_SENT_OK,
+                     op->starting_ret);
+  GNUNET_free (op);
 }
 
 
@@ -1022,132 +834,116 @@ change_service (struct GNUNET_ARM_Handle *h,
  * @param h handle to ARM
  * @param service_name name of the service
  * @param std_inheritance inheritance of std streams
- * @param timeout how long to wait before failing for good
  * @param cont callback to invoke after request is sent or not sent
  * @param cont_cls closure for @a cont
+ * @return handle for the operation, NULL on error
  */
-void
+struct GNUNET_ARM_Operation *
 GNUNET_ARM_request_service_start (struct GNUNET_ARM_Handle *h,
                                   const char *service_name,
                                   enum GNUNET_OS_InheritStdioFlags std_inheritance,
-                                  struct GNUNET_TIME_Relative timeout,
                                   GNUNET_ARM_ResultCallback cont,
                                   void *cont_cls)
 {
-  struct ARMControlMessage *cm;
-  size_t slen;
+  struct GNUNET_ARM_Operation *op;
+  enum GNUNET_ARM_Result ret;
 
   LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Asked to start service `%s' within %s\n", service_name,
-       GNUNET_STRINGS_relative_time_to_string (timeout, GNUNET_NO));
-  if (0 == strcasecmp ("arm", service_name))
+       "Starting service `%s'\n",
+       service_name);
+  if (0 != strcasecmp ("arm",
+                       service_name))
+    return change_service (h,
+                           service_name,
+                           cont,
+                           cont_cls,
+                           GNUNET_MESSAGE_TYPE_ARM_START);
+
+  /* Possible cases:
+   * 1) We're connected to ARM already. Invoke the callback immediately.
+   * 2) We're not connected to ARM.
+   *    Cancel any reconnection attempts temporarily, then perform
+   *    a service test.
+   */
+  if (GNUNET_YES == h->currently_up)
   {
-    /* Possible cases:
-     * 1) We're connected to ARM already. Invoke the callback immediately.
-     * 2) We're not connected to ARM.
-     *    Cancel any reconnection attempts temporarily, then perform
-     *    a service test.
-     */
-    if (GNUNET_NO == h->currently_down)
-    {
-      LOG (GNUNET_ERROR_TYPE_DEBUG,
-           "ARM is already running\n");
-      if (NULL != cont)
-        cont (cont_cls,
-              GNUNET_ARM_REQUEST_SENT_OK,
-              "arm",
-              GNUNET_ARM_RESULT_IS_STARTED_ALREADY);
-    }
-    else if (GNUNET_NO == h->service_test_is_active)
-    {
-      if (NULL != h->cth)
-      {
-        GNUNET_CLIENT_notify_transmit_ready_cancel (h->cth);
-        h->cth = NULL;
-      }
-      if (NULL != h->client)
-      {
-        GNUNET_CLIENT_disconnect (h->client);
-        h->client = NULL;
-      }
-      if (NULL != h->reconnect_task)
-      {
-        GNUNET_SCHEDULER_cancel (h->reconnect_task);
-        h->reconnect_task = NULL;
-      }
-
-      LOG (GNUNET_ERROR_TYPE_DEBUG,
-          "Not connected to ARM, will do a service test\n");
-
-      slen = strlen ("arm") + 1;
-      cm = GNUNET_malloc (sizeof (struct ARMControlMessage) + slen);
-      cm->h = h;
-      cm->result_cont = cont;
-      cm->cont_cls = cont_cls;
-      cm->timeout = GNUNET_TIME_relative_to_absolute (timeout);
-      cm->std_inheritance = std_inheritance;
-      memcpy (&cm[1], service_name, slen);
-      h->service_test_is_active = GNUNET_YES;
-      GNUNET_CLIENT_service_test ("arm",
-                                  h->cfg,
-                                  timeout,
-                                  &arm_service_report,
-                                  cm);
-    }
-    else
-    {
-      /* Service test is already running - tell user to chill out and try
-       * again later.
-       */
-      LOG (GNUNET_ERROR_TYPE_DEBUG,
-           "Service test is already in progress, we're busy\n");
-      if (NULL != cont)
-        cont (cont_cls,
-              GNUNET_ARM_REQUEST_BUSY,
-              NULL, 0);
-    }
-    return;
-  }
-  change_service (h,
-                  service_name,
-                  timeout,
-                  cont, cont_cls,
-                  GNUNET_MESSAGE_TYPE_ARM_START);
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "ARM is already running\n");
+    op = GNUNET_new (struct GNUNET_ARM_Operation);
+    op->h = h;
+    op->result_cont = cont;
+    op->cont_cls = cont_cls;
+    GNUNET_CONTAINER_DLL_insert_tail (h->operation_pending_head,
+                                      h->operation_pending_tail,
+                                      op);
+    op->async = GNUNET_SCHEDULER_add_now (&notify_running,
+                                          op);
+    return op;
+  }
+  /* This is an inherently uncertain choice, as it is of course
+     theoretically possible that ARM is up and we just did not
+     yet complete the MQ handshake.  However, given that users
+     are unlikely to hammer 'gnunet-arm -s' on a busy system,
+     the above check should catch 99.99% of the cases where ARM
+     is already running. */
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Starting ARM service\n");
+  ret = start_arm_service (h,
+                           std_inheritance);
+  if (GNUNET_ARM_RESULT_STARTING == ret)
+    reconnect_arm (h);
+  op = GNUNET_new (struct GNUNET_ARM_Operation);
+  op->h = h;
+  op->result_cont = cont;
+  op->cont_cls = cont_cls;
+  GNUNET_CONTAINER_DLL_insert_tail (h->operation_pending_head,
+                                    h->operation_pending_tail,
+                                    op);
+  op->starting_ret = ret;
+  op->async = GNUNET_SCHEDULER_add_now (&notify_starting,
+                                        op);
+  return op;
 }
 
 
 /**
- * Request a service to be stopped.
- * Stopping arm itself will not invalidate its handle, and
- * ARM API will try to restore connection to the ARM service,
- * even if ARM connection was lost because you asked for ARM to be stopped.
- * Call #GNUNET_ARM_disconnect_and_free() to free the handle and prevent
+ * Request a service to be stopped.  Stopping arm itself will not
+ * invalidate its handle, and ARM API will try to restore connection
+ * to the ARM service, even if ARM connection was lost because you
+ * asked for ARM to be stopped.  Call
+ * #GNUNET_ARM_disconnect() to free the handle and prevent
  * further connection attempts.
  *
  * @param h handle to ARM
  * @param service_name name of the service
- * @param timeout how long to wait before failing for good
  * @param cont callback to invoke after request is sent or is not sent
  * @param cont_cls closure for @a cont
+ * @return handle for the operation, NULL on error
  */
-void
+struct GNUNET_ARM_Operation *
 GNUNET_ARM_request_service_stop (struct GNUNET_ARM_Handle *h,
                                  const char *service_name,
-                                 struct GNUNET_TIME_Relative timeout,
                                  GNUNET_ARM_ResultCallback cont,
                                  void *cont_cls)
 {
+  struct GNUNET_ARM_Operation *op;
+
   LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Stopping service `%s' within %s\n",
-       service_name,
-       GNUNET_STRINGS_relative_time_to_string (timeout,
-                                               GNUNET_NO));
-  change_service (h,
-                  service_name,
-                  timeout,
-                  cont,
-                  cont_cls,
-                  GNUNET_MESSAGE_TYPE_ARM_STOP);
+       "Stopping service `%s'\n",
+       service_name);
+  op = change_service (h,
+                       service_name,
+                       cont,
+                       cont_cls,
+                       GNUNET_MESSAGE_TYPE_ARM_STOP);
+  if (NULL == op)
+    return NULL;
+  /* If the service is ARM, set a flag as we will use MQ errors
+     to detect that the process is really gone. */
+  if (0 == strcasecmp (service_name,
+                       "arm"))
+    op->is_arm_stop = GNUNET_YES;
+  return op;
 }
 
 
@@ -1155,43 +951,38 @@ GNUNET_ARM_request_service_stop (struct GNUNET_ARM_Handle *h,
  * Request a list of running services.
  *
  * @param h handle to ARM
- * @param timeout how long to wait before failing for good
  * @param cont callback to invoke after request is sent or is not sent
  * @param cont_cls closure for @a cont
+ * @return handle for the operation, NULL on error
  */
-void
+struct GNUNET_ARM_Operation *
 GNUNET_ARM_request_service_list (struct GNUNET_ARM_Handle *h,
-                                 struct GNUNET_TIME_Relative timeout,
                                  GNUNET_ARM_ServiceListCallback cont,
                                  void *cont_cls)
 {
-  struct ARMControlMessage *cm;
+  struct GNUNET_ARM_Operation *op;
+  struct GNUNET_MQ_Envelope *env;
   struct GNUNET_ARM_Message *msg;
 
   LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Requesting LIST from ARM service with timeout: %s\n",
-       GNUNET_STRINGS_relative_time_to_string (timeout,
-                                               GNUNET_YES));
-  cm = GNUNET_new (struct ARMControlMessage);
-  cm->h = h;
-  cm->list_cont = cont;
-  cm->cont_cls = cont_cls;
-  cm->timeout = GNUNET_TIME_relative_to_absolute (timeout);
-  msg = GNUNET_malloc (sizeof (struct GNUNET_ARM_Message));
-  msg->header.size = htons (sizeof (struct GNUNET_ARM_Message));
-  msg->header.type = htons (GNUNET_MESSAGE_TYPE_ARM_LIST);
+       "Requesting LIST from ARM service\n");
+  if (0 == h->request_id_counter)
+    h->request_id_counter++;
+  op = GNUNET_new (struct GNUNET_ARM_Operation);
+  op->h = h;
+  op->list_cont = cont;
+  op->cont_cls = cont_cls;
+  op->id = h->request_id_counter++;
+  GNUNET_CONTAINER_DLL_insert_tail (h->operation_pending_head,
+                                    h->operation_pending_tail,
+                                    op);
+  env = GNUNET_MQ_msg (msg,
+                       GNUNET_MESSAGE_TYPE_ARM_LIST);
   msg->reserved = htonl (0);
-  cm->msg = msg;
-  GNUNET_CONTAINER_DLL_insert_tail (h->control_pending_head,
-                                    h->control_pending_tail,
-                                    cm);
-  cm->timeout_task_id =
-      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining
-                                    (cm->timeout),
-                                    &control_message_timeout,
-                                    cm);
-  trigger_next_request (h,
-                        GNUNET_NO);
+  msg->request_id = GNUNET_htonll (op->id);
+  GNUNET_MQ_send (h->mq,
+                  env);
+  return op;
 }