src/util/mq.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2012-2014 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      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      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  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, "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    * usually points to the end of the containing GNUNET_MQ_Envelope
  49    */
  50   struct GNUNET_MessageHeader *mh;
  51
  52   /**
  53    * Queue the message is queued in, NULL if message is not queued.
  54    */
  55   struct GNUNET_MQ_Handle *parent_queue;
  56
  57   /**
  58    * Called after the message was sent irrevocably.
  59    */
  60   GNUNET_MQ_NotifyCallback sent_cb;
  61
  62   /**
  63    * Closure for send_cb
  64    */
  65   void *sent_cls;
  66 };
  67
  68
  69 /**
  70  * Handle to a message queue.
  71  */
  72 struct GNUNET_MQ_Handle
  73 {
  74   /**
  75    * Handlers array, or NULL if the queue should not receive messages
  76    */
  77   const struct GNUNET_MQ_MessageHandler *handlers;
  78
  79   /**
  80    * Closure for the handler callbacks,
  81    * as well as for the error handler.
  82    */
  83   void *handlers_cls;
  84
  85   /**
  86    * Actual implementation of message sending,
  87    * called when a message is added
  88    */
  89   GNUNET_MQ_SendImpl send_impl;
  90
  91   /**
  92    * Implementation-dependent queue destruction function
  93    */
  94   GNUNET_MQ_DestroyImpl destroy_impl;
  95
  96   /**
  97    * Implementation-dependent send cancel function
  98    */
  99   GNUNET_MQ_CancelImpl cancel_impl;
 100
 101   /**
 102    * Implementation-specific state
 103    */
 104   void *impl_state;
 105
 106   /**
 107    * Callback will be called when an error occurs.
 108    */
 109   GNUNET_MQ_ErrorHandler error_handler;
 110
 111   /**
 112    * Linked list of messages pending to be sent
 113    */
 114   struct GNUNET_MQ_Envelope *envelope_head;
 115
 116   /**
 117    * Linked list of messages pending to be sent
 118    */
 119   struct GNUNET_MQ_Envelope *envelope_tail;
 120
 121   /**
 122    * Message that is currently scheduled to be
 123    * sent. Not the head of the message queue, as the implementation
 124    * needs to know if sending has been already scheduled or not.
 125    */
 126   struct GNUNET_MQ_Envelope *current_envelope;
 127
 128   /**
 129    * Map of associations, lazily allocated
 130    */
 131   struct GNUNET_CONTAINER_MultiHashMap32 *assoc_map;
 132
 133   /**
 134    * Task scheduled during #GNUNET_MQ_impl_send_continue.
 135    */
 136   GNUNET_SCHEDULER_TaskIdentifier continue_task;
 137
 138   /**
 139    * Next id that should be used for the assoc_map,
 140    * initialized lazily to a random value together with
 141    * assoc_map
 142    */
 143   uint32_t assoc_id;
 144 };
 145
 146
 147 /**
 148  * Implementation-specific state for connection to
 149  * client (MQ for server).
 150  */
 151 struct ServerClientSocketState
 152 {
 153   /**
 154    * Handle of the client that connected to the server.
 155    */
 156   struct GNUNET_SERVER_Client *client;
 157
 158   /**
 159    * Active transmission request to the client.
 160    */
 161   struct GNUNET_SERVER_TransmitHandle* th;
 162 };
 163
 164
 165 /**
 166  * Implementation-specific state for connection to
 167  * service (MQ for clients).
 168  */
 169 struct ClientConnectionState
 170 {
 171   /**
 172    * Did we call receive alread alreadyy?
 173    */
 174   int receive_active;
 175
 176   /**
 177    * Do we also want to receive?
 178    */
 179   int receive_requested;
 180
 181   /**
 182    * Connection to the service.
 183    */
 184   struct GNUNET_CLIENT_Connection *connection;
 185
 186   /**
 187    * Active transmission request (or NULL).
 188    */
 189   struct GNUNET_CLIENT_TransmitHandle *th;
 190 };
 191
 192
 193 /**
 194  * Call the message message handler that was registered
 195  * for the type of the given message in the given message queue.
 196  *
 197  * This function is indended to be used for the implementation
 198  * of message queues.
 199  *
 200  * @param mq message queue with the handlers
 201  * @param mh message to dispatch
 202  */
 203 void
 204 GNUNET_MQ_inject_message (struct GNUNET_MQ_Handle *mq,
 205                           const struct GNUNET_MessageHeader *mh)
 206 {
 207   const struct GNUNET_MQ_MessageHandler *handler;
 208   int handled = GNUNET_NO;
 209
 210   handler = mq->handlers;
 211   if (NULL == handler)
 212   {
 213     LOG (GNUNET_ERROR_TYPE_WARNING,
 214          "No handler for message of type %d\n",
 215          ntohs (mh->type));
 216     return;
 217   }
 218   for (; NULL != handler->cb; handler++)
 219   {
 220     if (handler->type == ntohs (mh->type))
 221     {
 222       handler->cb (mq->handlers_cls, mh);
 223       handled = GNUNET_YES;
 224     }
 225   }
 226
 227   if (GNUNET_NO == handled)
 228     LOG (GNUNET_ERROR_TYPE_WARNING,
 229          "No handler for message of type %d\n",
 230          ntohs (mh->type));
 231 }
 232
 233
 234 /**
 235  * Call the error handler of a message queue with the given
 236  * error code.  If there is no error handler, log a warning.
 237  *
 238  * This function is intended to be used by the implementation
 239  * of message queues.
 240  *
 241  * @param mq message queue
 242  * @param error the error type
 243  */
 244 void
 245 GNUNET_MQ_inject_error (struct GNUNET_MQ_Handle *mq,
 246                         enum GNUNET_MQ_Error error)
 247 {
 248   if (NULL == mq->error_handler)
 249   {
 250     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 251                 "mq: got error %d, but no handler installed\n",
 252                 (int) error);
 253     return;
 254   }
 255   mq->error_handler (mq->handlers_cls, error);
 256 }
 257
 258
 259 void
 260 GNUNET_MQ_discard (struct GNUNET_MQ_Envelope *mqm)
 261 {
 262   GNUNET_assert (NULL == mqm->parent_queue);
 263   GNUNET_free (mqm);
 264 }
 265
 266
 267 /**
 268  * Send a message with the give message queue.
 269  * May only be called once per message.
 270  *
 271  * @param mq message queue
 272  * @param ev the envelope with the message to send.
 273  */
 274 void
 275 GNUNET_MQ_send (struct GNUNET_MQ_Handle *mq,
 276                 struct GNUNET_MQ_Envelope *ev)
 277 {
 278   GNUNET_assert (NULL != mq);
 279   GNUNET_assert (NULL == ev->parent_queue);
 280
 281   ev->parent_queue = mq;
 282   /* is the implementation busy? queue it! */
 283   if (NULL != mq->current_envelope)
 284   {
 285     GNUNET_CONTAINER_DLL_insert_tail (mq->envelope_head,
 286                                       mq->envelope_tail,
 287                                       ev);
 288     return;
 289   }
 290   mq->current_envelope = ev;
 291   mq->send_impl (mq, ev->mh, mq->impl_state);
 292 }
 293
 294
 295 /**
 296  * Task run to call the send implementation for the next queued
 297  * message, if any.  Only useful for implementing message queues,
 298  * results in undefined behavior if not used carefully.
 299  *
 300  * @param cls message queue to send the next message with
 301  * @param tc scheduler context
 302  */
 303 static void
 304 impl_send_continue (void *cls,
 305                     const struct GNUNET_SCHEDULER_TaskContext *tc)
 306 {
 307   struct GNUNET_MQ_Handle *mq = cls;
 308   struct GNUNET_MQ_Envelope *current_envelope;
 309
 310   if ((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
 311     return;
 312
 313   mq->continue_task = GNUNET_SCHEDULER_NO_TASK;
 314   /* call is only valid if we're actually currently sending
 315    * a message */
 316   current_envelope = mq->current_envelope;
 317   GNUNET_assert (NULL != current_envelope);
 318   current_envelope->parent_queue = NULL;
 319   if (NULL == mq->envelope_head)
 320   {
 321     mq->current_envelope = NULL;
 322   }
 323   else
 324   {
 325     mq->current_envelope = mq->envelope_head;
 326     GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
 327                                  mq->envelope_tail,
 328                                  mq->current_envelope);
 329     mq->send_impl (mq, mq->current_envelope->mh, mq->impl_state);
 330   }
 331   if (NULL != current_envelope->sent_cb)
 332     current_envelope->sent_cb (current_envelope->sent_cls);
 333   GNUNET_free (current_envelope);
 334 }
 335
 336
 337 /**
 338  * Call the send implementation for the next queued message,
 339  * if any.
 340  * Only useful for implementing message queues,
 341  * results in undefined behavior if not used carefully.
 342  *
 343  * @param mq message queue to send the next message with
 344  */
 345 void
 346 GNUNET_MQ_impl_send_continue (struct GNUNET_MQ_Handle *mq)
 347 {
 348   GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == mq->continue_task);
 349   mq->continue_task = GNUNET_SCHEDULER_add_now (&impl_send_continue,
 350                                                 mq);
 351 }
 352
 353
 354 /**
 355  * Create a message queue for the specified handlers.
 356  *
 357  * @param send function the implements sending messages
 358  * @param destroy function that implements destroying the queue
 359  * @param cancel function that implements canceling a message
 360  * @param impl_state for the queue, passed to 'send' and 'destroy'
 361  * @param handlers array of message handlers
 362  * @param error_handler handler for read and write errors
 363  * @param cls closure for message handlers and error handler
 364  * @return a new message queue
 365  */
 366 struct GNUNET_MQ_Handle *
 367 GNUNET_MQ_queue_for_callbacks (GNUNET_MQ_SendImpl send,
 368                                GNUNET_MQ_DestroyImpl destroy,
 369                                GNUNET_MQ_CancelImpl cancel,
 370                                void *impl_state,
 371                                const struct GNUNET_MQ_MessageHandler *handlers,
 372                                GNUNET_MQ_ErrorHandler error_handler,
 373                                void *cls)
 374 {
 375   struct GNUNET_MQ_Handle *mq;
 376
 377   mq = GNUNET_new (struct GNUNET_MQ_Handle);
 378   mq->send_impl = send;
 379   mq->destroy_impl = destroy;
 380   mq->cancel_impl = cancel;
 381   mq->handlers = handlers;
 382   mq->handlers_cls = cls;
 383   mq->impl_state = impl_state;
 384
 385   return mq;
 386 }
 387
 388
 389 /**
 390  * Get the message that should currently be sent.
 391  * Fails if there is no current message.
 392  * Only useful for implementing message queues,
 393  * results in undefined behavior if not used carefully.
 394  *
 395  * @param mq message queue with the current message
 396  * @return message to send, never NULL
 397  */
 398 const struct GNUNET_MessageHeader *
 399 GNUNET_MQ_impl_current (struct GNUNET_MQ_Handle *mq)
 400 {
 401   if (NULL == mq->current_envelope)
 402     GNUNET_abort ();
 403   if (NULL == mq->current_envelope->mh)
 404     GNUNET_abort ();
 405   return mq->current_envelope->mh;
 406 }
 407
 408
 409 /**
 410  * Get the implementation state associated with the
 411  * message queue.
 412  *
 413  * While the GNUNET_MQ_Impl* callbacks receive the
 414  * implementation state, continuations that are scheduled
 415  * by the implementation function often only have one closure
 416  * argument, with this function it is possible to get at the
 417  * implementation state when only passing the GNUNET_MQ_Handle
 418  * as closure.
 419  *
 420  * @param mq message queue with the current message
 421  * @return message to send, never NULL
 422  */
 423 void *
 424 GNUNET_MQ_impl_state (struct GNUNET_MQ_Handle *mq)
 425 {
 426   return mq->impl_state;
 427 }
 428
 429
 430 struct GNUNET_MQ_Envelope *
 431 GNUNET_MQ_msg_ (struct GNUNET_MessageHeader **mhp,
 432                 uint16_t size,
 433                 uint16_t type)
 434 {
 435   struct GNUNET_MQ_Envelope *mqm;
 436
 437   mqm = GNUNET_malloc (sizeof *mqm + size);
 438   mqm->mh = (struct GNUNET_MessageHeader *) &mqm[1];
 439   mqm->mh->size = htons (size);
 440   mqm->mh->type = htons (type);
 441   if (NULL != mhp)
 442     *mhp = mqm->mh;
 443   return mqm;
 444 }
 445
 446
 447 /**
 448  * Implementation of the GNUNET_MQ_msg_nested_mh macro.
 449  *
 450  * @param mhp pointer to the message header pointer that will be changed to allocate at
 451  *        the newly allocated space for the message.
 452  * @param base_size size of the data before the nested message
 453  * @param type type of the message in the envelope
 454  * @param nested_mh the message to append to the message after base_size
 455  */
 456 struct GNUNET_MQ_Envelope *
 457 GNUNET_MQ_msg_nested_mh_ (struct GNUNET_MessageHeader **mhp,
 458                           uint16_t base_size,
 459                           uint16_t type,
 460                           const struct GNUNET_MessageHeader *nested_mh)
 461 {
 462   struct GNUNET_MQ_Envelope *mqm;
 463   uint16_t size;
 464
 465   if (NULL == nested_mh)
 466     return GNUNET_MQ_msg_ (mhp, base_size, type);
 467
 468   size = base_size + ntohs (nested_mh->size);
 469
 470   /* check for uint16_t overflow */
 471   if (size < base_size)
 472     return NULL;
 473
 474   mqm = GNUNET_MQ_msg_ (mhp, size, type);
 475   memcpy ((char *) mqm->mh + base_size, nested_mh, ntohs (nested_mh->size));
 476
 477   return mqm;
 478 }
 479
 480
 481 /**
 482  * Transmit a queued message to the session's client.
 483  *
 484  * @param cls consensus session
 485  * @param size number of bytes available in buf
 486  * @param buf where the callee should write the message
 487  * @return number of bytes written to buf
 488  */
 489 static size_t
 490 transmit_queued (void *cls, size_t size,
 491                  void *buf)
 492 {
 493   struct GNUNET_MQ_Handle *mq = cls;
 494   struct ServerClientSocketState *state = GNUNET_MQ_impl_state (mq);
 495   const struct GNUNET_MessageHeader *msg = GNUNET_MQ_impl_current (mq);
 496   size_t msg_size;
 497
 498   GNUNET_assert (NULL != buf);
 499
 500   msg_size = ntohs (msg->size);
 501   GNUNET_assert (size >= msg_size);
 502   memcpy (buf, msg, msg_size);
 503   state->th = NULL;
 504
 505   GNUNET_MQ_impl_send_continue (mq);
 506
 507   return msg_size;
 508 }
 509
 510
 511 static void
 512 server_client_destroy_impl (struct GNUNET_MQ_Handle *mq,
 513                             void *impl_state)
 514 {
 515   struct ServerClientSocketState *state = impl_state;
 516
 517   if (NULL != state->th)
 518   {
 519     GNUNET_SERVER_notify_transmit_ready_cancel (state->th);
 520     state->th = NULL;
 521   }
 522
 523   GNUNET_assert (NULL != mq);
 524   GNUNET_assert (NULL != state);
 525   GNUNET_SERVER_client_drop (state->client);
 526   GNUNET_free (state);
 527 }
 528
 529
 530 static void
 531 server_client_send_impl (struct GNUNET_MQ_Handle *mq,
 532                          const struct GNUNET_MessageHeader *msg,
 533                          void *impl_state)
 534 {
 535   struct ServerClientSocketState *state = impl_state;
 536
 537   GNUNET_assert (NULL != mq);
 538   GNUNET_assert (NULL != state);
 539   state->th =
 540       GNUNET_SERVER_notify_transmit_ready (state->client, ntohs (msg->size),
 541                                            GNUNET_TIME_UNIT_FOREVER_REL,
 542                                            &transmit_queued, mq);
 543 }
 544
 545
 546 struct GNUNET_MQ_Handle *
 547 GNUNET_MQ_queue_for_server_client (struct GNUNET_SERVER_Client *client)
 548 {
 549   struct GNUNET_MQ_Handle *mq;
 550   struct ServerClientSocketState *scss;
 551
 552   mq = GNUNET_new (struct GNUNET_MQ_Handle);
 553   scss = GNUNET_new (struct ServerClientSocketState);
 554   mq->impl_state = scss;
 555   scss->client = client;
 556   GNUNET_SERVER_client_keep (client);
 557   mq->send_impl = server_client_send_impl;
 558   mq->destroy_impl = server_client_destroy_impl;
 559   return mq;
 560 }
 561
 562
 563 /**
 564  * Type of a function to call when we receive a message
 565  * from the service.
 566  *
 567  * @param cls closure
 568  * @param msg message received, NULL on timeout or fatal error
 569  */
 570 static void
 571 handle_client_message (void *cls,
 572                        const struct GNUNET_MessageHeader *msg)
 573 {
 574   struct GNUNET_MQ_Handle *mq = cls;
 575   struct ClientConnectionState *state;
 576
 577   state = mq->impl_state;
 578
 579   if (NULL == msg)
 580   {
 581     GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_READ);
 582     return;
 583   }
 584
 585   GNUNET_CLIENT_receive (state->connection, handle_client_message, mq,
 586                          GNUNET_TIME_UNIT_FOREVER_REL);
 587
 588   GNUNET_MQ_inject_message (mq, msg);
 589 }
 590
 591
 592 /**
 593  * Transmit a queued message to the session's client.
 594  *
 595  * @param cls consensus session
 596  * @param size number of bytes available in @a buf
 597  * @param buf where the callee should write the message
 598  * @return number of bytes written to buf
 599  */
 600 static size_t
 601 connection_client_transmit_queued (void *cls,
 602                                    size_t size,
 603                                    void *buf)
 604 {
 605   struct GNUNET_MQ_Handle *mq = cls;
 606   const struct GNUNET_MessageHeader *msg;
 607   struct ClientConnectionState *state = mq->impl_state;
 608   size_t msg_size;
 609
 610   GNUNET_assert (NULL != mq);
 611   msg = GNUNET_MQ_impl_current (mq);
 612
 613   if (NULL == buf)
 614   {
 615     GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_READ);
 616     return 0;
 617   }
 618
 619   if ( (GNUNET_YES == state->receive_requested) &&
 620        (GNUNET_NO == state->receive_active) )
 621   {
 622     state->receive_active = GNUNET_YES;
 623     GNUNET_CLIENT_receive (state->connection, handle_client_message, mq,
 624                            GNUNET_TIME_UNIT_FOREVER_REL);
 625   }
 626
 627   msg_size = ntohs (msg->size);
 628   GNUNET_assert (size >= msg_size);
 629   memcpy (buf, msg, msg_size);
 630   state->th = NULL;
 631
 632   GNUNET_MQ_impl_send_continue (mq);
 633
 634   return msg_size;
 635 }
 636
 637
 638 static void
 639 connection_client_destroy_impl (struct GNUNET_MQ_Handle *mq,
 640                                 void *impl_state)
 641 {
 642   GNUNET_free (impl_state);
 643 }
 644
 645
 646 static void
 647 connection_client_send_impl (struct GNUNET_MQ_Handle *mq,
 648                              const struct GNUNET_MessageHeader *msg,
 649                              void *impl_state)
 650 {
 651   struct ClientConnectionState *state = impl_state;
 652
 653   GNUNET_assert (NULL != state);
 654   GNUNET_assert (NULL == state->th);
 655   state->th =
 656       GNUNET_CLIENT_notify_transmit_ready (state->connection, ntohs (msg->size),
 657                                            GNUNET_TIME_UNIT_FOREVER_REL, GNUNET_NO,
 658                                            &connection_client_transmit_queued, mq);
 659   GNUNET_assert (NULL != state->th);
 660 }
 661
 662
 663 static void
 664 connection_client_cancel_impl (struct GNUNET_MQ_Handle *mq,
 665                                void *impl_state)
 666 {
 667   struct ClientConnectionState *state = impl_state;
 668   GNUNET_assert (NULL != state->th);
 669   GNUNET_CLIENT_notify_transmit_ready_cancel (state->th);
 670   state->th = NULL;
 671 }
 672
 673
 674 struct GNUNET_MQ_Handle *
 675 GNUNET_MQ_queue_for_connection_client (struct GNUNET_CLIENT_Connection *connection,
 676                                        const struct GNUNET_MQ_MessageHandler *handlers,
 677                                        GNUNET_MQ_ErrorHandler error_handler,
 678                                        void *cls)
 679 {
 680   struct GNUNET_MQ_Handle *mq;
 681   struct ClientConnectionState *state;
 682
 683   GNUNET_assert (NULL != connection);
 684
 685   mq = GNUNET_new (struct GNUNET_MQ_Handle);
 686   mq->handlers = handlers;
 687   mq->error_handler = error_handler;
 688   mq->handlers_cls = cls;
 689   state = GNUNET_new (struct ClientConnectionState);
 690   state->connection = connection;
 691   mq->impl_state = state;
 692   mq->send_impl = connection_client_send_impl;
 693   mq->destroy_impl = connection_client_destroy_impl;
 694   mq->cancel_impl = connection_client_cancel_impl;
 695   if (NULL != handlers)
 696     state->receive_requested = GNUNET_YES;
 697
 698   return mq;
 699 }
 700
 701
 702 void
 703 GNUNET_MQ_replace_handlers (struct GNUNET_MQ_Handle *mq,
 704                             const struct GNUNET_MQ_MessageHandler *new_handlers,
 705                             void *cls)
 706 {
 707   /* FIXME: notify implementation? */
 708   /* FIXME: what about NULL handlers? abort receive? */
 709   mq->handlers = new_handlers;
 710   mq->handlers_cls = cls;
 711 }
 712
 713
 714 /**
 715  * Associate the assoc_data in mq with a unique request id.
 716  *
 717  * @param mq message queue, id will be unique for the queue
 718  * @param assoc_data to associate
 719  */
 720 uint32_t
 721 GNUNET_MQ_assoc_add (struct GNUNET_MQ_Handle *mq,
 722                      void *assoc_data)
 723 {
 724   uint32_t id;
 725
 726   if (NULL == mq->assoc_map)
 727   {
 728     mq->assoc_map = GNUNET_CONTAINER_multihashmap32_create (8);
 729     mq->assoc_id = 1;
 730   }
 731   id = mq->assoc_id++;
 732   GNUNET_CONTAINER_multihashmap32_put (mq->assoc_map, id, assoc_data,
 733                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
 734   return id;
 735 }
 736
 737
 738 void *
 739 GNUNET_MQ_assoc_get (struct GNUNET_MQ_Handle *mq,
 740                      uint32_t request_id)
 741 {
 742   if (NULL == mq->assoc_map)
 743     return NULL;
 744   return GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map, request_id);
 745 }
 746
 747
 748 void *
 749 GNUNET_MQ_assoc_remove (struct GNUNET_MQ_Handle *mq,
 750                         uint32_t request_id)
 751 {
 752   void *val;
 753
 754   if (NULL == mq->assoc_map)
 755     return NULL;
 756   val = GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map, request_id);
 757   GNUNET_CONTAINER_multihashmap32_remove_all (mq->assoc_map, request_id);
 758   return val;
 759 }
 760
 761
 762 void
 763 GNUNET_MQ_notify_sent (struct GNUNET_MQ_Envelope *mqm,
 764                        GNUNET_MQ_NotifyCallback cb,
 765                        void *cls)
 766 {
 767   mqm->sent_cb = cb;
 768   mqm->sent_cls = cls;
 769 }
 770
 771
 772 void
 773 GNUNET_MQ_destroy (struct GNUNET_MQ_Handle *mq)
 774 {
 775   if (NULL != mq->destroy_impl)
 776   {
 777     mq->destroy_impl (mq, mq->impl_state);
 778   }
 779   if (GNUNET_SCHEDULER_NO_TASK != mq->continue_task)
 780   {
 781     GNUNET_SCHEDULER_cancel (mq->continue_task);
 782     mq->continue_task = GNUNET_SCHEDULER_NO_TASK;
 783   }
 784   while (NULL != mq->envelope_head)
 785   {
 786     struct GNUNET_MQ_Envelope *ev;
 787     ev = mq->envelope_head;
 788     ev->parent_queue = NULL;
 789     GNUNET_CONTAINER_DLL_remove (mq->envelope_head, mq->envelope_tail, ev);
 790     GNUNET_MQ_discard (ev);
 791   }
 792
 793   if (NULL != mq->current_envelope)
 794   {
 795     /* we can only discard envelopes that
 796      * are not queued! */
 797     mq->current_envelope->parent_queue = NULL;
 798     GNUNET_MQ_discard (mq->current_envelope);
 799     mq->current_envelope = NULL;
 800   }
 801
 802   if (NULL != mq->assoc_map)
 803   {
 804     GNUNET_CONTAINER_multihashmap32_destroy (mq->assoc_map);
 805     mq->assoc_map = NULL;
 806   }
 807
 808   GNUNET_free (mq);
 809 }
 810
 811
 812 const struct GNUNET_MessageHeader *
 813 GNUNET_MQ_extract_nested_mh_ (const struct GNUNET_MessageHeader *mh,
 814                               uint16_t base_size)
 815 {
 816   uint16_t whole_size;
 817   uint16_t nested_size;
 818   const struct GNUNET_MessageHeader *nested_msg;
 819
 820   whole_size = ntohs (mh->size);
 821   GNUNET_assert (whole_size >= base_size);
 822   nested_size = whole_size - base_size;
 823   if (0 == nested_size)
 824     return NULL;
 825   if (nested_size < sizeof (struct GNUNET_MessageHeader))
 826   {
 827     GNUNET_break_op (0);
 828     return NULL;
 829   }
 830   nested_msg = (const struct GNUNET_MessageHeader *) ((char *) mh + base_size);
 831   if (ntohs (nested_msg->size) != nested_size)
 832   {
 833     GNUNET_break_op (0);
 834     return NULL;
 835   }
 836   return nested_msg;
 837 }
 838
 839
 840 /**
 841  * Cancel sending the message. Message must have been sent with
 842  * #GNUNET_MQ_send before.  May not be called after the notify sent
 843  * callback has been called
 844  *
 845  * @param ev queued envelope to cancel
 846  */
 847 void
 848 GNUNET_MQ_send_cancel (struct GNUNET_MQ_Envelope *ev)
 849 {
 850   struct GNUNET_MQ_Handle *mq = ev->parent_queue;
 851
 852   GNUNET_assert (NULL != mq);
 853   GNUNET_assert (NULL != mq->cancel_impl);
 854
 855   if (mq->current_envelope == ev) {
 856     // complex case, we already started with transmitting
 857     // the message
 858     mq->cancel_impl (mq, mq->impl_state);
 859     // continue sending the next message, if any
 860     if (NULL == mq->envelope_head)
 861     {
 862       mq->current_envelope = NULL;
 863     }
 864     else
 865     {
 866       mq->current_envelope = mq->envelope_head;
 867       GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
 868                                    mq->envelope_tail,
 869                                    mq->current_envelope);
 870       mq->send_impl (mq, mq->current_envelope->mh, mq->impl_state);
 871     }
 872   } else {
 873     // simple case, message is still waiting in the queue
 874     GNUNET_CONTAINER_DLL_remove (mq->envelope_head, mq->envelope_tail, ev);
 875   }
 876
 877   ev->parent_queue = NULL;
 878   ev->mh = NULL;
 879   GNUNET_free (ev);
 880 }
 881
 882 /* end of mq.c */