2 This file is part of GNUnet.
3 Copyright (C) 2012-2014 Christian Grothoff (and other contributing authors)
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., 59 Temple Place - Suite 330,
18 Boston, MA 02111-1307, 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
300 * @param tc scheduler context
303 impl_send_continue (void *cls,
304 const struct GNUNET_SCHEDULER_TaskContext *tc)
306 struct GNUNET_MQ_Handle *mq = cls;
307 struct GNUNET_MQ_Envelope *current_envelope;
309 if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
312 mq->continue_task = NULL;
313 /* call is only valid if we're actually currently sending
315 current_envelope = mq->current_envelope;
316 GNUNET_assert (NULL != current_envelope);
317 current_envelope->parent_queue = NULL;
318 if (NULL == mq->envelope_head)
320 mq->current_envelope = NULL;
324 mq->current_envelope = mq->envelope_head;
325 GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
327 mq->current_envelope);
328 mq->send_impl (mq, mq->current_envelope->mh, mq->impl_state);
330 if (NULL != current_envelope->sent_cb)
331 current_envelope->sent_cb (current_envelope->sent_cls);
332 GNUNET_free (current_envelope);
337 * Call the send implementation for the next queued message,
339 * Only useful for implementing message queues,
340 * results in undefined behavior if not used carefully.
342 * @param mq message queue to send the next message with
345 GNUNET_MQ_impl_send_continue (struct GNUNET_MQ_Handle *mq)
347 GNUNET_assert (NULL == mq->continue_task);
348 mq->continue_task = GNUNET_SCHEDULER_add_now (&impl_send_continue,
354 * Create a message queue for the specified handlers.
356 * @param send function the implements sending messages
357 * @param destroy function that implements destroying the queue
358 * @param cancel function that implements canceling a message
359 * @param impl_state for the queue, passed to 'send' and 'destroy'
360 * @param handlers array of message handlers
361 * @param error_handler handler for read and write errors
362 * @param cls closure for message handlers and error handler
363 * @return a new message queue
365 struct GNUNET_MQ_Handle *
366 GNUNET_MQ_queue_for_callbacks (GNUNET_MQ_SendImpl send,
367 GNUNET_MQ_DestroyImpl destroy,
368 GNUNET_MQ_CancelImpl cancel,
370 const struct GNUNET_MQ_MessageHandler *handlers,
371 GNUNET_MQ_ErrorHandler error_handler,
374 struct GNUNET_MQ_Handle *mq;
376 mq = GNUNET_new (struct GNUNET_MQ_Handle);
377 mq->send_impl = send;
378 mq->destroy_impl = destroy;
379 mq->cancel_impl = cancel;
380 mq->handlers = handlers;
381 mq->handlers_cls = cls;
382 mq->impl_state = impl_state;
389 * Get the message that should currently be sent.
390 * Fails if there is no current message.
391 * Only useful for implementing message queues,
392 * results in undefined behavior if not used carefully.
394 * @param mq message queue with the current message
395 * @return message to send, never NULL
397 const struct GNUNET_MessageHeader *
398 GNUNET_MQ_impl_current (struct GNUNET_MQ_Handle *mq)
400 if (NULL == mq->current_envelope)
402 if (NULL == mq->current_envelope->mh)
404 return mq->current_envelope->mh;
409 * Get the implementation state associated with the
412 * While the GNUNET_MQ_Impl* callbacks receive the
413 * implementation state, continuations that are scheduled
414 * by the implementation function often only have one closure
415 * argument, with this function it is possible to get at the
416 * implementation state when only passing the GNUNET_MQ_Handle
419 * @param mq message queue with the current message
420 * @return message to send, never NULL
423 GNUNET_MQ_impl_state (struct GNUNET_MQ_Handle *mq)
425 return mq->impl_state;
429 struct GNUNET_MQ_Envelope *
430 GNUNET_MQ_msg_ (struct GNUNET_MessageHeader **mhp,
434 struct GNUNET_MQ_Envelope *mqm;
436 mqm = GNUNET_malloc (sizeof *mqm + size);
437 mqm->mh = (struct GNUNET_MessageHeader *) &mqm[1];
438 mqm->mh->size = htons (size);
439 mqm->mh->type = htons (type);
447 * Implementation of the GNUNET_MQ_msg_nested_mh macro.
449 * @param mhp pointer to the message header pointer that will be changed to allocate at
450 * the newly allocated space for the message.
451 * @param base_size size of the data before the nested message
452 * @param type type of the message in the envelope
453 * @param nested_mh the message to append to the message after base_size
455 struct GNUNET_MQ_Envelope *
456 GNUNET_MQ_msg_nested_mh_ (struct GNUNET_MessageHeader **mhp,
459 const struct GNUNET_MessageHeader *nested_mh)
461 struct GNUNET_MQ_Envelope *mqm;
464 if (NULL == nested_mh)
465 return GNUNET_MQ_msg_ (mhp, base_size, type);
467 size = base_size + ntohs (nested_mh->size);
469 /* check for uint16_t overflow */
470 if (size < base_size)
473 mqm = GNUNET_MQ_msg_ (mhp, size, type);
474 memcpy ((char *) mqm->mh + base_size, nested_mh, ntohs (nested_mh->size));
481 * Transmit a queued message to the session's client.
483 * @param cls consensus session
484 * @param size number of bytes available in buf
485 * @param buf where the callee should write the message
486 * @return number of bytes written to buf
489 transmit_queued (void *cls, size_t size,
492 struct GNUNET_MQ_Handle *mq = cls;
493 struct ServerClientSocketState *state = GNUNET_MQ_impl_state (mq);
494 const struct GNUNET_MessageHeader *msg = GNUNET_MQ_impl_current (mq);
497 GNUNET_assert (NULL != buf);
499 msg_size = ntohs (msg->size);
500 GNUNET_assert (size >= msg_size);
501 memcpy (buf, msg, msg_size);
504 GNUNET_MQ_impl_send_continue (mq);
511 server_client_destroy_impl (struct GNUNET_MQ_Handle *mq,
514 struct ServerClientSocketState *state = impl_state;
516 if (NULL != state->th)
518 GNUNET_SERVER_notify_transmit_ready_cancel (state->th);
522 GNUNET_assert (NULL != mq);
523 GNUNET_assert (NULL != state);
524 GNUNET_SERVER_client_drop (state->client);
530 server_client_send_impl (struct GNUNET_MQ_Handle *mq,
531 const struct GNUNET_MessageHeader *msg,
534 struct ServerClientSocketState *state = impl_state;
536 GNUNET_assert (NULL != mq);
537 GNUNET_assert (NULL != state);
539 GNUNET_SERVER_notify_transmit_ready (state->client, ntohs (msg->size),
540 GNUNET_TIME_UNIT_FOREVER_REL,
541 &transmit_queued, mq);
545 struct GNUNET_MQ_Handle *
546 GNUNET_MQ_queue_for_server_client (struct GNUNET_SERVER_Client *client)
548 struct GNUNET_MQ_Handle *mq;
549 struct ServerClientSocketState *scss;
551 mq = GNUNET_new (struct GNUNET_MQ_Handle);
552 scss = GNUNET_new (struct ServerClientSocketState);
553 mq->impl_state = scss;
554 scss->client = client;
555 GNUNET_SERVER_client_keep (client);
556 mq->send_impl = server_client_send_impl;
557 mq->destroy_impl = server_client_destroy_impl;
563 * Type of a function to call when we receive a message
567 * @param msg message received, NULL on timeout or fatal error
570 handle_client_message (void *cls,
571 const struct GNUNET_MessageHeader *msg)
573 struct GNUNET_MQ_Handle *mq = cls;
574 struct ClientConnectionState *state;
576 state = mq->impl_state;
580 GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_READ);
584 GNUNET_CLIENT_receive (state->connection, handle_client_message, mq,
585 GNUNET_TIME_UNIT_FOREVER_REL);
587 GNUNET_MQ_inject_message (mq, msg);
592 * Transmit a queued message to the session's client.
594 * @param cls consensus session
595 * @param size number of bytes available in @a buf
596 * @param buf where the callee should write the message
597 * @return number of bytes written to buf
600 connection_client_transmit_queued (void *cls,
604 struct GNUNET_MQ_Handle *mq = cls;
605 const struct GNUNET_MessageHeader *msg;
606 struct ClientConnectionState *state = mq->impl_state;
609 GNUNET_assert (NULL != mq);
610 msg = GNUNET_MQ_impl_current (mq);
614 GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_READ);
618 if ( (GNUNET_YES == state->receive_requested) &&
619 (GNUNET_NO == state->receive_active) )
621 state->receive_active = GNUNET_YES;
622 GNUNET_CLIENT_receive (state->connection, handle_client_message, mq,
623 GNUNET_TIME_UNIT_FOREVER_REL);
626 msg_size = ntohs (msg->size);
627 GNUNET_assert (size >= msg_size);
628 memcpy (buf, msg, msg_size);
631 GNUNET_MQ_impl_send_continue (mq);
638 connection_client_destroy_impl (struct GNUNET_MQ_Handle *mq,
641 GNUNET_free (impl_state);
646 connection_client_send_impl (struct GNUNET_MQ_Handle *mq,
647 const struct GNUNET_MessageHeader *msg,
650 struct ClientConnectionState *state = impl_state;
652 GNUNET_assert (NULL != state);
653 GNUNET_assert (NULL == state->th);
655 GNUNET_CLIENT_notify_transmit_ready (state->connection, ntohs (msg->size),
656 GNUNET_TIME_UNIT_FOREVER_REL, GNUNET_NO,
657 &connection_client_transmit_queued, mq);
658 GNUNET_assert (NULL != state->th);
663 connection_client_cancel_impl (struct GNUNET_MQ_Handle *mq,
666 struct ClientConnectionState *state = impl_state;
667 GNUNET_assert (NULL != state->th);
668 GNUNET_CLIENT_notify_transmit_ready_cancel (state->th);
673 struct GNUNET_MQ_Handle *
674 GNUNET_MQ_queue_for_connection_client (struct GNUNET_CLIENT_Connection *connection,
675 const struct GNUNET_MQ_MessageHandler *handlers,
676 GNUNET_MQ_ErrorHandler error_handler,
679 struct GNUNET_MQ_Handle *mq;
680 struct ClientConnectionState *state;
682 GNUNET_assert (NULL != connection);
684 mq = GNUNET_new (struct GNUNET_MQ_Handle);
685 mq->handlers = handlers;
686 mq->error_handler = error_handler;
687 mq->handlers_cls = cls;
688 state = GNUNET_new (struct ClientConnectionState);
689 state->connection = connection;
690 mq->impl_state = state;
691 mq->send_impl = connection_client_send_impl;
692 mq->destroy_impl = connection_client_destroy_impl;
693 mq->cancel_impl = connection_client_cancel_impl;
694 if (NULL != handlers)
695 state->receive_requested = GNUNET_YES;
702 GNUNET_MQ_replace_handlers (struct GNUNET_MQ_Handle *mq,
703 const struct GNUNET_MQ_MessageHandler *new_handlers,
706 /* FIXME: notify implementation? */
707 /* FIXME: what about NULL handlers? abort receive? */
708 mq->handlers = new_handlers;
709 mq->handlers_cls = cls;
714 * Associate the assoc_data in mq with a unique request id.
716 * @param mq message queue, id will be unique for the queue
717 * @param assoc_data to associate
720 GNUNET_MQ_assoc_add (struct GNUNET_MQ_Handle *mq,
725 if (NULL == mq->assoc_map)
727 mq->assoc_map = GNUNET_CONTAINER_multihashmap32_create (8);
731 GNUNET_CONTAINER_multihashmap32_put (mq->assoc_map, id, assoc_data,
732 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
738 GNUNET_MQ_assoc_get (struct GNUNET_MQ_Handle *mq,
741 if (NULL == mq->assoc_map)
743 return GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map, request_id);
748 GNUNET_MQ_assoc_remove (struct GNUNET_MQ_Handle *mq,
753 if (NULL == mq->assoc_map)
755 val = GNUNET_CONTAINER_multihashmap32_get (mq->assoc_map, request_id);
756 GNUNET_CONTAINER_multihashmap32_remove_all (mq->assoc_map, request_id);
762 GNUNET_MQ_notify_sent (struct GNUNET_MQ_Envelope *mqm,
763 GNUNET_MQ_NotifyCallback cb,
772 GNUNET_MQ_destroy (struct GNUNET_MQ_Handle *mq)
774 if (NULL != mq->destroy_impl)
776 mq->destroy_impl (mq, mq->impl_state);
778 if (NULL != mq->continue_task)
780 GNUNET_SCHEDULER_cancel (mq->continue_task);
781 mq->continue_task = NULL;
783 while (NULL != mq->envelope_head)
785 struct GNUNET_MQ_Envelope *ev;
786 ev = mq->envelope_head;
787 ev->parent_queue = NULL;
788 GNUNET_CONTAINER_DLL_remove (mq->envelope_head, mq->envelope_tail, ev);
789 GNUNET_MQ_discard (ev);
792 if (NULL != mq->current_envelope)
794 /* we can only discard envelopes that
796 mq->current_envelope->parent_queue = NULL;
797 GNUNET_MQ_discard (mq->current_envelope);
798 mq->current_envelope = NULL;
801 if (NULL != mq->assoc_map)
803 GNUNET_CONTAINER_multihashmap32_destroy (mq->assoc_map);
804 mq->assoc_map = NULL;
811 const struct GNUNET_MessageHeader *
812 GNUNET_MQ_extract_nested_mh_ (const struct GNUNET_MessageHeader *mh,
816 uint16_t nested_size;
817 const struct GNUNET_MessageHeader *nested_msg;
819 whole_size = ntohs (mh->size);
820 GNUNET_assert (whole_size >= base_size);
821 nested_size = whole_size - base_size;
822 if (0 == nested_size)
824 if (nested_size < sizeof (struct GNUNET_MessageHeader))
829 nested_msg = (const struct GNUNET_MessageHeader *) ((char *) mh + base_size);
830 if (ntohs (nested_msg->size) != nested_size)
840 * Cancel sending the message. Message must have been sent with
841 * #GNUNET_MQ_send before. May not be called after the notify sent
842 * callback has been called
844 * @param ev queued envelope to cancel
847 GNUNET_MQ_send_cancel (struct GNUNET_MQ_Envelope *ev)
849 struct GNUNET_MQ_Handle *mq = ev->parent_queue;
851 GNUNET_assert (NULL != mq);
852 GNUNET_assert (NULL != mq->cancel_impl);
854 if (mq->current_envelope == ev) {
855 // complex case, we already started with transmitting
857 mq->cancel_impl (mq, mq->impl_state);
858 // continue sending the next message, if any
859 if (NULL == mq->envelope_head)
861 mq->current_envelope = NULL;
865 mq->current_envelope = mq->envelope_head;
866 GNUNET_CONTAINER_DLL_remove (mq->envelope_head,
868 mq->current_envelope);
869 mq->send_impl (mq, mq->current_envelope->mh, mq->impl_state);
872 // simple case, message is still waiting in the queue
873 GNUNET_CONTAINER_DLL_remove (mq->envelope_head, mq->envelope_tail, ev);
876 ev->parent_queue = NULL;