src/util/mq.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2012-2014 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, "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_MQ_NotifyCallback 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    * Linked list of messages pending to be sent
 132    */
 133   struct GNUNET_MQ_Envelope *envelope_head;
 134
 135   /**
 136    * Linked list of messages pending to be sent
 137    */
 138   struct GNUNET_MQ_Envelope *envelope_tail;
 139
 140   /**
 141    * Message that is currently scheduled to be
 142    * sent. Not the head of the message queue, as the implementation
 143    * needs to know if sending has been already scheduled or not.
 144    */
 145   struct GNUNET_MQ_Envelope *current_envelope;
 146
 147   /**
 148    * GNUNET_YES if the sent notification was called
 149    * for the current envelope.
 150    */
 151   int send_notification_called;
 152
 153   /**
 154    * Map of associations, lazily allocated
 155    */
 156   struct GNUNET_CONTAINER_MultiHashMap32 *assoc_map;
 157
 158   /**
 159    * Task scheduled during #GNUNET_MQ_impl_send_continue
 160    * or #GNUNET_MQ_impl_send_in_flight
 161    */
 162   struct GNUNET_SCHEDULER_Task *send_task;
 163
 164   /**
 165    * Functions to call on queue destruction; kept in a DLL.
 166    */
 167   struct GNUNET_MQ_DestroyNotificationHandle *dnh_head;
 168
 169   /**
 170    * Functions to call on queue destruction; kept in a DLL.
 171    */
 172   struct GNUNET_MQ_DestroyNotificationHandle *dnh_tail;
 173
 174   /**
 175    * Additional options buffer set for this queue by
 176    * #GNUNET_MQ_set_options().  Default is 0.
 177    */
 178   const void *default_extra;
 179
 180   /**
 181    * Flags that were set for this queue by
 182    * #GNUNET_MQ_set_options().   Default is 0.
 183    */
 184   uint64_t default_flags;
 185
 186   /**
 187    * Next id that should be used for the @e assoc_map,
 188    * initialized lazily to a random value together with
 189    * @e assoc_map
 190    */
 191   uint32_t assoc_id;
 192
 193   /**
 194    * Number of entries we have in the envelope-DLL.
 195    */
 196   unsigned int queue_length;
 197
 198   /**
 199    * GNUNET_YES if GNUNET_MQ_impl_evacuate was called.
 200    */
 201   int evacuate_called;
 202 };
 203
 204
 205 /**
 206  * Implementation-specific state for connection to
 207  * client (MQ for server).
 208  */
 209 struct ServerClientSocketState
 210 {
 211   /**
 212    * Handle of the client that connected to the server.
 213    */
 214   struct GNUNET_SERVER_Client *client;
 215
 216   /**
 217    * Active transmission request to the client.
 218    */
 219   struct GNUNET_SERVER_TransmitHandle *th;
 220 };
 221
 222
 223 /**
 224  * Implementation-specific state for connection to
 225  * service (MQ for clients).
 226  */
 227 struct ClientConnectionState
 228 {
 229   /**
 230    * Did we call receive alread alreadyy?
 231    */
 232   int receive_active;
 233
 234   /**
 235    * Do we also want to receive?
 236    */
 237   int receive_requested;
 238
 239   /**
 240    * Connection to the service.
 241    */
 242   struct GNUNET_CLIENT_Connection *connection;
 243
 244   /**
 245    * Active transmission request (or NULL).
 246    */
 247   struct GNUNET_CLIENT_TransmitHandle *th;
 248 };
 249
 250
 251 /**
 252  * Call the message message handler that was registered
 253  * for the type of the given message in the given message queue.
 254  *
 255  * This function is indended to be used for the implementation
 256  * of message queues.
 257  *
 258  * @param mq message queue with the handlers
 259  * @param mh message to dispatch
 260  */
 261 void
 262 GNUNET_MQ_inject_message (struct GNUNET_MQ_Handle *mq,
 263                           const struct GNUNET_MessageHeader *mh)
 264 {
 265   const struct GNUNET_MQ_MessageHandler *handler;
 266   int handled = GNUNET_NO;
 267   uint16_t ms = ntohs (mh->size);
 268
 269   if (NULL == mq->handlers)
 270     goto done;
 271   for (handler = mq->handlers; NULL != handler->cb; handler++)
 272   {
 273     if (handler->type == ntohs (mh->type))
 274     {
 275       handled = GNUNET_YES;
 276       if ( (handler->expected_size > ms) ||
 277            ( (handler->expected_size != ms) &&
 278              (NULL == handler->mv) ) )
 279       {
 280         /* Too small, or not an exact size and
 281            no 'mv' handler to check rest */
 282         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 283                     "Received malformed message of type %u\n",
 284                     (unsigned int) handler->type);
 285         GNUNET_MQ_inject_error (mq,
 286                                 GNUNET_MQ_ERROR_MALFORMED);
 287         break;
 288       }
 289       if ( (NULL == handler->mv) ||
 290            (GNUNET_OK ==
 291             handler->mv (handler->cls, mh)) )
 292       {
 293         /* message well-formed, pass to handler */
 294         handler->cb (handler->cls, mh);
 295       }
 296       else
 297       {
 298         /* Message rejected by check routine */
 299         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 300                     "Received malformed message of type %u\n",
 301                     (unsigned int) handler->type);
 302         GNUNET_MQ_inject_error (mq,
 303                                 GNUNET_MQ_ERROR_MALFORMED);
 304       }
 305       break;
 306     }
 307   }
 308  done:
 309   if (GNUNET_NO == handled)
 310     LOG (GNUNET_ERROR_TYPE_INFO,
 311          "No handler for message of type %d and size %d\n",
 312          ntohs (mh->type),
 313          ntohs (mh->size));
 314 }
 315
 316
 317 /**
 318  * Call the error handler of a message queue with the given
 319  * error code.  If there is no error handler, log a warning.
 320  *
 321  * This function is intended to be used by the implementation
 322  * of message queues.
 323  *
 324  * @param mq message queue
 325  * @param error the error type
 326  */
 327 void
 328 GNUNET_MQ_inject_error (struct GNUNET_MQ_Handle *mq,
 329                         enum GNUNET_MQ_Error error)
 330 {
 331   if (NULL == mq->error_handler)
 332   {
 333     LOG (GNUNET_ERROR_TYPE_WARNING,
 334          "Got error %d, but no handler installed\n",
 335          (int) error);
 336     return;
 337   }
 338   mq->error_handler (mq->error_handler_cls,
 339                      error);
 340 }
 341
 342
 343 /**
 344  * Discard the message queue message, free all
 345  * allocated resources. Must be called in the event
 346  * that a message is created but should not actually be sent.
 347  *
 348  * @param mqm the message to discard
 349  */
 350 void
 351 GNUNET_MQ_discard (struct GNUNET_MQ_Envelope *ev)
 352 {
 353   GNUNET_assert (NULL == ev->parent_queue);
 354   GNUNET_free (ev);
 355 }
 356
 357
 358 /**
 359  * Obtain the current length of the message queue.
 360  *
 361  * @param mq queue to inspect
 362  * @return number of queued, non-transmitted messages
 363  */
 364 unsigned int
 365 GNUNET_MQ_get_length (struct GNUNET_MQ_Handle *mq)
 366 {
 367   return mq->queue_length;
 368 }
 369
 370
 371 /**
 372  * Send a message with the given message queue.
 373  * May only be called once per message.
 374  *
 375  * @param mq message queue
 376  * @param ev the envelope with the message to send.
 377  */
 378 void
 379 GNUNET_MQ_send (struct GNUNET_MQ_Handle *mq,
 380                 struct GNUNET_MQ_Envelope *ev)
 381 {
 382   GNUNET_assert (NULL != mq);
 383   GNUNET_assert (NULL == ev->parent_queue);
 384
 385   mq->queue_length++;
 386   ev->parent_queue = mq;
 387   /* is the implementation busy? queue it! */
 388   if (NULL != mq->current_envelope)
 389   {
 390     GNUNET_CONTAINER_DLL_insert_tail (mq->envelope_head,
 391                                       mq->envelope_tail,
 392                                       ev);
 393     return;
 394   }
 395   mq->current_envelope = ev;
 396   mq->send_impl (mq,
 397                  ev->mh,
 398                  mq->impl_state);
 399 }
 400
 401
 402 /**
 403  * Send a copy of a message with the given message queue.
 404  * Can be called repeatedly on the same envelope.
 405  *
 406  * @param mq message queue
 407  * @param ev the envelope with the message to send.
 408  */
 409 void
 410 GNUNET_MQ_send_copy (struct GNUNET_MQ_Handle *mq,
 411                      const struct GNUNET_MQ_Envelope *ev)
 412 {
 413   struct GNUNET_MQ_Envelope *env;
 414   uint16_t msize;
 415
 416   msize = ntohs (ev->mh->size);
 417   env = GNUNET_malloc (sizeof (struct GNUNET_MQ_Envelope) +
 418                        msize);
 419   env->mh = (struct GNUNET_MessageHeader *) &env[1];
 420   env->sent_cb = ev->sent_cb;
 421   env->sent_cls = ev->sent_cls;
 422   GNUNET_memcpy (&env[1],
 423           ev->mh,
 424           msize);
 425   GNUNET_MQ_send (mq,
 426                   env);
 427 }
 428
 429
 430 /**
 431  * Task run to call the send notification for the next queued
 432  * message, if any.  Only useful for implementing message queues,
 433  * results in undefined behavior if not used carefully.
 434  *
 435  * @param cls message queue to send the next message with
 436  */
 437 static void
 438 impl_send_in_flight (void *cls)
 439 {
 440   struct GNUNET_MQ_Handle *mq = cls;
 441   struct GNUNET_MQ_Envelope *current_envelope;
 442
 443   mq->send_task = NULL;
 444   /* call is only valid if we're actually currently sending
 445    * a message */
 446   current_envelope = mq->current_envelope;
 447   GNUNET_assert (NULL != current_envelope);
 448   /* can't call cancel from now on anymore */
 449   current_envelope->parent_queue = NULL;
 450   if ( (GNUNET_NO == mq->send_notification_called) &&
 451        (NULL != current_envelope->sent_cb) )
 452   {
 453     current_envelope->sent_cb (current_envelope->sent_cls);
 454   }
 455   mq->send_notification_called = GNUNET_YES;
 456 }
 457
 458
 459 /**
 460  * Task run to call the send implementation for the next queued
 461  * message, if any.  Only useful for implementing message queues,
 462  * results in undefined behavior if not used carefully.
 463  *
 464  * @param cls message queue to send the next message with
 465  */
 466 static void
 467 impl_send_continue (void *cls)
 468 {
 469   struct GNUNET_MQ_Handle *mq = cls;
 470   struct GNUNET_MQ_Envelope *current_envelope;
 471
 472   mq->send_task = NULL;
 473   /* call is only valid if we're actually currently sending
 474    * a message */
 475   current_envelope = mq->current_envelope;
 476   GNUNET_assert (NULL != current_envelope);
 477   impl_send_in_flight (mq);
 478   GNUNET_assert (0 < mq->queue_length);
 479   mq->queue_length--;
 480   if (NULL == mq->envelope_head)
 481   {
 482     mq->current_envelope = NULL;
 483   }
 484   else
 485   {
 486     mq->current_envelope = mq->envelope_head;
 487     GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
 488                                  mq->envelope_tail,
 489                                  mq->current_envelope);
 490     mq->send_notification_called = GNUNET_NO;
 491     mq->send_impl (mq,
 492                    mq->current_envelope->mh,
 493                    mq->impl_state);
 494   }
 495   GNUNET_free (current_envelope);
 496 }
 497
 498
 499 /**
 500  * Call the send implementation for the next queued message, if any.
 501  * Only useful for implementing message queues, results in undefined
 502  * behavior if not used carefully.
 503  *
 504  * @param mq message queue to send the next message with
 505  */
 506 void
 507 GNUNET_MQ_impl_send_continue (struct GNUNET_MQ_Handle *mq)
 508 {
 509   /* maybe #GNUNET_MQ_impl_send_in_flight was called? */
 510   if (NULL != mq->send_task)
 511   {
 512     GNUNET_SCHEDULER_cancel (mq->send_task);
 513   }
 514   mq->send_task = GNUNET_SCHEDULER_add_now (&impl_send_continue,
 515                                             mq);
 516 }
 517
 518
 519 /**
 520  * Call the send notification for the current message, but do not
 521  * try to send the message until #gnunet_mq_impl_send_continue
 522  * is called.
 523  *
 524  * only useful for implementing message queues, results in undefined
 525  * behavior if not used carefully.
 526  *
 527  * @param mq message queue to send the next message with
 528  */
 529 void
 530 GNUNET_MQ_impl_send_in_flight (struct GNUNET_MQ_Handle *mq)
 531 {
 532   GNUNET_assert (NULL == mq->send_task);
 533   mq->send_task = GNUNET_SCHEDULER_add_now (&impl_send_in_flight,
 534                                             mq);
 535 }
 536
 537
 538 /**
 539  * Create a message queue for the specified handlers.
 540  *
 541  * @param send function the implements sending messages
 542  * @param destroy function that implements destroying the queue
 543  * @param cancel function that implements canceling a message
 544  * @param impl_state for the queue, passed to 'send' and 'destroy'
 545  * @param handlers array of message handlers
 546  * @param error_handler handler for read and write errors
 547  * @param error_handler_cls closure for @a error_handler
 548  * @return a new message queue
 549  */
 550 struct GNUNET_MQ_Handle *
 551 GNUNET_MQ_queue_for_callbacks (GNUNET_MQ_SendImpl send,
 552                                GNUNET_MQ_DestroyImpl destroy,
 553                                GNUNET_MQ_CancelImpl cancel,
 554                                void *impl_state,
 555                                const struct GNUNET_MQ_MessageHandler *handlers,
 556                                GNUNET_MQ_ErrorHandler error_handler,
 557                                void *error_handler_cls)
 558 {
 559   struct GNUNET_MQ_Handle *mq;
 560   unsigned int i;
 561
 562   mq = GNUNET_new (struct GNUNET_MQ_Handle);
 563   mq->send_impl = send;
 564   mq->destroy_impl = destroy;
 565   mq->cancel_impl = cancel;
 566   if (NULL != handlers)
 567   {
 568     for (i=0;NULL != handlers[i].cb; i++) ;
 569     mq->handlers = GNUNET_new_array (i + 1,
 570                                      struct GNUNET_MQ_MessageHandler);
 571     GNUNET_memcpy (mq->handlers,
 572             handlers,
 573             i * sizeof (struct GNUNET_MQ_MessageHandler));
 574   }
 575   mq->error_handler = error_handler;
 576   mq->error_handler_cls = error_handler_cls;
 577   mq->impl_state = impl_state;
 578
 579   return mq;
 580 }
 581
 582
 583 /**
 584  * Change the closure argument in all of the `handlers` of the
 585  * @a mq.
 586  *
 587  * @param mq to modify
 588  * @param handlers_cls new closure to use
 589  */
 590 void
 591 GNUNET_MQ_set_handlers_closure (struct GNUNET_MQ_Handle *mq,
 592                                 void *handlers_cls)
 593 {
 594   unsigned int i;
 595
 596   if (NULL == mq->handlers)
 597     return;
 598   for (i=0;NULL != mq->handlers[i].cb; i++)
 599     mq->handlers[i].cls = handlers_cls;
 600 }
 601
 602
 603 /**
 604  * Get the message that should currently be sent.
 605  * Fails if there is no current message.
 606  * Only useful for implementing message queues,
 607  * results in undefined behavior if not used carefully.
 608  *
 609  * @param mq message queue with the current message
 610  * @return message to send, never NULL
 611  */
 612 const struct GNUNET_MessageHeader *
 613 GNUNET_MQ_impl_current (struct GNUNET_MQ_Handle *mq)
 614 {
 615   if (NULL == mq->current_envelope)
 616     GNUNET_assert (0);
 617   if (NULL == mq->current_envelope->mh)
 618     GNUNET_assert (0);
 619   return mq->current_envelope->mh;
 620 }
 621
 622
 623 /**
 624  * Get the implementation state associated with the
 625  * message queue.
 626  *
 627  * While the GNUNET_MQ_Impl* callbacks receive the
 628  * implementation state, continuations that are scheduled
 629  * by the implementation function often only have one closure
 630  * argument, with this function it is possible to get at the
 631  * implementation state when only passing the GNUNET_MQ_Handle
 632  * as closure.
 633  *
 634  * @param mq message queue with the current message
 635  * @return message to send, never NULL
 636  */
 637 void *
 638 GNUNET_MQ_impl_state (struct GNUNET_MQ_Handle *mq)
 639 {
 640   return mq->impl_state;
 641 }
 642
 643
 644 struct GNUNET_MQ_Envelope *
 645 GNUNET_MQ_msg_ (struct GNUNET_MessageHeader **mhp,
 646                 uint16_t size,
 647                 uint16_t type)
 648 {
 649   struct GNUNET_MQ_Envelope *ev;
 650
 651   ev = GNUNET_malloc (size + sizeof (struct GNUNET_MQ_Envelope));
 652   ev->mh = (struct GNUNET_MessageHeader *) &ev[1];
 653   ev->mh->size = htons (size);
 654   ev->mh->type = htons (type);
 655   if (NULL != mhp)
 656     *mhp = ev->mh;
 657   return ev;
 658 }
 659
 660
 661 /**
 662  * Create a new envelope by copying an existing message.
 663  *
 664  * @param hdr header of the message to copy
 665  * @return envelope containing @a hdr
 666  */
 667 struct GNUNET_MQ_Envelope *
 668 GNUNET_MQ_msg_copy (const struct GNUNET_MessageHeader *hdr)
 669 {
 670   struct GNUNET_MQ_Envelope *mqm;
 671   uint16_t size = ntohs (hdr->size);
 672
 673   mqm = GNUNET_malloc (sizeof (*mqm) + size);
 674   mqm->mh = (struct GNUNET_MessageHeader *) &mqm[1];
 675   GNUNET_memcpy (mqm->mh,
 676           hdr,
 677           size);
 678   return mqm;
 679 }
 680
 681
 682 /**
 683  * Implementation of the #GNUNET_MQ_msg_nested_mh macro.
 684  *
 685  * @param mhp pointer to the message header pointer that will be changed to allocate at
 686  *        the newly allocated space for the message.
 687  * @param base_size size of the data before the nested message
 688  * @param type type of the message in the envelope
 689  * @param nested_mh the message to append to the message after base_size
 690  */
 691 struct GNUNET_MQ_Envelope *
 692 GNUNET_MQ_msg_nested_mh_ (struct GNUNET_MessageHeader **mhp,
 693                           uint16_t base_size,
 694                           uint16_t type,
 695                           const struct GNUNET_MessageHeader *nested_mh)
 696 {
 697   struct GNUNET_MQ_Envelope *mqm;
 698   uint16_t size;
 699
 700   if (NULL == nested_mh)
 701     return GNUNET_MQ_msg_ (mhp, base_size, type);
 702
 703   size = base_size + ntohs (nested_mh->size);
 704
 705   /* check for uint16_t overflow */
 706   if (size < base_size)
 707     return NULL;
 708
 709   mqm = GNUNET_MQ_msg_ (mhp, size, type);
 710   GNUNET_memcpy ((char *) mqm->mh + base_size,
 711                  nested_mh,
 712                  ntohs (nested_mh->size));
 713
 714   return mqm;
 715 }
 716
 717
 718 /**
 719  * Transmit a queued message to the session's client.
 720  *
 721  * @param cls consensus session
 722  * @param size number of bytes available in @a buf
 723  * @param buf where the callee should write the message
 724  * @return number of bytes written to @a buf
 725  */
 726 static size_t
 727 transmit_queued (void *cls,
 728                  size_t size,
 729                  void *buf)
 730 {
 731   struct GNUNET_MQ_Handle *mq = cls;
 732   struct ServerClientSocketState *state = GNUNET_MQ_impl_state (mq);
 733   const struct GNUNET_MessageHeader *msg = GNUNET_MQ_impl_current (mq);
 734   size_t msg_size;
 735
 736   GNUNET_assert (NULL != buf);
 737   msg_size = ntohs (msg->size);
 738   GNUNET_assert (size >= msg_size);
 739   GNUNET_memcpy (buf, msg, msg_size);
 740   state->th = NULL;
 741
 742   GNUNET_MQ_impl_send_continue (mq);
 743
 744   return msg_size;
 745 }
 746
 747
 748 static void
 749 server_client_destroy_impl (struct GNUNET_MQ_Handle *mq,
 750                             void *impl_state)
 751 {
 752   struct ServerClientSocketState *state = impl_state;
 753
 754   if (NULL != state->th)
 755   {
 756     GNUNET_SERVER_notify_transmit_ready_cancel (state->th);
 757     state->th = NULL;
 758   }
 759
 760   GNUNET_assert (NULL != mq);
 761   GNUNET_assert (NULL != state);
 762   GNUNET_SERVER_client_drop (state->client);
 763   GNUNET_free (state);
 764 }
 765
 766
 767 static void
 768 server_client_send_impl (struct GNUNET_MQ_Handle *mq,
 769                          const struct GNUNET_MessageHeader *msg,
 770                          void *impl_state)
 771 {
 772   struct ServerClientSocketState *state = impl_state;
 773
 774   GNUNET_assert (NULL != mq);
 775   state->th = GNUNET_SERVER_notify_transmit_ready (state->client,
 776                                                    ntohs (msg->size),
 777                                                    GNUNET_TIME_UNIT_FOREVER_REL,
 778                                                    &transmit_queued, mq);
 779 }
 780
 781
 782 struct GNUNET_MQ_Handle *
 783 GNUNET_MQ_queue_for_server_client (struct GNUNET_SERVER_Client *client)
 784 {
 785   struct GNUNET_MQ_Handle *mq;
 786   struct ServerClientSocketState *scss;
 787
 788   mq = GNUNET_new (struct GNUNET_MQ_Handle);
 789   scss = GNUNET_new (struct ServerClientSocketState);
 790   mq->impl_state = scss;
 791   scss->client = client;
 792   GNUNET_SERVER_client_keep (client);
 793   mq->send_impl = &server_client_send_impl;
 794   mq->destroy_impl = &server_client_destroy_impl;
 795   return mq;
 796 }
 797
 798
 799 /**
 800  * Type of a function to call when we receive a message
 801  * from the service.
 802  *
 803  * @param cls closure
 804  * @param msg message received, NULL on timeout or fatal error
 805  */
 806 static void
 807 handle_client_message (void *cls,
 808                        const struct GNUNET_MessageHeader *msg)
 809 {
 810   struct GNUNET_MQ_Handle *mq = cls;
 811   struct ClientConnectionState *state;
 812
 813   state = mq->impl_state;
 814   if (NULL == msg)
 815   {
 816     GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_READ);
 817     return;
 818   }
 819   GNUNET_CLIENT_receive (state->connection,
 820                          &handle_client_message,
 821                          mq,
 822                          GNUNET_TIME_UNIT_FOREVER_REL);
 823   GNUNET_MQ_inject_message (mq, msg);
 824 }
 825
 826
 827 /**
 828  * Transmit a queued message to the session's client.
 829  *
 830  * @param cls consensus session
 831  * @param size number of bytes available in @a buf
 832  * @param buf where the callee should write the message
 833  * @return number of bytes written to buf
 834  */
 835 static size_t
 836 connection_client_transmit_queued (void *cls,
 837                                    size_t size,
 838                                    void *buf)
 839 {
 840   struct GNUNET_MQ_Handle *mq = cls;
 841   const struct GNUNET_MessageHeader *msg;
 842   struct ClientConnectionState *state = mq->impl_state;
 843   size_t msg_size;
 844
 845   GNUNET_assert (NULL != mq);
 846   state->th = NULL;
 847   msg = GNUNET_MQ_impl_current (mq);
 848
 849   if (NULL == buf)
 850   {
 851     GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_READ);
 852     return 0;
 853   }
 854
 855   if ( (GNUNET_YES == state->receive_requested) &&
 856        (GNUNET_NO == state->receive_active) )
 857   {
 858     state->receive_active = GNUNET_YES;
 859     GNUNET_CLIENT_receive (state->connection,
 860                            &handle_client_message,
 861                            mq,
 862                            GNUNET_TIME_UNIT_FOREVER_REL);
 863   }
 864
 865   msg_size = ntohs (msg->size);
 866   GNUNET_assert (size >= msg_size);
 867   GNUNET_memcpy (buf, msg, msg_size);
 868   state->th = NULL;
 869
 870   GNUNET_MQ_impl_send_continue (mq);
 871
 872   return msg_size;
 873 }
 874
 875
 876 static void
 877 connection_client_destroy_impl (struct GNUNET_MQ_Handle *mq,
 878                                 void *impl_state)
 879 {
 880   struct ClientConnectionState *state = impl_state;
 881
 882   if (NULL != state->th)
 883   {
 884     GNUNET_CLIENT_notify_transmit_ready_cancel (state->th);
 885     state->th = NULL;
 886   }
 887   GNUNET_CLIENT_disconnect (state->connection);
 888   GNUNET_free (impl_state);
 889 }
 890
 891
 892 static void
 893 connection_client_send_impl (struct GNUNET_MQ_Handle *mq,
 894                              const struct GNUNET_MessageHeader *msg,
 895                              void *impl_state)
 896 {
 897   struct ClientConnectionState *state = impl_state;
 898
 899   GNUNET_assert (NULL != state);
 900   GNUNET_assert (NULL == state->th);
 901   state->th =
 902       GNUNET_CLIENT_notify_transmit_ready (state->connection,
 903                                            ntohs (msg->size),
 904                                            GNUNET_TIME_UNIT_FOREVER_REL,
 905                                            GNUNET_NO,
 906                                            &connection_client_transmit_queued,
 907                                            mq);
 908   GNUNET_assert (NULL != state->th);
 909 }
 910
 911
 912 static void
 913 connection_client_cancel_impl (struct GNUNET_MQ_Handle *mq,
 914                                void *impl_state)
 915 {
 916   struct ClientConnectionState *state = impl_state;
 917
 918   if (NULL != state->th)
 919   {
 920     GNUNET_CLIENT_notify_transmit_ready_cancel (state->th);
 921     state->th = NULL;
 922   }
 923   else if (NULL != mq->send_task)
 924   {
 925     GNUNET_SCHEDULER_cancel (mq->send_task);
 926     mq->send_task = NULL;
 927   }
 928   else
 929     GNUNET_assert (0);
 930 }
 931
 932
 933 struct GNUNET_MQ_Handle *
 934 GNUNET_MQ_queue_for_connection_client (struct GNUNET_CLIENT_Connection *connection,
 935                                        const struct GNUNET_MQ_MessageHandler *handlers,
 936                                        GNUNET_MQ_ErrorHandler error_handler,
 937                                        void *error_handler_cls)
 938 {
 939   struct GNUNET_MQ_Handle *mq;
 940   struct ClientConnectionState *state;
 941   unsigned int i;
 942
 943   mq = GNUNET_new (struct GNUNET_MQ_Handle);
 944   if (NULL != handlers)
 945   {
 946     for (i=0;NULL != handlers[i].cb; i++) ;
 947     mq->handlers = GNUNET_new_array (i + 1,
 948                                      struct GNUNET_MQ_MessageHandler);
 949     GNUNET_memcpy (mq->handlers,
 950                    handlers,
 951                    i * sizeof (struct GNUNET_MQ_MessageHandler));
 952   }
 953   mq->error_handler = error_handler;
 954   mq->error_handler_cls = error_handler_cls;
 955   state = GNUNET_new (struct ClientConnectionState);
 956   state->connection = connection;
 957   mq->impl_state = state;
 958   mq->send_impl = &connection_client_send_impl;
 959   mq->destroy_impl = &connection_client_destroy_impl;
 960   mq->cancel_impl = &connection_client_cancel_impl;
 961   if (NULL != handlers)
 962     state->receive_requested = GNUNET_YES;
 963
 964   return mq;
 965 }
 966
 967
 968 /**
 969  * Associate the assoc_data in mq with a unique request id.
 970  *
 971  * @param mq message queue, id will be unique for the queue
 972  * @param assoc_data to associate
 973  */
 974 uint32_t
 975 GNUNET_MQ_assoc_add (struct GNUNET_MQ_Handle *mq,
 976                      void *assoc_data)
 977 {
 978   uint32_t id;
 979
 980   if (NULL == mq->assoc_map)
 981   {
 982     mq->assoc_map = GNUNET_CONTAINER_multihashmap32_create (8);
 983     mq->assoc_id = 1;
 984   }
 985   id = mq->assoc_id++;
 986   GNUNET_CONTAINER_multihashmap32_put (mq->assoc_map, id, assoc_data,
 987                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
 988   return id;
 989 }
 990
 991
 992 void *
 993 GNUNET_MQ_assoc_get (struct GNUNET_MQ_Handle *mq,
 994                      uint32_t request_id)
 995 {
 996   if (NULL == mq->assoc_map)
 997     return NULL;
 998   return GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map, request_id);
 999 }
