src/core/core_api.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009, 2010 Christian Grothoff (and other contributing authors)
   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., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file core/core_api.c
  23  * @brief core service; this is the main API for encrypted P2P
  24  *        communications
  25  * @author Christian Grothoff
  26  *
  27  * TODO:
  28  * - implement atsi parsing and passing
  29  */
  30 #include "platform.h"
  31 #include "gnunet_constants.h"
  32 #include "gnunet_core_service.h"
  33 #include "core.h"
  34
  35
  36 /**
  37  * Information we track for each peer.
  38  */
  39 struct PeerRecord
  40 {
  41
  42   /**
  43    * We generally do NOT keep peer records in a DLL; this
  44    * DLL is only used IF this peer's 'pending_head' message
  45    * is ready for transmission.
  46    */
  47   struct PeerRecord *prev;
  48
  49   /**
  50    * We generally do NOT keep peer records in a DLL; this
  51    * DLL is only used IF this peer's 'pending_head' message
  52    * is ready for transmission.
  53    */
  54   struct PeerRecord *next;
  55
  56   /**
  57    * Peer the record is about.
  58    */
  59   struct GNUNET_PeerIdentity peer;
  60
  61   /**
  62    * Corresponding core handle.
  63    */
  64   struct GNUNET_CORE_Handle *ch;
  65
  66   /**
  67    * Head of doubly-linked list of pending requests.
  68    * Requests are sorted by deadline *except* for HEAD,
  69    * which is only modified upon transmission to core.
  70    */
  71   struct GNUNET_CORE_TransmitHandle *pending_head;
  72
  73   /**
  74    * Tail of doubly-linked list of pending requests.
  75    */
  76   struct GNUNET_CORE_TransmitHandle *pending_tail;
  77
  78   /**
  79    * Pending callback waiting for peer information, or NULL for none.
  80    */
  81   GNUNET_CORE_PeerConfigurationInfoCallback pcic;
  82
  83   /**
  84    * Closure for pcic.
  85    */
  86   void *pcic_cls;
  87
  88   /**
  89    * Request information ID for the given pcic (needed in case a
  90    * request is cancelled after being submitted to core and a new
  91    * one is generated; in this case, we need to avoid matching the
  92    * reply to the first (cancelled) request to the second request).
  93    */
  94   uint32_t rim_id;
  95
  96   /**
  97    * ID of timeout task for the 'pending_head' handle
  98    * which is the one with the smallest timeout.
  99    */
 100   GNUNET_SCHEDULER_TaskIdentifier timeout_task;
 101
 102   /**
 103    * Current size of the queue of pending requests.
 104    */
 105   unsigned int queue_size;
 106
 107   /**
 108    * SendMessageRequest ID generator for this peer.
 109    */
 110   uint16_t smr_id_gen;
 111
 112 };
 113
 114
 115 /**
 116  * Entry in a doubly-linked list of control messages to be transmitted
 117  * to the core service.  Control messages include traffic allocation,
 118  * connection requests and of course our initial 'init' request.
 119  *
 120  * The actual message is allocated at the end of this struct.
 121  */
 122 struct ControlMessage
 123 {
 124   /**
 125    * This is a doubly-linked list.
 126    */
 127   struct ControlMessage *next;
 128
 129   /**
 130    * This is a doubly-linked list.
 131    */
 132   struct ControlMessage *prev;
 133
 134   /**
 135    * Function to run after successful transmission (or call with
 136    * reason 'TIMEOUT' on error); called with scheduler context 'NULL'
 137    * on disconnect.
 138    */
 139   GNUNET_SCHEDULER_Task cont;
 140
 141   /**
 142    * Closure for 'cont'.
 143    */
 144   void *cont_cls;
 145
 146 };
 147
 148
 149
 150 /**
 151  * Context for the core service connection.
 152  */
 153 struct GNUNET_CORE_Handle
 154 {
 155
 156   /**
 157    * Configuration we're using.
 158    */
 159   const struct GNUNET_CONFIGURATION_Handle *cfg;
 160
 161   /**
 162    * Closure for the various callbacks.
 163    */
 164   void *cls;
 165
 166   /**
 167    * Function to call once we've handshaked with the core service.
 168    */
 169   GNUNET_CORE_StartupCallback init;
 170
 171   /**
 172    * Function to call whenever we're notified about a peer connecting.
 173    */
 174   GNUNET_CORE_ConnectEventHandler connects;
 175
 176   /**
 177    * Function to call whenever we're notified about a peer disconnecting.
 178    */
 179   GNUNET_CORE_DisconnectEventHandler disconnects;
 180
 181   /**
 182    * Function to call whenever we're notified about a peer changing status.
 183    */
 184   GNUNET_CORE_PeerStatusEventHandler status_events;
 185
 186   /**
 187    * Function to call whenever we receive an inbound message.
 188    */
 189   GNUNET_CORE_MessageCallback inbound_notify;
 190
 191   /**
 192    * Function to call whenever we receive an outbound message.
 193    */
 194   GNUNET_CORE_MessageCallback outbound_notify;
 195
 196   /**
 197    * Function handlers for messages of particular type.
 198    */
 199   const struct GNUNET_CORE_MessageHandler *handlers;
 200
 201   /**
 202    * Our connection to the service.
 203    */
 204   struct GNUNET_CLIENT_Connection *client;
 205
 206   /**
 207    * Handle for our current transmission request.
 208    */
 209   struct GNUNET_CLIENT_TransmitHandle *cth;
 210
 211   /**
 212    * Head of doubly-linked list of pending requests.
 213    */
 214   struct ControlMessage *pending_head;
 215
 216   /**
 217    * Tail of doubly-linked list of pending requests.
 218    */
 219   struct ControlMessage *pending_tail;
 220
 221   /**
 222    * Head of doubly-linked list of peers that are core-approved
 223    * to send their next message.
 224    */
 225   struct PeerRecord *ready_peer_head;
 226
 227   /**
 228    * Tail of doubly-linked list of peers that are core-approved
 229    * to send their next message.
 230    */
 231   struct PeerRecord *ready_peer_tail;
 232
 233   /**
 234    * Hash map listing all of the peers that we are currently
 235    * connected to.
 236    */
 237   struct GNUNET_CONTAINER_MultiHashMap *peers;
 238
 239   /**
 240    * Identity of this peer.
 241    */
 242   struct GNUNET_PeerIdentity me;
 243
 244   /**
 245    * ID of reconnect task (if any).
 246    */
 247   GNUNET_SCHEDULER_TaskIdentifier reconnect_task;
 248
 249   /**
 250    * Current delay we use for re-trying to connect to core.
 251    */
 252   struct GNUNET_TIME_Relative retry_backoff;
 253
 254   /**
 255    * Request information ID generator.
 256    */
 257   uint32_t rim_id_gen;
 258
 259   /**
 260    * Number of messages we are allowed to queue per target.
 261    */
 262   unsigned int queue_size;
 263
 264   /**
 265    * Number of entries in the handlers array.
 266    */
 267   unsigned int hcnt;
 268
 269   /**
 270    * For inbound notifications without a specific handler, do
 271    * we expect to only receive headers?
 272    */
 273   int inbound_hdr_only;
 274
 275   /**
 276    * For outbound notifications without a specific handler, do
 277    * we expect to only receive headers?
 278    */
 279   int outbound_hdr_only;
 280
 281   /**
 282    * Are we currently disconnected and hence unable to forward
 283    * requests?
 284    */
 285   int currently_down;
 286
 287 };
 288
 289
 290 /**
 291  * Handle for a transmission request.
 292  */
 293 struct GNUNET_CORE_TransmitHandle
 294 {
 295
 296   /**
 297    * We keep active transmit handles in a doubly-linked list.
 298    */
 299   struct GNUNET_CORE_TransmitHandle *next;
 300
 301   /**
 302    * We keep active transmit handles in a doubly-linked list.
 303    */
 304   struct GNUNET_CORE_TransmitHandle *prev;
 305
 306   /**
 307    * Corresponding peer record.
 308    */
 309   struct PeerRecord *peer;
 310
 311   /**
 312    * Corresponding SEND_REQUEST message.  Only non-NULL
 313    * while SEND_REQUEST message is pending.
 314    */
 315   struct ControlMessage *cm;
 316
 317   /**
 318    * Function that will be called to get the actual request
 319    * (once we are ready to transmit this request to the core).
 320    * The function will be called with a NULL buffer to signal
 321    * timeout.
 322    */
 323   GNUNET_CONNECTION_TransmitReadyNotify get_message;
 324
 325   /**
 326    * Closure for get_message.
 327    */
 328   void *get_message_cls;
 329
 330   /**
 331    * Timeout for this handle.
 332    */
 333   struct GNUNET_TIME_Absolute timeout;
 334
 335   /**
 336    * How important is this message?
 337    */
 338   uint32_t priority;
 339
 340   /**
 341    * Size of this request.
 342    */
 343   uint16_t msize;
 344
 345   /**
 346    * Send message request ID for this request.
 347    */
 348   uint16_t smr_id;
 349
 350 };
 351
 352
 353 /**
 354  * Our current client connection went down.  Clean it up
 355  * and try to reconnect!
 356  *
 357  * @param h our handle to the core service
 358  */
 359 static void
 360 reconnect (struct GNUNET_CORE_Handle *h);
 361
 362
 363 /**
 364  * Task schedule to try to re-connect to core.
 365  *
 366  * @param cls the 'struct GNUNET_CORE_Handle'
 367  * @param tc task context
 368  */
 369 static void
 370 reconnect_task (void *cls,
 371                 const struct GNUNET_SCHEDULER_TaskContext *tc)
 372 {
 373   struct GNUNET_CORE_Handle *h = cls;
 374
 375   h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
 376 #if DEBUG_CORE
 377   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 378               "Connecting to CORE service after delay\n");
 379 #endif
 380   reconnect (h);
 381 }
 382
 383
 384 /**
 385  * Notify clients about disconnect and free
 386  * the entry for connected peer.
 387  *
 388  * @param cls the 'struct GNUNET_CORE_Handle*'
 389  * @param key the peer identity (not used)
 390  * @param value the 'struct PeerRecord' to free.
 391  * @return GNUNET_YES (continue)
 392  */
 393 static int
 394 disconnect_and_free_peer_entry (void *cls,
 395                                 const GNUNET_HashCode *key,
 396                                 void *value)
 397 {
 398   static struct GNUNET_BANDWIDTH_Value32NBO zero;
 399   struct GNUNET_CORE_Handle *h = cls;
 400   struct GNUNET_CORE_TransmitHandle *th;
 401   struct PeerRecord *pr = value;
 402   GNUNET_CORE_PeerConfigurationInfoCallback pcic;
 403
 404   while (NULL != (th = pr->pending_head))
 405     {
 406       GNUNET_CONTAINER_DLL_remove (pr->pending_head,
 407                                    pr->pending_tail,
 408                                    th);
 409       pr->queue_size--;
 410       GNUNET_assert (0 ==
 411                      th->get_message (th->get_message_cls,
 412                                       0, NULL));
 413       GNUNET_free (th);
 414     }
 415   if (NULL != (pcic = pr->pcic))
 416     {
 417       pr->pcic = NULL;
 418       pcic (pr->pcic_cls,
 419             &pr->peer,
 420             zero,
 421             0, 0);
 422     }
 423   if (pr->timeout_task != GNUNET_SCHEDULER_NO_TASK)
 424     {
 425       GNUNET_SCHEDULER_cancel (pr->timeout_task);
 426       pr->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 427     }
 428   GNUNET_assert (pr->queue_size == 0);
 429   if ( (pr->prev != NULL) ||
 430        (pr->next != NULL) ||
 431        (h->ready_peer_head == pr) )
 432     GNUNET_CONTAINER_DLL_remove (h->ready_peer_head,
 433                                  h->ready_peer_tail,
 434                                  pr);
 435   if (h->disconnects != NULL)
 436     h->disconnects (h->cls,
 437                     &pr->peer);
 438   GNUNET_assert (GNUNET_YES ==
 439                  GNUNET_CONTAINER_multihashmap_remove (h->peers,
 440                                                        key,
 441                                                        pr));
 442   GNUNET_assert (pr->pending_head == NULL);
 443   GNUNET_assert (pr->pending_tail == NULL);
 444   GNUNET_assert (pr->ch = h);
 445   GNUNET_assert (pr->queue_size == 0);
 446   GNUNET_assert (pr->timeout_task == GNUNET_SCHEDULER_NO_TASK);
 447   GNUNET_free (pr);
 448   return GNUNET_YES;
 449 }
 450
 451
 452 /**
 453  * Close down any existing connection to the CORE service and
 454  * try re-establishing it later.
 455  *
 456  * @param h our handle
 457  */
 458 static void
 459 reconnect_later (struct GNUNET_CORE_Handle *h)
 460 {
 461   struct ControlMessage *cm;
 462
 463   if (h->client != NULL)
 464     {
 465       GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
 466       h->client = NULL;
 467       GNUNET_CONTAINER_multihashmap_iterate (h->peers,
 468                                              &disconnect_and_free_peer_entry,
 469                                              h);
 470     }
 471   h->currently_down = GNUNET_YES;
 472   while (NULL != (cm = h->pending_head))
 473     {
 474       GNUNET_CONTAINER_DLL_remove (h->pending_head,
 475                                    h->pending_tail,
 476                                    cm);
 477       cm->cont (cm->cont_cls, NULL);
 478       GNUNET_free (cm);
 479     }
 480   GNUNET_assert (h->reconnect_task == GNUNET_SCHEDULER_NO_TASK);
 481   h->retry_backoff = GNUNET_TIME_relative_min (GNUNET_TIME_UNIT_SECONDS,
 482                                                h->retry_backoff);
 483   h->reconnect_task = GNUNET_SCHEDULER_add_delayed (h->retry_backoff,
 484                                                     &reconnect_task,
 485                                                     h);
 486   h->retry_backoff = GNUNET_TIME_relative_multiply (h->retry_backoff, 2);
 487 }
 488
 489
 490 /**
 491  * Check the list of pending requests, send the next
 492  * one to the core.
 493  *
 494  * @param h core handle
 495  * @param ignore_currently_down transmit message even if not initialized?
 496  */
 497 static void
 498 trigger_next_request (struct GNUNET_CORE_Handle *h,
 499                       int ignore_currently_down);
 500
 501
 502 /**
 503  * The given request hit its timeout.  Remove from the
 504  * doubly-linked list and call the respective continuation.
 505  *
 506  * @param cls the transmit handle of the request that timed out
 507  * @param tc context, can be NULL (!)
 508  */
 509 static void
 510 transmission_timeout (void *cls,
 511                       const struct GNUNET_SCHEDULER_TaskContext *tc);
 512
 513
 514 /**
 515  * Control message was sent, mark it as such.
 516  *
 517  * @param cls the 'struct GNUNET_CORE_TransmitHandle*'
 518  * @param tc scheduler context
 519  */
 520 static void
 521 mark_control_message_sent (void *cls,
 522                            const struct GNUNET_SCHEDULER_TaskContext *tc)
 523 {
 524   struct GNUNET_CORE_TransmitHandle *th = cls;
 525
 526   th->cm = NULL;
 527 }
 528
 529
 530 /**
 531  * Send a control message to the peer asking for transmission
 532  * of the message in the given peer record.
 533  *
 534  * @param pr peer to request transmission to
 535  */
 536 static void
 537 request_next_transmission (struct PeerRecord *pr)
 538 {
 539   struct GNUNET_CORE_Handle *h = pr->ch;
 540   struct ControlMessage *cm;
 541   struct SendMessageRequest *smr;
 542   struct GNUNET_CORE_TransmitHandle *th;
 543
 544   if (pr->timeout_task != GNUNET_SCHEDULER_NO_TASK)
 545     {
 546       GNUNET_SCHEDULER_cancel (pr->timeout_task);
 547       pr->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 548     }
 549   if (NULL == (th = pr->pending_head))
 550     {
 551       trigger_next_request (h, GNUNET_NO);
 552       return;
 553     }
 554   GNUNET_assert (pr->prev == NULL);
 555   GNUNET_assert (pr->next == NULL);
 556   pr->timeout_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining (th->timeout),
 557                                                    &transmission_timeout,
 558                                                    pr);
 559   cm = GNUNET_malloc (sizeof (struct ControlMessage) +
 560                       sizeof (struct SendMessageRequest));
 561   cm->cont = &mark_control_message_sent;
 562   cm->cont_cls = th;
 563   th->cm = cm;
 564   smr = (struct SendMessageRequest*) &cm[1];
 565   smr->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_SEND_REQUEST);
 566   smr->header.size = htons (sizeof (struct SendMessageRequest));
 567   smr->priority = htonl (th->priority);
 568   smr->deadline = GNUNET_TIME_absolute_hton (th->timeout);
 569   smr->peer = pr->peer;
 570   smr->queue_size = htonl (pr->queue_size);
 571   smr->size = htons (th->msize);
 572   smr->smr_id = htons (th->smr_id = pr->smr_id_gen++);
 573   GNUNET_CONTAINER_DLL_insert_after (h->pending_head,
 574                                      h->pending_tail,
 575                                      h->pending_tail,
 576                                      cm);
 577 #if DEBUG_CORE
 578   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 579               "Adding SEND REQUEST for peer `%s' to message queue\n",
 580               GNUNET_i2s (&pr->peer));
 581 #endif
 582   trigger_next_request (h, GNUNET_NO);
 583 }
 584
 585
 586 /**
 587  * The given request hit its timeout.  Remove from the
 588  * doubly-linked list and call the respective continuation.
 589  *
 590  * @param cls the transmit handle of the request that timed out
 591  * @param tc context, can be NULL (!)
 592  */
 593 static void
 594 transmission_timeout (void *cls,
 595                       const struct GNUNET_SCHEDULER_TaskContext *tc)
 596 {
 597   struct PeerRecord *pr = cls;
 598   struct GNUNET_CORE_TransmitHandle *th;
 599
 600   pr->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 601   th = pr->pending_head;
 602   GNUNET_CONTAINER_DLL_remove (pr->pending_head,
 603                                pr->pending_tail,
 604                                th);
 605   pr->queue_size--;
 606 #if DEBUG_CORE
 607   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 608               "Signalling timeout of request for transmission to CORE service\n");
 609 #endif
 610   GNUNET_assert (0 == th->get_message (th->get_message_cls, 0, NULL));
 611   request_next_transmission (pr);
 612 }
 613
 614
 615 /**
 616  * Transmit the next message to the core service.
 617  */
 618 static size_t
 619 transmit_message (void *cls,
 620                   size_t size,
 621                   void *buf)
 622 {
 623   struct GNUNET_CORE_Handle *h = cls;
 624   struct ControlMessage *cm;
 625   struct GNUNET_CORE_TransmitHandle *th;
 626   struct PeerRecord *pr;
 627   struct SendMessage *sm;
 628   const struct GNUNET_MessageHeader *hdr;
 629   uint16_t msize;
 630   size_t ret;
 631
 632   h->cth = NULL;
 633   if (buf == NULL)
 634     {
 635 #if DEBUG_CORE
 636       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 637                   "Transmission failed, initiating reconnect\n");
 638 #endif
 639       reconnect_later (h);
 640       return 0;
 641     }
 642   /* first check for control messages */
 643   if (NULL != (cm = h->pending_head))
 644     {
 645       hdr = (const struct GNUNET_MessageHeader*) &cm[1];
 646       msize = ntohs (hdr->size);
 647       if (size < msize)
 648         {
 649           trigger_next_request (h, GNUNET_NO);
 650           return 0;
 651         }
 652 #if DEBUG_CORE
 653       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 654                   "Transmitting control message with %u bytes of type %u to core.\n",
 655                   (unsigned int) msize,
 656                   (unsigned int) ntohs (hdr->type));
 657 #endif
 658       memcpy (buf, hdr, msize);
 659       GNUNET_CONTAINER_DLL_remove (h->pending_head,
 660                                    h->pending_tail,
 661                                    cm);
 662       if (NULL != cm->cont)
 663         GNUNET_SCHEDULER_add_continuation (cm->cont,
 664                                            cm->cont_cls,
 665                                            GNUNET_SCHEDULER_REASON_PREREQ_DONE);
 666       GNUNET_free (cm);
 667       trigger_next_request (h, GNUNET_NO);
 668       return msize;
 669     }
 670   /* now check for 'ready' P2P messages */
 671   if (NULL != (pr = h->ready_peer_head))
 672     {
 673       th = pr->pending_head;
 674       if (size < th->msize + sizeof (struct SendMessage))
 675         {
 676           trigger_next_request (h, GNUNET_NO);
 677           return 0;
 678         }
 679       GNUNET_CONTAINER_DLL_remove (h->ready_peer_head,
 680                                    h->ready_peer_tail,
 681                                    pr);
 682       GNUNET_CONTAINER_DLL_remove (pr->pending_head,
 683                                    pr->pending_tail,
 684                                    th);
 685       pr->queue_size--;
 686       if (pr->timeout_task != GNUNET_SCHEDULER_NO_TASK)
 687         {
 688           GNUNET_SCHEDULER_cancel (pr->timeout_task);
 689           pr->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 690         }
 691 #if DEBUG_CORE
 692       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 693                   "Transmitting SEND request to `%s' with %u bytes.\n",
 694                   GNUNET_i2s (&pr->peer),
 695                   (unsigned int) th->msize);
 696 #endif
 697       sm = (struct SendMessage *) buf;
 698       sm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_SEND);
 699       sm->priority = htonl (th->priority);
 700       sm->deadline = GNUNET_TIME_absolute_hton (th->timeout);
 701       sm->peer = pr->peer;
 702       ret = th->get_message (th->get_message_cls,
 703                              size - sizeof (struct SendMessage),
 704                              &sm[1]);
 705
 706       if (0 == ret)
 707         {
 708 #if DEBUG_CORE
 709           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 710                       "Size of clients message to peer %s is 0!\n",
 711                       GNUNET_i2s(&pr->peer));
 712 #endif
 713           /* client decided to send nothing! */
 714           request_next_transmission (pr);
 715           return 0;
 716         }
 717 #if DEBUG_CORE
 718       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 719                   "Produced SEND message to core with %u bytes payload\n",
 720                   (unsigned int) ret);
 721 #endif
 722       GNUNET_assert (ret >= sizeof (struct GNUNET_MessageHeader));
 723       if (ret + sizeof (struct SendMessage) >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
 724         {
 725           GNUNET_break (0);
 726           request_next_transmission (pr);
 727           return 0;
 728         }
 729       ret += sizeof (struct SendMessage);
 730       sm->header.size = htons (ret);
 731       GNUNET_assert (ret <= size);
 732       GNUNET_free (th);
 733       request_next_transmission (pr);
 734       return ret;
 735     }
 736   return 0;
 737 }
 738
 739
 740 /**
 741  * Check the list of pending requests, send the next
 742  * one to the core.
 743  *
 744  * @param h core handle
 745  * @param ignore_currently_down transmit message even if not initialized?
 746  */
 747 static void
 748 trigger_next_request (struct GNUNET_CORE_Handle *h,
 749                       int ignore_currently_down)
 750 {
 751   uint16_t msize;
 752
 753   if ( (GNUNET_YES == h->currently_down) &&
 754        (ignore_currently_down == GNUNET_NO) )
 755     {
 756 #if DEBUG_CORE
 757       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 758                   "Core connection down, not processing queue\n");
 759 #endif
 760       return;
 761     }
 762   if (NULL != h->cth)
 763     {
 764 #if DEBUG_CORE
 765       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 766                   "Request pending, not processing queue\n");
 767 #endif
 768       return;
 769     }
 770   if (h->pending_head != NULL)
 771     msize = ntohs (((struct GNUNET_MessageHeader*) &h->pending_head[1])->size);
 772   else if (h->ready_peer_head != NULL)
 773     msize = h->ready_peer_head->pending_head->msize + sizeof (struct SendMessage);
 774   else
 775     {
 776 #if DEBUG_CORE
 777       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 778                   "Request queue empty, not processing queue\n");
 779 #endif
 780       return; /* no pending message */
 781     }
 782   h->cth = GNUNET_CLIENT_notify_transmit_ready (h->client,
 783                                                 msize,
 784                                                 GNUNET_TIME_UNIT_FOREVER_REL,
 785                                                 GNUNET_NO,
 786                                                 &transmit_message, h);
 787 }
 788
 789
 790 /**
 791  * Handler for notification messages received from the core.
 792  *
 793  * @param cls our "struct GNUNET_CORE_Handle"
 794  * @param msg the message received from the core service
 795  */
 796 static void
 797 main_notify_handler (void *cls,
 798                      const struct GNUNET_MessageHeader *msg)
 799 {
 800   struct GNUNET_CORE_Handle *h = cls;
 801   const struct InitReplyMessage *m;
 802   const struct ConnectNotifyMessage *cnm;
 803   const struct DisconnectNotifyMessage *dnm;
 804   const struct NotifyTrafficMessage *ntm;
 805   const struct GNUNET_MessageHeader *em;
 806   const struct ConfigurationInfoMessage *cim;
 807   const struct PeerStatusNotifyMessage *psnm;
 808   const struct SendMessageReady *smr;
 809   const struct GNUNET_CORE_MessageHandler *mh;
 810   GNUNET_CORE_StartupCallback init;
 811   GNUNET_CORE_PeerConfigurationInfoCallback pcic;
 812   struct PeerRecord *pr;
 813   struct GNUNET_CORE_TransmitHandle *th;
 814   unsigned int hpos;
 815   int trigger;
 816   uint16_t msize;
 817   uint16_t et;
 818
 819   if (msg == NULL)
 820     {
 821       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 822                   _
 823                   ("Client was disconnected from core service, trying to reconnect.\n"));
 824       reconnect_later (h);
 825       return;
 826     }
 827   msize = ntohs (msg->size);
 828 #if DEBUG_CORE
 829   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 830               "Processing message of type %u and size %u from core service\n",
 831               ntohs (msg->type), msize);
 832 #endif
 833   switch (ntohs (msg->type))
 834     {
 835     case GNUNET_MESSAGE_TYPE_CORE_INIT_REPLY:
 836       if (ntohs (msg->size) != sizeof (struct InitReplyMessage))
 837         {
 838           GNUNET_break (0);
 839           reconnect_later (h);
 840           return;
 841         }
 842       m = (const struct InitReplyMessage *) msg;
 843       GNUNET_break (0 == ntohl (m->reserved));
 844       /* start our message processing loop */
 845       if (GNUNET_YES == h->currently_down)
 846         {
 847           h->currently_down = GNUNET_NO;
 848           trigger_next_request (h, GNUNET_NO);
 849         }
 850       h->retry_backoff = GNUNET_TIME_UNIT_MILLISECONDS;
 851       if (NULL != (init = h->init))
 852         {
 853           /* mark so we don't call init on reconnect */
 854           h->init = NULL;
 855           GNUNET_CRYPTO_hash (&m->publicKey,
 856                               sizeof (struct
 857                                       GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
 858                               &h->me.hashPubKey);
 859 #if DEBUG_CORE
 860           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 861                       "Connected to core service of peer `%s'.\n",
 862                       GNUNET_i2s (&h->me));
 863 #endif
 864           init (h->cls, h, &h->me, &m->publicKey);
 865         }
 866       else
 867         {
 868 #if DEBUG_CORE
 869           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 870                       "Successfully reconnected to core service.\n");
 871 #endif
 872         }
 873       break;
 874     case GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT:
 875       if (msize != sizeof (struct ConnectNotifyMessage))
 876         {
 877           GNUNET_break (0);
 878           reconnect_later (h);
 879           return;
 880         }
 881       cnm = (const struct ConnectNotifyMessage *) msg;
 882 #if DEBUG_CORE
 883       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 884                   "Received notification about connection from `%s'.\n",
 885                   GNUNET_i2s (&cnm->peer));
 886 #endif
 887       if (0 == memcmp (&h->me,
 888                        &cnm->peer,
 889                        sizeof (struct GNUNET_PeerIdentity)))
 890         {
 891           /* disconnect from self!? */
 892           GNUNET_break (0);
 893           return;
 894         }
 895       pr = GNUNET_CONTAINER_multihashmap_get (h->peers,
 896                                               &cnm->peer.hashPubKey);
 897       if (pr != NULL)
 898         {
 899           GNUNET_break (0);
 900           reconnect_later (h);
 901           return;
 902         }
 903       pr = GNUNET_malloc (sizeof (struct PeerRecord));
 904       pr->peer = cnm->peer;
 905       pr->ch = h;
 906       GNUNET_assert (GNUNET_YES ==
 907                      GNUNET_CONTAINER_multihashmap_put (h->peers,
 908                                                         &cnm->peer.hashPubKey,
 909                                                         pr,
 910                                                         GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
 911       if (NULL != h->connects)
 912         h->connects (h->cls,
 913                      &cnm->peer,
 914                      NULL /* FIXME: atsi! */);
 915       break;
 916     case GNUNET_MESSAGE_TYPE_CORE_NOTIFY_DISCONNECT:
 917       if (msize != sizeof (struct DisconnectNotifyMessage))
 918         {
 919           GNUNET_break (0);
 920           reconnect_later (h);
 921           return;
 922         }
 923       dnm = (const struct DisconnectNotifyMessage *) msg;
 924       if (0 == memcmp (&h->me,
 925                        &dnm->peer,
 926                        sizeof (struct GNUNET_PeerIdentity)))
 927         {
 928           /* connection to self!? */
 929           GNUNET_break (0);
 930           return;
 931         }
 932 #if DEBUG_CORE
 933       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 934                   "Received notification about disconnect from `%s'.\n",
 935                   GNUNET_i2s (&dnm->peer));
 936 #endif
 937       pr = GNUNET_CONTAINER_multihashmap_get (h->peers,
 938                                               &dnm->peer.hashPubKey);
 939       if (pr == NULL)
 940         {
 941           GNUNET_break (0);
 942           reconnect_later (h);
 943           return;
 944         }
 945       trigger = ( (pr->prev != NULL) ||
 946                   (pr->next != NULL) ||
 947                   (h->ready_peer_head == pr) );
 948       disconnect_and_free_peer_entry (h, &dnm->peer.hashPubKey, pr);
 949       if (trigger)
 950         trigger_next_request (h, GNUNET_NO);
 951       break;
 952     case GNUNET_MESSAGE_TYPE_CORE_NOTIFY_STATUS_CHANGE:
 953       if (NULL == h->status_events)
 954         {
 955           GNUNET_break (0);
 956           break;
 957         }
 958       if (msize != sizeof (struct PeerStatusNotifyMessage))
 959         {
 960           GNUNET_break (0);
 961           reconnect_later (h);
 962           return;
 963         }
 964       psnm = (const struct PeerStatusNotifyMessage *) msg;
 965       if (0 == memcmp (&h->me,
 966                        &psnm->peer,
 967                        sizeof (struct GNUNET_PeerIdentity)))
 968         {
 969           /* self-change!? */
 970           GNUNET_break (0);
 971           return;
 972         }
 973 #if DEBUG_CORE
 974       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 975                   "Received notification about status change by `%s'.\n",
 976                   GNUNET_i2s (&psnm->peer));
 977 #endif
 978       pr = GNUNET_CONTAINER_multihashmap_get (h->peers,
 979                                               &psnm->peer.hashPubKey);
 980       if (pr == NULL)
 981         {
 982           GNUNET_break (0);
 983           reconnect_later (h);
 984           return;
 985         }
 986       h->status_events (h->cls,
 987                         &psnm->peer,
 988                         psnm->bandwidth_in,
 989                         psnm->bandwidth_out,
 990                         GNUNET_TIME_absolute_ntoh (psnm->timeout),
 991                         NULL /* FIXME: atsi */);
 992       break;
 993     case GNUNET_MESSAGE_TYPE_CORE_NOTIFY_INBOUND:
 994       if (msize <
 995           sizeof (struct NotifyTrafficMessage) +
 996           sizeof (struct GNUNET_MessageHeader))
 997         {
 998           GNUNET_break (0);
 999           reconnect_later (h);
1000           return;
1001         }
1002       ntm = (const struct NotifyTrafficMessage *) msg;
1003       if (0 == memcmp (&h->me,
1004                        &ntm->peer,
1005                        sizeof (struct GNUNET_PeerIdentity)))
1006         {
1007           /* self-change!? */
1008           GNUNET_break (0);
1009           return;
1010         }
1011       em = (const struct GNUNET_MessageHeader *) &ntm[1];
1012 #if DEBUG_CORE
1013       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1014                   "Received message of type %u and size %u from peer `%4s'\n",
1015                   ntohs (em->type),
1016                   ntohs (em->size),
1017                   GNUNET_i2s (&ntm->peer));
1018 #endif
1019       pr = GNUNET_CONTAINER_multihashmap_get (h->peers,
1020                                               &ntm->peer.hashPubKey);
1021       if (pr == NULL)
1022         {
1023           GNUNET_break (0);
1024           reconnect_later (h);
1025           return;
1026         }
1027       if ((GNUNET_NO == h->inbound_hdr_only) &&
1028           (msize != ntohs (em->size) + sizeof (struct NotifyTrafficMessage)))
1029         {
1030           GNUNET_break (0);
1031           reconnect_later (h);
1032           return;
1033         }
1034       et = ntohs (em->type);
1035       for (hpos = 0; hpos < h->hcnt; hpos++)
1036         {
1037           mh = &h->handlers[hpos];
1038           if (mh->type != et)
1039             continue;
1040           if ((mh->expected_size != ntohs (em->size)) &&
1041               (mh->expected_size != 0))
1042             {
1043               GNUNET_break (0);
1044               continue;
1045             }
1046           if (GNUNET_OK !=
1047               h->handlers[hpos].callback (h->cls, &ntm->peer, em,
1048                                           NULL /* FIXME: atsi */))
1049             {
1050               /* error in processing, do not process other messages! */
1051               break;
1052             }
1053         }
1054       if (NULL != h->inbound_notify)
1055         h->inbound_notify (h->cls, &ntm->peer, em,
1056                            NULL /* FIXME: atsi */);
1057       break;
1058     case GNUNET_MESSAGE_TYPE_CORE_NOTIFY_OUTBOUND:
1059       if (msize <
1060           sizeof (struct NotifyTrafficMessage) +
1061           sizeof (struct GNUNET_MessageHeader))
1062         {
1063           GNUNET_break (0);
1064           reconnect_later (h);
1065           return;
1066         }
1067       ntm = (const struct NotifyTrafficMessage *) msg;
1068       if (0 == memcmp (&h->me,
1069                        &ntm->peer,
1070                        sizeof (struct GNUNET_PeerIdentity)))
1071         {
1072           /* self-change!? */
1073           GNUNET_break (0);
1074           return;
1075         }
1076       em = (const struct GNUNET_MessageHeader *) &ntm[1];
1077       pr = GNUNET_CONTAINER_multihashmap_get (h->peers,
1078                                               &ntm->peer.hashPubKey);
1079       if (pr == NULL)
1080         {
1081           GNUNET_break (0);
1082           reconnect_later (h);
1083           return;
1084         }
1085 #if DEBUG_CORE
1086       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1087                   "Received notification about transmission to `%s'.\n",
1088                   GNUNET_i2s (&ntm->peer));
1089 #endif
1090       if ((GNUNET_NO == h->outbound_hdr_only) &&
1091           (msize != ntohs (em->size) + sizeof (struct NotifyTrafficMessage)))
1092         {
1093           GNUNET_break (0);
1094           reconnect_later (h);
1095           return;
1096         }
1097       if (NULL == h->outbound_notify)
1098         {
1099           GNUNET_break (0);
1100           break;
1101         }
1102       h->outbound_notify (h->cls, &ntm->peer, em,
1103                           NULL /* FIXME: atsi? */);
1104       break;
1105     case GNUNET_MESSAGE_TYPE_CORE_SEND_READY:
1106       if (msize != sizeof (struct SendMessageReady))
1107         {
1108           GNUNET_break (0);
1109           reconnect_later (h);
1110           return;
1111         }
1112       smr = (const struct SendMessageReady *) msg;
1113       if (0 == memcmp (&h->me,
1114                        &smr->peer,
1115                        sizeof (struct GNUNET_PeerIdentity)))
1116         {
1117           /* self-change!? */
1118           GNUNET_break (0);
1119           return;
1120         }
1121       pr = GNUNET_CONTAINER_multihashmap_get (h->peers,
1122                                               &smr->peer.hashPubKey);
1123       if (pr == NULL)
1124         {
1125           GNUNET_break (0);
1126           reconnect_later (h);
1127           return;
1128         }
1129 #if DEBUG_CORE
1130       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1131                   "Received notification about transmission readiness to `%s'.\n",
1132                   GNUNET_i2s (&smr->peer));
1133 #endif
1134       th = pr->pending_head;
1135       if (ntohs (smr->smr_id) != th->smr_id)
1136         {
1137           /* READY message is for expired or cancelled message,
1138              ignore! (we should have already sent another request) */
1139           break;
1140         }
1141       if ( (pr->prev != NULL) ||
1142            (pr->next != NULL) ||
1143            (h->ready_peer_head == pr) )
1144         {
1145           /* we should not already be on the ready list... */
1146           GNUNET_break (0);
1147           reconnect_later (h);
1148           return;
1149         }
1150       GNUNET_CONTAINER_DLL_insert (h->ready_peer_head,
1151                                    h->ready_peer_tail,
1152                                    pr);
1153       trigger_next_request (h, GNUNET_NO);
1154       break;
1155     case GNUNET_MESSAGE_TYPE_CORE_CONFIGURATION_INFO:
1156       if (ntohs (msg->size) != sizeof (struct ConfigurationInfoMessage))
1157         {
1158           GNUNET_break (0);
1159           reconnect_later (h);
1160           return;
1161         }
1162       cim = (const struct ConfigurationInfoMessage*) msg;
1163       if (0 == memcmp (&h->me,
1164                        &cim->peer,
1165                        sizeof (struct GNUNET_PeerIdentity)))
1166         {
1167           /* self-change!? */
1168           GNUNET_break (0);
1169           return;
1170         }
1171 #if DEBUG_CORE
1172       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1173                   "Received notification about configuration update for `%s'.\n",
1174                   GNUNET_i2s (&cim->peer));
1175 #endif
1176       pr = GNUNET_CONTAINER_multihashmap_get (h->peers,
1177                                               &cim->peer.hashPubKey);
1178       if (pr == NULL)
1179         {
1180           GNUNET_break (0);
1181           reconnect_later (h);
1182           return;
1183         }
1184       if (pr->rim_id != ntohl (cim->rim_id))
1185         break;
1186       pcic = pr->pcic;
1187       pr->pcic = NULL;
1188       if (pcic != NULL)
1189         pcic (pr->pcic_cls,
1190               &pr->peer,
1191               cim->bw_out,
1192               ntohl (cim->reserved_amount),
1193               GNUNET_ntohll (cim->preference));
1194       break;
1195     default:
1196       reconnect_later (h);
1197       return;
1198     }
1199   GNUNET_CLIENT_receive (h->client,
1200                          &main_notify_handler, h,
1201                          GNUNET_TIME_UNIT_FOREVER_REL);
1202 }
1203
1204
1205 /**
1206  * Task executed once we are done transmitting the INIT message.
1207  * Starts our 'receive' loop.
1208  *
1209  * @param cls the 'struct GNUNET_CORE_Handle'
1210  * @param tc task context
1211  */
1212 static void
1213 init_done_task (void *cls,
1214                 const struct GNUNET_SCHEDULER_TaskContext *tc)
1215 {
1216   struct GNUNET_CORE_Handle *h = cls;
1217
1218   if (tc == NULL)
1219     return; /* error */
1220   if (0 == (tc->reason & GNUNET_SCHEDULER_REASON_PREREQ_DONE))
1221     {
1222 #if DEBUG_CORE
1223       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1224                   "Failed to exchange INIT with core, retrying\n");
1225 #endif
1226       reconnect_later (h);
1227       return;
1228     }
1229   GNUNET_CLIENT_receive (h->client,
1230                          &main_notify_handler,
1231                          h,
1232                          GNUNET_TIME_UNIT_FOREVER_REL);
1233 }
1234
1235
1236 /**
1237  * Our current client connection went down.  Clean it up
1238  * and try to reconnect!
1239  *
1240  * @param h our handle to the core service
1241  */
1242 static void
1243 reconnect (struct GNUNET_CORE_Handle *h)
1244 {
1245   struct ControlMessage *cm;
1246   struct InitMessage *init;
1247   uint32_t opt;
1248   uint16_t msize;
1249   uint16_t *ts;
1250   unsigned int hpos;
1251
1252 #if DEBUG_CORE
1253   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1254               "Reconnecting to CORE service\n");
1255 #endif
1256   GNUNET_assert (h->client == NULL);
1257   GNUNET_assert (h->currently_down == GNUNET_YES);
1258   h->client = GNUNET_CLIENT_connect ("core", h->cfg);
1259   if (h->client == NULL)
1260     {
1261       reconnect_later (h);
1262       return;
1263     }
1264   msize = h->hcnt * sizeof (uint16_t) + sizeof (struct InitMessage);
1265   cm = GNUNET_malloc (sizeof (struct ControlMessage) +
1266                       msize);
1267   cm->cont = &init_done_task;
1268   cm->cont_cls = h;
1269   init = (struct InitMessage*) &cm[1];
1270   init->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_INIT);
1271   init->header.size = htons (msize);
1272   opt = GNUNET_CORE_OPTION_SEND_CONNECT | GNUNET_CORE_OPTION_SEND_DISCONNECT;
1273   if (h->status_events != NULL)
1274     opt |= GNUNET_CORE_OPTION_SEND_STATUS_CHANGE;
1275   if (h->inbound_notify != NULL)
1276     {
1277       if (h->inbound_hdr_only)
1278         opt |= GNUNET_CORE_OPTION_SEND_HDR_INBOUND;
1279       else
1280         opt |= GNUNET_CORE_OPTION_SEND_FULL_INBOUND;
1281     }
1282   if (h->outbound_notify != NULL)
1283     {
1284       if (h->outbound_hdr_only)
1285         opt |= GNUNET_CORE_OPTION_SEND_HDR_OUTBOUND;
1286       else
1287         opt |= GNUNET_CORE_OPTION_SEND_FULL_OUTBOUND;
1288     }
1289   init->options = htonl (opt);
1290   ts = (uint16_t *) &init[1];
1291   for (hpos = 0; hpos < h->hcnt; hpos++)
1292     ts[hpos] = htons (h->handlers[hpos].type);
1293   GNUNET_CONTAINER_DLL_insert (h->pending_head,
1294                                h->pending_tail,
1295                                cm);
1296   trigger_next_request (h, GNUNET_YES);
1297 }
1298
1299
1300
1301 /**
1302  * Connect to the core service.  Note that the connection may
1303  * complete (or fail) asynchronously.
1304  *
1305  * @param cfg configuration to use
1306  * @param queue_size size of the per-peer message queue
1307  * @param cls closure for the various callbacks that follow (including handlers in the handlers array)
1308  * @param init callback to call on timeout or once we have successfully
1309  *        connected to the core service; note that timeout is only meaningful if init is not NULL
1310  * @param connects function to call on peer connect, can be NULL
1311  * @param disconnects function to call on peer disconnect / timeout, can be NULL
1312  * @param status_events function to call on changes to peer connection status, can be NULL
1313  * @param inbound_notify function to call for all inbound messages, can be NULL
1314  * @param inbound_hdr_only set to GNUNET_YES if inbound_notify will only read the
1315  *                GNUNET_MessageHeader and hence we do not need to give it the full message;
1316  *                can be used to improve efficiency, ignored if inbound_notify is NULLL
1317  * @param outbound_notify function to call for all outbound messages, can be NULL
1318  * @param outbound_hdr_only set to GNUNET_YES if outbound_notify will only read the
1319  *                GNUNET_MessageHeader and hence we do not need to give it the full message
1320  *                can be used to improve efficiency, ignored if outbound_notify is NULLL
1321  * @param handlers callbacks for messages we care about, NULL-terminated
1322  * @return handle to the core service (only useful for disconnect until 'init' is called);
1323  *                NULL on error (in this case, init is never called)
1324  */
1325 struct GNUNET_CORE_Handle *
1326 GNUNET_CORE_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
1327                      unsigned int queue_size,
1328                      void *cls,
1329                      GNUNET_CORE_StartupCallback init,
1330                      GNUNET_CORE_ConnectEventHandler connects,
1331                      GNUNET_CORE_DisconnectEventHandler disconnects,
1332                      GNUNET_CORE_PeerStatusEventHandler status_events,
1333                      GNUNET_CORE_MessageCallback inbound_notify,
1334                      int inbound_hdr_only,
1335                      GNUNET_CORE_MessageCallback outbound_notify,
1336                      int outbound_hdr_only,
1337                      const struct GNUNET_CORE_MessageHandler *handlers)
1338 {
1339   struct GNUNET_CORE_Handle *h;
1340
1341   h = GNUNET_malloc (sizeof (struct GNUNET_CORE_Handle));
1342   h->cfg = cfg;
1343   h->queue_size = queue_size;
1344   h->cls = cls;
1345   h->init = init;
1346   h->connects = connects;
1347   h->disconnects = disconnects;
1348   h->status_events = status_events;
1349   h->inbound_notify = inbound_notify;
1350   h->outbound_notify = outbound_notify;
1351   h->inbound_hdr_only = inbound_hdr_only;
1352   h->outbound_hdr_only = outbound_hdr_only;
1353   h->handlers = handlers;
1354   h->hcnt = 0;
1355   h->currently_down = GNUNET_YES;
1356   h->peers = GNUNET_CONTAINER_multihashmap_create (128);
1357   h->retry_backoff = GNUNET_TIME_UNIT_MILLISECONDS;
1358   while (handlers[h->hcnt].callback != NULL)
1359     h->hcnt++;
1360   GNUNET_assert (h->hcnt <
1361                  (GNUNET_SERVER_MAX_MESSAGE_SIZE -
1362                   sizeof (struct InitMessage)) / sizeof (uint16_t));
1363 #if DEBUG_CORE
1364   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1365               "Connecting to CORE service\n");
1366 #endif
1367   reconnect (h);
1368   return h;
1369 }
1370
1371
1372 /**
1373  * Disconnect from the core service.  This function can only
1374  * be called *after* all pending 'GNUNET_CORE_notify_transmit_ready'
1375  * requests have been explicitly cancelled.
1376  *
1377  * @param handle connection to core to disconnect
1378  */
1379 void
1380 GNUNET_CORE_disconnect (struct GNUNET_CORE_Handle *handle)
1381 {
1382   struct ControlMessage *cm;
1383
1384 #if DEBUG_CORE
1385   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1386               "Disconnecting from CORE service\n");
1387 #endif
1388   if (handle->cth != NULL)
1389     {
1390       GNUNET_CLIENT_notify_transmit_ready_cancel (handle->cth);
1391       handle->cth = NULL;
1392     }
1393   if (handle->client != NULL)
1394     {
1395       GNUNET_CLIENT_disconnect (handle->client, GNUNET_NO);
1396       handle->client = NULL;
1397     }
1398   if (handle->reconnect_task != GNUNET_SCHEDULER_NO_TASK)
1399     {
1400       GNUNET_SCHEDULER_cancel (handle->reconnect_task);
1401       handle->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
1402     }
1403   while (NULL != (cm = handle->pending_head))
1404     {
1405       GNUNET_CONTAINER_DLL_remove (handle->pending_head,
1406                                    handle->pending_tail,
1407                                    cm);
1408       cm->cont (cm->cont_cls, NULL);
1409       GNUNET_free (cm);
1410     }
1411   GNUNET_CONTAINER_multihashmap_iterate (handle->peers,
1412                                          &disconnect_and_free_peer_entry,
1413                                          handle);
1414   GNUNET_CONTAINER_multihashmap_destroy (handle->peers);
1415   GNUNET_break (handle->ready_peer_head == NULL);
1416   GNUNET_free (handle);
1417 }
1418
1419
1420 /**
1421  * Ask the core to call "notify" once it is ready to transmit the
1422  * given number of bytes to the specified "target".  If we are not yet
1423  * connected to the specified peer, a call to this function will cause
1424  * us to try to establish a connection.
1425  *
1426  * @param handle connection to core service
1427  * @param priority how important is the message?
1428  * @param maxdelay how long can the message wait?
1429  * @param target who should receive the message,
1430  *        use NULL for this peer (loopback)
1431  * @param notify_size how many bytes of buffer space does notify want?
1432  * @param notify function to call when buffer space is available
1433  * @param notify_cls closure for notify
1434  * @return non-NULL if the notify callback was queued,
1435  *         NULL if we can not even queue the request (insufficient
1436  *         memory); if NULL is returned, "notify" will NOT be called.
1437  */
1438 struct GNUNET_CORE_TransmitHandle *
1439 GNUNET_CORE_notify_transmit_ready (struct GNUNET_CORE_Handle *handle,
1440                                    uint32_t priority,
1441                                    struct GNUNET_TIME_Relative maxdelay,
1442                                    const struct GNUNET_PeerIdentity *target,
1443                                    size_t notify_size,
1444                                    GNUNET_CONNECTION_TransmitReadyNotify notify,
1445                                    void *notify_cls)
1446 {
1447   struct PeerRecord *pr;
1448   struct GNUNET_CORE_TransmitHandle *th;
1449   struct GNUNET_CORE_TransmitHandle *pos;
1450   struct GNUNET_CORE_TransmitHandle *prev;
1451   struct GNUNET_CORE_TransmitHandle *minp;
1452
1453   pr = GNUNET_CONTAINER_multihashmap_get (handle->peers,
1454                                           &target->hashPubKey);
1455   if (NULL == pr)
1456     {
1457       /* attempt to send to peer that is not connected */
1458       GNUNET_log(GNUNET_ERROR_TYPE_WARNING,
1459                  "Attempting to send to peer `%s' from peer `%s', but not connected!\n",
1460                  GNUNET_i2s(target), GNUNET_h2s(&handle->me.hashPubKey));
1461       GNUNET_break (0);
1462       return NULL;
1463     }
1464   GNUNET_assert (notify_size + sizeof (struct SendMessage) <
1465                  GNUNET_SERVER_MAX_MESSAGE_SIZE);
1466   th = GNUNET_malloc (sizeof (struct GNUNET_CORE_TransmitHandle));
1467   th->peer = pr;
1468   th->get_message = notify;
1469   th->get_message_cls = notify_cls;
1470   th->timeout = GNUNET_TIME_relative_to_absolute (maxdelay);
1471   th->priority = priority;
1472   th->msize = notify_size;
1473   /* bound queue size */
1474   if (pr->queue_size == handle->queue_size)
1475     {
1476       /* find lowest-priority entry */
1477       minp = pr->pending_head;
1478       prev = minp->next;
1479       while (prev != NULL)
1480         {
1481           if (prev->priority < minp->priority)
1482             minp = prev;
1483           prev = prev->next;
1484         }
1485       if (minp == NULL)
1486         {
1487           GNUNET_break (handle->queue_size != 0);
1488           GNUNET_break (pr->queue_size == 0);
1489           return NULL;
1490         }
1491       if (priority <= minp->priority)
1492         {
1493 #if DEBUG_CORE
1494           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1495                       "Dropping transmission request: priority too low\n");
1496 #endif
1497           return NULL; /* priority too low */
1498         }
1499       GNUNET_CONTAINER_DLL_remove (pr->pending_head,
1500                                    pr->pending_tail,
1501                                    minp);
1502       pr->queue_size--;
1503       GNUNET_assert (0 ==
1504                      minp->get_message (minp->get_message_cls,
1505                                         0, NULL));
1506       GNUNET_free (minp);
1507     }
1508
1509   /* Order entries by deadline, but SKIP 'HEAD' if
1510      we're in the 'ready_peer_*' DLL */
1511   pos = pr->pending_head;
1512   if ( (pr->prev != NULL) ||
1513        (pr->next != NULL) ||
1514        (pr == handle->ready_peer_head) )
1515     {
1516       GNUNET_assert (pos != NULL);
1517       pos = pos->next; /* skip head */
1518     }
1519
1520   /* insertion sort */
1521   prev = pos;
1522   while ( (pos != NULL) &&
1523           (pos->timeout.abs_value < th->timeout.abs_value) )
1524     {
1525       prev = pos;
1526       pos = pos->next;
1527     }
1528   GNUNET_CONTAINER_DLL_insert_after (pr->pending_head,
1529                                      pr->pending_tail,
1530                                      prev,
1531                                      th);
1532   pr->queue_size++;
1533   /* was the request queue previously empty? */
1534 #if DEBUG_CORE
1535   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1536               "Transmission request added to queue\n");
1537 #endif
1538   if (pr->pending_head == th)
1539     request_next_transmission (pr);
1540   return th;
1541 }
1542
1543
1544 /**
1545  * Cancel the specified transmission-ready notification.
1546  *
1547  * @param th handle that was returned by "notify_transmit_ready".
1548  */
1549 void
1550 GNUNET_CORE_notify_transmit_ready_cancel (struct GNUNET_CORE_TransmitHandle
1551                                           *th)
1552 {
1553   struct PeerRecord *pr = th->peer;
1554   struct GNUNET_CORE_Handle *h = pr->ch;
1555   int was_head;
1556
1557   was_head = (pr->pending_head == th);
1558   GNUNET_CONTAINER_DLL_remove (pr->pending_head,
1559                                pr->pending_tail,
1560                                th);
1561   pr->queue_size--;
1562   if (th->cm != NULL)
1563     {
1564       /* we're currently in the control queue, remove */
1565       GNUNET_CONTAINER_DLL_remove (h->pending_head,
1566                                    h->pending_tail,
1567                                    th->cm);
1568       GNUNET_free (th->cm);
1569     }
1570   GNUNET_free (th);
1571   if (was_head)
1572     {
1573       if ( (pr->prev != NULL) ||
1574            (pr->next != NULL) ||
1575            (pr == h->ready_peer_head) )
1576         {
1577           /* the request that was 'approved' by core was
1578              cancelled before it could be transmitted; remove
1579              us from the 'ready' list */
1580           GNUNET_CONTAINER_DLL_remove (h->ready_peer_head,
1581                                        h->ready_peer_tail,
1582                                        pr);
1583         }
1584       request_next_transmission (pr);
1585     }
1586 }
1587
1588
1589 /* ****************** GNUNET_CORE_peer_request_connect ******************** */
1590
1591 /**
1592  * Handle for a request to the core to connect to
1593  * a particular peer.  Can be used to cancel the request
1594  * (before the 'cont'inuation is called).
1595  */
1596 struct GNUNET_CORE_PeerRequestHandle
1597 {
1598
1599   /**
1600    * Link to control message.
1601    */
1602   struct ControlMessage *cm;
1603
1604   /**
1605    * Core handle used.
1606    */
1607   struct GNUNET_CORE_Handle *h;
1608
1609   /**
1610    * Continuation to run when done.
1611    */
1612   GNUNET_SCHEDULER_Task cont;
1613
1614   /**
1615    * Closure for 'cont'.
1616    */
1617   void *cont_cls;
1618
1619 };
1620
1621
1622 /**
1623  * Continuation called when the control message was transmitted.
1624  * Calls the original continuation and frees the remaining
1625  * resources.
1626  *
1627  * @param cls the 'struct GNUNET_CORE_PeerRequestHandle'
1628  * @param tc scheduler context
1629  */
1630 static void
1631 peer_request_connect_cont (void *cls,
1632                            const struct GNUNET_SCHEDULER_TaskContext *tc)
1633 {
1634   struct GNUNET_CORE_PeerRequestHandle *ret = cls;
1635
1636   if (ret->cont != NULL)
1637     {
1638       if (tc == NULL)
1639         GNUNET_SCHEDULER_add_now (ret->cont,
1640                                   ret->cont_cls);
1641       else
1642         ret->cont (ret->cont_cls, tc);
1643     }
1644   GNUNET_free (ret);
1645 }
1646
1647
1648 /**
1649  * Request that the core should try to connect to a particular peer.
1650  * Once the request has been transmitted to the core, the continuation
1651  * function will be called.  Note that this does NOT mean that a
1652  * connection was successfully established -- it only means that the
1653  * core will now try.  Successful establishment of the connection
1654  * will be signalled to the 'connects' callback argument of
1655  * 'GNUNET_CORE_connect' only.  If the core service does not respond
1656  * to our connection attempt within the given time frame, 'cont' will
1657  * be called with the TIMEOUT reason code.
1658  *
1659  * @param h core handle
1660  * @param timeout how long to try to talk to core
1661  * @param peer who should we connect to
1662  * @param cont function to call once the request has been completed (or timed out)
1663  * @param cont_cls closure for cont
1664  * @return NULL on error (cont will not be called), otherwise handle for cancellation
1665  */
1666 struct GNUNET_CORE_PeerRequestHandle *
1667 GNUNET_CORE_peer_request_connect (struct GNUNET_CORE_Handle *h,
1668                                   struct GNUNET_TIME_Relative timeout,
1669                                   const struct GNUNET_PeerIdentity * peer,
1670                                   GNUNET_SCHEDULER_Task cont,
1671                                   void *cont_cls)
1672 {
1673   struct GNUNET_CORE_PeerRequestHandle *ret;
1674   struct ControlMessage *cm;
1675   struct ConnectMessage *msg;
1676
1677   cm = GNUNET_malloc (sizeof (struct ControlMessage) +
1678                       sizeof (struct ConnectMessage));
1679   msg = (struct ConnectMessage*) &cm[1];
1680   msg->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_REQUEST_CONNECT);
1681   msg->header.size = htons (sizeof (struct ConnectMessage));
1682   msg->reserved = htonl (0);
1683   msg->timeout = GNUNET_TIME_relative_hton (timeout);
1684   msg->peer = *peer;
1685   GNUNET_CONTAINER_DLL_insert (h->pending_head,
1686                                h->pending_tail,
1687                                cm);
1688   ret = GNUNET_malloc (sizeof (struct GNUNET_CORE_PeerRequestHandle));
1689   ret->h = h;
1690   ret->cm = cm;
1691   ret->cont = cont;
1692   ret->cont_cls = cont_cls;
1693   cm->cont = &peer_request_connect_cont;
1694   cm->cont_cls = ret;
1695 #if DEBUG_CORE
1696   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1697               "Queueing REQUEST_CONNECT request\n");
1698 #endif
1699   if (h->pending_head == cm)
1700     trigger_next_request (h, GNUNET_NO);
1701   return ret;
1702 }
1703
1704
1705 /**
1706  * Cancel a pending request to connect to a particular peer.  Must not
1707  * be called after the 'cont' function was invoked.
1708  *
1709  * @param req request handle that was returned for the original request
1710  */
1711 void
1712 GNUNET_CORE_peer_request_connect_cancel (struct GNUNET_CORE_PeerRequestHandle *req)
1713 {
1714   struct GNUNET_CORE_Handle *h = req->h;
1715   struct ControlMessage *cm = req->cm;
1716
1717   GNUNET_CONTAINER_DLL_remove (h->pending_head,
1718                                h->pending_tail,
1719                                cm);
1720   GNUNET_free (cm);
1721   GNUNET_free (req);
1722 }
1723
1724
1725 /* ****************** GNUNET_CORE_peer_change_preference ******************** */
1726
1727
1728 struct GNUNET_CORE_InformationRequestContext
1729 {
1730
1731   /**
1732    * Our connection to the service.
1733    */
1734   struct GNUNET_CORE_Handle *h;
1735
1736   /**
1737    * Function to call with the information.
1738    */
1739   GNUNET_CORE_PeerConfigurationInfoCallback info;
1740
1741   /**
1742    * Closure for info.
1743    */
1744   void *info_cls;
1745
1746   /**
1747    * Link to control message, NULL if CM was sent.
1748    */
1749   struct ControlMessage *cm;
1750
1751   /**
1752    * Link to peer record.
1753    */
1754   struct PeerRecord *pr;
1755 };
1756
1757
1758 /**
1759  * CM was sent, remove link so we don't double-free.
1760  *
1761  * @param cls the 'struct GNUNET_CORE_InformationRequestContext'
1762  * @param tc scheduler context
1763  */
1764 static void
1765 change_preference_send_continuation (void *cls,
1766                                      const struct GNUNET_SCHEDULER_TaskContext *tc)
1767 {
1768   struct GNUNET_CORE_InformationRequestContext *irc = cls;
1769
1770   irc->cm = NULL;
1771 }
1772
1773
1774 /**
1775  * Obtain statistics and/or change preferences for the given peer.
1776  *
1777  * @param h core handle
1778  * @param peer identifies the peer
1779  * @param timeout after how long should we give up (and call "info" with NULL
1780  *                for "peer" to signal an error)?
1781  * @param bw_out set to the current bandwidth limit (sending) for this peer,
1782  *                caller should set "bw_out" to "-1" to avoid changing
1783  *                the current value; otherwise "bw_out" will be lowered to
1784  *                the specified value; passing a pointer to "0" can be used to force
1785  *                us to disconnect from the peer; "bw_out" might not increase
1786  *                as specified since the upper bound is generally
1787  *                determined by the other peer!
1788  * @param amount reserve N bytes for receiving, negative
1789  *                amounts can be used to undo a (recent) reservation;
1790  * @param preference increase incoming traffic share preference by this amount;
1791  *                in the absence of "amount" reservations, we use this
1792  *                preference value to assign proportional bandwidth shares
1793  *                to all connected peers
1794  * @param info function to call with the resulting configuration information
1795  * @param info_cls closure for info
1796  * @return NULL on error
1797  */
1798 struct GNUNET_CORE_InformationRequestContext *
1799 GNUNET_CORE_peer_change_preference (struct GNUNET_CORE_Handle *h,
1800                                     const struct GNUNET_PeerIdentity *peer,
1801                                     struct GNUNET_TIME_Relative timeout,
1802                                     struct GNUNET_BANDWIDTH_Value32NBO bw_out,
1803                                     int32_t amount,
1804                                     uint64_t preference,
1805                                     GNUNET_CORE_PeerConfigurationInfoCallback info,
1806                                     void *info_cls)
1807 {
1808   struct GNUNET_CORE_InformationRequestContext *irc;
1809   struct PeerRecord *pr;
1810   struct RequestInfoMessage *rim;
1811   struct ControlMessage *cm;
1812
1813   pr = GNUNET_CONTAINER_multihashmap_get (h->peers,
1814                                           &peer->hashPubKey);
1815   if (NULL == pr)
1816     {
1817       /* attempt to change preference on peer that is not connected */
1818       GNUNET_break (0);
1819       return NULL;
1820     }
1821   if (pr->pcic != NULL)
1822     {
1823       /* second change before first one is done */
1824       GNUNET_break (0);
1825       return NULL;
1826     }
1827   irc = GNUNET_malloc (sizeof (struct GNUNET_CORE_InformationRequestContext));
1828   irc->h = h;
1829   irc->pr = pr;
1830   irc->info = info;
1831   irc->info_cls = info_cls;
1832   cm = GNUNET_malloc (sizeof (struct ControlMessage) +
1833                       sizeof (struct RequestInfoMessage));
1834   cm->cont = &change_preference_send_continuation;
1835   cm->cont_cls = irc;
1836   irc->cm = cm;
1837   rim = (struct RequestInfoMessage*) &cm[1];
1838   rim->header.size = htons (sizeof (struct RequestInfoMessage));
1839   rim->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_REQUEST_INFO);
1840   rim->rim_id = htonl (pr->rim_id = h->rim_id_gen++);
1841   rim->limit_outbound = bw_out;
1842   rim->reserve_inbound = htonl (amount);
1843   rim->preference_change = GNUNET_htonll(preference);
1844   rim->peer = *peer;
1845 #if DEBUG_CORE
1846   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1847               "Queueing CHANGE PREFERENCE request\n");
1848 #endif
1849   GNUNET_CONTAINER_DLL_insert (h->pending_head,
1850                                h->pending_tail,
1851                                cm);
1852   pr->pcic = info;
1853   pr->pcic_cls = info_cls;
1854   if (h->pending_head == cm)
1855     trigger_next_request (h, GNUNET_NO);
1856   return irc;
1857 }
1858
1859
1860 /**
1861  * Cancel request for getting information about a peer.
1862  * Note that an eventual change in preference, trust or bandwidth
1863  * assignment MAY have already been committed at the time,
1864  * so cancelling a request is NOT sure to undo the original
1865  * request.  The original request may or may not still commit.
1866  * The only thing cancellation ensures is that the callback
1867  * from the original request will no longer be called.
1868  *
1869  * @param irc context returned by the original GNUNET_CORE_peer_get_info call
1870  */
1871 void
1872 GNUNET_CORE_peer_change_preference_cancel (struct GNUNET_CORE_InformationRequestContext *irc)
1873 {
1874   struct GNUNET_CORE_Handle *h = irc->h;
1875   struct PeerRecord *pr = irc->pr;
1876
1877   if (irc->cm != NULL)
1878     {
1879       GNUNET_CONTAINER_DLL_remove (h->pending_head,
1880                                    h->pending_tail,
1881                                    irc->cm);
1882       GNUNET_free (irc->cm);
1883     }
1884   pr->pcic = NULL;
1885   pr->pcic_cls = NULL;
1886   GNUNET_free (irc);
1887 }
1888
1889
1890 #if NEW
1891 /* ********************* GNUNET_CORE_iterate_peers *********************** */
1892
1893 /**
1894  * Context for 'iterate_peers' helper function.
1895  */
1896 struct IterationContext
1897 {
1898   /**
1899    * Callback to call.
1900    */
1901   GNUNET_CORE_ConnectEventHandler peer_cb;
1902
1903   /**
1904    * Closure for 'peer_cb'.
1905    */
1906   void *cb_cls;
1907 };
1908
1909
1910 /**
1911  * Call callback for each peer.
1912  *
1913  * @param cls the 'struct IterationContext'
1914  * @param hc peer identity, not used
1915  * @param value the 'struct PeerRecord'
1916  * @return GNUNET_YES (continue iteration)
1917  */
1918 static int
1919 iterate_peers (void *cls,
1920                const GNUNET_HashCode *hc,
1921                void *value)
1922 {
1923   struct IterationContext *ic = cls;
1924   struct PeerRecord *pr = value;
1925
1926   ic->peer_cb (ic->cb_cls,
1927                &pr->peer,
1928                NULL /* FIXME: pass atsi? */);
1929   return GNUNET_YES;
1930 }
1931
1932
1933 /**
1934  * Obtain statistics and/or change preferences for the given peer.
1935  *
1936  * @param h handle to core
1937  * @param peer_cb function to call with the peer information
1938  * @param cb_cls closure for peer_cb
1939  * @return GNUNET_OK if iterating, GNUNET_SYSERR on error
1940  */
1941 int
1942 GNUNET_CORE_iterate_peers (struct GNUNET_CORE_Handle *h,
1943                            GNUNET_CORE_ConnectEventHandler peer_cb,
1944                            void *cb_cls)
1945 {
1946   struct IterationContext ic;
1947
1948   ic.peer_cb = peer_cb;
1949   ic.cb_cls = cb_cls;
1950   GNUNET_CONTAINER_multihashmap_iterate (h->peers,
1951                                          &iterate_peers,
1952                                          &ic);
1953   return GNUNET_OK;
1954 }
1955 #endif
1956
1957 /* end of core_api.c */