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 */