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   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 336               "MQ ERROR: %d\n",
 337               error);
 338   reconnect_later (h);
 339 }
 340
 341
 342 /**
 343  * Handle  init  reply message  received  from  CORE service.   Notify
 344  * application  that we  are now  connected  to the  CORE.  Also  fake
 345  * loopback connection.
 346  *
 347  * @param cls the `struct GNUNET_CORE_Handle`
 348  * @param m the init reply
 349  */
 350 static void
 351 handle_init_reply (void *cls,
 352                    const struct InitReplyMessage *m)
 353 {
 354   struct GNUNET_CORE_Handle *h = cls;
 355   GNUNET_CORE_StartupCallback init;
 356   struct PeerRecord *pr;
 357
 358   GNUNET_break (0 == ntohl (m->reserved));
 359   GNUNET_break (GNUNET_YES == h->currently_down);
 360   h->currently_down = GNUNET_NO;
 361   h->retry_backoff = GNUNET_TIME_UNIT_MILLISECONDS;
 362   if (NULL != (init = h->init))
 363   {
 364     /* mark so we don't call init on reconnect */
 365     h->init = NULL;
 366     h->me = m->my_identity;
 367     LOG (GNUNET_ERROR_TYPE_DEBUG,
 368          "Connected to core service of peer `%s'.\n",
 369          GNUNET_i2s (&h->me));
 370     h->have_init = GNUNET_YES;
 371     init (h->cls,
 372           &h->me);
 373   }
 374   else
 375   {
 376     LOG (GNUNET_ERROR_TYPE_DEBUG,
 377          "Successfully reconnected to core service.\n");
 378     if (GNUNET_NO == h->have_init)
 379     {
 380       h->me = m->my_identity;
 381       h->have_init = GNUNET_YES;
 382     }
 383     else
 384     {
 385       GNUNET_break (0 == memcmp (&h->me,
 386                                  &m->my_identity,
 387                                  sizeof (struct GNUNET_PeerIdentity)));
 388     }
 389   }
 390   /* fake 'connect to self' */
 391   pr = GNUNET_new (struct PeerRecord);
 392   pr->peer = h->me;
 393   pr->ch = h;
 394   GNUNET_assert (GNUNET_YES ==
 395                  GNUNET_CONTAINER_multipeermap_put (h->peers,
 396                                                     &h->me,
 397                                                     pr,
 398                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 399   if (NULL != h->connects)
 400     h->connects (h->cls,
 401                  &pr->peer);
 402 }
 403
 404
 405 /**
 406  * Handle connect message received from CORE service.
 407  * Notify the application about the new connection.
 408  *
 409  * @param cls the `struct GNUNET_CORE_Handle`
 410  * @param cnm the connect message
 411  */
 412 static void
 413 handle_connect_notify (void *cls,
 414                        const struct ConnectNotifyMessage *cnm)
 415 {
 416   struct GNUNET_CORE_Handle *h = cls;
 417   struct PeerRecord *pr;
 418
 419   GNUNET_break (GNUNET_NO == h->currently_down);
 420   LOG (GNUNET_ERROR_TYPE_DEBUG,
 421        "Received notification about connection from `%s'.\n",
 422        GNUNET_i2s (&cnm->peer));
 423   if (0 == memcmp (&h->me,
 424                    &cnm->peer,
 425                    sizeof (struct GNUNET_PeerIdentity)))
 426   {
 427     /* connect to self!? */
 428     GNUNET_break (0);
 429     return;
 430   }
 431   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 432                                           &cnm->peer);
 433   if (NULL != pr)
 434   {
 435     GNUNET_break (0);
 436     reconnect_later (h);
 437     return;
 438   }
 439   pr = GNUNET_new (struct PeerRecord);
 440   pr->peer = cnm->peer;
 441   pr->ch = h;
 442   GNUNET_assert (GNUNET_YES ==
 443                  GNUNET_CONTAINER_multipeermap_put (h->peers,
 444                                                     &cnm->peer,
 445                                                     pr,
 446                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 447   if (NULL != h->connects)
 448     h->connects (h->cls,
 449                  &pr->peer);
 450 }
 451
 452
 453 /**
 454  * Handle disconnect message received from CORE service.
 455  * Notify the application about the lost connection.
 456  *
 457  * @param cls the `struct GNUNET_CORE_Handle`
 458  * @param dnm message about the disconnect event
 459  */
 460 static void
 461 handle_disconnect_notify (void *cls,
 462                           const struct DisconnectNotifyMessage * dnm)
 463 {
 464   struct GNUNET_CORE_Handle *h = cls;
 465   struct PeerRecord *pr;
 466
 467   GNUNET_break (GNUNET_NO == h->currently_down);
 468   if (0 == memcmp (&h->me,
 469                    &dnm->peer,
 470                    sizeof (struct GNUNET_PeerIdentity)))
 471   {
 472     /* connection to self!? */
 473     GNUNET_break (0);
 474     return;
 475   }
 476   GNUNET_break (0 == ntohl (dnm->reserved));
 477   LOG (GNUNET_ERROR_TYPE_DEBUG,
 478        "Received notification about disconnect from `%s'.\n",
 479        GNUNET_i2s (&dnm->peer));
 480   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 481                                           &dnm->peer);
 482   if (NULL == pr)
 483   {
 484     GNUNET_break (0);
 485     reconnect_later (h);
 486     return;
 487   }
 488   disconnect_and_free_peer_entry (h,
 489                                   &dnm->peer,
 490                                   pr);
 491 }
 492
 493
 494 /**
 495  * Check that message received from CORE service is well-formed.
 496  *
 497  * @param cls the `struct GNUNET_CORE_Handle`
 498  * @param ntm the message we got
 499  * @return #GNUNET_OK if the message is well-formed
 500  */
 501 static int
 502 check_notify_inbound (void *cls,
 503                       const struct NotifyTrafficMessage *ntm)
 504 {
 505   struct GNUNET_CORE_Handle *h = cls;
 506   uint16_t msize;
 507   const struct GNUNET_MessageHeader *em;
 508
 509   GNUNET_break (GNUNET_NO == h->currently_down);
 510   msize = ntohs (ntm->header.size) - sizeof (struct NotifyTrafficMessage);
 511   if (msize < sizeof (struct GNUNET_MessageHeader))
 512   {
 513     GNUNET_break (0);
 514     return GNUNET_SYSERR;
 515   }
 516   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 517   if ( (GNUNET_NO == h->inbound_hdr_only) &&
 518        (msize != ntohs (em->size)) )
 519   {
 520     GNUNET_break (0);
 521     return GNUNET_SYSERR;
 522   }
 523   return GNUNET_OK;
 524 }
 525
 526
 527 /**
 528  * Handle inbound message received from CORE service.  If applicable,
 529  * notify the application.
 530  *
 531  * @param cls the `struct GNUNET_CORE_Handle`
 532  * @param ntm the message we got from CORE.
 533  */
 534 static void
 535 handle_notify_inbound (void *cls,
 536                        const struct NotifyTrafficMessage *ntm)
 537 {
 538   struct GNUNET_CORE_Handle *h = cls;
 539   const struct GNUNET_MessageHeader *em;
 540   struct PeerRecord *pr;
 541   uint16_t et;
 542
 543   GNUNET_break (GNUNET_NO == h->currently_down);
 544   LOG (GNUNET_ERROR_TYPE_DEBUG,
 545        "Received inbound message from `%s'.\n",
 546        GNUNET_i2s (&ntm->peer));
 547   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 548   et = ntohs (em->type);
 549   for (unsigned int hpos = 0; NULL != h->handlers[hpos].callback; hpos++)
 550   {
 551     const struct GNUNET_CORE_MessageHandler *mh;
 552
 553     mh = &h->handlers[hpos];
 554     if (mh->type != et)
 555       continue;
 556     if ( (mh->expected_size != ntohs (em->size)) &&
 557          (0 != mh->expected_size) )
 558     {
 559       LOG (GNUNET_ERROR_TYPE_ERROR,
 560            "Unexpected message size %u for message of type %u from peer `%s'\n",
 561            htons (em->size),
 562            mh->type,
 563            GNUNET_i2s (&ntm->peer));
 564       GNUNET_break_op (0);
 565       continue;
 566     }
 567     pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 568                                             &ntm->peer);
 569     if (NULL == pr)
 570     {
 571       GNUNET_break (0);
 572       reconnect_later (h);
 573       return;
 574     }
 575     if (GNUNET_OK !=
 576         h->handlers[hpos].callback (h->cls,
 577                                     &ntm->peer,
 578                                     em))
 579     {
 580       /* error in processing, do not process other messages! */
 581       break;
 582     }
 583   }
 584   if (NULL != h->inbound_notify)
 585     h->inbound_notify (h->cls,
 586                        &ntm->peer,
 587                        em);
 588 }
 589
 590
 591 /**
 592  * Check that message received from CORE service is well-formed.
 593  *
 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
 597  */
 598 static int
 599 check_notify_outbound (void *cls,
 600                        const struct NotifyTrafficMessage *ntm)
 601 {
 602   struct GNUNET_CORE_Handle *h = cls;
 603   uint16_t msize;
 604   const struct GNUNET_MessageHeader *em;
 605
 606   GNUNET_break (GNUNET_NO == h->currently_down);
 607   LOG (GNUNET_ERROR_TYPE_DEBUG,
 608        "Received outbound message from `%s'.\n",
 609        GNUNET_i2s (&ntm->peer));
 610   msize = ntohs (ntm->header.size) - sizeof (struct NotifyTrafficMessage);
 611   if (msize < sizeof (struct GNUNET_MessageHeader))
 612   {
 613     GNUNET_break (0);
 614     return GNUNET_SYSERR;
 615   }
 616   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 617   if ( (GNUNET_NO == h->outbound_hdr_only) &&
 618        (msize != ntohs (em->size)) )
 619   {
 620     GNUNET_break (0);
 621     return GNUNET_SYSERR;
 622   }
 623   return GNUNET_OK;
 624 }
 625
 626
 627 /**
 628  * Handle outbound message received from CORE service.  If applicable,
 629  * notify the application.
 630  *
 631  * @param cls the `struct GNUNET_CORE_Handle`
 632  * @param ntm the message we got
 633  */
 634 static void
 635 handle_notify_outbound (void *cls,
 636                         const struct NotifyTrafficMessage *ntm)
 637 {
 638   struct GNUNET_CORE_Handle *h = cls;
 639   const struct GNUNET_MessageHeader *em;
 640
 641   GNUNET_break (GNUNET_NO == h->currently_down);
 642   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 643   LOG (GNUNET_ERROR_TYPE_DEBUG,
 644        "Received notification about transmission to `%s'.\n",
 645        GNUNET_i2s (&ntm->peer));
 646   if (NULL == h->outbound_notify)
 647   {
 648     GNUNET_break (0);
 649     return;
 650   }
 651   h->outbound_notify (h->cls,
 652                       &ntm->peer,
 653                       em);
 654 }
 655
 656
 657 /**
 658  * Handle message received from CORE service notifying us that we are
 659  * now allowed to send a message to a peer.  If that message is still
 660  * pending, put it into the queue to be transmitted.
 661  *
 662  * @param cls the `struct GNUNET_CORE_Handle`
 663  * @param ntm the message we got
 664  */
 665 static void
 666 handle_send_ready (void *cls,
 667                    const struct SendMessageReady *smr)
 668 {
 669   struct GNUNET_CORE_Handle *h = cls;
 670   struct PeerRecord *pr;
 671   struct GNUNET_CORE_TransmitHandle *th;
 672   struct SendMessage *sm;
 673   struct GNUNET_MQ_Envelope *env;
 674   struct GNUNET_TIME_Relative delay;
 675   struct GNUNET_TIME_Relative overdue;
 676   unsigned int ret;
 677
 678   GNUNET_break (GNUNET_NO == h->currently_down);
 679   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 680                                           &smr->peer);
 681   if (NULL == pr)
 682   {
 683     GNUNET_break (0);
 684     reconnect_later (h);
 685     return;
 686   }
 687   LOG (GNUNET_ERROR_TYPE_DEBUG,
 688        "Received notification about transmission readiness to `%s'.\n",
 689        GNUNET_i2s (&smr->peer));
 690   if (NULL == pr->th.peer)
 691   {
 692     /* request must have been cancelled between the original request
 693      * and the response from CORE, ignore CORE's readiness */
 694     return;
 695   }
 696   th = &pr->th;
 697   if (ntohs (smr->smr_id) != th->smr_id)
 698   {
 699     /* READY message is for expired or cancelled message,
 700      * ignore! (we should have already sent another request) */
 701     return;
 702   }
 703   /* ok, all good, send message out! */
 704   th->peer = NULL;
 705   env = GNUNET_MQ_msg_extra (sm,
 706                              th->msize,
 707                              GNUNET_MESSAGE_TYPE_CORE_SEND);
 708   sm->priority = htonl ((uint32_t) th->priority);
 709   sm->deadline = GNUNET_TIME_absolute_hton (th->deadline);
 710   sm->peer = pr->peer;
 711   sm->cork = htonl ((uint32_t) th->cork);
 712   sm->reserved = htonl (0);
 713   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 714               "Calling get_message with buffer of %u bytes\n",
 715               (unsigned int) th->msize);
 716   /* FIXME: this is ugly and a bit brutal, but "get_message"
 717      may call GNUNET_CORE_notify_transmit_ready() which
 718      may call GNUNET_MQ_send() as well, and we MUST get this
 719      message out before the next SEND_REQUEST.  So we queue
 720      it (even though incomplete) and then---relying on MQ being
 721      nice and not actually touching 'env' until much later---
 722      fill it afterwards.  This is horrible style, and once
 723      the core_api abandons GNUNET_CORE_notify_transmit_ready
 724      in favor of an MQ-style API, this hack should no longer
 725      be required */
 726   GNUNET_MQ_send (h->mq,
 727                   env);
 728   ret = th->get_message (th->get_message_cls,
 729                          th->msize,
 730                          &sm[1]);
 731   sm->header.size = htons (ret + sizeof (struct SendMessage));
 732   delay = GNUNET_TIME_absolute_get_duration (th->request_time);
 733   overdue = GNUNET_TIME_absolute_get_duration (th->deadline);
 734   if (overdue.rel_value_us > GNUNET_CONSTANTS_LATENCY_WARN.rel_value_us)
 735     LOG (GNUNET_ERROR_TYPE_WARNING,
 736          "Transmitting overdue %u bytes to `%s' at priority %u with %s delay %s\n",
 737          ret,
 738          GNUNET_i2s (&pr->peer),
 739          (unsigned int) th->priority,
 740          GNUNET_STRINGS_relative_time_to_string (delay,
 741                                                  GNUNET_YES),
 742          (th->cork) ? " (corked)" : "");
 743   else
 744     LOG (GNUNET_ERROR_TYPE_DEBUG,
 745          "Transmitting %u bytes to `%s' at priority %u with %s delay %s\n",
 746          ret,
 747          GNUNET_i2s (&pr->peer),
 748          (unsigned int) th->priority,
 749          GNUNET_STRINGS_relative_time_to_string (delay,
 750                                                  GNUNET_YES),
 751          (th->cork) ? " (corked)" : "");
 752 }
 753
 754
 755 /**
 756  * Our current client connection went down.  Clean it up and try to
 757  * reconnect!
 758  *
 759  * @param h our handle to the core service
 760  */
 761 static void
 762 reconnect (struct GNUNET_CORE_Handle *h)
 763 {
 764   GNUNET_MQ_hd_fixed_size (init_reply,
 765                            GNUNET_MESSAGE_TYPE_CORE_INIT_REPLY,
 766                            struct InitReplyMessage);
 767   GNUNET_MQ_hd_fixed_size (connect_notify,
 768                            GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT,
 769                            struct ConnectNotifyMessage);
 770   GNUNET_MQ_hd_fixed_size (disconnect_notify,
 771                            GNUNET_MESSAGE_TYPE_CORE_NOTIFY_DISCONNECT,
 772                            struct DisconnectNotifyMessage);
 773   GNUNET_MQ_hd_var_size (notify_inbound,
 774                          GNUNET_MESSAGE_TYPE_CORE_NOTIFY_INBOUND,
 775                          struct NotifyTrafficMessage);
 776   GNUNET_MQ_hd_var_size (notify_outbound,
 777                          GNUNET_MESSAGE_TYPE_CORE_NOTIFY_OUTBOUND,
 778                          struct NotifyTrafficMessage);
 779   GNUNET_MQ_hd_fixed_size (send_ready,
 780                            GNUNET_MESSAGE_TYPE_CORE_SEND_READY,
 781                            struct SendMessageReady);
 782  struct GNUNET_MQ_MessageHandler handlers[] = {
 783     make_init_reply_handler (h),
 784     make_connect_notify_handler (h),
 785     make_disconnect_notify_handler (h),
 786     make_notify_inbound_handler (h),
 787     make_notify_outbound_handler (h),
 788     make_send_ready_handler (h),
 789     GNUNET_MQ_handler_end ()
 790   };
 791   struct InitMessage *init;
 792   struct GNUNET_MQ_Envelope *env;
 793   uint32_t opt;
 794   uint16_t *ts;
 795
 796   GNUNET_assert (NULL == h->mq);
 797   GNUNET_assert (GNUNET_YES == h->currently_down);
 798   h->mq = GNUNET_CLIENT_connecT (h->cfg,
 799                                  "core",
 800                                  handlers,
 801                                  &handle_mq_error,
 802                                  h);
 803   if (NULL == h->mq)
 804   {
 805     reconnect_later (h);
 806     return;
 807   }
 808   env = GNUNET_MQ_msg_extra (init,
 809                              sizeof (uint16_t) * h->hcnt,
 810                              GNUNET_MESSAGE_TYPE_CORE_INIT);
 811   opt = 0;
 812   if (NULL != h->inbound_notify)
 813   {
 814     if (h->inbound_hdr_only)
 815       opt |= GNUNET_CORE_OPTION_SEND_HDR_INBOUND;
 816     else
 817       opt |= GNUNET_CORE_OPTION_SEND_FULL_INBOUND;
 818   }
 819   if (NULL != h->outbound_notify)
 820   {
 821     if (h->outbound_hdr_only)
 822       opt |= GNUNET_CORE_OPTION_SEND_HDR_OUTBOUND;
 823     else
 824       opt |= GNUNET_CORE_OPTION_SEND_FULL_OUTBOUND;
 825   }
 826   LOG (GNUNET_ERROR_TYPE_INFO,
 827        "(Re)connecting to CORE service, monitoring messages of type %u\n",
 828        opt);
 829   init->options = htonl (opt);
 830   ts = (uint16_t *) &init[1];
 831   for (unsigned int hpos = 0; hpos < h->hcnt; hpos++)
 832     ts[hpos] = htons (h->handlers[hpos].type);
 833   GNUNET_MQ_send (h->mq,
 834                   env);
 835 }
 836
 837
 838 /**
 839  * Connect to the core service.  Note that the connection may complete
 840  * (or fail) asynchronously.
 841  *
 842  * @param cfg configuration to use
 843  * @param cls closure for the various callbacks that follow (including handlers in the handlers array)
 844  * @param init callback to call once we have successfully
 845  *        connected to the core service
 846  * @param connects function to call on peer connect, can be NULL
 847  * @param disconnects function to call on peer disconnect / timeout, can be NULL
 848  * @param inbound_notify function to call for all inbound messages, can be NULL
 849  * @param inbound_hdr_only set to #GNUNET_YES if inbound_notify will only read the
 850  *                GNUNET_MessageHeader and hence we do not need to give it the full message;
 851  *                can be used to improve efficiency, ignored if @a inbound_notify is NULL
 852  * @param outbound_notify function to call for all outbound messages, can be NULL
 853  * @param outbound_hdr_only set to #GNUNET_YES if outbound_notify will only read the
 854  *                GNUNET_MessageHeader and hence we do not need to give it the full message
 855  *                can be used to improve efficiency, ignored if @a outbound_notify is NULL
 856  * @param handlers callbacks for messages we care about, NULL-terminated
 857  * @return handle to the core service (only useful for disconnect until @a init is called);
 858  *                NULL on error (in this case, init is never called)
 859  */
 860 struct GNUNET_CORE_Handle *
 861 GNUNET_CORE_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
 862                      void *cls,
 863                      GNUNET_CORE_StartupCallback init,
 864                      GNUNET_CORE_ConnectEventHandler connects,
 865                      GNUNET_CORE_DisconnectEventHandler disconnects,
 866                      GNUNET_CORE_MessageCallback inbound_notify,
 867                      int inbound_hdr_only,
 868                      GNUNET_CORE_MessageCallback outbound_notify,
 869                      int outbound_hdr_only,
 870                      const struct GNUNET_CORE_MessageHandler *handlers)
 871 {
 872   struct GNUNET_CORE_Handle *h;
 873   unsigned int hcnt;
 874
 875   h = GNUNET_new (struct GNUNET_CORE_Handle);
 876   h->cfg = cfg;
 877   h->cls = cls;
 878   h->init = init;
 879   h->connects = connects;
 880   h->disconnects = disconnects;
 881   h->inbound_notify = inbound_notify;
 882   h->outbound_notify = outbound_notify;
 883   h->inbound_hdr_only = inbound_hdr_only;
 884   h->outbound_hdr_only = outbound_hdr_only;
 885   h->currently_down = GNUNET_YES;
 886   h->peers = GNUNET_CONTAINER_multipeermap_create (128, GNUNET_NO);
 887   hcnt = 0;
 888   if (NULL != handlers)
 889     while (NULL != handlers[hcnt].callback)
 890       hcnt++;
 891   h->handlers = GNUNET_new_array (hcnt + 1,
 892                                   struct GNUNET_CORE_MessageHandler);
 893   if (NULL != handlers)
 894     memcpy (h->handlers,
 895             handlers,
 896             hcnt * sizeof (struct GNUNET_CORE_MessageHandler));
 897   h->hcnt = hcnt;
 898   GNUNET_assert (hcnt <
 899                  (GNUNET_SERVER_MAX_MESSAGE_SIZE -
 900                   sizeof (struct InitMessage)) / sizeof (uint16_t));
 901   LOG (GNUNET_ERROR_TYPE_DEBUG,
 902        "Connecting to CORE service\n");
 903   reconnect (h);
 904   if (NULL == h->mq)
 905   {
 906     GNUNET_CORE_disconnect (h);
 907     return NULL;
 908   }
 909   return h;
 910 }
 911
 912
 913 /**
 914  * Disconnect from the core service.  This function can only
 915  * be called *after* all pending #GNUNET_CORE_notify_transmit_ready()
 916  * requests have been explicitly canceled.
 917  *
 918  * @param handle connection to core to disconnect
 919  */
 920 void
 921 GNUNET_CORE_disconnect (struct GNUNET_CORE_Handle *handle)
 922 {
 923   LOG (GNUNET_ERROR_TYPE_DEBUG,
 924        "Disconnecting from CORE service\n");
 925   GNUNET_CONTAINER_multipeermap_iterate (handle->peers,
 926                                          &disconnect_and_free_peer_entry,
 927                                          handle);
 928   GNUNET_CONTAINER_multipeermap_destroy (handle->peers);
 929   handle->peers = NULL;
 930   if (NULL != handle->reconnect_task)
 931   {
 932     GNUNET_SCHEDULER_cancel (handle->reconnect_task);
 933     handle->reconnect_task = NULL;
 934   }
 935   if (NULL != handle->mq)
 936   {
 937     GNUNET_MQ_destroy (handle->mq);
 938     handle->mq = NULL;
 939   }
 940   GNUNET_free (handle->handlers);
 941   GNUNET_free (handle);
 942 }
 943
 944
 945 /**
 946  * Ask the core to call @a notify once it is ready to transmit the
 947  * given number of bytes to the specified @a target.  Must only be
 948  * called after a connection to the respective peer has been
 949  * established (and the client has been informed about this).  You may
 950  * have one request of this type pending for each connected peer at
 951  * any time.  If a peer disconnects, the application MUST call
 952  * #GNUNET_CORE_notify_transmit_ready_cancel on the respective
 953  * transmission request, if one such request is pending.
 954  *
 955  * @param handle connection to core service
 956  * @param cork is corking allowed for this transmission?
 957  * @param priority how important is the message?
 958  * @param maxdelay how long can the message wait? Only effective if @a cork is #GNUNET_YES
 959  * @param target who should receive the message, never NULL (can be this peer's identity for loopback)
 960  * @param notify_size how many bytes of buffer space does @a notify want?
 961  * @param notify function to call when buffer space is available;
 962  *        will be called with NULL on timeout; clients MUST cancel
 963  *        all pending transmission requests DURING the disconnect
 964  *        handler
 965  * @param notify_cls closure for @a notify
 966  * @return non-NULL if the notify callback was queued,
 967  *         NULL if we can not even queue the request (request already pending);
 968  *         if NULL is returned, @a notify will NOT be called.
 969  */
 970 struct GNUNET_CORE_TransmitHandle *
 971 GNUNET_CORE_notify_transmit_ready (struct GNUNET_CORE_Handle *handle,
 972                                    int cork,
 973                                    enum GNUNET_CORE_Priority priority,
 974                                    struct GNUNET_TIME_Relative maxdelay,
 975                                    const struct GNUNET_PeerIdentity *target,
 976                                    size_t notify_size,
 977                                    GNUNET_CONNECTION_TransmitReadyNotify notify,
 978                                    void *notify_cls)
 979 {
 980   struct PeerRecord *pr;
 981   struct GNUNET_CORE_TransmitHandle *th;
 982   struct SendMessageRequest *smr;
 983   struct GNUNET_MQ_Envelope *env;
 984
 985   if (NULL == handle->mq)
 986   {
 987     GNUNET_break (0); /* SEE #4588: do not call NTR from disconnect notification! */
 988     return NULL;
 989   }
 990   GNUNET_assert (NULL != notify);
 991   if ( (notify_size > GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE) ||
 992        (notify_size + sizeof (struct SendMessage) >= GNUNET_SERVER_MAX_MESSAGE_SIZE) )
 993   {
 994     GNUNET_break (0);
 995     return NULL;
 996   }
 997   LOG (GNUNET_ERROR_TYPE_DEBUG,
 998        "Asking core for transmission of %u bytes to `%s'\n",
 999        (unsigned int) notify_size,
1000        GNUNET_i2s (target));
1001   pr = GNUNET_CONTAINER_multipeermap_get (handle->peers,
1002                                           target);
1003   if (NULL == pr)
1004   {
1005     /* attempt to send to peer that is not connected */
1006     GNUNET_break (0);
1007     return NULL;
1008   }
1009   if (NULL != pr->th.peer)
1010   {
1011     /* attempting to queue a second request for the same destination */
1012     GNUNET_break (0);
1013     return NULL;
1014   }
1015   th = &pr->th;
1016   memset (th,
1017           0,
1018           sizeof (struct GNUNET_CORE_TransmitHandle));
1019   th->peer = pr;
1020   th->get_message = notify;
1021   th->get_message_cls = notify_cls;
1022   th->request_time = GNUNET_TIME_absolute_get ();
1023   if (GNUNET_YES == cork)
1024     th->deadline = GNUNET_TIME_relative_to_absolute (maxdelay);
1025   else
1026     th->deadline = th->request_time;
1027   th->priority = priority;
1028   th->msize = notify_size;
1029   th->cork = cork;
1030   if (NULL == handle->mq)
1031     return th; /* see #4588 (hack until we transition core fully to MQ) */
1032   env = GNUNET_MQ_msg (smr,
1033                        GNUNET_MESSAGE_TYPE_CORE_SEND_REQUEST);
1034   smr->priority = htonl ((uint32_t) th->priority);
1035   smr->deadline = GNUNET_TIME_absolute_hton (th->deadline);
1036   smr->peer = pr->peer;
1037   smr->reserved = htonl (0);
1038   smr->size = htons (th->msize);
1039   smr->smr_id = htons (th->smr_id = pr->smr_id_gen++);
1040   GNUNET_MQ_send (handle->mq,
1041                   env);
1042   LOG (GNUNET_ERROR_TYPE_DEBUG,
1043        "Transmission request added to queue\n");
1044   return th;
1045 }
1046
1047
1048 /**
1049  * Cancel the specified transmission-ready notification.
1050  *
1051  * @param th handle that was returned by #GNUNET_CORE_notify_transmit_ready().
1052  */
1053 void
1054 GNUNET_CORE_notify_transmit_ready_cancel (struct GNUNET_CORE_TransmitHandle *th)
1055 {
1056   struct PeerRecord *pr = th->peer;
1057
1058   LOG (GNUNET_ERROR_TYPE_DEBUG,
1059        "Aborting transmission request to core for %u bytes to `%s'\n",
1060        (unsigned int) th->msize,
1061        GNUNET_i2s (&pr->peer));
1062   th->peer = NULL;
1063 }
1064
1065
1066 /**
1067  * Check if the given peer is currently connected. This function is for special
1068  * cirumstances (GNUNET_TESTBED uses it), normal users of the CORE API are
1069  * expected to track which peers are connected based on the connect/disconnect
1070  * callbacks from #GNUNET_CORE_connect().  This function is NOT part of the
1071  * 'versioned', 'official' API. The difference between this function and the
1072  * function GNUNET_CORE_is_peer_connected() is that this one returns
1073  * synchronously after looking in the CORE API cache. The function
1074  * GNUNET_CORE_is_peer_connected() sends a message to the CORE service and hence
1075  * its response is given asynchronously.
1076  *
1077  * @param h the core handle
1078  * @param pid the identity of the peer to check if it has been connected to us
1079  * @return #GNUNET_YES if the peer is connected to us; #GNUNET_NO if not
1080  */
1081 int
1082 GNUNET_CORE_is_peer_connected_sync (const struct GNUNET_CORE_Handle *h,
1083                                     const struct GNUNET_PeerIdentity *pid)
1084 {
1085   return GNUNET_CONTAINER_multipeermap_contains (h->peers,
1086                                                  pid);
1087 }
1088
1089
1090 /* end of core_api.c */