add nonstring attribute to silence warning
[oweals/gnunet.git] / src / util / mq.c
1 /*
2      This file is part of GNUnet.
3      Copyright (C) 2012-2019 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   if (mq->queue_length >= 10000)
378   {
379     /* This would seem like a bug... */
380     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
381                 "MQ with %u entries extended by message of type %u (FC broken?)\n",
382                 (unsigned int) mq->queue_length,
383                 (unsigned int) ntohs (ev->mh->type));
384   }
385   ev->parent_queue = mq;
386   /* is the implementation busy? queue it! */
387   if ( (NULL != mq->current_envelope) ||
388        (NULL != mq->send_task) )
389   {
390     GNUNET_CONTAINER_DLL_insert_tail (mq->envelope_head,
391                                       mq->envelope_tail,
392                                       ev);
393     return;
394   }
395   GNUNET_assert (NULL == mq->envelope_head);
396   mq->current_envelope = ev;
397
398   LOG (GNUNET_ERROR_TYPE_DEBUG,
399        "sending message of type %u, queue empty (MQ: %p)\n",
400        ntohs(ev->mh->type),
401        mq);
402
403   mq->send_impl (mq,
404                  ev->mh,
405                  mq->impl_state);
406 }
407
408
409 /**
410  * Remove the first envelope that has not yet been sent from the message
411  * queue and return it.
412  *
413  * @param mq queue to remove envelope from
414  * @return NULL if queue is empty (or has no envelope that is not under transmission)
415  */
416 struct GNUNET_MQ_Envelope *
417 GNUNET_MQ_unsent_head (struct GNUNET_MQ_Handle *mq)
418 {
419   struct GNUNET_MQ_Envelope *env;
420
421   env = mq->envelope_head;
422   GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
423                                mq->envelope_tail,
424                                env);
425   mq->queue_length--;
426   env->parent_queue = NULL;
427   return env;
428 }
429
430
431 /**
432  * Function to copy an envelope.  The envelope must not yet
433  * be in any queue or have any options or callbacks set.
434  *
435  * @param env envelope to copy
436  * @return copy of @a env
437  */
438 struct GNUNET_MQ_Envelope *
439 GNUNET_MQ_env_copy (struct GNUNET_MQ_Envelope *env)
440 {
441   GNUNET_assert (NULL == env->next);
442   GNUNET_assert (NULL == env->parent_queue);
443   GNUNET_assert (NULL == env->sent_cb);
444   GNUNET_assert (GNUNET_NO == env->have_custom_options);
445   return GNUNET_MQ_msg_copy (env->mh);
446 }
447
448
449 /**
450  * Send a copy of a message with the given message queue.
451  * Can be called repeatedly on the same envelope.
452  *
453  * @param mq message queue
454  * @param ev the envelope with the message to send.
455  */
456 void
457 GNUNET_MQ_send_copy (struct GNUNET_MQ_Handle *mq,
458                      const struct GNUNET_MQ_Envelope *ev)
459 {
460   struct GNUNET_MQ_Envelope *env;
461   uint16_t msize;
462
463   msize = ntohs (ev->mh->size);
464   env = GNUNET_malloc (sizeof (struct GNUNET_MQ_Envelope) +
465                        msize);
466   env->mh = (struct GNUNET_MessageHeader *) &env[1];
467   env->sent_cb = ev->sent_cb;
468   env->sent_cls = ev->sent_cls;
469   GNUNET_memcpy (&env[1],
470           ev->mh,
471           msize);
472   GNUNET_MQ_send (mq,
473                   env);
474 }
475
476
477 /**
478  * Task run to call the send implementation for the next queued
479  * message, if any.  Only useful for implementing message queues,
480  * results in undefined behavior if not used carefully.
481  *
482  * @param cls message queue to send the next message with
483  */
484 static void
485 impl_send_continue (void *cls)
486 {
487   struct GNUNET_MQ_Handle *mq = cls;
488
489   mq->send_task = NULL;
490   /* call is only valid if we're actually currently sending
491    * a message */
492   if (NULL == mq->envelope_head)
493     return;
494   mq->current_envelope = mq->envelope_head;
495   GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
496                                mq->envelope_tail,
497                                mq->current_envelope);
498
499   LOG (GNUNET_ERROR_TYPE_DEBUG,
500        "sending message of type %u from queue\n",
501        ntohs(mq->current_envelope->mh->type));
502
503   mq->send_impl (mq,
504                  mq->current_envelope->mh,
505                  mq->impl_state);
506 }
507
508
509 /**
510  * Call the send implementation for the next queued message, if any.
511  * Only useful for implementing message queues, results in undefined
512  * behavior if not used carefully.
513  *
514  * @param mq message queue to send the next message with
515  */
516 void
517 GNUNET_MQ_impl_send_continue (struct GNUNET_MQ_Handle *mq)
518 {
519   struct GNUNET_MQ_Envelope *current_envelope;
520   GNUNET_SCHEDULER_TaskCallback cb;
521
522   GNUNET_assert (0 < mq->queue_length);
523   mq->queue_length--;
524   mq->in_flight = GNUNET_NO;
525   current_envelope = mq->current_envelope;
526   current_envelope->parent_queue = NULL;
527   mq->current_envelope = NULL;
528   GNUNET_assert (NULL == mq->send_task);
529   mq->send_task = GNUNET_SCHEDULER_add_now (&impl_send_continue,
530                                             mq);
531   if (NULL != (cb = current_envelope->sent_cb))
532   {
533     current_envelope->sent_cb = NULL;
534     cb (current_envelope->sent_cls);
535   }
536   GNUNET_free (current_envelope);
537 }
538
539
540 /**
541  * Call the send notification for the current message, but do not
542  * try to send the next message until #GNUNET_MQ_impl_send_continue
543  * is called.
544  *
545  * Only useful for implementing message queues, results in undefined
546  * behavior if not used carefully.
547  *
548  * @param mq message queue to send the next message with
549  */
550 void
551 GNUNET_MQ_impl_send_in_flight (struct GNUNET_MQ_Handle *mq)
552 {
553   struct GNUNET_MQ_Envelope *current_envelope;
554   GNUNET_SCHEDULER_TaskCallback cb;
555
556   mq->in_flight = GNUNET_YES;
557   /* call is only valid if we're actually currently sending
558    * a message */
559   current_envelope = mq->current_envelope;
560   GNUNET_assert (NULL != current_envelope);
561   /* can't call cancel from now on anymore */
562   current_envelope->parent_queue = NULL;
563   if (NULL != (cb = current_envelope->sent_cb))
564   {
565     current_envelope->sent_cb = NULL;
566     cb (current_envelope->sent_cls);
567   }
568 }
569
570
571 /**
572  * Create a message queue for the specified handlers.
573  *
574  * @param send function the implements sending messages
575  * @param destroy function that implements destroying the queue
576  * @param cancel function that implements canceling a message
577  * @param impl_state for the queue, passed to 'send' and 'destroy'
578  * @param handlers array of message handlers
579  * @param error_handler handler for read and write errors
580  * @param error_handler_cls closure for @a error_handler
581  * @return a new message queue
582  */
583 struct GNUNET_MQ_Handle *
584 GNUNET_MQ_queue_for_callbacks (GNUNET_MQ_SendImpl send,
585                                GNUNET_MQ_DestroyImpl destroy,
586                                GNUNET_MQ_CancelImpl cancel,
587                                void *impl_state,
588                                const struct GNUNET_MQ_MessageHandler *handlers,
589                                GNUNET_MQ_ErrorHandler error_handler,
590                                void *error_handler_cls)
591 {
592   struct GNUNET_MQ_Handle *mq;
593
594   mq = GNUNET_new (struct GNUNET_MQ_Handle);
595   mq->send_impl = send;
596   mq->destroy_impl = destroy;
597   mq->cancel_impl = cancel;
598   mq->handlers = GNUNET_MQ_copy_handlers (handlers);
599   mq->error_handler = error_handler;
600   mq->error_handler_cls = error_handler_cls;
601   mq->impl_state = impl_state;
602
603   return mq;
604 }
605
606
607 /**
608  * Change the closure argument in all of the `handlers` of the
609  * @a mq.
610  *
611  * @param mq to modify
612  * @param handlers_cls new closure to use
613  */
614 void
615 GNUNET_MQ_set_handlers_closure (struct GNUNET_MQ_Handle *mq,
616                                 void *handlers_cls)
617 {
618   if (NULL == mq->handlers)
619     return;
620   for (unsigned int i=0;NULL != mq->handlers[i].cb; i++)
621     mq->handlers[i].cls = handlers_cls;
622 }
623
624
625 /**
626  * Get the message that should currently be sent.
627  * Fails if there is no current message.
628  * Only useful for implementing message queues,
629  * results in undefined behavior if not used carefully.
630  *
631  * @param mq message queue with the current message
632  * @return message to send, never NULL
633  */
634 const struct GNUNET_MessageHeader *
635 GNUNET_MQ_impl_current (struct GNUNET_MQ_Handle *mq)
636 {
637   GNUNET_assert (NULL != mq->current_envelope);
638   GNUNET_assert (NULL != mq->current_envelope->mh);
639   return mq->current_envelope->mh;
640 }
641
642
643 /**
644  * Get the implementation state associated with the
645  * message queue.
646  *
647  * While the GNUNET_MQ_Impl* callbacks receive the
648  * implementation state, continuations that are scheduled
649  * by the implementation function often only have one closure
650  * argument, with this function it is possible to get at the
651  * implementation state when only passing the GNUNET_MQ_Handle
652  * as closure.
653  *
654  * @param mq message queue with the current message
655  * @return message to send, never NULL
656  */
657 void *
658 GNUNET_MQ_impl_state (struct GNUNET_MQ_Handle *mq)
659 {
660   return mq->impl_state;
661 }
662
663
664 struct GNUNET_MQ_Envelope *
665 GNUNET_MQ_msg_ (struct GNUNET_MessageHeader **mhp,
666                 uint16_t size,
667                 uint16_t type)
668 {
669   struct GNUNET_MQ_Envelope *ev;
670
671   ev = GNUNET_malloc (size + sizeof (struct GNUNET_MQ_Envelope));
672   ev->mh = (struct GNUNET_MessageHeader *) &ev[1];
673   ev->mh->size = htons (size);
674   ev->mh->type = htons (type);
675   if (NULL != mhp)
676     *mhp = ev->mh;
677   return ev;
678 }
679
680
681 /**
682  * Create a new envelope by copying an existing message.
683  *
684  * @param hdr header of the message to copy
685  * @return envelope containing @a hdr
686  */
687 struct GNUNET_MQ_Envelope *
688 GNUNET_MQ_msg_copy (const struct GNUNET_MessageHeader *hdr)
689 {
690   struct GNUNET_MQ_Envelope *mqm;
691   uint16_t size = ntohs (hdr->size);
692
693   mqm = GNUNET_malloc (sizeof (*mqm) + size);
694   mqm->mh = (struct GNUNET_MessageHeader *) &mqm[1];
695   GNUNET_memcpy (mqm->mh,
696           hdr,
697           size);
698   return mqm;
699 }
700
701
702 /**
703  * Implementation of the #GNUNET_MQ_msg_nested_mh macro.
704  *
705  * @param mhp pointer to the message header pointer that will be changed to allocate at
706  *        the newly allocated space for the message.
707  * @param base_size size of the data before the nested message
708  * @param type type of the message in the envelope
709  * @param nested_mh the message to append to the message after base_size
710  */
711 struct GNUNET_MQ_Envelope *
712 GNUNET_MQ_msg_nested_mh_ (struct GNUNET_MessageHeader **mhp,
713                           uint16_t base_size,
714                           uint16_t type,
715                           const struct GNUNET_MessageHeader *nested_mh)
716 {
717   struct GNUNET_MQ_Envelope *mqm;
718   uint16_t size;
719
720   if (NULL == nested_mh)
721     return GNUNET_MQ_msg_ (mhp, base_size, type);
722
723   size = base_size + ntohs (nested_mh->size);
724
725   /* check for uint16_t overflow */
726   if (size < base_size)
727     return NULL;
728
729   mqm = GNUNET_MQ_msg_ (mhp, size, type);
730   GNUNET_memcpy ((char *) mqm->mh + base_size,
731                  nested_mh,
732                  ntohs (nested_mh->size));
733
734   return mqm;
735 }
736
737
738 /**
739  * Associate the assoc_data in mq with a unique request id.
740  *
741  * @param mq message queue, id will be unique for the queue
742  * @param assoc_data to associate
743  */
744 uint32_t
745 GNUNET_MQ_assoc_add (struct GNUNET_MQ_Handle *mq,
746                      void *assoc_data)
747 {
748   uint32_t id;
749
750   if (NULL == mq->assoc_map)
751   {
752     mq->assoc_map = GNUNET_CONTAINER_multihashmap32_create (8);
753     mq->assoc_id = 1;
754   }
755   id = mq->assoc_id++;
756   GNUNET_assert (GNUNET_OK ==
757                  GNUNET_CONTAINER_multihashmap32_put (mq->assoc_map,
758                                                       id,
759                                                       assoc_data,
760                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
761   return id;
762 }
763
764
765 /**
766  * Get the data associated with a @a request_id in a queue
767  *
768  * @param mq the message queue with the association
769  * @param request_id the request id we are interested in
770  * @return the associated data
771  */
772 void *
773 GNUNET_MQ_assoc_get (struct GNUNET_MQ_Handle *mq,
774                      uint32_t request_id)
775 {
776   if (NULL == mq->assoc_map)
777     return NULL;
778   return GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map,
779                                               request_id);
780 }
781
782
783 /**
784  * Remove the association for a @a request_id
785  *
786  * @param mq the message queue with the association
787  * @param request_id the request id we want to remove
788  * @return the associated data
789  */
790 void *
791 GNUNET_MQ_assoc_remove (struct GNUNET_MQ_Handle *mq,
792                         uint32_t request_id)
793 {
794   void *val;
795
796   if (NULL == mq->assoc_map)
797     return NULL;
798   val = GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map,
799                                              request_id);
800   GNUNET_CONTAINER_multihashmap32_remove_all (mq->assoc_map,
801                                               request_id);
802   return val;
803 }
804
805
806 /**
807  * Call a callback once the envelope has been sent, that is,
808  * sending it can not be canceled anymore.
809  * There can be only one notify sent callback per envelope.
810  *
811  * @param ev message to call the notify callback for
812  * @param cb the notify callback
813  * @param cb_cls closure for the callback
814  */
815 void
816 GNUNET_MQ_notify_sent (struct GNUNET_MQ_Envelope *ev,
817                        GNUNET_SCHEDULER_TaskCallback cb,
818                        void *cb_cls)
819 {
820   /* allow setting *OR* clearing callback */
821   GNUNET_assert ( (NULL == ev->sent_cb) ||
822                   (NULL == cb) );
823   ev->sent_cb = cb;
824   ev->sent_cls = cb_cls;
825 }
826
827
828 /**
829  * Handle we return for callbacks registered to be
830  * notified when #GNUNET_MQ_destroy() is called on a queue.
831  */
832 struct GNUNET_MQ_DestroyNotificationHandle
833 {
834   /**
835    * Kept in a DLL.
836    */
837   struct GNUNET_MQ_DestroyNotificationHandle *prev;
838
839   /**
840    * Kept in a DLL.
841    */
842   struct GNUNET_MQ_DestroyNotificationHandle *next;
843
844   /**
845    * Queue to notify about.
846    */
847   struct GNUNET_MQ_Handle *mq;
848
849   /**
850    * Function to call.
851    */
852   GNUNET_SCHEDULER_TaskCallback cb;
853
854   /**
855    * Closure for @e cb.
856    */
857   void *cb_cls;
858 };
859
860
861 /**
862  * Destroy the message queue.
863  *
864  * @param mq message queue to destroy
865  */
866 void
867 GNUNET_MQ_destroy (struct GNUNET_MQ_Handle *mq)
868 {
869   struct GNUNET_MQ_DestroyNotificationHandle *dnh;
870
871   if (NULL != mq->destroy_impl)
872   {
873     mq->destroy_impl (mq, mq->impl_state);
874   }
875   if (NULL != mq->send_task)
876   {
877     GNUNET_SCHEDULER_cancel (mq->send_task);
878     mq->send_task = NULL;
879   }
880   while (NULL != mq->envelope_head)
881   {
882     struct GNUNET_MQ_Envelope *ev;
883
884     ev = mq->envelope_head;
885     ev->parent_queue = NULL;
886     GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
887                                  mq->envelope_tail,
888                                  ev);
889     GNUNET_assert (0 < mq->queue_length);
890     mq->queue_length--;
891     LOG (GNUNET_ERROR_TYPE_DEBUG,
892          "MQ destroy drops message of type %u\n",
893          ntohs (ev->mh->type));
894     GNUNET_MQ_discard (ev);
895   }
896   if (NULL != mq->current_envelope)
897   {
898     /* we can only discard envelopes that
899      * are not queued! */
900     mq->current_envelope->parent_queue = NULL;
901     LOG (GNUNET_ERROR_TYPE_DEBUG,
902          "MQ destroy drops current message of type %u\n",
903          ntohs (mq->current_envelope->mh->type));
904     GNUNET_MQ_discard (mq->current_envelope);
905     mq->current_envelope = NULL;
906     GNUNET_assert (0 < mq->queue_length);
907     mq->queue_length--;
908   }
909   GNUNET_assert (0 == mq->queue_length);
910   while (NULL != (dnh = mq->dnh_head))
911   {
912     dnh->cb (dnh->cb_cls);
913     GNUNET_MQ_destroy_notify_cancel (dnh);
914   }
915   if (NULL != mq->assoc_map)
916   {
917     GNUNET_CONTAINER_multihashmap32_destroy (mq->assoc_map);
918     mq->assoc_map = NULL;
919   }
920   GNUNET_free_non_null (mq->handlers);
921   GNUNET_free (mq);
922 }
923
924
925 const struct GNUNET_MessageHeader *
926 GNUNET_MQ_extract_nested_mh_ (const struct GNUNET_MessageHeader *mh,
927                               uint16_t base_size)
928 {
929   uint16_t whole_size;
930   uint16_t nested_size;
931   const struct GNUNET_MessageHeader *nested_msg;
932
933   whole_size = ntohs (mh->size);
934   GNUNET_assert (whole_size >= base_size);
935   nested_size = whole_size - base_size;
936   if (0 == nested_size)
937     return NULL;
938   if (nested_size < sizeof (struct GNUNET_MessageHeader))
939   {
940     GNUNET_break_op (0);
941     return NULL;
942   }
943   nested_msg = (const struct GNUNET_MessageHeader *) ((char *) mh + base_size);
944   if (ntohs (nested_msg->size) != nested_size)
945   {
946     GNUNET_break_op (0);
947     return NULL;
948   }
949   return nested_msg;
950 }
951
952
953 /**
954  * Cancel sending the message. Message must have been sent with
955  * #GNUNET_MQ_send before.  May not be called after the notify sent
956  * callback has been called
957  *
958  * @param ev queued envelope to cancel
959  */
960 void
961 GNUNET_MQ_send_cancel (struct GNUNET_MQ_Envelope *ev)
962 {
963   struct GNUNET_MQ_Handle *mq = ev->parent_queue;
964
965   GNUNET_assert (NULL != mq);
966   GNUNET_assert (NULL != mq->cancel_impl);
967
968   mq->evacuate_called = GNUNET_NO;
969
970   if (mq->current_envelope == ev)
971   {
972     /* complex case, we already started with transmitting
973        the message using the callbacks. */
974     GNUNET_assert (GNUNET_NO == mq->in_flight);
975     GNUNET_assert (0 < mq->queue_length);
976     mq->queue_length--;
977     mq->cancel_impl (mq,
978                      mq->impl_state);
979     /* continue sending the next message, if any */
980     mq->current_envelope = mq->envelope_head;
981     if (NULL != mq->current_envelope)
982     {
983       GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
984                                    mq->envelope_tail,
985                                    mq->current_envelope);
986
987       LOG (GNUNET_ERROR_TYPE_DEBUG,
988            "sending canceled message of type %u queue\n",
989            ntohs(ev->mh->type));
990
991       mq->send_impl (mq,
992                      mq->current_envelope->mh,
993                      mq->impl_state);
994     }
995   }
996   else
997   {
998     /* simple case, message is still waiting in the queue */
999     GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
1000                                  mq->envelope_tail,
1001                                  ev);
1002     GNUNET_assert (0 < mq->queue_length);
1003     mq->queue_length--;
1004   }
1005
1006   if (GNUNET_YES != mq->evacuate_called)
1007   {
1008     ev->parent_queue = NULL;
1009     ev->mh = NULL;
1010     /* also frees ev */
1011     GNUNET_free (ev);
1012   }
1013 }
1014
1015
1016 /**
1017  * Function to obtain the current envelope
1018  * from within #GNUNET_MQ_SendImpl implementations.
1019  *
1020  * @param mq message queue to interrogate
1021  * @return the current envelope
1022  */
1023 struct GNUNET_MQ_Envelope *
1024 GNUNET_MQ_get_current_envelope (struct GNUNET_MQ_Handle *mq)
1025 {
1026   return mq->current_envelope;
1027 }
1028
1029
1030 /**
1031  * Function to obtain the last envelope in the queue.
1032  *
1033  * @param mq message queue to interrogate
1034  * @return the last envelope in the queue
1035  */
1036 struct GNUNET_MQ_Envelope *
1037 GNUNET_MQ_get_last_envelope (struct GNUNET_MQ_Handle *mq)
1038 {
1039   if (NULL != mq->envelope_tail)
1040     return mq->envelope_tail;
1041
1042   return mq->current_envelope;
1043 }
1044
1045
1046 /**
1047  * Set application-specific options for this envelope.
1048  * Overrides the options set for the queue with
1049  * #GNUNET_MQ_set_options() for this message only.
1050  *
1051  * @param env message to set options for
1052  * @param flags flags to use (meaning is queue-specific)
1053  * @param extra additional buffer for further data (also queue-specific)
1054  */
1055 void
1056 GNUNET_MQ_env_set_options (struct GNUNET_MQ_Envelope *env,
1057                            uint64_t flags,
1058                            const void *extra)
1059 {
1060   env->flags = flags;
1061   env->extra = extra;
1062   env->have_custom_options = GNUNET_YES;
1063 }
1064
1065
1066 /**
1067  * Get application-specific options for this envelope.
1068  *
1069  * @param env message to set options for
1070  * @param[out] flags set to flags to use (meaning is queue-specific)
1071  * @return extra additional buffer for further data (also queue-specific)
1072  */
1073 const void *
1074 GNUNET_MQ_env_get_options (struct GNUNET_MQ_Envelope *env,
1075                            uint64_t *flags)
1076 {
1077   struct GNUNET_MQ_Handle *mq = env->parent_queue;
1078
1079   if (GNUNET_YES == env->have_custom_options)
1080   {
1081     *flags = env->flags;
1082     return env->extra;
1083   }
1084   if (NULL == mq)
1085   {
1086     *flags = 0;
1087     return NULL;
1088   }
1089   *flags = mq->default_flags;
1090   return mq->default_extra;
1091 }
1092
1093
1094 /**
1095  * Set application-specific options for this queue.
1096  *
1097  * @param mq message queue to set options for
1098  * @param flags flags to use (meaning is queue-specific)
1099  * @param extra additional buffer for further data (also queue-specific)
1100  */
1101 void
1102 GNUNET_MQ_set_options (struct GNUNET_MQ_Handle *mq,
1103                        uint64_t flags,
1104                        const void *extra)
1105 {
1106   mq->default_flags = flags;
1107   mq->default_extra = extra;
1108 }
1109
1110
1111 /**
1112  * Obtain message contained in envelope.
1113  *
1114  * @param env the envelope
1115  * @return message contained in the envelope
1116  */
1117 const struct GNUNET_MessageHeader *
1118 GNUNET_MQ_env_get_msg (const struct GNUNET_MQ_Envelope *env)
1119 {
1120   return env->mh;
1121 }
1122
1123
1124 /**
1125  * Return next envelope in queue.
1126  *
1127  * @param env a queued envelope
1128  * @return next one, or NULL
1129  */
1130 const struct GNUNET_MQ_Envelope *
1131 GNUNET_MQ_env_next (const struct GNUNET_MQ_Envelope *env)
1132 {
1133   return env->next;
1134 }
1135
1136
1137 /**
1138  * Register function to be called whenever @a mq is being
1139  * destroyed.
1140  *
1141  * @param mq message queue to watch
1142  * @param cb function to call on @a mq destruction
1143  * @param cb_cls closure for @a cb
1144  * @return handle for #GNUNET_MQ_destroy_notify_cancel().
1145  */
1146 struct GNUNET_MQ_DestroyNotificationHandle *
1147 GNUNET_MQ_destroy_notify (struct GNUNET_MQ_Handle *mq,
1148                           GNUNET_SCHEDULER_TaskCallback cb,
1149                           void *cb_cls)
1150 {
1151   struct GNUNET_MQ_DestroyNotificationHandle *dnh;
1152
1153   dnh = GNUNET_new (struct GNUNET_MQ_DestroyNotificationHandle);
1154   dnh->mq = mq;
1155   dnh->cb = cb;
1156   dnh->cb_cls = cb_cls;
1157   GNUNET_CONTAINER_DLL_insert (mq->dnh_head,
1158                                mq->dnh_tail,
1159                                dnh);
1160   return dnh;
1161 }
1162
1163
1164 /**
1165  * Cancel registration from #GNUNET_MQ_destroy_notify().
1166  *
1167  * @param dnh handle for registration to cancel
1168  */
1169 void
1170 GNUNET_MQ_destroy_notify_cancel (struct GNUNET_MQ_DestroyNotificationHandle *dnh)
1171 {
1172   struct GNUNET_MQ_Handle *mq = dnh->mq;
1173
1174   GNUNET_CONTAINER_DLL_remove (mq->dnh_head,
1175                                mq->dnh_tail,
1176                                dnh);
1177   GNUNET_free (dnh);
1178 }
1179
1180
1181 /**
1182  * Insert @a env into the envelope DLL starting at @a env_head
1183  * Note that @a env must not be in any MQ while this function
1184  * is used with DLLs defined outside of the MQ module.  This
1185  * is just in case some application needs to also manage a
1186  * FIFO of envelopes independent of MQ itself and wants to
1187  * re-use the pointers internal to @a env.  Use with caution.
1188  *
1189  * @param[in|out] env_head of envelope DLL
1190  * @param[in|out] env_tail tail of envelope DLL
1191  * @param[in|out] env element to insert at the tail
1192  */
1193 void
1194 GNUNET_MQ_dll_insert_tail (struct GNUNET_MQ_Envelope **env_head,
1195                            struct GNUNET_MQ_Envelope **env_tail,
1196                            struct GNUNET_MQ_Envelope *env)
1197 {
1198   GNUNET_CONTAINER_DLL_insert_tail (*env_head,
1199                                     *env_tail,
1200                                     env);
1201 }
1202
1203
1204 /**
1205  * Remove @a env from the envelope DLL starting at @a env_head.
1206  * Note that @a env must not be in any MQ while this function
1207  * is used with DLLs defined outside of the MQ module. This
1208  * is just in case some application needs to also manage a
1209  * FIFO of envelopes independent of MQ itself and wants to
1210  * re-use the pointers internal to @a env.  Use with caution.
1211  *
1212  * @param[in|out] env_head of envelope DLL
1213  * @param[in|out] env_tail tail of envelope DLL
1214  * @param[in|out] env element to remove from the DLL
1215  */
1216 void
1217 GNUNET_MQ_dll_remove (struct GNUNET_MQ_Envelope **env_head,
1218                       struct GNUNET_MQ_Envelope **env_tail,
1219                       struct GNUNET_MQ_Envelope *env)
1220 {
1221   GNUNET_CONTAINER_DLL_remove (*env_head,
1222                                *env_tail,
1223                                env);
1224 }
1225
1226
1227 /**
1228  * Copy an array of handlers.
1229  *
1230  * Useful if the array has been delared in local memory and needs to be
1231  * persisted for future use.
1232  *
1233  * @param handlers Array of handlers to be copied. Can be NULL (nothing done).
1234  * @return A newly allocated array of handlers.
1235  *         Needs to be freed with #GNUNET_free.
1236  */
1237 struct GNUNET_MQ_MessageHandler *
1238 GNUNET_MQ_copy_handlers (const struct GNUNET_MQ_MessageHandler *handlers)
1239 {
1240   struct GNUNET_MQ_MessageHandler *copy;
1241   unsigned int count;
1242
1243   if (NULL == handlers)
1244     return NULL;
1245
1246   count = GNUNET_MQ_count_handlers (handlers);
1247   copy = GNUNET_new_array (count + 1,
1248                            struct GNUNET_MQ_MessageHandler);
1249   GNUNET_memcpy (copy,
1250                  handlers,
1251                  count * sizeof (struct GNUNET_MQ_MessageHandler));
1252   return copy;
1253 }
1254
1255
1256 /**
1257  * Copy an array of handlers, appending AGPL handler.
1258  *
1259  * Useful if the array has been delared in local memory and needs to be
1260  * persisted for future use.
1261  *
1262  * @param handlers Array of handlers to be copied. Can be NULL (nothing done).
1263  * @param agpl_handler function to call for AGPL handling
1264  * @param agpl_cls closure for @a agpl_handler
1265  * @return A newly allocated array of handlers.
1266  *         Needs to be freed with #GNUNET_free.
1267  */
1268 struct GNUNET_MQ_MessageHandler *
1269 GNUNET_MQ_copy_handlers2 (const struct GNUNET_MQ_MessageHandler *handlers,
1270                           GNUNET_MQ_MessageCallback agpl_handler,
1271                           void *agpl_cls)
1272 {
1273   struct GNUNET_MQ_MessageHandler *copy;
1274   unsigned int count;
1275
1276   if (NULL == handlers)
1277     return NULL;
1278   count = GNUNET_MQ_count_handlers (handlers);
1279   copy = GNUNET_new_array (count + 2,
1280                            struct GNUNET_MQ_MessageHandler);
1281   GNUNET_memcpy (copy,
1282                  handlers,
1283                  count * sizeof (struct GNUNET_MQ_MessageHandler));
1284   copy[count].mv = NULL;
1285   copy[count].cb = agpl_handler;
1286   copy[count].cls = agpl_cls;
1287   copy[count].type = GNUNET_MESSAGE_TYPE_REQUEST_AGPL;
1288   copy[count].expected_size = sizeof (struct GNUNET_MessageHeader);
1289   return copy;
1290 }
1291
1292
1293 /**
1294  * Count the handlers in a handler array.
1295  *
1296  * @param handlers Array of handlers to be counted.
1297  * @return The number of handlers in the array.
1298  */
1299 unsigned int
1300 GNUNET_MQ_count_handlers (const struct GNUNET_MQ_MessageHandler *handlers)
1301 {
1302   unsigned int i;
1303
1304   if (NULL == handlers)
1305     return 0;
1306
1307   for (i=0; NULL != handlers[i].cb; i++) ;
1308
1309   return i;
1310 }
1311
1312
1313 /**
1314  * Convert an `enum GNUNET_MQ_PreferenceType` to a string
1315  *
1316  * @param type the preference type
1317  * @return a string or NULL if invalid
1318  */
1319 const char *
1320 GNUNET_MQ_preference_to_string (enum GNUNET_MQ_PreferenceKind type)
1321 {
1322   switch (type) {
1323   case GNUNET_MQ_PREFERENCE_NONE:
1324     return "NONE";
1325   case GNUNET_MQ_PREFERENCE_BANDWIDTH:
1326     return "BANDWIDTH";
1327   case GNUNET_MQ_PREFERENCE_LATENCY:
1328     return "LATENCY";
1329   case GNUNET_MQ_PREFERENCE_RELIABILITY:
1330     return "RELIABILITY";
1331   };
1332   return NULL;
1333 }
1334
1335
1336 /* end of mq.c */