src/dht/gnunet-service-dht_neighbours.c

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