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