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
   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., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, 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, "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  * Implementation-specific state for connection to
 206  * client (MQ for server).
 207  */
 208 struct ServerClientSocketState
 209 {
 210   /**
 211    * Handle of the client that connected to the server.
 212    */
 213   struct GNUNET_SERVER_Client *client;
 214
 215   /**
 216    * Active transmission request to the client.
 217    */
 218   struct GNUNET_SERVER_TransmitHandle *th;
 219 };
 220
 221
 222 /**
 223  * Call the message message handler that was registered
 224  * for the type of the given message in the given message queue.
 225  *
 226  * This function is indended to be used for the implementation
 227  * of message queues.
 228  *
 229  * @param mq message queue with the handlers
 230  * @param mh message to dispatch
 231  */
 232 void
 233 GNUNET_MQ_inject_message (struct GNUNET_MQ_Handle *mq,
 234                           const struct GNUNET_MessageHeader *mh)
 235 {
 236   const struct GNUNET_MQ_MessageHandler *handler;
 237   int handled = GNUNET_NO;
 238   uint16_t msize = ntohs (mh->size);
 239   uint16_t mtype = ntohs (mh->type);
 240
 241   LOG (GNUNET_ERROR_TYPE_DEBUG,
 242        "Received message of type %u and size %u\n",
 243        mtype, msize);
 244
 245   if (NULL == mq->handlers)
 246     goto done;
 247   for (handler = mq->handlers; NULL != handler->cb; handler++)
 248   {
 249     if (handler->type == mtype)
 250     {
 251       handled = GNUNET_YES;
 252       if ( (handler->expected_size > msize) ||
 253            ( (handler->expected_size != msize) &&
 254              (NULL == handler->mv) ) )
 255       {
 256         /* Too small, or not an exact size and
 257            no 'mv' handler to check rest */
 258         LOG (GNUNET_ERROR_TYPE_ERROR,
 259              "Received malformed message of type %u\n",
 260              (unsigned int) handler->type);
 261         GNUNET_MQ_inject_error (mq,
 262                                 GNUNET_MQ_ERROR_MALFORMED);
 263         break;
 264       }
 265       if ( (NULL == handler->mv) ||
 266            (GNUNET_OK ==
 267             handler->mv (handler->cls, mh)) )
 268       {
 269         /* message well-formed, pass to handler */
 270         handler->cb (handler->cls, mh);
 271       }
 272       else
 273       {
 274         /* Message rejected by check routine */
 275         LOG (GNUNET_ERROR_TYPE_ERROR,
 276              "Received malformed message of type %u\n",
 277              (unsigned int) handler->type);
 278         GNUNET_MQ_inject_error (mq,
 279                                 GNUNET_MQ_ERROR_MALFORMED);
 280       }
 281       break;
 282     }
 283   }
 284  done:
 285   if (GNUNET_NO == handled)
 286     LOG (GNUNET_ERROR_TYPE_INFO,
 287          "No handler for message of type %u and size %u\n",
 288          mtype, msize);
 289 }
 290
 291
 292 /**
 293  * Call the error handler of a message queue with the given
 294  * error code.  If there is no error handler, log a warning.
 295  *
 296  * This function is intended to be used by the implementation
 297  * of message queues.
 298  *
 299  * @param mq message queue
 300  * @param error the error type
 301  */
 302 void
 303 GNUNET_MQ_inject_error (struct GNUNET_MQ_Handle *mq,
 304                         enum GNUNET_MQ_Error error)
 305 {
 306   if (NULL == mq->error_handler)
 307   {
 308     LOG (GNUNET_ERROR_TYPE_WARNING,
 309          "Got error %d, but no handler installed\n",
 310          (int) error);
 311     return;
 312   }
 313   mq->error_handler (mq->error_handler_cls,
 314                      error);
 315 }
 316
 317
 318 /**
 319  * Discard the message queue message, free all
 320  * allocated resources. Must be called in the event
 321  * that a message is created but should not actually be sent.
 322  *
 323  * @param mqm the message to discard
 324  */
 325 void
 326 GNUNET_MQ_discard (struct GNUNET_MQ_Envelope *ev)
 327 {
 328   GNUNET_assert (NULL == ev->parent_queue);
 329   GNUNET_free (ev);
 330 }
 331
 332
 333 /**
 334  * Obtain the current length of the message queue.
 335  *
 336  * @param mq queue to inspect
 337  * @return number of queued, non-transmitted messages
 338  */
 339 unsigned int
 340 GNUNET_MQ_get_length (struct GNUNET_MQ_Handle *mq)
 341 {
 342   if (GNUNET_YES != mq->in_flight)
 343   {
 344     return mq->queue_length;
 345   }
 346   return mq->queue_length - 1;
 347 }
 348
 349
 350 /**
 351  * Send a message with the given message queue.
 352  * May only be called once per message.
 353  *
 354  * @param mq message queue
 355  * @param ev the envelope with the message to send.
 356  */
 357 void
 358 GNUNET_MQ_send (struct GNUNET_MQ_Handle *mq,
 359                 struct GNUNET_MQ_Envelope *ev)
 360 {
 361   GNUNET_assert (NULL != mq);
 362   GNUNET_assert (NULL == ev->parent_queue);
 363
 364   mq->queue_length++;
 365   GNUNET_break (mq->queue_length < 10000); /* This would seem like a bug... */
 366   ev->parent_queue = mq;
 367   /* is the implementation busy? queue it! */
 368   if ( (NULL != mq->current_envelope) ||
 369        (NULL != mq->send_task) )
 370   {
 371     GNUNET_CONTAINER_DLL_insert_tail (mq->envelope_head,
 372                                       mq->envelope_tail,
 373                                       ev);
 374     return;
 375   }
 376   GNUNET_assert (NULL == mq->envelope_head);
 377   mq->current_envelope = ev;
 378   mq->send_impl (mq,
 379                  ev->mh,
 380                  mq->impl_state);
 381 }
 382
 383
 384 /**
 385  * Remove the first envelope that has not yet been sent from the message
 386  * queue and return it.
 387  *
 388  * @param mq queue to remove envelope from
 389  * @return NULL if queue is empty (or has no envelope that is not under transmission)
 390  */
 391 struct GNUNET_MQ_Envelope *
 392 GNUNET_MQ_unsent_head (struct GNUNET_MQ_Handle *mq)
 393 {
 394   struct GNUNET_MQ_Envelope *env;
 395
 396   env = mq->envelope_head;
 397   GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
 398                                mq->envelope_tail,
 399                                env);
 400   mq->queue_length--;
 401   env->parent_queue = NULL;
 402   return env;
 403 }
 404
 405
 406 /**
 407  * Function to copy an envelope.  The envelope must not yet
 408  * be in any queue or have any options or callbacks set.
 409  *
 410  * @param env envelope to copy
 411  * @return copy of @a env
 412  */
 413 struct GNUNET_MQ_Envelope *
 414 GNUNET_MQ_env_copy (struct GNUNET_MQ_Envelope *env)
 415 {
 416   GNUNET_assert (NULL == env->next);
 417   GNUNET_assert (NULL == env->parent_queue);
 418   GNUNET_assert (NULL == env->sent_cb);
 419   GNUNET_assert (GNUNET_NO == env->have_custom_options);
 420   return GNUNET_MQ_msg_copy (env->mh);
 421 }
 422
 423
 424 /**
 425  * Send a copy of a message with the given message queue.
 426  * Can be called repeatedly on the same envelope.
 427  *
 428  * @param mq message queue
 429  * @param ev the envelope with the message to send.
 430  */
 431 void
 432 GNUNET_MQ_send_copy (struct GNUNET_MQ_Handle *mq,
 433                      const struct GNUNET_MQ_Envelope *ev)
 434 {
 435   struct GNUNET_MQ_Envelope *env;
 436   uint16_t msize;
 437
 438   msize = ntohs (ev->mh->size);
 439   env = GNUNET_malloc (sizeof (struct GNUNET_MQ_Envelope) +
 440                        msize);
 441   env->mh = (struct GNUNET_MessageHeader *) &env[1];
 442   env->sent_cb = ev->sent_cb;
 443   env->sent_cls = ev->sent_cls;
 444   GNUNET_memcpy (&env[1],
 445           ev->mh,
 446           msize);
 447   GNUNET_MQ_send (mq,
 448                   env);
 449 }
 450
 451
 452 /**
 453  * Task run to call the send implementation for the next queued
 454  * message, if any.  Only useful for implementing message queues,
 455  * results in undefined behavior if not used carefully.
 456  *
 457  * @param cls message queue to send the next message with
 458  */
 459 static void
 460 impl_send_continue (void *cls)
 461 {
 462   struct GNUNET_MQ_Handle *mq = cls;
 463
 464   mq->send_task = NULL;
 465   /* call is only valid if we're actually currently sending
 466    * a message */
 467   if (NULL == mq->envelope_head)
 468     return;
 469   mq->current_envelope = mq->envelope_head;
 470   GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
 471                                mq->envelope_tail,
 472                                mq->current_envelope);
 473   mq->send_impl (mq,
 474                  mq->current_envelope->mh,
 475                  mq->impl_state);
 476 }
 477
 478
 479 /**
 480  * Call the send implementation for the next queued message, if any.
 481  * Only useful for implementing message queues, results in undefined
 482  * behavior if not used carefully.
 483  *
 484  * @param mq message queue to send the next message with
 485  */
 486 void
 487 GNUNET_MQ_impl_send_continue (struct GNUNET_MQ_Handle *mq)
 488 {
 489   struct GNUNET_MQ_Envelope *current_envelope;
 490   GNUNET_SCHEDULER_TaskCallback cb;
 491
 492   GNUNET_assert (0 < mq->queue_length);
 493   mq->queue_length--;
 494   mq->in_flight = GNUNET_NO;
 495   current_envelope = mq->current_envelope;
 496   current_envelope->parent_queue = NULL;
 497   mq->current_envelope = NULL;
 498   GNUNET_assert (NULL == mq->send_task);
 499   mq->send_task = GNUNET_SCHEDULER_add_now (&impl_send_continue,
 500                                             mq);
 501   if (NULL != (cb = current_envelope->sent_cb))
 502   {
 503     current_envelope->sent_cb = NULL;
 504     cb (current_envelope->sent_cls);
 505   }
 506   GNUNET_free (current_envelope);
 507 }
 508
 509
 510 /**
 511  * Call the send notification for the current message, but do not
 512  * try to send the next message until #GNUNET_MQ_impl_send_continue
 513  * is called.
 514  *
 515  * Only useful for implementing message queues, results in undefined
 516  * behavior if not used carefully.
 517  *
 518  * @param mq message queue to send the next message with
 519  */
 520 void
 521 GNUNET_MQ_impl_send_in_flight (struct GNUNET_MQ_Handle *mq)
 522 {
 523   struct GNUNET_MQ_Envelope *current_envelope;
 524   GNUNET_SCHEDULER_TaskCallback cb;
 525
 526   mq->in_flight = GNUNET_YES;
 527   /* call is only valid if we're actually currently sending
 528    * a message */
 529   current_envelope = mq->current_envelope;
 530   GNUNET_assert (NULL != current_envelope);
 531   /* can't call cancel from now on anymore */
 532   current_envelope->parent_queue = NULL;
 533   if (NULL != (cb = current_envelope->sent_cb))
 534   {
 535     current_envelope->sent_cb = NULL;
 536     cb (current_envelope->sent_cls);
 537   }
 538 }
 539
 540
 541 /**
 542  * Create a message queue for the specified handlers.
 543  *
 544  * @param send function the implements sending messages
 545  * @param destroy function that implements destroying the queue
 546  * @param cancel function that implements canceling a message
 547  * @param impl_state for the queue, passed to 'send' and 'destroy'
 548  * @param handlers array of message handlers
 549  * @param error_handler handler for read and write errors
 550  * @param error_handler_cls closure for @a error_handler
 551  * @return a new message queue
 552  */
 553 struct GNUNET_MQ_Handle *
 554 GNUNET_MQ_queue_for_callbacks (GNUNET_MQ_SendImpl send,
 555                                GNUNET_MQ_DestroyImpl destroy,
 556                                GNUNET_MQ_CancelImpl cancel,
 557                                void *impl_state,
 558                                const struct GNUNET_MQ_MessageHandler *handlers,
 559                                GNUNET_MQ_ErrorHandler error_handler,
 560                                void *error_handler_cls)
 561 {
 562   struct GNUNET_MQ_Handle *mq;
 563
 564   mq = GNUNET_new (struct GNUNET_MQ_Handle);
 565   mq->send_impl = send;
 566   mq->destroy_impl = destroy;
 567   mq->cancel_impl = cancel;
 568   mq->handlers = GNUNET_MQ_copy_handlers (handlers);
 569   mq->error_handler = error_handler;
 570   mq->error_handler_cls = error_handler_cls;
 571   mq->impl_state = impl_state;
 572
 573   return mq;
 574 }
 575
 576
 577 /**
 578  * Change the closure argument in all of the `handlers` of the
 579  * @a mq.
 580  *
 581  * @param mq to modify
 582  * @param handlers_cls new closure to use
 583  */
 584 void
 585 GNUNET_MQ_set_handlers_closure (struct GNUNET_MQ_Handle *mq,
 586                                 void *handlers_cls)
 587 {
 588   unsigned int i;
 589
 590   if (NULL == mq->handlers)
 591     return;
 592   for (i=0;NULL != mq->handlers[i].cb; i++)
 593     mq->handlers[i].cls = handlers_cls;
 594 }
 595
 596
 597 /**
 598  * Get the message that should currently be sent.
 599  * Fails if there is no current message.
 600  * Only useful for implementing message queues,
 601  * results in undefined behavior if not used carefully.
 602  *
 603  * @param mq message queue with the current message
 604  * @return message to send, never NULL
 605  */
 606 const struct GNUNET_MessageHeader *
 607 GNUNET_MQ_impl_current (struct GNUNET_MQ_Handle *mq)
 608 {
 609   GNUNET_assert (NULL != mq->current_envelope);
 610   GNUNET_assert (NULL != mq->current_envelope->mh);
 611   return mq->current_envelope->mh;
 612 }
 613
 614
 615 /**
 616  * Get the implementation state associated with the
 617  * message queue.
 618  *
 619  * While the GNUNET_MQ_Impl* callbacks receive the
 620  * implementation state, continuations that are scheduled
 621  * by the implementation function often only have one closure
 622  * argument, with this function it is possible to get at the
 623  * implementation state when only passing the GNUNET_MQ_Handle
 624  * as closure.
 625  *
 626  * @param mq message queue with the current message
 627  * @return message to send, never NULL
 628  */
 629 void *
 630 GNUNET_MQ_impl_state (struct GNUNET_MQ_Handle *mq)
 631 {
 632   return mq->impl_state;
 633 }
 634
 635
 636 struct GNUNET_MQ_Envelope *
 637 GNUNET_MQ_msg_ (struct GNUNET_MessageHeader **mhp,
 638                 uint16_t size,
 639                 uint16_t type)
 640 {
 641   struct GNUNET_MQ_Envelope *ev;
 642
 643   ev = GNUNET_malloc (size + sizeof (struct GNUNET_MQ_Envelope));
 644   ev->mh = (struct GNUNET_MessageHeader *) &ev[1];
 645   ev->mh->size = htons (size);
 646   ev->mh->type = htons (type);
 647   if (NULL != mhp)
 648     *mhp = ev->mh;
 649   return ev;
 650 }
 651
 652
 653 /**
 654  * Create a new envelope by copying an existing message.
 655  *
 656  * @param hdr header of the message to copy
 657  * @return envelope containing @a hdr
 658  */
 659 struct GNUNET_MQ_Envelope *
 660 GNUNET_MQ_msg_copy (const struct GNUNET_MessageHeader *hdr)
 661 {
 662   struct GNUNET_MQ_Envelope *mqm;
 663   uint16_t size = ntohs (hdr->size);
 664
 665   mqm = GNUNET_malloc (sizeof (*mqm) + size);
 666   mqm->mh = (struct GNUNET_MessageHeader *) &mqm[1];
 667   GNUNET_memcpy (mqm->mh,
 668           hdr,
 669           size);
 670   return mqm;
 671 }
 672
 673
 674 /**
 675  * Implementation of the #GNUNET_MQ_msg_nested_mh macro.
 676  *
 677  * @param mhp pointer to the message header pointer that will be changed to allocate at
 678  *        the newly allocated space for the message.
 679  * @param base_size size of the data before the nested message
 680  * @param type type of the message in the envelope
 681  * @param nested_mh the message to append to the message after base_size
 682  */
 683 struct GNUNET_MQ_Envelope *
 684 GNUNET_MQ_msg_nested_mh_ (struct GNUNET_MessageHeader **mhp,
 685                           uint16_t base_size,
 686                           uint16_t type,
 687                           const struct GNUNET_MessageHeader *nested_mh)
 688 {
 689   struct GNUNET_MQ_Envelope *mqm;
 690   uint16_t size;
 691
 692   if (NULL == nested_mh)
 693     return GNUNET_MQ_msg_ (mhp, base_size, type);
 694
 695   size = base_size + ntohs (nested_mh->size);
 696
 697   /* check for uint16_t overflow */
 698   if (size < base_size)
 699     return NULL;
 700
 701   mqm = GNUNET_MQ_msg_ (mhp, size, type);
 702   GNUNET_memcpy ((char *) mqm->mh + base_size,
 703                  nested_mh,
 704                  ntohs (nested_mh->size));
 705
 706   return mqm;
 707 }
 708
 709
 710 /**
 711  * Transmit a queued message to the session's client.
 712  *
 713  * @param cls consensus session
 714  * @param size number of bytes available in @a buf
 715  * @param buf where the callee should write the message
 716  * @return number of bytes written to @a buf
 717  */
 718 static size_t
 719 transmit_queued (void *cls,
 720                  size_t size,
 721                  void *buf)
 722 {
 723   struct GNUNET_MQ_Handle *mq = cls;
 724   struct ServerClientSocketState *state = GNUNET_MQ_impl_state (mq);
 725   const struct GNUNET_MessageHeader *msg = GNUNET_MQ_impl_current (mq);
 726   size_t msg_size;
 727
 728   GNUNET_assert (NULL != buf);
 729   msg_size = ntohs (msg->size);
 730   GNUNET_assert (size >= msg_size);
 731   GNUNET_memcpy (buf, msg, msg_size);
 732   state->th = NULL;
 733
 734   GNUNET_MQ_impl_send_continue (mq);
 735
 736   return msg_size;
 737 }
 738
 739
 740 static void
 741 server_client_destroy_impl (struct GNUNET_MQ_Handle *mq,
 742                             void *impl_state)
 743 {
 744   struct ServerClientSocketState *state = impl_state;
 745
 746   if (NULL != state->th)
 747   {
 748     GNUNET_SERVER_notify_transmit_ready_cancel (state->th);
 749     state->th = NULL;
 750   }
 751
 752   GNUNET_assert (NULL != mq);
 753   GNUNET_assert (NULL != state);
 754   GNUNET_SERVER_client_drop (state->client);
 755   GNUNET_free (state);
 756 }
 757
 758
 759 static void
 760 server_client_send_impl (struct GNUNET_MQ_Handle *mq,
 761                          const struct GNUNET_MessageHeader *msg,
 762                          void *impl_state)
 763 {
 764   GNUNET_assert (NULL != mq);
 765
 766   LOG (GNUNET_ERROR_TYPE_DEBUG,
 767        "Sending message of type %u and size %u\n",
 768        ntohs (msg->type), ntohs (msg->size));
 769
 770   struct ServerClientSocketState *state = impl_state;
 771   state->th = GNUNET_SERVER_notify_transmit_ready (state->client,
 772                                                    ntohs (msg->size),
 773                                                    GNUNET_TIME_UNIT_FOREVER_REL,
 774                                                    &transmit_queued,
 775                                                    mq);
 776 }
 777
 778
 779 struct GNUNET_MQ_Handle *
 780 GNUNET_MQ_queue_for_server_client (struct GNUNET_SERVER_Client *client)
 781 {
 782   struct GNUNET_MQ_Handle *mq;
 783   struct ServerClientSocketState *scss;
 784
 785   mq = GNUNET_new (struct GNUNET_MQ_Handle);
 786   scss = GNUNET_new (struct ServerClientSocketState);
 787   mq->impl_state = scss;
 788   scss->client = client;
 789   GNUNET_SERVER_client_keep (client);
 790   mq->send_impl = &server_client_send_impl;
 791   mq->destroy_impl = &server_client_destroy_impl;
 792   return mq;
 793 }
 794
 795
 796 /**
 797  * Associate the assoc_data in mq with a unique request id.
 798  *
 799  * @param mq message queue, id will be unique for the queue
 800  * @param assoc_data to associate
 801  */
 802 uint32_t
 803 GNUNET_MQ_assoc_add (struct GNUNET_MQ_Handle *mq,
 804                      void *assoc_data)
 805 {
 806   uint32_t id;
 807
 808   if (NULL == mq->assoc_map)
 809   {
 810     mq->assoc_map = GNUNET_CONTAINER_multihashmap32_create (8);
 811     mq->assoc_id = 1;
 812   }
 813   id = mq->assoc_id++;
 814   GNUNET_CONTAINER_multihashmap32_put (mq->assoc_map, id, assoc_data,
 815                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
 816   return id;
 817 }
 818
 819
 820 void *
 821 GNUNET_MQ_assoc_get (struct GNUNET_MQ_Handle *mq,
 822                      uint32_t request_id)
 823 {
 824   if (NULL == mq->assoc_map)
 825     return NULL;
 826   return GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map, request_id);
 827 }
 828
 829
 830 void *
 831 GNUNET_MQ_assoc_remove (struct GNUNET_MQ_Handle *mq,
 832                         uint32_t request_id)
 833 {
 834   void *val;
 835
 836   if (NULL == mq->assoc_map)
 837     return NULL;
 838   val = GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map,
 839                                              request_id);
 840   GNUNET_CONTAINER_multihashmap32_remove_all (mq->assoc_map,
 841                                               request_id);
 842   return val;
 843 }
 844
 845
 846 /**
 847  * Call a callback once the envelope has been sent, that is,
 848  * sending it can not be canceled anymore.
 849  * There can be only one notify sent callback per envelope.
 850  *
 851  * @param ev message to call the notify callback for
 852  * @param cb the notify callback
 853  * @param cb_cls closure for the callback
 854  */
 855 void
 856 GNUNET_MQ_notify_sent (struct GNUNET_MQ_Envelope *ev,
 857                        GNUNET_SCHEDULER_TaskCallback cb,
 858                        void *cb_cls)
 859 {
 860   GNUNET_assert (NULL == ev->sent_cb);
 861   ev->sent_cb = cb;
 862   ev->sent_cls = cb_cls;
 863 }
 864
 865
 866 /**
 867  * Handle we return for callbacks registered to be
 868  * notified when #GNUNET_MQ_destroy() is called on a queue.
 869  */
 870 struct GNUNET_MQ_DestroyNotificationHandle
 871 {
 872   /**
 873    * Kept in a DLL.
 874    */
 875   struct GNUNET_MQ_DestroyNotificationHandle *prev;
 876
 877   /**
 878    * Kept in a DLL.
 879    */
 880   struct GNUNET_MQ_DestroyNotificationHandle *next;
 881
 882   /**
 883    * Queue to notify about.
 884    */
 885   struct GNUNET_MQ_Handle *mq;
 886
 887   /**
 888    * Function to call.
 889    */
 890   GNUNET_SCHEDULER_TaskCallback cb;
 891
 892   /**
 893    * Closure for @e cb.
 894    */
 895   void *cb_cls;
 896 };
 897
 898
 899 /**
 900  * Destroy the message queue.
 901  *
 902  * @param mq message queue to destroy
 903  */
 904 void
 905 GNUNET_MQ_destroy (struct GNUNET_MQ_Handle *mq)
 906 {
 907   struct GNUNET_MQ_DestroyNotificationHandle *dnh;
 908
 909   if (NULL != mq->destroy_impl)
 910   {
 911     mq->destroy_impl (mq, mq->impl_state);
 912   }
 913   if (NULL != mq->send_task)
 914   {
 915     GNUNET_SCHEDULER_cancel (mq->send_task);
 916     mq->send_task = NULL;
 917   }
 918   while (NULL != mq->envelope_head)
 919   {
 920     struct GNUNET_MQ_Envelope *ev;
 921
 922     ev = mq->envelope_head;
 923     ev->parent_queue = NULL;
 924     GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
 925                                  mq->envelope_tail,
 926                                  ev);
 927     GNUNET_assert (0 < mq->queue_length);
 928     mq->queue_length--;
 929     GNUNET_MQ_discard (ev);
 930   }
 931   if (NULL != mq->current_envelope)
 932   {
 933     /* we can only discard envelopes that
 934      * are not queued! */
 935     mq->current_envelope->parent_queue = NULL;
 936     GNUNET_MQ_discard (mq->current_envelope);
 937     mq->current_envelope = NULL;
 938     GNUNET_assert (0 < mq->queue_length);
 939     mq->queue_length--;
 940   }
 941   GNUNET_assert (0 == mq->queue_length);
 942   while (NULL != (dnh = mq->dnh_head))
 943   {
 944     dnh->cb (dnh->cb_cls);
 945     GNUNET_MQ_destroy_notify_cancel (dnh);
 946   }
 947   if (NULL != mq->assoc_map)
 948   {
 949     GNUNET_CONTAINER_multihashmap32_destroy (mq->assoc_map);
 950     mq->assoc_map = NULL;
 951   }
 952   GNUNET_free_non_null (mq->handlers);
 953   GNUNET_free (mq);
 954 }
 955
 956
 957 const struct GNUNET_MessageHeader *
 958 GNUNET_MQ_extract_nested_mh_ (const struct GNUNET_MessageHeader *mh,
 959                               uint16_t base_size)
 960 {
 961   uint16_t whole_size;
 962   uint16_t nested_size;
 963   const struct GNUNET_MessageHeader *nested_msg;
 964
 965   whole_size = ntohs (mh->size);
 966   GNUNET_assert (whole_size >= base_size);
 967   nested_size = whole_size - base_size;
 968   if (0 == nested_size)
 969     return NULL;
 970   if (nested_size < sizeof (struct GNUNET_MessageHeader))
 971   {
 972     GNUNET_break_op (0);
 973     return NULL;
 974   }
 975   nested_msg = (const struct GNUNET_MessageHeader *) ((char *) mh + base_size);
 976   if (ntohs (nested_msg->size) != nested_size)
 977   {
 978     GNUNET_break_op (0);
 979     return NULL;
 980   }
 981   return nested_msg;
 982 }
 983
 984
 985 /**
 986  * Cancel sending the message. Message must have been sent with
 987  * #GNUNET_MQ_send before.  May not be called after the notify sent
 988  * callback has been called
 989  *
 990  * @param ev queued envelope to cancel
 991  */
 992 void
 993 GNUNET_MQ_send_cancel (struct GNUNET_MQ_Envelope *ev)
 994 {
 995   struct GNUNET_MQ_Handle *mq = ev->parent_queue;
 996
 997   GNUNET_assert (NULL != mq);
 998   GNUNET_assert (NULL != mq->cancel_impl);
 999
1000   mq->evacuate_called = GNUNET_NO;
1001
1002   if (mq->current_envelope == ev)
1003   {
1004     /* complex case, we already started with transmitting
1005        the message using the callbacks. */
1006     GNUNET_assert (0 < mq->queue_length);
1007     mq->queue_length--;
1008     mq->cancel_impl (mq,
1009                      mq->impl_state);
1010     /* continue sending the next message, if any */
1011     mq->current_envelope = mq->envelope_head;
1012     if (NULL != mq->current_envelope)
1013     {
1014       GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
1015                                    mq->envelope_tail,
1016                                    mq->current_envelope);
1017       mq->send_impl (mq,
1018                      mq->current_envelope->mh,
1019                      mq->impl_state);
1020     }
1021   }
1022   else
1023   {
1024     /* simple case, message is still waiting in the queue */
1025     GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
1026                                  mq->envelope_tail,
1027                                  ev);
1028     GNUNET_assert (0 < mq->queue_length);
1029     mq->queue_length--;
1030   }
1031
1032   if (GNUNET_YES != mq->evacuate_called)
1033   {
1034     ev->parent_queue = NULL;
1035     ev->mh = NULL;
1036     /* also frees ev */
1037     GNUNET_free (ev);
1038   }
1039 }
1040
1041
1042 /**
1043  * Function to obtain the current envelope
1044  * from within #GNUNET_MQ_SendImpl implementations.
1045  *
1046  * @param mq message queue to interrogate
1047  * @return the current envelope
1048  */
1049 struct GNUNET_MQ_Envelope *
1050 GNUNET_MQ_get_current_envelope (struct GNUNET_MQ_Handle *mq)
1051 {
1052   return mq->current_envelope;
1053 }
1054
1055
1056 /**
1057  * Function to obtain the last envelope in the queue.
1058  *
1059  * @param mq message queue to interrogate
1060  * @return the last envelope in the queue
1061  */
1062 struct GNUNET_MQ_Envelope *
1063 GNUNET_MQ_get_last_envelope (struct GNUNET_MQ_Handle *mq)
1064 {
1065   if (NULL != mq->envelope_tail)
1066     return mq->envelope_tail;
1067
1068   return mq->current_envelope;
1069 }
1070
1071
1072 /**
1073  * Set application-specific options for this envelope.
1074  * Overrides the options set for the queue with
1075  * #GNUNET_MQ_set_options() for this message only.
1076  *
1077  * @param env message to set options for
1078  * @param flags flags to use (meaning is queue-specific)
1079  * @param extra additional buffer for further data (also queue-specific)
1080  */
1081 void
1082 GNUNET_MQ_env_set_options (struct GNUNET_MQ_Envelope *env,
1083                            uint64_t flags,
1084                            const void *extra)
1085 {
1086   env->flags = flags;
1087   env->extra = extra;
1088   env->have_custom_options = GNUNET_YES;
1089 }
1090
1091
1092 /**
1093  * Get application-specific options for this envelope.
1094  *
1095  * @param env message to set options for
1096  * @param[out] flags set to flags to use (meaning is queue-specific)
1097  * @return extra additional buffer for further data (also queue-specific)
1098  */
1099 const void *
1100 GNUNET_MQ_env_get_options (struct GNUNET_MQ_Envelope *env,
1101                            uint64_t *flags)
1102 {
1103   struct GNUNET_MQ_Handle *mq = env->parent_queue;
1104
1105   if (GNUNET_YES == env->have_custom_options)
1106   {
1107     *flags = env->flags;
1108     return env->extra;
1109   }
1110   if (NULL == mq)
1111   {
1112     *flags = 0;
1113     return NULL;
1114   }
1115   *flags = mq->default_flags;
1116   return mq->default_extra;
1117 }
1118
1119
1120 /**
1121  * Set application-specific options for this queue.
1122  *
1123  * @param mq message queue to set options for
1124  * @param flags flags to use (meaning is queue-specific)
1125  * @param extra additional buffer for further data (also queue-specific)
1126  */
1127 void
1128 GNUNET_MQ_set_options (struct GNUNET_MQ_Handle *mq,
1129                        uint64_t flags,
1130                        const void *extra)
1131 {
1132   mq->default_flags = flags;
1133   mq->default_extra = extra;
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  * Count the handlers in a handler array.
1258  *
1259  * @param handlers Array of handlers to be counted.
1260  * @return The number of handlers in the array.
1261  */
1262 unsigned int
1263 GNUNET_MQ_count_handlers (const struct GNUNET_MQ_MessageHandler *handlers)
1264 {
1265   unsigned int i;
1266
1267   if (NULL == handlers)
1268     return 0;
1269
1270   for (i=0; NULL != handlers[i].cb; i++) ;
1271
1272   return i;
1273 }
1274
1275
1276
1277 /* end of mq.c */