towards adding mq destruction notification
[oweals/gnunet.git] / src / util / mq.c
index d4c5b164419ff6039e1dba3257ae68b75e5a372c..01cdf764b90aa9c4e8b285d5015c92e256155c95 100644 (file)
@@ -63,6 +63,26 @@ struct GNUNET_MQ_Envelope
    * Closure for @e send_cb
    */
   void *sent_cls;
+
+  /**
+   * Flags that were set for this envelope by
+   * #GNUNET_MQ_env_set_options().   Only valid if
+   * @e have_custom_options is set.
+   */
+  uint64_t flags;
+
+  /**
+   * Additional options buffer set for this envelope by
+   * #GNUNET_MQ_env_set_options().  Only valid if
+   * @e have_custom_options is set.
+   */
+  const void *extra;
+
+  /**
+   * Did the application call #GNUNET_MQ_env_set_options()?
+   */
+  int have_custom_options;
+
 };
 
 
@@ -134,6 +154,28 @@ struct GNUNET_MQ_Handle
    */
   struct GNUNET_SCHEDULER_Task *continue_task;
 
+  /**
+   * Functions to call on queue destruction; kept in a DLL.
+   */
+  struct GNUNET_MQ_DestroyNotificationHandle *dnh_head;
+
+  /**
+   * Functions to call on queue destruction; kept in a DLL.
+   */
+  struct GNUNET_MQ_DestroyNotificationHandle *dnh_tail;
+
+  /**
+   * Additional options buffer set for this queue by
+   * #GNUNET_MQ_set_options().  Default is 0.
+   */
+  const void *default_extra;
+
+  /**
+   * Flags that were set for this queue by
+   * #GNUNET_MQ_set_options().   Default is 0.
+   */
+  uint64_t default_flags;
+
   /**
    * Next id that should be used for the @e assoc_map,
    * initialized lazily to a random value together with
@@ -162,7 +204,7 @@ struct ServerClientSocketState
   /**
    * Active transmission request to the client.
    */
-  struct GNUNET_SERVER_TransmitHandleth;
+  struct GNUNET_SERVER_TransmitHandle *th;
 };
 
 
@@ -253,7 +295,7 @@ GNUNET_MQ_inject_message (struct GNUNET_MQ_Handle *mq,
   }
  done:
   if (GNUNET_NO == handled)
-    LOG (GNUNET_ERROR_TYPE_WARNING,
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
          "No handler for message of type %d\n",
          ntohs (mh->type));
 }
@@ -327,6 +369,7 @@ GNUNET_MQ_send (struct GNUNET_MQ_Handle *mq,
   GNUNET_assert (NULL != mq);
   GNUNET_assert (NULL == ev->parent_queue);
 
+  mq->queue_length++;
   ev->parent_queue = mq;
   /* is the implementation busy? queue it! */
   if (NULL != mq->current_envelope)
@@ -334,11 +377,12 @@ GNUNET_MQ_send (struct GNUNET_MQ_Handle *mq,
     GNUNET_CONTAINER_DLL_insert_tail (mq->envelope_head,
                                       mq->envelope_tail,
                                       ev);
-    mq->queue_length++;
     return;
   }
   mq->current_envelope = ev;
-  mq->send_impl (mq, ev->mh, mq->impl_state);
+  mq->send_impl (mq,
+                ev->mh,
+                mq->impl_state);
 }
 
 
