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   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 545   et = ntohs (em->type);
 546   LOG (GNUNET_ERROR_TYPE_DEBUG,
 547        "Received inbound message of type %d from `%s'.\n",
 548        (int) et,
 549        GNUNET_i2s (&ntm->peer));
 550   for (unsigned int hpos = 0; NULL != h->handlers[hpos].callback; hpos++)
 551   {
 552     const struct GNUNET_CORE_MessageHandler *mh;
 553
 554     mh = &h->handlers[hpos];
 555     if (mh->type != et)
 556       continue;
 557     if ( (mh->expected_size != ntohs (em->size)) &&
 558          (0 != mh->expected_size) )
 559     {
 560       LOG (GNUNET_ERROR_TYPE_ERROR,
 561            "Unexpected message size %u for message of type %u from peer `%s'\n",
 562            htons (em->size),
 563            mh->type,
 564            GNUNET_i2s (&ntm->peer));
 565       GNUNET_break_op (0);
 566       continue;
 567     }
 568     pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 569                                             &ntm->peer);
 570     if (NULL == pr)
 571     {
 572       GNUNET_break (0);
 573       reconnect_later (h);
 574       return;
 575     }
 576     if (GNUNET_OK !=
 577         h->handlers[hpos].callback (h->cls,
 578                                     &ntm->peer,
 579                                     em))
 580     {
 581       /* error in processing, do not process other messages! */
 582       break;
 583     }
 584   }
 585   if (NULL != h->inbound_notify)
 586     h->inbound_notify (h->cls,
 587                        &ntm->peer,
 588                        em);
 589 }
 590
 591
 592 /**
 593  * Check that message received from CORE service is well-formed.
 594  *
 595  * @param cls the `struct GNUNET_CORE_Handle`
 596  * @param ntm the message we got
 597  * @return #GNUNET_OK if the message is well-formed
 598  */
 599 static int
 600 check_notify_outbound (void *cls,
 601                        const struct NotifyTrafficMessage *ntm)
 602 {
 603   struct GNUNET_CORE_Handle *h = cls;
 604   uint16_t msize;
 605   const struct GNUNET_MessageHeader *em;
 606
 607   GNUNET_break (GNUNET_NO == h->currently_down);
 608   LOG (GNUNET_ERROR_TYPE_DEBUG,
 609        "Received outbound message from `%s'.\n",
 610        GNUNET_i2s (&ntm->peer));
 611   msize = ntohs (ntm->header.size) - sizeof (struct NotifyTrafficMessage);
 612   if (msize < sizeof (struct GNUNET_MessageHeader))
 613   {
 614     GNUNET_break (0);
 615     return GNUNET_SYSERR;
 616   }
 617   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 618   if ( (GNUNET_NO == h->outbound_hdr_only) &&
 619        (msize != ntohs (em->size)) )
 620   {
 621     GNUNET_break (0);
 622     return GNUNET_SYSERR;
 623   }
 624   return GNUNET_OK;
 625 }
 626
 627
 628 /**
 629  * Handle outbound message received from CORE service.  If applicable,
 630  * notify the application.
 631  *
 632  * @param cls the `struct GNUNET_CORE_Handle`
 633  * @param ntm the message we got
 634  */
 635 static void
 636 handle_notify_outbound (void *cls,
 637                         const struct NotifyTrafficMessage *ntm)
 638 {
 639   struct GNUNET_CORE_Handle *h = cls;
 640   const struct GNUNET_MessageHeader *em;
 641
 642   GNUNET_break (GNUNET_NO == h->currently_down);
 643   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 644   LOG (GNUNET_ERROR_TYPE_DEBUG,
 645        "Received notification about transmission to `%s'.\n",
 646        GNUNET_i2s (&ntm->peer));
 647   if (NULL == h->outbound_notify)
 648   {
 649     GNUNET_break (0);
 650     return;
 651   }
 652   h->outbound_notify (h->cls,
 653                       &ntm->peer,
 654                       em);
 655 }
 656
 657
 658 /**
 659  * Handle message received from CORE service notifying us that we are
 660  * now allowed to send a message to a peer.  If that message is still
 661  * pending, put it into the queue to be transmitted.
 662  *
 663  * @param cls the `struct GNUNET_CORE_Handle`
 664  * @param ntm the message we got
 665  */
 666 static void
 667 handle_send_ready (void *cls,
 668                    const struct SendMessageReady *smr)
 669 {
 670   struct GNUNET_CORE_Handle *h = cls;
 671   struct PeerRecord *pr;
 672   struct GNUNET_CORE_TransmitHandle *th;
 673   struct SendMessage *sm;
 674   struct GNUNET_MQ_Envelope *env;
 675   struct GNUNET_TIME_Relative delay;
 676   struct GNUNET_TIME_Relative overdue;
 677   unsigned int ret;
 678   unsigned int priority;
 679   int cork;
 680
 681   GNUNET_break (GNUNET_NO == h->currently_down);
 682   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 683                                           &smr->peer);
 684   if (NULL == pr)
 685   {
 686     GNUNET_break (0);
 687     reconnect_later (h);
 688     return;
 689   }
 690   LOG (GNUNET_ERROR_TYPE_DEBUG,
 691        "Received notification about transmission readiness to `%s'.\n",
 692        GNUNET_i2s (&smr->peer));
 693   if (NULL == pr->th.peer)
 694   {
 695     /* request must have been cancelled between the original request
 696      * and the response from CORE, ignore CORE's readiness */
 697     return;
 698   }
 699   th = &pr->th;
 700   if (ntohs (smr->smr_id) != th->smr_id)
 701   {
 702     /* READY message is for expired or cancelled message,
 703      * ignore! (we should have already sent another request) */
 704     return;
 705   }
 706   /* ok, all good, send message out! */
 707   th->peer = NULL;
 708   env = GNUNET_MQ_msg_extra (sm,
 709                              th->msize,
 710                              GNUNET_MESSAGE_TYPE_CORE_SEND);
 711   sm->priority = htonl ((uint32_t) th->priority);
 712   sm->deadline = GNUNET_TIME_absolute_hton (th->deadline);
 713   sm->peer = pr->peer;
 714   sm->cork = htonl ((uint32_t) (cork = th->cork));
 715   sm->reserved = htonl (0);
 716   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 717               "Calling get_message with buffer of %u bytes (%s)\n",
 718               (unsigned int) th->msize,
 719               cork ? "corked" : "uncorked");
 720   /* FIXME: this is ugly and a bit brutal, but "get_message"
 721      may call GNUNET_CORE_notify_transmit_ready() which
 722      may call GNUNET_MQ_send() as well, and we MUST get this
 723      message out before the next SEND_REQUEST.  So we queue
 724      it (even though incomplete) and then---relying on MQ being
 725      nice and not actually touching 'env' until much later---
 726      fill it afterwards.  This is horrible style, and once
 727      the core_api abandons GNUNET_CORE_notify_transmit_ready
 728      in favor of an MQ-style API, this hack should no longer
 729      be required */
 730   GNUNET_MQ_send (h->mq,
 731                   env);
 732   delay = GNUNET_TIME_absolute_get_duration (th->request_time);
 733   overdue = GNUNET_TIME_absolute_get_duration (th->deadline);
 734   priority = th->priority;
 735   ret = th->get_message (th->get_message_cls,
 736                          th->msize,
 737                          &sm[1]);
 738   /* after this point, 'th' should not be used anymore, it
 739      may now be about another message! */
 740   sm->header.size = htons (ret + sizeof (struct SendMessage));
 741   if (overdue.rel_value_us > GNUNET_CONSTANTS_LATENCY_WARN.rel_value_us)
 742     LOG (GNUNET_ERROR_TYPE_WARNING,
 743          "Transmitting overdue %u bytes to `%s' at priority %u with %s delay %s\n",
 744          ret,
 745          GNUNET_i2s (&pr->peer),
 746          priority,
 747          GNUNET_STRINGS_relative_time_to_string (delay,
 748                                                  GNUNET_YES),
 749          (cork) ? " (corked)" : " (uncorked)");
 750   else
 751     LOG (GNUNET_ERROR_TYPE_DEBUG,
 752          "Transmitting %u bytes to `%s' at priority %u with %s delay %s\n",
 753          ret,
 754          GNUNET_i2s (&pr->peer),
 755          priority,
 756          GNUNET_STRINGS_relative_time_to_string (delay,
 757                                                  GNUNET_YES),
 758          (cork) ? " (corked)" : " (uncorked)");
 759 }
 760
 761
 762 /**
 763  * Our current client connection went down.  Clean it up and try to
 764  * reconnect!
 765  *
 766  * @param h our handle to the core service
 767  */
 768 static void
 769 reconnect (struct GNUNET_CORE_Handle *h)
 770 {
 771   struct GNUNET_MQ_MessageHandler handlers[] = {
 772     GNUNET_MQ_hd_fixed_size (init_reply,
 773                              GNUNET_MESSAGE_TYPE_CORE_INIT_REPLY,
 774                              struct InitReplyMessage,
 775                              h),
 776     GNUNET_MQ_hd_fixed_size (connect_notify,
 777                              GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT,
 778                              struct ConnectNotifyMessage,
 779                              h),
 780     GNUNET_MQ_hd_fixed_size (disconnect_notify,
 781                              GNUNET_MESSAGE_TYPE_CORE_NOTIFY_DISCONNECT,
 782                              struct DisconnectNotifyMessage,
 783                              h),
 784     GNUNET_MQ_hd_var_size (notify_inbound,
 785                            GNUNET_MESSAGE_TYPE_CORE_NOTIFY_INBOUND,
 786                            struct NotifyTrafficMessage,
 787                            h),
 788     GNUNET_MQ_hd_var_size (notify_outbound,
 789                            GNUNET_MESSAGE_TYPE_CORE_NOTIFY_OUTBOUND,
 790                            struct NotifyTrafficMessage,
 791                            h),
 792     GNUNET_MQ_hd_fixed_size (send_ready,
 793                              GNUNET_MESSAGE_TYPE_CORE_SEND_READY,
 794                              struct SendMessageReady,
 795                              h),
 796     GNUNET_MQ_handler_end ()
 797   };
 798   struct InitMessage *init;
 799   struct GNUNET_MQ_Envelope *env;
 800   uint32_t opt;
 801   uint16_t *ts;
 802
 803   GNUNET_assert (NULL == h->mq);
 804   GNUNET_assert (GNUNET_YES == h->currently_down);
 805   h->mq = GNUNET_CLIENT_connecT (h->cfg,
 806                                  "core",
 807                                  handlers,
 808                                  &handle_mq_error,
 809                                  h);
 810   if (NULL == h->mq)
 811   {
 812     reconnect_later (h);
 813     return;
 814   }
 815   env = GNUNET_MQ_msg_extra (init,
 816                              sizeof (uint16_t) * h->hcnt,
 817                              GNUNET_MESSAGE_TYPE_CORE_INIT);
 818   opt = 0;
 819   if (NULL != h->inbound_notify)
 820   {
 821     if (h->inbound_hdr_only)
 822       opt |= GNUNET_CORE_OPTION_SEND_HDR_INBOUND;
 823     else
 824       opt |= GNUNET_CORE_OPTION_SEND_FULL_INBOUND;
 825   }
 826   if (NULL != h->outbound_notify)
 827   {
 828     if (h->outbound_hdr_only)
 829       opt |= GNUNET_CORE_OPTION_SEND_HDR_OUTBOUND;
 830     else
 831       opt |= GNUNET_CORE_OPTION_SEND_FULL_OUTBOUND;
 832   }
 833   LOG (GNUNET_ERROR_TYPE_INFO,
 834        "(Re)connecting to CORE service, monitoring messages of type %u\n",
 835        opt);
 836   init->options = htonl (opt);
 837   ts = (uint16_t *) &init[1];
 838   for (unsigned int hpos = 0; hpos < h->hcnt; hpos++)
 839     ts[hpos] = htons (h->handlers[hpos].type);
 840   GNUNET_MQ_send (h->mq,
 841                   env);
 842 }
 843
 844
 845 /**
 846  * Connect to the core service.  Note that the connection may complete
 847  * (or fail) asynchronously.
 848  *
 849  * @param cfg configuration to use
 850  * @param cls closure for the various callbacks that follow (including handlers in the handlers array)
 851  * @param init callback to call once we have successfully
 852  *        connected to the core service
 853  * @param connects function to call on peer connect, can be NULL
 854  * @param disconnects function to call on peer disconnect / timeout, can be NULL
 855  * @param inbound_notify function to call for all inbound messages, can be NULL
 856  * @param inbound_hdr_only set to #GNUNET_YES if inbound_notify will only read the
 857  *                GNUNET_MessageHeader and hence we do not need to give it the full message;
 858  *                can be used to improve efficiency, ignored if @a inbound_notify is NULL
 859  * @param outbound_notify function to call for all outbound messages, can be NULL
 860  * @param outbound_hdr_only set to #GNUNET_YES if outbound_notify will only read the
 861  *                GNUNET_MessageHeader and hence we do not need to give it the full message
 862  *                can be used to improve efficiency, ignored if @a outbound_notify is NULL
 863  * @param handlers callbacks for messages we care about, NULL-terminated
 864  * @return handle to the core service (only useful for disconnect until @a init is called);
 865  *                NULL on error (in this case, init is never called)
 866  */
 867 struct GNUNET_CORE_Handle *
 868 GNUNET_CORE_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
 869                      void *cls,
 870                      GNUNET_CORE_StartupCallback init,
 871                      GNUNET_CORE_ConnectEventHandler connects,
 872                      GNUNET_CORE_DisconnectEventHandler disconnects,
 873                      GNUNET_CORE_MessageCallback inbound_notify,
 874                      int inbound_hdr_only,
 875                      GNUNET_CORE_MessageCallback outbound_notify,
 876                      int outbound_hdr_only,
 877                      const struct GNUNET_CORE_MessageHandler *handlers)
 878 {
 879   struct GNUNET_CORE_Handle *h;
 880   unsigned int hcnt;
 881
 882   h = GNUNET_new (struct GNUNET_CORE_Handle);
 883   h->cfg = cfg;
 884   h->cls = cls;
 885   h->init = init;
 886   h->connects = connects;
 887   h->disconnects = disconnects;
 888   h->inbound_notify = inbound_notify;
 889   h->outbound_notify = outbound_notify;
 890   h->inbound_hdr_only = inbound_hdr_only;
 891   h->outbound_hdr_only = outbound_hdr_only;
 892   h->currently_down = GNUNET_YES;
 893   h->peers = GNUNET_CONTAINER_multipeermap_create (128, GNUNET_NO);
 894   hcnt = 0;
 895   if (NULL != handlers)
 896     while (NULL != handlers[hcnt].callback)
 897       hcnt++;
 898   h->handlers = GNUNET_new_array (hcnt + 1,
 899                                   struct GNUNET_CORE_MessageHandler);
 900   if (NULL != handlers)
 901     GNUNET_memcpy (h->handlers,
 902             handlers,
 903             hcnt * sizeof (struct GNUNET_CORE_MessageHandler));
 904   h->hcnt = hcnt;
 905   GNUNET_assert (hcnt <
 906                  (GNUNET_SERVER_MAX_MESSAGE_SIZE -
 907                   sizeof (struct InitMessage)) / sizeof (uint16_t));
 908   LOG (GNUNET_ERROR_TYPE_DEBUG,
 909        "Connecting to CORE service\n");
 910   reconnect (h);
 911   if (NULL == h->mq)
 912   {
 913     GNUNET_CORE_disconnect (h);
 914     return NULL;
 915   }
 916   return h;
 917 }
 918
 919
 920 /**
 921  * Disconnect from the core service.  This function can only
 922  * be called *after* all pending #GNUNET_CORE_notify_transmit_ready()
 923  * requests have been explicitly canceled.
 924  *
 925  * @param handle connection to core to disconnect
 926  */
 927 void
 928 GNUNET_CORE_disconnect (struct GNUNET_CORE_Handle *handle)
 929 {
 930   LOG (GNUNET_ERROR_TYPE_DEBUG,
 931        "Disconnecting from CORE service\n");
 932   GNUNET_CONTAINER_multipeermap_iterate (handle->peers,
 933                                          &disconnect_and_free_peer_entry,
 934                                          handle);
 935   GNUNET_CONTAINER_multipeermap_destroy (handle->peers);
 936   handle->peers = NULL;
 937   if (NULL != handle->reconnect_task)
 938   {
 939     GNUNET_SCHEDULER_cancel (handle->reconnect_task);
 940     handle->reconnect_task = NULL;
 941   }
 942   if (NULL != handle->mq)
 943   {
 944     GNUNET_MQ_destroy (handle->mq);
 945     handle->mq = NULL;
 946   }
 947   GNUNET_free (handle->handlers);
 948   GNUNET_free (handle);
 949 }
 950
 951
 952 /**
 953  * Ask the core to call @a notify once it is ready to transmit the
 954  * given number of bytes to the specified @a target.  Must only be
 955  * called after a connection to the respective peer has been
 956  * established (and the client has been informed about this).  You may
 957  * have one request of this type pending for each connected peer at
 958  * any time.  If a peer disconnects, the application MUST call
 959  * #GNUNET_CORE_notify_transmit_ready_cancel on the respective
 960  * transmission request, if one such request is pending.
 961  *
 962  * @param handle connection to core service
 963  * @param cork is corking allowed for this transmission?
 964  * @param priority how important is the message?
 965  * @param maxdelay how long can the message wait? Only effective if @a cork is #GNUNET_YES
 966  * @param target who should receive the message, never NULL (can be this peer's identity for loopback)
 967  * @param notify_size how many bytes of buffer space does @a notify want?
 968  * @param notify function to call when buffer space is available;
 969  *        will be called with NULL on timeout; clients MUST cancel
 970  *        all pending transmission requests DURING the disconnect
 971  *        handler
 972  * @param notify_cls closure for @a notify
 973  * @return non-NULL if the notify callback was queued,
 974  *         NULL if we can not even queue the request (request already pending);
 975  *         if NULL is returned, @a notify will NOT be called.
 976  */
 977 struct GNUNET_CORE_TransmitHandle *
 978 GNUNET_CORE_notify_transmit_ready (struct GNUNET_CORE_Handle *handle,
 979                                    int cork,
 980                                    enum GNUNET_CORE_Priority priority,
 981                                    struct GNUNET_TIME_Relative maxdelay,
 982                                    const struct GNUNET_PeerIdentity *target,
 983                                    size_t notify_size,
 984                                    GNUNET_CONNECTION_TransmitReadyNotify notify,
 985                                    void *notify_cls)
 986 {
 987   struct PeerRecord *pr;
 988   struct GNUNET_CORE_TransmitHandle *th;
 989   struct SendMessageRequest *smr;
 990   struct GNUNET_MQ_Envelope *env;
 991
 992   if (NULL == handle->mq)
 993   {
 994     GNUNET_break (0); /* SEE #4588: do not call NTR from disconnect notification! */
 995     return NULL;
 996   }
 997   GNUNET_assert (NULL != notify);
 998   if ( (notify_size > GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE) ||
 999        (notify_size + sizeof (struct SendMessage) >= GNUNET_SERVER_MAX_MESSAGE_SIZE) )
1000   {
1001     GNUNET_break (0);
1002     return NULL;
1003   }
1004   LOG (GNUNET_ERROR_TYPE_DEBUG,
1005        "Asking core for transmission of %u bytes to `%s'%s\n",
1006        (unsigned int) notify_size,
1007        GNUNET_i2s (target),
1008        cork ? " (corked)" : "");
1009   pr = GNUNET_CONTAINER_multipeermap_get (handle->peers,
1010                                           target);
1011   if (NULL == pr)
1012   {
1013     /* attempt to send to peer that is not connected */
1014     GNUNET_break (0);
1015     return NULL;
1016   }
1017   if (NULL != pr->th.peer)
1018   {
1019     /* attempting to queue a second request for the same destination */
1020     GNUNET_break (0);
1021     return NULL;
1022   }
1023   th = &pr->th;
1024   memset (th,
1025           0,
1026           sizeof (struct GNUNET_CORE_TransmitHandle));
1027   th->peer = pr;
1028   th->get_message = notify;
1029   th->get_message_cls = notify_cls;
1030   th->request_time = GNUNET_TIME_absolute_get ();
1031   if (GNUNET_YES == cork)
1032     th->deadline = GNUNET_TIME_relative_to_absolute (maxdelay);
1033   else
1034     th->deadline = th->request_time;
1035   th->priority = priority;
1036   th->msize = notify_size;
1037   th->cork = cork;
1038   if (NULL == handle->mq)
1039     return th; /* see #4588 (hack until we transition core fully to MQ) */
1040   env = GNUNET_MQ_msg (smr,
1041                        GNUNET_MESSAGE_TYPE_CORE_SEND_REQUEST);
1042   smr->priority = htonl ((uint32_t) th->priority);
1043   smr->deadline = GNUNET_TIME_absolute_hton (th->deadline);
1044   smr->peer = pr->peer;
1045   smr->reserved = htonl (0);
1046   smr->size = htons (th->msize);
1047   smr->smr_id = htons (th->smr_id = pr->smr_id_gen++);
1048   GNUNET_MQ_send (handle->mq,
1049                   env);
1050   LOG (GNUNET_ERROR_TYPE_DEBUG,
1051        "Transmission request added to queue\n");
1052   return th;
1053 }
1054
1055
1056 /**
1057  * Cancel the specified transmission-ready notification.
1058  *
1059  * @param th handle that was returned by #GNUNET_CORE_notify_transmit_ready().
1060  */
1061 void
1062 GNUNET_CORE_notify_transmit_ready_cancel (struct GNUNET_CORE_TransmitHandle *th)
1063 {
1064   struct PeerRecord *pr = th->peer;
1065
1066   LOG (GNUNET_ERROR_TYPE_DEBUG,
1067        "Aborting transmission request to core for %u bytes to `%s'\n",
1068        (unsigned int) th->msize,
1069        GNUNET_i2s (&pr->peer));
1070   th->peer = NULL;
1071 }
1072
1073
1074 /**
1075  * Check if the given peer is currently connected. This function is for special
1076  * cirumstances (GNUNET_TESTBED uses it), normal users of the CORE API are
1077  * expected to track which peers are connected based on the connect/disconnect
1078  * callbacks from #GNUNET_CORE_connect().  This function is NOT part of the
1079  * 'versioned', 'official' API. The difference between this function and the
1080  * function GNUNET_CORE_is_peer_connected() is that this one returns
1081  * synchronously after looking in the CORE API cache. The function
1082  * GNUNET_CORE_is_peer_connected() sends a message to the CORE service and hence
1083  * its response is given asynchronously.
1084  *
1085  * @param h the core handle
1086  * @param pid the identity of the peer to check if it has been connected to us
1087  * @return #GNUNET_YES if the peer is connected to us; #GNUNET_NO if not
1088  */
1089 int
1090 GNUNET_CORE_is_peer_connected_sync (const struct GNUNET_CORE_Handle *h,
1091                                     const struct GNUNET_PeerIdentity *pid)
1092 {
1093   return GNUNET_CONTAINER_multipeermap_contains (h->peers,
1094                                                  pid);
1095 }
1096
1097
1098 /* end of core_api.c */