convert dns api to MQ

1 files changed, 108 insertions, 239 deletions

diff --git a/src/dns/dns_api.c b/src/dns/dns_api.c
index f96ac3142..1d212e5d1 100644
--- a/src/dns/dns_api.c
+++ b/src/dns/dns_api.c
@@ -1,6 +1,6 @@
 /*
       This file is part of GNUnet
-      Copyright (C) 2012 GNUnet e.V.
+      Copyright (C) 2012, 2016 GNUnet e.V.
 
       GNUnet is free software; you can redistribute it and/or modify
       it under the terms of the GNU General Public License as published
@@ -29,29 +29,6 @@
 
 
 /**
- * Reply to send to service.
- */
-struct ReplyQueueEntry
-{
-  /**
-   * Kept in DLL.
-   */
-  struct ReplyQueueEntry *next;
-
-  /**
-   * Kept in DLL.
-   */
-  struct ReplyQueueEntry *prev;
-
-  /**
-   * Message to transmit, allocated at the end of this struct.
-   */
-  const struct GNUNET_MessageHeader *msg;
-
-};
-
-
-/**
  * Handle to identify an individual DNS request.
  */
 struct GNUNET_DNS_RequestHandle
@@ -84,12 +61,7 @@ struct GNUNET_DNS_Handle
   /**
    * Connection to DNS service, or NULL.
    */
-  struct GNUNET_CLIENT_Connection *dns_connection;
-
-  /**
-   * Handle to active transmission request, or NULL.
-   */
-  struct GNUNET_CLIENT_TransmitHandle *dns_transmit_handle;
+  struct GNUNET_MQ_Handle *mq;
 
   /**
    * Configuration to use.
@@ -102,24 +74,14 @@ struct GNUNET_DNS_Handle
   GNUNET_DNS_RequestHandler rh;
 
   /**
-   * Closure for 'rh'.
+   * Closure for @e rh.
    */
   void *rh_cls;
 
   /**
-   * Head of replies to transmit.
-   */
-  struct ReplyQueueEntry *rq_head;
-
-  /**
-   * Tail of replies to transmit.
-   */
-  struct ReplyQueueEntry *rq_tail;
-
-  /**
    * Task to reconnect to the service.
    */
-  struct GNUNET_SCHEDULER_Task * reconnect_task;
+  struct GNUNET_SCHEDULER_Task *reconnect_task;
 
   /**
    * Re-connect counter, to make sure we did not reconnect in the meantime.
@@ -132,11 +94,6 @@ struct GNUNET_DNS_Handle
   enum GNUNET_DNS_Flags flags;
 
   /**
-   * Did we start the receive loop yet?
-   */
-  int in_receive;
-
-  /**
    * Number of GNUNET_DNS_RequestHandles we have outstanding. Must be 0 before
    * we can be disconnected.
    */
@@ -145,73 +102,71 @@ struct GNUNET_DNS_Handle
 
 
 /**
- * Add the given reply to our transmission queue and trigger sending if needed.
+ * Reconnect to the DNS service.
+ *
+ * @param cls handle with the connection to connect
+ * @param tc scheduler context (unused)
+ */
+static void
+reconnect (void *cls);
+
+
+/**
+ * Drop the existing connection and reconnect to the DNS service.
  *
  * @param dh handle with the connection
- * @param qe reply to queue
  */
 static void
-queue_reply (struct GNUNET_DNS_Handle *dh,
-             struct ReplyQueueEntry *qe);
+force_reconnect (struct GNUNET_DNS_Handle *dh)
+{
+  if (NULL != dh->mq)
+  {
+    GNUNET_MQ_destroy (dh->mq);
+    dh->mq = NULL;
+  }
+  dh->reconnect_task =
+    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
+                                  &reconnect,
+                                  dh);
+}
 
 
 /**
- * Reconnect to the DNS service.
+ * 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 handle with the connection to connect
- * @param tc scheduler context (unused)
+ * @param cls closure with the `struct GNUNET_DNS_Handle *`
+ * @param error error code
  */
 static void
