2 This file is part of GNUnet.
3 Copyright (C) 2012-2014 GNUnet e.V.
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.
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.
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.
22 * @author Florian Dold
24 * @brief general purpose request queue
27 #include "gnunet_util_lib.h"
29 #define LOG(kind,...) GNUNET_log_from (kind, "mq",__VA_ARGS__)
32 struct GNUNET_MQ_Envelope
35 * Messages are stored in a linked list.
36 * Each queue has its own list of envelopes.
38 struct GNUNET_MQ_Envelope *next;
41 * Messages are stored in a linked list
42 * Each queue has its own list of envelopes.
44 struct GNUNET_MQ_Envelope *prev;
47 * Actual allocated message header,
48 * usually points to the end of the containing GNUNET_MQ_Envelope
50 struct GNUNET_MessageHeader *mh;
53 * Queue the message is queued in, NULL if message is not queued.
55 struct GNUNET_MQ_Handle *parent_queue;
58 * Called after the message was sent irrevocably.
60 GNUNET_MQ_NotifyCallback sent_cb;
63 * Closure for @e send_cb
70 * Handle to a message queue.
72 struct GNUNET_MQ_Handle
75 * Handlers array, or NULL if the queue should not receive messages
77 const struct GNUNET_MQ_MessageHandler *handlers;
80 * Closure for the handler callbacks,
81 * as well as for the error handler.
86 * Actual implementation of message sending,
87 * called when a message is added
89 GNUNET_MQ_SendImpl send_impl;
92 * Implementation-dependent queue destruction function
94 GNUNET_MQ_DestroyImpl destroy_impl;
97 * Implementation-dependent send cancel function
99 GNUNET_MQ_CancelImpl cancel_impl;
102 * Implementation-specific state
107 * Callback will be called when an error occurs.
109 GNUNET_MQ_ErrorHandler error_handler;
112 * Linked list of messages pending to be sent
114 struct GNUNET_MQ_Envelope *envelope_head;
117 * Linked list of messages pending to be sent
119 struct GNUNET_MQ_Envelope *envelope_tail;
122 * Message that is currently scheduled to be
123 * sent. Not the head of the message queue, as the implementation
124 * needs to know if sending has been already scheduled or not.
126 struct GNUNET_MQ_Envelope *current_envelope;
129 * Map of associations, lazily allocated
131 struct GNUNET_CONTAINER_MultiHashMap32 *assoc_map;
134 * Task scheduled during #GNUNET_MQ_impl_send_continue.
136 struct GNUNET_SCHEDULER_Task * continue_task;
139 * Next id that should be used for the @e assoc_map,
140 * initialized lazily to a random value together with
148 * Implementation-specific state for connection to
149 * client (MQ for server).
151 struct ServerClientSocketState
154 * Handle of the client that connected to the server.
156 struct GNUNET_SERVER_Client *client;
159 * Active transmission request to the client.
161 struct GNUNET_SERVER_TransmitHandle* th;
166 * Implementation-specific state for connection to
167 * service (MQ for clients).
169 struct ClientConnectionState
172 * Did we call receive alread alreadyy?
177 * Do we also want to receive?
179 int receive_requested;
182 * Connection to the service.
184 struct GNUNET_CLIENT_Connection *connection;
187 * Active transmission request (or NULL).
189 struct GNUNET_CLIENT_TransmitHandle *th;
194 * Call the message message handler that was registered
195 * for the type of the given message in the given message queue.
197 * This function is indended to be used for the implementation
200 * @param mq message queue with the handlers
201 * @param mh message to dispatch
204 GNUNET_MQ_inject_message (struct GNUNET_MQ_Handle *mq,
205 const struct GNUNET_MessageHeader *mh)
207 const struct GNUNET_MQ_MessageHandler *handler;
208 int handled = GNUNET_NO;
210 if (NULL == mq->handlers)
212 LOG (GNUNET_ERROR_TYPE_WARNING,
213 "No handler for message of type %d\n",
217 for (handler = mq->handlers; NULL != handler->cb; handler++)
219 if (handler->type == ntohs (mh->type))
221 handler->cb (mq->handlers_cls, mh);
222 handled = GNUNET_YES;
226 if (GNUNET_NO == handled)
227 LOG (GNUNET_ERROR_TYPE_WARNING,
228 "No handler for message of type %d\n",
234 * Call the error handler of a message queue with the given
235 * error code. If there is no error handler, log a warning.
237 * This function is intended to be used by the implementation
240 * @param mq message queue
241 * @param error the error type
244 GNUNET_MQ_inject_error (struct GNUNET_MQ_Handle *mq,
245 enum GNUNET_MQ_Error error)
247 if (NULL == mq->error_handler)
249 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
250 "mq: got error %d, but no handler installed\n",
254 mq->error_handler (mq->handlers_cls, error);
259 GNUNET_MQ_discard (struct GNUNET_MQ_Envelope *mqm)
261 GNUNET_assert (NULL == mqm->parent_queue);
267 * Send a message with the give message queue.
268 * May only be called once per message.
270 * @param mq message queue
271 * @param ev the envelope with the message to send.
274 GNUNET_MQ_send (struct GNUNET_MQ_Handle *mq,
275 struct GNUNET_MQ_Envelope *ev)
277 GNUNET_assert (NULL != mq);
278 GNUNET_assert (NULL == ev->parent_queue);
280 ev->parent_queue = mq;
281 /* is the implementation busy? queue it! */
282 if (NULL != mq->current_envelope)
284 GNUNET_CONTAINER_DLL_insert_tail (mq->envelope_head,
289 mq->current_envelope = ev;
290 mq->send_impl (mq, ev->mh, mq->impl_state);
295 * Task run to call the send implementation for the next queued
296 * message, if any. Only useful for implementing message queues,
297 * results in undefined behavior if not used carefully.
299 * @param cls message queue to send the next message with
302 impl_send_continue (void *cls)
304 struct GNUNET_MQ_Handle *mq = cls;
305 struct GNUNET_MQ_Envelope *current_envelope;
307 mq->continue_task = NULL;
308 /* call is only valid if we're actually currently sending
310 current_envelope = mq->current_envelope;
311 GNUNET_assert (NULL != current_envelope);
312 current_envelope->parent_queue = NULL;
313 if (NULL == mq->envelope_head)
315 mq->current_envelope = NULL;
319 mq->current_envelope = mq->envelope_head;
320 GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
322 mq->current_envelope);
324 mq->current_envelope->mh,
327 if (NULL != current_envelope->sent_cb)
328 current_envelope->sent_cb (current_envelope->sent_cls);
329 GNUNET_free (current_envelope);
334 * Call the send implementation for the next queued message, if any.
335 * Only useful for implementing message queues, results in undefined
336 * behavior if not used carefully.
338 * @param mq message queue to send the next message with
341 GNUNET_MQ_impl_send_continue (struct GNUNET_MQ_Handle *mq)
343 GNUNET_assert (NULL == mq->continue_task);
344 mq->continue_task = GNUNET_SCHEDULER_add_now (&impl_send_continue,
350 * Create a message queue for the specified handlers.
352 * @param send function the implements sending messages
353 * @param destroy function that implements destroying the queue
354 * @param cancel function that implements canceling a message
355 * @param impl_state for the queue, passed to 'send' and 'destroy'
356 * @param handlers array of message handlers
357 * @param error_handler handler for read and write errors
358 * @param cls closure for message handlers and error handler
359 * @return a new message queue
361 struct GNUNET_MQ_Handle *
362 GNUNET_MQ_queue_for_callbacks (GNUNET_MQ_SendImpl send,
363 GNUNET_MQ_DestroyImpl destroy,
364 GNUNET_MQ_CancelImpl cancel,
366 const struct GNUNET_MQ_MessageHandler *handlers,
367 GNUNET_MQ_ErrorHandler error_handler,
370 struct GNUNET_MQ_Handle *mq;
372 mq = GNUNET_new (struct GNUNET_MQ_Handle);
373 mq->send_impl = send;
374 mq->destroy_impl = destroy;
375 mq->cancel_impl = cancel;
376 mq->handlers = handlers;
377 mq->handlers_cls = cls;
378 mq->impl_state = impl_state;
385 * Get the message that should currently be sent.
386 * Fails if there is no current message.
387 * Only useful for implementing message queues,
388 * results in undefined behavior if not used carefully.
390 * @param mq message queue with the current message
391 * @return message to send, never NULL
393 const struct GNUNET_MessageHeader *
394 GNUNET_MQ_impl_current (struct GNUNET_MQ_Handle *mq)
396 if (NULL == mq->current_envelope)
398 if (NULL == mq->current_envelope->mh)
400 return mq->current_envelope->mh;
405 * Get the implementation state associated with the
408 * While the GNUNET_MQ_Impl* callbacks receive the
409 * implementation state, continuations that are scheduled
410 * by the implementation function often only have one closure
411 * argument, with this function it is possible to get at the
412 * implementation state when only passing the GNUNET_MQ_Handle
415 * @param mq message queue with the current message
416 * @return message to send, never NULL
419 GNUNET_MQ_impl_state (struct GNUNET_MQ_Handle *mq)
421 return mq->impl_state;
425 struct GNUNET_MQ_Envelope *
426 GNUNET_MQ_msg_ (struct GNUNET_MessageHeader **mhp,
430 struct GNUNET_MQ_Envelope *mqm;
432 mqm = GNUNET_malloc (sizeof *mqm + size);
433 mqm->mh = (struct GNUNET_MessageHeader *) &mqm[1];
434 mqm->mh->size = htons (size);
435 mqm->mh->type = htons (type);
443 * Implementation of the GNUNET_MQ_msg_nested_mh macro.
445 * @param mhp pointer to the message header pointer that will be changed to allocate at
446 * the newly allocated space for the message.
447 * @param base_size size of the data before the nested message
448 * @param type type of the message in the envelope
449 * @param nested_mh the message to append to the message after base_size
451 struct GNUNET_MQ_Envelope *
452 GNUNET_MQ_msg_nested_mh_ (struct GNUNET_MessageHeader **mhp,
455 const struct GNUNET_MessageHeader *nested_mh)
457 struct GNUNET_MQ_Envelope *mqm;
460 if (NULL == nested_mh)
461 return GNUNET_MQ_msg_ (mhp, base_size, type);
463 size = base_size + ntohs (nested_mh->size);
465 /* check for uint16_t overflow */
466 if (size < base_size)
469 mqm = GNUNET_MQ_msg_ (mhp, size, type);
470 memcpy ((char *) mqm->mh + base_size,
472 ntohs (nested_mh->size));
479 * Transmit a queued message to the session's client.
481 * @param cls consensus session
482 * @param size number of bytes available in @a buf
483 * @param buf where the callee should write the message
484 * @return number of bytes written to @a buf
487 transmit_queued (void *cls, size_t size,
490 struct GNUNET_MQ_Handle *mq = cls;
491 struct ServerClientSocketState *state = GNUNET_MQ_impl_state (mq);
492 const struct GNUNET_MessageHeader *msg = GNUNET_MQ_impl_current (mq);
495 GNUNET_assert (NULL != buf);
497 msg_size = ntohs (msg->size);
498 GNUNET_assert (size >= msg_size);
499 memcpy (buf, msg, msg_size);
502 GNUNET_MQ_impl_send_continue (mq);
509 server_client_destroy_impl (struct GNUNET_MQ_Handle *mq,
512 struct ServerClientSocketState *state = impl_state;
514 if (NULL != state->th)
516 GNUNET_SERVER_notify_transmit_ready_cancel (state->th);
520 GNUNET_assert (NULL != mq);
521 GNUNET_assert (NULL != state);
522 GNUNET_SERVER_client_drop (state->client);
528 server_client_send_impl (struct GNUNET_MQ_Handle *mq,
529 const struct GNUNET_MessageHeader *msg,
532 struct ServerClientSocketState *state = impl_state;
534 GNUNET_assert (NULL != mq);
535 GNUNET_assert (NULL != state);
536 state->th = GNUNET_SERVER_notify_transmit_ready (state->client,
538 GNUNET_TIME_UNIT_FOREVER_REL,
539 &transmit_queued, mq);
543 struct GNUNET_MQ_Handle *
544 GNUNET_MQ_queue_for_server_client (struct GNUNET_SERVER_Client *client)
546 struct GNUNET_MQ_Handle *mq;
547 struct ServerClientSocketState *scss;
549 mq = GNUNET_new (struct GNUNET_MQ_Handle);
550 scss = GNUNET_new (struct ServerClientSocketState);
551 mq->impl_state = scss;
552 scss->client = client;
553 GNUNET_SERVER_client_keep (client);
554 mq->send_impl = server_client_send_impl;
555 mq->destroy_impl = server_client_destroy_impl;
561 * Type of a function to call when we receive a message
565 * @param msg message received, NULL on timeout or fatal error
568 handle_client_message (void *cls,
569 const struct GNUNET_MessageHeader *msg)
571 struct GNUNET_MQ_Handle *mq = cls;
572 struct ClientConnectionState *state;
574 state = mq->impl_state;
578 GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_READ);
581 GNUNET_CLIENT_receive (state->connection,
582 &handle_client_message,
584 GNUNET_TIME_UNIT_FOREVER_REL);
585 GNUNET_MQ_inject_message (mq, msg);
590 * Transmit a queued message to the session's client.
592 * @param cls consensus session
593 * @param size number of bytes available in @a buf
594 * @param buf where the callee should write the message
595 * @return number of bytes written to buf
598 connection_client_transmit_queued (void *cls,
602 struct GNUNET_MQ_Handle *mq = cls;
603 const struct GNUNET_MessageHeader *msg;
604 struct ClientConnectionState *state = mq->impl_state;
607 GNUNET_assert (NULL != mq);
608 msg = GNUNET_MQ_impl_current (mq);
612 GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_READ);
616 if ( (GNUNET_YES == state->receive_requested) &&
617 (GNUNET_NO == state->receive_active) )
619 state->receive_active = GNUNET_YES;
620 GNUNET_CLIENT_receive (state->connection, handle_client_message, mq,
621 GNUNET_TIME_UNIT_FOREVER_REL);
624 msg_size = ntohs (msg->size);
625 GNUNET_assert (size >= msg_size);
626 memcpy (buf, msg, msg_size);
629 GNUNET_MQ_impl_send_continue (mq);
636 connection_client_destroy_impl (struct GNUNET_MQ_Handle *mq,
639 GNUNET_free (impl_state);
644 connection_client_send_impl (struct GNUNET_MQ_Handle *mq,
645 const struct GNUNET_MessageHeader *msg,
648 struct ClientConnectionState *state = impl_state;
650 GNUNET_assert (NULL != state);
651 GNUNET_assert (NULL == state->th);
653 GNUNET_CLIENT_notify_transmit_ready (state->connection,
655 GNUNET_TIME_UNIT_FOREVER_REL, GNUNET_NO,
656 &connection_client_transmit_queued, mq);
657 GNUNET_assert (NULL != state->th);
662 connection_client_cancel_impl (struct GNUNET_MQ_Handle *mq,
665 struct ClientConnectionState *state = impl_state;
666 GNUNET_assert (NULL != state->th);
667 GNUNET_CLIENT_notify_transmit_ready_cancel (state->th);
672 struct GNUNET_MQ_Handle *
673 GNUNET_MQ_queue_for_connection_client (struct GNUNET_CLIENT_Connection *connection,
674 const struct GNUNET_MQ_MessageHandler *handlers,
675 GNUNET_MQ_ErrorHandler error_handler,
678 struct GNUNET_MQ_Handle *mq;
679 struct ClientConnectionState *state;
681 GNUNET_assert (NULL != connection);
683 mq = GNUNET_new (struct GNUNET_MQ_Handle);
684 mq->handlers = handlers;
685 mq->error_handler = error_handler;
686 mq->handlers_cls = cls;
687 state = GNUNET_new (struct ClientConnectionState);
688 state->connection = connection;
689 mq->impl_state = state;
690 mq->send_impl = connection_client_send_impl;
691 mq->destroy_impl = connection_client_destroy_impl;
692 mq->cancel_impl = connection_client_cancel_impl;
693 if (NULL != handlers)
694 state->receive_requested = GNUNET_YES;
701 GNUNET_MQ_replace_handlers (struct GNUNET_MQ_Handle *mq,
702 const struct GNUNET_MQ_MessageHandler *new_handlers,
705 /* FIXME: notify implementation? */
706 /* FIXME: what about NULL handlers? abort receive? */
707 mq->handlers = new_handlers;
708 mq->handlers_cls = cls;
713 * Associate the assoc_data in mq with a unique request id.
715 * @param mq message queue, id will be unique for the queue
716 * @param assoc_data to associate
719 GNUNET_MQ_assoc_add (struct GNUNET_MQ_Handle *mq,
724 if (NULL == mq->assoc_map)
726 mq->assoc_map = GNUNET_CONTAINER_multihashmap32_create (8);
730 GNUNET_CONTAINER_multihashmap32_put (mq->assoc_map, id, assoc_data,
731 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
737 GNUNET_MQ_assoc_get (struct GNUNET_MQ_Handle *mq,
740 if (NULL == mq->assoc_map)
742 return GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map, request_id);
747 GNUNET_MQ_assoc_remove (struct GNUNET_MQ_Handle *mq,
752 if (NULL == mq->assoc_map)
754 val = GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map,
756 GNUNET_CONTAINER_multihashmap32_remove_all (mq->assoc_map,
763 GNUNET_MQ_notify_sent (struct GNUNET_MQ_Envelope *mqm,
764 GNUNET_MQ_NotifyCallback cb,
773 GNUNET_MQ_destroy (struct GNUNET_MQ_Handle *mq)
775 if (NULL != mq->destroy_impl)
777 mq->destroy_impl (mq, mq->impl_state);
779 if (NULL != mq->continue_task)
781 GNUNET_SCHEDULER_cancel (mq->continue_task);
782 mq->continue_task = NULL;
784 while (NULL != mq->envelope_head)
786 struct GNUNET_MQ_Envelope *ev;
787 ev = mq->envelope_head;
788 ev->parent_queue = NULL;
789 GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
792 GNUNET_MQ_discard (ev);
794 if (NULL != mq->current_envelope)
796 /* we can only discard envelopes that
798 mq->current_envelope->parent_queue = NULL;
799 GNUNET_MQ_discard (mq->current_envelope);
800 mq->current_envelope = NULL;
802 if (NULL != mq->assoc_map)
804 GNUNET_CONTAINER_multihashmap32_destroy (mq->assoc_map);
805 mq->assoc_map = NULL;
812 const struct GNUNET_MessageHeader *
813 GNUNET_MQ_extract_nested_mh_ (const struct GNUNET_MessageHeader *mh,
817 uint16_t nested_size;
818 const struct GNUNET_MessageHeader *nested_msg;
820 whole_size = ntohs (mh->size);
821 GNUNET_assert (whole_size >= base_size);
822 nested_size = whole_size - base_size;
823 if (0 == nested_size)
825 if (nested_size < sizeof (struct GNUNET_MessageHeader))
830 nested_msg = (const struct GNUNET_MessageHeader *) ((char *) mh + base_size);
831 if (ntohs (nested_msg->size) != nested_size)
841 * Cancel sending the message. Message must have been sent with
842 * #GNUNET_MQ_send before. May not be called after the notify sent
843 * callback has been called
845 * @param ev queued envelope to cancel
848 GNUNET_MQ_send_cancel (struct GNUNET_MQ_Envelope *ev)
850 struct GNUNET_MQ_Handle *mq = ev->parent_queue;
852 GNUNET_assert (NULL != mq);
853 GNUNET_assert (NULL != mq->cancel_impl);
855 if (mq->current_envelope == ev)
857 // complex case, we already started with transmitting
861 // continue sending the next message, if any
862 if (NULL == mq->envelope_head)
864 mq->current_envelope = NULL;
868 mq->current_envelope = mq->envelope_head;
869 GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
871 mq->current_envelope);
873 mq->current_envelope->mh,
879 // simple case, message is still waiting in the queue
880 GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
885 ev->parent_queue = NULL;