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 next 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   GNUNET_assert (NULL != mq->current_envelope);
 616   GNUNET_assert (NULL == mq->current_envelope->mh);
 617   return mq->current_envelope->mh;
 618 }
 619
 620
 621 /**
 622  * Get the implementation state associated with the
 623  * message queue.
 624  *
 625  * While the GNUNET_MQ_Impl* callbacks receive the
 626  * implementation state, continuations that are scheduled
 627  * by the implementation function often only have one closure
 628  * argument, with this function it is possible to get at the
 629  * implementation state when only passing the GNUNET_MQ_Handle
 630  * as closure.
 631  *
 632  * @param mq message queue with the current message
 633  * @return message to send, never NULL
 634  */
 635 void *
 636 GNUNET_MQ_impl_state (struct GNUNET_MQ_Handle *mq)
 637 {
 638   return mq->impl_state;
 639 }
 640
 641
 642 struct GNUNET_MQ_Envelope *
 643 GNUNET_MQ_msg_ (struct GNUNET_MessageHeader **mhp,
 644                 uint16_t size,
 645                 uint16_t type)
 646 {
 647   struct GNUNET_MQ_Envelope *ev;
 648
 649   ev = GNUNET_malloc (size + sizeof (struct GNUNET_MQ_Envelope));
 650   ev->mh = (struct GNUNET_MessageHeader *) &ev[1];
 651   ev->mh->size = htons (size);
 652   ev->mh->type = htons (type);
 653   if (NULL != mhp)
 654     *mhp = ev->mh;
 655   return ev;
 656 }
 657
 658
 659 /**
 660  * Create a new envelope by copying an existing message.
 661  *
 662  * @param hdr header of the message to copy
 663  * @return envelope containing @a hdr
 664  */
 665 struct GNUNET_MQ_Envelope *
 666 GNUNET_MQ_msg_copy (const struct GNUNET_MessageHeader *hdr)
 667 {
 668   struct GNUNET_MQ_Envelope *mqm;
 669   uint16_t size = ntohs (hdr->size);
 670
 671   mqm = GNUNET_malloc (sizeof (*mqm) + size);
 672   mqm->mh = (struct GNUNET_MessageHeader *) &mqm[1];
 673   GNUNET_memcpy (mqm->mh,
 674           hdr,
 675           size);
 676   return mqm;
 677 }
 678
 679
 680 /**
 681  * Implementation of the #GNUNET_MQ_msg_nested_mh macro.
 682  *
 683  * @param mhp pointer to the message header pointer that will be changed to allocate at
 684  *        the newly allocated space for the message.
 685  * @param base_size size of the data before the nested message
 686  * @param type type of the message in the envelope
 687  * @param nested_mh the message to append to the message after base_size
 688  */
 689 struct GNUNET_MQ_Envelope *
 690 GNUNET_MQ_msg_nested_mh_ (struct GNUNET_MessageHeader **mhp,
 691                           uint16_t base_size,
 692                           uint16_t type,
 693                           const struct GNUNET_MessageHeader *nested_mh)
 694 {
 695   struct GNUNET_MQ_Envelope *mqm;
 696   uint16_t size;
 697
 698   if (NULL == nested_mh)
 699     return GNUNET_MQ_msg_ (mhp, base_size, type);
 700
 701   size = base_size + ntohs (nested_mh->size);
 702
 703   /* check for uint16_t overflow */
 704   if (size < base_size)
 705     return NULL;
 706
 707   mqm = GNUNET_MQ_msg_ (mhp, size, type);
 708   GNUNET_memcpy ((char *) mqm->mh + base_size,
 709                  nested_mh,
 710                  ntohs (nested_mh->size));
 711
 712   return mqm;
 713 }
 714
 715
 716 /**
 717  * Transmit a queued message to the session's client.
 718  *
 719  * @param cls consensus session
 720  * @param size number of bytes available in @a buf
 721  * @param buf where the callee should write the message
 722  * @return number of bytes written to @a buf
 723  */
 724 static size_t
 725 transmit_queued (void *cls,
 726                  size_t size,
 727                  void *buf)
 728 {
 729   struct GNUNET_MQ_Handle *mq = cls;
 730   struct ServerClientSocketState *state = GNUNET_MQ_impl_state (mq);
 731   const struct GNUNET_MessageHeader *msg = GNUNET_MQ_impl_current (mq);
 732   size_t msg_size;
 733
 734   GNUNET_assert (NULL != buf);
 735   msg_size = ntohs (msg->size);
 736   GNUNET_assert (size >= msg_size);
 737   GNUNET_memcpy (buf, msg, msg_size);
 738   state->th = NULL;
 739
 740   GNUNET_MQ_impl_send_continue (mq);
 741
 742   return msg_size;
 743 }
 744
 745
 746 static void
 747 server_client_destroy_impl (struct GNUNET_MQ_Handle *mq,
 748                             void *impl_state)
 749 {
 750   struct ServerClientSocketState *state = impl_state;
 751
 752   if (NULL != state->th)
 753   {
 754     GNUNET_SERVER_notify_transmit_ready_cancel (state->th);
 755     state->th = NULL;
 756   }
 757
 758   GNUNET_assert (NULL != mq);
 759   GNUNET_assert (NULL != state);
 760   GNUNET_SERVER_client_drop (state->client);
 761   GNUNET_free (state);
 762 }
 763
 764
 765 static void
 766 server_client_send_impl (struct GNUNET_MQ_Handle *mq,
 767                          const struct GNUNET_MessageHeader *msg,
 768                          void *impl_state)
 769 {
 770   struct ServerClientSocketState *state = impl_state;
 771
 772   GNUNET_assert (NULL != mq);
 773   state->th = GNUNET_SERVER_notify_transmit_ready (state->client,
 774                                                    ntohs (msg->size),
 775                                                    GNUNET_TIME_UNIT_FOREVER_REL,
 776                                                    &transmit_queued, mq);
 777 }
 778
 779
 780 struct GNUNET_MQ_Handle *
 781 GNUNET_MQ_queue_for_server_client (struct GNUNET_SERVER_Client *client)
 782 {
 783   struct GNUNET_MQ_Handle *mq;
 784   struct ServerClientSocketState *scss;
 785
 786   mq = GNUNET_new (struct GNUNET_MQ_Handle);
 787   scss = GNUNET_new (struct ServerClientSocketState);
 788   mq->impl_state = scss;
 789   scss->client = client;
 790   GNUNET_SERVER_client_keep (client);
 791   mq->send_impl = &server_client_send_impl;
 792   mq->destroy_impl = &server_client_destroy_impl;
 793   return mq;
 794 }
 795
 796
 797 /**
 798  * Type of a function to call when we receive a message
 799  * from the service.
 800  *
 801  * @param cls closure
 802  * @param msg message received, NULL on timeout or fatal error
 803  */
 804 static void
 805 handle_client_message (void *cls,
 806                        const struct GNUNET_MessageHeader *msg)
 807 {
 808   struct GNUNET_MQ_Handle *mq = cls;
 809   struct ClientConnectionState *state;
 810
 811   state = mq->impl_state;
 812   if (NULL == msg)
 813   {
 814     GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_READ);
 815     return;
 816   }
 817   GNUNET_CLIENT_receive (state->connection,
 818                          &handle_client_message,
 819                          mq,
 820                          GNUNET_TIME_UNIT_FOREVER_REL);
 821   GNUNET_MQ_inject_message (mq, msg);
 822 }
 823
 824
 825 /**
 826  * Transmit a queued message to the session's client.
 827  *
 828  * @param cls consensus session
 829  * @param size number of bytes available in @a buf
 830  * @param buf where the callee should write the message
 831  * @return number of bytes written to buf
 832  */
 833 static size_t
 834 connection_client_transmit_queued (void *cls,
 835                                    size_t size,
 836                                    void *buf)
 837 {
 838   struct GNUNET_MQ_Handle *mq = cls;
 839   const struct GNUNET_MessageHeader *msg;
 840   struct ClientConnectionState *state = mq->impl_state;
 841   size_t msg_size;
 842
 843   GNUNET_assert (NULL != mq);
 844   state->th = NULL;
 845   msg = GNUNET_MQ_impl_current (mq);
 846
 847   if (NULL == buf)
 848   {
 849     GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_READ);
 850     return 0;
 851   }
 852
 853   if ( (GNUNET_YES == state->receive_requested) &&
 854        (GNUNET_NO == state->receive_active) )
 855   {
 856     state->receive_active = GNUNET_YES;
 857     GNUNET_CLIENT_receive (state->connection,
 858                            &handle_client_message,
 859                            mq,
 860                            GNUNET_TIME_UNIT_FOREVER_REL);
 861   }
 862
 863   msg_size = ntohs (msg->size);
 864   GNUNET_assert (size >= msg_size);
 865   GNUNET_memcpy (buf, msg, msg_size);
 866   state->th = NULL;
 867
 868   GNUNET_MQ_impl_send_continue (mq);
 869
 870   return msg_size;
 871 }
 872
 873
 874 static void
 875 connection_client_destroy_impl (struct GNUNET_MQ_Handle *mq,
 876                                 void *impl_state)
 877 {
 878   struct ClientConnectionState *state = impl_state;
 879
 880   if (NULL != state->th)
 881   {
 882     GNUNET_CLIENT_notify_transmit_ready_cancel (state->th);
 883     state->th = NULL;
 884   }
 885   GNUNET_CLIENT_disconnect (state->connection);
 886   GNUNET_free (impl_state);
 887 }
 888
 889
 890 static void
 891 connection_client_send_impl (struct GNUNET_MQ_Handle *mq,
 892                              const struct GNUNET_MessageHeader *msg,
 893                              void *impl_state)
 894 {
 895   struct ClientConnectionState *state = impl_state;
 896
 897   GNUNET_assert (NULL != state);
 898   GNUNET_assert (NULL == state->th);
 899   state->th =
 900       GNUNET_CLIENT_notify_transmit_ready (state->connection,
 901                                            ntohs (msg->size),
 902                                            GNUNET_TIME_UNIT_FOREVER_REL,
 903                                            GNUNET_NO,
 904                                            &connection_client_transmit_queued,
 905                                            mq);
 906   GNUNET_assert (NULL != state->th);
 907 }
 908
 909
 910 static void
 911 connection_client_cancel_impl (struct GNUNET_MQ_Handle *mq,
 912                                void *impl_state)
 913 {
 914   struct ClientConnectionState *state = impl_state;
 915
 916   if (NULL != state->th)
 917   {
 918     GNUNET_CLIENT_notify_transmit_ready_cancel (state->th);
 919     state->th = NULL;
 920   }
 921   else if (NULL != mq->send_task)
 922   {
 923     GNUNET_SCHEDULER_cancel (mq->send_task);
 924     mq->send_task = NULL;
 925   }
 926   else
 927     GNUNET_assert (0);
 928 }
 929
 930
 931 struct GNUNET_MQ_Handle *
 932 GNUNET_MQ_queue_for_connection_client (struct GNUNET_CLIENT_Connection *connection,
 933                                        const struct GNUNET_MQ_MessageHandler *handlers,
 934                                        GNUNET_MQ_ErrorHandler error_handler,
 935                                        void *error_handler_cls)
 936 {
 937   struct GNUNET_MQ_Handle *mq;
 938   struct ClientConnectionState *state;
 939   unsigned int i;
 940
 941   mq = GNUNET_new (struct GNUNET_MQ_Handle);
 942   if (NULL != handlers)
 943   {
 944     for (i=0;NULL != handlers[i].cb; i++) ;
 945     mq->handlers = GNUNET_new_array (i + 1,
 946                                      struct GNUNET_MQ_MessageHandler);
 947     GNUNET_memcpy (mq->handlers,
 948                    handlers,
 949                    i * sizeof (struct GNUNET_MQ_MessageHandler));
 950   }
 951   mq->error_handler = error_handler;
 952   mq->error_handler_cls = error_handler_cls;
 953   state = GNUNET_new (struct ClientConnectionState);
 954   state->connection = connection;
 955   mq->impl_state = state;
 956   mq->send_impl = &connection_client_send_impl;
 957   mq->destroy_impl = &connection_client_destroy_impl;
 958   mq->cancel_impl = &connection_client_cancel_impl;
 959   if (NULL != handlers)
 960     state->receive_requested = GNUNET_YES;
 961
 962   return mq;
 963 }
 964
 965
 966 /**
 967  * Associate the assoc_data in mq with a unique request id.
 968  *
 969  * @param mq message queue, id will be unique for the queue
 970  * @param assoc_data to associate
 971  */
 972 uint32_t
 973 GNUNET_MQ_assoc_add (struct GNUNET_MQ_Handle *mq,
 974                      void *assoc_data)
 975 {
 976   uint32_t id;
 977
 978   if (NULL == mq->assoc_map)
 979   {
 980     mq->assoc_map = GNUNET_CONTAINER_multihashmap32_create (8);
 981     mq->assoc_id = 1;
 982   }
 983   id = mq->assoc_id++;
 984   GNUNET_CONTAINER_multihashmap32_put (mq->assoc_map, id, assoc_data,
 985                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
 986   return id;
 987 }
 988
 989
 990 void *
 991 GNUNET_MQ_assoc_get (struct GNUNET_MQ_Handle *mq,
 992                      uint32_t request_id)
 993 {
 994   if (NULL == mq->assoc_map)
 995     return NULL;
 996   return GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map, request_id);
 997 }
 998
 999