-reconnect (void *cls)
+mq_error_handler (void *cls,
+                  enum GNUNET_MQ_Error error)
 {
   struct GNUNET_DNS_Handle *dh = cls;
-  struct ReplyQueueEntry *qe;
-  struct GNUNET_DNS_Register *msg;
 
-  dh->reconnect_task = NULL;
-  dh->dns_connection = GNUNET_CLIENT_connect ("dns", dh->cfg);
-  if (NULL == dh->dns_connection)
-    return;
-  dh->generation++;
-  qe = GNUNET_malloc (sizeof (struct ReplyQueueEntry) +
-                      sizeof (struct GNUNET_DNS_Register));
-  msg = (struct GNUNET_DNS_Register*) &qe[1];
-  qe->msg = &msg->header;
-  msg->header.size = htons (sizeof (struct GNUNET_DNS_Register));
-  msg->header.type = htons (GNUNET_MESSAGE_TYPE_DNS_CLIENT_INIT);
-  msg->flags = htonl (dh->flags);
-  queue_reply (dh, qe);
+  force_reconnect (dh);
 }
 
 
+
 /**
- * Disconnect from the DNS service.
+ * This receives packets from the DNS service and calls the application to
+ * check that the request is well-formed
  *
- * @param dh handle with the connection to disconnect
+ * @param cls the struct GNUNET_DNS_Handle
+ * @param req message from the service (request)
  */
-static void
-disconnect (struct GNUNET_DNS_Handle *dh)
+static int
+check_request (void *cls,
+               const struct GNUNET_DNS_Request *req)
 {
-  struct ReplyQueueEntry *qe;
-
-  if (NULL != dh->dns_transmit_handle)
-  {
-    GNUNET_CLIENT_notify_transmit_ready_cancel (dh->dns_transmit_handle);
-    dh->dns_transmit_handle = NULL;
-  }
-  if (NULL != dh->dns_connection)
+  if (0 != ntohl (req->reserved))
   {
-    GNUNET_CLIENT_disconnect (dh->dns_connection);
-    dh->dns_connection = NULL;
-  }
-  while (NULL != (qe = dh->rq_head))
-  {
-    GNUNET_CONTAINER_DLL_remove (dh->rq_head,
-                                 dh->rq_tail,
-                                 qe);
-    GNUNET_free (qe);
+    GNUNET_break (0);
+    return GNUNET_SYSERR;
   }
-  dh->in_receive = GNUNET_NO;
+  return GNUNET_OK;
 }
 
 
@@ -219,44 +174,17 @@ disconnect (struct GNUNET_DNS_Handle *dh)
  * This receives packets from the DNS service and calls the application to
  * handle it.
  *
- * @param cls the struct GNUNET_DNS_Handle
+ * @param cls the `struct GNUNET_DNS_Handle *`
  * @param msg message from the service (request)
  */
 static void
-request_handler (void *cls,
-                 const struct GNUNET_MessageHeader *msg)
+handle_request (void *cls,
+                const struct GNUNET_DNS_Request *req)
 {
   struct GNUNET_DNS_Handle *dh = cls;
-  const struct GNUNET_DNS_Request *req;
+  size_t payload_length = ntohs (req->header.size) - sizeof (*req);
   struct GNUNET_DNS_RequestHandle *rh;
-  size_t payload_length;
 
-  /* the service disconnected, reconnect after short wait */
-  if (msg == NULL)
-  {
-    disconnect (dh);
-    dh->reconnect_task =
-        GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
-                                      &reconnect, dh);
-    return;
-  }
-  if ( (ntohs (msg->type) != GNUNET_MESSAGE_TYPE_DNS_CLIENT_REQUEST) ||
-       (ntohs (msg->size) < sizeof (struct GNUNET_DNS_Request)) )
-  {
-    /* the service did something strange, reconnect immediately */
-    GNUNET_break (0);
-    disconnect (dh);
-    dh->reconnect_task = GNUNET_SCHEDULER_add_now (&reconnect, dh);
-    return;
-  }
-  req = (const struct GNUNET_DNS_Request *) msg;
-  GNUNET_break (ntohl (req->reserved) == 0);
-  payload_length = ntohs (req->header.size) - sizeof (struct GNUNET_DNS_Request);
-  GNUNET_CLIENT_receive (dh->dns_connection,
-                         &request_handler, dh,
-                         GNUNET_TIME_UNIT_FOREVER_REL);
-
-  /* finally, pass request to callback for answers */
   rh = GNUNET_new (struct GNUNET_DNS_RequestHandle);
   rh->dh =dh;
   rh->request_id = req->request_id;