1000
1001
1002 void *
1003 GNUNET_MQ_assoc_remove (struct GNUNET_MQ_Handle *mq,
1004                         uint32_t request_id)
1005 {
1006   void *val;
1007
1008   if (NULL == mq->assoc_map)
1009     return NULL;
1010   val = GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map,
1011                                              request_id);
1012   GNUNET_CONTAINER_multihashmap32_remove_all (mq->assoc_map,
1013                                               request_id);
1014   return val;
1015 }
1016
1017
1018 /**
1019  * Call a callback once the envelope has been sent, that is,
1020  * sending it can not be canceled anymore.
1021  * There can be only one notify sent callback per envelope.
1022  *
1023  * @param ev message to call the notify callback for
1024  * @param cb the notify callback
1025  * @param cb_cls closure for the callback
1026  */
1027 void
1028 GNUNET_MQ_notify_sent (struct GNUNET_MQ_Envelope *mqm,
1029                        GNUNET_MQ_NotifyCallback cb,
1030                        void *cb_cls)
1031 {
1032   mqm->sent_cb = cb;
1033   mqm->sent_cls = cb_cls;
1034 }
1035
1036
1037 /**
1038  * Handle we return for callbacks registered to be
1039  * notified when #GNUNET_MQ_destroy() is called on a queue.
1040  */
1041 struct GNUNET_MQ_DestroyNotificationHandle
1042 {
1043   /**
1044    * Kept in a DLL.
1045    */
1046   struct GNUNET_MQ_DestroyNotificationHandle *prev;
1047
1048   /**
1049    * Kept in a DLL.
1050    */
1051   struct GNUNET_MQ_DestroyNotificationHandle *next;
1052
1053   /**
1054    * Queue to notify about.
1055    */
1056   struct GNUNET_MQ_Handle *mq;
1057
1058   /**
1059    * Function to call.
1060    */
1061   GNUNET_SCHEDULER_TaskCallback cb;
1062
1063   /**
1064    * Closure for @e cb.
1065    */
1066   void *cb_cls;
1067 };
1068
1069
1070 /**
1071  * Destroy the message queue.
1072  *
1073  * @param mq message queue to destroy
1074  */
1075 void
1076 GNUNET_MQ_destroy (struct GNUNET_MQ_Handle *mq)
1077 {
1078   struct GNUNET_MQ_DestroyNotificationHandle *dnh;
1079
1080   if (NULL != mq->destroy_impl)
1081   {
1082     mq->destroy_impl (mq, mq->impl_state);
1083   }
1084   if (NULL != mq->send_task)
1085   {
1086     GNUNET_SCHEDULER_cancel (mq->send_task);
1087     mq->send_task = NULL;
1088   }
1089   while (NULL != mq->envelope_head)
1090   {
1091     struct GNUNET_MQ_Envelope *ev;
1092
1093     ev = mq->envelope_head;
1094     ev->parent_queue = NULL;
1095     GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
1096                                  mq->envelope_tail,
1097                                  ev);
1098     GNUNET_assert (0 < mq->queue_length);
1099     mq->queue_length--;
1100     GNUNET_MQ_discard (ev);
1101   }
1102   if (NULL != mq->current_envelope)
1103   {
1104     /* we can only discard envelopes that
1105      * are not queued! */
1106     mq->current_envelope->parent_queue = NULL;
1107     GNUNET_MQ_discard (mq->current_envelope);
1108     mq->current_envelope = NULL;
1109     GNUNET_assert (0 < mq->queue_length);
1110     mq->queue_length--;
1111   }
1112   GNUNET_assert (0 == mq->queue_length);
1113   while (NULL != (dnh = mq->dnh_head))
1114   {
1115     dnh->cb (dnh->cb_cls);
1116     GNUNET_MQ_destroy_notify_cancel (dnh);
1117   }
1118   if (NULL != mq->assoc_map)
1119   {
1120     GNUNET_CONTAINER_multihashmap32_destroy (mq->assoc_map);
1121     mq->assoc_map = NULL;
1122   }
1123   GNUNET_free_non_null (mq->handlers);
1124   GNUNET_free (mq);
1125 }
1126
1127
1128 const struct GNUNET_MessageHeader *
1129 GNUNET_MQ_extract_nested_mh_ (const struct GNUNET_MessageHeader *mh,
1130                               uint16_t base_size)
1131 {
1132   uint16_t whole_size;
1133   uint16_t nested_size;
1134   const struct GNUNET_MessageHeader *nested_msg;
1135
1136   whole_size = ntohs (mh->size);
1137   GNUNET_assert (whole_size >= base_size);
1138   nested_size = whole_size - base_size;
1139   if (0 == nested_size)
1140     return NULL;
1141   if (nested_size < sizeof (struct GNUNET_MessageHeader))
1142   {
1143     GNUNET_break_op (0);
1144     return NULL;
1145   }
1146   nested_msg = (const struct GNUNET_MessageHeader *) ((char *) mh + base_size);
1147   if (ntohs (nested_msg->size) != nested_size)
1148   {
1149     GNUNET_break_op (0);
1150     return NULL;
1151   }
1152   return nested_msg;
1153 }
1154
1155
1156 /**
1157  * Cancel sending the message. Message must have been sent with
1158  * #GNUNET_MQ_send before.  May not be called after the notify sent
1159  * callback has been called
1160  *
1161  * @param ev queued envelope to cancel
1162  */
1163 void
1164 GNUNET_MQ_send_cancel (struct GNUNET_MQ_Envelope *ev)
1165 {
1166   struct GNUNET_MQ_Handle *mq = ev->parent_queue;
1167
1168   GNUNET_assert (NULL != mq);
1169   GNUNET_assert (NULL != mq->cancel_impl);
1170
1171   mq->evacuate_called = GNUNET_NO;
1172
1173   if (mq->current_envelope == ev)
1174   {
1175     // complex case, we already started with transmitting
1176     // the message
1177     GNUNET_assert (0 < mq->queue_length);
1178     mq->queue_length--;
1179     mq->cancel_impl (mq,
1180                      mq->impl_state);
1181     // continue sending the next message, if any
1182     if (NULL == mq->envelope_head)
1183     {
1184       mq->current_envelope = NULL;
1185     }
1186     else
1187     {
1188       mq->current_envelope = mq->envelope_head;
1189       GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
1190                                    mq->envelope_tail,
1191                                    mq->current_envelope);
1192       mq->send_notification_called = GNUNET_NO;
1193       mq->send_impl (mq,
1194                      mq->current_envelope->mh,
1195                      mq->impl_state);
1196     }
1197   }
1198   else
1199   {
1200     // simple case, message is still waiting in the queue
1201     GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
1202                                  mq->envelope_tail,
1203                                  ev);
1204     GNUNET_assert (0 < mq->queue_length);
1205     mq->queue_length--;
1206   }
1207
1208   if (GNUNET_YES != mq->evacuate_called)
1209   {
1210     ev->parent_queue = NULL;
1211     ev->mh = NULL;
1212     /* also frees ev */
1213     GNUNET_free (ev);
1214   }
1215 }
1216
1217
1218 /**
1219  * Function to obtain the current envelope
1220  * from within #GNUNET_MQ_SendImpl implementations.
1221  *
1222  * @param mq message queue to interrogate
1223  * @return the current envelope
1224  */
1225 struct GNUNET_MQ_Envelope *
1226 GNUNET_MQ_get_current_envelope (struct GNUNET_MQ_Handle *mq)
1227 {
1228   return mq->current_envelope;
1229 }
1230
1231
1232 /**
1233  * Function to obtain the last envelope in the queue.
1234  *
1235  * @param mq message queue to interrogate
1236  * @return the last envelope in the queue
1237  */
1238 struct GNUNET_MQ_Envelope *
1239 GNUNET_MQ_get_last_envelope (struct GNUNET_MQ_Handle *mq)
1240 {
1241   if (NULL != mq->envelope_tail)
1242     return mq->envelope_tail;
1243
1244   return mq->current_envelope;
1245 }
1246
1247
1248 /**
1249  * Set application-specific options for this envelope.
1250  * Overrides the options set for the queue with
1251  * #GNUNET_MQ_set_options() for this message only.
1252  *
1253  * @param env message to set options for
1254  * @param flags flags to use (meaning is queue-specific)
1255  * @param extra additional buffer for further data (also queue-specific)
1256  */
1257 void
1258 GNUNET_MQ_env_set_options (struct GNUNET_MQ_Envelope *env,
1259                            uint64_t flags,
1260                            const void *extra)
1261 {
1262   env->flags = flags;
1263   env->extra = extra;
1264   env->have_custom_options = GNUNET_YES;
1265 }
1266
1267
1268 /**
1269  * Get application-specific options for this envelope.
1270  *
1271  * @param env message to set options for
1272  * @param[out] flags set to flags to use (meaning is queue-specific)
1273  * @return extra additional buffer for further data (also queue-specific)
1274  */
1275 const void *
1276 GNUNET_MQ_env_get_options (struct GNUNET_MQ_Envelope *env,
1277                            uint64_t *flags)
1278 {
1279   struct GNUNET_MQ_Handle *mq = env->parent_queue;
1280
1281   if (GNUNET_YES == env->have_custom_options)
1282   {
1283     *flags = env->flags;
1284     return env->extra;
1285   }
1286   if (NULL == mq)
1287   {
1288     *flags = 0;
1289     return NULL;
1290   }
1291   *flags = mq->default_flags;
1292   return mq->default_extra;
1293 }
1294
1295
1296 /**
1297  * Set application-specific options for this queue.
1298  *
1299  * @param mq message queue to set options for
1300  * @param flags flags to use (meaning is queue-specific)
1301  * @param extra additional buffer for further data (also queue-specific)
1302  */
1303 void
1304 GNUNET_MQ_set_options (struct GNUNET_MQ_Handle *mq,
1305                        uint64_t flags,
1306                        const void *extra)
1307 {
1308   mq->default_flags = flags;
1309   mq->default_extra = extra;
1310 }
1311
1312
1313 /**
1314  * Register function to be called whenever @a mq is being
1315  * destroyed.
1316  *
1317  * @param mq message queue to watch
1318  * @param cb function to call on @a mq destruction
1319  * @param cb_cls closure for @a cb
1320  * @return handle for #GNUNET_MQ_destroy_notify_cancel().
1321  */
1322 struct GNUNET_MQ_DestroyNotificationHandle *
1323 GNUNET_MQ_destroy_notify (struct GNUNET_MQ_Handle *mq,
1324                           GNUNET_SCHEDULER_TaskCallback cb,
1325                           void *cb_cls)
1326 {
1327   struct GNUNET_MQ_DestroyNotificationHandle *dnh;
1328
1329   dnh = GNUNET_new (struct GNUNET_MQ_DestroyNotificationHandle);
1330   dnh->mq = mq;
1331   dnh->cb = cb;
1332   dnh->cb_cls = cb_cls;
1333   GNUNET_CONTAINER_DLL_insert (mq->dnh_head,
1334                                mq->dnh_tail,
1335                                dnh);
1336   return dnh;
1337 }
1338
1339
1340 /**
1341  * Cancel registration from #GNUNET_MQ_destroy_notify().
1342  *
1343  * @param dnh handle for registration to cancel
1344  */
1345 void
1346 GNUNET_MQ_destroy_notify_cancel (struct GNUNET_MQ_DestroyNotificationHandle *dnh)
1347 {
1348   struct GNUNET_MQ_Handle *mq = dnh->mq;
1349
1350   GNUNET_CONTAINER_DLL_remove (mq->dnh_head,
1351                                mq->dnh_tail,
1352                                dnh);
1353   GNUNET_free (dnh);
1354 }
1355
1356
1357 /* end of mq.c */