2 This file is part of GNUnet.
3 Copyright (C) 2009-2016 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.
21 * @file core/core_api_2.c
22 * @brief core service; this is the main API for encrypted P2P
24 * @author Christian Grothoff
27 #include "gnunet_util_lib.h"
28 #include "gnunet_constants.h"
29 #include "gnunet_core_service.h"
32 #define LOG(kind,...) GNUNET_log_from (kind, "core-api",__VA_ARGS__)
36 * Information we track for each peer.
42 * Corresponding CORE handle.
44 struct GNUNET_CORE_Handle *h;
47 * Message queue for the peer.
49 struct GNUNET_MQ_Handle *mq;
52 * Message we are currently trying to pass to the CORE service
53 * for this peer (from @e mq).
55 struct GNUNET_MQ_Envelope *env;
58 * Value the client returned when we connected, used
59 * as the closure in various places.
64 * Peer the record is about.
66 struct GNUNET_PeerIdentity peer;
69 * SendMessageRequest ID generator for this peer.
77 * Context for the core service connection.
79 struct GNUNET_CORE_Handle
83 * Configuration we're using.
85 const struct GNUNET_CONFIGURATION_Handle *cfg;
88 * Closure for the various callbacks.
93 * Function to call once we've handshaked with the core service.
95 GNUNET_CORE_StartupCallback init;
98 * Function to call whenever we're notified about a peer connecting.
100 GNUNET_CORE_ConnecTEventHandler connects;
103 * Function to call whenever we're notified about a peer disconnecting.
105 GNUNET_CORE_DisconnecTEventHandler disconnects;
108 * Function handlers for messages of particular type.
110 struct GNUNET_MQ_MessageHandler *handlers;
113 * Our message queue for transmissions to the service.
115 struct GNUNET_MQ_Handle *mq;
118 * Hash map listing all of the peers that we are currently
121 struct GNUNET_CONTAINER_MultiPeerMap *peers;
124 * Identity of this peer.
126 struct GNUNET_PeerIdentity me;
129 * ID of reconnect task (if any).
131 struct GNUNET_SCHEDULER_Task *reconnect_task;
134 * Current delay we use for re-trying to connect to core.
136 struct GNUNET_TIME_Relative retry_backoff;
139 * Number of entries in the handlers array.
144 * Did we ever get INIT?
152 * Our current client connection went down. Clean it up
153 * and try to reconnect!
155 * @param h our handle to the core service
158 reconnect (struct GNUNET_CORE_Handle *h);
162 * Task schedule to try to re-connect to core.
164 * @param cls the `struct GNUNET_CORE_Handle`
165 * @param tc task context
168 reconnect_task (void *cls)
170 struct GNUNET_CORE_Handle *h = cls;
172 h->reconnect_task = NULL;
173 LOG (GNUNET_ERROR_TYPE_DEBUG,
174 "Connecting to CORE service after delay\n");
180 * Notify clients about disconnect and free the entry for connected
183 * @param cls the `struct GNUNET_CORE_Handle *`
184 * @param key the peer identity (not used)
185 * @param value the `struct PeerRecord` to free.
186 * @return #GNUNET_YES (continue)
189 disconnect_and_free_peer_entry (void *cls,
190 const struct GNUNET_PeerIdentity *key,
193 struct GNUNET_CORE_Handle *h = cls;
194 struct PeerRecord *pr = value;
196 GNUNET_assert (pr->h == h);
197 if (NULL != h->disconnects)
198 h->disconnects (h->cls,
201 GNUNET_assert (GNUNET_YES ==
202 GNUNET_CONTAINER_multipeermap_remove (h->peers,
205 GNUNET_MQ_destroy (pr->mq);
206 GNUNET_assert (NULL == pr->mq);
213 * Close down any existing connection to the CORE service and
214 * try re-establishing it later.
216 * @param h our handle
219 reconnect_later (struct GNUNET_CORE_Handle *h)
221 GNUNET_assert (NULL == h->reconnect_task);
224 GNUNET_MQ_destroy (h->mq);
227 GNUNET_assert (NULL == h->reconnect_task);
229 GNUNET_SCHEDULER_add_delayed (h->retry_backoff,
232 GNUNET_CONTAINER_multipeermap_iterate (h->peers,
233 &disconnect_and_free_peer_entry,
235 h->retry_backoff = GNUNET_TIME_STD_BACKOFF (h->retry_backoff);
240 * Error handler for the message queue to the CORE service.
241 * On errors, we reconnect.
243 * @param cls closure, a `struct GNUNET_CORE_Handle *`
244 * @param error error code
247 handle_mq_error (void *cls,
248 enum GNUNET_MQ_Error error)
250 struct GNUNET_CORE_Handle *h = cls;
252 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
260 * Inquire with CORE what options should be set for a message
261 * so that it is transmitted with the given @a priority and
262 * the given @a cork value.
264 * @param cork desired corking
265 * @param priority desired message priority
266 * @param[out] flags set to `flags` value for #GNUNET_MQ_set_options()
267 * @return `extra` argument to give to #GNUNET_MQ_set_options()
270 GNUNET_CORE_get_mq_options (int cork,
271 enum GNUNET_CORE_Priority priority,
274 *flags = ((uint64_t) priority) + (((uint64_t) cork) << 32);
280 * Implement sending functionality of a message queue for
281 * us sending messages to a peer.
283 * @param mq the message queue
284 * @param msg the message to send
285 * @param impl_state state of the implementation
288 core_mq_send_impl (struct GNUNET_MQ_Handle *mq,
289 const struct GNUNET_MessageHeader *msg,
292 struct PeerRecord *pr = impl_state;
293 struct GNUNET_CORE_Handle *h = pr->h;
294 struct SendMessageRequest *smr;
295 struct SendMessage *sm;
296 struct GNUNET_MQ_Envelope *env;
300 enum GNUNET_CORE_Priority priority;
304 /* We're currently reconnecting, pretend this worked */
305 GNUNET_MQ_impl_send_continue (mq);
308 GNUNET_assert (NULL == pr->env);
309 /* extract options from envelope */
310 env = GNUNET_MQ_get_current_envelope (mq);
311 GNUNET_break (NULL ==
312 GNUNET_MQ_env_get_options (env,
314 cork = (int) (flags >> 32);
315 priority = (uint32_t) flags;
317 /* check message size for sanity */
318 msize = ntohs (msg->size);
319 if (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE - sizeof (struct SendMessage))
322 GNUNET_MQ_impl_send_continue (mq);
326 /* ask core for transmission */
327 LOG (GNUNET_ERROR_TYPE_DEBUG,
328 "Asking core for transmission of %u bytes to `%s'\n",
329 (unsigned int) msize,
330 GNUNET_i2s (&pr->peer));
331 env = GNUNET_MQ_msg (smr,
332 GNUNET_MESSAGE_TYPE_CORE_SEND_REQUEST);
333 smr->priority = htonl ((uint32_t) priority);
334 // smr->deadline = GNUNET_TIME_absolute_hton (deadline);
335 smr->peer = pr->peer;
336 smr->reserved = htonl (0);
337 smr->size = htons (msize);
338 smr->smr_id = htons (++pr->smr_id_gen);
339 GNUNET_MQ_send (h->mq,
342 /* prepare message with actual transmission data */
343 pr->env = GNUNET_MQ_msg_nested_mh (sm,
344 GNUNET_MESSAGE_TYPE_CORE_SEND,
346 sm->priority = htonl ((uint32_t) priority);
347 // sm->deadline = GNUNET_TIME_absolute_hton (deadline);
349 sm->cork = htonl ((uint32_t) cork);
350 sm->reserved = htonl (0);
351 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
352 "Calling get_message with buffer of %u bytes (%s)\n",
353 (unsigned int) msize,
354 cork ? "corked" : "uncorked");
359 * Handle destruction of a message queue. Implementations must not
360 * free @a mq, but should take care of @a impl_state.
362 * @param mq the message queue to destroy
363 * @param impl_state state of the implementation
366 core_mq_destroy_impl (struct GNUNET_MQ_Handle *mq,
369 struct PeerRecord *pr = impl_state;
371 GNUNET_assert (mq == pr->mq);
377 * Implementation function that cancels the currently sent message.
378 * Should basically undo whatever #mq_send_impl() did.
380 * @param mq message queue
381 * @param impl_state state specific to the implementation
384 core_mq_cancel_impl (struct GNUNET_MQ_Handle *mq,
387 struct PeerRecord *pr = impl_state;
389 GNUNET_assert (NULL != pr->env);
390 GNUNET_MQ_discard (pr->env);
396 * We had an error processing a message we forwarded from a peer to
397 * the CORE service. We should just complain about it but otherwise
398 * continue processing.
401 * @param error error code
404 core_mq_error_handler (void *cls,
405 enum GNUNET_MQ_Error error)
407 /* struct PeerRecord *pr = cls; */
414 * Add the given peer to the list of our connected peers
415 * and create the respective data structures and notify
418 * @param h the core handle
419 * @param peer the peer that is connecting to us
422 connect_peer (struct GNUNET_CORE_Handle *h,
423 const struct GNUNET_PeerIdentity *peer)
425 struct PeerRecord *pr;
429 pr = GNUNET_new (struct PeerRecord);
432 GNUNET_assert (GNUNET_YES ==
433 GNUNET_CONTAINER_multipeermap_put (h->peers,
436 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
437 pr->mq = GNUNET_MQ_queue_for_callbacks (&core_mq_send_impl,
438 &core_mq_destroy_impl,
439 &core_mq_cancel_impl,
442 &core_mq_error_handler,
444 /* get our default options */
445 extra = GNUNET_CORE_get_mq_options (GNUNET_NO,
446 GNUNET_CORE_PRIO_BEST_EFFORT,
448 GNUNET_MQ_set_options (pr->mq,
451 if (NULL != h->connects)
453 pr->client_cls = h->connects (h->cls,
456 GNUNET_MQ_set_handlers_closure (pr->mq,
463 * Handle init reply message received from CORE service. Notify
464 * application that we are now connected to the CORE. Also fake
465 * loopback connection.
467 * @param cls the `struct GNUNET_CORE_Handle`
468 * @param m the init reply
471 handle_init_reply (void *cls,
472 const struct InitReplyMessage *m)
474 struct GNUNET_CORE_Handle *h = cls;
475 GNUNET_CORE_StartupCallback init;
477 GNUNET_break (0 == ntohl (m->reserved));
478 h->retry_backoff = GNUNET_TIME_UNIT_MILLISECONDS;
479 if (NULL != (init = h->init))
481 /* mark so we don't call init on reconnect */
483 h->me = m->my_identity;
484 LOG (GNUNET_ERROR_TYPE_DEBUG,
485 "Connected to core service of peer `%s'.\n",
486 GNUNET_i2s (&h->me));
487 h->have_init = GNUNET_YES;
493 LOG (GNUNET_ERROR_TYPE_DEBUG,
494 "Successfully reconnected to core service.\n");
495 if (GNUNET_NO == h->have_init)
497 h->me = m->my_identity;
498 h->have_init = GNUNET_YES;
502 GNUNET_break (0 == memcmp (&h->me,
504 sizeof (struct GNUNET_PeerIdentity)));
507 /* fake 'connect to self' */
514 * Handle connect message received from CORE service.
515 * Notify the application about the new connection.
517 * @param cls the `struct GNUNET_CORE_Handle`
518 * @param cnm the connect message
521 handle_connect_notify (void *cls,
522 const struct ConnectNotifyMessage *cnm)
524 struct GNUNET_CORE_Handle *h = cls;
525 struct PeerRecord *pr;
527 LOG (GNUNET_ERROR_TYPE_DEBUG,
528 "Received notification about connection from `%s'.\n",
529 GNUNET_i2s (&cnm->peer));
530 if (0 == memcmp (&h->me,
532 sizeof (struct GNUNET_PeerIdentity)))
534 /* connect to self!? */
538 pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
552 * Handle disconnect message received from CORE service.
553 * Notify the application about the lost connection.
555 * @param cls the `struct GNUNET_CORE_Handle`
556 * @param dnm message about the disconnect event
559 handle_disconnect_notify (void *cls,
560 const struct DisconnectNotifyMessage *dnm)
562 struct GNUNET_CORE_Handle *h = cls;
563 struct PeerRecord *pr;
565 if (0 == memcmp (&h->me,
567 sizeof (struct GNUNET_PeerIdentity)))
569 /* disconnect from self!? */
573 GNUNET_break (0 == ntohl (dnm->reserved));
574 LOG (GNUNET_ERROR_TYPE_DEBUG,
575 "Received notification about disconnect from `%s'.\n",
576 GNUNET_i2s (&dnm->peer));
577 pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
585 disconnect_and_free_peer_entry (h,
592 * Check that message received from CORE service is well-formed.
594 * @param cls the `struct GNUNET_CORE_Handle`
595 * @param ntm the message we got
596 * @return #GNUNET_OK if the message is well-formed
599 check_notify_inbound (void *cls,
600 const struct NotifyTrafficMessage *ntm)
603 const struct GNUNET_MessageHeader *em;
605 msize = ntohs (ntm->header.size) - sizeof (struct NotifyTrafficMessage);
606 if (msize < sizeof (struct GNUNET_MessageHeader))
609 return GNUNET_SYSERR;
611 em = (const struct GNUNET_MessageHeader *) &ntm[1];
612 if (msize != ntohs (em->size))
615 return GNUNET_SYSERR;
622 * Handle inbound message received from CORE service. If applicable,
623 * notify the application.
625 * @param cls the `struct GNUNET_CORE_Handle`
626 * @param ntm the message we got from CORE.
629 handle_notify_inbound (void *cls,
630 const struct NotifyTrafficMessage *ntm)
632 struct GNUNET_CORE_Handle *h = cls;
633 const struct GNUNET_MessageHeader *em;
634 struct PeerRecord *pr;
636 LOG (GNUNET_ERROR_TYPE_DEBUG,
637 "Received inbound message from `%s'.\n",
638 GNUNET_i2s (&ntm->peer));
639 em = (const struct GNUNET_MessageHeader *) &ntm[1];
640 pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
648 GNUNET_MQ_inject_message (pr->mq,
654 * Handle message received from CORE service notifying us that we are
655 * now allowed to send a message to a peer. If that message is still
656 * pending, put it into the queue to be transmitted.
658 * @param cls the `struct GNUNET_CORE_Handle`
659 * @param smr the message we got
662 handle_send_ready (void *cls,
663 const struct SendMessageReady *smr)
665 struct GNUNET_CORE_Handle *h = cls;
666 struct PeerRecord *pr;
668 pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
676 LOG (GNUNET_ERROR_TYPE_DEBUG,
677 "Received notification about transmission readiness to `%s'.\n",
678 GNUNET_i2s (&smr->peer));
681 /* request must have been cancelled between the original request
682 * and the response from CORE, ignore CORE's readiness */
685 if (ntohs (smr->smr_id) != pr->smr_id_gen)
687 /* READY message is for expired or cancelled message,
688 * ignore! (we should have already sent another request) */
692 /* ok, all good, send message out! */
693 GNUNET_MQ_send (h->mq,
696 GNUNET_MQ_impl_send_continue (pr->mq);
701 * Our current client connection went down. Clean it up and try to
704 * @param h our handle to the core service
707 reconnect (struct GNUNET_CORE_Handle *h)
709 struct GNUNET_MQ_MessageHandler handlers[] = {
710 GNUNET_MQ_hd_fixed_size (init_reply,
711 GNUNET_MESSAGE_TYPE_CORE_INIT_REPLY,
712 struct InitReplyMessage,
714 GNUNET_MQ_hd_fixed_size (connect_notify,
715 GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT,
716 struct ConnectNotifyMessage,
718 GNUNET_MQ_hd_fixed_size (disconnect_notify,
719 GNUNET_MESSAGE_TYPE_CORE_NOTIFY_DISCONNECT,
720 struct DisconnectNotifyMessage,
722 GNUNET_MQ_hd_var_size (notify_inbound,
723 GNUNET_MESSAGE_TYPE_CORE_NOTIFY_INBOUND,
724 struct NotifyTrafficMessage,
726 GNUNET_MQ_hd_fixed_size (send_ready,
727 GNUNET_MESSAGE_TYPE_CORE_SEND_READY,
728 struct SendMessageReady,
730 GNUNET_MQ_handler_end ()
732 struct InitMessage *init;
733 struct GNUNET_MQ_Envelope *env;
736 GNUNET_assert (NULL == h->mq);
737 h->mq = GNUNET_CLIENT_connecT (h->cfg,
747 env = GNUNET_MQ_msg_extra (init,
748 sizeof (uint16_t) * h->hcnt,
749 GNUNET_MESSAGE_TYPE_CORE_INIT);
750 LOG (GNUNET_ERROR_TYPE_INFO,
751 "(Re)connecting to CORE service\n");
752 init->options = htonl (0);
753 ts = (uint16_t *) &init[1];
754 for (unsigned int hpos = 0; hpos < h->hcnt; hpos++)
755 ts[hpos] = htons (h->handlers[hpos].type);
756 GNUNET_MQ_send (h->mq,
762 * Connect to the core service. Note that the connection may complete
763 * (or fail) asynchronously.
765 * @param cfg configuration to use
766 * @param cls closure for the various callbacks that follow (including handlers in the handlers array)
767 * @param init callback to call once we have successfully
768 * connected to the core service
769 * @param connects function to call on peer connect, can be NULL
770 * @param disconnects function to call on peer disconnect / timeout, can be NULL
771 * @param handlers callbacks for messages we care about, NULL-terminated
772 * @return handle to the core service (only useful for disconnect until @a init is called);
773 * NULL on error (in this case, init is never called)
775 struct GNUNET_CORE_Handle *
776 GNUNET_CORE_connecT (const struct GNUNET_CONFIGURATION_Handle *cfg,
778 GNUNET_CORE_StartupCallback init,
779 GNUNET_CORE_ConnecTEventHandler connects,
780 GNUNET_CORE_DisconnecTEventHandler disconnects,
781 const struct GNUNET_MQ_MessageHandler *handlers)
783 struct GNUNET_CORE_Handle *h;
786 h = GNUNET_new (struct GNUNET_CORE_Handle);
790 h->connects = connects;
791 h->disconnects = disconnects;
792 h->peers = GNUNET_CONTAINER_multipeermap_create (128,
795 if (NULL != handlers)
796 while (NULL != handlers[hcnt].cb)
798 h->handlers = GNUNET_new_array (hcnt + 1,
799 struct GNUNET_MQ_MessageHandler);
800 if (NULL != handlers)
801 GNUNET_memcpy (h->handlers,
803 hcnt * sizeof (struct GNUNET_MQ_MessageHandler));
805 GNUNET_assert (hcnt <
806 (GNUNET_SERVER_MAX_MESSAGE_SIZE -
807 sizeof (struct InitMessage)) / sizeof (uint16_t));
808 LOG (GNUNET_ERROR_TYPE_DEBUG,
809 "Connecting to CORE service\n");
813 GNUNET_CORE_disconnect (h);
821 * Disconnect from the core service.
823 * @param handle connection to core to disconnect
826 GNUNET_CORE_disconnecT (struct GNUNET_CORE_Handle *handle)
828 LOG (GNUNET_ERROR_TYPE_DEBUG,
829 "Disconnecting from CORE service\n");
830 GNUNET_CONTAINER_multipeermap_iterate (handle->peers,
831 &disconnect_and_free_peer_entry,
833 GNUNET_CONTAINER_multipeermap_destroy (handle->peers);
834 handle->peers = NULL;
835 if (NULL != handle->reconnect_task)
837 GNUNET_SCHEDULER_cancel (handle->reconnect_task);
838 handle->reconnect_task = NULL;
840 if (NULL != handle->mq)
842 GNUNET_MQ_destroy (handle->mq);
845 GNUNET_free (handle->handlers);
846 GNUNET_free (handle);
851 * Obtain the message queue for a connected peer.
853 * @param h the core handle
854 * @param pid the identity of the peer to check if it has been connected to us
855 * @return NULL if peer is not connected
857 struct GNUNET_MQ_Handle *
858 GNUNET_CORE_get_mq (const struct GNUNET_CORE_Handle *h,
859 const struct GNUNET_PeerIdentity *pid)
861 struct PeerRecord *pr;
863 pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
871 /* end of core_api.c */