@@ -270,96 +198,38 @@ request_handler (void *cls,
 
 
 /**
- * Callback called by notify_transmit_ready; sends DNS replies
- * to the DNS service.
+ * Reconnect to the DNS service.
  *
- * @param cls the struct GNUNET_DNS_Handle
- * @param size number of bytes available in buf
- * @param buf where to copy the message for transmission
- * @return number of bytes copied to buf
+ * @param cls handle with the connection to connect
  */
-static size_t
-send_response (void *cls, size_t size, void *buf)
+static void
+reconnect (void *cls)
 {
+  GNUNET_MQ_hd_var_size (request,
+                         GNUNET_MESSAGE_TYPE_DNS_CLIENT_REQUEST,
+                         struct GNUNET_DNS_Request);
   struct GNUNET_DNS_Handle *dh = cls;
-  struct ReplyQueueEntry *qe;
-  size_t len;
-
-  dh->dns_transmit_handle = NULL;
-  if (NULL == buf)
-  {
-    disconnect (dh);
-    dh->reconnect_task =
-      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
-                                    &reconnect, dh);
-    return 0;
-  }
-  qe = dh->rq_head;
-  if (NULL == qe)
-    return 0;
-  len = ntohs (qe->msg->size);
-  if (len > size)
-  {
-    dh->dns_transmit_handle =
-      GNUNET_CLIENT_notify_transmit_ready (dh->dns_connection,
-                                           len,
-                                           GNUNET_TIME_UNIT_FOREVER_REL,
-                                           GNUNET_NO,
-                                           &send_response, dh);
-    return 0;
-  }
-  memcpy (buf, qe->msg, len);
-  GNUNET_CONTAINER_DLL_remove (dh->rq_head,
-                               dh->rq_tail,
-                               qe);
-  GNUNET_free (qe);
-  if (GNUNET_NO == dh->in_receive)
-  {
-    dh->in_receive = GNUNET_YES;
-    GNUNET_CLIENT_receive (dh->dns_connection,
-                           &request_handler, dh,
-                           GNUNET_TIME_UNIT_FOREVER_REL);
-  }
-  if (NULL != (qe = dh->rq_head))
-  {
-    dh->dns_transmit_handle =
-      GNUNET_CLIENT_notify_transmit_ready (dh->dns_connection,
-                                           ntohs (qe->msg->size),
-                                           GNUNET_TIME_UNIT_FOREVER_REL,
-                                           GNUNET_NO,
-                                           &send_response, dh);
-  }
-  return len;
-}
-
+  struct GNUNET_MQ_MessageHandler handlers[] = {
+    make_request_handler (dh),
+    GNUNET_MQ_handler_end ()
+  };
+  struct GNUNET_MQ_Envelope *env;
+  struct GNUNET_DNS_Register *msg;
 
