src/dht/gnunet-service-dht_neighbours.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2009-2017 GNUnet e.V.
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, USA.
  19 */
  20
  21 /**
  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 #include "platform.h"
  28 #include "gnunet_util_lib.h"
  29 #include "gnunet_block_lib.h"
  30 #include "gnunet_hello_lib.h"
  31 #include "gnunet_constants.h"
  32 #include "gnunet_protocols.h"
  33 #include "gnunet_nse_service.h"
  34 #include "gnunet_ats_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_datacache.h"
  43 #include "gnunet-service-dht_hello.h"
  44 #include "gnunet-service-dht_neighbours.h"
  45 #include "gnunet-service-dht_nse.h"
  46 #include "gnunet-service-dht_routing.h"
  47 #include "dht.h"
  48
  49 #define LOG_TRAFFIC(kind,...) GNUNET_log_from (kind, "dht-traffic",__VA_ARGS__)
  50
  51 /**
  52  * Enable slow sanity checks to debug issues.
  53  */
  54 #define SANITY_CHECKS 1
  55
  56 /**
  57  * How many buckets will we allow total.
  58  */
  59 #define MAX_BUCKETS sizeof (struct GNUNET_HashCode) * 8
  60
  61 /**
  62  * What is the maximum number of peers in a given bucket.
  63  */
  64 #define DEFAULT_BUCKET_SIZE 8
  65
  66 /**
  67  * Desired replication level for FIND PEER requests
  68  */
  69 #define FIND_PEER_REPLICATION_LEVEL 4
  70
  71 /**
  72  * Maximum allowed replication level for all requests.
  73  */
  74 #define MAXIMUM_REPLICATION_LEVEL 16
  75
  76 /**
  77  * Maximum allowed number of pending messages per peer.
  78  */
  79 #define MAXIMUM_PENDING_PER_PEER 64
  80
  81 /**
  82  * How long at least to wait before sending another find peer request.
  83  */
  84 #define DHT_MINIMUM_FIND_PEER_INTERVAL GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS, 30)
  85
  86 /**
  87  * How long at most to wait before sending another find peer request.
  88  */
  89 #define DHT_MAXIMUM_FIND_PEER_INTERVAL GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 10)
  90
  91 /**
  92  * How long at most to wait for transmission of a GET request to another peer?
  93  */
  94 #define GET_TIMEOUT GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 2)
  95
  96 /**
  97  * Hello address expiration
  98  */
  99 extern struct GNUNET_TIME_Relative hello_expiration;
 100
 101
 102 GNUNET_NETWORK_STRUCT_BEGIN
 103
 104 /**
 105  * P2P PUT message
 106  */
 107 struct PeerPutMessage
 108 {
 109   /**
 110    * Type: #GNUNET_MESSAGE_TYPE_DHT_P2P_PUT
 111    */
 112   struct GNUNET_MessageHeader header;
 113
 114   /**
 115    * Processing options
 116    */
 117   uint32_t options GNUNET_PACKED;
 118
 119   /**
 120    * Content type.
 121    */
 122   uint32_t type GNUNET_PACKED;
 123
 124   /**
 125    * Hop count
 126    */
 127   uint32_t hop_count GNUNET_PACKED;
 128
 129   /**
 130    * Replication level for this message
 131    */
 132   uint32_t desired_replication_level GNUNET_PACKED;
 133
 134   /**
 135    * Length of the PUT path that follows (if tracked).
 136    */
 137   uint32_t put_path_length GNUNET_PACKED;
 138
 139   /**
 140    * When does the content expire?
 141    */
 142   struct GNUNET_TIME_AbsoluteNBO expiration_time;
 143
 144   /**
 145    * Bloomfilter (for peer identities) to stop circular routes
 146    */
 147   char bloomfilter[DHT_BLOOM_SIZE];
 148
 149   /**
 150    * The key we are storing under.
 151    */
 152   struct GNUNET_HashCode key;
 153
 154   /* put path (if tracked) */
 155
 156   /* Payload */
 157
 158 };
 159
 160
 161 /**
 162  * P2P Result message
 163  */
 164 struct PeerResultMessage
 165 {
 166   /**
 167    * Type: #GNUNET_MESSAGE_TYPE_DHT_P2P_RESULT
 168    */
 169   struct GNUNET_MessageHeader header;
 170
 171   /**
 172    * Content type.
 173    */
 174   uint32_t type GNUNET_PACKED;
 175
 176   /**
 177    * Length of the PUT path that follows (if tracked).
 178    */
 179   uint32_t put_path_length GNUNET_PACKED;
 180
 181   /**
 182    * Length of the GET path that follows (if tracked).
 183    */
 184   uint32_t get_path_length GNUNET_PACKED;
 185
 186   /**
 187    * When does the content expire?
 188    */
 189   struct GNUNET_TIME_AbsoluteNBO expiration_time;
 190
 191   /**
 192    * The key of the corresponding GET request.
 193    */
 194   struct GNUNET_HashCode key;
 195
 196   /* put path (if tracked) */
 197
 198   /* get path (if tracked) */
 199
 200   /* Payload */
 201
 202 };
 203
 204
 205 /**
 206  * P2P GET message
 207  */
 208 struct PeerGetMessage
 209 {
 210   /**
 211    * Type: #GNUNET_MESSAGE_TYPE_DHT_P2P_GET
 212    */
 213   struct GNUNET_MessageHeader header;
 214
 215   /**
 216    * Processing options
 217    */
 218   uint32_t options GNUNET_PACKED;
 219
 220   /**
 221    * Desired content type.
 222    */
 223   uint32_t type GNUNET_PACKED;
 224
 225   /**
 226    * Hop count
 227    */
 228   uint32_t hop_count GNUNET_PACKED;
 229
 230   /**
 231    * Desired replication level for this request.
 232    */
 233   uint32_t desired_replication_level GNUNET_PACKED;
 234
 235   /**
 236    * Size of the extended query.
 237    */
 238   uint32_t xquery_size;
 239
 240   /**
 241    * Bloomfilter mutator.
 242    */
 243   uint32_t bf_mutator;
 244
 245   /**
 246    * Bloomfilter (for peer identities) to stop circular routes
 247    */
 248   char bloomfilter[DHT_BLOOM_SIZE];
 249
 250   /**
 251    * The key we are looking for.
 252    */
 253   struct GNUNET_HashCode key;
 254
 255   /* xquery */
 256
 257   /* result bloomfilter */
 258
 259 };
 260 GNUNET_NETWORK_STRUCT_END
 261
 262
 263 /**
 264  * Entry for a peer in a bucket.
 265  */
 266 struct PeerInfo
 267 {
 268   /**
 269    * Next peer entry (DLL)
 270    */
 271   struct PeerInfo *next;
 272
 273   /**
 274    *  Prev peer entry (DLL)
 275    */
 276   struct PeerInfo *prev;
 277
 278   /**
 279    * Handle for sending messages to this peer.
 280    */
 281   struct GNUNET_MQ_Handle *mq;
 282
 283   /**
 284    * What is the identity of the peer?
 285    */
 286   const struct GNUNET_PeerIdentity *id;
 287
 288   /**
 289    * Hash of @e id.
 290    */
 291   struct GNUNET_HashCode phash;
 292
 293   /**
 294    * Which bucket is this peer in?
 295    */
 296   int peer_bucket;
 297
 298 };
 299
 300
 301 /**
 302  * Peers are grouped into buckets.
 303  */
 304 struct PeerBucket
 305 {
 306   /**
 307    * Head of DLL
 308    */
 309   struct PeerInfo *head;
 310
 311   /**
 312    * Tail of DLL
 313    */
 314   struct PeerInfo *tail;
 315
 316   /**
 317    * Number of peers in the bucket.
 318    */
 319   unsigned int peers_size;
 320 };
 321
 322
 323 /**
 324  * Information about a peer that we would like to connect to.
 325  */
 326 struct ConnectInfo
 327 {
 328
 329   /**
 330    * Handle to active HELLO offer operation, or NULL.
 331    */
 332   struct GNUNET_TRANSPORT_OfferHelloHandle *oh;
 333
 334   /**
 335    * Handle to active connectivity suggestion operation, or NULL.
 336    */
 337   struct GNUNET_ATS_ConnectivitySuggestHandle *sh;
 338
 339   /**
 340    * How much would we like to connect to this peer?
 341    */
 342   uint32_t strength;
 343 };
 344
 345
 346 /**
 347  * Do we cache all results that we are routing in the local datacache?
 348  */
 349 static int cache_results;
 350
 351 /**
 352  * Should routing details be logged to stderr (for debugging)?
 353  */
 354 static int log_route_details_stderr;
 355
 356 /**
 357  * The lowest currently used bucket, initially 0 (for 0-bits matching bucket).
 358  */
 359 static unsigned int closest_bucket;
 360
 361 /**
 362  * How many peers have we added since we sent out our last
 363  * find peer request?
 364  */
 365 static unsigned int newly_found_peers;
 366
 367 /**
 368  * Option for testing that disables the 'connect' function of the DHT.
 369  */
 370 static int disable_try_connect;
 371
 372 /**
 373  * The buckets.  Array of size #MAX_BUCKETS.  Offset 0 means 0 bits matching.
 374  */
 375 static struct PeerBucket k_buckets[MAX_BUCKETS];
 376
 377 /**
 378  * Hash map of all CORE-connected peers, for easy removal from
 379  * #k_buckets on disconnect.  Values are of type `struct PeerInfo`.
 380  */
 381 static struct GNUNET_CONTAINER_MultiPeerMap *all_connected_peers;
 382
 383 /**
 384  * Hash map of all peers we would like to be connected to.
 385  * Values are of type `struct ConnectInfo`.
 386  */
 387 static struct GNUNET_CONTAINER_MultiPeerMap *all_desired_peers;
 388
 389 /**
 390  * Maximum size for each bucket.
 391  */
 392 static unsigned int bucket_size = DEFAULT_BUCKET_SIZE;
 393
 394 /**
 395  * Task that sends FIND PEER requests.
 396  */
 397 static struct GNUNET_SCHEDULER_Task *find_peer_task;
 398
 399 /**
 400  * Identity of this peer.
 401  */
 402 static struct GNUNET_PeerIdentity my_identity;
 403
 404 /**
 405  * Hash of the identity of this peer.
 406  */
 407 static struct GNUNET_HashCode my_identity_hash;
 408
 409 /**
 410  * Handle to CORE.
 411  */
 412 static struct GNUNET_CORE_Handle *core_api;
 413
 414 /**
 415  * Handle to ATS connectivity.
 416  */
 417 static struct GNUNET_ATS_ConnectivityHandle *ats_ch;
 418
 419
 420 /**
 421  * Find the optimal bucket for this key.
 422  *
 423  * @param hc the hashcode to compare our identity to
 424  * @return the proper bucket index, or GNUNET_SYSERR
 425  *         on error (same hashcode)
 426  */
 427 static int
 428 find_bucket (const struct GNUNET_HashCode *hc)
 429 {
 430   unsigned int bits;
 431
 432   bits = GNUNET_CRYPTO_hash_matching_bits (&my_identity_hash, hc);
 433   if (bits == MAX_BUCKETS)
 434   {
 435     /* How can all bits match? Got my own ID? */
 436     GNUNET_break (0);
 437     return GNUNET_SYSERR;
 438   }
 439   return MAX_BUCKETS - bits - 1;
 440 }
 441
 442
 443 /**
 444  * Function called when #GNUNET_TRANSPORT_offer_hello() is done.
 445  * Clean up the "oh" field in the @a cls
 446  *
 447  * @param cls a `struct ConnectInfo`
 448  */
 449 static void
 450 offer_hello_done (void *cls)
 451 {
 452   struct ConnectInfo *ci = cls;
 453
 454   ci->oh = NULL;
 455 }
 456
 457
 458 /**
 459  * Function called for all entries in #all_desired_peers to clean up.
 460  *
 461  * @param cls NULL
 462  * @param peer peer the entry is for
 463  * @param value the value to remove
 464  * @return #GNUNET_YES
 465  */
 466 static int
 467 free_connect_info (void *cls,
 468                    const struct GNUNET_PeerIdentity *peer,
 469                    void *value)
 470 {
 471   struct ConnectInfo *ci = value;
 472
 473   GNUNET_assert (GNUNET_YES ==
 474                  GNUNET_CONTAINER_multipeermap_remove (all_desired_peers,
 475                                                        peer,
 476                                                        ci));
 477   if (NULL != ci->sh)
 478   {
 479     GNUNET_ATS_connectivity_suggest_cancel (ci->sh);
 480     ci->sh = NULL;
 481   }
 482   if (NULL != ci->oh)
 483   {
 484     GNUNET_TRANSPORT_offer_hello_cancel (ci->oh);
 485     ci->oh = NULL;
 486   }
 487   GNUNET_free (ci);
 488   return GNUNET_YES;
 489 }
 490
 491
 492 /**
 493  * Consider if we want to connect to a given peer, and if so
 494  * let ATS know.  If applicable, the HELLO is offered to the
 495  * TRANSPORT service.
 496  *
 497  * @param pid peer to consider connectivity requirements for
 498  * @param h a HELLO message, or NULL
 499  */
 500 static void
 501 try_connect (const struct GNUNET_PeerIdentity *pid,
 502              const struct GNUNET_MessageHeader *h)
 503 {
 504   int bucket;
 505   struct GNUNET_HashCode pid_hash;
 506   struct ConnectInfo *ci;
 507   uint32_t strength;
 508
 509   GNUNET_CRYPTO_hash (pid,
 510                       sizeof (struct GNUNET_PeerIdentity),
 511                       &pid_hash);
 512   bucket = find_bucket (&pid_hash);
 513   if (bucket < 0)
 514     return; /* self? */
 515   ci = GNUNET_CONTAINER_multipeermap_get (all_desired_peers,
 516                                           pid);
 517
 518   if (k_buckets[bucket].peers_size < bucket_size)
 519     strength = (bucket_size - k_buckets[bucket].peers_size) * bucket;
 520   else
 521     strength = bucket; /* minimum value of connectivity */
 522   if (GNUNET_YES ==
 523       GNUNET_CONTAINER_multipeermap_contains (all_connected_peers,
 524                                               pid))
 525     strength *= 2; /* double for connected peers */
 526   else if (k_buckets[bucket].peers_size > bucket_size)
 527     strength = 0; /* bucket full, we really do not care about more */
 528
 529   if ( (0 == strength) &&
 530        (NULL != ci) )
 531   {
 532     /* release request */
 533     GNUNET_assert (GNUNET_YES ==
 534                    free_connect_info (NULL,
 535                                       pid,
 536                                       ci));
 537     return;
 538   }
 539   if (NULL == ci)
 540   {
 541     ci = GNUNET_new (struct ConnectInfo);
 542     GNUNET_assert (GNUNET_OK ==
 543                    GNUNET_CONTAINER_multipeermap_put (all_desired_peers,
 544                                                       pid,
 545                                                       ci,
 546                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 547   }
 548   if ( (NULL != ci->oh) &&
 549        (NULL != h) )
 550     GNUNET_TRANSPORT_offer_hello_cancel (ci->oh);
 551   if (NULL != h)
 552     ci->oh = GNUNET_TRANSPORT_offer_hello (GDS_cfg,
 553                                            h,
 554                                            &offer_hello_done,
 555                                            ci);
 556   if ( (NULL != ci->sh) &&
 557        (ci->strength != strength) )
 558     GNUNET_ATS_connectivity_suggest_cancel (ci->sh);
 559   if (ci->strength != strength)
 560     ci->sh = GNUNET_ATS_connectivity_suggest (ats_ch,
 561                                               pid,
 562                                               strength);
 563   ci->strength = strength;
 564 }
 565
 566
 567 /**
 568  * Function called for each peer in #all_desired_peers during
 569  * #update_connect_preferences() if we have reason to adjust
 570  * the strength of our desire to keep connections to certain
 571  * peers.  Calls #try_connect() to update the calculations for
 572  * the given @a pid.
 573  *
 574  * @param cls NULL
 575  * @param pid peer to update
 576  * @param value unused
 577  * @return #GNUNET_YES (continue to iterate)
 578  */
 579 static int
 580 update_desire_strength (void *cls,
 581                         const struct GNUNET_PeerIdentity *pid,
 582                         void *value)
 583 {
 584   try_connect (pid, NULL);
 585   return GNUNET_YES;
 586 }
 587
 588
 589 /**
 590  * Update our preferences for connectivity as given to ATS.
 591  *
 592  * @param cls the `struct PeerInfo` of the peer
 593  * @param tc scheduler context.
 594  */
 595 static void
 596 update_connect_preferences ()
 597 {
 598   GNUNET_CONTAINER_multipeermap_iterate (all_desired_peers,
 599                                          &update_desire_strength,
 600                                          NULL);
 601 }
 602
 603
 604 /**
 605  * Add each of the peers we already know to the bloom filter of
 606  * the request so that we don't get duplicate HELLOs.
 607  *
 608  * @param cls the `struct GNUNET_BLOCK_Group`
 609  * @param key peer identity to add to the bloom filter
 610  * @param value value the peer information (unused)
 611  * @return #GNUNET_YES (we should continue to iterate)
 612  */
 613 static int
 614 add_known_to_bloom (void *cls,
 615                     const struct GNUNET_PeerIdentity *key,
 616                     void *value)
 617 {
 618   struct GNUNET_BLOCK_Group *bg = cls;
 619   struct GNUNET_HashCode key_hash;
 620
 621   GNUNET_CRYPTO_hash (key,
 622                       sizeof (struct GNUNET_PeerIdentity),
 623                       &key_hash);
 624   GNUNET_BLOCK_group_set_seen (bg,
 625                                &key_hash,
 626                                1);
 627   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 628               "Adding known peer (%s) to bloomfilter for FIND PEER\n",
 629               GNUNET_i2s (key));
 630   return GNUNET_YES;
 631 }
 632
 633
 634 /**
 635  * Task to send a find peer message for our own peer identifier
 636  * so that we can find the closest peers in the network to ourselves
 637  * and attempt to connect to them.
 638  *
 639  * @param cls closure for this task
 640  */
 641 static void
 642 send_find_peer_message (void *cls)
 643 {
 644   struct GNUNET_TIME_Relative next_send_time;
 645   struct GNUNET_BLOCK_Group *bg;
 646   struct GNUNET_CONTAINER_BloomFilter *peer_bf;
 647
 648   find_peer_task = NULL;
 649   if (newly_found_peers > bucket_size)
 650   {
 651     /* If we are finding many peers already, no need to send out our request right now! */
 652     find_peer_task =
 653         GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
 654                                       &send_find_peer_message,
 655                                       NULL);
 656     newly_found_peers = 0;
 657     return;
 658   }
 659   bg = GNUNET_BLOCK_group_create (GDS_block_context,
 660                                   GNUNET_BLOCK_TYPE_DHT_HELLO,
 661                                   GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
 662                                                             UINT32_MAX),
 663                                   NULL,
 664                                   0,
 665                                   "filter-size",
 666                                   DHT_BLOOM_SIZE,
 667                                   NULL);
 668   GNUNET_CONTAINER_multipeermap_iterate (all_connected_peers,
 669                                          &add_known_to_bloom,
 670                                          bg);
 671   GNUNET_STATISTICS_update (GDS_stats,
 672                             gettext_noop ("# FIND PEER messages initiated"),
 673                             1,
 674                             GNUNET_NO);
 675   peer_bf
 676     = GNUNET_CONTAINER_bloomfilter_init (NULL,
 677                                          DHT_BLOOM_SIZE,
 678                                          GNUNET_CONSTANTS_BLOOMFILTER_K);
 679   // FIXME: pass priority!?
 680   GDS_NEIGHBOURS_handle_get (GNUNET_BLOCK_TYPE_DHT_HELLO,
 681                              GNUNET_DHT_RO_FIND_PEER | GNUNET_DHT_RO_RECORD_ROUTE,
 682                              FIND_PEER_REPLICATION_LEVEL,
 683                              0,
 684                              &my_identity_hash,
 685                              NULL,
 686                              0,
 687                              bg,
 688                              peer_bf);
 689   GNUNET_CONTAINER_bloomfilter_free (peer_bf);
 690   GNUNET_BLOCK_group_destroy (bg);
 691   /* schedule next round */
 692   next_send_time.rel_value_us =
 693       DHT_MINIMUM_FIND_PEER_INTERVAL.rel_value_us +
 694       GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
 695                                 DHT_MAXIMUM_FIND_PEER_INTERVAL.rel_value_us /
 696                                 (newly_found_peers + 1));
 697   newly_found_peers = 0;
 698   GNUNET_assert (NULL == find_peer_task);
 699   find_peer_task =
 700       GNUNET_SCHEDULER_add_delayed (next_send_time,
 701                                     &send_find_peer_message,
 702                                     NULL);
 703 }
 704
 705
 706 /**
 707  * Method called whenever a peer connects.
 708  *
 709  * @param cls closure
 710  * @param peer peer identity this notification is about
 711  * @param mq message queue for sending messages to @a peer
 712  * @return our `struct PeerInfo` for @a peer
 713  */
 714 static void *
 715 handle_core_connect (void *cls,
 716                      const struct GNUNET_PeerIdentity *peer,
 717                      struct GNUNET_MQ_Handle *mq)
 718 {
 719   struct PeerInfo *pi;
 720
 721   /* Check for connect to self message */
 722   if (0 == memcmp (&my_identity,
 723                    peer,
 724                    sizeof (struct GNUNET_PeerIdentity)))
 725     return NULL;
 726   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 727               "Connected to %s\n",
 728               GNUNET_i2s (peer));
 729   GNUNET_assert (GNUNET_NO ==
 730                  GNUNET_CONTAINER_multipeermap_get (all_connected_peers,
 731                                                     peer));
 732   GNUNET_STATISTICS_update (GDS_stats,
 733                             gettext_noop ("# peers connected"),
 734                             1,
 735                             GNUNET_NO);
 736   pi = GNUNET_new (struct PeerInfo);
 737   pi->id = peer;
 738   pi->mq = mq;
 739   GNUNET_CRYPTO_hash (peer,
 740                       sizeof (struct GNUNET_PeerIdentity),
 741                       &pi->phash);
 742   pi->peer_bucket = find_bucket (&pi->phash);
 743   GNUNET_assert ( (pi->peer_bucket >= 0) &&
 744                   (pi->peer_bucket < MAX_BUCKETS) );
 745   GNUNET_CONTAINER_DLL_insert_tail (k_buckets[pi->peer_bucket].head,
 746                                     k_buckets[pi->peer_bucket].tail,
 747                                     pi);
 748   k_buckets[pi->peer_bucket].peers_size++;
 749   closest_bucket = GNUNET_MAX (closest_bucket,
 750                                pi->peer_bucket);
 751   GNUNET_assert (GNUNET_OK ==
 752                  GNUNET_CONTAINER_multipeermap_put (all_connected_peers,
 753                                                     pi->id,
 754                                                     pi,
 755                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 756   if ( (pi->peer_bucket > 0) &&
 757        (k_buckets[pi->peer_bucket].peers_size <= bucket_size))
 758   {
 759     update_connect_preferences ();
 760     newly_found_peers++;
 761   }
 762   if ( (1 == GNUNET_CONTAINER_multipeermap_size (all_connected_peers)) &&
 763        (GNUNET_YES != disable_try_connect))
 764   {
 765     /* got a first connection, good time to start with FIND PEER requests... */
 766     GNUNET_assert (NULL == find_peer_task);
 767     find_peer_task = GNUNET_SCHEDULER_add_now (&send_find_peer_message,
 768                                                NULL);
 769   }
 770   return pi;
 771 }
 772
 773
 774 /**
 775  * Method called whenever a peer disconnects.
 776  *
 777  * @param cls closure
 778  * @param peer peer identity this notification is about
 779  * @param internal_cls our `struct PeerInfo` for @a peer
 780  */
 781 static void
 782 handle_core_disconnect (void *cls,
 783                         const struct GNUNET_PeerIdentity *peer,
 784                         void *internal_cls)
 785 {
 786   struct PeerInfo *to_remove = internal_cls;
 787
 788   /* Check for disconnect from self message */
 789   if (NULL == to_remove)
 790     return;
 791   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 792               "Disconnected %s\n",
 793               GNUNET_i2s (peer));
 794   GNUNET_STATISTICS_update (GDS_stats,
 795                             gettext_noop ("# peers connected"),
 796                             -1,
 797                             GNUNET_NO);
 798   GNUNET_assert (GNUNET_YES ==
 799                  GNUNET_CONTAINER_multipeermap_remove (all_connected_peers,
 800                                                        peer,
 801                                                        to_remove));
 802   if ( (0 == GNUNET_CONTAINER_multipeermap_size (all_connected_peers)) &&
 803        (GNUNET_YES != disable_try_connect) )
 804   {
 805     GNUNET_SCHEDULER_cancel (find_peer_task);
 806     find_peer_task = NULL;
 807   }
 808   GNUNET_assert (to_remove->peer_bucket >= 0);
 809   GNUNET_CONTAINER_DLL_remove (k_buckets[to_remove->peer_bucket].head,
 810                                k_buckets[to_remove->peer_bucket].tail,
 811                                to_remove);
 812   GNUNET_assert (k_buckets[to_remove->peer_bucket].peers_size > 0);
 813   k_buckets[to_remove->peer_bucket].peers_size--;
 814   while ( (closest_bucket > 0) &&
 815           (0 == k_buckets[to_remove->peer_bucket].peers_size) )
 816     closest_bucket--;
 817   if (k_buckets[to_remove->peer_bucket].peers_size < bucket_size)
 818     update_connect_preferences ();
 819   GNUNET_free (to_remove);
 820 }
 821
 822
 823 /**
 824  * To how many peers should we (on average) forward the request to
 825  * obtain the desired target_replication count (on average).
 826  *
 827  * @param hop_count number of hops the message has traversed
 828  * @param target_replication the number of total paths desired
 829  * @return Some number of peers to forward the message to
 830  */
 831 static unsigned int
 832 get_forward_count (uint32_t hop_count,
 833                    uint32_t target_replication)
 834 {
 835   uint32_t random_value;
 836   uint32_t forward_count;
 837   float target_value;
 838
 839   if (hop_count > GDS_NSE_get () * 4.0)
 840   {
 841     /* forcefully terminate */
 842     GNUNET_STATISTICS_update (GDS_stats,
 843                               gettext_noop ("# requests TTL-dropped"),
 844                               1, GNUNET_NO);
 845     return 0;
 846   }
 847   if (hop_count > GDS_NSE_get () * 2.0)
 848   {
 849     /* Once we have reached our ideal number of hops, only forward to 1 peer */
 850     return 1;
 851   }
 852   /* bound by system-wide maximum */
 853   target_replication =
 854       GNUNET_MIN (MAXIMUM_REPLICATION_LEVEL, target_replication);
 855   target_value =
 856       1 + (target_replication - 1.0) / (GDS_NSE_get () +
 857                                         ((float) (target_replication - 1.0) *
 858                                          hop_count));
 859   /* Set forward count to floor of target_value */
 860   forward_count = (uint32_t) target_value;
 861   /* Subtract forward_count (floor) from target_value (yields value between 0 and 1) */
 862   target_value = target_value - forward_count;
 863   random_value =
 864       GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, UINT32_MAX);
 865   if (random_value < (target_value * UINT32_MAX))
 866     forward_count++;
 867   return forward_count;
 868 }
 869
 870
 871 /**
 872  * Compute the distance between have and target as a 32-bit value.
 873  * Differences in the lower bits must count stronger than differences
 874  * in the higher bits.
 875  *
 876  * @param target
 877  * @param have
 878  * @return 0 if have==target, otherwise a number
 879  *           that is larger as the distance between
 880  *           the two hash codes increases
 881  */
 882 static unsigned int
 883 get_distance (const struct GNUNET_HashCode *target,
 884               const struct GNUNET_HashCode *have)
 885 {
 886   unsigned int bucket;
 887   unsigned int msb;
 888   unsigned int lsb;
 889   unsigned int i;
 890
 891   /* We have to represent the distance between two 2^9 (=512)-bit
 892    * numbers as a 2^5 (=32)-bit number with "0" being used for the
 893    * two numbers being identical; furthermore, we need to
 894    * guarantee that a difference in the number of matching
 895    * bits is always represented in the result.
 896    *
 897    * We use 2^32/2^9 numerical values to distinguish between
 898    * hash codes that have the same LSB bit distance and
 899    * use the highest 2^9 bits of the result to signify the
 900    * number of (mis)matching LSB bits; if we have 0 matching
 901    * and hence 512 mismatching LSB bits we return -1 (since
 902    * 512 itself cannot be represented with 9 bits) */
 903
 904   /* first, calculate the most significant 9 bits of our
 905    * result, aka the number of LSBs */
 906   bucket = GNUNET_CRYPTO_hash_matching_bits (target,
 907                                              have);
 908   /* bucket is now a value between 0 and 512 */
 909   if (bucket == 512)
 910     return 0;                   /* perfect match */
 911   if (bucket == 0)
 912     return (unsigned int) -1;   /* LSB differs; use max (if we did the bit-shifting
 913                                  * below, we'd end up with max+1 (overflow)) */
 914
 915   /* calculate the most significant bits of the final result */
 916   msb = (512 - bucket) << (32 - 9);
 917   /* calculate the 32-9 least significant bits of the final result by
 918    * looking at the differences in the 32-9 bits following the
 919    * mismatching bit at 'bucket' */
 920   lsb = 0;
 921   for (i = bucket + 1;
 922        (i < sizeof (struct GNUNET_HashCode) * 8) && (i < bucket + 1 + 32 - 9); i++)
 923   {
 924     if (GNUNET_CRYPTO_hash_get_bit (target, i) !=
 925         GNUNET_CRYPTO_hash_get_bit (have, i))
 926       lsb |= (1 << (bucket + 32 - 9 - i));      /* first bit set will be 10,
 927                                                  * last bit set will be 31 -- if
 928                                                  * i does not reach 512 first... */
 929   }
 930   return msb | lsb;
 931 }
 932
 933
 934 /**
 935  * Check whether my identity is closer than any known peers.  If a
 936  * non-null bloomfilter is given, check if this is the closest peer
 937  * that hasn't already been routed to.
 938  *
 939  * @param key hash code to check closeness to
 940  * @param bloom bloomfilter, exclude these entries from the decision
 941  * @return #GNUNET_YES if node location is closest,
 942  *         #GNUNET_NO otherwise.
 943  */
 944 static int
 945 am_closest_peer (const struct GNUNET_HashCode *key,
 946                  const struct GNUNET_CONTAINER_BloomFilter *bloom)
 947 {
 948   int bits;
 949   int other_bits;
 950   int bucket_num;
 951   int count;
 952   struct PeerInfo *pos;
 953
 954   if (0 == memcmp (&my_identity_hash, key, sizeof (struct GNUNET_HashCode)))
 955     return GNUNET_YES;
 956   bucket_num = find_bucket (key);
 957   GNUNET_assert (bucket_num >= 0);
 958   bits = GNUNET_CRYPTO_hash_matching_bits (&my_identity_hash,
 959                                            key);
 960   pos = k_buckets[bucket_num].head;
 961   count = 0;
 962   while ((NULL != pos) && (count < bucket_size))
 963   {
 964     if ((NULL != bloom) &&
 965         (GNUNET_YES ==
 966          GNUNET_CONTAINER_bloomfilter_test (bloom,
 967                                             &pos->phash)))
 968     {
 969       pos = pos->next;
 970       continue;                 /* Skip already checked entries */
 971     }
 972     other_bits = GNUNET_CRYPTO_hash_matching_bits (&pos->phash,
 973                                                    key);
 974     if (other_bits > bits)
 975       return GNUNET_NO;
 976     if (other_bits == bits)     /* We match the same number of bits */
 977       return GNUNET_YES;
 978     pos = pos->next;
 979   }
 980   /* No peers closer, we are the closest! */
 981   return GNUNET_YES;
 982 }
 983
 984
 985 /**
 986  * Select a peer from the routing table that would be a good routing
 987  * destination for sending a message for "key".  The resulting peer
 988  * must not be in the set of blocked peers.<p>
 989  *
 990  * Note that we should not ALWAYS select the closest peer to the
 991  * target, peers further away from the target should be chosen with
 992  * exponentially declining probability.
 993  *
 994  * FIXME: double-check that this is fine
 995  *
 996  *
 997  * @param key the key we are selecting a peer to route to
 998  * @param bloom a bloomfilter containing entries this request has seen already
 999  * @param hops how many hops has this message traversed thus far
1000  * @return Peer to route to, or NULL on error
1001  */
1002 static struct PeerInfo *
1003 select_peer (const struct GNUNET_HashCode *key,
1004              const struct GNUNET_CONTAINER_BloomFilter *bloom,
1005              uint32_t hops)
1006 {
1007   unsigned int bc;
1008   unsigned int count;
1009   unsigned int selected;
1010   struct PeerInfo *pos;
1011   unsigned int dist;
1012   unsigned int smallest_distance;
1013   struct PeerInfo *chosen;
1014
1015   if (hops >= GDS_NSE_get ())
1016   {
1017     /* greedy selection (closest peer that is not in bloomfilter) */
1018     smallest_distance = UINT_MAX;
1019     chosen = NULL;
1020     for (bc = 0; bc <= closest_bucket; bc++)
1021     {
1022       pos = k_buckets[bc].head;
1023       count = 0;
1024       while ((pos != NULL) && (count < bucket_size))
1025       {
1026         if ( (NULL == bloom) ||
1027              (GNUNET_NO ==
1028               GNUNET_CONTAINER_bloomfilter_test (bloom,
1029                                                  &pos->phash)))
1030         {
1031           dist = get_distance (key,
1032                                &pos->phash);
1033           if (dist < smallest_distance)
1034           {
1035             chosen = pos;
1036             smallest_distance = dist;
1037           }
1038         }
1039         else
1040         {
1041           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1042                       "Excluded peer `%s' due to BF match in greedy routing for %s\n",
1043                       GNUNET_i2s (pos->id),
1044                       GNUNET_h2s (key));
1045           GNUNET_STATISTICS_update (GDS_stats,
1046                                     gettext_noop ("# Peers excluded from routing due to Bloomfilter"),
1047                                     1,
1048                                     GNUNET_NO);
1049           dist = get_distance (key,
1050                                &pos->phash);
1051           if (dist < smallest_distance)
1052           {
1053             chosen = NULL;
1054             smallest_distance = dist;
1055           }
1056         }
1057         count++;
1058         pos = pos->next;
1059       }
1060     }
1061     if (NULL == chosen)
1062       GNUNET_STATISTICS_update (GDS_stats,
1063                                 gettext_noop ("# Peer selection failed"),
1064                                 1,
1065                                 GNUNET_NO);
1066     else
1067       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1068                   "Selected peer `%s' in greedy routing for %s\n",
1069                   GNUNET_i2s (chosen->id),
1070                   GNUNET_h2s (key));
1071     return chosen;
1072   }
1073
1074   /* select "random" peer */
1075   /* count number of peers that are available and not filtered */
1076   count = 0;
1077   for (bc = 0; bc <= closest_bucket; bc++)
1078   {
1079     pos = k_buckets[bc].head;
1080     while ( (NULL != pos) && (count < bucket_size) )
1081     {
1082       if ( (NULL != bloom) &&
1083            (GNUNET_YES ==
1084             GNUNET_CONTAINER_bloomfilter_test (bloom,
1085                                                &pos->phash)) )
1086       {
1087         GNUNET_STATISTICS_update (GDS_stats,
1088                                   gettext_noop
1089                                   ("# Peers excluded from routing due to Bloomfilter"),
1090                                   1, GNUNET_NO);
1091         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1092                     "Excluded peer `%s' due to BF match in random routing for %s\n",
1093                     GNUNET_i2s (pos->id),
1094                     GNUNET_h2s (key));
1095         pos = pos->next;
1096         continue;               /* Ignore bloomfiltered peers */
1097       }
1098       count++;
1099       pos = pos->next;
1100     }
1101   }
1102   if (0 == count)               /* No peers to select from! */
1103   {
1104     GNUNET_STATISTICS_update (GDS_stats,
1105                               gettext_noop ("# Peer selection failed"), 1,
1106                               GNUNET_NO);
1107     return NULL;
1108   }
1109   /* Now actually choose a peer */
1110   selected = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1111                                        count);
1112   count = 0;
1113   for (bc = 0; bc <= closest_bucket; bc++)
1114   {
1115     for (pos = k_buckets[bc].head; ((pos != NULL) && (count < bucket_size)); pos = pos->next)
1116     {
1117       if ((bloom != NULL) &&
1118           (GNUNET_YES ==
1119            GNUNET_CONTAINER_bloomfilter_test (bloom,
1120                                               &pos->phash)))
1121       {
1122         continue;               /* Ignore bloomfiltered peers */
1123       }
1124       if (0 == selected--)
1125       {
1126         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1127                     "Selected peer `%s' in random routing for %s\n",
1128                     GNUNET_i2s (pos->id),
1129                     GNUNET_h2s (key));
1130         return pos;
1131       }
1132     }
1133   }
1134   GNUNET_break (0);
1135   return NULL;
1136 }
1137
1138
1139 /**
1140  * Compute the set of peers that the given request should be
1141  * forwarded to.
1142  *
1143  * @param key routing key
1144  * @param bloom bloom filter excluding peers as targets, all selected
1145  *        peers will be added to the bloom filter
1146  * @param hop_count number of hops the request has traversed so far
1147  * @param target_replication desired number of replicas
1148  * @param targets where to store an array of target peers (to be
1149  *         free'd by the caller)
1150  * @return number of peers returned in 'targets'.
1151  */
1152 static unsigned int
1153 get_target_peers (const struct GNUNET_HashCode *key,
1154                   struct GNUNET_CONTAINER_BloomFilter *bloom,
1155                   uint32_t hop_count,
1156                   uint32_t target_replication,
1157                   struct PeerInfo ***targets)
1158 {
1159   unsigned int ret;
1160   unsigned int off;
1161   struct PeerInfo **rtargets;
1162   struct PeerInfo *nxt;
1163
1164   GNUNET_assert (NULL != bloom);
1165   ret = get_forward_count (hop_count,
1166                            target_replication);
1167   if (0 == ret)
1168   {
1169     *targets = NULL;
1170     return 0;
1171   }
1172   rtargets = GNUNET_new_array (ret,
1173                                struct PeerInfo *);
1174   for (off = 0; off < ret; off++)
1175   {
1176     nxt = select_peer (key,
1177                        bloom,
1178                        hop_count);
1179     if (NULL == nxt)
1180       break;
1181     rtargets[off] = nxt;
1182     GNUNET_break (GNUNET_NO ==
1183                   GNUNET_CONTAINER_bloomfilter_test (bloom,
1184                                                      &nxt->phash));
1185     GNUNET_CONTAINER_bloomfilter_add (bloom,
1186                                       &nxt->phash);
1187   }
1188   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1189               "Selected %u/%u peers at hop %u for %s (target was %u)\n",
1190               off,
1191               GNUNET_CONTAINER_multipeermap_size (all_connected_peers),
1192               (unsigned int) hop_count,
1193               GNUNET_h2s (key),
1194               ret);
1195   if (0 == off)
1196   {
1197     GNUNET_free (rtargets);
1198     *targets = NULL;
1199     return 0;
1200   }
1201   *targets = rtargets;
1202   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1203               "Forwarding query `%s' to %u peers (goal was %u peers)\n",
1204               GNUNET_h2s (key),
1205               off,
1206               ret);
1207   return off;
1208 }
1209
1210
1211 /**
1212  * Perform a PUT operation.   Forwards the given request to other
1213  * peers.   Does not store the data locally.  Does not give the
1214  * data to local clients.  May do nothing if this is the only
1215  * peer in the network (or if we are the closest peer in the
1216  * network).
1217  *
1218  * @param type type of the block
1219  * @param options routing options
1220  * @param desired_replication_level desired replication count
1221  * @param expiration_time when does the content expire
1222  * @param hop_count how many hops has this message traversed so far
1223  * @param bf Bloom filter of peers this PUT has already traversed
1224  * @param key key for the content
1225  * @param put_path_length number of entries in @a put_path
1226  * @param put_path peers this request has traversed so far (if tracked)
1227  * @param data payload to store
1228  * @param data_size number of bytes in @a data
1229  * @return #GNUNET_OK if the request was forwarded, #GNUNET_NO if not
1230  */
1231 int
1232 GDS_NEIGHBOURS_handle_put (enum GNUNET_BLOCK_Type type,
1233                            enum GNUNET_DHT_RouteOption options,
1234                            uint32_t desired_replication_level,
1235                            struct GNUNET_TIME_Absolute expiration_time,
1236                            uint32_t hop_count,
1237                            struct GNUNET_CONTAINER_BloomFilter *bf,
1238                            const struct GNUNET_HashCode *key,
1239                            unsigned int put_path_length,
1240                            struct GNUNET_PeerIdentity *put_path,
1241                            const void *data,
1242                            size_t data_size)
1243 {
1244   unsigned int target_count;
1245   unsigned int i;
1246   struct PeerInfo **targets;
1247   struct PeerInfo *target;
1248   size_t msize;
1249   struct GNUNET_MQ_Envelope *env;
1250   struct PeerPutMessage *ppm;
1251   struct GNUNET_PeerIdentity *pp;
1252   unsigned int skip_count;
1253
1254   GNUNET_assert (NULL != bf);
1255   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1256               "Adding myself (%s) to PUT bloomfilter for %s\n",
1257               GNUNET_i2s (&my_identity),
1258               GNUNET_h2s (key));
1259   GNUNET_CONTAINER_bloomfilter_add (bf,
1260                                     &my_identity_hash);
1261   GNUNET_STATISTICS_update (GDS_stats,
1262                             gettext_noop ("# PUT requests routed"),
1263                             1,
1264                             GNUNET_NO);
1265   target_count
1266     = get_target_peers (key,
1267                         bf,
1268                         hop_count,
1269                         desired_replication_level,
1270                         &targets);
1271   if (0 == target_count)
1272   {
1273     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1274                 "Routing PUT for %s terminates after %u hops at %s\n",
1275                 GNUNET_h2s (key),
1276                 (unsigned int) hop_count,
1277                 GNUNET_i2s (&my_identity));
1278     return GNUNET_NO;
1279   }
1280   msize = put_path_length * sizeof (struct GNUNET_PeerIdentity) + data_size;
1281   if (msize + sizeof (struct PeerPutMessage)
1282       >= GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE)
1283   {
1284     put_path_length = 0;
1285     msize = data_size;
1286   }
1287   if (msize + sizeof (struct PeerPutMessage)
1288       >= GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE)
1289   {
1290     GNUNET_break (0);
1291     GNUNET_free (targets);
1292     return GNUNET_NO;
1293   }
1294   GNUNET_STATISTICS_update (GDS_stats,
1295                             gettext_noop ("# PUT messages queued for transmission"),
1296                             target_count,
1297                             GNUNET_NO);
1298   skip_count = 0;
1299   for (i = 0; i < target_count; i++)
1300   {
1301     target = targets[i];
1302     if (GNUNET_MQ_get_length (target->mq) >= MAXIMUM_PENDING_PER_PEER)
1303     {
1304       /* skip */
1305       GNUNET_STATISTICS_update (GDS_stats,
1306                                 gettext_noop ("# P2P messages dropped due to full queue"),
1307                                 1,
1308                                 GNUNET_NO);
1309       skip_count++;
1310       continue;
1311     }
1312     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1313                 "Routing PUT for %s after %u hops to %s\n",
1314                 GNUNET_h2s (key),
1315                 (unsigned int) hop_count,
1316                 GNUNET_i2s (target->id));
1317     env = GNUNET_MQ_msg_extra (ppm,
1318                                msize,
1319                                GNUNET_MESSAGE_TYPE_DHT_P2P_PUT);
1320     ppm->options = htonl (options);
1321     ppm->type = htonl (type);
1322     ppm->hop_count = htonl (hop_count + 1);
1323     ppm->desired_replication_level = htonl (desired_replication_level);
1324     ppm->put_path_length = htonl (put_path_length);
1325     ppm->expiration_time = GNUNET_TIME_absolute_hton (expiration_time);
1326     GNUNET_break (GNUNET_YES ==
1327                   GNUNET_CONTAINER_bloomfilter_test (bf,
1328                                                      &target->phash));
1329     GNUNET_assert (GNUNET_OK ==
1330                    GNUNET_CONTAINER_bloomfilter_get_raw_data (bf,
1331                                                               ppm->bloomfilter,
1332                                                               DHT_BLOOM_SIZE));
1333     ppm->key = *key;
1334     pp = (struct GNUNET_PeerIdentity *) &ppm[1];
1335     GNUNET_memcpy (pp,
1336                    put_path,
1337                    sizeof (struct GNUNET_PeerIdentity) * put_path_length);
1338     GNUNET_memcpy (&pp[put_path_length],
1339                    data,
1340                    data_size);
1341     GNUNET_MQ_send (target->mq,
1342                     env);
1343   }
1344   GNUNET_free (targets);
1345   return (skip_count < target_count) ? GNUNET_OK : GNUNET_NO;
1346 }
1347
1348
1349 /**
1350  * Perform a GET operation.  Forwards the given request to other
1351  * peers.  Does not lookup the key locally.  May do nothing if this is
1352  * the only peer in the network (or if we are the closest peer in the
1353  * network).
1354  *
1355  * @param type type of the block
1356  * @param options routing options
1357  * @param desired_replication_level desired replication count
1358  * @param hop_count how many hops did this request traverse so far?
1359  * @param key key for the content
1360  * @param xquery extended query
1361  * @param xquery_size number of bytes in @a xquery
1362  * @param bg group to use for filtering replies
1363  * @param peer_bf filter for peers not to select (again)
1364  * @return #GNUNET_OK if the request was forwarded, #GNUNET_NO if not
1365  */
1366 int
1367 GDS_NEIGHBOURS_handle_get (enum GNUNET_BLOCK_Type type,
1368                            enum GNUNET_DHT_RouteOption options,
1369                            uint32_t desired_replication_level,
1370                            uint32_t hop_count,
1371                            const struct GNUNET_HashCode *key,
1372                            const void *xquery,
1373                            size_t xquery_size,
1374                            struct GNUNET_BLOCK_Group *bg,
1375                            struct GNUNET_CONTAINER_BloomFilter *peer_bf)
1376 {
1377   unsigned int target_count;
1378   struct PeerInfo **targets;
1379   struct PeerInfo *target;
1380   struct GNUNET_MQ_Envelope *env;
1381   size_t msize;
1382   struct PeerGetMessage *pgm;
1383   char *xq;
1384   size_t reply_bf_size;
1385   void *reply_bf;
1386   unsigned int skip_count;
1387   uint32_t bf_nonce;
1388
1389   GNUNET_assert (NULL != peer_bf);
1390   GNUNET_STATISTICS_update (GDS_stats,
1391                             gettext_noop ("# GET requests routed"),
1392                             1,
1393                             GNUNET_NO);
1394   target_count = get_target_peers (key,
1395                                    peer_bf,
1396                                    hop_count,
1397                                    desired_replication_level,
1398                                    &targets);
1399   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1400               "Adding myself (%s) to GET bloomfilter for %s\n",
1401               GNUNET_i2s (&my_identity),
1402               GNUNET_h2s (key));
1403   GNUNET_CONTAINER_bloomfilter_add (peer_bf,
1404                                     &my_identity_hash);
1405   if (0 == target_count)
1406   {
1407     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1408                 "Routing GET for %s terminates after %u hops at %s\n",
1409                 GNUNET_h2s (key),
1410                 (unsigned int) hop_count,
1411                 GNUNET_i2s (&my_identity));
1412     return GNUNET_NO;
1413   }
1414   if (GNUNET_OK !=
1415       GNUNET_BLOCK_group_serialize (bg,
1416                                     &bf_nonce,
1417                                     &reply_bf,
1418                                     &reply_bf_size))
1419   {
1420     reply_bf = NULL;
1421     reply_bf_size = 0;
1422     bf_nonce = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1423                                          UINT32_MAX);
1424   }
1425   msize = xquery_size + reply_bf_size;
1426   if (msize + sizeof (struct PeerGetMessage) >= GNUNET_MAX_MESSAGE_SIZE)
1427   {
1428     GNUNET_break (0);
1429     GNUNET_free_non_null (reply_bf);
1430     GNUNET_free (targets);
1431     return GNUNET_NO;
1432   }
1433   GNUNET_STATISTICS_update (GDS_stats,
1434                             gettext_noop ("# GET messages queued for transmission"),
1435                             target_count,
1436                             GNUNET_NO);
1437   /* forward request */
1438   skip_count = 0;
1439   for (unsigned int i = 0; i < target_count; i++)
1440   {
1441     target = targets[i];
1442     if (GNUNET_MQ_get_length (target->mq) >= MAXIMUM_PENDING_PER_PEER)
1443     {
1444       /* skip */
1445       GNUNET_STATISTICS_update (GDS_stats,
1446                                 gettext_noop ("# P2P messages dropped due to full queue"),
1447                                 1, GNUNET_NO);
1448       skip_count++;
1449       continue;
1450     }
1451     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1452                 "Routing GET for %s after %u hops to %s\n",
1453                 GNUNET_h2s (key),
1454                 (unsigned int) hop_count,
1455                 GNUNET_i2s (target->id));
1456     env = GNUNET_MQ_msg_extra (pgm,
1457                                msize,
1458                                GNUNET_MESSAGE_TYPE_DHT_P2P_GET);
1459     pgm->options = htonl (options);
1460     pgm->type = htonl (type);
1461     pgm->hop_count = htonl (hop_count + 1);
1462     pgm->desired_replication_level = htonl (desired_replication_level);
1463     pgm->xquery_size = htonl (xquery_size);
1464     pgm->bf_mutator = bf_nonce;
1465     GNUNET_break (GNUNET_YES ==
1466                   GNUNET_CONTAINER_bloomfilter_test (peer_bf,
1467                                                      &target->phash));
1468     GNUNET_assert (GNUNET_OK ==
1469                    GNUNET_CONTAINER_bloomfilter_get_raw_data (peer_bf,
1470                                                               pgm->bloomfilter,
1471                                                               DHT_BLOOM_SIZE));
1472     pgm->key = *key;
1473     xq = (char *) &pgm[1];
1474     GNUNET_memcpy (xq,
1475                    xquery,
1476                    xquery_size);
1477     GNUNET_memcpy (&xq[xquery_size],
1478                    reply_bf,
1479                    reply_bf_size);
1480     GNUNET_MQ_send (target->mq,
1481                     env);
1482   }
1483   GNUNET_free (targets);
1484   GNUNET_free_non_null (reply_bf);
1485   return (skip_count < target_count) ? GNUNET_OK : GNUNET_NO;
1486 }
1487
1488
1489 /**
1490  * Handle a reply (route to origin).  Only forwards the reply back to
1491  * the given peer.  Does not do local caching or forwarding to local
1492  * clients.
1493  *
1494  * @param target neighbour that should receive the block (if still connected)
1495  * @param type type of the block
1496  * @param expiration_time when does the content expire
1497  * @param key key for the content
1498  * @param put_path_length number of entries in @a put_path
1499  * @param put_path peers the original PUT traversed (if tracked)
1500  * @param get_path_length number of entries in @a get_path
1501  * @param get_path peers this reply has traversed so far (if tracked)
1502  * @param data payload of the reply
1503  * @param data_size number of bytes in @a data
1504  */
1505 void
1506 GDS_NEIGHBOURS_handle_reply (const struct GNUNET_PeerIdentity *target,
1507                              enum GNUNET_BLOCK_Type type,
1508                              struct GNUNET_TIME_Absolute expiration_time,
1509                              const struct GNUNET_HashCode *key,
1510                              unsigned int put_path_length,
1511                              const struct GNUNET_PeerIdentity *put_path,
1512                              unsigned int get_path_length,
1513                              const struct GNUNET_PeerIdentity *get_path,
1514                              const void *data,
1515                              size_t data_size)
1516 {
1517   struct PeerInfo *pi;
1518   struct GNUNET_MQ_Envelope *env;
1519   size_t msize;
1520   struct PeerResultMessage *prm;
1521   struct GNUNET_PeerIdentity *paths;
1522
1523   msize = data_size + (get_path_length + put_path_length) *
1524       sizeof (struct GNUNET_PeerIdentity);
1525   if ((msize + sizeof (struct PeerResultMessage) >= GNUNET_MAX_MESSAGE_SIZE) ||
1526       (get_path_length >
1527        GNUNET_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)) ||
1528       (put_path_length >
1529        GNUNET_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)) ||
1530       (data_size > GNUNET_MAX_MESSAGE_SIZE))
1531   {
1532     GNUNET_break (0);
1533     return;
1534   }
1535   pi = GNUNET_CONTAINER_multipeermap_get (all_connected_peers,
1536                                           target);
1537   if (NULL == pi)
1538   {
1539     /* peer disconnected in the meantime, drop reply */
1540     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1541                 "No matching peer for reply for key %s\n",
1542                 GNUNET_h2s (key));
1543     return;
1544   }
1545   if (GNUNET_MQ_get_length (pi->mq) >= MAXIMUM_PENDING_PER_PEER)
1546   {
1547     /* skip */
1548     GNUNET_STATISTICS_update (GDS_stats,
1549                               gettext_noop ("# P2P messages dropped due to full queue"),
1550                               1,
1551                               GNUNET_NO);
1552     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1553                 "Peer queue full, ignoring reply for key %s\n",
1554                 GNUNET_h2s (key));
1555     return;
1556   }
1557
1558   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1559               "Forwarding reply for key %s to peer %s\n",
1560               GNUNET_h2s (key),
1561               GNUNET_i2s (target));
1562   GNUNET_STATISTICS_update (GDS_stats,
1563                             gettext_noop
1564                             ("# RESULT messages queued for transmission"), 1,
1565                             GNUNET_NO);
1566   env = GNUNET_MQ_msg_extra (prm,
1567                              msize,
1568                              GNUNET_MESSAGE_TYPE_DHT_P2P_RESULT);
1569   prm->type = htonl (type);
1570   prm->put_path_length = htonl (put_path_length);
1571   prm->get_path_length = htonl (get_path_length);
1572   prm->expiration_time = GNUNET_TIME_absolute_hton (expiration_time);
1573   prm->key = *key;
1574   paths = (struct GNUNET_PeerIdentity *) &prm[1];
1575   GNUNET_memcpy (paths,
1576                  put_path,
1577                  put_path_length * sizeof (struct GNUNET_PeerIdentity));
1578   GNUNET_memcpy (&paths[put_path_length],
1579                  get_path,
1580                  get_path_length * sizeof (struct GNUNET_PeerIdentity));
1581   GNUNET_memcpy (&paths[put_path_length + get_path_length],
1582                  data,
1583                  data_size);
1584   GNUNET_MQ_send (pi->mq,
1585                   env);
1586 }
1587
1588
1589 /**
1590  * To be called on core init/fail.
1591  *
1592  * @param cls service closure
1593  * @param identity the public identity of this peer
1594  */
1595 static void
1596 core_init (void *cls,
1597            const struct GNUNET_PeerIdentity *identity)
1598 {
1599   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1600               "CORE called, I am %s\n",
1601               GNUNET_i2s (identity));
1602   my_identity = *identity;
1603   GNUNET_CRYPTO_hash (identity,
1604                       sizeof (struct GNUNET_PeerIdentity),
1605                       &my_identity_hash);
1606   GNUNET_SERVICE_resume (GDS_service);
1607 }
1608
1609
1610 /**
1611  * Check validity of a p2p put request.
1612  *
1613  * @param cls closure with the `struct PeerInfo` of the sender
1614  * @param message message
1615  * @return #GNUNET_OK if the message is valid
1616  */
1617 static int
1618 check_dht_p2p_put (void *cls,
1619                    const struct PeerPutMessage *put)
1620 {
1621   uint32_t putlen;
1622   uint16_t msize;
1623
1624   msize = ntohs (put->header.size);
1625   putlen = ntohl (put->put_path_length);
1626   if ((msize <
1627        sizeof (struct PeerPutMessage) +
1628        putlen * sizeof (struct GNUNET_PeerIdentity)) ||
1629       (putlen >
1630        GNUNET_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)))
1631   {
1632     GNUNET_break_op (0);
1633     return GNUNET_SYSERR;
1634   }
1635   return GNUNET_OK;
1636 }
1637
1638
1639 /**
1640  * Core handler for p2p put requests.
1641  *
1642  * @param cls closure with the `struct PeerInfo` of the sender
1643  * @param message message
1644  */
1645 static void
1646 handle_dht_p2p_put (void *cls,
1647                     const struct PeerPutMessage *put)
1648 {
1649   struct PeerInfo *peer = cls;
1650   const struct GNUNET_PeerIdentity *put_path;
1651   const void *payload;
1652   uint32_t putlen;
1653   uint16_t msize;
1654   size_t payload_size;
1655   enum GNUNET_DHT_RouteOption options;
1656   struct GNUNET_CONTAINER_BloomFilter *bf;
1657   struct GNUNET_HashCode test_key;
1658   int forwarded;
1659
1660   msize = ntohs (put->header.size);
1661   putlen = ntohl (put->put_path_length);
1662   GNUNET_STATISTICS_update (GDS_stats,
1663                             gettext_noop ("# P2P PUT requests received"),
1664                             1,
1665                             GNUNET_NO);
1666   GNUNET_STATISTICS_update (GDS_stats,
1667                             gettext_noop ("# P2P PUT bytes received"),
1668                             msize,
1669                             GNUNET_NO);
1670   put_path = (const struct GNUNET_PeerIdentity *) &put[1];
1671   payload = &put_path[putlen];
1672   options = ntohl (put->options);
1673   payload_size = msize - (sizeof (struct PeerPutMessage) +
1674                           putlen * sizeof (struct GNUNET_PeerIdentity));
1675
1676   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1677               "PUT for `%s' from %s\n",
1678               GNUNET_h2s (&put->key),
1679               GNUNET_i2s (peer->id));
1680   if (GNUNET_YES == log_route_details_stderr)
1681   {
1682     char *tmp;
1683     char *pp;
1684
1685     pp = GNUNET_STRINGS_pp2s (put_path,
1686                               putlen);
1687     tmp = GNUNET_strdup (GNUNET_i2s (&my_identity));
1688     LOG_TRAFFIC (GNUNET_ERROR_TYPE_DEBUG,
1689                  "R5N PUT %s: %s->%s (%u, %u=>%u, PP: %s)\n",
1690                  GNUNET_h2s (&put->key),
1691                  GNUNET_i2s (peer->id),
1692                  tmp,
1693                  ntohl(put->hop_count),
1694                  GNUNET_CRYPTO_hash_matching_bits (&peer->phash,
1695                                                    &put->key),
1696                  GNUNET_CRYPTO_hash_matching_bits (&my_identity_hash,
1697                                                    &put->key),
1698                  pp);
1699     GNUNET_free (pp);
1700     GNUNET_free (tmp);
1701   }
1702   switch (GNUNET_BLOCK_get_key
1703           (GDS_block_context,
1704            ntohl (put->type),
1705            payload,
1706            payload_size,
1707            &test_key))
1708   {
1709   case GNUNET_YES:
1710     if (0 != memcmp (&test_key,
1711                      &put->key,
1712                      sizeof (struct GNUNET_HashCode)))
1713     {
1714       char *put_s = GNUNET_strdup (GNUNET_h2s_full (&put->key));
1715
1716       GNUNET_break_op (0);
1717       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1718                   "PUT with key `%s' for block with key %s\n",
1719                   put_s,
1720                   GNUNET_h2s_full (&test_key));
1721       GNUNET_free (put_s);
1722       return;
1723     }
1724     break;
1725   case GNUNET_NO:
1726     GNUNET_break_op (0);
1727     return;
1728   case GNUNET_SYSERR:
1729     /* cannot verify, good luck */
1730     break;
1731   }
1732   if (ntohl (put->type) == GNUNET_BLOCK_TYPE_REGEX) /* FIXME: do for all tpyes */
1733   {
1734     switch (GNUNET_BLOCK_evaluate (GDS_block_context,
1735                                    ntohl (put->type),
1736                                    NULL, /* query group */
1737                                    GNUNET_BLOCK_EO_NONE,
1738                                    NULL,    /* query */
1739                                    NULL, 0, /* xquery */
1740                                    payload,
1741                                    payload_size))
1742     {
1743     case GNUNET_BLOCK_EVALUATION_OK_MORE:
1744     case GNUNET_BLOCK_EVALUATION_OK_LAST:
1745       break;
1746
1747     case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
1748     case GNUNET_BLOCK_EVALUATION_RESULT_INVALID:
1749     case GNUNET_BLOCK_EVALUATION_RESULT_IRRELEVANT:
1750     case GNUNET_BLOCK_EVALUATION_REQUEST_VALID:
1751     case GNUNET_BLOCK_EVALUATION_REQUEST_INVALID:
1752     case GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED:
1753     default:
1754       GNUNET_break_op (0);
1755       return;
1756     }
1757   }
1758
1759   bf = GNUNET_CONTAINER_bloomfilter_init (put->bloomfilter,
1760                                           DHT_BLOOM_SIZE,
1761                                           GNUNET_CONSTANTS_BLOOMFILTER_K);
1762   GNUNET_break_op (GNUNET_YES ==
1763                    GNUNET_CONTAINER_bloomfilter_test (bf,
1764                                                       &peer->phash));
1765   {
1766     struct GNUNET_PeerIdentity pp[putlen + 1];
1767
1768     /* extend 'put path' by sender */
1769     if (0 != (options & GNUNET_DHT_RO_RECORD_ROUTE))
1770     {
1771 #if SANITY_CHECKS
1772       for (unsigned int i=0;i<=putlen;i++)
1773       {
1774         for (unsigned int j=0;j<i;j++)
1775         {
1776           GNUNET_break (0 != memcmp (&pp[i],
1777                                      &pp[j],
1778                                      sizeof (struct GNUNET_PeerIdentity)));
1779         }
1780         GNUNET_break (0 != memcmp (&pp[i],
1781                                    peer->id,
1782                                    sizeof (struct GNUNET_PeerIdentity)));
1783       }
1784 #endif
1785       GNUNET_memcpy (pp,
1786                      put_path,
1787                      putlen * sizeof (struct GNUNET_PeerIdentity));
1788       pp[putlen] = *peer->id;
1789       putlen++;
1790     }
1791     else
1792       putlen = 0;
1793
1794     /* give to local clients */
1795     GDS_CLIENTS_handle_reply (GNUNET_TIME_absolute_ntoh (put->expiration_time),
1796                               &put->key,
1797                               0,
1798                               NULL,
1799                               putlen,
1800                               pp,
1801                               ntohl (put->type),
1802                               payload_size,
1803                               payload);
1804     /* store locally */
1805     if ((0 != (options & GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE)) ||
1806         (am_closest_peer (&put->key, bf)))
1807       GDS_DATACACHE_handle_put (GNUNET_TIME_absolute_ntoh (put->expiration_time),
1808                                 &put->key,
1809                                 putlen,
1810                                 pp,
1811                                 ntohl (put->type),
1812                                 payload_size,
1813                                 payload);
1814     /* route to other peers */
1815     forwarded = GDS_NEIGHBOURS_handle_put (ntohl (put->type),
1816                                            options,
1817                                            ntohl (put->desired_replication_level),
1818                                            GNUNET_TIME_absolute_ntoh (put->expiration_time),
1819                                            ntohl (put->hop_count),
1820                                            bf,
1821                                            &put->key,
1822                                            putlen,
1823                                            pp,
1824                                            payload,
1825                                            payload_size);
1826     /* notify monitoring clients */
1827     GDS_CLIENTS_process_put (options
1828                              | ( (GNUNET_OK == forwarded)
1829                                  ? GNUNET_DHT_RO_LAST_HOP
1830                                  : 0 ),
1831                              ntohl (put->type),
1832                              ntohl (put->hop_count),
1833                              ntohl (put->desired_replication_level),
1834                              putlen, pp,
1835                              GNUNET_TIME_absolute_ntoh (put->expiration_time),
1836                              &put->key,
1837                              payload,
1838                              payload_size);
1839   }
1840   GNUNET_CONTAINER_bloomfilter_free (bf);
1841 }
1842
1843
1844 /**
1845  * We have received a FIND PEER request.  Send matching
1846  * HELLOs back.
1847  *
1848  * @param sender sender of the FIND PEER request
1849  * @param key peers close to this key are desired
1850  * @param bg group for filtering peers
1851  */
1852 static void
1853 handle_find_peer (const struct GNUNET_PeerIdentity *sender,
1854                   const struct GNUNET_HashCode *key,
1855                   struct GNUNET_BLOCK_Group *bg)
1856 {
1857   int bucket_idx;
1858   struct PeerBucket *bucket;
1859   struct PeerInfo *peer;
1860   unsigned int choice;
1861   const struct GNUNET_HELLO_Message *hello;
1862   size_t hello_size;
1863
1864   /* first, check about our own HELLO */
1865   if (NULL != GDS_my_hello)
1866   {
1867     hello_size = GNUNET_HELLO_size ((const struct GNUNET_HELLO_Message *) GDS_my_hello);
1868     GNUNET_break (hello_size >= sizeof (struct GNUNET_MessageHeader));
1869     if (GNUNET_BLOCK_EVALUATION_OK_MORE ==
1870         GNUNET_BLOCK_evaluate (GDS_block_context,
1871                                GNUNET_BLOCK_TYPE_DHT_HELLO,
1872                                bg,
1873                                GNUNET_BLOCK_EO_LOCAL_SKIP_CRYPTO,
1874                                &my_identity_hash,
1875                                NULL, 0,
1876                                GDS_my_hello,
1877                                hello_size))
1878     {
1879       GDS_NEIGHBOURS_handle_reply (sender,
1880                                    GNUNET_BLOCK_TYPE_DHT_HELLO,
1881                                    GNUNET_TIME_relative_to_absolute (hello_expiration),
1882                                    key,
1883                                    0,
1884                                    NULL,
1885                                    0,
1886                                    NULL,
1887                                    GDS_my_hello,
1888                                    hello_size);
1889     }
1890     else
1891     {
1892       GNUNET_STATISTICS_update (GDS_stats,
1893                                 gettext_noop ("# FIND PEER requests ignored due to Bloomfilter"),
1894                                 1,
1895                                 GNUNET_NO);
1896     }
1897   }
1898   else
1899   {
1900     GNUNET_STATISTICS_update (GDS_stats,
1901                               gettext_noop ("# FIND PEER requests ignored due to lack of HELLO"),
1902                               1,
1903                               GNUNET_NO);
1904   }
1905
1906   /* then, also consider sending a random HELLO from the closest bucket */
1907   if (0 == memcmp (&my_identity_hash,
1908                    key,
1909                    sizeof (struct GNUNET_HashCode)))
1910     bucket_idx = closest_bucket;
1911   else
1912     bucket_idx = GNUNET_MIN (closest_bucket,
1913                              find_bucket (key));
1914   if (bucket_idx == GNUNET_SYSERR)
1915     return;
1916   bucket = &k_buckets[bucket_idx];
1917   if (bucket->peers_size == 0)
1918     return;
1919   choice = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1920                                      bucket->peers_size);
1921   peer = bucket->head;
1922   while (choice > 0)
1923   {
1924     GNUNET_assert (NULL != peer);
1925     peer = peer->next;
1926     choice--;
1927   }
1928   choice = bucket->peers_size;
1929   do
1930   {
1931     peer = peer->next;
1932     if (0 == choice--)
1933       return;                   /* no non-masked peer available */
1934     if (NULL == peer)
1935       peer = bucket->head;
1936     hello = GDS_HELLO_get (peer->id);
1937   } while ( (NULL == hello) ||
1938             (GNUNET_BLOCK_EVALUATION_OK_MORE !=
1939              GNUNET_BLOCK_evaluate (GDS_block_context,
1940                                     GNUNET_BLOCK_TYPE_DHT_HELLO,
1941                                     bg,
1942                                     GNUNET_BLOCK_EO_LOCAL_SKIP_CRYPTO,
1943                                     &peer->phash,
1944                                     NULL, 0,
1945                                     hello,
1946                                     (hello_size = GNUNET_HELLO_size (hello)))) );
1947   GDS_NEIGHBOURS_handle_reply (sender,
1948                                GNUNET_BLOCK_TYPE_DHT_HELLO,
1949                                GNUNET_TIME_relative_to_absolute
1950                                (GNUNET_CONSTANTS_HELLO_ADDRESS_EXPIRATION),
1951                                key,
1952                                0,
1953                                NULL,
1954                                0,
1955                                NULL,
1956                                hello,
1957                                hello_size);
1958 }
1959
1960
1961 /**
1962  * Handle a result from local datacache for a GET operation.
1963  *
1964  * @param cls the `struct PeerInfo` for which this is a reply
1965  * @param type type of the block
1966  * @param expiration_time when does the content expire
1967  * @param key key for the content
1968  * @param put_path_length number of entries in @a put_path
1969  * @param put_path peers the original PUT traversed (if tracked)
1970  * @param get_path_length number of entries in @a get_path
1971  * @param get_path peers this reply has traversed so far (if tracked)
1972  * @param data payload of the reply
1973  * @param data_size number of bytes in @a data
1974  */
1975 static void
1976 handle_local_result (void *cls,
1977                      enum GNUNET_BLOCK_Type type,
1978                      struct GNUNET_TIME_Absolute expiration_time,
1979                      const struct GNUNET_HashCode *key,
1980                      unsigned int put_path_length,
1981                      const struct GNUNET_PeerIdentity *put_path,
1982                      unsigned int get_path_length,
1983                      const struct GNUNET_PeerIdentity *get_path,
1984                      const void *data,
1985                      size_t data_size)
1986 {
1987   struct PeerInfo *peer = cls;
1988   char *pp;
1989
1990   pp = GNUNET_STRINGS_pp2s (put_path,
1991                             put_path_length);
1992   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1993               "Found local result for %s (PP: %s)\n",
1994               GNUNET_h2s (key),
1995               pp);
1996   GNUNET_free (pp);
1997   GDS_NEIGHBOURS_handle_reply (peer->id,
1998                                type,
1999                                expiration_time,
2000                                key,
2001                                put_path_length, put_path,
2002                                get_path_length, get_path,
2003                                data, data_size);
2004 }
2005
2006
2007 /**
2008  * Check validity of p2p get request.
2009  *
2010  * @param cls closure with the `struct PeerInfo` of the sender
2011  * @param get the message
2012  * @return #GNUNET_OK if the message is well-formed
2013  */
2014 static int
2015 check_dht_p2p_get (void *cls,
2016                    const struct PeerGetMessage *get)
2017 {
2018   uint32_t xquery_size;
2019   uint16_t msize;
2020
2021   msize = ntohs (get->header.size);
2022   xquery_size = ntohl (get->xquery_size);
2023   if (msize < sizeof (struct PeerGetMessage) + xquery_size)
2024   {
2025     GNUNET_break_op (0);
2026     return GNUNET_SYSERR;
2027   }
2028   return GNUNET_OK;
2029 }
2030
2031
2032 /**
2033  * Core handler for p2p get requests.
2034  *
2035  * @param cls closure with the `struct PeerInfo` of the sender
2036  * @param get the message
2037  */
2038 static void
2039 handle_dht_p2p_get (void *cls,
2040                     const struct PeerGetMessage *get)
2041 {
2042   struct PeerInfo *peer = cls;
2043   uint32_t xquery_size;
2044   size_t reply_bf_size;
2045   uint16_t msize;
2046   enum GNUNET_BLOCK_Type type;
2047   enum GNUNET_DHT_RouteOption options;
2048   enum GNUNET_BLOCK_EvaluationResult eval;
2049   struct GNUNET_BLOCK_Group *bg;
2050   struct GNUNET_CONTAINER_BloomFilter *peer_bf;
2051   const char *xquery;
2052   int forwarded;
2053
2054   /* parse and validate message */
2055   msize = ntohs (get->header.size);
2056   xquery_size = ntohl (get->xquery_size);
2057   reply_bf_size = msize - (sizeof (struct PeerGetMessage) + xquery_size);
2058   type = ntohl (get->type);
2059   options = ntohl (get->options);
2060   xquery = (const char *) &get[1];
2061   GNUNET_STATISTICS_update (GDS_stats,
2062                             gettext_noop ("# P2P GET requests received"),
2063                             1,
2064                             GNUNET_NO);
2065   GNUNET_STATISTICS_update (GDS_stats,
2066                             gettext_noop ("# P2P GET bytes received"),
2067                             msize,
2068                             GNUNET_NO);
2069   if (GNUNET_YES == log_route_details_stderr)
2070   {
2071     char *tmp;
2072
2073     tmp = GNUNET_strdup (GNUNET_i2s (&my_identity));
2074     LOG_TRAFFIC (GNUNET_ERROR_TYPE_DEBUG,
2075                  "R5N GET %s: %s->%s (%u, %u=>%u) xq: %.*s\n",
2076                  GNUNET_h2s (&get->key),
2077                  GNUNET_i2s (peer->id),
2078                  tmp,
2079                  ntohl(get->hop_count),
2080                  GNUNET_CRYPTO_hash_matching_bits (&peer->phash,
2081                                                    &get->key),
2082                  GNUNET_CRYPTO_hash_matching_bits (&my_identity_hash,
2083                                                    &get->key),
2084                  ntohl(get->xquery_size),
2085                  xquery);
2086     GNUNET_free (tmp);
2087   }
2088   eval
2089     = GNUNET_BLOCK_evaluate (GDS_block_context,
2090                              type,
2091                              NULL,
2092                              GNUNET_BLOCK_EO_NONE,
2093                              &get->key,
2094                              xquery,
2095                              xquery_size,
2096                              NULL,
2097                              0);
2098   if (eval != GNUNET_BLOCK_EVALUATION_REQUEST_VALID)
2099   {
2100     /* request invalid or block type not supported */
2101     GNUNET_break_op (eval == GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED);
2102     return;
2103   }
2104   peer_bf = GNUNET_CONTAINER_bloomfilter_init (get->bloomfilter,
2105                                                DHT_BLOOM_SIZE,
2106                                                GNUNET_CONSTANTS_BLOOMFILTER_K);
2107   GNUNET_break_op (GNUNET_YES ==
2108                    GNUNET_CONTAINER_bloomfilter_test (peer_bf,
2109                                                       &peer->phash));
2110   bg = GNUNET_BLOCK_group_create (GDS_block_context,
2111                                   type,
2112                                   get->bf_mutator,
2113                                   &xquery[xquery_size],
2114                                   reply_bf_size,
2115                                   "filter-size",
2116                                   reply_bf_size,
2117                                   NULL);
2118   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2119               "GET for %s at %s after %u hops\n",
2120               GNUNET_h2s (&get->key),
2121               GNUNET_i2s (&my_identity),
2122               (unsigned int) ntohl (get->hop_count));
2123   /* local lookup (this may update the reply_bf) */
2124   if ((0 != (options & GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE)) ||
2125       (am_closest_peer (&get->key,
2126                         peer_bf)))
2127   {
2128     if ((0 != (options & GNUNET_DHT_RO_FIND_PEER)))
2129     {
2130       GNUNET_STATISTICS_update (GDS_stats,
2131                                 gettext_noop ("# P2P FIND PEER requests processed"),
2132                                 1,
2133                                 GNUNET_NO);
2134       handle_find_peer (peer->id,
2135                         &get->key,
2136                         bg);
2137     }
2138     else
2139     {
2140       eval = GDS_DATACACHE_handle_get (&get->key,
2141                                        type,
2142                                        xquery,
2143                                        xquery_size,
2144                                        bg,
2145                                        &handle_local_result,
2146                                        peer);
2147     }
2148   }
2149   else
2150   {
2151     GNUNET_STATISTICS_update (GDS_stats,
2152                               gettext_noop ("# P2P GET requests ONLY routed"),
2153                               1,
2154                               GNUNET_NO);
2155   }
2156
2157   /* remember request for routing replies */
2158   GDS_ROUTING_add (peer->id,
2159                         type,
2160                         bg, /* bg now owned by routing, but valid at least until end of this function! */
2161                         options,
2162                         &get->key,
2163                         xquery,
2164                         xquery_size);
2165
2166   /* P2P forwarding */
2167   forwarded = GNUNET_NO;
2168   if (eval != GNUNET_BLOCK_EVALUATION_OK_LAST)
2169     forwarded = GDS_NEIGHBOURS_handle_get (type,
2170                                            options,
2171                                            ntohl (get->desired_replication_level),
2172                                            ntohl (get->hop_count),
2173                                            &get->key,
2174                                            xquery,
2175                                            xquery_size,
2176                                            bg,
2177                                            peer_bf);
2178   GDS_CLIENTS_process_get (options
2179                            | (GNUNET_OK == forwarded)
2180                            ? GNUNET_DHT_RO_LAST_HOP : 0,
2181                            type,
2182                            ntohl (get->hop_count),
2183                            ntohl (get->desired_replication_level),
2184                            0,
2185                            NULL,
2186                            &get->key);
2187
2188   /* clean up; note that 'bg' is owned by routing now! */
2189   GNUNET_CONTAINER_bloomfilter_free (peer_bf);
2190 }
2191
2192
2193 /**
2194  * Check validity of p2p result message.
2195  *
2196  * @param cls closure
2197  * @param message message
2198  * @return #GNUNET_YES if the message is well-formed
2199  */
2200 static int
2201 check_dht_p2p_result (void *cls,
2202                       const struct PeerResultMessage *prm)
2203 {
2204   uint32_t get_path_length;
2205   uint32_t put_path_length;
2206   uint16_t msize;
2207
2208   msize = ntohs (prm->header.size);
2209   put_path_length = ntohl (prm->put_path_length);
2210   get_path_length = ntohl (prm->get_path_length);
2211   if ((msize <
2212        sizeof (struct PeerResultMessage) + (get_path_length +
2213                                             put_path_length) *
2214        sizeof (struct GNUNET_PeerIdentity)) ||
2215       (get_path_length >
2216        GNUNET_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)) ||
2217       (put_path_length >
2218        GNUNET_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)))
2219   {
2220     GNUNET_break_op (0);
2221     return GNUNET_SYSERR;
2222   }
2223   return GNUNET_OK;
2224 }
2225
2226
2227 /**
2228  * Process a reply, after the @a get_path has been updated.
2229  *
2230  * @param expiration_time when does the reply expire
2231  * @param key key matching the query
2232  * @param get_path_length number of entries in @a get_path
2233  * @param get_path path the reply has taken
2234  * @param put_path_length number of entries in @a put_path
2235  * @param put_path path the PUT has taken
2236  * @param type type of the block
2237  * @param data_size number of bytes in @a data
2238  * @param data payload of the reply
2239  */
2240 static void
2241 process_reply_with_path (struct GNUNET_TIME_Absolute expiration_time,
2242                          const struct GNUNET_HashCode *key,
2243                          unsigned int get_path_length,
2244                          const struct GNUNET_PeerIdentity *get_path,
2245                          unsigned int put_path_length,
2246                          const struct GNUNET_PeerIdentity *put_path,
2247                          enum GNUNET_BLOCK_Type type,
2248                          size_t data_size,
2249                          const void *data)
2250 {
2251   /* forward to local clients */
2252   GDS_CLIENTS_handle_reply (expiration_time,
2253                             key,
2254                             get_path_length,
2255                             get_path,
2256                             put_path_length,
2257                             put_path,
2258                             type,
2259                             data_size,
2260                             data);
2261   GDS_CLIENTS_process_get_resp (type,
2262                                 get_path,
2263                                 get_path_length,
2264                                 put_path,
2265                                 put_path_length,
2266                                 expiration_time,
2267                                 key,
2268                                 data,
2269                                 data_size);
2270   if (GNUNET_YES == cache_results)
2271   {
2272     struct GNUNET_PeerIdentity xput_path[get_path_length + 1 + put_path_length];
2273
2274     GNUNET_memcpy (xput_path,
2275                    put_path,
2276                    put_path_length * sizeof (struct GNUNET_PeerIdentity));
2277     GNUNET_memcpy (&xput_path[put_path_length],
2278                    get_path,
2279                    get_path_length * sizeof (struct GNUNET_PeerIdentity));
2280
2281     GDS_DATACACHE_handle_put (expiration_time,
2282                               key,
2283                               get_path_length + put_path_length,
2284                               xput_path,
2285                               type,
2286                               data_size,
2287                               data);
2288   }
2289   /* forward to other peers */
2290   GDS_ROUTING_process (type,
2291                        expiration_time,
2292                        key,
2293                        put_path_length,
2294                        put_path,
2295                        get_path_length,
2296                        get_path,
2297                        data,
2298                        data_size);
2299 }
2300
2301
2302 /**
2303  * Core handler for p2p result messages.
2304  *
2305  * @param cls closure
2306  * @param message message
2307  */
2308 static void
2309 handle_dht_p2p_result (void *cls,
2310                        const struct PeerResultMessage *prm)
2311 {
2312   struct PeerInfo *peer = cls;
2313   const struct GNUNET_PeerIdentity *put_path;
2314   const struct GNUNET_PeerIdentity *get_path;
2315   const void *data;
2316   uint32_t get_path_length;
2317   uint32_t put_path_length;
2318   uint16_t msize;
2319   size_t data_size;
2320   enum GNUNET_BLOCK_Type type;
2321
2322   /* parse and validate message */
2323   msize = ntohs (prm->header.size);
2324   put_path_length = ntohl (prm->put_path_length);
2325   get_path_length = ntohl (prm->get_path_length);
2326   put_path = (const struct GNUNET_PeerIdentity *) &prm[1];
2327   get_path = &put_path[put_path_length];
2328   type = ntohl (prm->type);
2329   data = (const void *) &get_path[get_path_length];
2330   data_size = msize - (sizeof (struct PeerResultMessage) +
2331                        (get_path_length +
2332                         put_path_length) * sizeof (struct GNUNET_PeerIdentity));
2333   GNUNET_STATISTICS_update (GDS_stats,
2334                             gettext_noop ("# P2P RESULTS received"),
2335                             1,
2336                             GNUNET_NO);
2337   GNUNET_STATISTICS_update (GDS_stats,
2338                             gettext_noop ("# P2P RESULT bytes received"),
2339                             msize,
2340                             GNUNET_NO);
2341   if (GNUNET_YES == log_route_details_stderr)
2342   {
2343     char *tmp;
2344     char *pp;
2345     char *gp;
2346
2347     gp = GNUNET_STRINGS_pp2s (get_path,
2348                               get_path_length);
2349     pp = GNUNET_STRINGS_pp2s (put_path,
2350                               put_path_length);
2351     tmp = GNUNET_strdup (GNUNET_i2s (&my_identity));
2352     LOG_TRAFFIC (GNUNET_ERROR_TYPE_DEBUG,
2353                  "R5N RESULT %s: %s->%s (GP: %s, PP: %s)\n",
2354                  GNUNET_h2s (&prm->key),
2355                  GNUNET_i2s (peer->id),
2356                  tmp,
2357                  gp,
2358                  pp);
2359     GNUNET_free (gp);
2360     GNUNET_free (pp);
2361     GNUNET_free (tmp);
2362   }
2363   /* if we got a HELLO, consider it for our own routing table */
2364   if (GNUNET_BLOCK_TYPE_DHT_HELLO == type)
2365   {
2366     const struct GNUNET_MessageHeader *h;
2367     struct GNUNET_PeerIdentity pid;
2368
2369     /* Should be a HELLO, validate and consider using it! */
2370     if (data_size < sizeof (struct GNUNET_HELLO_Message))
2371     {
2372       GNUNET_break_op (0);
2373       return;
2374     }
2375     h = data;
2376     if (data_size != ntohs (h->size))
2377     {
2378       GNUNET_break_op (0);
2379       return;
2380     }
2381     if (GNUNET_OK !=
2382         GNUNET_HELLO_get_id ((const struct GNUNET_HELLO_Message *) h,
2383                              &pid))
2384     {
2385       GNUNET_break_op (0);
2386       return;
2387     }
2388     if ( (GNUNET_YES != disable_try_connect) &&
2389          (0 != memcmp (&my_identity,
2390                        &pid,
2391                        sizeof (struct GNUNET_PeerIdentity))) )
2392       try_connect (&pid,
2393                    h);
2394   }
2395
2396
2397   /* First, check if 'peer' is already on the path, and if
2398      so, truncate it instead of expanding. */
2399   for (unsigned int i=0;i<=get_path_length;i++)
2400     if (0 == memcmp (&get_path[i],
2401                      peer->id,
2402                      sizeof (struct GNUNET_PeerIdentity)))
2403     {
2404       process_reply_with_path (GNUNET_TIME_absolute_ntoh (prm->expiration_time),
2405                                &prm->key,
2406                                i,
2407                                get_path,
2408                                put_path_length,
2409                                put_path,
2410                                type,
2411                                data_size,
2412                                data);
2413       return;
2414     }
2415
2416   /* Need to append 'peer' to 'get_path' (normal case) */
2417   {
2418     struct GNUNET_PeerIdentity xget_path[get_path_length + 1];
2419
2420     GNUNET_memcpy (xget_path,
2421                    get_path,
2422                    get_path_length * sizeof (struct GNUNET_PeerIdentity));
2423     xget_path[get_path_length] = *peer->id;
2424
2425     process_reply_with_path (GNUNET_TIME_absolute_ntoh (prm->expiration_time),
2426                              &prm->key,
2427                              get_path_length + 1,
2428                              xget_path,
2429                              put_path_length,
2430                              put_path,
2431                              type,
2432                              data_size,
2433                              data);
2434   }
2435 }
2436
2437
2438 /**
2439  * Initialize neighbours subsystem.
2440  *
2441  * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
2442  */
2443 int
2444 GDS_NEIGHBOURS_init ()
2445 {
2446   struct GNUNET_MQ_MessageHandler core_handlers[] = {
2447     GNUNET_MQ_hd_var_size (dht_p2p_get,
2448                            GNUNET_MESSAGE_TYPE_DHT_P2P_GET,
2449                            struct PeerGetMessage,
2450                            NULL),
2451     GNUNET_MQ_hd_var_size (dht_p2p_put,
2452                            GNUNET_MESSAGE_TYPE_DHT_P2P_PUT,
2453                            struct PeerPutMessage,
2454                            NULL),
2455     GNUNET_MQ_hd_var_size (dht_p2p_result,
2456                            GNUNET_MESSAGE_TYPE_DHT_P2P_RESULT,
2457                            struct PeerResultMessage,
2458                            NULL),
2459     GNUNET_MQ_handler_end ()
2460   };
2461   unsigned long long temp_config_num;
2462
2463   disable_try_connect
2464     = GNUNET_CONFIGURATION_get_value_yesno (GDS_cfg,
2465                                             "DHT",
2466                                             "DISABLE_TRY_CONNECT");
2467   if (GNUNET_OK ==
2468       GNUNET_CONFIGURATION_get_value_number (GDS_cfg,
2469                                              "DHT",
2470                                              "bucket_size",
2471                                              &temp_config_num))
2472     bucket_size = (unsigned int) temp_config_num;
2473   cache_results
2474     = GNUNET_CONFIGURATION_get_value_yesno (GDS_cfg,
2475                                             "DHT",
2476                                             "CACHE_RESULTS");
2477
2478   log_route_details_stderr =
2479     (NULL != getenv("GNUNET_DHT_ROUTE_DEBUG")) ? GNUNET_YES : GNUNET_NO;
2480   ats_ch = GNUNET_ATS_connectivity_init (GDS_cfg);
2481   core_api = GNUNET_CORE_connect (GDS_cfg,
2482                                   NULL,
2483                                   &core_init,
2484                                   &handle_core_connect,
2485                                   &handle_core_disconnect,
2486                                   core_handlers);
2487   if (NULL == core_api)
2488     return GNUNET_SYSERR;
2489   all_connected_peers = GNUNET_CONTAINER_multipeermap_create (256,
2490                                                               GNUNET_YES);
2491   all_desired_peers = GNUNET_CONTAINER_multipeermap_create (256,
2492                                                             GNUNET_NO);
2493   return GNUNET_OK;
2494 }
2495
2496
2497 /**
2498  * Shutdown neighbours subsystem.
2499  */
2500 void
2501 GDS_NEIGHBOURS_done ()
2502 {
2503   if (NULL == core_api)
2504     return;
2505   GNUNET_CORE_disconnect (core_api);
2506   core_api = NULL;
2507   GNUNET_assert (0 ==
2508                  GNUNET_CONTAINER_multipeermap_size (all_connected_peers));
2509   GNUNET_CONTAINER_multipeermap_destroy (all_connected_peers);
2510   all_connected_peers = NULL;
2511   GNUNET_CONTAINER_multipeermap_iterate (all_desired_peers,
2512                                          &free_connect_info,
2513                                          NULL);
2514   GNUNET_CONTAINER_multipeermap_destroy (all_desired_peers);
2515   all_desired_peers = NULL;
2516   GNUNET_ATS_connectivity_done (ats_ch);
2517   ats_ch = NULL;
2518   GNUNET_assert (NULL == find_peer_task);
2519 }
2520
2521
2522 /**
2523  * Get the ID of the local node.
2524  *
2525  * @return identity of the local node
2526  */
2527 struct GNUNET_PeerIdentity *
2528 GDS_NEIGHBOURS_get_id ()
2529 {
2530   return &my_identity;
2531 }
2532
2533
2534 /* end of gnunet-service-dht_neighbours.c */