1000 void *
1001 GNUNET_MQ_assoc_remove (struct GNUNET_MQ_Handle *mq,
1002                         uint32_t request_id)
1003 {
1004   void *val;
1005
1006   if (NULL == mq->assoc_map)
1007     return NULL;
1008   val = GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map,
1009                                              request_id);
1010   GNUNET_CONTAINER_multihashmap32_remove_all (mq->assoc_map,
1011                                               request_id);
1012   return val;
1013 }
1014
1015
1016 /**
1017  * Call a callback once the envelope has been sent, that is,
1018  * sending it can not be canceled anymore.
1019  * There can be only one notify sent callback per envelope.
1020  *
1021  * @param ev message to call the notify callback for
1022  * @param cb the notify callback
1023  * @param cb_cls closure for the callback
1024  */
1025 void
1026 GNUNET_MQ_notify_sent (struct GNUNET_MQ_Envelope *mqm,
1027                        GNUNET_MQ_NotifyCallback cb,
1028                        void *cb_cls)
1029 {
1030   mqm->sent_cb = cb;
1031   mqm->sent_cls = cb_cls;
1032 }
1033
1034
1035 /**
1036  * Handle we return for callbacks registered to be
1037  * notified when #GNUNET_MQ_destroy() is called on a queue.
1038  */
1039 struct GNUNET_MQ_DestroyNotificationHandle
1040 {
1041   /**
1042    * Kept in a DLL.
1043    */
1044   struct GNUNET_MQ_DestroyNotificationHandle *prev;
1045
1046   /**
1047    * Kept in a DLL.
1048    */
1049   struct GNUNET_MQ_DestroyNotificationHandle *next;
1050
1051   /**
1052    * Queue to notify about.
1053    */
1054   struct GNUNET_MQ_Handle *mq;
1055
1056   /**
1057    * Function to call.
1058    */
1059   GNUNET_SCHEDULER_TaskCallback cb;
1060
1061   /**
1062    * Closure for @e cb.
1063    */
1064   void *cb_cls;
1065 };
1066
1067
1068 /**
1069  * Destroy the message queue.
1070  *
1071  * @param mq message queue to destroy
1072  */
1073 void
1074 GNUNET_MQ_destroy (struct GNUNET_MQ_Handle *mq)
1075 {
1076   struct GNUNET_MQ_DestroyNotificationHandle *dnh;
1077
1078   if (NULL != mq->destroy_impl)
1079   {
1080     mq->destroy_impl (mq, mq->impl_state);
1081   }
1082   if (NULL != mq->send_task)
1083   {
1084     GNUNET_SCHEDULER_cancel (mq->send_task);
1085     mq->send_task = NULL;
1086   }
1087   while (NULL != mq->envelope_head)
1088   {
1089     struct GNUNET_MQ_Envelope *ev;
1090
1091     ev = mq->envelope_head;
1092     ev->parent_queue = NULL;
1093     GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
1094                                  mq->envelope_tail,
1095                                  ev);
1096     GNUNET_assert (0 < mq->queue_length);
1097     mq->queue_length--;
1098     GNUNET_MQ_discard (ev);
1099   }
1100   if (NULL != mq->current_envelope)
1101   {
1102     /* we can only discard envelopes that
1103      * are not queued! */
1104     mq->current_envelope->parent_queue = NULL;
1105     GNUNET_MQ_discard (mq->current_envelope);
1106     mq->current_envelope = NULL;
1107     GNUNET_assert (0 < mq->queue_length);
1108     mq->queue_length--;
1109   }
1110   GNUNET_assert (0 == mq->queue_length);
1111   while (NULL != (dnh = mq->dnh_head))
1112   {
1113     dnh->cb (dnh->cb_cls);
1114     GNUNET_MQ_destroy_notify_cancel (dnh);
1115   }
1116   if (NULL != mq->assoc_map)
1117   {
1118     GNUNET_CONTAINER_multihashmap32_destroy (mq->assoc_map);
1119     mq->assoc_map = NULL;
1120   }
1121   GNUNET_free_non_null (mq->handlers);
1122   GNUNET_free (mq);
1123 }
1124
1125
1126 const struct GNUNET_MessageHeader *
1127 GNUNET_MQ_extract_nested_mh_ (const struct GNUNET_MessageHeader *mh,
1128                               uint16_t base_size)
1129 {
1130   uint16_t whole_size;
1131   uint16_t nested_size;
1132   const struct GNUNET_MessageHeader *nested_msg;
1133
1134   whole_size = ntohs (mh->size);
1135   GNUNET_assert (whole_size >= base_size);
1136   nested_size = whole_size - base_size;
1137   if (0 == nested_size)
1138     return NULL;
1139   if (nested_size < sizeof (struct GNUNET_MessageHeader))
1140   {
1141     GNUNET_break_op (0);
1142     return NULL;
1143   }
1144   nested_msg = (const struct GNUNET_MessageHeader *) ((char *) mh + base_size);
1145   if (ntohs (nested_msg->size) != nested_size)
1146   {
1147     GNUNET_break_op (0);
1148     return NULL;
1149   }
1150   return nested_msg;
1151 }
1152
1153
1154 /**
1155  * Cancel sending the message. Message must have been sent with
1156  * #GNUNET_MQ_send before.  May not be called after the notify sent
1157  * callback has been called
1158  *
1159  * @param ev queued envelope to cancel
1160  */
1161 void
1162 GNUNET_MQ_send_cancel (struct GNUNET_MQ_Envelope *ev)
1163 {
1164   struct GNUNET_MQ_Handle *mq = ev->parent_queue;
1165
1166   GNUNET_assert (NULL != mq);
1167   GNUNET_assert (NULL != mq->cancel_impl);
1168
1169   mq->evacuate_called = GNUNET_NO;
1170
1171   if (mq->current_envelope == ev)
1172   {
1173     // complex case, we already started with transmitting
1174     // the message
1175     GNUNET_assert (0 < mq->queue_length);
1176     mq->queue_length--;
1177     mq->cancel_impl (mq,
1178                      mq->impl_state);
1179     // continue sending the next message, if any
1180     if (NULL == mq->envelope_head)
1181     {
1182       mq->current_envelope = NULL;
1183     }
1184     else
1185     {
1186       mq->current_envelope = mq->envelope_head;
1187       GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
1188                                    mq->envelope_tail,
1189                                    mq->current_envelope);
1190       mq->send_notification_called = GNUNET_NO;
1191       mq->send_impl (mq,
1192                      mq->current_envelope->mh,
1193                      mq->impl_state);
1194     }
1195   }
1196   else
1197   {
1198     // simple case, message is still waiting in the queue
1199     GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
1200                                  mq->envelope_tail,
1201                                  ev);
1202     GNUNET_assert (0 < mq->queue_length);
1203     mq->queue_length--;
1204   }
1205
1206   if (GNUNET_YES != mq->evacuate_called)
1207   {
1208     ev->parent_queue = NULL;
1209     ev->mh = NULL;
1210     /* also frees ev */
1211     GNUNET_free (ev);
1212   }
1213 }
1214
1215
1216 /**
1217  * Function to obtain the current envelope
1218  * from within #GNUNET_MQ_SendImpl implementations.
1219  *
1220  * @param mq message queue to interrogate
1221  * @return the current envelope
1222  */
1223 struct GNUNET_MQ_Envelope *
1224 GNUNET_MQ_get_current_envelope (struct GNUNET_MQ_Handle *mq)
1225 {
1226   return mq->current_envelope;
1227 }
1228
1229
1230 /**
1231  * Function to obtain the last envelope in the queue.
1232  *
1233  * @param mq message queue to interrogate
1234  * @return the last envelope in the queue
1235  */
1236 struct GNUNET_MQ_Envelope *
1237 GNUNET_MQ_get_last_envelope (struct GNUNET_MQ_Handle *mq)
1238 {
1239   if (NULL != mq->envelope_tail)
1240     return mq->envelope_tail;
1241
1242   return mq->current_envelope;
1243 }
1244
1245
1246 /**
1247  * Set application-specific options for this envelope.
1248  * Overrides the options set for the queue with
1249  * #GNUNET_MQ_set_options() for this message only.
1250  *
1251  * @param env message to set options for
1252  * @param flags flags to use (meaning is queue-specific)
1253  * @param extra additional buffer for further data (also queue-specific)
1254  */
1255 void
1256 GNUNET_MQ_env_set_options (struct GNUNET_MQ_Envelope *env,
1257                            uint64_t flags,
1258                            const void *extra)
1259 {
1260   env->flags = flags;
1261   env->extra = extra;
1262   env->have_custom_options = GNUNET_YES;
1263 }
1264
1265
1266 /**
1267  * Get application-specific options for this envelope.
1268  *
1269  * @param env message to set options for
1270  * @param[out] flags set to flags to use (meaning is queue-specific)
1271  * @return extra additional buffer for further data (also queue-specific)
1272  */
1273 const void *
1274 GNUNET_MQ_env_get_options (struct GNUNET_MQ_Envelope *env,
1275                            uint64_t *flags)
1276 {
1277   struct GNUNET_MQ_Handle *mq = env->parent_queue;
1278
1279   if (GNUNET_YES == env->have_custom_options)
1280   {
1281     *flags = env->flags;
1282     return env->extra;
1283   }
1284   if (NULL == mq)
1285   {
1286     *flags = 0;
1287     return NULL;
1288   }
1289   *flags = mq->default_flags;
1290   return mq->default_extra;
1291 }
1292
1293
1294 /**
1295  * Set application-specific options for this queue.
1296  *
1297  * @param mq message queue to set options for
1298  * @param flags flags to use (meaning is queue-specific)
1299  * @param extra additional buffer for further data (also queue-specific)
1300  */
1301 void
1302 GNUNET_MQ_set_options (struct GNUNET_MQ_Handle *mq,
1303                        uint64_t flags,
1304                        const void *extra)
1305 {
1306   mq->default_flags = flags;
1307   mq->default_extra = extra;
1308 }
1309
1310
1311 /**
1312  * Register function to be called whenever @a mq is being
1313  * destroyed.
1314  *
1315  * @param mq message queue to watch
1316  * @param cb function to call on @a mq destruction
1317  * @param cb_cls closure for @a cb
1318  * @return handle for #GNUNET_MQ_destroy_notify_cancel().
1319  */
1320 struct GNUNET_MQ_DestroyNotificationHandle *
1321 GNUNET_MQ_destroy_notify (struct GNUNET_MQ_Handle *mq,
1322                           GNUNET_SCHEDULER_TaskCallback cb,
1323                           void *cb_cls)
1324 {
1325   struct GNUNET_MQ_DestroyNotificationHandle *dnh;
1326
1327   dnh = GNUNET_new (struct GNUNET_MQ_DestroyNotificationHandle);
1328   dnh->mq = mq;
1329   dnh->cb = cb;
1330   dnh->cb_cls = cb_cls;
1331   GNUNET_CONTAINER_DLL_insert (mq->dnh_head,
1332                                mq->dnh_tail,
1333                                dnh);
1334   return dnh;
1335 }
1336
1337
1338 /**
1339  * Cancel registration from #GNUNET_MQ_destroy_notify().
1340  *
1341  * @param dnh handle for registration to cancel
1342  */
1343 void
1344 GNUNET_MQ_destroy_notify_cancel (struct GNUNET_MQ_DestroyNotificationHandle *dnh)
1345 {
1346   struct GNUNET_MQ_Handle *mq = dnh->mq;
1347
1348   GNUNET_CONTAINER_DLL_remove (mq->dnh_head,
1349                                mq->dnh_tail,
1350                                dnh);
1351   GNUNET_free (dnh);
1352 }
1353
1354
1355 /* end of mq.c */