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