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