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