-/**
- * Add the given reply to our transmission queue and trigger sending if needed.
- *
- * @param dh handle with the connection
- * @param qe reply to queue
- */
-static void
-queue_reply (struct GNUNET_DNS_Handle *dh,
-             struct ReplyQueueEntry *qe)
-{
-  if (NULL == dh->dns_connection)
-  {
-    GNUNET_free (qe);
-    return;
-  }
-  GNUNET_CONTAINER_DLL_insert_tail (dh->rq_head,
-                                    dh->rq_tail,
-                                    qe);
-  if (NULL != dh->dns_transmit_handle)
+  dh->reconnect_task = NULL;
+  dh->mq = GNUNET_CLIENT_connecT (dh->cfg,
+                                  "dns",
+                                  handlers,
+                                  &mq_error_handler,
+                                  dh);
+  if (NULL == dh->mq)
     return;
-  /* trigger sending */
-  dh->dns_transmit_handle =
-    GNUNET_CLIENT_notify_transmit_ready (dh->dns_connection,
-                                         ntohs (dh->rq_head->msg->size),
-                                         GNUNET_TIME_UNIT_FOREVER_REL,
-                                         GNUNET_NO,
-                                         &send_response, dh);
+  dh->generation++;
+  env = GNUNET_MQ_msg (msg,
+                       GNUNET_MESSAGE_TYPE_DNS_CLIENT_INIT);
+  msg->flags = htonl (dh->flags);
+  GNUNET_MQ_send (dh->mq,
+                  env);
 }
 
 
@@ -376,7 +246,7 @@ queue_reply (struct GNUNET_DNS_Handle *dh,
 void
 GNUNET_DNS_request_forward (struct GNUNET_DNS_RequestHandle *rh)
 {
-  struct ReplyQueueEntry *qe;
+  struct GNUNET_MQ_Envelope *env;
   struct GNUNET_DNS_Response *resp;
 
   GNUNET_assert (0 < rh->dh->pending_requests--);
@@ -385,15 +255,12 @@ GNUNET_DNS_request_forward (struct GNUNET_DNS_RequestHandle *rh)
     GNUNET_free (rh);
     return;
   }
-  qe = GNUNET_malloc (sizeof (struct ReplyQueueEntry) +
-                      sizeof (struct GNUNET_DNS_Response));
-  resp = (struct GNUNET_DNS_Response*) &qe[1];
-  qe->msg = &resp->header;
-  resp->header.size = htons (sizeof (struct GNUNET_DNS_Response));
-  resp->header.type = htons (GNUNET_MESSAGE_TYPE_DNS_CLIENT_RESPONSE);
+  env = GNUNET_MQ_msg (resp,
+                       GNUNET_MESSAGE_TYPE_DNS_CLIENT_RESPONSE);
   resp->drop_flag = htonl (1);
   resp->request_id = rh->request_id;
-  queue_reply (rh->dh, qe);
+  GNUNET_MQ_send (rh->dh->mq,
+                  env);
   GNUNET_free (rh);
 }
 
@@ -407,7 +274,7 @@ GNUNET_DNS_request_forward (struct GNUNET_DNS_RequestHandle *rh)
 void
 GNUNET_DNS_request_drop (struct GNUNET_DNS_RequestHandle *rh)
 {
-  struct ReplyQueueEntry *qe;
+  struct GNUNET_MQ_Envelope *env;
   struct GNUNET_DNS_Response *resp;
 
   GNUNET_assert (0 < rh->dh->pending_requests--);
@@ -416,15 +283,12 @@ GNUNET_DNS_request_drop (struct GNUNET_DNS_RequestHandle *rh)
       GNUNET_free (rh);
       return;
   }
-  qe = GNUNET_malloc (sizeof (struct ReplyQueueEntry) +
-                      sizeof (struct GNUNET_DNS_Response));
-  resp = (struct GNUNET_DNS_Response*) &qe[1];
-  qe->msg = &resp->header;
-  resp->header.size = htons (sizeof (struct GNUNET_DNS_Response));
-  resp->header.type = htons (GNUNET_MESSAGE_TYPE_DNS_CLIENT_RESPONSE);
+  env = GNUNET_MQ_msg (resp,
+                       GNUNET_MESSAGE_TYPE_DNS_CLIENT_RESPONSE);
   resp->request_id = rh->request_id;
   resp->drop_flag = htonl (0);
