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