src/dht/gnunet-service-dht_neighbours.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009, 2010, 2011 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 dht/gnunet-service-dht_neighbours.c
  23  * @brief GNUnet DHT service's bucket and neighbour management code
  24  * @author Christian Grothoff
  25  * @author Nathan Evans
  26  */
  27
  28 #include "platform.h"
  29 #include "gnunet_block_lib.h"
  30 #include "gnunet_util_lib.h"
  31 #include "gnunet_hello_lib.h"
  32 #include "gnunet_constants.h"
  33 #include "gnunet_protocols.h"
  34 #include "gnunet_nse_service.h"
  35 #include "gnunet_core_service.h"
  36 #include "gnunet_datacache_lib.h"
  37 #include "gnunet_transport_service.h"
  38 #include "gnunet_hello_lib.h"
  39 #include "gnunet_dht_service.h"
  40 #include "gnunet_statistics_service.h"
  41 #include "gnunet-service-dht.h"
  42 #include "gnunet-service-dht_clients.h"
  43 #include "gnunet-service-dht_datacache.h"
  44 #include "gnunet-service-dht_hello.h"
  45 #include "gnunet-service-dht_neighbours.h"
  46 #include "gnunet-service-dht_nse.h"
  47 #include "gnunet-service-dht_routing.h"
  48 #include <fenv.h>
  49 #include "dht.h"
  50
  51 /**
  52  * How many buckets will we allow total.
  53  */
  54 #define MAX_BUCKETS sizeof (GNUNET_HashCode) * 8
  55
  56 /**
  57  * What is the maximum number of peers in a given bucket.
  58  */
  59 #define DEFAULT_BUCKET_SIZE 4
  60
  61 /**
  62  * Desired replication level for FIND PEER requests
  63  */
  64 #define FIND_PEER_REPLICATION_LEVEL 4
  65
  66 /**
  67  * Maximum allowed replication level for all requests.
  68  */
  69 #define MAXIMUM_REPLICATION_LEVEL 16
  70
  71 /**
  72  * How often to update our preference levels for peers in our routing tables.
  73  */
  74 #define DHT_DEFAULT_PREFERENCE_INTERVAL GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 2)
  75
  76 /**
  77  * How long at least to wait before sending another find peer request.
  78  */
  79 #define DHT_MINIMUM_FIND_PEER_INTERVAL GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS, 30)
  80
  81 /**
  82  * How long at most to wait before sending another find peer request.
  83  */
  84 #define DHT_MAXIMUM_FIND_PEER_INTERVAL GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 10)
  85
  86 /**
  87  * How long at most to wait for transmission of a GET request to another peer?
  88  */
  89 #define GET_TIMEOUT GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 2)
  90
  91
  92 /**
  93  * P2P PUT message
  94  */
  95 struct PeerPutMessage
  96 {
  97   /**
  98    * Type: GNUNET_MESSAGE_TYPE_DHT_P2P_PUT
  99    */
 100   struct GNUNET_MessageHeader header;
 101
 102   /**
 103    * Processing options
 104    */
 105   uint32_t options GNUNET_PACKED;
 106
 107   /**
 108    * Content type.
 109    */
 110   uint32_t type GNUNET_PACKED;
 111
 112   /**
 113    * Hop count
 114    */
 115   uint32_t hop_count GNUNET_PACKED;
 116
 117   /**
 118    * Replication level for this message
 119    */
 120   uint32_t desired_replication_level GNUNET_PACKED;
 121
 122   /**
 123    * Length of the PUT path that follows (if tracked).
 124    */
 125   uint32_t put_path_length GNUNET_PACKED;
 126
 127   /**
 128    * When does the content expire?
 129    */
 130   struct GNUNET_TIME_AbsoluteNBO expiration_time;
 131
 132   /**
 133    * Bloomfilter (for peer identities) to stop circular routes
 134    */
 135   char bloomfilter[DHT_BLOOM_SIZE];
 136
 137   /**
 138    * The key we are storing under.
 139    */
 140   GNUNET_HashCode key;
 141
 142   /* put path (if tracked) */
 143
 144   /* Payload */
 145
 146 };
 147
 148
 149 /**
 150  * P2P Result message
 151  */
 152 struct PeerResultMessage
 153 {
 154   /**
 155    * Type: GNUNET_MESSAGE_TYPE_DHT_P2P_RESULT
 156    */
 157   struct GNUNET_MessageHeader header;
 158
 159   /**
 160    * Content type.
 161    */
 162   uint32_t type GNUNET_PACKED;
 163
 164   /**
 165    * Length of the PUT path that follows (if tracked).
 166    */
 167   uint32_t put_path_length GNUNET_PACKED;
 168
 169   /**
 170    * Length of the GET path that follows (if tracked).
 171    */
 172   uint32_t get_path_length GNUNET_PACKED;
 173
 174   /**
 175    * When does the content expire?
 176    */
 177   struct GNUNET_TIME_AbsoluteNBO expiration_time;
 178
 179   /**
 180    * The key of the corresponding GET request.
 181    */
 182   GNUNET_HashCode key;
 183
 184   /* put path (if tracked) */
 185
 186   /* get path (if tracked) */
 187
 188   /* Payload */
 189
 190 };
 191
 192
 193 /**
 194  * P2P GET message
 195  */
 196 struct PeerGetMessage
 197 {
 198   /**
 199    * Type: GNUNET_MESSAGE_TYPE_DHT_P2P_GET
 200    */
 201   struct GNUNET_MessageHeader header;
 202
 203   /**
 204    * Processing options
 205    */
 206   uint32_t options GNUNET_PACKED;
 207
 208   /**
 209    * Desired content type.
 210    */
 211   uint32_t type GNUNET_PACKED;
 212
 213   /**
 214    * Hop count
 215    */
 216   uint32_t hop_count GNUNET_PACKED;
 217
 218   /**
 219    * Desired replication level for this request.
 220    */
 221   uint32_t desired_replication_level GNUNET_PACKED;
 222
 223   /**
 224    * Size of the extended query.
 225    */
 226   uint32_t xquery_size;
 227
 228   /**
 229    * Bloomfilter mutator.
 230    */
 231   uint32_t bf_mutator;
 232
 233   /**
 234    * Bloomfilter (for peer identities) to stop circular routes
 235    */
 236   char bloomfilter[DHT_BLOOM_SIZE];
 237
 238   /**
 239    * The key we are looking for.
 240    */
 241   GNUNET_HashCode key;
 242
 243   /* xquery */
 244
 245   /* result bloomfilter */
 246
 247 };
 248
 249
 250 /**
 251  * Linked list of messages to send to a particular other peer.
 252  */
 253 struct P2PPendingMessage
 254 {
 255   /**
 256    * Pointer to next item in the list
 257    */
 258   struct P2PPendingMessage *next;
 259
 260   /**
 261    * Pointer to previous item in the list
 262    */
 263   struct P2PPendingMessage *prev;
 264
 265   /**
 266    * Message importance level.  FIXME: used? useful?
 267    */
 268   unsigned int importance;
 269
 270   /**
 271    * When does this message time out?
 272    */
 273   struct GNUNET_TIME_Absolute timeout;
 274
 275   /**
 276    * Actual message to be sent, allocated at the end of the struct:
 277    * // msg = (cast) &pm[1];
 278    * // memcpy (&pm[1], data, len);
 279    */
 280   const struct GNUNET_MessageHeader *msg;
 281
 282 };
 283
 284
 285 /**
 286  * Entry for a peer in a bucket.
 287  */
 288 struct PeerInfo
 289 {
 290   /**
 291    * Next peer entry (DLL)
 292    */
 293   struct PeerInfo *next;
 294
 295   /**
 296    *  Prev peer entry (DLL)
 297    */
 298   struct PeerInfo *prev;
 299
 300   /**
 301    * Count of outstanding messages for peer.  FIXME: NEEDED?
 302    * FIXME: bound queue size!?
 303    */
 304   unsigned int pending_count;
 305
 306   /**
 307    * Head of pending messages to be sent to this peer.
 308    */
 309   struct P2PPendingMessage *head;
 310
 311   /**
 312    * Tail of pending messages to be sent to this peer.
 313    */
 314   struct P2PPendingMessage *tail;
 315
 316   /**
 317    * Core handle for sending messages to this peer.
 318    */
 319   struct GNUNET_CORE_TransmitHandle *th;
 320
 321   /**
 322    * Preference update context
 323    */
 324   struct GNUNET_CORE_InformationRequestContext *info_ctx;
 325
 326   /**
 327    * Task for scheduling preference updates
 328    */
 329   GNUNET_SCHEDULER_TaskIdentifier preference_task;
 330
 331   /**
 332    * What is the identity of the peer?
 333    */
 334   struct GNUNET_PeerIdentity id;
 335
 336 #if 0
 337   /**
 338    * What is the average latency for replies received?
 339    */
 340   struct GNUNET_TIME_Relative latency;
 341
 342   /**
 343    * Transport level distance to peer.
 344    */
 345   unsigned int distance;
 346 #endif
 347
 348 };
 349
 350
 351 /**
 352  * Peers are grouped into buckets.
 353  */
 354 struct PeerBucket
 355 {
 356   /**
 357    * Head of DLL
 358    */
 359   struct PeerInfo *head;
 360
 361   /**
 362    * Tail of DLL
 363    */
 364   struct PeerInfo *tail;
 365
 366   /**
 367    * Number of peers in the bucket.
 368    */
 369   unsigned int peers_size;
 370 };
 371
 372
 373 /**
 374  * The lowest currently used bucket, initially 0 (for 0-bits matching bucket).
 375  */
 376 static unsigned int closest_bucket;
 377
 378 /**
 379  * How many peers have we added since we sent out our last
 380  * find peer request?
 381  */
 382 static unsigned int newly_found_peers;
 383
 384 /**
 385  * The buckets.  Array of size MAX_BUCKET_SIZE.  Offset 0 means 0 bits matching.
 386  */
 387 static struct PeerBucket k_buckets[MAX_BUCKETS];
 388
 389 /**
 390  * Hash map of all known peers, for easy removal from k_buckets on disconnect.
 391  */
 392 static struct GNUNET_CONTAINER_MultiHashMap *all_known_peers;
 393
 394 /**
 395  * Maximum size for each bucket.
 396  */
 397 static unsigned int bucket_size = DEFAULT_BUCKET_SIZE;
 398
 399 /**
 400  * Task that sends FIND PEER requests.
 401  */
 402 static GNUNET_SCHEDULER_TaskIdentifier find_peer_task;
 403
 404 /**
 405  * Identity of this peer.
 406  */
 407 static struct GNUNET_PeerIdentity my_identity;
 408
 409 /**
 410  * Handle to GNUnet core.
 411  */
 412 static struct GNUNET_CORE_Handle *coreAPI;
 413
 414
 415
 416 /**
 417  * Find the optimal bucket for this key.
 418  *
 419  * @param hc the hashcode to compare our identity to
 420  * @return the proper bucket index, or GNUNET_SYSERR
 421  *         on error (same hashcode)
 422  */
 423 static int
 424 find_bucket (const GNUNET_HashCode * hc)
 425 {
 426   unsigned int bits;
 427
 428   bits = GNUNET_CRYPTO_hash_matching_bits (&my_identity.hashPubKey, hc);
 429   if (bits == MAX_BUCKETS)
 430     {
 431       /* How can all bits match? Got my own ID? */
 432       GNUNET_break (0);
 433       return GNUNET_SYSERR;
 434     }
 435   return MAX_BUCKETS - bits - 1;
 436 }
 437
 438
 439 /**
 440  * Let GNUnet core know that we like the given peer.
 441  *
 442  * @param cls the 'struct PeerInfo' of the peer
 443  * @param tc scheduler context.
 444  */
 445 static void
 446 update_core_preference (void *cls,
 447                         const struct GNUNET_SCHEDULER_TaskContext *tc);
 448
 449
 450 /**
 451  * Function called with statistics about the given peer.
 452  *
 453  * @param cls closure
 454  * @param peer identifies the peer
 455  * @param bpm_out set to the current bandwidth limit (sending) for this peer
 456  * @param amount set to the amount that was actually reserved or unreserved;
 457  *               either the full requested amount or zero (no partial reservations)
 458  * @param res_delay if the reservation could not be satisfied (amount was 0), how
 459  *        long should the client wait until re-trying?
 460  * @param preference current traffic preference for the given peer
 461  */
 462 static void
 463 update_core_preference_finish (void *cls,
 464                                const struct GNUNET_PeerIdentity *peer,
 465                                struct GNUNET_BANDWIDTH_Value32NBO bpm_out,
 466                                int32_t amount,
 467                                struct GNUNET_TIME_Relative res_delay,
 468                                uint64_t preference)
 469 {
 470   struct PeerInfo *peer_info = cls;
 471
 472   peer_info->info_ctx = NULL;
 473   peer_info->preference_task
 474     = GNUNET_SCHEDULER_add_delayed (DHT_DEFAULT_PREFERENCE_INTERVAL,
 475                                     &update_core_preference, peer_info);
 476 }
 477
 478
 479 /**
 480  * Let GNUnet core know that we like the given peer.
 481  *
 482  * @param cls the 'struct PeerInfo' of the peer
 483  * @param tc scheduler context.
 484  */
 485 static void
 486 update_core_preference (void *cls,
 487                         const struct GNUNET_SCHEDULER_TaskContext *tc)
 488 {
 489   struct PeerInfo *peer = cls;
 490   uint64_t preference;
 491   unsigned int matching;
 492   int bucket;
 493
 494   peer->preference_task = GNUNET_SCHEDULER_NO_TASK;
 495   if ((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
 496     return;
 497   matching =
 498     GNUNET_CRYPTO_hash_matching_bits (&my_identity.hashPubKey,
 499                                       &peer->id.hashPubKey);
 500   if (matching >= 64)
 501     matching = 63;
 502   bucket = find_bucket (&peer->id.hashPubKey);
 503   if (bucket == GNUNET_SYSERR)
 504     preference = 0;
 505   else
 506     preference = (1LL << matching) / k_buckets[bucket].peers_size;
 507   if (preference == 0)
 508     {
 509       peer->preference_task
 510         = GNUNET_SCHEDULER_add_delayed (DHT_DEFAULT_PREFERENCE_INTERVAL,
 511                                         &update_core_preference, peer);
 512       return;
 513     }
 514   GNUNET_STATISTICS_update (GDS_stats,
 515                             gettext_noop ("# Preference updates given to core"), 1,
 516                             GNUNET_NO);
 517   peer->info_ctx =
 518     GNUNET_CORE_peer_change_preference (coreAPI, &peer->id,
 519                                         GNUNET_TIME_UNIT_FOREVER_REL,
 520                                         GNUNET_BANDWIDTH_VALUE_MAX, 0,
 521                                         preference,
 522                                         &update_core_preference_finish, peer);
 523 }
 524
 525
 526 /**
 527  * Closure for 'add_known_to_bloom'.
 528  */
 529 struct BloomConstructorContext
 530 {
 531   /**
 532    * Bloom filter under construction.
 533    */
 534   struct GNUNET_CONTAINER_BloomFilter *bloom;
 535
 536   /**
 537    * Mutator to use.
 538    */
 539   uint32_t bf_mutator;
 540 };
 541
 542
 543 /**
 544  * Add each of the peers we already know to the bloom filter of
 545  * the request so that we don't get duplicate HELLOs.
 546  *
 547  * @param cls the 'struct BloomConstructorContext'.
 548  * @param key peer identity to add to the bloom filter
 549  * @param value value the peer information (unused)
 550  * @return GNUNET_YES (we should continue to iterate)
 551  */
 552 static int
 553 add_known_to_bloom (void *cls, const GNUNET_HashCode * key, void *value)
 554 {
 555   struct BloomConstructorContext *ctx = cls;
 556   GNUNET_HashCode mh;
 557
 558   GNUNET_BLOCK_mingle_hash (key, ctx->bf_mutator, &mh);
 559   GNUNET_CONTAINER_bloomfilter_add (ctx->bloom, &mh);
 560   return GNUNET_YES;
 561 }
 562
 563
 564 /**
 565  * Task to send a find peer message for our own peer identifier
 566  * so that we can find the closest peers in the network to ourselves
 567  * and attempt to connect to them.
 568  *
 569  * @param cls closure for this task
 570  * @param tc the context under which the task is running
 571  */
 572 static void
 573 send_find_peer_message (void *cls,
 574                         const struct GNUNET_SCHEDULER_TaskContext *tc)
 575 {
 576   struct GNUNET_TIME_Relative next_send_time;
 577   struct BloomConstructorContext bcc;
 578   struct GNUNET_CONTAINER_BloomFilter *peer_bf;
 579
 580   find_peer_task = GNUNET_SCHEDULER_NO_TASK;
 581   if ((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
 582     return;
 583   if (newly_found_peers > bucket_size)
 584   {
 585     /* If we are finding many peers already, no need to send out our request right now! */
 586     find_peer_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
 587                                                    &send_find_peer_message, NULL);
 588     newly_found_peers = 0;
 589     return;
 590   }
 591   bcc.bf_mutator = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, UINT32_MAX);
 592   bcc.bloom =
 593     GNUNET_CONTAINER_bloomfilter_init (NULL, DHT_BLOOM_SIZE, GNUNET_CONSTANTS_BLOOMFILTER_K);
 594   GNUNET_CONTAINER_multihashmap_iterate (all_known_peers,
 595                                          &add_known_to_bloom,
 596                                          &bcc);
 597   GNUNET_STATISTICS_update (GDS_stats,
 598                             gettext_noop ("# FIND PEER messages initiated"), 1,
 599                             GNUNET_NO);
 600   peer_bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
 601                                                DHT_BLOOM_SIZE,
 602                                                GNUNET_CONSTANTS_BLOOMFILTER_K);
 603   // FIXME: pass priority!?
 604   GDS_NEIGHBOURS_handle_get (GNUNET_BLOCK_TYPE_DHT_HELLO,
 605                              GNUNET_DHT_RO_FIND_PEER,
 606                              FIND_PEER_REPLICATION_LEVEL,
 607                              0,
 608                              &my_identity.hashPubKey,
 609                              NULL, 0,
 610                              bcc.bloom, bcc.bf_mutator,
 611                              peer_bf);
 612   GNUNET_CONTAINER_bloomfilter_free (peer_bf);
 613   GNUNET_CONTAINER_bloomfilter_free (bcc.bloom);
 614   /* schedule next round */
 615   next_send_time.rel_value =
 616     DHT_MINIMUM_FIND_PEER_INTERVAL.rel_value +
 617     GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
 618                               DHT_MAXIMUM_FIND_PEER_INTERVAL.rel_value / (newly_found_peers+1));
 619   newly_found_peers = 0;
 620   find_peer_task = GNUNET_SCHEDULER_add_delayed (next_send_time,
 621                                                  &send_find_peer_message,
 622                                                  NULL);
 623 }
 624
 625
 626 /**
 627  * Method called whenever a peer connects.
 628  *
 629  * @param cls closure
 630  * @param peer peer identity this notification is about
 631  * @param atsi performance data
 632  */
 633 static void
 634 handle_core_connect (void *cls, const struct GNUNET_PeerIdentity *peer,
 635                      const struct GNUNET_TRANSPORT_ATS_Information *atsi)
 636 {
 637   struct PeerInfo *ret;
 638   int peer_bucket;
 639
 640   /* Check for connect to self message */
 641   if (0 == memcmp (&my_identity, peer, sizeof (struct GNUNET_PeerIdentity)))
 642     return;
 643   GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 644               "Connected %s to %s\n",
 645               GNUNET_i2s (&my_identity),
 646               GNUNET_h2s (&peer->hashPubKey));
 647   if (GNUNET_YES ==
 648       GNUNET_CONTAINER_multihashmap_contains (all_known_peers,
 649                                               &peer->hashPubKey))
 650   {
 651     GNUNET_break (0);
 652     return;
 653   }
 654   GNUNET_STATISTICS_update (GDS_stats,
 655                             gettext_noop ("# Peers connected"), 1,
 656                             GNUNET_NO);
 657   peer_bucket = find_bucket (&peer->hashPubKey);
 658   GNUNET_assert ( (peer_bucket >= 0) && (peer_bucket < MAX_BUCKETS) );
 659   ret = GNUNET_malloc (sizeof (struct PeerInfo));
 660 #if 0
 661   ret->latency = latency;
 662   ret->distance = distance;
 663 #endif
 664   ret->id = *peer;
 665   GNUNET_CONTAINER_DLL_insert_tail (k_buckets[peer_bucket].head,
 666                                     k_buckets[peer_bucket].tail, ret);
 667   k_buckets[peer_bucket].peers_size++;
 668   closest_bucket = GNUNET_MAX (closest_bucket,
 669                                peer_bucket);
 670   if ( (peer_bucket > 0) &&
 671        (k_buckets[peer_bucket].peers_size <= bucket_size) )
 672   {
 673     ret->preference_task = GNUNET_SCHEDULER_add_now (&update_core_preference, ret);
 674     newly_found_peers++;
 675   }
 676   GNUNET_assert (GNUNET_OK ==
 677                  GNUNET_CONTAINER_multihashmap_put (all_known_peers,
 678                                                     &peer->hashPubKey, ret,
 679                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 680   if (1 == GNUNET_CONTAINER_multihashmap_size (all_known_peers))
 681   {
 682     /* got a first connection, good time to start with FIND PEER requests... */
 683     find_peer_task = GNUNET_SCHEDULER_add_now (&send_find_peer_message,
 684                                                NULL);
 685   }
 686 }
 687
 688
 689 /**
 690  * Method called whenever a peer disconnects.
 691  *
 692  * @param cls closure
 693  * @param peer peer identity this notification is about
 694  */
 695 static void
 696 handle_core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer)
 697 {
 698   struct PeerInfo *to_remove;
 699   int current_bucket;
 700   struct P2PPendingMessage *pos;
 701
 702   /* Check for disconnect from self message */
 703   if (0 == memcmp (&my_identity, peer, sizeof (struct GNUNET_PeerIdentity)))
 704     return;
 705   GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 706               "Disconnected %s from %s\n",
 707               GNUNET_i2s (&my_identity),
 708               GNUNET_h2s (&peer->hashPubKey));
 709   to_remove =
 710       GNUNET_CONTAINER_multihashmap_get (all_known_peers, &peer->hashPubKey);
 711   if (NULL == to_remove)
 712     {
 713       GNUNET_break (0);
 714       return;
 715     }
 716   GNUNET_STATISTICS_update (GDS_stats,
 717                             gettext_noop ("# Peers connected"), -1,
 718                             GNUNET_NO);
 719   GNUNET_assert (GNUNET_YES ==
 720                  GNUNET_CONTAINER_multihashmap_remove (all_known_peers,
 721                                                        &peer->hashPubKey,
 722                                                        to_remove));
 723   if (NULL != to_remove->info_ctx)
 724   {
 725     GNUNET_CORE_peer_change_preference_cancel (to_remove->info_ctx);
 726     to_remove->info_ctx = NULL;
 727   }
 728   if (GNUNET_SCHEDULER_NO_TASK != to_remove->preference_task)
 729   {
 730     GNUNET_SCHEDULER_cancel (to_remove->preference_task);
 731     to_remove->preference_task = GNUNET_SCHEDULER_NO_TASK;
 732   }
 733   current_bucket = find_bucket (&to_remove->id.hashPubKey);
 734   GNUNET_CONTAINER_DLL_remove (k_buckets[current_bucket].head,
 735                                k_buckets[current_bucket].tail,
 736                                to_remove);
 737   GNUNET_assert (k_buckets[current_bucket].peers_size > 0);
 738   k_buckets[current_bucket].peers_size--;
 739   while ( (closest_bucket > 0) &&
 740           (k_buckets[closest_bucket].peers_size == 0) )
 741     closest_bucket--;
 742
 743   if (to_remove->th != NULL)
 744   {
 745     GNUNET_CORE_notify_transmit_ready_cancel (to_remove->th);
 746     to_remove->th = NULL;
 747   }
 748   while (NULL != (pos = to_remove->head))
 749   {
 750     GNUNET_CONTAINER_DLL_remove (to_remove->head,
 751                                  to_remove->tail,
 752                                  pos);
 753     GNUNET_free (pos);
 754   }
 755 }
 756
 757
 758 /**
 759  * Called when core is ready to send a message we asked for
 760  * out to the destination.
 761  *
 762  * @param cls the 'struct PeerInfo' of the target peer
 763  * @param size number of bytes available in buf
 764  * @param buf where the callee should write the message
 765  * @return number of bytes written to buf
 766  */
 767 static size_t
 768 core_transmit_notify (void *cls, size_t size, void *buf)
 769 {
 770   struct PeerInfo *peer = cls;
 771   char *cbuf = buf;
 772   struct P2PPendingMessage *pending;
 773   size_t off;
 774   size_t msize;
 775
 776   peer->th = NULL;
 777   if (buf == NULL)
 778   {
 779     /* client disconnected */
 780     return 0;
 781   }
 782   if (peer->head == NULL)
 783   {
 784     /* no messages pending */
 785     return 0;
 786   }
 787   off = 0;
 788   while ( (NULL != (pending = peer->head)) &&
 789           (size - off >= (msize = ntohs (pending->msg->size))) )
 790   {
 791     GNUNET_STATISTICS_update (GDS_stats,
 792                               gettext_noop ("# Bytes transmitted to other peers"), msize,
 793                               GNUNET_NO);
 794     memcpy (&cbuf[off], pending->msg, msize);
 795     off += msize;
 796     peer->pending_count--;
 797     GNUNET_CONTAINER_DLL_remove (peer->head, peer->tail, pending);
 798     GNUNET_free (pending);
 799   }
 800   if (peer->head != NULL)
 801     {
 802       peer->th
 803         = GNUNET_CORE_notify_transmit_ready (coreAPI, GNUNET_YES,
 804                                              pending->importance,
 805                                              GNUNET_TIME_absolute_get_remaining (pending->timeout),
 806                                              &peer->id, msize,
 807                                              &core_transmit_notify, peer);
 808     }
 809   return off;
 810 }
 811
 812
 813 /**
 814  * Transmit all messages in the peer's message queue.
 815  *
 816  * @param peer message queue to process
 817  */
 818 static void
 819 process_peer_queue (struct PeerInfo *peer)
 820 {
 821   struct P2PPendingMessage *pending;
 822
 823   if (NULL == (pending = peer->head))
 824     return;
 825   if (NULL != peer->th)
 826     return;
 827   GNUNET_STATISTICS_update (GDS_stats,
 828                             gettext_noop ("# Bytes of bandwdith requested from core"),
 829                             ntohs (pending->msg->size),
 830                             GNUNET_NO);
 831   peer->th
 832     = GNUNET_CORE_notify_transmit_ready (coreAPI, GNUNET_YES,
 833                                          pending->importance,
 834                                          GNUNET_TIME_absolute_get_remaining (pending->timeout),
 835                                          &peer->id,
 836                                          ntohs (pending->msg->size),
 837                                          &core_transmit_notify, peer);
 838 }
 839
 840
 841 /**
 842  * To how many peers should we (on average) forward the request to
 843  * obtain the desired target_replication count (on average).
 844  *
 845  * @param hop_count number of hops the message has traversed
 846  * @param target_replication the number of total paths desired
 847  * @return Some number of peers to forward the message to
 848  */
 849 static unsigned int
 850 get_forward_count (uint32_t hop_count,
 851                    uint32_t target_replication)
 852 {
 853   uint32_t random_value;
 854   uint32_t forward_count;
 855   float target_value;
 856
 857   if (hop_count > GDS_NSE_get () * 4.0)
 858   {
 859     /* forcefully terminate */
 860     return 0;
 861   }
 862   if (hop_count > GDS_NSE_get () * 2.0)
 863   {
 864     /* Once we have reached our ideal number of hops, only forward to 1 peer */
 865     return 1;
 866   }
 867   /* bound by system-wide maximum */
 868   target_replication = GNUNET_MIN (MAXIMUM_REPLICATION_LEVEL,
 869                                    target_replication);
 870   target_value =
 871     1 + (target_replication - 1.0) / (GDS_NSE_get () +
 872                                       ((float) (target_replication - 1.0) *
 873                                        hop_count));
 874   /* Set forward count to floor of target_value */
 875   forward_count = (uint32_t) target_value;
 876   /* Subtract forward_count (floor) from target_value (yields value between 0 and 1) */
 877   target_value = target_value - forward_count;
 878   random_value =
 879     GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, UINT32_MAX);
 880   if (random_value < (target_value * UINT32_MAX))
 881     forward_count++;
 882   return forward_count;
 883 }
 884
 885
 886 /**
 887  * Compute the distance between have and target as a 32-bit value.
 888  * Differences in the lower bits must count stronger than differences
 889  * in the higher bits.
 890  *
 891  * @return 0 if have==target, otherwise a number
 892  *           that is larger as the distance between
 893  *           the two hash codes increases
 894  */
 895 static unsigned int
 896 get_distance (const GNUNET_HashCode * target, const GNUNET_HashCode * have)
 897 {
 898   unsigned int bucket;
 899   unsigned int msb;
 900   unsigned int lsb;
 901   unsigned int i;
 902
 903   /* We have to represent the distance between two 2^9 (=512)-bit
 904    * numbers as a 2^5 (=32)-bit number with "0" being used for the
 905    * two numbers being identical; furthermore, we need to
 906    * guarantee that a difference in the number of matching
 907    * bits is always represented in the result.
 908    *
 909    * We use 2^32/2^9 numerical values to distinguish between
 910    * hash codes that have the same LSB bit distance and
 911    * use the highest 2^9 bits of the result to signify the
 912    * number of (mis)matching LSB bits; if we have 0 matching
 913    * and hence 512 mismatching LSB bits we return -1 (since
 914    * 512 itself cannot be represented with 9 bits) */
 915
 916   /* first, calculate the most significant 9 bits of our
 917    * result, aka the number of LSBs */
 918   bucket = GNUNET_CRYPTO_hash_matching_bits (target, have);
 919   /* bucket is now a value between 0 and 512 */
 920   if (bucket == 512)
 921     return 0;                   /* perfect match */
 922   if (bucket == 0)
 923     return (unsigned int) -1;   /* LSB differs; use max (if we did the bit-shifting
 924                                  * below, we'd end up with max+1 (overflow)) */
 925
 926   /* calculate the most significant bits of the final result */
 927   msb = (512 - bucket) << (32 - 9);
 928   /* calculate the 32-9 least significant bits of the final result by
 929    * looking at the differences in the 32-9 bits following the
 930    * mismatching bit at 'bucket' */
 931   lsb = 0;
 932   for (i = bucket + 1;
 933        (i < sizeof (GNUNET_HashCode) * 8) && (i < bucket + 1 + 32 - 9); i++)
 934   {
 935     if (GNUNET_CRYPTO_hash_get_bit (target, i) !=
 936         GNUNET_CRYPTO_hash_get_bit (have, i))
 937       lsb |= (1 << (bucket + 32 - 9 - i));      /* first bit set will be 10,
 938                                                  * last bit set will be 31 -- if
 939                                                  * i does not reach 512 first... */
 940   }
 941   return msb | lsb;
 942 }
 943
 944
 945 /**
 946  * Check whether my identity is closer than any known peers.  If a
 947  * non-null bloomfilter is given, check if this is the closest peer
 948  * that hasn't already been routed to.
 949  *
 950  * @param key hash code to check closeness to
 951  * @param bloom bloomfilter, exclude these entries from the decision
 952  * @return GNUNET_YES if node location is closest,
 953  *         GNUNET_NO otherwise.
 954  */
 955 static int
 956 am_closest_peer (const GNUNET_HashCode *key,
 957                  const struct GNUNET_CONTAINER_BloomFilter *bloom)
 958 {
 959   int bits;
 960   int other_bits;
 961   int bucket_num;
 962   int count;
 963   struct PeerInfo *pos;
 964
 965   if (0 == memcmp (&my_identity.hashPubKey, key, sizeof (GNUNET_HashCode)))
 966     return GNUNET_YES;
 967   bucket_num = find_bucket (key);
 968   bits = GNUNET_CRYPTO_hash_matching_bits (&my_identity.hashPubKey, key);
 969   pos = k_buckets[bucket_num].head;
 970   count = 0;
 971   while ((pos != NULL) && (count < bucket_size))
 972   {
 973     if ((bloom != NULL) &&
 974         (GNUNET_YES ==
 975          GNUNET_CONTAINER_bloomfilter_test (bloom, &pos->id.hashPubKey)))
 976     {
 977       pos = pos->next;
 978       continue;                 /* Skip already checked entries */
 979     }
 980     other_bits = GNUNET_CRYPTO_hash_matching_bits (&pos->id.hashPubKey, key);
 981     if (other_bits > bits)
 982       return GNUNET_NO;
 983     if (other_bits == bits)        /* We match the same number of bits */
 984       return GNUNET_YES;
 985     pos = pos->next;
 986   }
 987   /* No peers closer, we are the closest! */
 988   return GNUNET_YES;
 989 }
 990
 991
 992 /**
 993  * Select a peer from the routing table that would be a good routing
 994  * destination for sending a message for "key".  The resulting peer
 995  * must not be in the set of blocked peers.<p>
 996  *
 997  * Note that we should not ALWAYS select the closest peer to the
 998  * target, peers further away from the target should be chosen with
 999  * exponentially declining probability.
1000  *
1001  * FIXME: double-check that this is fine
1002  *
1003  *
1004  * @param key the key we are selecting a peer to route to
1005  * @param bloom a bloomfilter containing entries this request has seen already
1006  * @param hops how many hops has this message traversed thus far
1007  * @return Peer to route to, or NULL on error
1008  */
1009 static struct PeerInfo *
1010 select_peer (const GNUNET_HashCode *key,
1011              const struct GNUNET_CONTAINER_BloomFilter *bloom,
1012              uint32_t hops)
1013 {
1014   unsigned int bc;
1015   unsigned int count;
1016   unsigned int selected;
1017   struct PeerInfo *pos;
1018   unsigned int dist;
1019   unsigned int smallest_distance;
1020   struct PeerInfo *chosen;
1021
1022   if (hops >= GDS_NSE_get ())
1023   {
1024     /* greedy selection (closest peer that is not in bloomfilter) */
1025     smallest_distance = UINT_MAX;
1026     chosen = NULL;
1027     for (bc = 0; bc < closest_bucket; bc++)
1028     {
1029       pos = k_buckets[bc].head;
1030       count = 0;
1031       while ((pos != NULL) && (count < bucket_size))
1032       {
1033         if ( (bloom == NULL) ||
1034              (GNUNET_NO ==
1035               GNUNET_CONTAINER_bloomfilter_test (bloom, &pos->id.hashPubKey)) )
1036         {
1037           dist = get_distance (key, &pos->id.hashPubKey);
1038           if (dist < smallest_distance)
1039           {
1040             chosen = pos;
1041             smallest_distance = dist;
1042           }
1043         }
1044         else
1045         {
1046           GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1047                       "Excluded peer `%s' due to BF match in greedy routing for %s\n",
1048                       GNUNET_i2s (&pos->id),
1049                       GNUNET_h2s (key));
1050           GNUNET_STATISTICS_update (GDS_stats,
1051                                     gettext_noop ("# Peers excluded from routing due to Bloomfilter"), 1,
1052                                     GNUNET_NO);
1053         }
1054         count++;
1055         pos = pos->next;
1056       }
1057     }
1058     if (NULL == chosen)
1059       GNUNET_STATISTICS_update (GDS_stats,
1060                                 gettext_noop ("# Peer selection failed"), 1,
1061                                 GNUNET_NO);
1062     return chosen;
1063   }
1064
1065   /* select "random" peer */
1066   /* count number of peers that are available and not filtered */
1067   count = 0;
1068   for (bc = closest_bucket; bc < MAX_BUCKETS; bc++)
1069   {
1070     pos = k_buckets[bc].head;
1071     while ((pos != NULL) && (count < bucket_size))
1072     {
1073       if ( (bloom != NULL) &&
1074            (GNUNET_YES ==
1075             GNUNET_CONTAINER_bloomfilter_test (bloom, &pos->id.hashPubKey)) )
1076       {
1077         GNUNET_STATISTICS_update (GDS_stats,
1078                                   gettext_noop ("# Peers excluded from routing due to Bloomfilter"), 1,
1079                                   GNUNET_NO);
1080         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1081                     "Excluded peer `%s' due to BF match in random routing for %s\n",
1082                     GNUNET_i2s (&pos->id),
1083                     GNUNET_h2s (key));
1084         pos = pos->next;
1085         continue;               /* Ignore bloomfiltered peers */
1086       }
1087       count++;
1088       pos = pos->next;
1089     }
1090   }
1091   if (count == 0)               /* No peers to select from! */
1092   {
1093     GNUNET_STATISTICS_update (GDS_stats,
1094                               gettext_noop ("# Peer selection failed"), 1,
1095                               GNUNET_NO);
1096     return NULL;
1097   }
1098   /* Now actually choose a peer */
1099   selected = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, count);
1100   count = 0;
1101   for (bc = closest_bucket; bc < MAX_BUCKETS; bc++)
1102   {
1103     pos = k_buckets[bc].head;
1104     while ((pos != NULL) && (count < bucket_size))
1105     {
1106       if ( (bloom != NULL) &&
1107            (GNUNET_YES ==
1108             GNUNET_CONTAINER_bloomfilter_test (bloom, &pos->id.hashPubKey)) )
1109       {
1110         pos = pos->next;
1111         continue;               /* Ignore bloomfiltered peers */
1112       }
1113       if (0 == selected--)
1114         return pos;
1115       pos = pos->next;
1116     }
1117   }
1118   GNUNET_break (0);
1119   return NULL;
1120 }
1121
1122
1123 /**
1124  * Compute the set of peers that the given request should be
1125  * forwarded to.
1126  *
1127  * @param key routing key
1128  * @param bloom bloom filter excluding peers as targets, all selected
1129  *        peers will be added to the bloom filter
1130  * @param hop_count number of hops the request has traversed so far
1131  * @param target_replication desired number of replicas
1132  * @param targets where to store an array of target peers (to be
1133  *         free'd by the caller)
1134  * @return number of peers returned in 'targets'.
1135  */
1136 static unsigned int
1137 get_target_peers (const GNUNET_HashCode *key,
1138                   struct GNUNET_CONTAINER_BloomFilter *bloom,
1139                   uint32_t hop_count,
1140                   uint32_t target_replication,
1141                   struct PeerInfo ***targets)
1142 {
1143   unsigned int ret;
1144   unsigned int off;
1145   struct PeerInfo **rtargets;
1146   struct PeerInfo *nxt;
1147
1148   GNUNET_assert (NULL != bloom);
1149   ret = get_forward_count (hop_count, target_replication);
1150   if (ret == 0)
1151   {
1152     *targets = NULL;
1153     return 0;
1154   }
1155   rtargets = GNUNET_malloc (sizeof (struct PeerInfo*) * ret);
1156   for (off = 0; off < ret; off++)
1157   {
1158     nxt = select_peer (key, bloom, hop_count);
1159     if (nxt == NULL)
1160       break;
1161     rtargets[off] = nxt;
1162     GNUNET_break (GNUNET_NO ==
1163                   GNUNET_CONTAINER_bloomfilter_test (bloom, &nxt->id.hashPubKey));
1164   }
1165   GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1166               "Selected %u/%u peers at hop %u for %s (target was %u)\n",
1167               off,
1168               GNUNET_CONTAINER_multihashmap_size (all_known_peers),
1169               (unsigned int) hop_count,
1170               GNUNET_h2s (key),
1171               ret);
1172
1173   if (0 == off)
1174   {
1175     GNUNET_free (rtargets);
1176     *targets = NULL;
1177     return 0;
1178   }
1179   *targets = rtargets;
1180   for (ret = 0; ret < off; ret++)
1181     GNUNET_CONTAINER_bloomfilter_add (bloom, &rtargets[ret]->id.hashPubKey);
1182   return off;
1183 }
1184
1185
1186 /**
1187  * Perform a PUT operation.   Forwards the given request to other
1188  * peers.   Does not store the data locally.  Does not give the
1189  * data to local clients.  May do nothing if this is the only
1190  * peer in the network (or if we are the closest peer in the
1191  * network).
1192  *
1193  * @param type type of the block
1194  * @param options routing options
1195  * @param desired_replication_level desired replication count
1196  * @param expiration_time when does the content expire
1197  * @param hop_count how many hops has this message traversed so far
1198  * @param bf Bloom filter of peers this PUT has already traversed
1199  * @param key key for the content
1200  * @param put_path_length number of entries in put_path
1201  * @param put_path peers this request has traversed so far (if tracked)
1202  * @param data payload to store
1203  * @param data_size number of bytes in data
1204  */
1205 void
1206 GDS_NEIGHBOURS_handle_put (enum GNUNET_BLOCK_Type type,
1207                            enum GNUNET_DHT_RouteOption options,
1208                            uint32_t desired_replication_level,
1209                            struct GNUNET_TIME_Absolute expiration_time,
1210                            uint32_t hop_count,
1211                            struct GNUNET_CONTAINER_BloomFilter *bf,
1212                            const GNUNET_HashCode *key,
1213                            unsigned int put_path_length,
1214                            struct GNUNET_PeerIdentity *put_path,
1215                            const void *data,
1216                            size_t data_size)
1217 {
1218   unsigned int target_count;
1219   unsigned int i;
1220   struct PeerInfo **targets;
1221   struct PeerInfo *target;
1222   struct P2PPendingMessage *pending;
1223   size_t msize;
1224   struct PeerPutMessage *ppm;
1225   struct GNUNET_PeerIdentity *pp;
1226
1227   GNUNET_assert (NULL != bf);
1228   GNUNET_CONTAINER_bloomfilter_add (bf, &my_identity.hashPubKey);
1229   GNUNET_STATISTICS_update (GDS_stats,
1230                             gettext_noop ("# PUT requests routed"), 1,
1231                             GNUNET_NO);
1232   target_count = get_target_peers (key, bf, hop_count,
1233                                    desired_replication_level,
1234                                    &targets);
1235   if (0 == target_count)
1236     {
1237       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1238                   "Routing PUT for %s terminates after %u hops at %s\n",
1239                   GNUNET_h2s (key),
1240                   (unsigned int) hop_count,
1241                   GNUNET_i2s (&my_identity));
1242       return;
1243     }
1244   msize = put_path_length * sizeof (struct GNUNET_PeerIdentity) + data_size + sizeof (struct PeerPutMessage);
1245   if (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
1246   {
1247     put_path_length = 0;
1248     msize = data_size + sizeof (struct PeerPutMessage);
1249   }
1250   if (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
1251   {
1252     GNUNET_break (0);
1253     return;
1254   }
1255   GNUNET_STATISTICS_update (GDS_stats,
1256                             gettext_noop ("# Peers selected as targets for PUT requests"), target_count,
1257                             GNUNET_NO);
1258   for (i=0;i<target_count;i++)
1259   {
1260     target = targets[i];
1261     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1262                 "Routing PUT for %s after %u hops to %s\n",
1263                 GNUNET_h2s (key),
1264                 (unsigned int) hop_count,
1265                 GNUNET_i2s (&target->id));
1266     pending = GNUNET_malloc (sizeof (struct P2PPendingMessage) + msize);
1267     pending->importance = 0; /* FIXME */
1268     pending->timeout = expiration_time;
1269     ppm = (struct PeerPutMessage*) &pending[1];
1270     pending->msg = &ppm->header;
1271     ppm->header.size = htons (msize);
1272     ppm->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_P2P_PUT);
1273     ppm->options = htonl (options);
1274     ppm->type = htonl (type);
1275     ppm->hop_count = htonl (hop_count + 1);
1276     ppm->desired_replication_level = htonl (desired_replication_level);
1277     ppm->put_path_length = htonl (put_path_length);
1278     ppm->expiration_time = GNUNET_TIME_absolute_hton (expiration_time);
1279     GNUNET_break (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (bf, &target->id.hashPubKey));
1280     GNUNET_assert (GNUNET_OK ==
1281                    GNUNET_CONTAINER_bloomfilter_get_raw_data (bf,
1282                                                               ppm->bloomfilter,
1283                                                               DHT_BLOOM_SIZE));
1284     ppm->key = *key;
1285     pp = (struct GNUNET_PeerIdentity*) &ppm[1];
1286     memcpy (pp, put_path, sizeof (struct GNUNET_PeerIdentity) * put_path_length);
1287     memcpy (&pp[put_path_length], data, data_size);
1288     GNUNET_CONTAINER_DLL_insert_tail (target->head,
1289                                       target->tail,
1290                                       pending);
1291     target->pending_count++;
1292     process_peer_queue (target);
1293   }
1294   GNUNET_free (targets);
1295 }
1296
1297
1298 /**
1299  * Perform a GET operation.  Forwards the given request to other
1300  * peers.  Does not lookup the key locally.  May do nothing if this is
1301  * the only peer in the network (or if we are the closest peer in the
1302  * network).
1303  *
1304  * @param type type of the block
1305  * @param options routing options
1306  * @param desired_replication_level desired replication count
1307  * @param hop_count how many hops did this request traverse so far?
1308  * @param key key for the content
1309  * @param xquery extended query
1310  * @param xquery_size number of bytes in xquery
1311  * @param reply_bf bloomfilter to filter duplicates
1312  * @param reply_bf_mutator mutator for reply_bf
1313  * @param peer_bf filter for peers not to select (again)
1314  */
1315 void
1316 GDS_NEIGHBOURS_handle_get (enum GNUNET_BLOCK_Type type,
1317                            enum GNUNET_DHT_RouteOption options,
1318                            uint32_t desired_replication_level,
1319                            uint32_t hop_count,
1320                            const GNUNET_HashCode *key,
1321                            const void *xquery,
1322                            size_t xquery_size,
1323                            const struct GNUNET_CONTAINER_BloomFilter *reply_bf,
1324                            uint32_t reply_bf_mutator,
1325                            struct GNUNET_CONTAINER_BloomFilter *peer_bf)
1326 {
1327   unsigned int target_count;
1328   unsigned int i;
1329   struct PeerInfo **targets;
1330   struct PeerInfo *target;
1331   struct P2PPendingMessage *pending;
1332   size_t msize;
1333   struct PeerGetMessage *pgm;
1334   char *xq;
1335   size_t reply_bf_size;
1336
1337   GNUNET_assert (NULL != peer_bf);
1338   GNUNET_STATISTICS_update (GDS_stats,
1339                             gettext_noop ("# GET requests routed"), 1,
1340                             GNUNET_NO);
1341   target_count = get_target_peers (key, peer_bf, hop_count,
1342                                    desired_replication_level,
1343                                    &targets);
1344   GNUNET_CONTAINER_bloomfilter_add (peer_bf, &my_identity.hashPubKey);
1345   if (0 == target_count)
1346     {
1347       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1348                   "Routing GET for %s terminates after %u hops at %s\n",
1349                   GNUNET_h2s (key),
1350                   (unsigned int) hop_count,
1351                   GNUNET_i2s (&my_identity));
1352       return;
1353     }
1354   reply_bf_size = GNUNET_CONTAINER_bloomfilter_get_size (reply_bf);
1355   msize = xquery_size + sizeof (struct PeerGetMessage) + reply_bf_size;
1356   if (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
1357   {
1358     GNUNET_break (0);
1359     return;
1360   }
1361   GNUNET_STATISTICS_update (GDS_stats,
1362                             gettext_noop ("# Peers selected as targets for GET requests"), target_count,
1363                             GNUNET_NO);
1364   /* forward request */
1365   for (i=0;i<target_count;i++)
1366   {
1367     target = targets[i];
1368     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1369                 "Routing GET for %s after %u hops to %s\n",
1370                 GNUNET_h2s (key),
1371                 (unsigned int) hop_count,
1372                 GNUNET_i2s (&target->id));
1373     pending = GNUNET_malloc (sizeof (struct P2PPendingMessage) + msize);
1374     pending->importance = 0; /* FIXME */
1375     pending->timeout = GNUNET_TIME_relative_to_absolute (GET_TIMEOUT);
1376     pgm = (struct PeerGetMessage*) &pending[1];
1377     pending->msg = &pgm->header;
1378     pgm->header.size = htons (msize);
1379     pgm->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_P2P_GET);
1380     pgm->options = htonl (options);
1381     pgm->type = htonl (type);
1382     pgm->hop_count = htonl (hop_count + 1);
1383     pgm->desired_replication_level = htonl (desired_replication_level);
1384     pgm->xquery_size = htonl (xquery_size);
1385     pgm->bf_mutator = reply_bf_mutator;
1386     GNUNET_break (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (peer_bf, &target->id.hashPubKey));
1387     GNUNET_assert (GNUNET_OK ==
1388                    GNUNET_CONTAINER_bloomfilter_get_raw_data (peer_bf,
1389                                                               pgm->bloomfilter,
1390                                                               DHT_BLOOM_SIZE));
1391     pgm->key = *key;
1392     xq = (char *) &pgm[1];
1393     memcpy (xq, xquery, xquery_size);
1394     if (NULL != reply_bf)
1395       GNUNET_assert (GNUNET_OK ==
1396                      GNUNET_CONTAINER_bloomfilter_get_raw_data (reply_bf,
1397                                                                 &xq[xquery_size],
1398                                                                 reply_bf_size));
1399     GNUNET_CONTAINER_DLL_insert_tail (target->head,
1400                                       target->tail,
1401                                       pending);
1402     target->pending_count++;
1403     process_peer_queue (target);
1404   }
1405   GNUNET_free (targets);
1406 }
1407
1408
1409 /**
1410  * Handle a reply (route to origin).  Only forwards the reply back to
1411  * the given peer.  Does not do local caching or forwarding to local
1412  * clients.
1413  *
1414  * @param target neighbour that should receive the block (if still connected)
1415  * @param type type of the block
1416  * @param expiration_time when does the content expire
1417  * @param key key for the content
1418  * @param put_path_length number of entries in put_path
1419  * @param put_path peers the original PUT traversed (if tracked)
1420  * @param get_path_length number of entries in put_path
1421  * @param get_path peers this reply has traversed so far (if tracked)
1422  * @param data payload of the reply
1423  * @param data_size number of bytes in data
1424  */
1425 void
1426 GDS_NEIGHBOURS_handle_reply (const struct GNUNET_PeerIdentity *target,
1427                              enum GNUNET_BLOCK_Type type,
1428                              struct GNUNET_TIME_Absolute expiration_time,
1429                              const GNUNET_HashCode *key,
1430                              unsigned int put_path_length,
1431                              const struct GNUNET_PeerIdentity *put_path,
1432                              unsigned int get_path_length,
1433                              const struct GNUNET_PeerIdentity *get_path,
1434                              const void *data,
1435                              size_t data_size)
1436 {
1437   struct PeerInfo *pi;
1438   struct P2PPendingMessage *pending;
1439   size_t msize;
1440   struct PeerResultMessage *prm;
1441   struct GNUNET_PeerIdentity *paths;
1442
1443   msize = data_size + sizeof (struct PeerResultMessage) +
1444     (get_path_length + put_path_length) * sizeof (struct GNUNET_PeerIdentity);
1445   if ( (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
1446        (get_path_length > GNUNET_SERVER_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)) ||
1447        (put_path_length > GNUNET_SERVER_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)) ||
1448        (data_size > GNUNET_SERVER_MAX_MESSAGE_SIZE) )
1449   {
1450     GNUNET_break (0);
1451     return;
1452   }
1453   pi = GNUNET_CONTAINER_multihashmap_get (all_known_peers,
1454                                           &target->hashPubKey);
1455   if (NULL == pi)
1456   {
1457     /* peer disconnected in the meantime, drop reply */
1458     return;
1459   }
1460   GNUNET_STATISTICS_update (GDS_stats,
1461                             gettext_noop ("# REPLIES routed"), 1,
1462                             GNUNET_NO);
1463   pending = GNUNET_malloc (sizeof (struct P2PPendingMessage) + msize);
1464   pending->importance = 0; /* FIXME */
1465   pending->timeout = expiration_time;
1466   prm = (struct PeerResultMessage*) &pending[1];
1467   pending->msg = &prm->header;
1468   prm->header.size = htons (msize);
1469   prm->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_P2P_RESULT);
1470   prm->type = htonl (type);
1471   prm->put_path_length = htonl (put_path_length);
1472   prm->get_path_length = htonl (get_path_length);
1473   prm->expiration_time = GNUNET_TIME_absolute_hton (expiration_time);
1474   prm->key = *key;
1475   paths = (struct GNUNET_PeerIdentity*) &prm[1];
1476   memcpy (paths, put_path, put_path_length * sizeof (struct GNUNET_PeerIdentity));
1477   memcpy (&paths[put_path_length],
1478           get_path, get_path_length * sizeof (struct GNUNET_PeerIdentity));
1479   memcpy (&paths[put_path_length + get_path_length],
1480           data, data_size);
1481   GNUNET_CONTAINER_DLL_insert (pi->head,
1482                                pi->tail,
1483                                pending);
1484   pi->pending_count++;
1485   process_peer_queue (pi);
1486 }
1487
1488
1489 /**
1490  * To be called on core init/fail.
1491  *
1492  * @param cls service closure
1493  * @param server handle to the server for this service
1494  * @param identity the public identity of this peer
1495  * @param publicKey the public key of this peer
1496  */
1497 static void
1498 core_init (void *cls, struct GNUNET_CORE_Handle *server,
1499            const struct GNUNET_PeerIdentity *identity,
1500            const struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded *publicKey)
1501 {
1502   GNUNET_assert (server != NULL);
1503   my_identity = *identity;
1504 }
1505
1506
1507 /**
1508  * Core handler for p2p put requests.
1509  *
1510  * @param cls closure
1511  * @param peer sender of the request
1512  * @param message message
1513  * @param peer peer identity this notification is about
1514  * @param atsi performance data
1515  * @return GNUNET_OK to keep the connection open,
1516  *         GNUNET_SYSERR to close it (signal serious error)
1517  */
1518 static int
1519 handle_dht_p2p_put (void *cls,
1520                     const struct GNUNET_PeerIdentity *peer,
1521                     const struct GNUNET_MessageHeader *message,
1522                     const struct GNUNET_TRANSPORT_ATS_Information
1523                     *atsi)
1524 {
1525   const struct PeerPutMessage *put;
1526   const struct GNUNET_PeerIdentity *put_path;
1527   const void *payload;
1528   uint32_t putlen;
1529   uint16_t msize;
1530   size_t payload_size;
1531   enum GNUNET_DHT_RouteOption options;
1532   struct GNUNET_CONTAINER_BloomFilter *bf;
1533   GNUNET_HashCode test_key;
1534
1535   msize = ntohs (message->size);
1536   if (msize < sizeof (struct PeerPutMessage))
1537   {
1538     GNUNET_break_op (0);
1539     return GNUNET_YES;
1540   }
1541   put = (const struct PeerPutMessage*) message;
1542   putlen = ntohl (put->put_path_length);
1543   if ( (msize < sizeof (struct PeerPutMessage) + putlen * sizeof (struct GNUNET_PeerIdentity)) ||
1544        (putlen > GNUNET_SERVER_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)) )
1545     {
1546       GNUNET_break_op (0);
1547       return GNUNET_YES;
1548     }
1549   GNUNET_STATISTICS_update (GDS_stats,
1550                             gettext_noop ("# P2P PUT requests received"), 1,
1551                             GNUNET_NO);
1552   put_path = (const struct GNUNET_PeerIdentity*) &put[1];
1553   payload = &put_path[putlen];
1554   options = ntohl (put->options);
1555   payload_size = msize - (sizeof (struct PeerPutMessage) +
1556                           putlen * sizeof (struct GNUNET_PeerIdentity));
1557   switch (GNUNET_BLOCK_get_key (GDS_block_context,
1558                                 ntohl (put->type),
1559                                 payload, payload_size,
1560                                 &test_key))
1561   {
1562   case GNUNET_YES:
1563     if (0 != memcmp (&test_key, &put->key, sizeof (GNUNET_HashCode)))
1564     {
1565       GNUNET_break_op (0);
1566       return GNUNET_YES;
1567     }
1568     break;
1569   case GNUNET_NO:
1570     GNUNET_break_op (0);
1571     return GNUNET_YES;
1572   case GNUNET_SYSERR:
1573     /* cannot verify, good luck */
1574     break;
1575   }
1576   GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1577               "PUT for %s at %s\n",
1578               GNUNET_h2s (&put->key),
1579               GNUNET_i2s (&my_identity));
1580   bf = GNUNET_CONTAINER_bloomfilter_init (put->bloomfilter,
1581                                           DHT_BLOOM_SIZE,
1582                                           GNUNET_CONSTANTS_BLOOMFILTER_K);
1583   GNUNET_break_op (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (bf, &peer->hashPubKey));
1584   {
1585     struct GNUNET_PeerIdentity pp[putlen+1];
1586
1587     /* extend 'put path' by sender */
1588     if (0 != (options & GNUNET_DHT_RO_RECORD_ROUTE))
1589     {
1590       memcpy (pp, put_path, putlen * sizeof (struct GNUNET_PeerIdentity));
1591       pp[putlen] = *peer;
1592       putlen++;
1593     }
1594     else
1595       putlen = 0;
1596
1597     /* give to local clients */
1598     GDS_CLIENTS_handle_reply (GNUNET_TIME_absolute_ntoh (put->expiration_time),
1599                              &put->key,
1600                              0, NULL,
1601                              putlen,
1602                              pp,
1603                              ntohl (put->type),
1604                              payload_size,
1605                              payload);
1606     /* store locally */
1607     if ( (0 != (options & GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE)) ||
1608          (am_closest_peer (&put->key,
1609                            bf) ) )
1610       GDS_DATACACHE_handle_put (GNUNET_TIME_absolute_ntoh (put->expiration_time),
1611                                 &put->key,
1612                                 putlen, pp,
1613                                 ntohl (put->type),
1614                                 payload_size,
1615                                 payload);
1616     /* route to other peers */
1617     GDS_NEIGHBOURS_handle_put (ntohl (put->type),
1618                                options,
1619                                ntohl (put->desired_replication_level),
1620                                GNUNET_TIME_absolute_ntoh (put->expiration_time),
1621                                ntohl (put->hop_count),
1622                                bf,
1623                                &put->key,
1624                                putlen, pp,
1625                                payload,
1626                                payload_size);
1627   }
1628   GNUNET_CONTAINER_bloomfilter_free (bf);
1629   return GNUNET_YES;
1630 }
1631
1632
1633 /**
1634  * We have received a FIND PEER request.  Send matching
1635  * HELLOs back.
1636  *
1637  * @param sender sender of the FIND PEER request
1638  * @param key peers close to this key are desired
1639  * @param bf peers matching this bf are excluded
1640  * @param bf_mutator mutator for bf
1641  */
1642 static void
1643 handle_find_peer (const struct GNUNET_PeerIdentity *sender,
1644                   const GNUNET_HashCode *key,
1645                   struct GNUNET_CONTAINER_BloomFilter *bf,
1646                   uint32_t bf_mutator)
1647 {
1648   int bucket_idx;
1649   struct PeerBucket *bucket;
1650   struct PeerInfo *peer;
1651   unsigned int choice;
1652   GNUNET_HashCode mhash;
1653   const struct GNUNET_HELLO_Message *hello;
1654
1655   /* first, check about our own HELLO */
1656   if (NULL != GDS_my_hello)
1657   {
1658     GNUNET_BLOCK_mingle_hash (&my_identity.hashPubKey, bf_mutator, &mhash);
1659     if ( (NULL == bf) ||
1660          (GNUNET_YES != GNUNET_CONTAINER_bloomfilter_test (bf, &mhash)) )
1661     {
1662       GDS_NEIGHBOURS_handle_reply (sender,
1663                                    GNUNET_BLOCK_TYPE_DHT_HELLO,
1664                                    GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_HELLO_ADDRESS_EXPIRATION),
1665                                    key,
1666                                    0, NULL,
1667                                    0, NULL,
1668                                    GDS_my_hello,
1669                                    GNUNET_HELLO_size ((const struct GNUNET_HELLO_Message*) GDS_my_hello));
1670     }
1671     else
1672     {
1673       GNUNET_STATISTICS_update (GDS_stats,
1674                                 gettext_noop ("# FIND PEER requests ignored due to Bloomfilter"), 1,
1675                                 GNUNET_NO);
1676     }
1677   }
1678   else
1679   {
1680     GNUNET_STATISTICS_update (GDS_stats,
1681                               gettext_noop ("# FIND PEER requests ignored due to lack of HELLO"), 1,
1682                               GNUNET_NO);
1683   }
1684
1685   /* then, also consider sending a random HELLO from the closest bucket */
1686   if (0 == memcmp (&my_identity.hashPubKey, key, sizeof (GNUNET_HashCode)))
1687     bucket_idx = closest_bucket;
1688   else
1689     bucket_idx = GNUNET_MIN (closest_bucket, find_bucket (key));
1690   if (bucket_idx == GNUNET_SYSERR)
1691     return;
1692   bucket = &k_buckets[bucket_idx];
1693   if (bucket->peers_size == 0)
1694     return;
1695   choice = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1696                                      bucket->peers_size);
1697   peer = bucket->head;
1698   while (choice > 0)
1699   {
1700     GNUNET_assert (peer != NULL);
1701     peer = peer->next;
1702     choice--;
1703   }
1704   choice = bucket->peers_size;
1705   do
1706     {
1707       peer = peer->next;
1708       if (choice-- == 0)
1709         return; /* no non-masked peer available */
1710       if (peer == NULL)
1711         peer = bucket->head;
1712       GNUNET_BLOCK_mingle_hash (&peer->id.hashPubKey, bf_mutator, &mhash);
1713       hello = GDS_HELLO_get (&peer->id);
1714     }
1715   while ( (hello == NULL) ||
1716           (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (bf, &mhash)) );
1717   GDS_NEIGHBOURS_handle_reply (sender,
1718                                GNUNET_BLOCK_TYPE_DHT_HELLO,
1719                                GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_HELLO_ADDRESS_EXPIRATION),
1720                                key,
1721                                0, NULL,
1722                                0, NULL,
1723                                hello,
1724                                GNUNET_HELLO_size (hello));
1725 }
1726
1727
1728 /**
1729  * Core handler for p2p get requests.
1730  *
1731  * @param cls closure
1732  * @param peer sender of the request
1733  * @param message message
1734  * @param peer peer identity this notification is about
1735  * @param atsi performance data
1736  * @return GNUNET_OK to keep the connection open,
1737  *         GNUNET_SYSERR to close it (signal serious error)
1738  */
1739 static int
1740 handle_dht_p2p_get (void *cls, const struct GNUNET_PeerIdentity *peer,
1741                     const struct GNUNET_MessageHeader *message,
1742                     const struct GNUNET_TRANSPORT_ATS_Information
1743                     *atsi)
1744 {
1745   struct PeerGetMessage *get;
1746   uint32_t xquery_size;
1747   size_t reply_bf_size;
1748   uint16_t msize;
1749   enum GNUNET_BLOCK_Type type;
1750   enum GNUNET_DHT_RouteOption options;
1751   enum GNUNET_BLOCK_EvaluationResult eval;
1752   struct GNUNET_CONTAINER_BloomFilter *reply_bf;
1753   struct GNUNET_CONTAINER_BloomFilter *peer_bf;
1754   const char *xquery;
1755
1756   GNUNET_break (0 != memcmp (peer, &my_identity, sizeof (struct GNUNET_PeerIdentity)));
1757   /* parse and validate message */
1758   msize = ntohs (message->size);
1759   if (msize < sizeof (struct PeerGetMessage))
1760   {
1761     GNUNET_break_op (0);
1762     return GNUNET_YES;
1763   }
1764   get = (struct PeerGetMessage *) message;
1765   xquery_size = ntohl (get->xquery_size);
1766   if (msize < sizeof (struct PeerGetMessage) + xquery_size)
1767   {
1768     GNUNET_break_op (0);
1769     return GNUNET_YES;
1770   }
1771   GNUNET_STATISTICS_update (GDS_stats,
1772                             gettext_noop ("# P2P GET requests received"), 1,
1773                             GNUNET_NO);
1774   reply_bf_size = msize - (sizeof (struct PeerGetMessage) + xquery_size);
1775   type = ntohl (get->type);
1776   options = ntohl (get->options);
1777   xquery = (const char*) &get[1];
1778   reply_bf = NULL;
1779   if (reply_bf_size > 0)
1780     reply_bf = GNUNET_CONTAINER_bloomfilter_init (&xquery[xquery_size],
1781                                                   reply_bf_size,
1782                                                   GNUNET_CONSTANTS_BLOOMFILTER_K);
1783   eval = GNUNET_BLOCK_evaluate (GDS_block_context,
1784                                 type,
1785                                 &get->key,
1786                                 &reply_bf,
1787                                 get->bf_mutator,
1788                                 xquery, xquery_size,
1789                                 NULL, 0);
1790   if (eval != GNUNET_BLOCK_EVALUATION_REQUEST_VALID)
1791   {
1792     /* request invalid or block type not supported */
1793     GNUNET_break_op (eval == GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED);
1794     if (NULL != reply_bf)
1795       GNUNET_CONTAINER_bloomfilter_free (reply_bf);
1796     return GNUNET_YES;
1797   }
1798   peer_bf =
1799     GNUNET_CONTAINER_bloomfilter_init (get->bloomfilter,
1800                                        DHT_BLOOM_SIZE,
1801                                        GNUNET_CONSTANTS_BLOOMFILTER_K);
1802   GNUNET_break_op (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (peer_bf, &peer->hashPubKey));
1803   /* remember request for routing replies */
1804   GDS_ROUTING_add (peer,
1805                    type,
1806                    options,
1807                    &get->key,
1808                    xquery, xquery_size,
1809                    reply_bf, get->bf_mutator);
1810   GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1811               "GET for %s at %s after %u hops\n",
1812               GNUNET_h2s (&get->key),
1813               GNUNET_i2s (&my_identity),
1814               (unsigned int) ntohl (get->hop_count));
1815   /* local lookup (this may update the reply_bf) */
1816   if ( (0 != (options & GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE)) ||
1817        (am_closest_peer (&get->key,
1818                          peer_bf) ) )
1819     {
1820     if ( (0 != (options & GNUNET_DHT_RO_FIND_PEER)))
1821     {
1822       GNUNET_STATISTICS_update (GDS_stats,
1823                                 gettext_noop ("# P2P FIND PEER requests processed"), 1,
1824                                 GNUNET_NO);
1825       handle_find_peer (peer,
1826                         &get->key,
1827                         reply_bf,
1828                         get->bf_mutator);
1829     }
1830     else
1831     {
1832       eval = GDS_DATACACHE_handle_get (&get->key,
1833                                        type,
1834                                        xquery, xquery_size,
1835                                        &reply_bf,
1836                                        get->bf_mutator);
1837     }
1838   }
1839   else
1840   {
1841     GNUNET_STATISTICS_update (GDS_stats,
1842                               gettext_noop ("# P2P GET requests ONLY routed"), 1,
1843                               GNUNET_NO);
1844   }
1845
1846   /* P2P forwarding */
1847   if (eval != GNUNET_BLOCK_EVALUATION_OK_LAST)
1848     GDS_NEIGHBOURS_handle_get (type,
1849                                options,
1850                                ntohl (get->desired_replication_level),
1851                                ntohl (get->hop_count),
1852                                &get->key,
1853                                xquery, xquery_size,
1854                                reply_bf,
1855                                get->bf_mutator,
1856                                peer_bf);
1857   /* clean up */
1858   if (NULL != reply_bf)
1859     GNUNET_CONTAINER_bloomfilter_free (reply_bf);
1860   GNUNET_CONTAINER_bloomfilter_free (peer_bf);
1861   return GNUNET_YES;
1862 }
1863
1864
1865 /**
1866  * Core handler for p2p result messages.
1867  *
1868  * @param cls closure
1869  * @param message message
1870  * @param peer peer identity this notification is about
1871  * @param atsi performance data
1872  * @return GNUNET_YES (do not cut p2p connection)
1873  */
1874 static int
1875 handle_dht_p2p_result (void *cls, const struct GNUNET_PeerIdentity *peer,
1876                        const struct GNUNET_MessageHeader *message,
1877                        const struct GNUNET_TRANSPORT_ATS_Information
1878                        *atsi)
1879 {
1880   const struct PeerResultMessage *prm;
1881   const struct GNUNET_PeerIdentity *put_path;
1882   const struct GNUNET_PeerIdentity *get_path;
1883   const void *data;
1884   uint32_t get_path_length;
1885   uint32_t put_path_length;
1886   uint16_t msize;
1887   size_t data_size;
1888   enum GNUNET_BLOCK_Type type;
1889
1890   /* parse and validate message */
1891   msize = ntohs (message->size);
1892   if (msize < sizeof (struct PeerResultMessage))
1893   {
1894     GNUNET_break_op (0);
1895     return GNUNET_YES;
1896   }
1897   prm = (struct PeerResultMessage *) message;
1898   put_path_length = ntohl (prm->put_path_length);
1899   get_path_length = ntohl (prm->get_path_length);
1900   if ( (msize < sizeof (struct PeerResultMessage) +
1901         (get_path_length + put_path_length) * sizeof (struct GNUNET_PeerIdentity)) ||
1902        (get_path_length > GNUNET_SERVER_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)) ||
1903        (put_path_length > GNUNET_SERVER_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)) )
1904   {
1905     GNUNET_break_op (0);
1906     return GNUNET_YES;
1907   }
1908   GNUNET_STATISTICS_update (GDS_stats,
1909                             gettext_noop ("# P2P RESULTS received"), 1,
1910                             GNUNET_NO);
1911   put_path = (const struct GNUNET_PeerIdentity*) &prm[1];
1912   get_path = &put_path[put_path_length];
1913   type = ntohl (prm->type);
1914   data = (const void*) &get_path[get_path_length];
1915   data_size = msize - (sizeof (struct PeerResultMessage) +
1916                        (get_path_length + put_path_length) * sizeof (struct GNUNET_PeerIdentity));
1917
1918   /* if we got a HELLO, consider it for our own routing table */
1919   if (type == GNUNET_BLOCK_TYPE_DHT_HELLO)
1920   {
1921     const struct GNUNET_MessageHeader *h;
1922     struct GNUNET_PeerIdentity pid;
1923     int bucket;
1924
1925     /* Should be a HELLO, validate and consider using it! */
1926     if (data_size < sizeof (struct GNUNET_MessageHeader))
1927     {
1928       GNUNET_break_op (0);
1929       return GNUNET_YES;
1930     }
1931     h = data;
1932     if (data_size != ntohs (h->size))
1933     {
1934       GNUNET_break_op (0);
1935       return GNUNET_YES;
1936     }
1937     if (GNUNET_OK !=
1938         GNUNET_HELLO_get_id ((const struct GNUNET_HELLO_Message*) h,
1939                              &pid))
1940     {
1941       GNUNET_break_op (0);
1942       return GNUNET_YES;
1943     }
1944     if (0 != memcmp (&my_identity, &pid, sizeof (struct GNUNET_PeerIdentity)))
1945     {
1946       bucket = find_bucket (&pid.hashPubKey);
1947       if ( (bucket >= 0) &&
1948            (k_buckets[bucket].peers_size < bucket_size) )
1949         {
1950           if (NULL != GDS_transport_handle)
1951             GNUNET_TRANSPORT_offer_hello (GDS_transport_handle,
1952                                           h, NULL, NULL);
1953           (void) GNUNET_CORE_peer_request_connect (coreAPI,
1954                                                    &pid,
1955                                                    NULL, NULL);
1956         }
1957     }
1958   }
1959
1960   /* append 'peer' to 'get_path' */
1961   {
1962     struct GNUNET_PeerIdentity xget_path[get_path_length+1];
1963
1964     memcpy (xget_path, get_path, get_path_length * sizeof (struct GNUNET_PeerIdentity));
1965     xget_path[get_path_length] = *peer;
1966     get_path_length++;
1967
1968     /* forward to local clients */
1969     GDS_CLIENTS_handle_reply (GNUNET_TIME_absolute_ntoh (prm->expiration_time),
1970                              &prm->key,
1971                              get_path_length,
1972                              xget_path,
1973                              put_path_length,
1974                              put_path,
1975                              type,
1976                              data_size,
1977                              data);
1978
1979     /* forward to other peers */
1980     GDS_ROUTING_process (type,
1981                          GNUNET_TIME_absolute_ntoh (prm->expiration_time),
1982                          &prm->key,
1983                          put_path_length,
1984                          put_path,
1985                          get_path_length,
1986                          xget_path,
1987                          data,
1988                          data_size);
1989   }
1990   return GNUNET_YES;
1991 }
1992
1993
1994 /**
1995  * Initialize neighbours subsystem.
1996  *
1997  * @return GNUNET_OK on success, GNUNET_SYSERR on error
1998  */
1999 int
2000 GDS_NEIGHBOURS_init ()
2001 {
2002   static struct GNUNET_CORE_MessageHandler core_handlers[] = {
2003     {&handle_dht_p2p_get, GNUNET_MESSAGE_TYPE_DHT_P2P_GET, 0},
2004     {&handle_dht_p2p_put, GNUNET_MESSAGE_TYPE_DHT_P2P_PUT, 0},
2005     {&handle_dht_p2p_result, GNUNET_MESSAGE_TYPE_DHT_P2P_RESULT, 0},
2006     {NULL, 0, 0}
2007   };
2008   unsigned long long temp_config_num;
2009
2010   if (GNUNET_OK ==
2011       GNUNET_CONFIGURATION_get_value_number (GDS_cfg, "DHT", "bucket_size",
2012                                              &temp_config_num))
2013     bucket_size = (unsigned int) temp_config_num;
2014   coreAPI = GNUNET_CORE_connect (GDS_cfg,
2015                                  1,
2016                                  NULL,
2017                                  &core_init,
2018                                  &handle_core_connect,
2019                                  &handle_core_disconnect,
2020                                  NULL,  /* Do we care about "status" updates? */
2021                                  NULL, GNUNET_NO,
2022                                  NULL, GNUNET_NO,
2023                                  core_handlers);
2024   if (coreAPI == NULL)
2025     return GNUNET_SYSERR;
2026   all_known_peers = GNUNET_CONTAINER_multihashmap_create (256);
2027   return GNUNET_OK;
2028 }
2029
2030
2031 /**
2032  * Shutdown neighbours subsystem.
2033  */
2034 void
2035 GDS_NEIGHBOURS_done ()
2036 {
2037   if (coreAPI == NULL)
2038     return;
2039   GNUNET_CORE_disconnect (coreAPI);
2040   coreAPI = NULL;
2041   GNUNET_assert (0 == GNUNET_CONTAINER_multihashmap_size (all_known_peers));
2042   GNUNET_CONTAINER_multihashmap_destroy (all_known_peers);
2043   all_known_peers = NULL;
2044   if (GNUNET_SCHEDULER_NO_TASK != find_peer_task)
2045   {
2046     GNUNET_SCHEDULER_cancel (find_peer_task);
2047     find_peer_task = GNUNET_SCHEDULER_NO_TASK;
2048   }
2049 }
2050
2051
2052 /* end of gnunet-service-dht_neighbours.c */