src/core/core_api.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2009-2016 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  * @file core/core_api.c
  22  * @brief core service; this is the main API for encrypted P2P
  23  *        communications
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_constants.h"
  29 #include "gnunet_core_service.h"
  30 #include "core.h"
  31
  32 #define LOG(kind,...) GNUNET_log_from (kind, "core-api",__VA_ARGS__)
  33
  34
  35 /**
  36  * Handle for a transmission request.
  37  */
  38 struct GNUNET_CORE_TransmitHandle
  39 {
  40
  41   /**
  42    * Corresponding peer record.
  43    */
  44   struct PeerRecord *peer;
  45
  46   /**
  47    * Function that will be called to get the actual request
  48    * (once we are ready to transmit this request to the core).
  49    * The function will be called with a NULL buffer to signal
  50    * timeout.
  51    */
  52   GNUNET_CONNECTION_TransmitReadyNotify get_message;
  53
  54   /**
  55    * Closure for @e get_message.
  56    */
  57   void *get_message_cls;
  58
  59   /**
  60    * Deadline for the transmission (the request does not get cancelled
  61    * at this time, this is merely how soon the application wants this out).
  62    */
  63   struct GNUNET_TIME_Absolute deadline;
  64
  65   /**
  66    * When did this request get queued?
  67    */
  68   struct GNUNET_TIME_Absolute request_time;
  69
  70   /**
  71    * How important is this message?
  72    */
  73   enum GNUNET_CORE_Priority priority;
  74
  75   /**
  76    * Is corking allowed?
  77    */
  78   int cork;
  79
  80   /**
  81    * Size of this request.
  82    */
  83   uint16_t msize;
  84
  85   /**
  86    * Send message request ID for this request.
  87    */
  88   uint16_t smr_id;
  89
  90 };
  91
  92
  93 /**
  94  * Information we track for each peer.
  95  */
  96 struct PeerRecord
  97 {
  98
  99   /**
 100    * Corresponding CORE handle.
 101    */
 102   struct GNUNET_CORE_Handle *ch;
 103
 104   /**
 105    * Pending request, if any. 'th->peer' is set to NULL if the
 106    * request is not active.
 107    */
 108   struct GNUNET_CORE_TransmitHandle th;
 109
 110   /**
 111    * Peer the record is about.
 112    */
 113   struct GNUNET_PeerIdentity peer;
 114
 115   /**
 116    * SendMessageRequest ID generator for this peer.
 117    */
 118   uint16_t smr_id_gen;
 119
 120 };
 121
 122
 123 /**
 124  * Context for the core service connection.
 125  */
 126 struct GNUNET_CORE_Handle
 127 {
 128
 129   /**
 130    * Configuration we're using.
 131    */
 132   const struct GNUNET_CONFIGURATION_Handle *cfg;
 133
 134   /**
 135    * Closure for the various callbacks.
 136    */
 137   void *cls;
 138
 139   /**
 140    * Function to call once we've handshaked with the core service.
 141    */
 142   GNUNET_CORE_StartupCallback init;
 143
 144   /**
 145    * Function to call whenever we're notified about a peer connecting.
 146    */
 147   GNUNET_CORE_ConnectEventHandler connects;
 148
 149   /**
 150    * Function to call whenever we're notified about a peer disconnecting.
 151    */
 152   GNUNET_CORE_DisconnectEventHandler disconnects;
 153
 154   /**
 155    * Function to call whenever we receive an inbound message.
 156    */
 157   GNUNET_CORE_MessageCallback inbound_notify;
 158
 159   /**
 160    * Function to call whenever we receive an outbound message.
 161    */
 162   GNUNET_CORE_MessageCallback outbound_notify;
 163
 164   /**
 165    * Function handlers for messages of particular type.
 166    */
 167   struct GNUNET_CORE_MessageHandler *handlers;
 168
 169   /**
 170    * Our message queue for transmissions to the service.
 171    */
 172   struct GNUNET_MQ_Handle *mq;
 173
 174   /**
 175    * Hash map listing all of the peers that we are currently
 176    * connected to.
 177    */
 178   struct GNUNET_CONTAINER_MultiPeerMap *peers;
 179
 180   /**
 181    * Identity of this peer.
 182    */
 183   struct GNUNET_PeerIdentity me;
 184
 185   /**
 186    * ID of reconnect task (if any).
 187    */
 188   struct GNUNET_SCHEDULER_Task *reconnect_task;
 189
 190   /**
 191    * Current delay we use for re-trying to connect to core.
 192    */
 193   struct GNUNET_TIME_Relative retry_backoff;
 194
 195   /**
 196    * Number of entries in the handlers array.
 197    */
 198   unsigned int hcnt;
 199
 200   /**
 201    * For inbound notifications without a specific handler, do
 202    * we expect to only receive headers?
 203    */
 204   int inbound_hdr_only;
 205
 206   /**
 207    * For outbound notifications without a specific handler, do
 208    * we expect to only receive headers?
 209    */
 210   int outbound_hdr_only;
 211
 212   /**
 213    * Are we currently disconnected and hence unable to forward
 214    * requests?
 215    */
 216   int currently_down;
 217
 218   /**
 219    * Did we ever get INIT?
 220    */
 221   int have_init;
 222
 223 };
 224
 225
 226 /**
 227  * Our current client connection went down.  Clean it up
 228  * and try to reconnect!
 229  *
 230  * @param h our handle to the core service
 231  */
 232 static void
 233 reconnect (struct GNUNET_CORE_Handle *h);
 234
 235
 236 /**
 237  * Task schedule to try to re-connect to core.
 238  *
 239  * @param cls the `struct GNUNET_CORE_Handle`
 240  * @param tc task context
 241  */
 242 static void
 243 reconnect_task (void *cls)
 244 {
 245   struct GNUNET_CORE_Handle *h = cls;
 246
 247   h->reconnect_task = NULL;
 248   LOG (GNUNET_ERROR_TYPE_DEBUG,
 249        "Connecting to CORE service after delay\n");
 250   reconnect (h);
 251 }
 252
 253
 254 /**
 255  * Notify clients about disconnect and free the entry for connected
 256  * peer.
 257  *
 258  * @param cls the `struct GNUNET_CORE_Handle *`
 259  * @param key the peer identity (not used)
 260  * @param value the `struct PeerRecord` to free.
 261  * @return #GNUNET_YES (continue)
 262  */
 263 static int
 264 disconnect_and_free_peer_entry (void *cls,
 265                                 const struct GNUNET_PeerIdentity *key,
 266                                 void *value)
 267 {
 268   struct GNUNET_CORE_Handle *h = cls;
 269   struct GNUNET_CORE_TransmitHandle *th;
 270   struct PeerRecord *pr = value;
 271
 272   if (NULL != h->disconnects)
 273     h->disconnects (h->cls,
 274                     &pr->peer);
 275   /* all requests should have been cancelled, clean up anyway, just in case */
 276   th = &pr->th;
 277   if (NULL != th->peer)
 278   {
 279     GNUNET_break (0);
 280     th->peer = NULL;
 281   }
 282   /* done with 'voluntary' cleanups, now on to normal freeing */
 283   GNUNET_assert (GNUNET_YES ==
 284                  GNUNET_CONTAINER_multipeermap_remove (h->peers,
 285                                                        key,
 286                                                        pr));
 287   GNUNET_assert (pr->ch == h);
 288   GNUNET_free (pr);
 289   return GNUNET_YES;
 290 }
 291
 292
 293 /**
 294  * Close down any existing connection to the CORE service and
 295  * try re-establishing it later.
 296  *
 297  * @param h our handle
 298  */
 299 static void
 300 reconnect_later (struct GNUNET_CORE_Handle *h)
 301 {
 302   GNUNET_assert (NULL == h->reconnect_task);
 303   if (NULL != h->mq)
 304   {
 305     GNUNET_MQ_destroy (h->mq);
 306     h->mq = NULL;
 307   }
 308   h->currently_down = GNUNET_YES;
 309   GNUNET_assert (h->reconnect_task == NULL);
 310   h->reconnect_task =
 311       GNUNET_SCHEDULER_add_delayed (h->retry_backoff,
 312                                     &reconnect_task,
 313                                     h);
 314   GNUNET_CONTAINER_multipeermap_iterate (h->peers,
 315                                          &disconnect_and_free_peer_entry,
 316                                          h);
 317   h->retry_backoff = GNUNET_TIME_STD_BACKOFF (h->retry_backoff);
 318 }
 319
 320
 321 /**
 322  * Generic error handler, called with the appropriate error code and
 323  * the same closure specified at the creation of the message queue.
 324  * Not every message queue implementation supports an error handler.
 325  *
 326  * @param cls closure, a `struct GNUNET_CORE_Handle *`
 327  * @param error error code
 328  */
 329 static void
 330 handle_mq_error (void *cls,
 331                  enum GNUNET_MQ_Error error)
 332 {
 333   struct GNUNET_CORE_Handle *h = cls;
 334
 335   reconnect_later (h);
 336 }
 337
 338
 339 /**
 340  * Handle  init  reply message  received  from  CORE service.   Notify
 341  * application  that we  are now  connected  to the  CORE.  Also  fake
 342  * loopback connection.
 343  *
 344  * @param cls the `struct GNUNET_CORE_Handle`
 345  * @param m the init reply
 346  */
 347 static void
 348 handle_init_reply (void *cls,
 349                    const struct InitReplyMessage *m)
 350 {
 351   struct GNUNET_CORE_Handle *h = cls;
 352   GNUNET_CORE_StartupCallback init;
 353   struct PeerRecord *pr;
 354
 355   GNUNET_break (0 == ntohl (m->reserved));
 356   GNUNET_break (GNUNET_YES == h->currently_down);
 357   h->currently_down = GNUNET_NO;
 358   h->retry_backoff = GNUNET_TIME_UNIT_MILLISECONDS;
 359   if (NULL != (init = h->init))
 360   {
 361     /* mark so we don't call init on reconnect */
 362     h->init = NULL;
 363     h->me = m->my_identity;
 364     LOG (GNUNET_ERROR_TYPE_DEBUG,
 365          "Connected to core service of peer `%s'.\n",
 366          GNUNET_i2s (&h->me));
 367     h->have_init = GNUNET_YES;
 368     init (h->cls,
 369           &h->me);
 370   }
 371   else
 372   {
 373     LOG (GNUNET_ERROR_TYPE_DEBUG,
 374          "Successfully reconnected to core service.\n");
 375     if (GNUNET_NO == h->have_init)
 376     {
 377       h->me = m->my_identity;
 378       h->have_init = GNUNET_YES;
 379     }
 380     else
 381     {
 382       GNUNET_break (0 == memcmp (&h->me,
 383                                  &m->my_identity,
 384                                  sizeof (struct GNUNET_PeerIdentity)));
 385     }
 386   }
 387   /* fake 'connect to self' */
 388   pr = GNUNET_new (struct PeerRecord);
 389   pr->peer = h->me;
 390   pr->ch = h;
 391   GNUNET_assert (GNUNET_YES ==
 392                  GNUNET_CONTAINER_multipeermap_put (h->peers,
 393                                                     &h->me,
 394                                                     pr,
 395                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 396   if (NULL != h->connects)
 397     h->connects (h->cls,
 398                  &pr->peer);
 399 }
 400
 401
 402 /**
 403  * Handle connect message received from CORE service.
 404  * Notify the application about the new connection.
 405  *
 406  * @param cls the `struct GNUNET_CORE_Handle`
 407  * @param cnm the connect message
 408  */
 409 static void
 410 handle_connect_notify (void *cls,
 411                        const struct ConnectNotifyMessage * cnm)
 412 {
 413   struct GNUNET_CORE_Handle *h = cls;
 414   struct PeerRecord *pr;
 415
 416   GNUNET_break (GNUNET_NO == h->currently_down);
 417   LOG (GNUNET_ERROR_TYPE_DEBUG,
 418        "Received notification about connection from `%s'.\n",
 419        GNUNET_i2s (&cnm->peer));
 420   if (0 == memcmp (&h->me,
 421                    &cnm->peer,
 422                    sizeof (struct GNUNET_PeerIdentity)))
 423   {
 424     /* connect to self!? */
 425     GNUNET_break (0);
 426     return;
 427   }
 428   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 429                                           &cnm->peer);
 430   if (NULL != pr)
 431   {
 432     GNUNET_break (0);
 433     reconnect_later (h);
 434     return;
 435   }
 436   pr = GNUNET_new (struct PeerRecord);
 437   pr->peer = cnm->peer;
 438   pr->ch = h;
 439   GNUNET_assert (GNUNET_YES ==
 440                  GNUNET_CONTAINER_multipeermap_put (h->peers,
 441                                                     &cnm->peer,
 442                                                     pr,
 443                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 444   if (NULL != h->connects)
 445     h->connects (h->cls,
 446                  &pr->peer);
 447 }
 448
 449
 450 /**
 451  * Handle disconnect message received from CORE service.
 452  * Notify the application about the lost connection.
 453  *
 454  * @param cls the `struct GNUNET_CORE_Handle`
 455  * @param dnm message about the disconnect event
 456  */
 457 static void
 458 handle_disconnect_notify (void *cls,
 459                           const struct DisconnectNotifyMessage * dnm)
 460 {
 461   struct GNUNET_CORE_Handle *h = cls;
 462   struct PeerRecord *pr;
 463
 464   GNUNET_break (GNUNET_NO == h->currently_down);
 465   if (0 == memcmp (&h->me,
 466                    &dnm->peer,
 467                    sizeof (struct GNUNET_PeerIdentity)))
 468   {
 469     /* connection to self!? */
 470     GNUNET_break (0);
 471     return;
 472   }
 473   GNUNET_break (0 == ntohl (dnm->reserved));
 474   LOG (GNUNET_ERROR_TYPE_DEBUG,
 475        "Received notification about disconnect from `%s'.\n",
 476        GNUNET_i2s (&dnm->peer));
 477   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 478                                           &dnm->peer);
 479   if (NULL == pr)
 480   {
 481     GNUNET_break (0);
 482     reconnect_later (h);
 483     return;
 484   }
 485   disconnect_and_free_peer_entry (h,
 486                                   &dnm->peer,
 487                                   pr);
 488 }
 489
 490
 491 /**
 492  * Check that message received from CORE service is well-formed.
 493  *
 494  * @param cls the `struct GNUNET_CORE_Handle`
 495  * @param ntm the message we got
 496  * @return #GNUNET_OK if the message is well-formed
 497  */
 498 static int
 499 check_notify_inbound (void *cls,
 500                       const struct NotifyTrafficMessage *ntm)
 501 {
 502   struct GNUNET_CORE_Handle *h = cls;
 503   uint16_t msize;
 504   const struct GNUNET_MessageHeader *em;
 505
 506   GNUNET_break (GNUNET_NO == h->currently_down);
 507   msize = ntohs (ntm->header.size) - sizeof (struct NotifyTrafficMessage);
 508   if (msize < sizeof (struct GNUNET_MessageHeader))
 509   {
 510     GNUNET_break (0);
 511     return GNUNET_SYSERR;
 512   }
 513   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 514   if ( (GNUNET_NO == h->inbound_hdr_only) &&
 515        (msize != ntohs (em->size)) )
 516   {
 517     GNUNET_break (0);
 518     return GNUNET_SYSERR;
 519   }
 520   return GNUNET_OK;
 521 }
 522
 523
 524 /**
 525  * Handle inbound message received from CORE service.  If applicable,
 526  * notify the application.
 527  *
 528  * @param cls the `struct GNUNET_CORE_Handle`
 529  * @param ntm the message we got from CORE.
 530  */
 531 static void
 532 handle_notify_inbound (void *cls,
 533                        const struct NotifyTrafficMessage *ntm)
 534 {
 535   struct GNUNET_CORE_Handle *h = cls;
 536   const struct GNUNET_MessageHeader *em;
 537   struct PeerRecord *pr;
 538   uint16_t et;
 539
 540   GNUNET_break (GNUNET_NO == h->currently_down);
 541   LOG (GNUNET_ERROR_TYPE_DEBUG,
 542        "Received inbound message from `%s'.\n",
 543        GNUNET_i2s (&ntm->peer));
 544   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 545   et = ntohs (em->type);
 546   for (unsigned int hpos = 0; NULL != h->handlers[hpos].callback; hpos++)
 547   {
 548     const struct GNUNET_CORE_MessageHandler *mh;
 549
 550     mh = &h->handlers[hpos];
 551     if (mh->type != et)
 552       continue;
 553     if ( (mh->expected_size != ntohs (em->size)) &&
 554          (0 != mh->expected_size) )
 555     {
 556       LOG (GNUNET_ERROR_TYPE_ERROR,
 557            "Unexpected message size %u for message of type %u from peer `%s'\n",
 558            htons (em->size),
 559            mh->type,
 560            GNUNET_i2s (&ntm->peer));
 561       GNUNET_break_op (0);
 562       continue;
 563     }
 564     pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 565                                             &ntm->peer);
 566     if (NULL == pr)
 567     {
 568       GNUNET_break (0);
 569       reconnect_later (h);
 570       return;
 571     }
 572     if (GNUNET_OK !=
 573         h->handlers[hpos].callback (h->cls,
 574                                     &ntm->peer,
 575                                     em))
 576     {
 577       /* error in processing, do not process other messages! */
 578       break;
 579     }
 580   }
 581   if (NULL != h->inbound_notify)
 582     h->inbound_notify (h->cls,
 583                        &ntm->peer,
 584                        em);
 585 }
 586
 587
 588 /**
 589  * Check that message received from CORE service is well-formed.
 590  *
 591  * @param cls the `struct GNUNET_CORE_Handle`
 592  * @param ntm the message we got
 593  * @return #GNUNET_OK if the message is well-formed
 594  */
 595 static int
 596 check_notify_outbound (void *cls,
 597                        const struct NotifyTrafficMessage *ntm)
 598 {
 599   struct GNUNET_CORE_Handle *h = cls;
 600   uint16_t msize;
 601   const struct GNUNET_MessageHeader *em;
 602
 603   GNUNET_break (GNUNET_NO == h->currently_down);
 604   LOG (GNUNET_ERROR_TYPE_DEBUG,
 605        "Received outbound message from `%s'.\n",
 606        GNUNET_i2s (&ntm->peer));
 607   msize = ntohs (ntm->header.size) - sizeof (struct NotifyTrafficMessage);
 608   if (msize < sizeof (struct GNUNET_MessageHeader))
 609   {
 610     GNUNET_break (0);
 611     return GNUNET_SYSERR;
 612   }
 613   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 614   if ( (GNUNET_NO == h->outbound_hdr_only) &&
 615        (msize != ntohs (em->size)) )
 616   {
 617     GNUNET_break (0);
 618     return GNUNET_SYSERR;
 619   }
 620   return GNUNET_OK;
 621 }
 622
 623
 624 /**
 625  * Handle outbound message received from CORE service.  If applicable,
 626  * notify the application.
 627  *
 628  * @param cls the `struct GNUNET_CORE_Handle`
 629  * @param ntm the message we got
 630  */
 631 static void
 632 handle_notify_outbound (void *cls,
 633                         const struct NotifyTrafficMessage *ntm)
 634 {
 635   struct GNUNET_CORE_Handle *h = cls;
 636   const struct GNUNET_MessageHeader *em;
 637
 638   GNUNET_break (GNUNET_NO == h->currently_down);
 639   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 640   LOG (GNUNET_ERROR_TYPE_DEBUG,
 641        "Received notification about transmission to `%s'.\n",
 642        GNUNET_i2s (&ntm->peer));
 643   if (NULL == h->outbound_notify)
 644   {
 645     GNUNET_break (0);
 646     return;
 647   }
 648   h->outbound_notify (h->cls,
 649                       &ntm->peer,
 650                       em);
 651 }
 652
 653
 654 /**
 655  * Handle message received from CORE service notifying us that we are
 656  * now allowed to send a message to a peer.  If that message is still
 657  * pending, put it into the queue to be transmitted.
 658  *
 659  * @param cls the `struct GNUNET_CORE_Handle`
 660  * @param ntm the message we got
 661  */
 662 static void
 663 handle_send_ready (void *cls,
 664                    const struct SendMessageReady *smr)
 665 {
 666   struct GNUNET_CORE_Handle *h = cls;
 667   struct PeerRecord *pr;
 668   struct GNUNET_CORE_TransmitHandle *th;
 669   struct SendMessage *sm;
 670   struct GNUNET_MQ_Envelope *env;
 671   struct GNUNET_TIME_Relative delay;
 672   struct GNUNET_TIME_Relative overdue;
 673   unsigned int ret;
 674
 675   GNUNET_break (GNUNET_NO == h->currently_down);
 676   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 677                                           &smr->peer);
 678   if (NULL == pr)
 679   {
 680     GNUNET_break (0);
 681     reconnect_later (h);
 682     return;
 683   }
 684   LOG (GNUNET_ERROR_TYPE_DEBUG,
 685        "Received notification about transmission readiness to `%s'.\n",
 686        GNUNET_i2s (&smr->peer));
 687   if (NULL == pr->th.peer)
 688   {
 689     /* request must have been cancelled between the original request
 690      * and the response from CORE, ignore CORE's readiness */
 691     return;
 692   }
 693   th = &pr->th;
 694   if (ntohs (smr->smr_id) != th->smr_id)
 695   {
 696     /* READY message is for expired or cancelled message,
 697      * ignore! (we should have already sent another request) */
 698     return;
 699   }
 700   /* ok, all good, send message out! */
 701   th->peer = NULL;
 702   env = GNUNET_MQ_msg_extra (sm,
 703                              th->msize,
 704                              GNUNET_MESSAGE_TYPE_CORE_SEND);
 705   sm->priority = htonl ((uint32_t) th->priority);
 706   sm->deadline = GNUNET_TIME_absolute_hton (th->deadline);
 707   sm->peer = pr->peer;
 708   sm->cork = htonl ((uint32_t) th->cork);
 709   sm->reserved = htonl (0);
 710   ret = th->get_message (th->get_message_cls,
 711                          th->msize,
 712                          &sm[1]);
 713   sm->header.size = htons (ret + sizeof (struct SendMessage));
 714   th->msize = ret;
 715   // GNUNET_assert (ret == th->msize); /* NOTE: API change! */
 716   delay = GNUNET_TIME_absolute_get_duration (th->request_time);
 717   overdue = GNUNET_TIME_absolute_get_duration (th->deadline);
 718   if (overdue.rel_value_us > GNUNET_CONSTANTS_LATENCY_WARN.rel_value_us)
 719     LOG (GNUNET_ERROR_TYPE_WARNING,
 720          "Transmitting overdue %u bytes to `%s' at priority %u with %s delay %s\n",
 721          ret,
 722          GNUNET_i2s (&pr->peer),
 723          (unsigned int) th->priority,
 724          GNUNET_STRINGS_relative_time_to_string (delay,
 725                                                  GNUNET_YES),
 726          (th->cork) ? " (corked)" : "");
 727   else
 728     LOG (GNUNET_ERROR_TYPE_DEBUG,
 729          "Transmitting %u bytes to `%s' at priority %u with %s delay %s\n",
 730          ret,
 731          GNUNET_i2s (&pr->peer),
 732          (unsigned int) th->priority,
 733          GNUNET_STRINGS_relative_time_to_string (delay,
 734                                                  GNUNET_YES),
 735          (th->cork) ? " (corked)" : "");
 736   GNUNET_MQ_send (h->mq,
 737                   env);
 738 }
 739
 740
 741 /**
 742  * Our current client connection went down.  Clean it up and try to
 743  * reconnect!
 744  *
 745  * @param h our handle to the core service
 746  */
 747 static void
 748 reconnect (struct GNUNET_CORE_Handle *h)
 749 {
 750   GNUNET_MQ_hd_fixed_size (init_reply,
 751                            GNUNET_MESSAGE_TYPE_CORE_INIT_REPLY,
 752                            struct InitReplyMessage);
 753   GNUNET_MQ_hd_fixed_size (connect_notify,
 754                            GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT,
 755                            struct ConnectNotifyMessage);
 756   GNUNET_MQ_hd_fixed_size (disconnect_notify,
 757                            GNUNET_MESSAGE_TYPE_CORE_NOTIFY_DISCONNECT,
 758                            struct DisconnectNotifyMessage);
 759   GNUNET_MQ_hd_var_size (notify_inbound,
 760                          GNUNET_MESSAGE_TYPE_CORE_NOTIFY_INBOUND,
 761                          struct NotifyTrafficMessage);
 762   GNUNET_MQ_hd_var_size (notify_outbound,
 763                          GNUNET_MESSAGE_TYPE_CORE_NOTIFY_OUTBOUND,
 764                          struct NotifyTrafficMessage);
 765   GNUNET_MQ_hd_fixed_size (send_ready,
 766                            GNUNET_MESSAGE_TYPE_CORE_SEND_READY,
 767                            struct SendMessageReady);
 768  struct GNUNET_MQ_MessageHandler handlers[] = {
 769     make_init_reply_handler (h),
 770     make_connect_notify_handler (h),
 771     make_disconnect_notify_handler (h),
 772     make_notify_inbound_handler (h),
 773     make_notify_outbound_handler (h),
 774     make_send_ready_handler (h),
 775     GNUNET_MQ_handler_end ()
 776   };
 777   struct InitMessage *init;
 778   struct GNUNET_MQ_Envelope *env;
 779   uint32_t opt;
 780   uint16_t *ts;
 781
 782   GNUNET_assert (NULL == h->mq);
 783   GNUNET_assert (GNUNET_YES == h->currently_down);
 784   h->mq = GNUNET_CLIENT_connecT (h->cfg,
 785                                  "core",
 786                                  handlers,
 787                                  &handle_mq_error,
 788                                  h);
 789   if (NULL == h->mq)
 790   {
 791     reconnect_later (h);
 792     return;
 793   }
 794   env = GNUNET_MQ_msg_extra (init,
 795                              sizeof (uint16_t) * h->hcnt,
 796                              GNUNET_MESSAGE_TYPE_CORE_INIT);
 797   opt = 0;
 798   if (NULL != h->inbound_notify)
 799   {
 800     if (h->inbound_hdr_only)
 801       opt |= GNUNET_CORE_OPTION_SEND_HDR_INBOUND;
 802     else
 803       opt |= GNUNET_CORE_OPTION_SEND_FULL_INBOUND;
 804   }
 805   if (NULL != h->outbound_notify)
 806   {
 807     if (h->outbound_hdr_only)
 808       opt |= GNUNET_CORE_OPTION_SEND_HDR_OUTBOUND;
 809     else
 810       opt |= GNUNET_CORE_OPTION_SEND_FULL_OUTBOUND;
 811   }
 812   LOG (GNUNET_ERROR_TYPE_INFO,
 813        "(Re)connecting to CORE service, monitoring messages of type %u\n",
 814        opt);
 815   init->options = htonl (opt);
 816   ts = (uint16_t *) &init[1];
 817   for (unsigned int hpos = 0; hpos < h->hcnt; hpos++)
 818     ts[hpos] = htons (h->handlers[hpos].type);
 819   GNUNET_MQ_send (h->mq,
 820                   env);
 821 }
 822
 823
 824 /**
 825  * Connect to the core service.  Note that the connection may complete
 826  * (or fail) asynchronously.
 827  *
 828  * @param cfg configuration to use
 829  * @param cls closure for the various callbacks that follow (including handlers in the handlers array)
 830  * @param init callback to call once we have successfully
 831  *        connected to the core service
 832  * @param connects function to call on peer connect, can be NULL
 833  * @param disconnects function to call on peer disconnect / timeout, can be NULL
 834  * @param inbound_notify function to call for all inbound messages, can be NULL
 835  * @param inbound_hdr_only set to #GNUNET_YES if inbound_notify will only read the
 836  *                GNUNET_MessageHeader and hence we do not need to give it the full message;
 837  *                can be used to improve efficiency, ignored if @a inbound_notify is NULLL
 838  * @param outbound_notify function to call for all outbound messages, can be NULL
 839  * @param outbound_hdr_only set to #GNUNET_YES if outbound_notify will only read the
 840  *                GNUNET_MessageHeader and hence we do not need to give it the full message
 841  *                can be used to improve efficiency, ignored if @a outbound_notify is NULLL
 842  * @param handlers callbacks for messages we care about, NULL-terminated
 843  * @return handle to the core service (only useful for disconnect until 'init' is called);
 844  *                NULL on error (in this case, init is never called)
 845  */
 846 struct GNUNET_CORE_Handle *
 847 GNUNET_CORE_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
 848                      void *cls,
 849                      GNUNET_CORE_StartupCallback init,
 850                      GNUNET_CORE_ConnectEventHandler connects,
 851                      GNUNET_CORE_DisconnectEventHandler disconnects,
 852                      GNUNET_CORE_MessageCallback inbound_notify,
 853                      int inbound_hdr_only,
 854                      GNUNET_CORE_MessageCallback outbound_notify,
 855                      int outbound_hdr_only,
 856                      const struct GNUNET_CORE_MessageHandler *handlers)
 857 {
 858   struct GNUNET_CORE_Handle *h;
 859   unsigned int hcnt;
 860
 861   h = GNUNET_new (struct GNUNET_CORE_Handle);
 862   h->cfg = cfg;
 863   h->cls = cls;
 864   h->init = init;
 865   h->connects = connects;
 866   h->disconnects = disconnects;
 867   h->inbound_notify = inbound_notify;
 868   h->outbound_notify = outbound_notify;
 869   h->inbound_hdr_only = inbound_hdr_only;
 870   h->outbound_hdr_only = outbound_hdr_only;
 871   h->currently_down = GNUNET_YES;
 872   h->peers = GNUNET_CONTAINER_multipeermap_create (128, GNUNET_NO);
 873   hcnt = 0;
 874   if (NULL != handlers)
 875     while (NULL != handlers[hcnt].callback)
 876       hcnt++;
 877   h->handlers = GNUNET_new_array (hcnt + 1,
 878                                   struct GNUNET_CORE_MessageHandler);
 879   if (NULL != handlers)
 880     memcpy (h->handlers,
 881             handlers,
 882             hcnt * sizeof (struct GNUNET_CORE_MessageHandler));
 883   h->hcnt = hcnt;
 884   GNUNET_assert (hcnt <
 885                  (GNUNET_SERVER_MAX_MESSAGE_SIZE -
 886                   sizeof (struct InitMessage)) / sizeof (uint16_t));
 887   LOG (GNUNET_ERROR_TYPE_DEBUG,
 888        "Connecting to CORE service\n");
 889   reconnect (h);
 890   return h;
 891 }
 892
 893
 894 /**
 895  * Disconnect from the core service.  This function can only
 896  * be called *after* all pending #GNUNET_CORE_notify_transmit_ready()
 897  * requests have been explicitly canceled.
 898  *
 899  * @param handle connection to core to disconnect
 900  */
 901 void
 902 GNUNET_CORE_disconnect (struct GNUNET_CORE_Handle *handle)
 903 {
 904   LOG (GNUNET_ERROR_TYPE_DEBUG,
 905        "Disconnecting from CORE service\n");
 906   GNUNET_CONTAINER_multipeermap_iterate (handle->peers,
 907                                          &disconnect_and_free_peer_entry,
 908                                          handle);
 909   GNUNET_CONTAINER_multipeermap_destroy (handle->peers);
 910   handle->peers = NULL;
 911   if (NULL != handle->reconnect_task)
 912   {
 913     GNUNET_SCHEDULER_cancel (handle->reconnect_task);
 914     handle->reconnect_task = NULL;
 915   }
 916   if (NULL != handle->mq)
 917   {
 918     GNUNET_MQ_destroy (handle->mq);
 919     handle->mq = NULL;
 920   }
 921   GNUNET_free (handle->handlers);
 922   GNUNET_free (handle);
 923 }
 924
 925
 926 /**
 927  * Ask the core to call @a notify once it is ready to transmit the
 928  * given number of bytes to the specified @a target.  Must only be
 929  * called after a connection to the respective peer has been
 930  * established (and the client has been informed about this).  You may
 931  * have one request of this type pending for each connected peer at
 932  * any time.  If a peer disconnects, the application MUST call
 933  * #GNUNET_CORE_notify_transmit_ready_cancel on the respective
 934  * transmission request, if one such request is pending.
 935  *
 936  * @param handle connection to core service
 937  * @param cork is corking allowed for this transmission?
 938  * @param priority how important is the message?
 939  * @param maxdelay how long can the message wait? Only effective if @a cork is #GNUNET_YES
 940  * @param target who should receive the message, never NULL (can be this peer's identity for loopback)
 941  * @param notify_size how many bytes of buffer space does @a notify want?
 942  * @param notify function to call when buffer space is available;
 943  *        will be called with NULL on timeout; clients MUST cancel
 944  *        all pending transmission requests DURING the disconnect
 945  *        handler
 946  * @param notify_cls closure for @a notify
 947  * @return non-NULL if the notify callback was queued,
 948  *         NULL if we can not even queue the request (request already pending);
 949  *         if NULL is returned, @a notify will NOT be called.
 950  */
 951 struct GNUNET_CORE_TransmitHandle *
 952 GNUNET_CORE_notify_transmit_ready (struct GNUNET_CORE_Handle *handle,
 953                                    int cork,
 954                                    enum GNUNET_CORE_Priority priority,
 955                                    struct GNUNET_TIME_Relative maxdelay,
 956                                    const struct GNUNET_PeerIdentity *target,
 957                                    size_t notify_size,
 958                                    GNUNET_CONNECTION_TransmitReadyNotify notify,
 959                                    void *notify_cls)
 960 {
 961   struct PeerRecord *pr;
 962   struct GNUNET_CORE_TransmitHandle *th;
 963   struct SendMessageRequest *smr;
 964   struct GNUNET_MQ_Envelope *env;
 965
 966   GNUNET_assert (NULL != notify);
 967   if ( (notify_size > GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE) ||
 968        (notify_size + sizeof (struct SendMessage) >= GNUNET_SERVER_MAX_MESSAGE_SIZE) )
 969   {
 970     GNUNET_break (0);
 971     return NULL;
 972   }
 973   LOG (GNUNET_ERROR_TYPE_DEBUG,
 974        "Asking core for transmission of %u bytes to `%s'\n",
 975        (unsigned int) notify_size,
 976        GNUNET_i2s (target));
 977   pr = GNUNET_CONTAINER_multipeermap_get (handle->peers,
 978                                           target);
 979   if (NULL == pr)
 980   {
 981     /* attempt to send to peer that is not connected */
 982     GNUNET_break (0);
 983     return NULL;
 984   }
 985   if (NULL != pr->th.peer)
 986   {
 987     /* attempting to queue a second request for the same destination */
 988     GNUNET_break (0);
 989     return NULL;
 990   }
 991   th = &pr->th;
 992   memset (th,
 993           0,
 994           sizeof (struct GNUNET_CORE_TransmitHandle));
 995   th->peer = pr;
 996   th->get_message = notify;
 997   th->get_message_cls = notify_cls;
 998   th->request_time = GNUNET_TIME_absolute_get ();
 999   if (GNUNET_YES == cork)
1000     th->deadline = GNUNET_TIME_relative_to_absolute (maxdelay);
1001   else
1002     th->deadline = th->request_time;
1003   th->priority = priority;
1004   th->msize = notify_size;
1005   th->cork = cork;
1006   env = GNUNET_MQ_msg (smr,
1007                        GNUNET_MESSAGE_TYPE_CORE_SEND_REQUEST);
1008   smr->priority = htonl ((uint32_t) th->priority);
1009   smr->deadline = GNUNET_TIME_absolute_hton (th->deadline);
1010   smr->peer = pr->peer;
1011   smr->reserved = htonl (0);
1012   smr->size = htons (th->msize);
1013   smr->smr_id = htons (th->smr_id = pr->smr_id_gen++);
1014   GNUNET_MQ_send (handle->mq,
1015                   env);
1016   LOG (GNUNET_ERROR_TYPE_DEBUG,
1017        "Transmission request added to queue\n");
1018   return th;
1019 }
1020
1021
1022 /**
1023  * Cancel the specified transmission-ready notification.
1024  *
1025  * @param th handle that was returned by #GNUNET_CORE_notify_transmit_ready().
1026  */
1027 void
1028 GNUNET_CORE_notify_transmit_ready_cancel (struct GNUNET_CORE_TransmitHandle *th)
1029 {
1030   struct PeerRecord *pr = th->peer;
1031
1032   LOG (GNUNET_ERROR_TYPE_DEBUG,
1033        "Aborting transmission request to core for %u bytes to `%s'\n",
1034        (unsigned int) th->msize,
1035        GNUNET_i2s (&pr->peer));
1036   th->peer = NULL;
1037 }
1038
1039
1040 /**
1041  * Check if the given peer is currently connected. This function is for special
1042  * cirumstances (GNUNET_TESTBED uses it), normal users of the CORE API are
1043  * expected to track which peers are connected based on the connect/disconnect
1044  * callbacks from #GNUNET_CORE_connect().  This function is NOT part of the
1045  * 'versioned', 'official' API. The difference between this function and the
1046  * function GNUNET_CORE_is_peer_connected() is that this one returns
1047  * synchronously after looking in the CORE API cache. The function
1048  * GNUNET_CORE_is_peer_connected() sends a message to the CORE service and hence
1049  * its response is given asynchronously.
1050  *
1051  * @param h the core handle
1052  * @param pid the identity of the peer to check if it has been connected to us
1053  * @return #GNUNET_YES if the peer is connected to us; #GNUNET_NO if not
1054  */
1055 int
1056 GNUNET_CORE_is_peer_connected_sync (const struct GNUNET_CORE_Handle *h,
1057                                     const struct GNUNET_PeerIdentity *pid)
1058 {
1059   return GNUNET_CONTAINER_multipeermap_contains (h->peers,
1060                                                  pid);
1061 }
1062
1063
1064 /* end of core_api.c */