[GNUnet-SVN] r38160 - in gnunet/src: include util

[gnunet]

Author: dold
Date: 2016-10-17 00:07:40 +0200 (Mon, 17 Oct 2016)
New Revision: 38160

Modified:
   gnunet/src/include/gnunet_mq_lib.h
   gnunet/src/util/mq.c
Log:
implement impl_in_flight API for MQ, replacing evacuation


Modified: gnunet/src/include/gnunet_mq_lib.h
===================================================================
--- gnunet/src/include/gnunet_mq_lib.h  2016-10-16 20:12:00 UTC (rev 38159)
+++ gnunet/src/include/gnunet_mq_lib.h  2016-10-16 22:07:40 UTC (rev 38160)
@@ -702,6 +702,9 @@
 
 /**
  * Call the send implementation for the next queued message, if any.
+ * Calls the send notification for the current message unless
+ * #GNUNET_MQ_impl_send_in_flight was called for this envelope.
+ *
  * Only useful for implementing message queues, results in undefined
  * behavior if not used carefully.
  *
@@ -712,35 +715,17 @@
 
 
 /**
- * Get the message that should currently be sent.  The returned
- * message is only valid until #GNUNET_MQ_impl_send_continue is
- * called.  Fails if there is no current message.  Only useful for
- * implementing message queues, results in undefined behavior if not
- * used carefully.
+ * Call the send notification for the current message, but do not
+ * try to send the message until #gnunet_mq_impl_send_continue
+ * is called.
  *
- * @param mq message queue with the current message, only valid
- *        until #GNUNET_MQ_impl_send_continue() is called
- * @return message to send, never NULL
- */
-const struct GNUNET_MessageHeader *
-GNUNET_MQ_impl_current (struct GNUNET_MQ_Handle *mq);
-
-
-/**
- * Get the message that is currently being sent when cancellation of that
- * message is requested.  The returned buffer must be freed by the caller.
+ * only useful for implementing message queues, results in undefined
+ * behavior if not used carefully.
  *
- * This function may be called at most once in the cancel_impl
- * function of a message queue.
- *
- * Use this function to avoid copying a half-sent message.
- *
- * @param mq message queue
- * @return pointer to store the message being canceled,
- *         must be freed by the caller
+ * @param mq message queue to send the next message with
  */
-struct GNUNET_MessageHeader *
-GNUNET_MQ_impl_cancel_evacuate (struct GNUNET_MQ_Handle *mq);
+void
+GNUNET_MQ_impl_send_in_flight (struct GNUNET_MQ_Handle *mq);
 
 
 /**

Modified: gnunet/src/util/mq.c
===================================================================
--- gnunet/src/util/mq.c        2016-10-16 20:12:00 UTC (rev 38159)
+++ gnunet/src/util/mq.c        2016-10-16 22:07:40 UTC (rev 38160)
@@ -145,14 +145,21 @@
   struct GNUNET_MQ_Envelope *current_envelope;
 
   /**
+   * GNUNET_YES if the sent notification was called 
+   * for the current envelope.
+   */
+  int send_notification_called;
+
+  /**
    * Map of associations, lazily allocated
    */
   struct GNUNET_CONTAINER_MultiHashMap32 *assoc_map;
 
   /**
-   * Task scheduled during #GNUNET_MQ_impl_send_continue.
+   * Task scheduled during #GNUNET_MQ_impl_send_continue
+   * or #GNUNET_MQ_impl_send_in_flight
    */
-  struct GNUNET_SCHEDULER_Task *continue_task;
+  struct GNUNET_SCHEDULER_Task *send_task;
 
   /**
    * Functions to call on queue destruction; kept in a DLL.
@@ -344,8 +351,7 @@
 GNUNET_MQ_discard (struct GNUNET_MQ_Envelope *ev)
 {
   GNUNET_assert (NULL == ev->parent_queue);
-  /* also frees ev */
-  GNUNET_free (ev->mh);
+  GNUNET_free (ev);
 }
 
 
@@ -421,7 +427,35 @@
 }
 
 
+/**
+ * Task run to call the send notification for the next queued
+ * message, if any.  Only useful for implementing message queues,
+ * results in undefined behavior if not used carefully.
+ *
+ * @param cls message queue to send the next message with
+ */
+static void
+impl_send_in_flight (void *cls)
+{
+  struct GNUNET_MQ_Handle *mq = cls;
+  struct GNUNET_MQ_Envelope *current_envelope;
 
+  mq->send_task = NULL;
+  /* call is only valid if we're actually currently sending
+   * a message */
+  current_envelope = mq->current_envelope;
+  GNUNET_assert (NULL != current_envelope);
+  /* can't call cancel from now on anymore */
+  current_envelope->parent_queue = NULL;
+  if ( (GNUNET_NO == mq->send_notification_called) &&
+       (NULL != current_envelope->sent_cb) )
+  {
+    current_envelope->sent_cb (current_envelope->sent_cls);
+  }
+  mq->send_notification_called = GNUNET_YES;
+}
+
+
 /**
  * Task run to call the send implementation for the next queued
  * message, if any.  Only useful for implementing message queues,
@@ -435,12 +469,12 @@
   struct GNUNET_MQ_Handle *mq = cls;
   struct GNUNET_MQ_Envelope *current_envelope;
 
-  mq->continue_task = NULL;
+  mq->send_task = NULL;
   /* call is only valid if we're actually currently sending
    * a message */
   current_envelope = mq->current_envelope;
   GNUNET_assert (NULL != current_envelope);
