src/dht/gnunet-service-dht_neighbours.c

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