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