-  current_envelope->parent_queue = NULL;
+  impl_send_in_flight (mq);
   GNUNET_assert (0 < mq->queue_length);
   mq->queue_length--;
   if (NULL == mq->envelope_head)
@@ -453,14 +487,12 @@
     GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
                                  mq->envelope_tail,
                                  mq->current_envelope);
+    mq->send_notification_called = GNUNET_NO;
     mq->send_impl (mq,
                   mq->current_envelope->mh,
                   mq->impl_state);
   }
-  if (NULL != current_envelope->sent_cb)
-    current_envelope->sent_cb (current_envelope->sent_cls);
-  /* also frees current_envelope */
-  GNUNET_free (current_envelope->mh);
+  GNUNET_free (current_envelope);
 }
 
 
@@ -474,13 +506,36 @@
 void
 GNUNET_MQ_impl_send_continue (struct GNUNET_MQ_Handle *mq)
 {
-  GNUNET_assert (NULL == mq->continue_task);
-  mq->continue_task = GNUNET_SCHEDULER_add_now (&impl_send_continue,
-                                                mq);
+  /* maybe #GNUNET_MQ_impl_send_in_flight was called? */
+  if (NULL != mq->send_task)
+  {
+    GNUNET_SCHEDULER_cancel (mq->send_task);
+  }
+  mq->send_task = GNUNET_SCHEDULER_add_now (&impl_send_continue,
+                                            mq);
 }
 
 
 /**
+ * Call the send notification for the current message, but do not
+ * try to send the message until #gnunet_mq_impl_send_continue
+ * is called.
+ *
+ * only useful for implementing message queues, results in undefined
+ * behavior if not used carefully.
+ *
+ * @param mq message queue to send the next message with
+ */
+void
+GNUNET_MQ_impl_send_in_flight (struct GNUNET_MQ_Handle *mq)
+{
+  GNUNET_assert (NULL == mq->send_task);
+  mq->send_task = GNUNET_SCHEDULER_add_now (&impl_send_in_flight,
+                                            mq);
+}
+
+
+/**
  * Create a message queue for the specified handlers.
  *
  * @param send function the implements sending messages
@@ -592,11 +647,9 @@
                 uint16_t type)
 {
   struct GNUNET_MQ_Envelope *ev;
-  void *mem;
 
-  mem = GNUNET_malloc (size + sizeof (struct GNUNET_MQ_Envelope));
-  ev = mem + size;
-  ev->mh = mem;
+  ev = GNUNET_malloc (size + sizeof (struct GNUNET_MQ_Envelope));
+  ev->mh = (struct GNUNET_MessageHeader *) &ev[1];
   ev->mh->size = htons (size);
   ev->mh->type = htons (type);
   if (NULL != mhp)
@@ -867,10 +920,10 @@
     GNUNET_CLIENT_notify_transmit_ready_cancel (state->th);
     state->th = NULL;
   }
-  else if (NULL != mq->continue_task)
+  else if (NULL != mq->send_task)
   {
-    GNUNET_SCHEDULER_cancel (mq->continue_task);
-    mq->continue_task = NULL;
+    GNUNET_SCHEDULER_cancel (mq->send_task);
+    mq->send_task = NULL;
   }
   else
     GNUNET_assert (0);
@@ -1028,10 +1081,10 @@
   {
     mq->destroy_impl (mq, mq->impl_state);
   }
-  if (NULL != mq->continue_task)
+  if (NULL != mq->send_task)
   {
-    GNUNET_SCHEDULER_cancel (mq->continue_task);
-    mq->continue_task = NULL;
+    GNUNET_SCHEDULER_cancel (mq->send_task);
+    mq->send_task = NULL;
   }
   while (NULL != mq->envelope_head)
   {
@@ -1136,6 +1189,7 @@
       GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
                                    mq->envelope_tail,
                                    mq->current_envelope);
+      mq->send_notification_called = GNUNET_NO;
       mq->send_impl (mq,
                     mq->current_envelope->mh,
                     mq->impl_state);
@@ -1154,8 +1208,9 @@
   if (GNUNET_YES != mq->evacuate_called)
   {
     ev->parent_queue = NULL;
+    ev->mh = NULL;
     /* also frees ev */
-    GNUNET_free (ev->mh);
+    GNUNET_free (ev);
   }
 }
 
@@ -1299,34 +1354,4 @@
 }
 
 
-/**
- * Get the message that is currently being sent when cancellation of that
- * message is requested.  The returned buffer must be freed by the caller.
- *
- * This function may be called at most once in the cancel_impl
- * function of a message queue.
- *
- * Use this function to avoid copying a half-sent message.
- *
- * @param mq message queue
- * @return pointer to store the message being canceled,
- *         must be freed by the caller
- */
-struct GNUNET_MessageHeader *
-GNUNET_MQ_impl_cancel_evacuate (struct GNUNET_MQ_Handle *mq)
-{
-  struct GNUNET_MessageHeader *mh;
-
-  GNUNET_assert (GNUNET_NO == mq->evacuate_called);
-  GNUNET_assert (NULL != mq->current_envelope);
-
-  mq->evacuate_called = GNUNET_YES;
-  mh = mq->current_envelope->mh;
-  mq->current_envelope->parent_queue = NULL;
-  mq->current_envelope = NULL;
-
-  return mh;
-}
-
-
 /* end of mq.c */