Merge branch 'master' of gnunet.org:gnunet
[oweals/gnunet.git] / src / util / mq.c
1 /*
2      This file is part of GNUnet.
3      Copyright (C) 2012-2017 GNUnet e.V.
4
5      GNUnet is free software: you can redistribute it and/or modify it
6      under the terms of the GNU Affero General Public License as published
7      by the Free Software Foundation, either version 3 of the License,
8      or (at your option) any later version.
9
10      GNUnet is distributed in the hope that it will be useful, but
11      WITHOUT ANY WARRANTY; without even the implied warranty of
12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13      Affero General Public License for more details.
14
15      You should have received a copy of the GNU Affero General Public License
16      along with this program.  If not, see <http://www.gnu.org/licenses/>.
17
18      SPDX-License-Identifier: AGPL3.0-or-later
19 */
20
21 /**
22  * @author Florian Dold
23  * @file util/mq.c
24  * @brief general purpose request queue
25  */
26 #include "platform.h"
27 #include "gnunet_util_lib.h"
28
29 #define LOG(kind,...) GNUNET_log_from (kind, "util-mq",__VA_ARGS__)
30
31
32 struct GNUNET_MQ_Envelope
33 {
34   /**
35    * Messages are stored in a linked list.
36    * Each queue has its own list of envelopes.
37    */
38   struct GNUNET_MQ_Envelope *next;
39
40   /**
41    * Messages are stored in a linked list
42    * Each queue has its own list of envelopes.
43    */
44   struct GNUNET_MQ_Envelope *prev;
45
46   /**
47    * Actual allocated message header.
48    * The GNUNET_MQ_Envelope header is allocated at
49    * the end of the message.
50    */
51   struct GNUNET_MessageHeader *mh;
52
53   /**
54    * Queue the message is queued in, NULL if message is not queued.
55    */
56   struct GNUNET_MQ_Handle *parent_queue;
57
58   /**
59    * Called after the message was sent irrevocably.
60    */
61   GNUNET_SCHEDULER_TaskCallback sent_cb;
62
63   /**
64    * Closure for @e send_cb
65    */
66   void *sent_cls;
67
68   /**
69    * Flags that were set for this envelope by
70    * #GNUNET_MQ_env_set_options().   Only valid if
71    * @e have_custom_options is set.
72    */
73   uint64_t flags;
74
75   /**
76    * Additional options buffer set for this envelope by
77    * #GNUNET_MQ_env_set_options().  Only valid if
78    * @e have_custom_options is set.
79    */
80   const void *extra;
81
82   /**
83    * Did the application call #GNUNET_MQ_env_set_options()?
84    */
85   int have_custom_options;
86 };
87
88
89 /**
90  * Handle to a message queue.
91  */
92 struct GNUNET_MQ_Handle
93 {
94   /**
95    * Handlers array, or NULL if the queue should not receive messages
96    */
97   struct GNUNET_MQ_MessageHandler *handlers;
98
99   /**
100    * Actual implementation of message sending,
101    * called when a message is added
102    */
103   GNUNET_MQ_SendImpl send_impl;
104
105   /**
106    * Implementation-dependent queue destruction function
107    */
108   GNUNET_MQ_DestroyImpl destroy_impl;
109
110   /**
111    * Implementation-dependent send cancel function
112    */
113   GNUNET_MQ_CancelImpl cancel_impl;
114
115   /**
116    * Implementation-specific state
117    */
118   void *impl_state;
119
120   /**
121    * Callback will be called when an error occurs.
122    */
123   GNUNET_MQ_ErrorHandler error_handler;
124
125   /**
126    * Closure for the error handler.
127    */
128   void *error_handler_cls;
129
130   /**
131    * Task to asynchronously run #impl_send_continue().
132    */
133   struct GNUNET_SCHEDULER_Task *send_task;
134
135   /**
136    * Linked list of messages pending to be sent
137    */
138   struct GNUNET_MQ_Envelope *envelope_head;
139
140   /**
141    * Linked list of messages pending to be sent
142    */
143   struct GNUNET_MQ_Envelope *envelope_tail;
144
145   /**
146    * Message that is currently scheduled to be
147    * sent. Not the head of the message queue, as the implementation
148    * needs to know if sending has been already scheduled or not.
149    */
150   struct GNUNET_MQ_Envelope *current_envelope;
151
152   /**
153    * Map of associations, lazily allocated
154    */
155   struct GNUNET_CONTAINER_MultiHashMap32 *assoc_map;
156
157   /**
158    * Functions to call on queue destruction; kept in a DLL.
159    */
160   struct GNUNET_MQ_DestroyNotificationHandle *dnh_head;
161
162   /**
163    * Functions to call on queue destruction; kept in a DLL.
164    */
165   struct GNUNET_MQ_DestroyNotificationHandle *dnh_tail;
166
167   /**
168    * Additional options buffer set for this queue by
169    * #GNUNET_MQ_set_options().  Default is 0.
170    */
171   const void *default_extra;
172
173   /**
174    * Flags that were set for this queue by
175    * #GNUNET_MQ_set_options().   Default is 0.
176    */
177   uint64_t default_flags;
178
179   /**
180    * Next id that should be used for the @e assoc_map,
181    * initialized lazily to a random value together with
182    * @e assoc_map
183    */
184   uint32_t assoc_id;
185
186   /**
187    * Number of entries we have in the envelope-DLL.
188    */
189   unsigned int queue_length;
190
191   /**
192    * #GNUNET_YES if GNUNET_MQ_impl_evacuate was called.
193    * FIXME: is this dead?
194    */
195   int evacuate_called;
196
197   /**
198    * #GNUNET_YES if GNUNET_MQ_impl_send_in_flight() was called.
199    */
200   int in_flight;
201 };
202
203
204 /**
205  * Call the message message handler that was registered
206  * for the type of the given message in the given message queue.
207  *
208  * This function is indended to be used for the implementation
209  * of message queues.
210  *
211  * @param mq message queue with the handlers
212  * @param mh message to dispatch
213  */
214 void
215 GNUNET_MQ_inject_message (struct GNUNET_MQ_Handle *mq,
216                           const struct GNUNET_MessageHeader *mh)
217 {
218   int ret;
219
220   ret = GNUNET_MQ_handle_message (mq->handlers,
221                                   mh);
222   if (GNUNET_SYSERR == ret)
223   {
224     GNUNET_MQ_inject_error (mq,
225                             GNUNET_MQ_ERROR_MALFORMED);
226     return;
227   }
228 }
229
230
231 /**
232  * Call the message message handler that was registered
233  * for the type of the given message in the given @a handlers list.
234  *
235  * This function is indended to be used for the implementation
236  * of message queues.
237  *
238  * @param handlers a set of handlers
239  * @param mh message to dispatch
240  * @return #GNUNET_OK on success, #GNUNET_NO if no handler matched,
241  *         #GNUNET_SYSERR if message was rejected by check function
242  */
243 int
244 GNUNET_MQ_handle_message (const struct GNUNET_MQ_MessageHandler *handlers,
245                           const struct GNUNET_MessageHeader *mh)
246 {
247   const struct GNUNET_MQ_MessageHandler *handler;
248   int handled = GNUNET_NO;
249   uint16_t msize = ntohs (mh->size);
250   uint16_t mtype = ntohs (mh->type);
251
252   LOG (GNUNET_ERROR_TYPE_DEBUG,
253        "Received message of type %u and size %u\n",
254        mtype, msize);
255
256   if (NULL == handlers)
257     goto done;
258   for (handler = handlers; NULL != handler->cb; handler++)
259   {
260     if (handler->type == mtype)
261     {
262       handled = GNUNET_YES;
263       if ( (handler->expected_size > msize) ||
264            ( (handler->expected_size != msize) &&
265              (NULL == handler->mv) ) )
266       {
267         /* Too small, or not an exact size and
268            no 'mv' handler to check rest */
269         LOG (GNUNET_ERROR_TYPE_ERROR,
270              "Received malformed message of type %u\n",
271              (unsigned int) handler->type);
272         return GNUNET_SYSERR;
273       }
274       if ( (NULL == handler->mv) ||
275            (GNUNET_OK ==
276             handler->mv (handler->cls, mh)) )
277       {
278         /* message well-formed, pass to handler */
279         handler->cb (handler->cls, mh);
280       }
281       else
282       {
283         /* Message rejected by check routine */
284         LOG (GNUNET_ERROR_TYPE_ERROR,
285              "Received malformed message of type %u\n",
286              (unsigned int) handler->type);
287         return GNUNET_SYSERR;
288       }
289       break;
290     }
291   }
292  done:
293   if (GNUNET_NO == handled)
294   {
295     LOG (GNUNET_ERROR_TYPE_INFO,
296          "No handler for message of type %u and size %u\n",
297          mtype, msize);
298     return GNUNET_NO;
299   }
300   return GNUNET_OK;
301 }
302
303
304 /**
305  * Call the error handler of a message queue with the given
306  * error code.  If there is no error handler, log a warning.
307  *
308  * This function is intended to be used by the implementation
309  * of message queues.
310  *
311  * @param mq message queue
312  * @param error the error type
313  */
314 void
315 GNUNET_MQ_inject_error (struct GNUNET_MQ_Handle *mq,
316                         enum GNUNET_MQ_Error error)
317 {
318   if (NULL == mq->error_handler)
319   {
320     LOG (GNUNET_ERROR_TYPE_WARNING,
321          "Got error %d, but no handler installed\n",
322          (int) error);
323     return;
324   }
325   mq->error_handler (mq->error_handler_cls,
326                      error);
327 }
328
329
330 /**
331  * Discard the message queue message, free all
332  * allocated resources. Must be called in the event
333  * that a message is created but should not actually be sent.
334  *
335  * @param mqm the message to discard
336  */
337 void
338 GNUNET_MQ_discard (struct GNUNET_MQ_Envelope *ev)
339 {
340   GNUNET_assert (NULL == ev->parent_queue);
341   GNUNET_free (ev);
342 }
343
344
345 /**
346  * Obtain the current length of the message queue.
347  *
348  * @param mq queue to inspect
349  * @return number of queued, non-transmitted messages
350  */
351 unsigned int
352 GNUNET_MQ_get_length (struct GNUNET_MQ_Handle *mq)
353 {
354   if (GNUNET_YES != mq->in_flight)
355   {
356     return mq->queue_length;
357   }
358   return mq->queue_length - 1;
359 }
360
361
362 /**
363  * Send a message with the given message queue.
364  * May only be called once per message.
365  *
366  * @param mq message queue
367  * @param ev the envelope with the message to send.
368  */
369 void
370 GNUNET_MQ_send (struct GNUNET_MQ_Handle *mq,
371                 struct GNUNET_MQ_Envelope *ev)
372 {
373   GNUNET_assert (NULL != mq);
374   GNUNET_assert (NULL == ev->parent_queue);
375
376   mq->queue_length++;
377   GNUNET_break (mq->queue_length < 10000); /* This would seem like a bug... */
378   ev->parent_queue = mq;
379   /* is the implementation busy? queue it! */
380   if ( (NULL != mq->current_envelope) ||
381        (NULL != mq->send_task) )
382   {
383     GNUNET_CONTAINER_DLL_insert_tail (mq->envelope_head,
384                                       mq->envelope_tail,
385                                       ev);
386     return;
387   }
388   GNUNET_assert (NULL == mq->envelope_head);
389   mq->current_envelope = ev;
390
391   LOG (GNUNET_ERROR_TYPE_DEBUG,
392        "sending message of type %u, queue empty (MQ: %p)\n",
393        ntohs(ev->mh->type),
394        mq);
395
396   mq->send_impl (mq,
397                  ev->mh,
398                  mq->impl_state);
399 }
400
401
402 /**
403  * Remove the first envelope that has not yet been sent from the message
404  * queue and return it.
405  *
406  * @param mq queue to remove envelope from
407  * @return NULL if queue is empty (or has no envelope that is not under transmission)
408  */
409 struct GNUNET_MQ_Envelope *
410 GNUNET_MQ_unsent_head (struct GNUNET_MQ_Handle *mq)
411 {
412   struct GNUNET_MQ_Envelope *env;
413
414   env = mq->envelope_head;
415   GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
416                                mq->envelope_tail,
417                                env);
418   mq->queue_length--;
419   env->parent_queue = NULL;
420   return env;
421 }
422
423
424 /**
425  * Function to copy an envelope.  The envelope must not yet
426  * be in any queue or have any options or callbacks set.
427  *
428  * @param env envelope to copy
429  * @return copy of @a env
430  */
431 struct GNUNET_MQ_Envelope *
432 GNUNET_MQ_env_copy (struct GNUNET_MQ_Envelope *env)
433 {
434   GNUNET_assert (NULL == env->next);
435   GNUNET_assert (NULL == env->parent_queue);
436   GNUNET_assert (NULL == env->sent_cb);
437   GNUNET_assert (GNUNET_NO == env->have_custom_options);
438   return GNUNET_MQ_msg_copy (env->mh);
439 }
440
441
442 /**
443  * Send a copy of a message with the given message queue.
444  * Can be called repeatedly on the same envelope.
445  *
446  * @param mq message queue
447  * @param ev the envelope with the message to send.
448  */
449 void
450 GNUNET_MQ_send_copy (struct GNUNET_MQ_Handle *mq,
451                      const struct GNUNET_MQ_Envelope *ev)
452 {
453   struct GNUNET_MQ_Envelope *env;
454   uint16_t msize;
455
456   msize = ntohs (ev->mh->size);
457   env = GNUNET_malloc (sizeof (struct GNUNET_MQ_Envelope) +
458                        msize);
459   env->mh = (struct GNUNET_MessageHeader *) &env[1];
460   env->sent_cb = ev->sent_cb;
461   env->sent_cls = ev->sent_cls;
462   GNUNET_memcpy (&env[1],
463           ev->mh,
464           msize);
465   GNUNET_MQ_send (mq,
466                   env);
467 }
468
469
470 /**
471  * Task run to call the send implementation for the next queued
472  * message, if any.  Only useful for implementing message queues,
473  * results in undefined behavior if not used carefully.
474  *
475  * @param cls message queue to send the next message with
476  */
477 static void
478 impl_send_continue (void *cls)
479 {
480   struct GNUNET_MQ_Handle *mq = cls;
481
482   mq->send_task = NULL;
483   /* call is only valid if we're actually currently sending
484    * a message */
485   if (NULL == mq->envelope_head)
486     return;
487   mq->current_envelope = mq->envelope_head;
488   GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
489                                mq->envelope_tail,
490                                mq->current_envelope);
491
492   LOG (GNUNET_ERROR_TYPE_DEBUG,
493        "sending message of type %u from queue\n",
494        ntohs(mq->current_envelope->mh->type));
495
496   mq->send_impl (mq,
497                  mq->current_envelope->mh,
498                  mq->impl_state);
499 }
500
501
502 /**
503  * Call the send implementation for the next queued message, if any.
504  * Only useful for implementing message queues, results in undefined
505  * behavior if not used carefully.
506  *
507  * @param mq message queue to send the next message with
508  */
509 void
510 GNUNET_MQ_impl_send_continue (struct GNUNET_MQ_Handle *mq)
511 {
512   struct GNUNET_MQ_Envelope *current_envelope;
513   GNUNET_SCHEDULER_TaskCallback cb;
514
515   GNUNET_assert (0 < mq->queue_length);
516   mq->queue_length--;
517   mq->in_flight = GNUNET_NO;
518   current_envelope = mq->current_envelope;
519   current_envelope->parent_queue = NULL;
520   mq->current_envelope = NULL;
521   GNUNET_assert (NULL == mq->send_task);
522   mq->send_task = GNUNET_SCHEDULER_add_now (&impl_send_continue,
523                                             mq);
524   if (NULL != (cb = current_envelope->sent_cb))
525   {
526     current_envelope->sent_cb = NULL;
527     cb (current_envelope->sent_cls);
528   }
529   GNUNET_free (current_envelope);
530 }
531
532
533 /**
534  * Call the send notification for the current message, but do not
535  * try to send the next message until #GNUNET_MQ_impl_send_continue
536  * is called.
537  *
538  * Only useful for implementing message queues, results in undefined
539  * behavior if not used carefully.
540  *
541  * @param mq message queue to send the next message with
542  */
543 void
544 GNUNET_MQ_impl_send_in_flight (struct GNUNET_MQ_Handle *mq)
545 {
546   struct GNUNET_MQ_Envelope *current_envelope;
547   GNUNET_SCHEDULER_TaskCallback cb;
548
549   mq->in_flight = GNUNET_YES;
550   /* call is only valid if we're actually currently sending
551    * a message */
552   current_envelope = mq->current_envelope;
553   GNUNET_assert (NULL != current_envelope);
554   /* can't call cancel from now on anymore */
555   current_envelope->parent_queue = NULL;
556   if (NULL != (cb = current_envelope->sent_cb))
557   {
558     current_envelope->sent_cb = NULL;
559     cb (current_envelope->sent_cls);
560   }
561 }
562
563
564 /**
565  * Create a message queue for the specified handlers.
566  *
567  * @param send function the implements sending messages
568  * @param destroy function that implements destroying the queue
569  * @param cancel function that implements canceling a message
570  * @param impl_state for the queue, passed to 'send' and 'destroy'
571  * @param handlers array of message handlers
572  * @param error_handler handler for read and write errors
573  * @param error_handler_cls closure for @a error_handler
574  * @return a new message queue
575  */
576 struct GNUNET_MQ_Handle *
577 GNUNET_MQ_queue_for_callbacks (GNUNET_MQ_SendImpl send,
578                                GNUNET_MQ_DestroyImpl destroy,
579                                GNUNET_MQ_CancelImpl cancel,
580                                void *impl_state,
581                                const struct GNUNET_MQ_MessageHandler *handlers,
582                                GNUNET_MQ_ErrorHandler error_handler,
583                                void *error_handler_cls)
584 {
585   struct GNUNET_MQ_Handle *mq;
586
587   mq = GNUNET_new (struct GNUNET_MQ_Handle);
588   mq->send_impl = send;
589   mq->destroy_impl = destroy;
590   mq->cancel_impl = cancel;
591   mq->handlers = GNUNET_MQ_copy_handlers (handlers);
592   mq->error_handler = error_handler;
593   mq->error_handler_cls = error_handler_cls;
594   mq->impl_state = impl_state;
595
596   return mq;
597 }
598
599
600 /**
601  * Change the closure argument in all of the `handlers` of the
602  * @a mq.
603  *
604  * @param mq to modify
605  * @param handlers_cls new closure to use
606  */
607 void
608 GNUNET_MQ_set_handlers_closure (struct GNUNET_MQ_Handle *mq,
609                                 void *handlers_cls)
610 {
611   if (NULL == mq->handlers)
612     return;
613   for (unsigned int i=0;NULL != mq->handlers[i].cb; i++)
614     mq->handlers[i].cls = handlers_cls;
615 }
616
617
618 /**
619  * Get the message that should currently be sent.
620  * Fails if there is no current message.
621  * Only useful for implementing message queues,
622  * results in undefined behavior if not used carefully.
623  *
624  * @param mq message queue with the current message
625  * @return message to send, never NULL
626  */
627 const struct GNUNET_MessageHeader *
628 GNUNET_MQ_impl_current (struct GNUNET_MQ_Handle *mq)
629 {
630   GNUNET_assert (NULL != mq->current_envelope);
631   GNUNET_assert (NULL != mq->current_envelope->mh);
632   return mq->current_envelope->mh;
633 }
634
635
636 /**
637  * Get the implementation state associated with the
638  * message queue.
639  *
640  * While the GNUNET_MQ_Impl* callbacks receive the
641  * implementation state, continuations that are scheduled
642  * by the implementation function often only have one closure
643  * argument, with this function it is possible to get at the
644  * implementation state when only passing the GNUNET_MQ_Handle
645  * as closure.
646  *
647  * @param mq message queue with the current message
648  * @return message to send, never NULL
649  */
650 void *
651 GNUNET_MQ_impl_state (struct GNUNET_MQ_Handle *mq)
652 {
653   return mq->impl_state;
654 }
655
656
657 struct GNUNET_MQ_Envelope *
658 GNUNET_MQ_msg_ (struct GNUNET_MessageHeader **mhp,
659                 uint16_t size,
660                 uint16_t type)
661 {
662   struct GNUNET_MQ_Envelope *ev;
663
664   ev = GNUNET_malloc (size + sizeof (struct GNUNET_MQ_Envelope));
665   ev->mh = (struct GNUNET_MessageHeader *) &ev[1];
666   ev->mh->size = htons (size);
667   ev->mh->type = htons (type);
668   if (NULL != mhp)
669     *mhp = ev->mh;
670   return ev;
671 }
672
673
674 /**
675  * Create a new envelope by copying an existing message.
676  *
677  * @param hdr header of the message to copy
678  * @return envelope containing @a hdr
679  */
680 struct GNUNET_MQ_Envelope *
681 GNUNET_MQ_msg_copy (const struct GNUNET_MessageHeader *hdr)
682 {
683   struct GNUNET_MQ_Envelope *mqm;
684   uint16_t size = ntohs (hdr->size);
685
686   mqm = GNUNET_malloc (sizeof (*mqm) + size);
687   mqm->mh = (struct GNUNET_MessageHeader *) &mqm[1];
688   GNUNET_memcpy (mqm->mh,
689           hdr,
690           size);
691   return mqm;
692 }
693
694
695 /**
696  * Implementation of the #GNUNET_MQ_msg_nested_mh macro.
697  *
698  * @param mhp pointer to the message header pointer that will be changed to allocate at
699  *        the newly allocated space for the message.
700  * @param base_size size of the data before the nested message
701  * @param type type of the message in the envelope
702  * @param nested_mh the message to append to the message after base_size
703  */
704 struct GNUNET_MQ_Envelope *
705 GNUNET_MQ_msg_nested_mh_ (struct GNUNET_MessageHeader **mhp,
706                           uint16_t base_size,
707                           uint16_t type,
708                           const struct GNUNET_MessageHeader *nested_mh)
709 {
710   struct GNUNET_MQ_Envelope *mqm;
711   uint16_t size;
712
713   if (NULL == nested_mh)
714     return GNUNET_MQ_msg_ (mhp, base_size, type);
715
716   size = base_size + ntohs (nested_mh->size);
717
718   /* check for uint16_t overflow */
719   if (size < base_size)
720     return NULL;
721
722   mqm = GNUNET_MQ_msg_ (mhp, size, type);
723   GNUNET_memcpy ((char *) mqm->mh + base_size,
724                  nested_mh,
725                  ntohs (nested_mh->size));
726
727   return mqm;
728 }
729
730
731 /**
732  * Associate the assoc_data in mq with a unique request id.
733  *
734  * @param mq message queue, id will be unique for the queue
735  * @param assoc_data to associate
736  */
737 uint32_t
738 GNUNET_MQ_assoc_add (struct GNUNET_MQ_Handle *mq,
739                      void *assoc_data)
740 {
741   uint32_t id;
742
743   if (NULL == mq->assoc_map)
744   {
745     mq->assoc_map = GNUNET_CONTAINER_multihashmap32_create (8);
746     mq->assoc_id = 1;
747   }
748   id = mq->assoc_id++;
749   GNUNET_assert (GNUNET_OK ==
750                  GNUNET_CONTAINER_multihashmap32_put (mq->assoc_map,
751                                                       id,
752                                                       assoc_data,
753                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
754   return id;
755 }
756
757
758 /**
759  * Get the data associated with a @a request_id in a queue
760  *
761  * @param mq the message queue with the association
762  * @param request_id the request id we are interested in
763  * @return the associated data
764  */
765 void *
766 GNUNET_MQ_assoc_get (struct GNUNET_MQ_Handle *mq,
767                      uint32_t request_id)
768 {
769   if (NULL == mq->assoc_map)
770     return NULL;
771   return GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map,
772                                               request_id);
773 }
774
775
776 /**
777  * Remove the association for a @a request_id
778  *
779  * @param mq the message queue with the association
780  * @param request_id the request id we want to remove
781  * @return the associated data
782  */
783 void *
784 GNUNET_MQ_assoc_remove (struct GNUNET_MQ_Handle *mq,
785                         uint32_t request_id)
786 {
787   void *val;
788
789   if (NULL == mq->assoc_map)
790     return NULL;
791   val = GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map,
792                                              request_id);
793   GNUNET_CONTAINER_multihashmap32_remove_all (mq->assoc_map,
794                                               request_id);
795   return val;
796 }
797
798
799 /**
800  * Call a callback once the envelope has been sent, that is,
801  * sending it can not be canceled anymore.
802  * There can be only one notify sent callback per envelope.
803  *
804  * @param ev message to call the notify callback for
805  * @param cb the notify callback
806  * @param cb_cls closure for the callback
807  */
808 void
809 GNUNET_MQ_notify_sent (struct GNUNET_MQ_Envelope *ev,
810                        GNUNET_SCHEDULER_TaskCallback cb,
811                        void *cb_cls)
812 {
813   /* allow setting *OR* clearing callback */
814   GNUNET_assert ( (NULL == ev->sent_cb) ||
815                   (NULL == cb) );
816   ev->sent_cb = cb;
817   ev->sent_cls = cb_cls;
818 }
819
820
821 /**
822  * Handle we return for callbacks registered to be
823  * notified when #GNUNET_MQ_destroy() is called on a queue.
824  */
825 struct GNUNET_MQ_DestroyNotificationHandle
826 {
827   /**
828    * Kept in a DLL.
829    */
830   struct GNUNET_MQ_DestroyNotificationHandle *prev;
831
832   /**
833    * Kept in a DLL.
834    */
835   struct GNUNET_MQ_DestroyNotificationHandle *next;
836
837   /**
838    * Queue to notify about.
839    */
840   struct GNUNET_MQ_Handle *mq;
841
842   /**
843    * Function to call.
844    */
845   GNUNET_SCHEDULER_TaskCallback cb;
846
847   /**
848    * Closure for @e cb.
849    */
850   void *cb_cls;
851 };
852
853
854 /**
855  * Destroy the message queue.
856  *
857  * @param mq message queue to destroy
858  */
859 void
860 GNUNET_MQ_destroy (struct GNUNET_MQ_Handle *mq)
861 {
862   struct GNUNET_MQ_DestroyNotificationHandle *dnh;
863
864   if (NULL != mq->destroy_impl)
865   {
866     mq->destroy_impl (mq, mq->impl_state);
867   }
868   if (NULL != mq->send_task)
869   {
870     GNUNET_SCHEDULER_cancel (mq->send_task);
871     mq->send_task = NULL;
872   }
873   while (NULL != mq->envelope_head)
874   {
875     struct GNUNET_MQ_Envelope *ev;
876
877     ev = mq->envelope_head;
878     ev->parent_queue = NULL;
879     GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
880                                  mq->envelope_tail,
881                                  ev);
882     GNUNET_assert (0 < mq->queue_length);
883     mq->queue_length--;
884     LOG (GNUNET_ERROR_TYPE_DEBUG,
885          "MQ destroy drops message of type %u\n",
886          ntohs (ev->mh->type));
887     GNUNET_MQ_discard (ev);
888   }
889   if (NULL != mq->current_envelope)
890   {
891     /* we can only discard envelopes that
892      * are not queued! */
893     mq->current_envelope->parent_queue = NULL;
894     LOG (GNUNET_ERROR_TYPE_DEBUG,
895          "MQ destroy drops current message of type %u\n",
896          ntohs (mq->current_envelope->mh->type));
897     GNUNET_MQ_discard (mq->current_envelope);
898     mq->current_envelope = NULL;
899     GNUNET_assert (0 < mq->queue_length);
900     mq->queue_length--;
901   }
902   GNUNET_assert (0 == mq->queue_length);
903   while (NULL != (dnh = mq->dnh_head))
904   {
905     dnh->cb (dnh->cb_cls);
906     GNUNET_MQ_destroy_notify_cancel (dnh);
907   }
908   if (NULL != mq->assoc_map)
909   {
910     GNUNET_CONTAINER_multihashmap32_destroy (mq->assoc_map);
911     mq->assoc_map = NULL;
912   }
913   GNUNET_free_non_null (mq->handlers);
914   GNUNET_free (mq);
915 }
916
917
918 const struct GNUNET_MessageHeader *
919 GNUNET_MQ_extract_nested_mh_ (const struct GNUNET_MessageHeader *mh,
920                               uint16_t base_size)
921 {
922   uint16_t whole_size;
923   uint16_t nested_size;
924   const struct GNUNET_MessageHeader *nested_msg;
925
926   whole_size = ntohs (mh->size);
927   GNUNET_assert (whole_size >= base_size);
928   nested_size = whole_size - base_size;
929   if (0 == nested_size)
930     return NULL;
931   if (nested_size < sizeof (struct GNUNET_MessageHeader))
932   {
933     GNUNET_break_op (0);
934     return NULL;
935   }
936   nested_msg = (const struct GNUNET_MessageHeader *) ((char *) mh + base_size);
937   if (ntohs (nested_msg->size) != nested_size)
938   {
939     GNUNET_break_op (0);
940     return NULL;
941   }
942   return nested_msg;
943 }
944
945
946 /**
947  * Cancel sending the message. Message must have been sent with
948  * #GNUNET_MQ_send before.  May not be called after the notify sent
949  * callback has been called
950  *
951  * @param ev queued envelope to cancel
952  */
953 void
954 GNUNET_MQ_send_cancel (struct GNUNET_MQ_Envelope *ev)
955 {
956   struct GNUNET_MQ_Handle *mq = ev->parent_queue;
957
958   GNUNET_assert (NULL != mq);
959   GNUNET_assert (NULL != mq->cancel_impl);
960
961   mq->evacuate_called = GNUNET_NO;
962
963   if (mq->current_envelope == ev)
964   {
965     /* complex case, we already started with transmitting
966        the message using the callbacks. */
967     GNUNET_assert (GNUNET_NO == mq->in_flight);
968     GNUNET_assert (0 < mq->queue_length);
969     mq->queue_length--;
970     mq->cancel_impl (mq,
971                      mq->impl_state);
972     /* continue sending the next message, if any */
973     mq->current_envelope = mq->envelope_head;
974     if (NULL != mq->current_envelope)
975     {
976       GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
977                                    mq->envelope_tail,
978                                    mq->current_envelope);
979
980       LOG (GNUNET_ERROR_TYPE_DEBUG,
981            "sending canceled message of type %u queue\n",
982            ntohs(ev->mh->type));
983
984       mq->send_impl (mq,
985                      mq->current_envelope->mh,
986                      mq->impl_state);
987     }
988   }
989   else
990   {
991     /* simple case, message is still waiting in the queue */
992     GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
993                                  mq->envelope_tail,
994                                  ev);
995     GNUNET_assert (0 < mq->queue_length);
996     mq->queue_length--;
997   }
998
999   if (GNUNET_YES != mq->evacuate_called)
1000   {
1001     ev->parent_queue = NULL;
1002     ev->mh = NULL;
1003     /* also frees ev */
1004     GNUNET_free (ev);
1005   }
1006 }
1007
1008
1009 /**
1010  * Function to obtain the current envelope
1011  * from within #GNUNET_MQ_SendImpl implementations.
1012  *
1013  * @param mq message queue to interrogate
1014  * @return the current envelope
1015  */
1016 struct GNUNET_MQ_Envelope *
1017 GNUNET_MQ_get_current_envelope (struct GNUNET_MQ_Handle *mq)
1018 {
1019   return mq->current_envelope;
1020 }
1021
1022
1023 /**
1024  * Function to obtain the last envelope in the queue.
1025  *
1026  * @param mq message queue to interrogate
1027  * @return the last envelope in the queue
1028  */
1029 struct GNUNET_MQ_Envelope *
1030 GNUNET_MQ_get_last_envelope (struct GNUNET_MQ_Handle *mq)
1031 {
1032   if (NULL != mq->envelope_tail)
1033     return mq->envelope_tail;
1034
1035   return mq->current_envelope;
1036 }
1037
1038
1039 /**
1040  * Set application-specific options for this envelope.
1041  * Overrides the options set for the queue with
1042  * #GNUNET_MQ_set_options() for this message only.
1043  *
1044  * @param env message to set options for
1045  * @param flags flags to use (meaning is queue-specific)
1046  * @param extra additional buffer for further data (also queue-specific)
1047  */
1048 void
1049 GNUNET_MQ_env_set_options (struct GNUNET_MQ_Envelope *env,
1050                            uint64_t flags,
1051                            const void *extra)
1052 {
1053   env->flags = flags;
1054   env->extra = extra;
1055   env->have_custom_options = GNUNET_YES;
1056 }
1057
1058
1059 /**
1060  * Get application-specific options for this envelope.
1061  *
1062  * @param env message to set options for
1063  * @param[out] flags set to flags to use (meaning is queue-specific)
1064  * @return extra additional buffer for further data (also queue-specific)
1065  */
1066 const void *
1067 GNUNET_MQ_env_get_options (struct GNUNET_MQ_Envelope *env,
1068                            uint64_t *flags)
1069 {
1070   struct GNUNET_MQ_Handle *mq = env->parent_queue;
1071
1072   if (GNUNET_YES == env->have_custom_options)
1073   {
1074     *flags = env->flags;
1075     return env->extra;
1076   }
1077   if (NULL == mq)
1078   {
1079     *flags = 0;
1080     return NULL;
1081   }
1082   *flags = mq->default_flags;
1083   return mq->default_extra;
1084 }
1085
1086
1087 /**
1088  * Set application-specific options for this queue.
1089  *
1090  * @param mq message queue to set options for
1091  * @param flags flags to use (meaning is queue-specific)
1092  * @param extra additional buffer for further data (also queue-specific)
1093  */
1094 void
1095 GNUNET_MQ_set_options (struct GNUNET_MQ_Handle *mq,
1096                        uint64_t flags,
1097                        const void *extra)
1098 {
1099   mq->default_flags = flags;
1100   mq->default_extra = extra;
1101 }
1102
1103
1104 /**
1105  * Obtain message contained in envelope.
1106  *
1107  * @param env the envelope
1108  * @return message contained in the envelope
1109  */
1110 const struct GNUNET_MessageHeader *
1111 GNUNET_MQ_env_get_msg (const struct GNUNET_MQ_Envelope *env)
1112 {
1113   return env->mh;
1114 }
1115
1116
1117 /**
1118  * Return next envelope in queue.
1119  *
1120  * @param env a queued envelope
1121  * @return next one, or NULL
1122  */
1123 const struct GNUNET_MQ_Envelope *
1124 GNUNET_MQ_env_next (const struct GNUNET_MQ_Envelope *env)
1125 {
1126   return env->next;
1127 }
1128
1129
1130 /**
1131  * Register function to be called whenever @a mq is being
1132  * destroyed.
1133  *
1134  * @param mq message queue to watch
1135  * @param cb function to call on @a mq destruction
1136  * @param cb_cls closure for @a cb
1137  * @return handle for #GNUNET_MQ_destroy_notify_cancel().
1138  */
1139 struct GNUNET_MQ_DestroyNotificationHandle *
1140 GNUNET_MQ_destroy_notify (struct GNUNET_MQ_Handle *mq,
1141                           GNUNET_SCHEDULER_TaskCallback cb,
1142                           void *cb_cls)
1143 {
1144   struct GNUNET_MQ_DestroyNotificationHandle *dnh;
1145
1146   dnh = GNUNET_new (struct GNUNET_MQ_DestroyNotificationHandle);
1147   dnh->mq = mq;
1148   dnh->cb = cb;
1149   dnh->cb_cls = cb_cls;
1150   GNUNET_CONTAINER_DLL_insert (mq->dnh_head,
1151                                mq->dnh_tail,
1152                                dnh);
1153   return dnh;
1154 }
1155
1156
1157 /**
1158  * Cancel registration from #GNUNET_MQ_destroy_notify().
1159  *
1160  * @param dnh handle for registration to cancel
1161  */
1162 void
1163 GNUNET_MQ_destroy_notify_cancel (struct GNUNET_MQ_DestroyNotificationHandle *dnh)
1164 {
1165   struct GNUNET_MQ_Handle *mq = dnh->mq;
1166
1167   GNUNET_CONTAINER_DLL_remove (mq->dnh_head,
1168                                mq->dnh_tail,
1169                                dnh);
1170   GNUNET_free (dnh);
1171 }
1172
1173
1174 /**
1175  * Insert @a env into the envelope DLL starting at @a env_head
1176  * Note that @a env must not be in any MQ while this function
1177  * is used with DLLs defined outside of the MQ module.  This
1178  * is just in case some application needs to also manage a
1179  * FIFO of envelopes independent of MQ itself and wants to
1180  * re-use the pointers internal to @a env.  Use with caution.
1181  *
1182  * @param[in|out] env_head of envelope DLL
1183  * @param[in|out] env_tail tail of envelope DLL
1184  * @param[in|out] env element to insert at the tail
1185  */
1186 void
1187 GNUNET_MQ_dll_insert_tail (struct GNUNET_MQ_Envelope **env_head,
1188                            struct GNUNET_MQ_Envelope **env_tail,
1189                            struct GNUNET_MQ_Envelope *env)
1190 {
1191   GNUNET_CONTAINER_DLL_insert_tail (*env_head,
1192                                     *env_tail,
1193                                     env);
1194 }
1195
1196
1197 /**
1198  * Remove @a env from the envelope DLL starting at @a env_head.
1199  * Note that @a env must not be in any MQ while this function
1200  * is used with DLLs defined outside of the MQ module. This
1201  * is just in case some application needs to also manage a
1202  * FIFO of envelopes independent of MQ itself and wants to
1203  * re-use the pointers internal to @a env.  Use with caution.
1204  *
1205  * @param[in|out] env_head of envelope DLL
1206  * @param[in|out] env_tail tail of envelope DLL
1207  * @param[in|out] env element to remove from the DLL
1208  */
1209 void
1210 GNUNET_MQ_dll_remove (struct GNUNET_MQ_Envelope **env_head,
1211                       struct GNUNET_MQ_Envelope **env_tail,
1212                       struct GNUNET_MQ_Envelope *env)
1213 {
1214   GNUNET_CONTAINER_DLL_remove (*env_head,
1215                                *env_tail,
1216                                env);
1217 }
1218
1219
1220 /**
1221  * Copy an array of handlers.
1222  *
1223  * Useful if the array has been delared in local memory and needs to be
1224  * persisted for future use.
1225  *
1226  * @param handlers Array of handlers to be copied. Can be NULL (nothing done).
1227  * @return A newly allocated array of handlers.
1228  *         Needs to be freed with #GNUNET_free.
1229  */
1230 struct GNUNET_MQ_MessageHandler *
1231 GNUNET_MQ_copy_handlers (const struct GNUNET_MQ_MessageHandler *handlers)
1232 {
1233   struct GNUNET_MQ_MessageHandler *copy;
1234   unsigned int count;
1235
1236   if (NULL == handlers)
1237     return NULL;
1238
1239   count = GNUNET_MQ_count_handlers (handlers);
1240   copy = GNUNET_new_array (count + 1,
1241                            struct GNUNET_MQ_MessageHandler);
1242   GNUNET_memcpy (copy,
1243                  handlers,
1244                  count * sizeof (struct GNUNET_MQ_MessageHandler));
1245   return copy;
1246 }
1247
1248
1249 /**
1250  * Copy an array of handlers, appending AGPL handler.
1251  *
1252  * Useful if the array has been delared in local memory and needs to be
1253  * persisted for future use.
1254  *
1255  * @param handlers Array of handlers to be copied. Can be NULL (nothing done).
1256  * @param agpl_handler function to call for AGPL handling
1257  * @param agpl_cls closure for @a agpl_handler
1258  * @return A newly allocated array of handlers.
1259  *         Needs to be freed with #GNUNET_free.
1260  */
1261 struct GNUNET_MQ_MessageHandler *
1262 GNUNET_MQ_copy_handlers2 (const struct GNUNET_MQ_MessageHandler *handlers,
1263                           GNUNET_MQ_MessageCallback agpl_handler,
1264                           void *agpl_cls)
1265 {
1266   struct GNUNET_MQ_MessageHandler *copy;
1267   unsigned int count;
1268
1269   if (NULL == handlers)
1270     return NULL;
1271   count = GNUNET_MQ_count_handlers (handlers);
1272   copy = GNUNET_new_array (count + 2,
1273                            struct GNUNET_MQ_MessageHandler);
1274   GNUNET_memcpy (copy,
1275                  handlers,
1276                  count * sizeof (struct GNUNET_MQ_MessageHandler));
1277   copy[count].mv = NULL;
1278   copy[count].cb = agpl_handler;
1279   copy[count].cls = agpl_cls;
1280   copy[count].type = GNUNET_MESSAGE_TYPE_REQUEST_AGPL;
1281   copy[count].expected_size = sizeof (struct GNUNET_MessageHeader);
1282   return copy;
1283 }
1284
1285
1286 /**
1287  * Count the handlers in a handler array.
1288  *
1289  * @param handlers Array of handlers to be counted.
1290  * @return The number of handlers in the array.
1291  */
1292 unsigned int
1293 GNUNET_MQ_count_handlers (const struct GNUNET_MQ_MessageHandler *handlers)
1294 {
1295   unsigned int i;
1296
1297   if (NULL == handlers)
1298     return 0;
1299
1300   for (i=0; NULL != handlers[i].cb; i++) ;
1301
1302   return i;
1303 }
1304
1305
1306 /**
1307  * Convert an `enum GNUNET_MQ_PreferenceType` to a string
1308  *
1309  * @param type the preference type
1310  * @return a string or NULL if invalid
1311  */
1312 const char *
1313 GNUNET_MQ_preference_to_string (enum GNUNET_MQ_PreferenceKind type)
1314 {
1315   switch (type) {
1316   case GNUNET_MQ_PREFERENCE_NONE:
1317     return "NONE";
1318   case GNUNET_MQ_PREFERENCE_BANDWIDTH:
1319     return "BANDWIDTH";
1320   case GNUNET_MQ_PREFERENCE_LATENCY:
1321     return "LATENCY";
1322   case GNUNET_MQ_PREFERENCE_RELIABILITY:
1323     return "RELIABILITY";
1324   };
1325   return NULL;
1326 }
1327
1328
1329 /* end of mq.c */