-  queue_reply (rh->dh, qe);
+  GNUNET_MQ_send (rh->dh->mq,
+                  env);
   GNUNET_free (rh);
 }
 
@@ -435,7 +299,7 @@ GNUNET_DNS_request_drop (struct GNUNET_DNS_RequestHandle *rh)
  * the modifications the function might have made).
  *
  * @param rh request that should now be answered
- * @param reply_length size of reply (uint16_t to force sane size)
+ * @param reply_length size of @a reply (uint16_t to force sane size)
  * @param reply reply data
  */
 void
@@ -443,7 +307,7 @@ GNUNET_DNS_request_answer (struct GNUNET_DNS_RequestHandle *rh,
                            uint16_t reply_length,
                            const char *reply)
 {
-  struct ReplyQueueEntry *qe;
+  struct GNUNET_MQ_Envelope *env;
   struct GNUNET_DNS_Response *resp;
 
   GNUNET_assert (0 < rh->dh->pending_requests--);
@@ -452,22 +316,23 @@ GNUNET_DNS_request_answer (struct GNUNET_DNS_RequestHandle *rh,
       GNUNET_free (rh);
       return;
   }
-  if (reply_length + sizeof (struct GNUNET_DNS_Response) >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
+  if (reply_length + sizeof (struct GNUNET_DNS_Response)
+      >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
   {
     GNUNET_break (0);
     GNUNET_free (rh);
     return;
   }
-  qe = GNUNET_malloc (sizeof (struct ReplyQueueEntry) +
-                      sizeof (struct GNUNET_DNS_Response) + reply_length);
-  resp = (struct GNUNET_DNS_Response*) &qe[1];
-  qe->msg = &resp->header;
-  resp->header.size = htons (sizeof (struct GNUNET_DNS_Response) + reply_length);
-  resp->header.type = htons (GNUNET_MESSAGE_TYPE_DNS_CLIENT_RESPONSE);
+  env = GNUNET_MQ_msg_extra (resp,
+                             reply_length,
+                             GNUNET_MESSAGE_TYPE_DNS_CLIENT_RESPONSE);
   resp->drop_flag = htonl (2);
   resp->request_id = rh->request_id;
-  memcpy (&resp[1], reply, reply_length);
-  queue_reply (rh->dh, qe);
+  memcpy (&resp[1],
+          reply,
+          reply_length);
+  GNUNET_MQ_send (rh->dh->mq,
+                  env);
   GNUNET_free (rh);
 }
 
@@ -476,9 +341,9 @@ GNUNET_DNS_request_answer (struct GNUNET_DNS_RequestHandle *rh,
  * Connect to the service-dns
  *
  * @param cfg configuration to use
- * @param flags when to call rh
+ * @param flags when to call @a rh
  * @param rh function to call with DNS requests
- * @param rh_cls closure to pass to rh
+ * @param rh_cls closure to pass to @a rh
  * @return DNS handle
  */
 struct GNUNET_DNS_Handle *
@@ -507,15 +372,19 @@ GNUNET_DNS_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
 void
 GNUNET_DNS_disconnect (struct GNUNET_DNS_Handle *dh)
 {
+  if (NULL != dh->mq)
+  {
+    GNUNET_MQ_destroy (dh->mq);
+    dh->mq = NULL;
+  }
   if (NULL != dh->reconnect_task)
   {
     GNUNET_SCHEDULER_cancel (dh->reconnect_task);
     dh->reconnect_task = NULL;
   }
-  disconnect (dh);
   /* make sure client has no pending requests left over! */
-  GNUNET_assert (0 == dh->pending_requests);
+  GNUNET_break (0 == dh->pending_requests);
   GNUNET_free (dh);
 }
 
-/* end of dns_api_new.c */
+/* end of dns_api.c */