@@ -362,7 +406,7 @@ GNUNET_MQ_send_copy (struct GNUNET_MQ_Handle *mq,
   env->mh = (struct GNUNET_MessageHeader *) &env[1];
   env->sent_cb = ev->sent_cb;
   env->sent_cls = ev->sent_cls;
-  memcpy (&env[1],
+  GNUNET_memcpy (&env[1],
           ev->mh,
           msize);
   GNUNET_MQ_send (mq,
@@ -390,6 +434,8 @@ impl_send_continue (void *cls)
   current_envelope = mq->current_envelope;
   GNUNET_assert (NULL != current_envelope);
   current_envelope->parent_queue = NULL;
+  GNUNET_assert (0 < mq->queue_length);
+  mq->queue_length--;
   if (NULL == mq->envelope_head)
   {
     mq->current_envelope = NULL;
@@ -400,7 +446,6 @@ impl_send_continue (void *cls)
     GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
                                  mq->envelope_tail,
                                  mq->current_envelope);
-    mq->queue_length--;
     mq->send_impl (mq,
                   mq->current_envelope->mh,
                   mq->impl_state);
@@ -460,7 +505,7 @@ GNUNET_MQ_queue_for_callbacks (GNUNET_MQ_SendImpl send,
     for (i=0;NULL != handlers[i].cb; i++) ;
     mq->handlers = GNUNET_new_array (i + 1,
                                     struct GNUNET_MQ_MessageHandler);
-    memcpy (mq->handlers,
+    GNUNET_memcpy (mq->handlers,
            handlers,
            i * sizeof (struct GNUNET_MQ_MessageHandler));
   }
@@ -472,6 +517,26 @@ GNUNET_MQ_queue_for_callbacks (GNUNET_MQ_SendImpl send,
 }
 
 
+/**
+ * Change the closure argument in all of the `handlers` of the
+ * @a mq.
+ *
+ * @param mq to modify
+ * @param handlers_cls new closure to use
+ */
+void
+GNUNET_MQ_set_handlers_closure (struct GNUNET_MQ_Handle *mq,
+                                void *handlers_cls)
+{
+  unsigned int i;
+
+  if (NULL == mq->handlers)
+    return;
+  for (i=0;NULL != mq->handlers[i].cb; i++)
+    mq->handlers[i].cls = handlers_cls;
+}
+
+
 /**
  * Get the message that should currently be sent.
  * Fails if there is no current message.
@@ -544,7 +609,7 @@ GNUNET_MQ_msg_copy (const struct GNUNET_MessageHeader *hdr)
 
   mqm = GNUNET_malloc (sizeof (*mqm) + size);
   mqm->mh = (struct GNUNET_MessageHeader *) &mqm[1];
-  memcpy (mqm->mh,
+  GNUNET_memcpy (mqm->mh,
           hdr,
           size);
   return mqm;
@@ -579,9 +644,9 @@ GNUNET_MQ_msg_nested_mh_ (struct GNUNET_MessageHeader **mhp,
     return NULL;
 
   mqm = GNUNET_MQ_msg_ (mhp, size, type);
-  memcpy ((char *) mqm->mh + base_size,
-         nested_mh,
-         ntohs (nested_mh->size));
+  GNUNET_memcpy ((char *) mqm->mh + base_size,
+                nested_mh,
+                ntohs (nested_mh->size));
 
   return mqm;
 }
@@ -608,7 +673,7 @@ transmit_queued (void *cls,
   GNUNET_assert (NULL != buf);
   msg_size = ntohs (msg->size);
   GNUNET_assert (size >= msg_size);
-  memcpy (buf, msg, msg_size);
+  GNUNET_memcpy (buf, msg, msg_size);
   state->th = NULL;
 
   GNUNET_MQ_impl_send_continue (mq);
@@ -644,7 +709,6 @@ server_client_send_impl (struct GNUNET_MQ_Handle *mq,
   struct ServerClientSocketState *state = impl_state;
 
   GNUNET_assert (NULL != mq);
-  GNUNET_assert (NULL != state);
   state->th = GNUNET_SERVER_notify_transmit_ready (state->client,
                                                   ntohs (msg->size),
                                                   GNUNET_TIME_UNIT_FOREVER_REL,
@@ -663,8 +727,8 @@ GNUNET_MQ_queue_for_server_client (struct GNUNET_SERVER_Client *client)
   mq->impl_state = scss;
   scss->client = client;
   GNUNET_SERVER_client_keep (client);
-  mq->send_impl = server_client_send_impl;
-  mq->destroy_impl = server_client_destroy_impl;
+  mq->send_impl = &server_client_send_impl;
+  mq->destroy_impl = &server_client_destroy_impl;
   return mq;
 }
 
@@ -737,7 +801,7 @@ connection_client_transmit_queued (void *cls,
 
   msg_size = ntohs (msg->size);
   GNUNET_assert (size >= msg_size);
-  memcpy (buf, msg, msg_size);
+  GNUNET_memcpy (buf, msg, msg_size);
   state->th = NULL;
 
   GNUNET_MQ_impl_send_continue (mq);
@@ -788,9 +852,18 @@ connection_client_cancel_impl (struct GNUNET_MQ_Handle *mq,
 {
   struct ClientConnectionState *state = impl_state;
 
-  GNUNET_assert (NULL != state->th);
-  GNUNET_CLIENT_notify_transmit_ready_cancel (state->th);
-  state->th = NULL;
+  if (NULL != state->th)
+  {
+    GNUNET_CLIENT_notify_transmit_ready_cancel (state->th);
+    state->th = NULL;
+  }
+  else if (NULL != mq->continue_task)
+  {
+    GNUNET_SCHEDULER_cancel (mq->continue_task);
+    mq->continue_task = NULL;
+  }
+  else
+    GNUNET_assert (0);
 }
 
 
@@ -808,11 +881,11 @@ GNUNET_MQ_queue_for_connection_client (struct GNUNET_CLIENT_Connection *connecti
   if (NULL != handlers)
   {
     for (i=0;NULL != handlers[i].cb; i++) ;
-    mq->handlers = GNUNET_new_array (i,
+    mq->handlers = GNUNET_new_array (i + 1,
                                     struct GNUNET_MQ_MessageHandler);
-    memcpy (mq->handlers,
-           handlers,
-           i * sizeof (struct GNUNET_MQ_MessageHandler));
+    GNUNET_memcpy (mq->handlers,
+                   handlers,
+                   i * sizeof (struct GNUNET_MQ_MessageHandler));
   }
   mq->error_handler = error_handler;
   mq->error_handler_cls = error_handler_cls;
@@ -913,7 +986,6 @@ GNUNET_MQ_destroy (struct GNUNET_MQ_Handle *mq)
     mq->queue_length--;
     GNUNET_MQ_discard (ev);
   }
-  GNUNET_assert (0 == mq->queue_length);
   if (NULL != mq->current_envelope)
   {
     /* we can only discard envelopes that
@@ -921,7 +993,9 @@ GNUNET_MQ_destroy (struct GNUNET_MQ_Handle *mq)
     mq->current_envelope->parent_queue = NULL;
     GNUNET_MQ_discard (mq->current_envelope);
     mq->current_envelope = NULL;
+    mq->queue_length--;
   }
+  GNUNET_assert (0 == mq->queue_length);
   if (NULL != mq->assoc_map)
   {
     GNUNET_CONTAINER_multihashmap32_destroy (mq->assoc_map);
@@ -979,6 +1053,7 @@ GNUNET_MQ_send_cancel (struct GNUNET_MQ_Envelope *ev)
   {
     // complex case, we already started with transmitting
     // the message
+    mq->queue_length--;
     mq->cancel_impl (mq,
                     mq->impl_state);
     // continue sending the next message, if any
@@ -1012,4 +1087,177 @@ GNUNET_MQ_send_cancel (struct GNUNET_MQ_Envelope *ev)
   GNUNET_free (ev);
 }
 
+
+/**
+ * Function to obtain the current envelope
+ * from within #GNUNET_MQ_SendImpl implementations.
+ *
+ * @param mq message queue to interrogate
+ * @return the current envelope
+ */
+struct GNUNET_MQ_Envelope *
+GNUNET_MQ_get_current_envelope (struct GNUNET_MQ_Handle *mq)
+{
+  return mq->current_envelope;
+}
+
+
+/**
+ * Function to obtain the last envelope in the queue.
+ *
+ * @param mq message queue to interrogate
+ * @return the last envelope in the queue
+ */
+struct GNUNET_MQ_Envelope *
+GNUNET_MQ_get_last_envelope (struct GNUNET_MQ_Handle *mq)
+{
+  if (NULL != mq->envelope_tail)
+    return mq->envelope_tail;
+
+  return mq->current_envelope;
+}
+
+
+/**
+ * Set application-specific options for this envelope.
+ * Overrides the options set for the queue with
+ * #GNUNET_MQ_set_options() for this message only.
+ *
+ * @param env message to set options for
+ * @param flags flags to use (meaning is queue-specific)
+ * @param extra additional buffer for further data (also queue-specific)
+ */
+void
+GNUNET_MQ_env_set_options (struct GNUNET_MQ_Envelope *env,
+                          uint64_t flags,
+                          const void *extra)
+{
+  env->flags = flags;
+  env->extra = extra;
+  env->have_custom_options = GNUNET_YES;
+}
+
+
+/**
+ * Get application-specific options for this envelope.
+ *
+ * @param env message to set options for
+ * @param[out] flags set to flags to use (meaning is queue-specific)
+ * @return extra additional buffer for further data (also queue-specific)
+ */
+const void *
+GNUNET_MQ_env_get_options (struct GNUNET_MQ_Envelope *env,
+                          uint64_t *flags)
+{
+  struct GNUNET_MQ_Handle *mq = env->parent_queue;
+
+  if (GNUNET_YES == env->have_custom_options)
+  {
+    *flags = env->flags;
+    return env->extra;
+  }
+  if (NULL == mq)
+  {
+    *flags = 0;
+    return NULL;
+  }
+  *flags = mq->default_flags;
+  return mq->default_extra;
+}
+
+
+/**
+ * Set application-specific options for this queue.
+ *
+ * @param mq message queue to set options for
+ * @param flags flags to use (meaning is queue-specific)
+ * @param extra additional buffer for further data (also queue-specific)
+ */
+void
+GNUNET_MQ_set_options (struct GNUNET_MQ_Handle *mq,
+                      uint64_t flags,
+                      const void *extra)
+{
+  mq->default_flags = flags;
+  mq->default_extra = extra;
+}
+
+
+/**
+ * Handle we return for callbacks registered to be
+ * notified when #GNUNET_MQ_destroy() is called on a queue.
+ */
+struct GNUNET_MQ_DestroyNotificationHandle
+{
+  /**
+   * Kept in a DLL.
+   */
+  struct GNUNET_MQ_DestroyNotificationHandle *prev;
+
+  /**
+   * Kept in a DLL.
+   */
+  struct GNUNET_MQ_DestroyNotificationHandle *next;
+
+  /**
+   * Queue to notify about.
+   */
+  struct GNUNET_MQ_Handle *mq;
+
+  /**
+   * Function to call.
+   */
+  GNUNET_SCHEDULER_TaskCallback cb;
+
+  /**
+   * Closure for @e cb.
+   */
+  void *cb_cls;
+};
+
+
+/**
+ * Register function to be called whenever @a mq is being
+ * destroyed.
+ *
+ * @param mq message queue to watch
+ * @param cb function to call on @a mq destruction
+ * @param cb_cls closure for @a cb
+ * @return handle for #GNUNET_MQ_destroy_notify_cancel().
+ */
+struct GNUNET_MQ_DestroyNotificationHandle *
+GNUNET_MQ_destroy_notify (struct GNUNET_MQ_Handle *mq,
+                         GNUNET_SCHEDULER_TaskCallback cb,
+                         void *cb_cls)
+{
+  struct GNUNET_MQ_DestroyNotificationHandle *dnh;
+
+  dnh = GNUNET_new (struct GNUNET_MQ_DestroyNotificationHandle);
+  dnh->mq = mq;
+  dnh->cb = cb;
+  dnh->cb_cls = cb_cls;
+  GNUNET_CONTAINER_DLL_insert (mq->dnh_head,
+                              mq->dnh_tail,
+                              dnh);
+  return dnh;
+}
+
+
+/**
+ * Cancel registration from #GNUNET_MQ_destroy_notify().
+ *
+ * @param dnh handle for registration to cancel
+ */
+void
+GNUNET_MQ_destroy_notify_cancel (struct GNUNET_MQ_DestroyNotificationHandle *dnh)
+{
+  struct GNUNET_MQ_Handle *mq = dnh->mq;
+  
+  GNUNET_CONTAINER_DLL_remove (mq->dnh_head,
+                              mq->dnh_tail,
+                              dnh);
+  GNUNET_free (dnh);
+}
+
+
 /* end of mq.c */