src/util/mq.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2012 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 2, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @author Florian Dold
  23  * @file util/mq.c
  24  * @brief general purpose request queue
  25  */
  26
  27 #include "platform.h"
  28 #include "gnunet_common.h"
  29 #include "gnunet_util_lib.h"
  30
  31 #define LOG(kind,...) GNUNET_log_from (kind, "mq",__VA_ARGS__)
  32
  33
  34 struct GNUNET_MQ_Envelope
  35 {
  36   /**
  37    * Messages are stored in a linked list.
  38    * Each queue has its own list of envelopes.
  39    */
  40   struct GNUNET_MQ_Envelope *next;
  41
  42   /**
  43    * Messages are stored in a linked list
  44    * Each queue has its own list of envelopes.
  45    */
  46   struct GNUNET_MQ_Envelope *prev;
  47
  48   /**
  49    * Actual allocated message header,
  50    * usually points to the end of the containing GNUNET_MQ_Envelope
  51    */
  52   struct GNUNET_MessageHeader *mh;
  53
  54   /**
  55    * Queue the message is queued in, NULL if message is not queued.
  56    */
  57   struct GNUNET_MQ_Handle *parent_queue;
  58
  59   /**
  60    * Called after the message was sent irrevocably.
  61    */
  62   GNUNET_MQ_NotifyCallback sent_cb;
  63
  64   /**
  65    * Closure for send_cb
  66    */
  67   void *sent_cls;
  68 };
  69
  70
  71 /**
  72  * Handle to a message queue.
  73  */
  74 struct GNUNET_MQ_Handle
  75 {
  76   /**
  77    * Handlers array, or NULL if the queue should not receive messages
  78    */
  79   const struct GNUNET_MQ_MessageHandler *handlers;
  80
  81   /**
  82    * Closure for the handler callbacks,
  83    * as well as for the error handler.
  84    */
  85   void *handlers_cls;
  86
  87   /**
  88    * Actual implementation of message sending,
  89    * called when a message is added
  90    */
  91   GNUNET_MQ_SendImpl send_impl;
  92
  93   /**
  94    * Implementation-dependent queue destruction function
  95    */
  96   GNUNET_MQ_DestroyImpl destroy_impl;
  97
  98   /**
  99    * Implementation-specific state
 100    */
 101   void *impl_state;
 102
 103   /**
 104    * Callback will be called when an error occurs.
 105    */
 106   GNUNET_MQ_ErrorHandler error_handler;
 107
 108   /**
 109    * Linked list of messages pending to be sent
 110    */
 111   struct GNUNET_MQ_Envelope *envelope_head;
 112
 113   /**
 114    * Linked list of messages pending to be sent
 115    */
 116   struct GNUNET_MQ_Envelope *envelope_tail;
 117
 118   /**
 119    * Message that is currently scheduled to be
 120    * sent. Not the head of the message queue, as the implementation
 121    * needs to know if sending has been already scheduled or not.
 122    */
 123   struct GNUNET_MQ_Envelope *current_envelope;
 124
 125   /**
 126    * Has the current envelope been commited?
 127    * Either GNUNET_YES or GNUNET_NO.
 128    */
 129   int commited;
 130
 131   /**
 132    * Map of associations, lazily allocated
 133    */
 134   struct GNUNET_CONTAINER_MultiHashMap32 *assoc_map;
 135
 136   /**
 137    * Next id that should be used for the assoc_map,
 138    * initialized lazily to a random value together with
 139    * assoc_map
 140    */
 141   uint32_t assoc_id;
 142 };
 143
 144
 145
 146
 147 struct ServerClientSocketState
 148 {
 149   struct GNUNET_SERVER_Client *client;
 150   struct GNUNET_SERVER_TransmitHandle* th;
 151 };
 152
 153
 154 struct ClientConnectionState
 155 {
 156   /**
 157    * Did we call receive alread alreadyy?
 158    */
 159   int receive_active;
 160
 161   /**
 162    * Do we also want to receive?
 163    */
 164   int receive_requested;
 165   struct GNUNET_CLIENT_Connection *connection;
 166   struct GNUNET_CLIENT_TransmitHandle *th;
 167 };
 168
 169
 170
 171
 172 /**
 173  * Call the right callback for a message.
 174  *
 175  * @param mq message queue with the handlers
 176  * @param mh message to dispatch
 177  */
 178 void
 179 GNUNET_MQ_inject_message (struct GNUNET_MQ_Handle *mq, const struct GNUNET_MessageHeader *mh)
 180 {
 181   const struct GNUNET_MQ_MessageHandler *handler;
 182   int handled = GNUNET_NO;
 183
 184   handler = mq->handlers;
 185   if (NULL == handler)
 186     return;
 187   for (; NULL != handler->cb; handler++)
 188   {
 189     if (handler->type == ntohs (mh->type))
 190     {
 191       handler->cb (mq->handlers_cls, mh);
 192       handled = GNUNET_YES;
 193     }
 194   }
 195
 196   if (GNUNET_NO == handled)
 197     LOG (GNUNET_ERROR_TYPE_WARNING, "no handler for message of type %d\n", ntohs (mh->type));
 198 }
 199
 200
 201 /**
 202  * Call the right callback for an error condition.
 203  *
 204  * @param mq message queue
 205  */
 206 void
 207 GNUNET_MQ_inject_error (struct GNUNET_MQ_Handle *mq,
 208                         enum GNUNET_MQ_Error error)
 209 {
 210   if (NULL == mq->error_handler)
 211   {
 212     /* FIXME: log what kind of error occured */
 213     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "mq: got error, but no handler installed\n");
 214     return;
 215   }
 216   mq->error_handler (mq->handlers_cls, error);
 217 }
 218
 219
 220 void
 221 GNUNET_MQ_discard (struct GNUNET_MQ_Envelope *mqm)
 222 {
 223   GNUNET_assert (NULL == mqm->parent_queue);
 224   GNUNET_free (mqm);
 225 }
 226
 227
 228 /**
 229  * Send a message with the give message queue.
 230  * May only be called once per message.
 231  *
 232  * @param mq message queue
 233  * @param ev the message to send.
 234  */
 235 void
 236 GNUNET_MQ_send (struct GNUNET_MQ_Handle *mq, struct GNUNET_MQ_Envelope *ev)
 237 {
 238   GNUNET_assert (NULL != mq);
 239   GNUNET_assert (NULL == ev->parent_queue);
 240
 241   /* is the implementation busy? queue it! */
 242   if (NULL != mq->current_envelope)
 243   {
 244     GNUNET_CONTAINER_DLL_insert_tail (mq->envelope_head, mq->envelope_tail, ev);
 245     return;
 246   }
 247   mq->current_envelope = ev;
 248   mq->send_impl (mq, ev->mh, mq->impl_state);
 249 }
 250
 251
 252 /**
 253  * Call the send implementation for the next queued message,
 254  * if any.
 255  * Only useful for implementing message queues,
 256  * results in undefined behavior if not used carefully.
 257  *
 258  * @param mq message queue to send the next message with
 259  */
 260 void
 261 GNUNET_MQ_impl_send_continue (struct GNUNET_MQ_Handle *mq)
 262 {
 263   /* call is only valid if we're actually currently sending
 264    * a message */
 265   GNUNET_assert (NULL != mq);
 266   GNUNET_assert (NULL != mq->current_envelope);
 267   GNUNET_assert (GNUNET_YES == mq->commited);
 268   mq->commited = GNUNET_NO;
 269   GNUNET_free (mq->current_envelope);
 270   if (NULL == mq->envelope_head)
 271   {
 272     mq->current_envelope = NULL;
 273     return;
 274   }
 275
 276
 277   GNUNET_assert (NULL != mq->envelope_tail);
 278   GNUNET_assert (NULL != mq->envelope_head);
 279   mq->current_envelope = mq->envelope_head;
 280   GNUNET_CONTAINER_DLL_remove (mq->envelope_head, mq->envelope_tail,
 281                                mq->current_envelope);
 282   mq->send_impl (mq, mq->current_envelope->mh, mq->impl_state);
 283 }
 284
 285
 286 /**
 287  * Create a message queue for the specified handlers.
 288  *
 289  * @param send function the implements sending messages
 290  * @param destroy function that implements destroying the queue
 291  * @param destroy function that implements canceling a message
 292  * @param state for the queue, passed to 'send' and 'destroy'
 293  * @param handlers array of message handlers
 294  * @param error_handler handler for read and write errors
 295  * @return a new message queue
 296  */
 297 struct GNUNET_MQ_Handle *
 298 GNUNET_MQ_queue_for_callbacks (GNUNET_MQ_SendImpl send,
 299                                GNUNET_MQ_DestroyImpl destroy,
 300                                GNUNET_MQ_CancelImpl cancel,
 301                                void *impl_state,
 302                                const struct GNUNET_MQ_MessageHandler *handlers,
 303                                GNUNET_MQ_ErrorHandler error_handler,
 304                                void *cls)
 305 {
 306   struct GNUNET_MQ_Handle *mq;
 307
 308   mq = GNUNET_new (struct GNUNET_MQ_Handle);
 309   mq->send_impl = send;
 310   mq->destroy_impl = destroy;
 311   mq->handlers = handlers;
 312   mq->handlers_cls = cls;
 313   mq->impl_state = impl_state;
 314
 315   return mq;
 316 }
 317
 318
 319 /**
 320  * Get the message that should currently be sent.
 321  * Fails if there is no current message.
 322  * Only useful for implementing message queues,
 323  * results in undefined behavior if not used carefully.
 324  *
 325  * @param mq message queue with the current message
 326  * @return message to send, never NULL
 327  */
 328 const struct GNUNET_MessageHeader *
 329 GNUNET_MQ_impl_current (struct GNUNET_MQ_Handle *mq)
 330 {
 331   if (NULL == mq->current_envelope)
 332     GNUNET_abort ();
 333   if (NULL == mq->current_envelope->mh)
 334     GNUNET_abort ();
 335   return mq->current_envelope->mh;
 336 }
 337
 338
 339 /**
 340  * Get the implementation state associated with the
 341  * message queue.
 342  *
 343  * While the GNUNET_MQ_Impl* callbacks receive the
 344  * implementation state, continuations that are scheduled
 345  * by the implementation function often only have one closure
 346  * argument, with this function it is possible to get at the
 347  * implementation state when only passing the GNUNET_MQ_Handle
 348  * as closure.
 349  *
 350  * @param mq message queue with the current message
 351  * @return message to send, never NULL
 352  */
 353 void *
 354 GNUNET_MQ_impl_state (struct GNUNET_MQ_Handle *mq)
 355 {
 356   return mq->impl_state;
 357 }
 358
 359
 360
 361 /**
 362  * Mark the current message as irrevocably sent, but do not
 363  * proceed with sending the next message.
 364  * Will call the appropriate GNUNET_MQ_NotifyCallback, if any.
 365  *
 366  * @param mq message queue
 367  */
 368 void
 369 GNUNET_MQ_impl_send_commit (struct GNUNET_MQ_Handle *mq)
 370 {
 371   GNUNET_assert (NULL != mq->current_envelope);
 372   GNUNET_assert (GNUNET_NO == mq->commited);
 373   mq->commited = GNUNET_YES;
 374   if (NULL != mq->current_envelope->sent_cb)
 375     mq->current_envelope->sent_cb (mq->current_envelope->sent_cls);
 376 }
 377
 378
 379 struct GNUNET_MQ_Envelope *
 380 GNUNET_MQ_msg_ (struct GNUNET_MessageHeader **mhp, uint16_t size, uint16_t type)
 381 {
 382   struct GNUNET_MQ_Envelope *mqm;
 383
 384   mqm = GNUNET_malloc (sizeof *mqm + size);
 385   mqm->mh = (struct GNUNET_MessageHeader *) &mqm[1];
 386   mqm->mh->size = htons (size);
 387   mqm->mh->type = htons (type);
 388   if (NULL != mhp)
 389     *mhp = mqm->mh;
 390   return mqm;
 391 }
 392
 393
 394 struct GNUNET_MQ_Envelope *
 395 GNUNET_MQ_msg_nested_mh_ (struct GNUNET_MessageHeader **mhp, uint16_t base_size, uint16_t type,
 396                           const struct GNUNET_MessageHeader *nested_mh)
 397 {
 398   struct GNUNET_MQ_Envelope *mqm;
 399   uint16_t size;
 400
 401   if (NULL == nested_mh)
 402     return GNUNET_MQ_msg_ (mhp, base_size, type);
 403
 404   size = base_size + ntohs (nested_mh->size);
 405
 406   /* check for uint16_t overflow */
 407   if (size < base_size)
 408     return NULL;
 409
 410   mqm = GNUNET_MQ_msg_ (mhp, size, type);
 411   memcpy ((char *) mqm->mh + base_size, nested_mh, ntohs (nested_mh->size));
 412
 413   return mqm;
 414 }
 415
 416
 417 /**
 418  * Transmit a queued message to the session's client.
 419  *
 420  * @param cls consensus session
 421  * @param size number of bytes available in buf
 422  * @param buf where the callee should write the message
 423  * @return number of bytes written to buf
 424  */
 425 static size_t
 426 transmit_queued (void *cls, size_t size,
 427                  void *buf)
 428 {
 429   struct GNUNET_MQ_Handle *mq = cls;
 430   struct ServerClientSocketState *state = GNUNET_MQ_impl_state (mq);
 431   const struct GNUNET_MessageHeader *msg = GNUNET_MQ_impl_current (mq);
 432   size_t msg_size;
 433
 434   GNUNET_assert (NULL != buf);
 435
 436   msg_size = ntohs (msg->size);
 437   GNUNET_assert (size >= msg_size);
 438   memcpy (buf, msg, msg_size);
 439   state->th = NULL;
 440
 441   GNUNET_MQ_impl_send_continue (mq);
 442
 443   return msg_size;
 444 }
 445
 446
 447
 448 static void
 449 server_client_destroy_impl (struct GNUNET_MQ_Handle *mq,
 450                             void *impl_state)
 451 {
 452   struct ServerClientSocketState *state = impl_state;
 453
 454   GNUNET_assert (NULL != mq);
 455   GNUNET_assert (NULL != state);
 456   GNUNET_SERVER_client_drop (state->client);
 457   GNUNET_free (state);
 458 }
 459
 460 static void
 461 server_client_send_impl (struct GNUNET_MQ_Handle *mq,
 462                          const struct GNUNET_MessageHeader *msg, void *impl_state)
 463 {
 464   struct ServerClientSocketState *state = impl_state;
 465
 466   GNUNET_assert (NULL != mq);
 467   GNUNET_assert (NULL != state);
 468
 469   GNUNET_MQ_impl_send_commit (mq);
 470
 471   state->th =
 472       GNUNET_SERVER_notify_transmit_ready (state->client, ntohs (msg->size),
 473                                            GNUNET_TIME_UNIT_FOREVER_REL,
 474                                            &transmit_queued, mq);
 475 }
 476
 477
 478 struct GNUNET_MQ_Handle *
 479 GNUNET_MQ_queue_for_server_client (struct GNUNET_SERVER_Client *client)
 480 {
 481   struct GNUNET_MQ_Handle *mq;
 482   struct ServerClientSocketState *scss;
 483
 484   mq = GNUNET_new (struct GNUNET_MQ_Handle);
 485   scss = GNUNET_new (struct ServerClientSocketState);
 486   mq->impl_state = scss;
 487   scss->client = client;
 488   GNUNET_SERVER_client_keep (client);
 489   mq->send_impl = server_client_send_impl;
 490   mq->destroy_impl = server_client_destroy_impl;
 491   return mq;
 492 }
 493
 494
 495 /**
 496  * Type of a function to call when we receive a message
 497  * from the service.
 498  *
 499  * @param cls closure
 500  * @param msg message received, NULL on timeout or fatal error
 501  */
 502 static void
 503 handle_client_message (void *cls,
 504                        const struct GNUNET_MessageHeader *msg)
 505 {
 506   struct GNUNET_MQ_Handle *mq = cls;
 507   struct ClientConnectionState *state;
 508
 509   state = mq->impl_state;
 510
 511   if (NULL == msg)
 512   {
 513     GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_READ);
 514     return;
 515   }
 516
 517   GNUNET_CLIENT_receive (state->connection, handle_client_message, mq,
 518                          GNUNET_TIME_UNIT_FOREVER_REL);
 519
 520   GNUNET_MQ_inject_message (mq, msg);
 521 }
 522
 523
 524 /**
 525  * Transmit a queued message to the session's client.
 526  *
 527  * @param cls consensus session
 528  * @param size number of bytes available in buf
 529  * @param buf where the callee should write the message
 530  * @return number of bytes written to buf
 531  */
 532 static size_t
 533 connection_client_transmit_queued (void *cls, size_t size,
 534                  void *buf)
 535 {
 536   struct GNUNET_MQ_Handle *mq = cls;
 537   const struct GNUNET_MessageHeader *msg;
 538   struct ClientConnectionState *state = mq->impl_state;
 539   size_t msg_size;
 540
 541   GNUNET_assert (NULL != mq);
 542   msg = GNUNET_MQ_impl_current (mq);
 543
 544   if (NULL == buf)
 545   {
 546     GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_READ);
 547     return 0;
 548   }
 549
 550   if ( (GNUNET_YES == state->receive_requested) &&
 551        (GNUNET_NO == state->receive_active) )
 552   {
 553     state->receive_active = GNUNET_YES;
 554     GNUNET_CLIENT_receive (state->connection, handle_client_message, mq,
 555                            GNUNET_TIME_UNIT_FOREVER_REL);
 556   }
 557
 558
 559   msg_size = ntohs (msg->size);
 560   GNUNET_assert (size >= msg_size);
 561   memcpy (buf, msg, msg_size);
 562   state->th = NULL;
 563
 564   GNUNET_MQ_impl_send_continue (mq);
 565
 566   return msg_size;
 567 }
 568
 569
 570
 571 static void
 572 connection_client_destroy_impl (struct GNUNET_MQ_Handle *mq, void *impl_state)
 573 {
 574   GNUNET_free (impl_state);
 575 }
 576
 577 static void
 578 connection_client_send_impl (struct GNUNET_MQ_Handle *mq,
 579                              const struct GNUNET_MessageHeader *msg, void *impl_state)
 580 {
 581   struct ClientConnectionState *state = impl_state;
 582
 583   GNUNET_assert (NULL != state);
 584   GNUNET_assert (NULL == state->th);
 585
 586   GNUNET_MQ_impl_send_commit (mq);
 587
 588   state->th =
 589       GNUNET_CLIENT_notify_transmit_ready (state->connection, ntohs (msg->size),
 590                                            GNUNET_TIME_UNIT_FOREVER_REL, GNUNET_NO,
 591                                            &connection_client_transmit_queued, mq);
 592 }
 593
 594
 595 struct GNUNET_MQ_Handle *
 596 GNUNET_MQ_queue_for_connection_client (struct GNUNET_CLIENT_Connection *connection,
 597                                        const struct GNUNET_MQ_MessageHandler *handlers,
 598                                        void *cls)
 599 {
 600   struct GNUNET_MQ_Handle *mq;
 601   struct ClientConnectionState *state;
 602
 603   GNUNET_assert (NULL != connection);
 604
 605   mq = GNUNET_new (struct GNUNET_MQ_Handle);
 606   mq->handlers = handlers;
 607   mq->handlers_cls = cls;
 608   state = GNUNET_new (struct ClientConnectionState);
 609   state->connection = connection;
 610   mq->impl_state = state;
 611   mq->send_impl = connection_client_send_impl;
 612   mq->destroy_impl = connection_client_destroy_impl;
 613   if (NULL != handlers)
 614     state->receive_requested = GNUNET_YES;
 615
 616   return mq;
 617 }
 618
 619
 620 void
 621 GNUNET_MQ_replace_handlers (struct GNUNET_MQ_Handle *mq,
 622                             const struct GNUNET_MQ_MessageHandler *new_handlers,
 623                             void *cls)
 624 {
 625   /* FIXME: notify implementation? */
 626   /* FIXME: what about NULL handlers? abort receive? */
 627   mq->handlers = new_handlers;
 628   mq->handlers_cls = cls;
 629 }
 630
 631
 632 /**
 633  * Associate the assoc_data in mq with a unique request id.
 634  *
 635  * @param mq message queue, id will be unique for the queue
 636  * @param mqm message to associate
 637  * @param assoc_data to associate
 638  */
 639 uint32_t
 640 GNUNET_MQ_assoc_add (struct GNUNET_MQ_Handle *mq,
 641                      void *assoc_data)
 642 {
 643   uint32_t id;
 644
 645   if (NULL == mq->assoc_map)
 646   {
 647     mq->assoc_map = GNUNET_CONTAINER_multihashmap32_create (8);
 648     mq->assoc_id = 1;
 649   }
 650   id = mq->assoc_id++;
 651   GNUNET_CONTAINER_multihashmap32_put (mq->assoc_map, id, assoc_data,
 652                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
 653   return id;
 654 }
 655
 656
 657
 658 void *
 659 GNUNET_MQ_assoc_get (struct GNUNET_MQ_Handle *mq, uint32_t request_id)
 660 {
 661   if (NULL == mq->assoc_map)
 662     return NULL;
 663   return GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map, request_id);
 664 }
 665
 666
 667 void *
 668 GNUNET_MQ_assoc_remove (struct GNUNET_MQ_Handle *mq, uint32_t request_id)
 669 {
 670   void *val;
 671
 672   if (NULL == mq->assoc_map)
 673     return NULL;
 674   val = GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map, request_id);
 675   GNUNET_assert (NULL != val);
 676   GNUNET_CONTAINER_multihashmap32_remove (mq->assoc_map, request_id, val);
 677   return val;
 678 }
 679
 680
 681 void
 682 GNUNET_MQ_notify_sent (struct GNUNET_MQ_Envelope *mqm,
 683                        GNUNET_MQ_NotifyCallback cb,
 684                        void *cls)
 685 {
 686   mqm->sent_cb = cb;
 687   mqm->sent_cls = cls;
 688 }
 689
 690
 691 void
 692 GNUNET_MQ_destroy (struct GNUNET_MQ_Handle *mq)
 693 {
 694   /* FIXME: destroy all pending messages in the queue */
 695
 696   if (NULL != mq->destroy_impl)
 697   {
 698     mq->destroy_impl (mq, mq->impl_state);
 699   }
 700
 701   GNUNET_free (mq);
 702 }
 703
 704
 705
 706 struct GNUNET_MessageHeader *
 707 GNUNET_MQ_extract_nested_mh_ (const struct GNUNET_MessageHeader *mh, uint16_t base_size)
 708 {
 709   uint16_t whole_size;
 710   uint16_t nested_size;
 711   struct GNUNET_MessageHeader *nested_msg;
 712
 713   whole_size = ntohs (mh->size);
 714   GNUNET_assert (whole_size >= base_size);
 715
 716   nested_size = whole_size - base_size;
 717
 718   if (0 == nested_size)
 719     return NULL;
 720
 721   if (nested_size < sizeof (struct GNUNET_MessageHeader))
 722   {
 723     GNUNET_break_op (0);
 724     return NULL;
 725   }
 726
 727   nested_msg = (struct GNUNET_MessageHeader *) ((char *) mh + base_size);
 728
 729   if (ntohs (nested_msg->size) != nested_size)
 730   {
 731     GNUNET_break_op (0);
 732     nested_msg->size = htons (nested_size);
 733   }
 734
 735   return nested_msg;
 736 }
 737
 738