src/fs/gnunet-service-fs.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009, 2010 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 fs/gnunet-service-fs.c
  23  * @brief gnunet anonymity protocol implementation
  24  * @author Christian Grothoff
  25  *
  26  * TODO:
  27  * - more statistics
  28  */
  29 #include "platform.h"
  30 #include <float.h>
  31 #include "gnunet_constants.h"
  32 #include "gnunet_core_service.h"
  33 #include "gnunet_dht_service.h"
  34 #include "gnunet_datastore_service.h"
  35 #include "gnunet_load_lib.h"
  36 #include "gnunet_peer_lib.h"
  37 #include "gnunet_protocols.h"
  38 #include "gnunet_signatures.h"
  39 #include "gnunet_statistics_service.h"
  40 #include "gnunet_util_lib.h"
  41 #include "gnunet-service-fs_indexing.h"
  42 #include "fs.h"
  43
  44 #define DEBUG_FS GNUNET_NO
  45
  46 /**
  47  * Should we introduce random latency in processing?  Required for proper
  48  * implementation of GAP, but can be disabled for performance evaluation of
  49  * the basic routing algorithm.
  50  *
  51  * Note that with delays enabled, performance can be significantly lower
  52  * (several orders of magnitude in 2-peer test runs); if you want to
  53  * measure throughput of other components, set this to NO.  Also, you
  54  * might want to consider changing 'RETRY_PROBABILITY_INV' to 1 for
  55  * a rather wasteful mode of operation (that might still get the highest
  56  * throughput overall).
  57  *
  58  * Performance measurements (for 50 MB file, 2 peers):
  59  *
  60  * - Without delays: 3300 kb/s
  61  * - With    delays:  101 kb/s
  62  */
  63 #define SUPPORT_DELAYS GNUNET_NO
  64
  65 /**
  66  * Size for the hash map for DHT requests from the FS
  67  * service.  Should be about the number of concurrent
  68  * DHT requests we plan to make.
  69  */
  70 #define FS_DHT_HT_SIZE 1024
  71
  72 /**
  73  * At what frequency should our datastore load decrease
  74  * automatically (since if we don't use it, clearly the
  75  * load must be going down).
  76  */
  77 #define DATASTORE_LOAD_AUTODECLINE GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, 250)
  78
  79 /**
  80  * How often do we flush trust values to disk?
  81  */
  82 #define TRUST_FLUSH_FREQ GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5)
  83
  84 /**
  85  * How often do we at most PUT content into the DHT?
  86  */
  87 #define MAX_DHT_PUT_FREQ GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
  88
  89 /**
  90  * Inverse of the probability that we will submit the same query
  91  * to the same peer again.  If the same peer already got the query
  92  * repeatedly recently, the probability is multiplied by the inverse
  93  * of this number each time.  Note that we only try about every TTL_DECREMENT/2
  94  * plus MAX_CORK_DELAY (so roughly every 3.5s).
  95  *
  96  * Note that this factor is a key influence to performance in small
  97  * networks (especially test networks of 2 peers) because if there is
  98  * only a single peer with the data, this value will determine how
  99  * soon we might re-try.  For example, a value of 3 can result in
 100  * 1.7 MB/s transfer rates for a 10 MB file when a value of 1 would
 101  * give us 5 MB/s.  OTOH, obviously re-trying the same peer can be
 102  * rather inefficient in larger networks, hence picking 1 is in
 103  * general not the best choice.
 104  *
 105  * Performance measurements (for 50 MB file, 2 peers, no delays):
 106  *
 107  * - 1: 3300 kb/s (consistently)
 108  * - 3: 2046 kb/s, 754 kb/s, 3490 kb/s
 109  * - 5:  759 kb/s, 968 kb/s, 1160 kb/s
 110  *
 111  * Note that this does NOT mean that the value should be 1 since
 112  * a 2-peer network is far from representative here (and this fails
 113  * to take into consideration bandwidth wasted by repeatedly
 114  * sending queries to peers that don't have the content).  Also,
 115  * it is expected that higher values lead to more inconsistent
 116  * measurements since this only affects lost messages towards the
 117  * end of the download.
 118  *
 119  * Finally, we should probably consider changing this and making
 120  * it dependent on the number of connected peers or a related
 121  * metric (bad magic constants...).
 122  */
 123 #define RETRY_PROBABILITY_INV 1
 124
 125 /**
 126  * What is the maximum delay for a P2P FS message (in our interaction
 127  * with core)?  FS-internal delays are another story.  The value is
 128  * chosen based on the 32k block size.  Given that peers typcially
 129  * have at least 1 kb/s bandwidth, 45s waits give us a chance to
 130  * transmit one message even to the lowest-bandwidth peers.
 131  */
 132 #define MAX_TRANSMIT_DELAY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 45)
 133
 134 /**
 135  * Maximum number of requests (from other peers, overall) that we're
 136  * willing to have pending at any given point in time.  Can be changed
 137  * via the configuration file (32k is just the default).
 138  */
 139 static unsigned long long max_pending_requests = (32 * 1024);
 140
 141
 142 /**
 143  * Information we keep for each pending reply.  The
 144  * actual message follows at the end of this struct.
 145  */
 146 struct PendingMessage;
 147
 148 /**
 149  * Function called upon completion of a transmission.
 150  *
 151  * @param cls closure
 152  * @param pid ID of receiving peer, 0 on transmission error
 153  */
 154 typedef void (*TransmissionContinuation)(void * cls,
 155                                          GNUNET_PEER_Id tpid);
 156
 157
 158 /**
 159  * Information we keep for each pending message (GET/PUT).  The
 160  * actual message follows at the end of this struct.
 161  */
 162 struct PendingMessage
 163 {
 164   /**
 165    * This is a doubly-linked list of messages to the same peer.
 166    */
 167   struct PendingMessage *next;
 168
 169   /**
 170    * This is a doubly-linked list of messages to the same peer.
 171    */
 172   struct PendingMessage *prev;
 173
 174   /**
 175    * Entry in pending message list for this pending message.
 176    */
 177   struct PendingMessageList *pml;
 178
 179   /**
 180    * Function to call immediately once we have transmitted this
 181    * message.
 182    */
 183   TransmissionContinuation cont;
 184
 185   /**
 186    * Closure for cont.
 187    */
 188   void *cont_cls;
 189
 190   /**
 191    * Do not transmit this pending message until this deadline.
 192    */
 193   struct GNUNET_TIME_Absolute delay_until;
 194
 195   /**
 196    * Size of the reply; actual reply message follows
 197    * at the end of this struct.
 198    */
 199   size_t msize;
 200
 201   /**
 202    * How important is this message for us?
 203    */
 204   uint32_t priority;
 205
 206 };
 207
 208
 209 /**
 210  * Information about a peer that we are connected to.
 211  * We track data that is useful for determining which
 212  * peers should receive our requests.  We also keep
 213  * a list of messages to transmit to this peer.
 214  */
 215 struct ConnectedPeer
 216 {
 217
 218   /**
 219    * List of the last clients for which this peer successfully
 220    * answered a query.
 221    */
 222   struct GNUNET_SERVER_Client *last_client_replies[CS2P_SUCCESS_LIST_SIZE];
 223
 224   /**
 225    * List of the last PIDs for which
 226    * this peer successfully answered a query;
 227    * We use 0 to indicate no successful reply.
 228    */
 229   GNUNET_PEER_Id last_p2p_replies[P2P_SUCCESS_LIST_SIZE];
 230
 231   /**
 232    * Average delay between sending the peer a request and
 233    * getting a reply (only calculated over the requests for
 234    * which we actually got a reply).   Calculated
 235    * as a moving average: new_delay = ((n-1)*last_delay+curr_delay) / n
 236    */
 237   struct GNUNET_TIME_Relative avg_delay;
 238
 239   /**
 240    * Point in time until which this peer does not want us to migrate content
 241    * to it.
 242    */
 243   struct GNUNET_TIME_Absolute migration_blocked;
 244
 245   /**
 246    * Time until when we blocked this peer from migrating
 247    * data to us.
 248    */
 249   struct GNUNET_TIME_Absolute last_migration_block;
 250
 251   /**
 252    * Transmission times for the last MAX_QUEUE_PER_PEER
 253    * requests for this peer.  Used as a ring buffer, current
 254    * offset is stored in 'last_request_times_off'.  If the
 255    * oldest entry is more recent than the 'avg_delay', we should
 256    * not send any more requests right now.
 257    */
 258   struct GNUNET_TIME_Absolute last_request_times[MAX_QUEUE_PER_PEER];
 259
 260   /**
 261    * Handle for an active request for transmission to this
 262    * peer, or NULL.
 263    */
 264   struct GNUNET_CORE_TransmitHandle *cth;
 265
 266   /**
 267    * Messages (replies, queries, content migration) we would like to
 268    * send to this peer in the near future.  Sorted by priority, head.
 269    */
 270   struct PendingMessage *pending_messages_head;
 271
 272   /**
 273    * Messages (replies, queries, content migration) we would like to
 274    * send to this peer in the near future.  Sorted by priority, tail.
 275    */
 276   struct PendingMessage *pending_messages_tail;
 277
 278   /**
 279    * How long does it typically take for us to transmit a message
 280    * to this peer?  (delay between the request being issued and
 281    * the callback being invoked).
 282    */
 283   struct GNUNET_LOAD_Value *transmission_delay;
 284
 285   /**
 286    * Time when the last transmission request was issued.
 287    */
 288   struct GNUNET_TIME_Absolute last_transmission_request_start;
 289
 290   /**
 291    * ID of delay task for scheduling transmission.
 292    */
 293   GNUNET_SCHEDULER_TaskIdentifier delayed_transmission_request_task;
 294
 295   /**
 296    * Average priority of successful replies.  Calculated
 297    * as a moving average: new_avg = ((n-1)*last_avg+curr_prio) / n
 298    */
 299   double avg_priority;
 300
 301   /**
 302    * Increase in traffic preference still to be submitted
 303    * to the core service for this peer.
 304    */
 305   uint64_t inc_preference;
 306
 307   /**
 308    * Trust rating for this peer
 309    */
 310   uint32_t trust;
 311
 312   /**
 313    * Trust rating for this peer on disk.
 314    */
 315   uint32_t disk_trust;
 316
 317   /**
 318    * The peer's identity.
 319    */
 320   GNUNET_PEER_Id pid;
 321
 322   /**
 323    * Size of the linked list of 'pending_messages'.
 324    */
 325   unsigned int pending_requests;
 326
 327   /**
 328    * Which offset in "last_p2p_replies" will be updated next?
 329    * (we go round-robin).
 330    */
 331   unsigned int last_p2p_replies_woff;
 332
 333   /**
 334    * Which offset in "last_client_replies" will be updated next?
 335    * (we go round-robin).
 336    */
 337   unsigned int last_client_replies_woff;
 338
 339   /**
 340    * Current offset into 'last_request_times' ring buffer.
 341    */
 342   unsigned int last_request_times_off;
 343
 344 };
 345
 346
 347 /**
 348  * Information we keep for each pending request.  We should try to
 349  * keep this struct as small as possible since its memory consumption
 350  * is key to how many requests we can have pending at once.
 351  */
 352 struct PendingRequest;
 353
 354
 355 /**
 356  * Doubly-linked list of requests we are performing
 357  * on behalf of the same client.
 358  */
 359 struct ClientRequestList
 360 {
 361
 362   /**
 363    * This is a doubly-linked list.
 364    */
 365   struct ClientRequestList *next;
 366
 367   /**
 368    * This is a doubly-linked list.
 369    */
 370   struct ClientRequestList *prev;
 371
 372   /**
 373    * Request this entry represents.
 374    */
 375   struct PendingRequest *req;
 376
 377   /**
 378    * Client list this request belongs to.
 379    */
 380   struct ClientList *client_list;
 381
 382 };
 383
 384
 385 /**
 386  * Replies to be transmitted to the client.  The actual
 387  * response message is allocated after this struct.
 388  */
 389 struct ClientResponseMessage
 390 {
 391   /**
 392    * This is a doubly-linked list.
 393    */
 394   struct ClientResponseMessage *next;
 395
 396   /**
 397    * This is a doubly-linked list.
 398    */
 399   struct ClientResponseMessage *prev;
 400
 401   /**
 402    * Client list entry this response belongs to.
 403    */
 404   struct ClientList *client_list;
 405
 406   /**
 407    * Number of bytes in the response.
 408    */
 409   size_t msize;
 410 };
 411
 412
 413 /**
 414  * Linked list of clients we are performing requests
 415  * for right now.
 416  */
 417 struct ClientList
 418 {
 419   /**
 420    * This is a linked list.
 421    */
 422   struct ClientList *next;
 423
 424   /**
 425    * ID of a client making a request, NULL if this entry is for a
 426    * peer.
 427    */
 428   struct GNUNET_SERVER_Client *client;
 429
 430   /**
 431    * Head of list of requests performed on behalf
 432    * of this client right now.
 433    */
 434   struct ClientRequestList *rl_head;
 435
 436   /**
 437    * Tail of list of requests performed on behalf
 438    * of this client right now.
 439    */
 440   struct ClientRequestList *rl_tail;
 441
 442   /**
 443    * Head of linked list of responses.
 444    */
 445   struct ClientResponseMessage *res_head;
 446
 447   /**
 448    * Tail of linked list of responses.
 449    */
 450   struct ClientResponseMessage *res_tail;
 451
 452   /**
 453    * Context for sending replies.
 454    */
 455   struct GNUNET_CONNECTION_TransmitHandle *th;
 456
 457 };
 458
 459
 460 /**
 461  * Information about a peer that we have forwarded this
 462  * request to already.
 463  */
 464 struct UsedTargetEntry
 465 {
 466   /**
 467    * What was the last time we have transmitted this request to this
 468    * peer?
 469    */
 470   struct GNUNET_TIME_Absolute last_request_time;
 471
 472   /**
 473    * How often have we transmitted this request to this peer?
 474    */
 475   unsigned int num_requests;
 476
 477   /**
 478    * PID of the target peer.
 479    */
 480   GNUNET_PEER_Id pid;
 481
 482 };
 483
 484
 485
 486
 487
 488 /**
 489  * Doubly-linked list of messages we are performing
 490  * due to a pending request.
 491  */
 492 struct PendingMessageList
 493 {
 494
 495   /**
 496    * This is a doubly-linked list of messages on behalf of the same request.
 497    */
 498   struct PendingMessageList *next;
 499
 500   /**
 501    * This is a doubly-linked list of messages on behalf of the same request.
 502    */
 503   struct PendingMessageList *prev;
 504
 505   /**
 506    * Message this entry represents.
 507    */
 508   struct PendingMessage *pm;
 509
 510   /**
 511    * Request this entry belongs to.
 512    */
 513   struct PendingRequest *req;
 514
 515   /**
 516    * Peer this message is targeted for.
 517    */
 518   struct ConnectedPeer *target;
 519
 520 };
 521
 522
 523 /**
 524  * Information we keep for each pending request.  We should try to
 525  * keep this struct as small as possible since its memory consumption
 526  * is key to how many requests we can have pending at once.
 527  */
 528 struct PendingRequest
 529 {
 530
 531   /**
 532    * If this request was made by a client, this is our entry in the
 533    * client request list; otherwise NULL.
 534    */
 535   struct ClientRequestList *client_request_list;
 536
 537   /**
 538    * Entry of peer responsible for this entry (if this request
 539    * was made by a peer).
 540    */
 541   struct ConnectedPeer *cp;
 542
 543   /**
 544    * If this is a namespace query, pointer to the hash of the public
 545    * key of the namespace; otherwise NULL.  Pointer will be to the
 546    * end of this struct (so no need to free it).
 547    */
 548   const GNUNET_HashCode *namespace;
 549
 550   /**
 551    * Bloomfilter we use to filter out replies that we don't care about
 552    * (anymore).  NULL as long as we are interested in all replies.
 553    */
 554   struct GNUNET_CONTAINER_BloomFilter *bf;
 555
 556   /**
 557    * Context of our GNUNET_CORE_peer_change_preference call.
 558    */
 559   struct GNUNET_CORE_InformationRequestContext *irc;
 560
 561   /**
 562    * Reference to DHT get operation for this request (or NULL).
 563    */
 564   struct GNUNET_DHT_GetHandle *dht_get;
 565
 566   /**
 567    * Hash code of all replies that we have seen so far (only valid
 568    * if client is not NULL since we only track replies like this for
 569    * our own clients).
 570    */
 571   GNUNET_HashCode *replies_seen;
 572
 573   /**
 574    * Node in the heap representing this entry; NULL
 575    * if we have no heap node.
 576    */
 577   struct GNUNET_CONTAINER_HeapNode *hnode;
 578
 579   /**
 580    * Head of list of messages being performed on behalf of this
 581    * request.
 582    */
 583   struct PendingMessageList *pending_head;
 584
 585   /**
 586    * Tail of list of messages being performed on behalf of this
 587    * request.
 588    */
 589   struct PendingMessageList *pending_tail;
 590
 591   /**
 592    * When did we first see this request (form this peer), or, if our
 593    * client is initiating, when did we last initiate a search?
 594    */
 595   struct GNUNET_TIME_Absolute start_time;
 596
 597   /**
 598    * The query that this request is for.
 599    */
 600   GNUNET_HashCode query;
 601
 602   /**
 603    * The task responsible for transmitting queries
 604    * for this request.
 605    */
 606   GNUNET_SCHEDULER_TaskIdentifier task;
 607
 608   /**
 609    * (Interned) Peer identifier that identifies a preferred target
 610    * for requests.
 611    */
 612   GNUNET_PEER_Id target_pid;
 613
 614   /**
 615    * (Interned) Peer identifiers of peers that have already
 616    * received our query for this content.
 617    */
 618   struct UsedTargetEntry *used_targets;
 619
 620   /**
 621    * Our entry in the queue (non-NULL while we wait for our
 622    * turn to interact with the local database).
 623    */
 624   struct GNUNET_DATASTORE_QueueEntry *qe;
 625
 626   /**
 627    * Size of the 'bf' (in bytes).
 628    */
 629   size_t bf_size;
 630
 631   /**
 632    * Desired anonymity level; only valid for requests from a local client.
 633    */
 634   uint32_t anonymity_level;
 635
 636   /**
 637    * How many entries in "used_targets" are actually valid?
 638    */
 639   unsigned int used_targets_off;
 640
 641   /**
 642    * How long is the "used_targets" array?
 643    */
 644   unsigned int used_targets_size;
 645
 646   /**
 647    * Number of results found for this request.
 648    */
 649   unsigned int results_found;
 650
 651   /**
 652    * How many entries in "replies_seen" are actually valid?
 653    */
 654   unsigned int replies_seen_off;
 655
 656   /**
 657    * How long is the "replies_seen" array?
 658    */
 659   unsigned int replies_seen_size;
 660
 661   /**
 662    * Priority with which this request was made.  If one of our clients
 663    * made the request, then this is the current priority that we are
 664    * using when initiating the request.  This value is used when
 665    * we decide to reward other peers with trust for providing a reply.
 666    */
 667   uint32_t priority;
 668
 669   /**
 670    * Priority points left for us to spend when forwarding this request
 671    * to other peers.
 672    */
 673   uint32_t remaining_priority;
 674
 675   /**
 676    * Number to mingle hashes for bloom-filter tests with.
 677    */
 678   int32_t mingle;
 679
 680   /**
 681    * TTL with which we saw this request (or, if we initiated, TTL that
 682    * we used for the request).
 683    */
 684   int32_t ttl;
 685
 686   /**
 687    * Type of the content that this request is for.
 688    */
 689   enum GNUNET_BLOCK_Type type;
 690
 691   /**
 692    * Remove this request after transmission of the current response.
 693    */
 694   int8_t do_remove;
 695
 696   /**
 697    * GNUNET_YES if we should not forward this request to other peers.
 698    */
 699   int8_t local_only;
 700
 701   /**
 702    * GNUNET_YES if we should not forward this request to other peers.
 703    */
 704   int8_t forward_only;
 705
 706 };
 707
 708
 709 /**
 710  * Block that is ready for migration to other peers.  Actual data is at the end of the block.
 711  */
 712 struct MigrationReadyBlock
 713 {
 714
 715   /**
 716    * This is a doubly-linked list.
 717    */
 718   struct MigrationReadyBlock *next;
 719
 720   /**
 721    * This is a doubly-linked list.
 722    */
 723   struct MigrationReadyBlock *prev;
 724
 725   /**
 726    * Query for the block.
 727    */
 728   GNUNET_HashCode query;
 729
 730   /**
 731    * When does this block expire?
 732    */
 733   struct GNUNET_TIME_Absolute expiration;
 734
 735   /**
 736    * Peers we would consider forwarding this
 737    * block to.  Zero for empty entries.
 738    */
 739   GNUNET_PEER_Id target_list[MIGRATION_LIST_SIZE];
 740
 741   /**
 742    * Size of the block.
 743    */
 744   size_t size;
 745
 746   /**
 747    *  Number of targets already used.
 748    */
 749   unsigned int used_targets;
 750
 751   /**
 752    * Type of the block.
 753    */
 754   enum GNUNET_BLOCK_Type type;
 755 };
 756
 757
 758 /**
 759  * Our connection to the datastore.
 760  */
 761 static struct GNUNET_DATASTORE_Handle *dsh;
 762
 763 /**
 764  * Our block context.
 765  */
 766 static struct GNUNET_BLOCK_Context *block_ctx;
 767
 768 /**
 769  * Our block configuration.
 770  */
 771 static struct GNUNET_CONFIGURATION_Handle *block_cfg;
 772
 773 /**
 774  * Our configuration.
 775  */
 776 static const struct GNUNET_CONFIGURATION_Handle *cfg;
 777
 778 /**
 779  * Map of peer identifiers to "struct ConnectedPeer" (for that peer).
 780  */
 781 static struct GNUNET_CONTAINER_MultiHashMap *connected_peers;
 782
 783 /**
 784  * Map of peer identifiers to "struct PendingRequest" (for that peer).
 785  */
 786 static struct GNUNET_CONTAINER_MultiHashMap *peer_request_map;
 787
 788 /**
 789  * Map of query identifiers to "struct PendingRequest" (for that query).
 790  */
 791 static struct GNUNET_CONTAINER_MultiHashMap *query_request_map;
 792
 793 /**
 794  * Heap with the request that will expire next at the top.  Contains
 795  * pointers of type "struct PendingRequest*"; these will *also* be
 796  * aliased from the "requests_by_peer" data structures and the
 797  * "requests_by_query" table.  Note that requests from our clients
 798  * don't expire and are thus NOT in the "requests_by_expiration"
 799  * (or the "requests_by_peer" tables).
 800  */
 801 static struct GNUNET_CONTAINER_Heap *requests_by_expiration_heap;
 802
 803 /**
 804  * Handle for reporting statistics.
 805  */
 806 static struct GNUNET_STATISTICS_Handle *stats;
 807
 808 /**
 809  * Linked list of clients we are currently processing requests for.
 810  */
 811 static struct ClientList *client_list;
 812
 813 /**
 814  * Pointer to handle to the core service (points to NULL until we've
 815  * connected to it).
 816  */
 817 static struct GNUNET_CORE_Handle *core;
 818
 819 /**
 820  * Head of linked list of blocks that can be migrated.
 821  */
 822 static struct MigrationReadyBlock *mig_head;
 823
 824 /**
 825  * Tail of linked list of blocks that can be migrated.
 826  */
 827 static struct MigrationReadyBlock *mig_tail;
 828
 829 /**
 830  * Request to datastore for migration (or NULL).
 831  */
 832 static struct GNUNET_DATASTORE_QueueEntry *mig_qe;
 833
 834 /**
 835  * Request to datastore for DHT PUTs (or NULL).
 836  */
 837 static struct GNUNET_DATASTORE_QueueEntry *dht_qe;
 838
 839 /**
 840  * Type we will request for the next DHT PUT round from the datastore.
 841  */
 842 static enum GNUNET_BLOCK_Type dht_put_type = GNUNET_BLOCK_TYPE_FS_KBLOCK;
 843
 844 /**
 845  * Where do we store trust information?
 846  */
 847 static char *trustDirectory;
 848
 849 /**
 850  * ID of task that collects blocks for migration.
 851  */
 852 static GNUNET_SCHEDULER_TaskIdentifier mig_task;
 853
 854 /**
 855  * ID of task that collects blocks for DHT PUTs.
 856  */
 857 static GNUNET_SCHEDULER_TaskIdentifier dht_task;
 858
 859 /**
 860  * What is the maximum frequency at which we are allowed to
 861  * poll the datastore for migration content?
 862  */
 863 static struct GNUNET_TIME_Relative min_migration_delay;
 864
 865 /**
 866  * Handle for DHT operations.
 867  */
 868 static struct GNUNET_DHT_Handle *dht_handle;
 869
 870 /**
 871  * Size of the doubly-linked list of migration blocks.
 872  */
 873 static unsigned int mig_size;
 874
 875 /**
 876  * Are we allowed to migrate content to this peer.
 877  */
 878 static int active_migration;
 879
 880 /**
 881  * How many entires with zero anonymity do we currently estimate
 882  * to have in the database?
 883  */
 884 static unsigned int zero_anonymity_count_estimate;
 885
 886 /**
 887  * Typical priorities we're seeing from other peers right now.  Since
 888  * most priorities will be zero, this value is the weighted average of
 889  * non-zero priorities seen "recently".  In order to ensure that new
 890  * values do not dramatically change the ratio, values are first
 891  * "capped" to a reasonable range (+N of the current value) and then
 892  * averaged into the existing value by a ratio of 1:N.  Hence
 893  * receiving the largest possible priority can still only raise our
 894  * "current_priorities" by at most 1.
 895  */
 896 static double current_priorities;
 897
 898 /**
 899  * Datastore 'GET' load tracking.
 900  */
 901 static struct GNUNET_LOAD_Value *datastore_get_load;
 902
 903 /**
 904  * Datastore 'PUT' load tracking.
 905  */
 906 static struct GNUNET_LOAD_Value *datastore_put_load;
 907
 908 /**
 909  * How long do requests typically stay in the routing table?
 910  */
 911 static struct GNUNET_LOAD_Value *rt_entry_lifetime;
 912
 913 /**
 914  * We've just now completed a datastore request.  Update our
 915  * datastore load calculations.
 916  *
 917  * @param start time when the datastore request was issued
 918  */
 919 static void
 920 update_datastore_delays (struct GNUNET_TIME_Absolute start)
 921 {
 922   struct GNUNET_TIME_Relative delay;
 923
 924   delay = GNUNET_TIME_absolute_get_duration (start);
 925   GNUNET_LOAD_update (datastore_get_load,
 926                       delay.rel_value);
 927 }
 928
 929
 930 /**
 931  * Get the filename under which we would store the GNUNET_HELLO_Message
 932  * for the given host and protocol.
 933  * @return filename of the form DIRECTORY/HOSTID
 934  */
 935 static char *
 936 get_trust_filename (const struct GNUNET_PeerIdentity *id)
 937 {
 938   struct GNUNET_CRYPTO_HashAsciiEncoded fil;
 939   char *fn;
 940
 941   GNUNET_CRYPTO_hash_to_enc (&id->hashPubKey, &fil);
 942   GNUNET_asprintf (&fn, "%s%s%s", trustDirectory, DIR_SEPARATOR_STR, &fil);
 943   return fn;
 944 }
 945
 946
 947
 948 /**
 949  * Transmit messages by copying it to the target buffer
 950  * "buf".  "buf" will be NULL and "size" zero if the socket was closed
 951  * for writing in the meantime.  In that case, do nothing
 952  * (the disconnect or shutdown handler will take care of the rest).
 953  * If we were able to transmit messages and there are still more
 954  * pending, ask core again for further calls to this function.
 955  *
 956  * @param cls closure, pointer to the 'struct ConnectedPeer*'
 957  * @param size number of bytes available in buf
 958  * @param buf where the callee should write the message
 959  * @return number of bytes written to buf
 960  */
 961 static size_t
 962 transmit_to_peer (void *cls,
 963                   size_t size, void *buf);
 964
 965
 966 /* ******************* clean up functions ************************ */
 967
 968 /**
 969  * Delete the given migration block.
 970  *
 971  * @param mb block to delete
 972  */
 973 static void
 974 delete_migration_block (struct MigrationReadyBlock *mb)
 975 {
 976   GNUNET_CONTAINER_DLL_remove (mig_head,
 977                                mig_tail,
 978                                mb);
 979   GNUNET_PEER_decrement_rcs (mb->target_list,
 980                              MIGRATION_LIST_SIZE);
 981   mig_size--;
 982   GNUNET_free (mb);
 983 }
 984
 985
 986 /**
 987  * Compare the distance of two peers to a key.
 988  *
 989  * @param key key
 990  * @param p1 first peer
 991  * @param p2 second peer
 992  * @return GNUNET_YES if P1 is closer to key than P2
 993  */
 994 static int
 995 is_closer (const GNUNET_HashCode *key,
 996            const struct GNUNET_PeerIdentity *p1,
 997            const struct GNUNET_PeerIdentity *p2)
 998 {
 999   return GNUNET_CRYPTO_hash_xorcmp (&p1->hashPubKey,
1000                                     &p2->hashPubKey,
1001                                     key);
1002 }
1003
1004
1005 /**
1006  * Consider migrating content to a given peer.
1007  *
1008  * @param cls 'struct MigrationReadyBlock*' to select
1009  *            targets for (or NULL for none)
1010  * @param key ID of the peer
1011  * @param value 'struct ConnectedPeer' of the peer
1012  * @return GNUNET_YES (always continue iteration)
1013  */
1014 static int
1015 consider_migration (void *cls,
1016                     const GNUNET_HashCode *key,
1017                     void *value)
1018 {
1019   struct MigrationReadyBlock *mb = cls;
1020   struct ConnectedPeer *cp = value;
1021   struct MigrationReadyBlock *pos;
1022   struct GNUNET_PeerIdentity cppid;
1023   struct GNUNET_PeerIdentity otherpid;
1024   struct GNUNET_PeerIdentity worstpid;
1025   size_t msize;
1026   unsigned int i;
1027   unsigned int repl;
1028
1029   /* consider 'cp' as a migration target for mb */
1030   if (GNUNET_TIME_absolute_get_remaining (cp->migration_blocked).rel_value > 0)
1031     return GNUNET_YES; /* peer has requested no migration! */
1032   if (mb != NULL)
1033     {
1034       GNUNET_PEER_resolve (cp->pid,
1035                            &cppid);
1036       repl = MIGRATION_LIST_SIZE;
1037       for (i=0;i<MIGRATION_LIST_SIZE;i++)
1038         {
1039           if (mb->target_list[i] == 0)
1040             {
1041               mb->target_list[i] = cp->pid;
1042               GNUNET_PEER_change_rc (mb->target_list[i], 1);
1043               repl = MIGRATION_LIST_SIZE;
1044               break;
1045             }
1046           GNUNET_PEER_resolve (mb->target_list[i],
1047                                &otherpid);
1048           if ( (repl == MIGRATION_LIST_SIZE) &&
1049                is_closer (&mb->query,
1050                           &cppid,
1051                           &otherpid))
1052             {
1053               repl = i;
1054               worstpid = otherpid;
1055             }
1056           else if ( (repl != MIGRATION_LIST_SIZE) &&
1057                     (is_closer (&mb->query,
1058                                 &worstpid,
1059                                 &otherpid) ) )
1060             {
1061               repl = i;
1062               worstpid = otherpid;
1063             }
1064         }
1065       if (repl != MIGRATION_LIST_SIZE)
1066         {
1067           GNUNET_PEER_change_rc (mb->target_list[repl], -1);
1068           mb->target_list[repl] = cp->pid;
1069           GNUNET_PEER_change_rc (mb->target_list[repl], 1);
1070         }
1071     }
1072
1073   /* consider scheduling transmission to cp for content migration */
1074   if (cp->cth != NULL)
1075     return GNUNET_YES;
1076   msize = 0;
1077   pos = mig_head;
1078   while (pos != NULL)
1079     {
1080       for (i=0;i<MIGRATION_LIST_SIZE;i++)
1081         {
1082           if (cp->pid == pos->target_list[i])
1083             {
1084               if (msize == 0)
1085                 msize = pos->size;
1086               else
1087                 msize = GNUNET_MIN (msize,
1088                                     pos->size);
1089               break;
1090             }
1091         }
1092       pos = pos->next;
1093     }
1094   if (msize == 0)
1095     return GNUNET_YES; /* no content available */
1096 #if DEBUG_FS
1097   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1098               "Trying to migrate at least %u bytes to peer `%s'\n",
1099               msize,
1100               GNUNET_h2s (key));
1101 #endif
1102   if (cp->delayed_transmission_request_task != GNUNET_SCHEDULER_NO_TASK)
1103     {
1104       GNUNET_SCHEDULER_cancel (cp->delayed_transmission_request_task);
1105       cp->delayed_transmission_request_task = GNUNET_SCHEDULER_NO_TASK;
1106     }
1107   cp->cth
1108     = GNUNET_CORE_notify_transmit_ready (core,
1109                                          0, GNUNET_TIME_UNIT_FOREVER_REL,
1110                                          (const struct GNUNET_PeerIdentity*) key,
1111                                          msize + sizeof (struct PutMessage),
1112                                          &transmit_to_peer,
1113                                          cp);
1114   return GNUNET_YES;
1115 }
1116
1117
1118 /**
1119  * Task that is run periodically to obtain blocks for content
1120  * migration
1121  *
1122  * @param cls unused
1123  * @param tc scheduler context (also unused)
1124  */
1125 static void
1126 gather_migration_blocks (void *cls,
1127                          const struct GNUNET_SCHEDULER_TaskContext *tc);
1128
1129
1130
1131
1132 /**
1133  * Task that is run periodically to obtain blocks for DHT PUTs.
1134  *
1135  * @param cls type of blocks to gather
1136  * @param tc scheduler context (unused)
1137  */
1138 static void
1139 gather_dht_put_blocks (void *cls,
1140                        const struct GNUNET_SCHEDULER_TaskContext *tc);
1141
1142
1143 /**
1144  * If the migration task is not currently running, consider
1145  * (re)scheduling it with the appropriate delay.
1146  */
1147 static void
1148 consider_migration_gathering ()
1149 {
1150   struct GNUNET_TIME_Relative delay;
1151
1152   if (dsh == NULL)
1153     return;
1154   if (mig_qe != NULL)
1155     return;
1156   if (mig_task != GNUNET_SCHEDULER_NO_TASK)
1157     return;
1158   delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
1159                                          mig_size);
1160   delay = GNUNET_TIME_relative_divide (delay,
1161                                        MAX_MIGRATION_QUEUE);
1162   delay = GNUNET_TIME_relative_max (delay,
1163                                     min_migration_delay);
1164   mig_task = GNUNET_SCHEDULER_add_delayed (delay,
1165                                            &gather_migration_blocks,
1166                                            NULL);
1167 }
1168
1169
1170 /**
1171  * If the DHT PUT gathering task is not currently running, consider
1172  * (re)scheduling it with the appropriate delay.
1173  */
1174 static void
1175 consider_dht_put_gathering (void *cls)
1176 {
1177   struct GNUNET_TIME_Relative delay;
1178
1179   if (dsh == NULL)
1180     return;
1181   if (dht_qe != NULL)
1182     return;
1183   if (dht_task != GNUNET_SCHEDULER_NO_TASK)
1184     return;
1185   if (zero_anonymity_count_estimate > 0)
1186     {
1187       delay = GNUNET_TIME_relative_divide (GNUNET_DHT_DEFAULT_REPUBLISH_FREQUENCY,
1188                                            zero_anonymity_count_estimate);
1189       delay = GNUNET_TIME_relative_min (delay,
1190                                         MAX_DHT_PUT_FREQ);
1191     }
1192   else
1193     {
1194       /* if we have NO zero-anonymity content yet, wait 5 minutes for some to
1195          (hopefully) appear */
1196       delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5);
1197     }
1198   dht_task = GNUNET_SCHEDULER_add_delayed (delay,
1199                                            &gather_dht_put_blocks,
1200                                            cls);
1201 }
1202
1203
1204 /**
1205  * Process content offered for migration.
1206  *
1207  * @param cls closure
1208  * @param key key for the content
1209  * @param size number of bytes in data
1210  * @param data content stored
1211  * @param type type of the content
1212  * @param priority priority of the content
1213  * @param anonymity anonymity-level for the content
1214  * @param expiration expiration time for the content
1215  * @param uid unique identifier for the datum;
1216  *        maybe 0 if no unique identifier is available
1217  */
1218 static void
1219 process_migration_content (void *cls,
1220                            const GNUNET_HashCode * key,
1221                            size_t size,
1222                            const void *data,
1223                            enum GNUNET_BLOCK_Type type,
1224                            uint32_t priority,
1225                            uint32_t anonymity,
1226                            struct GNUNET_TIME_Absolute
1227                            expiration, uint64_t uid)
1228 {
1229   struct MigrationReadyBlock *mb;
1230
1231   if (key == NULL)
1232     {
1233       mig_qe = NULL;
1234       if (mig_size < MAX_MIGRATION_QUEUE)
1235         consider_migration_gathering ();
1236       return;
1237     }
1238   if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
1239     {
1240       if (GNUNET_OK !=
1241           GNUNET_FS_handle_on_demand_block (key, size, data,
1242                                             type, priority, anonymity,
1243                                             expiration, uid,
1244                                             &process_migration_content,
1245                                             NULL))
1246         {
1247           GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1248         }
1249       return;
1250     }
1251 #if DEBUG_FS
1252   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1253               "Retrieved block `%s' of type %u for migration\n",
1254               GNUNET_h2s (key),
1255               type);
1256 #endif
1257   mb = GNUNET_malloc (sizeof (struct MigrationReadyBlock) + size);
1258   mb->query = *key;
1259   mb->expiration = expiration;
1260   mb->size = size;
1261   mb->type = type;
1262   memcpy (&mb[1], data, size);
1263   GNUNET_CONTAINER_DLL_insert_after (mig_head,
1264                                      mig_tail,
1265                                      mig_tail,
1266                                      mb);
1267   mig_size++;
1268   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1269                                          &consider_migration,
1270                                          mb);
1271   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1272 }
1273
1274
1275 /**
1276  * Function called upon completion of the DHT PUT operation.
1277  */
1278 static void
1279 dht_put_continuation (void *cls,
1280                       const struct GNUNET_SCHEDULER_TaskContext *tc)
1281 {
1282   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1283 }
1284
1285
1286 /**
1287  * Store content in DHT.
1288  *
1289  * @param cls closure
1290  * @param key key for the content
1291  * @param size number of bytes in data
1292  * @param data content stored
1293  * @param type type of the content
1294  * @param priority priority of the content
1295  * @param anonymity anonymity-level for the content
1296  * @param expiration expiration time for the content
1297  * @param uid unique identifier for the datum;
1298  *        maybe 0 if no unique identifier is available
1299  */
1300 static void
1301 process_dht_put_content (void *cls,
1302                          const GNUNET_HashCode * key,
1303                          size_t size,
1304                          const void *data,
1305                          enum GNUNET_BLOCK_Type type,
1306                          uint32_t priority,
1307                          uint32_t anonymity,
1308                          struct GNUNET_TIME_Absolute
1309                          expiration, uint64_t uid)
1310 {
1311   static unsigned int counter;
1312   static GNUNET_HashCode last_vhash;
1313   static GNUNET_HashCode vhash;
1314
1315   if (key == NULL)
1316     {
1317       dht_qe = NULL;
1318       consider_dht_put_gathering (cls);
1319       return;
1320     }
1321   /* slightly funky code to estimate the total number of values with zero
1322      anonymity from the maximum observed length of a monotonically increasing
1323      sequence of hashes over the contents */
1324   GNUNET_CRYPTO_hash (data, size, &vhash);
1325   if (GNUNET_CRYPTO_hash_cmp (&vhash, &last_vhash) <= 0)
1326     {
1327       if (zero_anonymity_count_estimate > 0)
1328         zero_anonymity_count_estimate /= 2;
1329       counter = 0;
1330     }
1331   last_vhash = vhash;
1332   if (counter < 31)
1333     counter++;
1334   if (zero_anonymity_count_estimate < (1 << counter))
1335     zero_anonymity_count_estimate = (1 << counter);
1336 #if DEBUG_FS
1337   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1338               "Retrieved block `%s' of type %u for DHT PUT\n",
1339               GNUNET_h2s (key),
1340               type);
1341 #endif
1342   GNUNET_DHT_put (dht_handle,
1343                   key,
1344                   DEFAULT_PUT_REPLICATION,
1345                   GNUNET_DHT_RO_NONE,
1346                   type,
1347                   size,
1348                   data,
1349                   expiration,
1350                   GNUNET_TIME_UNIT_FOREVER_REL,
1351                   &dht_put_continuation,
1352                   cls);
1353 }
1354
1355
1356 /**
1357  * Task that is run periodically to obtain blocks for content
1358  * migration
1359  *
1360  * @param cls unused
1361  * @param tc scheduler context (also unused)
1362  */
1363 static void
1364 gather_migration_blocks (void *cls,
1365                          const struct GNUNET_SCHEDULER_TaskContext *tc)
1366 {
1367   mig_task = GNUNET_SCHEDULER_NO_TASK;
1368   if (dsh != NULL)
1369     {
1370       mig_qe = GNUNET_DATASTORE_get_random (dsh, 0, UINT_MAX,
1371                                             GNUNET_TIME_UNIT_FOREVER_REL,
1372                                             &process_migration_content, NULL);
1373       GNUNET_assert (mig_qe != NULL);
1374     }
1375 }
1376
1377
1378 /**
1379  * Task that is run periodically to obtain blocks for DHT PUTs.
1380  *
1381  * @param cls type of blocks to gather
1382  * @param tc scheduler context (unused)
1383  */
1384 static void
1385 gather_dht_put_blocks (void *cls,
1386                        const struct GNUNET_SCHEDULER_TaskContext *tc)
1387 {
1388   dht_task = GNUNET_SCHEDULER_NO_TASK;
1389   if (dsh != NULL)
1390     {
1391       if (dht_put_type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
1392         dht_put_type = GNUNET_BLOCK_TYPE_FS_KBLOCK;
1393       dht_qe = GNUNET_DATASTORE_get_zero_anonymity (dsh, 0, UINT_MAX,
1394                                                     GNUNET_TIME_UNIT_FOREVER_REL,
1395                                                     dht_put_type++,
1396                                                     &process_dht_put_content, NULL);
1397       GNUNET_assert (dht_qe != NULL);
1398     }
1399 }
1400
1401
1402 /**
1403  * We're done with a particular message list entry.
1404  * Free all associated resources.
1405  *
1406  * @param pml entry to destroy
1407  */
1408 static void
1409 destroy_pending_message_list_entry (struct PendingMessageList *pml)
1410 {
1411   GNUNET_CONTAINER_DLL_remove (pml->req->pending_head,
1412                                pml->req->pending_tail,
1413                                pml);
1414   GNUNET_CONTAINER_DLL_remove (pml->target->pending_messages_head,
1415                                pml->target->pending_messages_tail,
1416                                pml->pm);
1417   pml->target->pending_requests--;
1418   GNUNET_free (pml->pm);
1419   GNUNET_free (pml);
1420 }
1421
1422
1423 /**
1424  * Destroy the given pending message (and call the respective
1425  * continuation).
1426  *
1427  * @param pm message to destroy
1428  * @param tpid id of peer that the message was delivered to, or 0 for none
1429  */
1430 static void
1431 destroy_pending_message (struct PendingMessage *pm,
1432                          GNUNET_PEER_Id tpid)
1433 {
1434   struct PendingMessageList *pml = pm->pml;
1435   TransmissionContinuation cont;
1436   void *cont_cls;
1437
1438   cont = pm->cont;
1439   cont_cls = pm->cont_cls;
1440   if (pml != NULL)
1441     {
1442       GNUNET_assert (pml->pm == pm);
1443       GNUNET_assert ( (tpid == 0) || (tpid == pml->target->pid) );
1444       destroy_pending_message_list_entry (pml);
1445     }
1446   else
1447     {
1448       GNUNET_free (pm);
1449     }
1450   if (cont != NULL)
1451     cont (cont_cls, tpid);
1452 }
1453
1454
1455 /**
1456  * We're done processing a particular request.
1457  * Free all associated resources.
1458  *
1459  * @param pr request to destroy
1460  */
1461 static void
1462 destroy_pending_request (struct PendingRequest *pr)
1463 {
1464   struct GNUNET_PeerIdentity pid;
1465   unsigned int i;
1466
1467   if (pr->hnode != NULL)
1468     {
1469       GNUNET_CONTAINER_heap_remove_node (requests_by_expiration_heap,
1470                                          pr->hnode);
1471       pr->hnode = NULL;
1472     }
1473   if (NULL == pr->client_request_list)
1474     {
1475       GNUNET_STATISTICS_update (stats,
1476                                 gettext_noop ("# P2P searches active"),
1477                                 -1,
1478                                 GNUNET_NO);
1479     }
1480   else
1481     {
1482       GNUNET_STATISTICS_update (stats,
1483                                 gettext_noop ("# client searches active"),
1484                                 -1,
1485                                 GNUNET_NO);
1486     }
1487   if (GNUNET_YES ==
1488       GNUNET_CONTAINER_multihashmap_remove (query_request_map,
1489                                             &pr->query,
1490                                             pr))
1491     {
1492       GNUNET_LOAD_update (rt_entry_lifetime,
1493                           GNUNET_TIME_absolute_get_duration (pr->start_time).rel_value);
1494     }
1495   if (pr->qe != NULL)
1496      {
1497       GNUNET_DATASTORE_cancel (pr->qe);
1498       pr->qe = NULL;
1499     }
1500   if (pr->dht_get != NULL)
1501     {
1502       GNUNET_DHT_get_stop (pr->dht_get);
1503       pr->dht_get = NULL;
1504     }
1505   if (pr->client_request_list != NULL)
1506     {
1507       GNUNET_CONTAINER_DLL_remove (pr->client_request_list->client_list->rl_head,
1508                                    pr->client_request_list->client_list->rl_tail,
1509                                    pr->client_request_list);
1510       GNUNET_free (pr->client_request_list);
1511       pr->client_request_list = NULL;
1512     }
1513   if (pr->cp != NULL)
1514     {
1515       GNUNET_PEER_resolve (pr->cp->pid,
1516                            &pid);
1517       (void) GNUNET_CONTAINER_multihashmap_remove (peer_request_map,
1518                                                    &pid.hashPubKey,
1519                                                    pr);
1520       pr->cp = NULL;
1521     }
1522   if (pr->bf != NULL)
1523     {
1524       GNUNET_CONTAINER_bloomfilter_free (pr->bf);
1525       pr->bf = NULL;
1526     }
1527   if (pr->irc != NULL)
1528     {
1529       GNUNET_CORE_peer_change_preference_cancel (pr->irc);
1530       pr->irc = NULL;
1531     }
1532   if (pr->replies_seen != NULL)
1533     {
1534       GNUNET_free (pr->replies_seen);
1535       pr->replies_seen = NULL;
1536     }
1537   if (pr->task != GNUNET_SCHEDULER_NO_TASK)
1538     {
1539       GNUNET_SCHEDULER_cancel (pr->task);
1540       pr->task = GNUNET_SCHEDULER_NO_TASK;
1541     }
1542   while (NULL != pr->pending_head)
1543     destroy_pending_message_list_entry (pr->pending_head);
1544   GNUNET_PEER_change_rc (pr->target_pid, -1);
1545   if (pr->used_targets != NULL)
1546     {
1547       for (i=0;i<pr->used_targets_off;i++)
1548         GNUNET_PEER_change_rc (pr->used_targets[i].pid, -1);
1549       GNUNET_free (pr->used_targets);
1550       pr->used_targets_off = 0;
1551       pr->used_targets_size = 0;
1552       pr->used_targets = NULL;
1553     }
1554   GNUNET_free (pr);
1555 }
1556
1557
1558 /**
1559  * Method called whenever a given peer connects.
1560  *
1561  * @param cls closure, not used
1562  * @param peer peer identity this notification is about
1563  * @param latency reported latency of the connection with 'other'
1564  * @param distance reported distance (DV) to 'other'
1565  */
1566 static void
1567 peer_connect_handler (void *cls,
1568                       const struct
1569                       GNUNET_PeerIdentity * peer,
1570                       struct GNUNET_TIME_Relative latency,
1571                       uint32_t distance)
1572 {
1573   struct ConnectedPeer *cp;
1574   struct MigrationReadyBlock *pos;
1575   char *fn;
1576   uint32_t trust;
1577
1578   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
1579                                           &peer->hashPubKey);
1580   if (NULL != cp)
1581     {
1582       GNUNET_break (0);
1583       return;
1584     }
1585   cp = GNUNET_malloc (sizeof (struct ConnectedPeer));
1586   cp->transmission_delay = GNUNET_LOAD_value_init (latency);
1587   cp->pid = GNUNET_PEER_intern (peer);
1588
1589   fn = get_trust_filename (peer);
1590   if ((GNUNET_DISK_file_test (fn) == GNUNET_YES) &&
1591       (sizeof (trust) == GNUNET_DISK_fn_read (fn, &trust, sizeof (trust))))
1592     cp->disk_trust = cp->trust = ntohl (trust);
1593   GNUNET_free (fn);
1594
1595   GNUNET_break (GNUNET_OK ==
1596                 GNUNET_CONTAINER_multihashmap_put (connected_peers,
1597                                                    &peer->hashPubKey,
1598                                                    cp,
1599                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1600
1601   pos = mig_head;
1602   while (NULL != pos)
1603     {
1604       (void) consider_migration (pos, &peer->hashPubKey, cp);
1605       pos = pos->next;
1606     }
1607 }
1608
1609
1610 /**
1611  * Method called whenever a given peer has a status change.
1612  *
1613  * @param cls closure
1614  * @param peer peer identity this notification is about
1615  * @param latency reported latency of the connection with 'other'
1616  * @param distance reported distance (DV) to 'other'
1617  * @param bandwidth_in available amount of inbound bandwidth
1618  * @param bandwidth_out available amount of outbound bandwidth
1619  * @param timeout absolute time when this peer will time out
1620  *        unless we see some further activity from it
1621  */
1622 static void
1623 peer_status_handler (void *cls,
1624                      const struct
1625                      GNUNET_PeerIdentity * peer,
1626                      struct GNUNET_TIME_Relative latency,
1627                      uint32_t distance,
1628                      struct GNUNET_BANDWIDTH_Value32NBO bandwidth_in,
1629                      struct GNUNET_BANDWIDTH_Value32NBO bandwidth_out,
1630                      struct GNUNET_TIME_Absolute timeout)
1631 {
1632   struct ConnectedPeer *cp;
1633
1634   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
1635                                           &peer->hashPubKey);
1636   if (cp == NULL)
1637     {
1638       GNUNET_break (0);
1639       return;
1640     }
1641   GNUNET_LOAD_value_set_decline (cp->transmission_delay,
1642                                  latency);
1643 }
1644
1645
1646
1647 /**
1648  * Increase the host credit by a value.
1649  *
1650  * @param host which peer to change the trust value on
1651  * @param value is the int value by which the
1652  *  host credit is to be increased or decreased
1653  * @returns the actual change in trust (positive or negative)
1654  */
1655 static int
1656 change_host_trust (struct ConnectedPeer *host, int value)
1657 {
1658   if (value == 0)
1659     return 0;
1660   GNUNET_assert (host != NULL);
1661   if (value > 0)
1662     {
1663       if (host->trust + value < host->trust)
1664         {
1665           value = UINT32_MAX - host->trust;
1666           host->trust = UINT32_MAX;
1667         }
1668       else
1669         host->trust += value;
1670     }
1671   else
1672     {
1673       if (host->trust < -value)
1674         {
1675           value = -host->trust;
1676           host->trust = 0;
1677         }
1678       else
1679         host->trust += value;
1680     }
1681   return value;
1682 }
1683
1684
1685 /**
1686  * Write host-trust information to a file - flush the buffer entry!
1687  */
1688 static int
1689 flush_trust (void *cls,
1690              const GNUNET_HashCode *key,
1691              void *value)
1692 {
1693   struct ConnectedPeer *host = value;
1694   char *fn;
1695   uint32_t trust;
1696   struct GNUNET_PeerIdentity pid;
1697
1698   if (host->trust == host->disk_trust)
1699     return GNUNET_OK;                     /* unchanged */
1700   GNUNET_PEER_resolve (host->pid,
1701                        &pid);
1702   fn = get_trust_filename (&pid);
1703   if (host->trust == 0)
1704     {
1705       if ((0 != UNLINK (fn)) && (errno != ENOENT))
1706         GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_WARNING |
1707                                   GNUNET_ERROR_TYPE_BULK, "unlink", fn);
1708     }
1709   else
1710     {
1711       trust = htonl (host->trust);
1712       if (sizeof(uint32_t) == GNUNET_DISK_fn_write (fn, &trust,
1713                                                     sizeof(uint32_t),
1714                                                     GNUNET_DISK_PERM_USER_READ | GNUNET_DISK_PERM_USER_WRITE
1715                                                     | GNUNET_DISK_PERM_GROUP_READ | GNUNET_DISK_PERM_OTHER_READ))
1716         host->disk_trust = host->trust;
1717     }
1718   GNUNET_free (fn);
1719   return GNUNET_OK;
1720 }
1721
1722 /**
1723  * Call this method periodically to scan data/hosts for new hosts.
1724  */
1725 static void
1726 cron_flush_trust (void *cls,
1727                   const struct GNUNET_SCHEDULER_TaskContext *tc)
1728 {
1729
1730   if (NULL == connected_peers)
1731     return;
1732   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1733                                          &flush_trust,
1734                                          NULL);
1735   if (NULL == tc)
1736     return;
1737   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1738     return;
1739   GNUNET_SCHEDULER_add_delayed (TRUST_FLUSH_FREQ, &cron_flush_trust, NULL);
1740 }
1741
1742
1743 /**
1744  * Free (each) request made by the peer.
1745  *
1746  * @param cls closure, points to peer that the request belongs to
1747  * @param key current key code
1748  * @param value value in the hash map
1749  * @return GNUNET_YES (we should continue to iterate)
1750  */
1751 static int
1752 destroy_request (void *cls,
1753                  const GNUNET_HashCode * key,
1754                  void *value)
1755 {
1756   const struct GNUNET_PeerIdentity * peer = cls;
1757   struct PendingRequest *pr = value;
1758
1759   GNUNET_break (GNUNET_YES ==
1760                 GNUNET_CONTAINER_multihashmap_remove (peer_request_map,
1761                                                       &peer->hashPubKey,
1762                                                       pr));
1763   destroy_pending_request (pr);
1764   return GNUNET_YES;
1765 }
1766
1767
1768 /**
1769  * Method called whenever a peer disconnects.
1770  *
1771  * @param cls closure, not used
1772  * @param peer peer identity this notification is about
1773  */
1774 static void
1775 peer_disconnect_handler (void *cls,
1776                          const struct
1777                          GNUNET_PeerIdentity * peer)
1778 {
1779   struct ConnectedPeer *cp;
1780   struct PendingMessage *pm;
1781   unsigned int i;
1782   struct MigrationReadyBlock *pos;
1783   struct MigrationReadyBlock *next;
1784
1785   GNUNET_CONTAINER_multihashmap_get_multiple (peer_request_map,
1786                                               &peer->hashPubKey,
1787                                               &destroy_request,
1788                                               (void*) peer);
1789   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
1790                                           &peer->hashPubKey);
1791   if (cp == NULL)
1792     return;
1793   for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
1794     {
1795       if (NULL != cp->last_client_replies[i])
1796         {
1797           GNUNET_SERVER_client_drop (cp->last_client_replies[i]);
1798           cp->last_client_replies[i] = NULL;
1799         }
1800     }
1801   GNUNET_break (GNUNET_YES ==
1802                 GNUNET_CONTAINER_multihashmap_remove (connected_peers,
1803                                                       &peer->hashPubKey,
1804                                                       cp));
1805   /* remove this peer from migration considerations; schedule
1806      alternatives */
1807   next = mig_head;
1808   while (NULL != (pos = next))
1809     {
1810       next = pos->next;
1811       for (i=0;i<MIGRATION_LIST_SIZE;i++)
1812         {
1813           if (pos->target_list[i] == cp->pid)
1814             {
1815               GNUNET_PEER_change_rc (pos->target_list[i], -1);
1816               pos->target_list[i] = 0;
1817             }
1818          }
1819       if (pos->used_targets >= GNUNET_CONTAINER_multihashmap_size (connected_peers))
1820         {
1821           delete_migration_block (pos);
1822           consider_migration_gathering ();
1823           continue;
1824         }
1825       GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1826                                              &consider_migration,
1827                                              pos);
1828     }
1829   GNUNET_PEER_change_rc (cp->pid, -1);
1830   GNUNET_PEER_decrement_rcs (cp->last_p2p_replies, P2P_SUCCESS_LIST_SIZE);
1831   if (NULL != cp->cth)
1832     {
1833       GNUNET_CORE_notify_transmit_ready_cancel (cp->cth);
1834       cp->cth = NULL;
1835     }
1836   if (cp->delayed_transmission_request_task != GNUNET_SCHEDULER_NO_TASK)
1837     {
1838       GNUNET_SCHEDULER_cancel (cp->delayed_transmission_request_task);
1839       cp->delayed_transmission_request_task = GNUNET_SCHEDULER_NO_TASK;
1840     }
1841   while (NULL != (pm = cp->pending_messages_head))
1842     destroy_pending_message (pm, 0 /* delivery failed */);
1843   GNUNET_LOAD_value_free (cp->transmission_delay);
1844   GNUNET_break (0 == cp->pending_requests);
1845   GNUNET_free (cp);
1846 }
1847
1848
1849 /**
1850  * Iterator over hash map entries that removes all occurences
1851  * of the given 'client' from the 'last_client_replies' of the
1852  * given connected peer.
1853  *
1854  * @param cls closure, the 'struct GNUNET_SERVER_Client*' to remove
1855  * @param key current key code (unused)
1856  * @param value value in the hash map (the 'struct ConnectedPeer*' to change)
1857  * @return GNUNET_YES (we should continue to iterate)
1858  */
1859 static int
1860 remove_client_from_last_client_replies (void *cls,
1861                                         const GNUNET_HashCode * key,
1862                                         void *value)
1863 {
1864   struct GNUNET_SERVER_Client *client = cls;
1865   struct ConnectedPeer *cp = value;
1866   unsigned int i;
1867
1868   for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
1869     {
1870       if (cp->last_client_replies[i] == client)
1871         {
1872           GNUNET_SERVER_client_drop (cp->last_client_replies[i]);
1873           cp->last_client_replies[i] = NULL;
1874         }
1875     }
1876   return GNUNET_YES;
1877 }
1878
1879
1880 /**
1881  * A client disconnected.  Remove all of its pending queries.
1882  *
1883  * @param cls closure, NULL
1884  * @param client identification of the client
1885  */
1886 static void
1887 handle_client_disconnect (void *cls,
1888                           struct GNUNET_SERVER_Client
1889                           * client)
1890 {
1891   struct ClientList *pos;
1892   struct ClientList *prev;
1893   struct ClientRequestList *rcl;
1894   struct ClientResponseMessage *creply;
1895
1896   if (client == NULL)
1897     return;
1898   prev = NULL;
1899   pos = client_list;
1900   while ( (NULL != pos) &&
1901           (pos->client != client) )
1902     {
1903       prev = pos;
1904       pos = pos->next;
1905     }
1906   if (pos == NULL)
1907     return; /* no requests pending for this client */
1908   while (NULL != (rcl = pos->rl_head))
1909     {
1910       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1911                   "Destroying pending request `%s' on disconnect\n",
1912                   GNUNET_h2s (&rcl->req->query));
1913       destroy_pending_request (rcl->req);
1914     }
1915   if (prev == NULL)
1916     client_list = pos->next;
1917   else
1918     prev->next = pos->next;
1919   if (pos->th != NULL)
1920     {
1921       GNUNET_CONNECTION_notify_transmit_ready_cancel (pos->th);
1922       pos->th = NULL;
1923     }
1924   while (NULL != (creply = pos->res_head))
1925     {
1926       GNUNET_CONTAINER_DLL_remove (pos->res_head,
1927                                    pos->res_tail,
1928                                    creply);
1929       GNUNET_free (creply);
1930     }
1931   GNUNET_SERVER_client_drop (pos->client);
1932   GNUNET_free (pos);
1933   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1934                                          &remove_client_from_last_client_replies,
1935                                          client);
1936 }
1937
1938
1939 /**
1940  * Iterator to free peer entries.
1941  *
1942  * @param cls closure, unused
1943  * @param key current key code
1944  * @param value value in the hash map (peer entry)
1945  * @return GNUNET_YES (we should continue to iterate)
1946  */
1947 static int
1948 clean_peer (void *cls,
1949             const GNUNET_HashCode * key,
1950             void *value)
1951 {
1952   peer_disconnect_handler (NULL, (const struct GNUNET_PeerIdentity*) key);
1953   return GNUNET_YES;
1954 }
1955
1956
1957 /**
1958  * Task run during shutdown.
1959  *
1960  * @param cls unused
1961  * @param tc unused
1962  */
1963 static void
1964 shutdown_task (void *cls,
1965                const struct GNUNET_SCHEDULER_TaskContext *tc)
1966 {
1967   if (mig_qe != NULL)
1968     {
1969       GNUNET_DATASTORE_cancel (mig_qe);
1970       mig_qe = NULL;
1971     }
1972   if (dht_qe != NULL)
1973     {
1974       GNUNET_DATASTORE_cancel (dht_qe);
1975       dht_qe = NULL;
1976     }
1977   if (GNUNET_SCHEDULER_NO_TASK != mig_task)
1978     {
1979       GNUNET_SCHEDULER_cancel (mig_task);
1980       mig_task = GNUNET_SCHEDULER_NO_TASK;
1981     }
1982   if (GNUNET_SCHEDULER_NO_TASK != dht_task)
1983     {
1984       GNUNET_SCHEDULER_cancel (dht_task);
1985       dht_task = GNUNET_SCHEDULER_NO_TASK;
1986     }
1987   while (client_list != NULL)
1988     handle_client_disconnect (NULL,
1989                               client_list->client);
1990   cron_flush_trust (NULL, NULL);
1991   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1992                                          &clean_peer,
1993                                          NULL);
1994   GNUNET_break (0 == GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap));
1995   GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
1996   requests_by_expiration_heap = 0;
1997   GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
1998   connected_peers = NULL;
1999   GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_size (query_request_map));
2000   GNUNET_CONTAINER_multihashmap_destroy (query_request_map);
2001   query_request_map = NULL;
2002   GNUNET_LOAD_value_free (rt_entry_lifetime);
2003   rt_entry_lifetime = NULL;
2004   GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_size (peer_request_map));
2005   GNUNET_CONTAINER_multihashmap_destroy (peer_request_map);
2006   peer_request_map = NULL;
2007   GNUNET_assert (NULL != core);
2008   GNUNET_CORE_disconnect (core);
2009   core = NULL;
2010   if (stats != NULL)
2011     {
2012       GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
2013       stats = NULL;
2014     }
2015   if (dsh != NULL)
2016     {
2017       GNUNET_DATASTORE_disconnect (dsh,
2018                                    GNUNET_NO);
2019       dsh = NULL;
2020     }
2021   while (mig_head != NULL)
2022     delete_migration_block (mig_head);
2023   GNUNET_assert (0 == mig_size);
2024   GNUNET_DHT_disconnect (dht_handle);
2025   dht_handle = NULL;
2026   GNUNET_LOAD_value_free (datastore_get_load);
2027   datastore_get_load = NULL;
2028   GNUNET_LOAD_value_free (datastore_put_load);
2029   datastore_put_load = NULL;
2030   GNUNET_BLOCK_context_destroy (block_ctx);
2031   block_ctx = NULL;
2032   GNUNET_CONFIGURATION_destroy (block_cfg);
2033   block_cfg = NULL;
2034   cfg = NULL;
2035   GNUNET_free_non_null (trustDirectory);
2036   trustDirectory = NULL;
2037 }
2038
2039
2040 /* ******************* Utility functions  ******************** */
2041
2042
2043 /**
2044  * We've had to delay a request for transmission to core, but now
2045  * we should be ready.  Run it.
2046  *
2047  * @param cls the 'struct ConnectedPeer' for which a request was delayed
2048  * @param tc task context (unused)
2049  */
2050 static void
2051 delayed_transmission_request (void *cls,
2052                               const struct GNUNET_SCHEDULER_TaskContext *tc)
2053 {
2054   struct ConnectedPeer *cp = cls;
2055   struct GNUNET_PeerIdentity pid;
2056   struct PendingMessage *pm;
2057
2058   pm = cp->pending_messages_head;
2059   cp->delayed_transmission_request_task = GNUNET_SCHEDULER_NO_TASK;
2060   GNUNET_assert (cp->cth == NULL);
2061   if (pm == NULL)
2062     return;
2063   GNUNET_PEER_resolve (cp->pid,
2064                        &pid);
2065   cp->last_transmission_request_start = GNUNET_TIME_absolute_get ();
2066   cp->cth = GNUNET_CORE_notify_transmit_ready (core,
2067                                                pm->priority,
2068                                                GNUNET_CONSTANTS_SERVICE_TIMEOUT,
2069                                                &pid,
2070                                                pm->msize,
2071                                                &transmit_to_peer,
2072                                                cp);
2073 }
2074
2075
2076 /**
2077  * Transmit messages by copying it to the target buffer
2078  * "buf".  "buf" will be NULL and "size" zero if the socket was closed
2079  * for writing in the meantime.  In that case, do nothing
2080  * (the disconnect or shutdown handler will take care of the rest).
2081  * If we were able to transmit messages and there are still more
2082  * pending, ask core again for further calls to this function.
2083  *
2084  * @param cls closure, pointer to the 'struct ConnectedPeer*'
2085  * @param size number of bytes available in buf
2086  * @param buf where the callee should write the message
2087  * @return number of bytes written to buf
2088  */
2089 static size_t
2090 transmit_to_peer (void *cls,
2091                   size_t size, void *buf)
2092 {
2093   struct ConnectedPeer *cp = cls;
2094   char *cbuf = buf;
2095   struct PendingMessage *pm;
2096   struct PendingMessage *next_pm;
2097   struct GNUNET_TIME_Absolute now;
2098   struct GNUNET_TIME_Relative min_delay;
2099   struct MigrationReadyBlock *mb;
2100   struct MigrationReadyBlock *next;
2101   struct PutMessage migm;
2102   size_t msize;
2103   unsigned int i;
2104   struct GNUNET_PeerIdentity pid;
2105
2106   cp->cth = NULL;
2107   if (NULL == buf)
2108     {
2109 #if DEBUG_FS
2110       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2111                   "Dropping message, core too busy.\n");
2112 #endif
2113       GNUNET_LOAD_update (cp->transmission_delay,
2114                           UINT64_MAX);
2115       return 0;
2116     }
2117   GNUNET_LOAD_update (cp->transmission_delay,
2118                       GNUNET_TIME_absolute_get_duration (cp->last_transmission_request_start).rel_value);
2119   now = GNUNET_TIME_absolute_get ();
2120   msize = 0;
2121   min_delay = GNUNET_TIME_UNIT_FOREVER_REL;
2122   next_pm = cp->pending_messages_head;
2123   while ( (NULL != (pm = next_pm) ) &&
2124           (pm->msize <= size) )
2125     {
2126       next_pm = pm->next;
2127       if (pm->delay_until.abs_value > now.abs_value)
2128         {
2129           min_delay = GNUNET_TIME_relative_min (min_delay,
2130                                                 GNUNET_TIME_absolute_get_remaining (pm->delay_until));
2131           continue;
2132         }
2133       memcpy (&cbuf[msize], &pm[1], pm->msize);
2134       msize += pm->msize;
2135       size -= pm->msize;
2136       if (NULL == pm->pml)
2137         {
2138           GNUNET_CONTAINER_DLL_remove (cp->pending_messages_head,
2139                                        cp->pending_messages_tail,
2140                                        pm);
2141           cp->pending_requests--;
2142         }
2143       destroy_pending_message (pm, cp->pid);
2144     }
2145   if (pm != NULL)
2146     min_delay = GNUNET_TIME_UNIT_ZERO;
2147   if (NULL != cp->pending_messages_head)
2148     {
2149       GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == cp->delayed_transmission_request_task);
2150       cp->delayed_transmission_request_task
2151         = GNUNET_SCHEDULER_add_delayed (min_delay,
2152                                         &delayed_transmission_request,
2153                                         cp);
2154     }
2155   if (pm == NULL)
2156     {
2157       GNUNET_PEER_resolve (cp->pid,
2158                            &pid);
2159       next = mig_head;
2160       while (NULL != (mb = next))
2161         {
2162           next = mb->next;
2163           for (i=0;i<MIGRATION_LIST_SIZE;i++)
2164             {
2165               if ( (cp->pid == mb->target_list[i]) &&
2166                    (mb->size + sizeof (migm) <= size) )
2167                 {
2168                   GNUNET_PEER_change_rc (mb->target_list[i], -1);
2169                   mb->target_list[i] = 0;
2170                   mb->used_targets++;
2171                   memset (&migm, 0, sizeof (migm));
2172                   migm.header.size = htons (sizeof (migm) + mb->size);
2173                   migm.header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
2174                   migm.type = htonl (mb->type);
2175                   migm.expiration = GNUNET_TIME_absolute_hton (mb->expiration);
2176                   memcpy (&cbuf[msize], &migm, sizeof (migm));
2177                   msize += sizeof (migm);
2178                   size -= sizeof (migm);
2179                   memcpy (&cbuf[msize], &mb[1], mb->size);
2180                   msize += mb->size;
2181                   size -= mb->size;
2182 #if DEBUG_FS
2183                   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2184                               "Pushing migration block `%s' (%u bytes) to `%s'\n",
2185                               GNUNET_h2s (&mb->query),
2186                               (unsigned int) mb->size,
2187                               GNUNET_i2s (&pid));
2188 #endif
2189                   break;
2190                 }
2191               else
2192                 {
2193 #if DEBUG_FS
2194                   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2195                               "Migration block `%s' (%u bytes) is not on migration list for peer `%s'\n",
2196                               GNUNET_h2s (&mb->query),
2197                               (unsigned int) mb->size,
2198                               GNUNET_i2s (&pid));
2199 #endif
2200                 }
2201             }
2202           if ( (mb->used_targets >= MIGRATION_TARGET_COUNT) ||
2203                (mb->used_targets >= GNUNET_CONTAINER_multihashmap_size (connected_peers)) )
2204             {
2205               delete_migration_block (mb);
2206               consider_migration_gathering ();
2207             }
2208         }
2209       consider_migration (NULL,
2210                           &pid.hashPubKey,
2211                           cp);
2212     }
2213 #if DEBUG_FS
2214   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2215               "Transmitting %u bytes to peer with PID %u\n",
2216               (unsigned int) msize,
2217               (unsigned int) cp->pid);
2218 #endif
2219   return msize;
2220 }
2221
2222
2223 /**
2224  * Add a message to the set of pending messages for the given peer.
2225  *
2226  * @param cp peer to send message to
2227  * @param pm message to queue
2228  * @param pr request on which behalf this message is being queued
2229  */
2230 static void
2231 add_to_pending_messages_for_peer (struct ConnectedPeer *cp,
2232                                   struct PendingMessage *pm,
2233                                   struct PendingRequest *pr)
2234 {
2235   struct PendingMessage *pos;
2236   struct PendingMessageList *pml;
2237   struct GNUNET_PeerIdentity pid;
2238
2239   GNUNET_assert (pm->next == NULL);
2240   GNUNET_assert (pm->pml == NULL);
2241   if (pr != NULL)
2242     {
2243       pml = GNUNET_malloc (sizeof (struct PendingMessageList));
2244       pml->req = pr;
2245       pml->target = cp;
2246       pml->pm = pm;
2247       pm->pml = pml;
2248       GNUNET_CONTAINER_DLL_insert (pr->pending_head,
2249                                    pr->pending_tail,
2250                                    pml);
2251     }
2252   pos = cp->pending_messages_head;
2253   while ( (pos != NULL) &&
2254           (pm->priority < pos->priority) )
2255     pos = pos->next;
2256   GNUNET_CONTAINER_DLL_insert_after (cp->pending_messages_head,
2257                                      cp->pending_messages_tail,
2258                                      pos,
2259                                      pm);
2260   cp->pending_requests++;
2261   if (cp->pending_requests > MAX_QUEUE_PER_PEER)
2262     {
2263       GNUNET_STATISTICS_update (stats,
2264                                 gettext_noop ("# P2P searches discarded (queue length bound)"),
2265                                 1,
2266                                 GNUNET_NO);
2267       destroy_pending_message (cp->pending_messages_tail, 0);
2268     }
2269   GNUNET_PEER_resolve (cp->pid, &pid);
2270   if (NULL != cp->cth)
2271     {
2272       GNUNET_CORE_notify_transmit_ready_cancel (cp->cth);
2273       cp->cth = NULL;
2274     }
2275   if (cp->delayed_transmission_request_task != GNUNET_SCHEDULER_NO_TASK)
2276     {
2277       GNUNET_SCHEDULER_cancel (cp->delayed_transmission_request_task);
2278       cp->delayed_transmission_request_task = GNUNET_SCHEDULER_NO_TASK;
2279     }
2280   /* need to schedule transmission */
2281   cp->last_transmission_request_start = GNUNET_TIME_absolute_get ();
2282   cp->cth = GNUNET_CORE_notify_transmit_ready (core,
2283                                                cp->pending_messages_head->priority,
2284                                                MAX_TRANSMIT_DELAY,
2285                                                &pid,
2286                                                cp->pending_messages_head->msize,
2287                                                &transmit_to_peer,
2288                                                cp);
2289   if (cp->cth == NULL)
2290     {
2291 #if DEBUG_FS
2292       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2293                   "Failed to schedule transmission with core!\n");
2294 #endif
2295       GNUNET_STATISTICS_update (stats,
2296                                 gettext_noop ("# CORE transmission failures"),
2297                                 1,
2298                                 GNUNET_NO);
2299     }
2300 }
2301
2302
2303 /**
2304  * Test if the DATABASE (GET) load on this peer is too high
2305  * to even consider processing the query at
2306  * all.
2307  *
2308  * @return GNUNET_YES if the load is too high to do anything (load high)
2309  *         GNUNET_NO to process normally (load normal)
2310  *         GNUNET_SYSERR to process for free (load low)
2311  */
2312 static int
2313 test_get_load_too_high (uint32_t priority)
2314 {
2315   double ld;
2316
2317   ld = GNUNET_LOAD_get_load (datastore_get_load);
2318   if (ld < 1)
2319     return GNUNET_SYSERR;
2320   if (ld <= priority)
2321     return GNUNET_NO;
2322   return GNUNET_YES;
2323 }
2324
2325
2326
2327
2328 /**
2329  * Test if the DATABASE (PUT) load on this peer is too high
2330  * to even consider processing the query at
2331  * all.
2332  *
2333  * @return GNUNET_YES if the load is too high to do anything (load high)
2334  *         GNUNET_NO to process normally (load normal or low)
2335  */
2336 static int
2337 test_put_load_too_high (uint32_t priority)
2338 {
2339   double ld;
2340
2341   if (GNUNET_LOAD_get_average (datastore_put_load) < 50)
2342     return GNUNET_NO; /* very fast */
2343   ld = GNUNET_LOAD_get_load (datastore_put_load);
2344   if (ld < 2.0 * (1 + priority))
2345     return GNUNET_NO;
2346   GNUNET_STATISTICS_update (stats,
2347                             gettext_noop ("# storage requests dropped due to high load"),
2348                             1,
2349                             GNUNET_NO);
2350   return GNUNET_YES;
2351 }
2352
2353
2354 /* ******************* Pending Request Refresh Task ******************** */
2355
2356
2357
2358 /**
2359  * We use a random delay to make the timing of requests less
2360  * predictable.  This function returns such a random delay.  We add a base
2361  * delay of MAX_CORK_DELAY (1s).
2362  *
2363  * FIXME: make schedule dependent on the specifics of the request?
2364  * Or bandwidth and number of connected peers and load?
2365  *
2366  * @return random delay to use for some request, between 1s and 1000+TTL_DECREMENT ms
2367  */
2368 static struct GNUNET_TIME_Relative
2369 get_processing_delay ()
2370 {
2371   return
2372     GNUNET_TIME_relative_add (GNUNET_CONSTANTS_MAX_CORK_DELAY,
2373                               GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
2374                                                              GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2375                                                                                        TTL_DECREMENT)));
2376 }
2377
2378
2379 /**
2380  * We're processing a GET request from another peer and have decided
2381  * to forward it to other peers.  This function is called periodically
2382  * and should forward the request to other peers until we have all
2383  * possible replies.  If we have transmitted the *only* reply to
2384  * the initiator we should destroy the pending request.  If we have
2385  * many replies in the queue to the initiator, we should delay sending
2386  * out more queries until the reply queue has shrunk some.
2387  *
2388  * @param cls our "struct ProcessGetContext *"
2389  * @param tc unused
2390  */
2391 static void
2392 forward_request_task (void *cls,
2393                       const struct GNUNET_SCHEDULER_TaskContext *tc);
2394
2395
2396 /**
2397  * Function called after we either failed or succeeded
2398  * at transmitting a query to a peer.
2399  *
2400  * @param cls the requests "struct PendingRequest*"
2401  * @param tpid ID of receiving peer, 0 on transmission error
2402  */
2403 static void
2404 transmit_query_continuation (void *cls,
2405                              GNUNET_PEER_Id tpid)
2406 {
2407   struct PendingRequest *pr = cls;
2408   unsigned int i;
2409
2410   GNUNET_STATISTICS_update (stats,
2411                             gettext_noop ("# queries scheduled for forwarding"),
2412                             -1,
2413                             GNUNET_NO);
2414   if (tpid == 0)
2415     {
2416 #if DEBUG_FS
2417       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2418                   "Transmission of request failed, will try again later.\n");
2419 #endif
2420       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
2421         pr->task = GNUNET_SCHEDULER_add_delayed (get_processing_delay (),
2422                                                  &forward_request_task,
2423                                                  pr);
2424       return;
2425     }
2426 #if DEBUG_FS
2427   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2428               "Transmitted query `%s'\n",
2429               GNUNET_h2s (&pr->query));
2430 #endif
2431   GNUNET_STATISTICS_update (stats,
2432                             gettext_noop ("# queries forwarded"),
2433                             1,
2434                             GNUNET_NO);
2435   for (i=0;i<pr->used_targets_off;i++)
2436     if (pr->used_targets[i].pid == tpid)
2437       break; /* found match! */
2438   if (i == pr->used_targets_off)
2439     {
2440       /* need to create new entry */
2441       if (pr->used_targets_off == pr->used_targets_size)
2442         GNUNET_array_grow (pr->used_targets,
2443                            pr->used_targets_size,
2444                            pr->used_targets_size * 2 + 2);
2445       GNUNET_PEER_change_rc (tpid, 1);
2446       pr->used_targets[pr->used_targets_off].pid = tpid;
2447       pr->used_targets[pr->used_targets_off].num_requests = 0;
2448       i = pr->used_targets_off++;
2449     }
2450   pr->used_targets[i].last_request_time = GNUNET_TIME_absolute_get ();
2451   pr->used_targets[i].num_requests++;
2452   if (pr->task == GNUNET_SCHEDULER_NO_TASK)
2453     pr->task = GNUNET_SCHEDULER_add_delayed (get_processing_delay (),
2454                                              &forward_request_task,
2455                                              pr);
2456 }
2457
2458
2459 /**
2460  * How many bytes should a bloomfilter be if we have already seen
2461  * entry_count responses?  Note that BLOOMFILTER_K gives us the number
2462  * of bits set per entry.  Furthermore, we should not re-size the
2463  * filter too often (to keep it cheap).
2464  *
2465  * Since other peers will also add entries but not resize the filter,
2466  * we should generally pick a slightly larger size than what the
2467  * strict math would suggest.
2468  *
2469  * @return must be a power of two and smaller or equal to 2^15.
2470  */
2471 static size_t
2472 compute_bloomfilter_size (unsigned int entry_count)
2473 {
2474   size_t size;
2475   unsigned int ideal = (entry_count * BLOOMFILTER_K) / 4;
2476   uint16_t max = 1 << 15;
2477
2478   if (entry_count > max)
2479     return max;
2480   size = 8;
2481   while ((size < max) && (size < ideal))
2482     size *= 2;
2483   if (size > max)
2484     return max;
2485   return size;
2486 }
2487
2488
2489 /**
2490  * Recalculate our bloom filter for filtering replies.  This function
2491  * will create a new bloom filter from scratch, so it should only be
2492  * called if we have no bloomfilter at all (and hence can create a
2493  * fresh one of minimal size without problems) OR if our peer is the
2494  * initiator (in which case we may resize to larger than mimimum size).
2495  *
2496  * @param pr request for which the BF is to be recomputed
2497  */
2498 static void
2499 refresh_bloomfilter (struct PendingRequest *pr)
2500 {
2501   unsigned int i;
2502   size_t nsize;
2503   GNUNET_HashCode mhash;
2504
2505   nsize = compute_bloomfilter_size (pr->replies_seen_off);
2506   if (nsize == pr->bf_size)
2507     return; /* size not changed */
2508   if (pr->bf != NULL)
2509     GNUNET_CONTAINER_bloomfilter_free (pr->bf);
2510   pr->bf_size = nsize;
2511   pr->mingle = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, -1);
2512   pr->bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
2513                                               pr->bf_size,
2514                                               BLOOMFILTER_K);
2515   for (i=0;i<pr->replies_seen_off;i++)
2516     {
2517       GNUNET_BLOCK_mingle_hash (&pr->replies_seen[i],
2518                                 pr->mingle,
2519                                 &mhash);
2520       GNUNET_CONTAINER_bloomfilter_add (pr->bf, &mhash);
2521     }
2522 }
2523
2524
2525 /**
2526  * Function called after we've tried to reserve a certain amount of
2527  * bandwidth for a reply.  Check if we succeeded and if so send our
2528  * query.
2529  *
2530  * @param cls the requests "struct PendingRequest*"
2531  * @param peer identifies the peer
2532  * @param bpm_in set to the current bandwidth limit (receiving) for this peer
2533  * @param bpm_out set to the current bandwidth limit (sending) for this peer
2534  * @param amount set to the amount that was actually reserved or unreserved
2535  * @param preference current traffic preference for the given peer
2536  */
2537 static void
2538 target_reservation_cb (void *cls,
2539                        const struct
2540                        GNUNET_PeerIdentity * peer,
2541                        struct GNUNET_BANDWIDTH_Value32NBO bpm_in,
2542                        struct GNUNET_BANDWIDTH_Value32NBO bpm_out,
2543                        int amount,
2544                        uint64_t preference)
2545 {
2546   struct PendingRequest *pr = cls;
2547   struct ConnectedPeer *cp;
2548   struct PendingMessage *pm;
2549   struct GetMessage *gm;
2550   GNUNET_HashCode *ext;
2551   char *bfdata;
2552   size_t msize;
2553   unsigned int k;
2554   int no_route;
2555   uint32_t bm;
2556   unsigned int i;
2557
2558   pr->irc = NULL;
2559   if (peer == NULL)
2560     {
2561       /* error in communication with core, try again later */
2562       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
2563         pr->task = GNUNET_SCHEDULER_add_delayed (get_processing_delay (),
2564                                                  &forward_request_task,
2565                                                  pr);
2566       return;
2567     }
2568   /* (3) transmit, update ttl/priority */
2569   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
2570                                           &peer->hashPubKey);
2571   if (cp == NULL)
2572     {
2573       /* Peer must have just left */
2574 #if DEBUG_FS
2575       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2576                   "Selected peer disconnected!\n");
2577 #endif
2578       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
2579         pr->task = GNUNET_SCHEDULER_add_delayed (get_processing_delay (),
2580                                                  &forward_request_task,
2581                                                  pr);
2582       return;
2583     }
2584   no_route = GNUNET_NO;
2585   if (amount == 0)
2586     {
2587       if (pr->cp == NULL)
2588         {
2589 #if DEBUG_FS > 1
2590           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2591                       "Failed to reserve bandwidth for reply (got %d/%u bytes only)!\n",
2592                       amount,
2593                       DBLOCK_SIZE);
2594 #endif
2595           GNUNET_STATISTICS_update (stats,
2596                                     gettext_noop ("# reply bandwidth reservation requests failed"),
2597                                     1,
2598                                     GNUNET_NO);
2599           if (pr->task == GNUNET_SCHEDULER_NO_TASK)
2600             pr->task = GNUNET_SCHEDULER_add_delayed (get_processing_delay (),
2601                                                      &forward_request_task,
2602                                                      pr);
2603           return;  /* this target round failed */
2604         }
2605       no_route = GNUNET_YES;
2606     }
2607
2608   GNUNET_STATISTICS_update (stats,
2609                             gettext_noop ("# queries scheduled for forwarding"),
2610                             1,
2611                             GNUNET_NO);
2612   for (i=0;i<pr->used_targets_off;i++)
2613     if (pr->used_targets[i].pid == cp->pid)
2614       {
2615         GNUNET_STATISTICS_update (stats,
2616                                   gettext_noop ("# queries retransmitted to same target"),
2617                                   1,
2618                                   GNUNET_NO);
2619         break;
2620       }
2621
2622   /* build message and insert message into priority queue */
2623 #if DEBUG_FS
2624   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2625               "Forwarding request `%s' to `%4s'!\n",
2626               GNUNET_h2s (&pr->query),
2627               GNUNET_i2s (peer));
2628 #endif
2629   k = 0;
2630   bm = 0;
2631   if (GNUNET_YES == no_route)
2632     {
2633       bm |= GET_MESSAGE_BIT_RETURN_TO;
2634       k++;
2635     }
2636   if (pr->namespace != NULL)
2637     {
2638       bm |= GET_MESSAGE_BIT_SKS_NAMESPACE;
2639       k++;
2640     }
2641   if (pr->target_pid != 0)
2642     {
2643       bm |= GET_MESSAGE_BIT_TRANSMIT_TO;
2644       k++;
2645     }
2646   msize = sizeof (struct GetMessage) + pr->bf_size + k * sizeof(GNUNET_HashCode);
2647   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
2648   pm = GNUNET_malloc (sizeof (struct PendingMessage) + msize);
2649   pm->msize = msize;
2650   gm = (struct GetMessage*) &pm[1];
2651   gm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_GET);
2652   gm->header.size = htons (msize);
2653   gm->type = htonl (pr->type);
2654   pr->remaining_priority /= 2;
2655   gm->priority = htonl (pr->remaining_priority);
2656   gm->ttl = htonl (pr->ttl);
2657   gm->filter_mutator = htonl(pr->mingle);
2658   gm->hash_bitmap = htonl (bm);
2659   gm->query = pr->query;
2660   ext = (GNUNET_HashCode*) &gm[1];
2661   k = 0;
2662   if (GNUNET_YES == no_route)
2663     GNUNET_PEER_resolve (pr->cp->pid, (struct GNUNET_PeerIdentity*) &ext[k++]);
2664   if (pr->namespace != NULL)
2665     memcpy (&ext[k++], pr->namespace, sizeof (GNUNET_HashCode));
2666   if (pr->target_pid != 0)
2667     GNUNET_PEER_resolve (pr->target_pid, (struct GNUNET_PeerIdentity*) &ext[k++]);
2668   bfdata = (char *) &ext[k];
2669   if (pr->bf != NULL)
2670     GNUNET_CONTAINER_bloomfilter_get_raw_data (pr->bf,
2671                                                bfdata,
2672                                                pr->bf_size);
2673   pm->cont = &transmit_query_continuation;
2674   pm->cont_cls = pr;
2675   cp->last_request_times[(cp->last_request_times_off++) % MAX_QUEUE_PER_PEER] = GNUNET_TIME_absolute_get ();
2676   add_to_pending_messages_for_peer (cp, pm, pr);
2677 }
2678
2679
2680 /**
2681  * Closure used for "target_peer_select_cb".
2682  */
2683 struct PeerSelectionContext
2684 {
2685   /**
2686    * The request for which we are selecting
2687    * peers.
2688    */
2689   struct PendingRequest *pr;
2690
2691   /**
2692    * Current "prime" target.
2693    */
2694   struct GNUNET_PeerIdentity target;
2695
2696   /**
2697    * How much do we like this target?
2698    */
2699   double target_score;
2700
2701 };
2702
2703
2704 /**
2705  * Function called for each connected peer to determine
2706  * which one(s) would make good targets for forwarding.
2707  *
2708  * @param cls closure (struct PeerSelectionContext)
2709  * @param key current key code (peer identity)
2710  * @param value value in the hash map (struct ConnectedPeer)
2711  * @return GNUNET_YES if we should continue to
2712  *         iterate,
2713  *         GNUNET_NO if not.
2714  */
2715 static int
2716 target_peer_select_cb (void *cls,
2717                        const GNUNET_HashCode * key,
2718                        void *value)
2719 {
2720   struct PeerSelectionContext *psc = cls;
2721   struct ConnectedPeer *cp = value;
2722   struct PendingRequest *pr = psc->pr;
2723   struct GNUNET_TIME_Relative delay;
2724   double score;
2725   unsigned int i;
2726   unsigned int pc;
2727
2728   /* 1) check that this peer is not the initiator */
2729   if (cp == pr->cp)
2730     {
2731 #if DEBUG_FS
2732       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2733                   "Skipping initiator in forwarding selection\n");
2734 #endif
2735       return GNUNET_YES; /* skip */
2736     }
2737
2738   /* 2) check if we have already (recently) forwarded to this peer */
2739   /* 2a) this particular request */
2740   pc = 0;
2741   for (i=0;i<pr->used_targets_off;i++)
2742     if (pr->used_targets[i].pid == cp->pid)
2743       {
2744         pc = pr->used_targets[i].num_requests;
2745         GNUNET_assert (pc > 0);
2746         if (0 != GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2747                                            RETRY_PROBABILITY_INV * pc))
2748           {
2749 #if DEBUG_FS
2750             GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2751                         "NOT re-trying query that was previously transmitted %u times\n",
2752                         (unsigned int) pc);
2753 #endif
2754             return GNUNET_YES; /* skip */
2755           }
2756         break;
2757       }
2758 #if DEBUG_FS
2759   if (0 < pc)
2760     {
2761       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
2762                   "Re-trying query that was previously transmitted %u times to this peer\n",
2763                   (unsigned int) pc);
2764     }
2765 #endif
2766   /* 2b) many other requests to this peer */
2767   delay = GNUNET_TIME_absolute_get_duration (cp->last_request_times[cp->last_request_times_off % MAX_QUEUE_PER_PEER]);
2768   if (delay.rel_value <= cp->avg_delay.rel_value)
2769     {
2770 #if DEBUG_FS
2771       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2772                   "NOT sending query since we send %u others to this peer in the last %llums\n",
2773                   MAX_QUEUE_PER_PEER,
2774                   cp->avg_delay.rel_value);
2775 #endif
2776       return GNUNET_YES; /* skip */
2777     }
2778
2779   /* 3) calculate how much we'd like to forward to this peer,
2780      starting with a random value that is strong enough
2781      to at least give any peer a chance sometimes
2782      (compared to the other factors that come later) */
2783   /* 3a) count successful (recent) routes from cp for same source */
2784   if (pr->cp != NULL)
2785     {
2786       score = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2787                                         P2P_SUCCESS_LIST_SIZE);
2788       for (i=0;i<P2P_SUCCESS_LIST_SIZE;i++)
2789         if (cp->last_p2p_replies[i] == pr->cp->pid)
2790           score += 1.0; /* likely successful based on hot path */
2791     }
2792   else
2793     {
2794       score = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2795                                         CS2P_SUCCESS_LIST_SIZE);
2796       for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
2797         if (cp->last_client_replies[i] == pr->client_request_list->client_list->client)
2798           score += 1.0; /* likely successful based on hot path */
2799     }
2800   /* 3b) include latency */
2801   if (cp->avg_delay.rel_value < 4 * TTL_DECREMENT)
2802     score += 1.0; /* likely fast based on latency */
2803   /* 3c) include priorities */
2804   if (cp->avg_priority <= pr->remaining_priority / 2.0)
2805     score += 1.0; /* likely successful based on priorities */
2806   /* 3d) penalize for queue size */
2807   score -= (2.0 * cp->pending_requests / (double) MAX_QUEUE_PER_PEER);
2808   /* 3e) include peer proximity */
2809   score -= (2.0 * (GNUNET_CRYPTO_hash_distance_u32 (key,
2810                                                     &pr->query)) / (double) UINT32_MAX);
2811   /* 4) super-bonus for being the known target */
2812   if (pr->target_pid == cp->pid)
2813     score += 100.0;
2814   /* store best-fit in closure */
2815 #if DEBUG_FS
2816   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2817               "Peer `%s' gets score %f for forwarding query, max is %f\n",
2818               GNUNET_h2s (key),
2819               score,
2820               psc->target_score);
2821 #endif
2822   score++; /* avoid zero */
2823   if (score > psc->target_score)
2824     {
2825       psc->target_score = score;
2826       psc->target.hashPubKey = *key;
2827     }
2828   return GNUNET_YES;
2829 }
2830
2831
2832 /**
2833  * The priority level imposes a bound on the maximum
2834  * value for the ttl that can be requested.
2835  *
2836  * @param ttl_in requested ttl
2837  * @param prio given priority
2838  * @return ttl_in if ttl_in is below the limit,
2839  *         otherwise the ttl-limit for the given priority
2840  */
2841 static int32_t
2842 bound_ttl (int32_t ttl_in, uint32_t prio)
2843 {
2844   unsigned long long allowed;
2845
2846   if (ttl_in <= 0)
2847     return ttl_in;
2848   allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000;
2849   if (ttl_in > allowed)
2850     {
2851       if (allowed >= (1 << 30))
2852         return 1 << 30;
2853       return allowed;
2854     }
2855   return ttl_in;
2856 }
2857
2858
2859 /**
2860  * Iterator called on each result obtained for a DHT
2861  * operation that expects a reply
2862  *
2863  * @param cls closure
2864  * @param exp when will this value expire
2865  * @param key key of the result
2866  * @param get_path NULL-terminated array of pointers
2867  *                 to the peers on reverse GET path (or NULL if not recorded)
2868  * @param put_path NULL-terminated array of pointers
2869  *                 to the peers on the PUT path (or NULL if not recorded)
2870  * @param type type of the result
2871  * @param size number of bytes in data
2872  * @param data pointer to the result data
2873  */
2874 static void
2875 process_dht_reply (void *cls,
2876                    struct GNUNET_TIME_Absolute exp,
2877                    const GNUNET_HashCode * key,
2878                    const struct GNUNET_PeerIdentity * const *get_path,
2879                    const struct GNUNET_PeerIdentity * const *put_path,
2880                    enum GNUNET_BLOCK_Type type,
2881                    size_t size,
2882                    const void *data);
2883
2884
2885 /**
2886  * We're processing a GET request and have decided
2887  * to forward it to other peers.  This function is called periodically
2888  * and should forward the request to other peers until we have all
2889  * possible replies.  If we have transmitted the *only* reply to
2890  * the initiator we should destroy the pending request.  If we have
2891  * many replies in the queue to the initiator, we should delay sending
2892  * out more queries until the reply queue has shrunk some.
2893  *
2894  * @param cls our "struct ProcessGetContext *"
2895  * @param tc unused
2896  */
2897 static void
2898 forward_request_task (void *cls,
2899                      const struct GNUNET_SCHEDULER_TaskContext *tc)
2900 {
2901   struct PendingRequest *pr = cls;
2902   struct PeerSelectionContext psc;
2903   struct ConnectedPeer *cp;
2904   struct GNUNET_TIME_Relative delay;
2905
2906   pr->task = GNUNET_SCHEDULER_NO_TASK;
2907   if (pr->irc != NULL)
2908     {
2909 #if DEBUG_FS
2910       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2911                   "Forwarding of query `%s' not attempted due to pending local lookup!\n",
2912                   GNUNET_h2s (&pr->query));
2913 #endif
2914       return; /* already pending */
2915     }
2916   if (GNUNET_YES == pr->local_only)
2917     return; /* configured to not do P2P search */
2918   /* (0) try DHT */
2919   if ( (0 == pr->anonymity_level) &&
2920        (GNUNET_YES != pr->forward_only) &&
2921        (pr->type != GNUNET_BLOCK_TYPE_FS_DBLOCK) &&
2922        (pr->type != GNUNET_BLOCK_TYPE_FS_IBLOCK) )
2923     {
2924       pr->dht_get = GNUNET_DHT_get_start (dht_handle,
2925                                           GNUNET_TIME_UNIT_FOREVER_REL,
2926                                           pr->type,
2927                                           &pr->query,
2928                                           DEFAULT_GET_REPLICATION,
2929                                           GNUNET_DHT_RO_NONE,
2930                                           pr->bf,
2931                                           pr->mingle,
2932                                           pr->namespace,
2933                                           (pr->namespace != NULL) ? sizeof (GNUNET_HashCode) : 0,
2934                                           &process_dht_reply,
2935                                           pr);
2936     }
2937   /* (1) select target */
2938   psc.pr = pr;
2939   psc.target_score = -DBL_MAX;
2940   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
2941                                          &target_peer_select_cb,
2942                                          &psc);
2943   if (psc.target_score == -DBL_MAX)
2944     {
2945       delay = get_processing_delay ();
2946 #if DEBUG_FS
2947       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2948                   "No peer selected for forwarding of query `%s', will try again in %llu ms!\n",
2949                   GNUNET_h2s (&pr->query),
2950                   delay.rel_value);
2951 #endif
2952       pr->task = GNUNET_SCHEDULER_add_delayed (delay,
2953                                                &forward_request_task,
2954                                                pr);
2955       return; /* nobody selected */
2956     }
2957   /* (3) update TTL/priority */
2958   if (pr->client_request_list != NULL)
2959     {
2960       /* FIXME: use better algorithm!? */
2961       if (0 == GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2962                                          4))
2963         pr->priority++;
2964       /* bound priority we use by priorities we see from other peers
2965          rounded up (must round up so that we can see non-zero
2966          priorities, but round up as little as possible to make it
2967          plausible that we forwarded another peers request) */
2968       if (pr->priority > current_priorities + 1.0)
2969         pr->priority = (uint32_t) current_priorities + 1.0;
2970       pr->ttl = bound_ttl (pr->ttl + TTL_DECREMENT * 2,
2971                            pr->priority);
2972 #if DEBUG_FS
2973       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2974                   "Trying query `%s' with priority %u and TTL %d.\n",
2975                   GNUNET_h2s (&pr->query),
2976                   pr->priority,
2977                   pr->ttl);
2978 #endif
2979     }
2980
2981   /* (3) reserve reply bandwidth */
2982   if (GNUNET_NO == pr->forward_only)
2983     {
2984       cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
2985                                               &psc.target.hashPubKey);
2986       GNUNET_assert (NULL != cp);
2987       pr->irc = GNUNET_CORE_peer_change_preference (cfg,
2988                                                     &psc.target,
2989                                                     GNUNET_CONSTANTS_SERVICE_TIMEOUT,
2990                                                     GNUNET_BANDWIDTH_value_init (UINT32_MAX),
2991                                                     DBLOCK_SIZE * 2,
2992                                                     cp->inc_preference,
2993                                                     &target_reservation_cb,
2994                                                     pr);
2995       cp->inc_preference = 0;
2996     }
2997   else
2998     {
2999       /* force forwarding */
3000       static struct GNUNET_BANDWIDTH_Value32NBO zerobw;
3001       target_reservation_cb (pr, &psc.target,
3002                              zerobw, zerobw, 0, 0.0);
3003     }
3004 }
3005
3006
3007 /* **************************** P2P PUT Handling ************************ */
3008
3009
3010 /**
3011  * Function called after we either failed or succeeded
3012  * at transmitting a reply to a peer.
3013  *
3014  * @param cls the requests "struct PendingRequest*"
3015  * @param tpid ID of receiving peer, 0 on transmission error
3016  */
3017 static void
3018 transmit_reply_continuation (void *cls,
3019                              GNUNET_PEER_Id tpid)
3020 {
3021   struct PendingRequest *pr = cls;
3022
3023   switch (pr->type)
3024     {
3025     case GNUNET_BLOCK_TYPE_FS_DBLOCK:
3026     case GNUNET_BLOCK_TYPE_FS_IBLOCK:
3027       /* only one reply expected, done with the request! */
3028       destroy_pending_request (pr);
3029       break;
3030     case GNUNET_BLOCK_TYPE_ANY:
3031     case GNUNET_BLOCK_TYPE_FS_KBLOCK:
3032     case GNUNET_BLOCK_TYPE_FS_SBLOCK:
3033       break;
3034     default:
3035       GNUNET_break (0);
3036       break;
3037     }
3038 }
3039
3040
3041 /**
3042  * Transmit the given message by copying it to the target buffer
3043  * "buf".  "buf" will be NULL and "size" zero if the socket was closed
3044  * for writing in the meantime.  In that case, do nothing
3045  * (the disconnect or shutdown handler will take care of the rest).
3046  * If we were able to transmit messages and there are still more
3047  * pending, ask core again for further calls to this function.
3048  *
3049  * @param cls closure, pointer to the 'struct ClientList*'
3050  * @param size number of bytes available in buf
3051  * @param buf where the callee should write the message
3052  * @return number of bytes written to buf
3053  */
3054 static size_t
3055 transmit_to_client (void *cls,
3056                   size_t size, void *buf)
3057 {
3058   struct ClientList *cl = cls;
3059   char *cbuf = buf;
3060   struct ClientResponseMessage *creply;
3061   size_t msize;
3062
3063   cl->th = NULL;
3064   if (NULL == buf)
3065     {
3066 #if DEBUG_FS
3067       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3068                   "Not sending reply, client communication problem.\n");
3069 #endif
3070       return 0;
3071     }
3072   msize = 0;
3073   while ( (NULL != (creply = cl->res_head) ) &&
3074           (creply->msize <= size) )
3075     {
3076       memcpy (&cbuf[msize], &creply[1], creply->msize);
3077       msize += creply->msize;
3078       size -= creply->msize;
3079       GNUNET_CONTAINER_DLL_remove (cl->res_head,
3080                                    cl->res_tail,
3081                                    creply);
3082       GNUNET_free (creply);
3083     }
3084   if (NULL != creply)
3085     cl->th = GNUNET_SERVER_notify_transmit_ready (cl->client,
3086                                                   creply->msize,
3087                                                   GNUNET_TIME_UNIT_FOREVER_REL,
3088                                                   &transmit_to_client,
3089                                                   cl);
3090 #if DEBUG_FS
3091   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3092               "Transmitted %u bytes to client\n",
3093               (unsigned int) msize);
3094 #endif
3095   return msize;
3096 }
3097
3098
3099 /**
3100  * Closure for "process_reply" function.
3101  */
3102 struct ProcessReplyClosure
3103 {
3104   /**
3105    * The data for the reply.
3106    */
3107   const void *data;
3108
3109   /**
3110    * Who gave us this reply? NULL for local host (or DHT)
3111    */
3112   struct ConnectedPeer *sender;
3113
3114   /**
3115    * When the reply expires.
3116    */
3117   struct GNUNET_TIME_Absolute expiration;
3118
3119   /**
3120    * Size of data.
3121    */
3122   size_t size;
3123
3124   /**
3125    * Type of the block.
3126    */
3127   enum GNUNET_BLOCK_Type type;
3128
3129   /**
3130    * How much was this reply worth to us?
3131    */
3132   uint32_t priority;
3133
3134   /**
3135    * Evaluation result (returned).
3136    */
3137   enum GNUNET_BLOCK_EvaluationResult eval;
3138
3139   /**
3140    * Did we finish processing the associated request?
3141    */
3142   int finished;
3143
3144   /**
3145    * Did we find a matching request?
3146    */
3147   int request_found;
3148 };
3149
3150
3151 /**
3152  * We have received a reply; handle it!
3153  *
3154  * @param cls response (struct ProcessReplyClosure)
3155  * @param key our query
3156  * @param value value in the hash map (info about the query)
3157  * @return GNUNET_YES (we should continue to iterate)
3158  */
3159 static int
3160 process_reply (void *cls,
3161                const GNUNET_HashCode * key,
3162                void *value)
3163 {
3164   struct ProcessReplyClosure *prq = cls;
3165   struct PendingRequest *pr = value;
3166   struct PendingMessage *reply;
3167   struct ClientResponseMessage *creply;
3168   struct ClientList *cl;
3169   struct PutMessage *pm;
3170   struct ConnectedPeer *cp;
3171   struct GNUNET_TIME_Relative cur_delay;
3172 #if SUPPORT_DELAYS
3173 struct GNUNET_TIME_Relative art_delay;
3174 #endif
3175   size_t msize;
3176   unsigned int i;
3177
3178 #if DEBUG_FS
3179   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3180               "Matched result (type %u) for query `%s' with pending request\n",
3181               (unsigned int) prq->type,
3182               GNUNET_h2s (key));
3183 #endif
3184   GNUNET_STATISTICS_update (stats,
3185                             gettext_noop ("# replies received and matched"),
3186                             1,
3187                             GNUNET_NO);
3188   if (prq->sender != NULL)
3189     {
3190       for (i=0;i<pr->used_targets_off;i++)
3191         if (pr->used_targets[i].pid == prq->sender->pid)
3192           break;
3193       if (i < pr->used_targets_off)
3194         {
3195           cur_delay = GNUNET_TIME_absolute_get_duration (pr->used_targets[i].last_request_time);
3196           prq->sender->avg_delay.rel_value
3197             = (prq->sender->avg_delay.rel_value *
3198                (RUNAVG_DELAY_N - 1) + cur_delay.rel_value) / RUNAVG_DELAY_N;
3199           prq->sender->avg_priority
3200             = (prq->sender->avg_priority *
3201                (RUNAVG_DELAY_N - 1) + pr->priority) / (double) RUNAVG_DELAY_N;
3202         }
3203       if (pr->cp != NULL)
3204         {
3205           GNUNET_PEER_change_rc (prq->sender->last_p2p_replies
3206                                  [prq->sender->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE],
3207                                  -1);
3208           GNUNET_PEER_change_rc (pr->cp->pid, 1);
3209           prq->sender->last_p2p_replies
3210             [(prq->sender->last_p2p_replies_woff++) % P2P_SUCCESS_LIST_SIZE]
3211             = pr->cp->pid;
3212         }
3213       else
3214         {
3215           if (NULL != prq->sender->last_client_replies
3216               [(prq->sender->last_client_replies_woff) % CS2P_SUCCESS_LIST_SIZE])
3217             GNUNET_SERVER_client_drop (prq->sender->last_client_replies
3218                                        [(prq->sender->last_client_replies_woff) % CS2P_SUCCESS_LIST_SIZE]);
3219           prq->sender->last_client_replies
3220             [(prq->sender->last_client_replies_woff++) % CS2P_SUCCESS_LIST_SIZE]
3221             = pr->client_request_list->client_list->client;
3222           GNUNET_SERVER_client_keep (pr->client_request_list->client_list->client);
3223         }
3224     }
3225   prq->eval = GNUNET_BLOCK_evaluate (block_ctx,
3226                                      prq->type,
3227                                      key,
3228                                      &pr->bf,
3229                                      pr->mingle,
3230                                      pr->namespace, (pr->namespace != NULL) ? sizeof (GNUNET_HashCode) : 0,
3231                                      prq->data,
3232                                      prq->size);
3233   switch (prq->eval)
3234     {
3235     case GNUNET_BLOCK_EVALUATION_OK_MORE:
3236       break;
3237     case GNUNET_BLOCK_EVALUATION_OK_LAST:
3238       while (NULL != pr->pending_head)
3239         destroy_pending_message_list_entry (pr->pending_head);
3240       if (pr->qe != NULL)
3241         {
3242           if (pr->client_request_list != NULL)
3243             GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
3244                                         GNUNET_YES);
3245           GNUNET_DATASTORE_cancel (pr->qe);
3246           pr->qe = NULL;
3247         }
3248       pr->do_remove = GNUNET_YES;
3249       if (pr->task != GNUNET_SCHEDULER_NO_TASK)
3250         {
3251           GNUNET_SCHEDULER_cancel (pr->task);
3252           pr->task = GNUNET_SCHEDULER_NO_TASK;
3253         }
3254       GNUNET_break (GNUNET_YES ==
3255                     GNUNET_CONTAINER_multihashmap_remove (query_request_map,
3256                                                           key,
3257                                                           pr));
3258       GNUNET_LOAD_update (rt_entry_lifetime,
3259                           GNUNET_TIME_absolute_get_duration (pr->start_time).rel_value);
3260       break;
3261     case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
3262       GNUNET_STATISTICS_update (stats,
3263                                 gettext_noop ("# duplicate replies discarded (bloomfilter)"),
3264                                 1,
3265                                 GNUNET_NO);
3266 #if DEBUG_FS
3267 /*      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3268                   "Duplicate response `%s', discarding.\n",
3269                   GNUNET_h2s (&mhash));*/
3270 #endif
3271       return GNUNET_YES; /* duplicate */
3272     case GNUNET_BLOCK_EVALUATION_RESULT_INVALID:
3273       return GNUNET_YES; /* wrong namespace */
3274     case GNUNET_BLOCK_EVALUATION_REQUEST_VALID:
3275       GNUNET_break (0);
3276       return GNUNET_YES;
3277     case GNUNET_BLOCK_EVALUATION_REQUEST_INVALID:
3278       GNUNET_break (0);
3279       return GNUNET_YES;
3280     case GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED:
3281       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
3282                   _("Unsupported block type %u\n"),
3283                   prq->type);
3284       return GNUNET_NO;
3285     }
3286   if (pr->client_request_list != NULL)
3287     {
3288       if (pr->replies_seen_size == pr->replies_seen_off)
3289         GNUNET_array_grow (pr->replies_seen,
3290                            pr->replies_seen_size,
3291                            pr->replies_seen_size * 2 + 4);
3292       GNUNET_CRYPTO_hash (prq->data,
3293                           prq->size,
3294                           &pr->replies_seen[pr->replies_seen_off++]);
3295       refresh_bloomfilter (pr);
3296     }
3297   if (NULL == prq->sender)
3298     {
3299 #if DEBUG_FS
3300       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3301                   "Found result for query `%s' in local datastore\n",
3302                   GNUNET_h2s (key));
3303 #endif
3304       GNUNET_STATISTICS_update (stats,
3305                                 gettext_noop ("# results found locally"),
3306                                 1,
3307                                 GNUNET_NO);
3308     }
3309   prq->priority += pr->remaining_priority;
3310   pr->remaining_priority = 0;
3311   pr->results_found++;
3312   prq->request_found = GNUNET_YES;
3313   if (NULL != pr->client_request_list)
3314     {
3315       GNUNET_STATISTICS_update (stats,
3316                                 gettext_noop ("# replies received for local clients"),
3317                                 1,
3318                                 GNUNET_NO);
3319       cl = pr->client_request_list->client_list;
3320       msize = sizeof (struct PutMessage) + prq->size;
3321       creply = GNUNET_malloc (msize + sizeof (struct ClientResponseMessage));
3322       creply->msize = msize;
3323       creply->client_list = cl;
3324       GNUNET_CONTAINER_DLL_insert_after (cl->res_head,
3325                                          cl->res_tail,
3326                                          cl->res_tail,
3327                                          creply);
3328       pm = (struct PutMessage*) &creply[1];
3329       pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
3330       pm->header.size = htons (msize);
3331       pm->type = htonl (prq->type);
3332       pm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
3333       memcpy (&pm[1], prq->data, prq->size);
3334       if (NULL == cl->th)
3335         {
3336 #if DEBUG_FS
3337           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3338                       "Transmitting result for query `%s' to client\n",
3339                       GNUNET_h2s (key));
3340 #endif
3341           cl->th = GNUNET_SERVER_notify_transmit_ready (cl->client,
3342                                                         msize,
3343                                                         GNUNET_TIME_UNIT_FOREVER_REL,
3344                                                         &transmit_to_client,
3345                                                         cl);
3346         }
3347       GNUNET_break (cl->th != NULL);
3348       if (pr->do_remove)
3349         {
3350           prq->finished = GNUNET_YES;
3351           destroy_pending_request (pr);
3352         }
3353     }
3354   else
3355     {
3356       cp = pr->cp;
3357 #if DEBUG_FS
3358       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3359                   "Transmitting result for query `%s' to other peer (PID=%u)\n",
3360                   GNUNET_h2s (key),
3361                   (unsigned int) cp->pid);
3362 #endif
3363       GNUNET_STATISTICS_update (stats,
3364                                 gettext_noop ("# replies received for other peers"),
3365                                 1,
3366                                 GNUNET_NO);
3367       msize = sizeof (struct PutMessage) + prq->size;
3368       reply = GNUNET_malloc (msize + sizeof (struct PendingMessage));
3369       reply->cont = &transmit_reply_continuation;
3370       reply->cont_cls = pr;
3371 #if SUPPORT_DELAYS
3372       art_delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
3373                                                  GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
3374                                                                            TTL_DECREMENT));
3375       reply->delay_until
3376         = GNUNET_TIME_relative_to_absolute (art_delay);
3377       GNUNET_STATISTICS_update (stats,
3378                                 gettext_noop ("cummulative artificial delay introduced (ms)"),
3379                                 art_delay.abs_value,
3380                                 GNUNET_NO);
3381 #endif
3382       reply->msize = msize;
3383       reply->priority = UINT32_MAX; /* send replies first! */
3384       pm = (struct PutMessage*) &reply[1];
3385       pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
3386       pm->header.size = htons (msize);
3387       pm->type = htonl (prq->type);
3388       pm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
3389       memcpy (&pm[1], prq->data, prq->size);
3390       add_to_pending_messages_for_peer (cp, reply, pr);
3391     }
3392   return GNUNET_YES;
3393 }
3394
3395
3396 /**
3397  * Iterator called on each result obtained for a DHT
3398  * operation that expects a reply
3399  *
3400  * @param cls closure
3401  * @param exp when will this value expire
3402  * @param key key of the result
3403  * @param get_path NULL-terminated array of pointers
3404  *                 to the peers on reverse GET path (or NULL if not recorded)
3405  * @param put_path NULL-terminated array of pointers
3406  *                 to the peers on the PUT path (or NULL if not recorded)
3407  * @param type type of the result
3408  * @param size number of bytes in data
3409  * @param data pointer to the result data
3410  */
3411 static void
3412 process_dht_reply (void *cls,
3413                    struct GNUNET_TIME_Absolute exp,
3414                    const GNUNET_HashCode * key,
3415                    const struct GNUNET_PeerIdentity * const *get_path,
3416                    const struct GNUNET_PeerIdentity * const *put_path,
3417                    enum GNUNET_BLOCK_Type type,
3418                    size_t size,
3419                    const void *data)
3420 {
3421   struct PendingRequest *pr = cls;
3422   struct ProcessReplyClosure prq;
3423
3424   memset (&prq, 0, sizeof (prq));
3425   prq.data = data;
3426   prq.expiration = exp;
3427   prq.size = size;
3428   prq.type = type;
3429   process_reply (&prq, key, pr);
3430 }
3431
3432
3433
3434 /**
3435  * Continuation called to notify client about result of the
3436  * operation.
3437  *
3438  * @param cls closure
3439  * @param success GNUNET_SYSERR on failure
3440  * @param msg NULL on success, otherwise an error message
3441  */
3442 static void
3443 put_migration_continuation (void *cls,
3444                             int success,
3445                             const char *msg)
3446 {
3447   struct GNUNET_TIME_Absolute *start = cls;
3448   struct GNUNET_TIME_Relative delay;
3449
3450   delay = GNUNET_TIME_absolute_get_duration (*start);
3451   GNUNET_free (start);
3452   GNUNET_LOAD_update (datastore_put_load,
3453                       delay.rel_value);
3454   if (GNUNET_OK == success)
3455     return;
3456   GNUNET_STATISTICS_update (stats,
3457                             gettext_noop ("# datastore 'put' failures"),
3458                             1,
3459                             GNUNET_NO);
3460 }
3461
3462
3463 /**
3464  * Handle P2P "PUT" message.
3465  *
3466  * @param cls closure, always NULL
3467  * @param other the other peer involved (sender or receiver, NULL
3468  *        for loopback messages where we are both sender and receiver)
3469  * @param message the actual message
3470  * @param latency reported latency of the connection with 'other'
3471  * @param distance reported distance (DV) to 'other'
3472  * @return GNUNET_OK to keep the connection open,
3473  *         GNUNET_SYSERR to close it (signal serious error)
3474  */
3475 static int
3476 handle_p2p_put (void *cls,
3477                 const struct GNUNET_PeerIdentity *other,
3478                 const struct GNUNET_MessageHeader *message,
3479                 struct GNUNET_TIME_Relative latency,
3480                 uint32_t distance)
3481 {
3482   const struct PutMessage *put;
3483   uint16_t msize;
3484   size_t dsize;
3485   enum GNUNET_BLOCK_Type type;
3486   struct GNUNET_TIME_Absolute expiration;
3487   GNUNET_HashCode query;
3488   struct ProcessReplyClosure prq;
3489   struct GNUNET_TIME_Absolute *start;
3490   struct GNUNET_TIME_Relative block_time;
3491   double putl;
3492   struct ConnectedPeer *cp;
3493   struct PendingMessage *pm;
3494   struct MigrationStopMessage *msm;
3495
3496   msize = ntohs (message->size);
3497   if (msize < sizeof (struct PutMessage))
3498     {
3499       GNUNET_break_op(0);
3500       return GNUNET_SYSERR;
3501     }
3502   put = (const struct PutMessage*) message;
3503   dsize = msize - sizeof (struct PutMessage);
3504   type = ntohl (put->type);
3505   expiration = GNUNET_TIME_absolute_ntoh (put->expiration);
3506
3507   if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
3508     return GNUNET_SYSERR;
3509   if (GNUNET_OK !=
3510       GNUNET_BLOCK_get_key (block_ctx,
3511                             type,
3512                             &put[1],
3513                             dsize,
3514                             &query))
3515     {
3516       GNUNET_break_op (0);
3517       return GNUNET_SYSERR;
3518     }
3519 #if DEBUG_FS
3520   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3521               "Received result for query `%s' from peer `%4s'\n",
3522               GNUNET_h2s (&query),
3523               GNUNET_i2s (other));
3524 #endif
3525   GNUNET_STATISTICS_update (stats,
3526                             gettext_noop ("# replies received (overall)"),
3527                             1,
3528                             GNUNET_NO);
3529   /* now, lookup 'query' */
3530   prq.data = (const void*) &put[1];
3531   if (other != NULL)
3532     prq.sender = GNUNET_CONTAINER_multihashmap_get (connected_peers,
3533                                                     &other->hashPubKey);
3534   else
3535     prq.sender = NULL;
3536   prq.size = dsize;
3537   prq.type = type;
3538   prq.expiration = expiration;
3539   prq.priority = 0;
3540   prq.finished = GNUNET_NO;
3541   prq.request_found = GNUNET_NO;
3542   GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
3543                                               &query,
3544                                               &process_reply,
3545                                               &prq);
3546   if (prq.sender != NULL)
3547     {
3548       prq.sender->inc_preference += CONTENT_BANDWIDTH_VALUE + 1000 * prq.priority;
3549       change_host_trust (prq.sender, prq.priority);
3550     }
3551   if ( (GNUNET_YES == active_migration) &&
3552        (GNUNET_NO == test_put_load_too_high (prq.priority)) )
3553     {
3554 #if DEBUG_FS
3555       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3556                   "Replicating result for query `%s' with priority %u\n",
3557                   GNUNET_h2s (&query),
3558                   prq.priority);
3559 #endif
3560       start = GNUNET_malloc (sizeof (struct GNUNET_TIME_Absolute));
3561       *start = GNUNET_TIME_absolute_get ();
3562       GNUNET_DATASTORE_put (dsh,
3563                             0, &query, dsize, &put[1],
3564                             type, prq.priority, 1 /* anonymity */,
3565                             expiration,
3566                             1 + prq.priority, MAX_DATASTORE_QUEUE,
3567                             GNUNET_CONSTANTS_SERVICE_TIMEOUT,
3568                             &put_migration_continuation,
3569                             start);
3570     }
3571   putl = GNUNET_LOAD_get_load (datastore_put_load);
3572   if ( (GNUNET_NO == prq.request_found) &&
3573        ( (GNUNET_YES != active_migration) ||
3574          (putl > 2.5 * (1 + prq.priority)) ) )
3575     {
3576       cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
3577                                               &other->hashPubKey);
3578       if (GNUNET_TIME_absolute_get_duration (cp->last_migration_block).rel_value < 5000)
3579         return GNUNET_OK; /* already blocked */
3580       /* We're too busy; send MigrationStop message! */
3581       if (GNUNET_YES != active_migration)
3582         putl = 1.0 + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 5);
3583       block_time = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
3584                                                   5000 + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
3585                                                                                    (unsigned int) (60000 * putl * putl)));
3586
3587       cp->last_migration_block = GNUNET_TIME_relative_to_absolute (block_time);
3588       pm = GNUNET_malloc (sizeof (struct PendingMessage) +
3589                           sizeof (struct MigrationStopMessage));
3590       pm->msize = sizeof (struct MigrationStopMessage);
3591       pm->priority = UINT32_MAX;
3592       msm = (struct MigrationStopMessage*) &pm[1];
3593       msm->header.size = htons (sizeof (struct MigrationStopMessage));
3594       msm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP);
3595       msm->duration = GNUNET_TIME_relative_hton (block_time);
3596       add_to_pending_messages_for_peer (cp,
3597                                         pm,
3598                                         NULL);
3599     }
3600   return GNUNET_OK;
3601 }
3602
3603
3604 /**
3605  * Handle P2P "MIGRATION_STOP" message.
3606  *
3607  * @param cls closure, always NULL
3608  * @param other the other peer involved (sender or receiver, NULL
3609  *        for loopback messages where we are both sender and receiver)
3610  * @param message the actual message
3611  * @param latency reported latency of the connection with 'other'
3612  * @param distance reported distance (DV) to 'other'
3613  * @return GNUNET_OK to keep the connection open,
3614  *         GNUNET_SYSERR to close it (signal serious error)
3615  */
3616 static int
3617 handle_p2p_migration_stop (void *cls,
3618                            const struct GNUNET_PeerIdentity *other,
3619                            const struct GNUNET_MessageHeader *message,
3620                            struct GNUNET_TIME_Relative latency,
3621                            uint32_t distance)
3622 {
3623   struct ConnectedPeer *cp;
3624   const struct MigrationStopMessage *msm;
3625
3626   msm = (const struct MigrationStopMessage*) message;
3627   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
3628                                           &other->hashPubKey);
3629   if (cp == NULL)
3630     {
3631       GNUNET_break (0);
3632       return GNUNET_OK;
3633     }
3634   cp->migration_blocked = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_relative_ntoh (msm->duration));
3635   return GNUNET_OK;
3636 }
3637
3638
3639
3640 /* **************************** P2P GET Handling ************************ */
3641
3642
3643 /**
3644  * Closure for 'check_duplicate_request_{peer,client}'.
3645  */
3646 struct CheckDuplicateRequestClosure
3647 {
3648   /**
3649    * The new request we should check if it already exists.
3650    */
3651   const struct PendingRequest *pr;
3652
3653   /**
3654    * Existing request found by the checker, NULL if none.
3655    */
3656   struct PendingRequest *have;
3657 };
3658
3659
3660 /**
3661  * Iterator over entries in the 'query_request_map' that
3662  * tries to see if we have the same request pending from
3663  * the same client already.
3664  *
3665  * @param cls closure (our 'struct CheckDuplicateRequestClosure')
3666  * @param key current key code (query, ignored, must match)
3667  * @param value value in the hash map (a 'struct PendingRequest'
3668  *              that already exists)
3669  * @return GNUNET_YES if we should continue to
3670  *         iterate (no match yet)
3671  *         GNUNET_NO if not (match found).
3672  */
3673 static int
3674 check_duplicate_request_client (void *cls,
3675                                 const GNUNET_HashCode * key,
3676                                 void *value)
3677 {
3678   struct CheckDuplicateRequestClosure *cdc = cls;
3679   struct PendingRequest *have = value;
3680
3681   if (have->client_request_list == NULL)
3682     return GNUNET_YES;
3683   if ( (cdc->pr->client_request_list->client_list->client == have->client_request_list->client_list->client) &&
3684        (cdc->pr != have) )
3685     {
3686       cdc->have = have;
3687       return GNUNET_NO;
3688     }
3689   return GNUNET_YES;
3690 }
3691
3692
3693 /**
3694  * We're processing (local) results for a search request
3695  * from another peer.  Pass applicable results to the
3696  * peer and if we are done either clean up (operation
3697  * complete) or forward to other peers (more results possible).
3698  *
3699  * @param cls our closure (struct LocalGetContext)
3700  * @param key key for the content
3701  * @param size number of bytes in data
3702  * @param data content stored
3703  * @param type type of the content
3704  * @param priority priority of the content
3705  * @param anonymity anonymity-level for the content
3706  * @param expiration expiration time for the content
3707  * @param uid unique identifier for the datum;
3708  *        maybe 0 if no unique identifier is available
3709  */
3710 static void
3711 process_local_reply (void *cls,
3712                      const GNUNET_HashCode * key,
3713                      size_t size,
3714                      const void *data,
3715                      enum GNUNET_BLOCK_Type type,
3716                      uint32_t priority,
3717                      uint32_t anonymity,
3718                      struct GNUNET_TIME_Absolute
3719                      expiration,
3720                      uint64_t uid)
3721 {
3722   struct PendingRequest *pr = cls;
3723   struct ProcessReplyClosure prq;
3724   struct CheckDuplicateRequestClosure cdrc;
3725   GNUNET_HashCode query;
3726   unsigned int old_rf;
3727
3728   if (NULL == key)
3729     {
3730 #if DEBUG_FS > 1
3731       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3732                   "Done processing local replies, forwarding request to other peers.\n");
3733 #endif
3734       pr->qe = NULL;
3735       if (pr->client_request_list != NULL)
3736         {
3737           GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
3738                                       GNUNET_YES);
3739           /* Figure out if this is a duplicate request and possibly
3740              merge 'struct PendingRequest' entries */
3741           cdrc.have = NULL;
3742           cdrc.pr = pr;
3743           GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
3744                                                       &pr->query,
3745                                                       &check_duplicate_request_client,
3746                                                       &cdrc);
3747           if (cdrc.have != NULL)
3748             {
3749 #if DEBUG_FS
3750               GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3751                           "Received request for block `%s' twice from client, will only request once.\n",
3752                           GNUNET_h2s (&pr->query));
3753 #endif
3754
3755               destroy_pending_request (pr);
3756               return;
3757             }
3758         }
3759       if (pr->local_only == GNUNET_YES)
3760         {
3761           destroy_pending_request (pr);
3762           return;
3763         }
3764       /* no more results */
3765       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
3766         pr->task = GNUNET_SCHEDULER_add_now (&forward_request_task,
3767                                              pr);
3768       return;
3769     }
3770 #if DEBUG_FS
3771   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3772               "New local response to `%s' of type %u.\n",
3773               GNUNET_h2s (key),
3774               type);
3775 #endif
3776   if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
3777     {
3778 #if DEBUG_FS
3779       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3780                   "Found ONDEMAND block, performing on-demand encoding\n");
3781 #endif
3782       GNUNET_STATISTICS_update (stats,
3783                                 gettext_noop ("# on-demand blocks matched requests"),
3784                                 1,
3785                                 GNUNET_NO);
3786       if (GNUNET_OK !=
3787           GNUNET_FS_handle_on_demand_block (key, size, data, type, priority,
3788                                             anonymity, expiration, uid,
3789                                             &process_local_reply,
3790                                             pr))
3791       if (pr->qe != NULL)
3792         {
3793           GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
3794         }
3795       return;
3796     }
3797   old_rf = pr->results_found;
3798   memset (&prq, 0, sizeof (prq));
3799   prq.data = data;
3800   prq.expiration = expiration;
3801   prq.size = size;
3802   if (GNUNET_OK !=
3803       GNUNET_BLOCK_get_key (block_ctx,
3804                             type,
3805                             data,
3806                             size,
3807                             &query))
3808     {
3809       GNUNET_break (0);
3810       GNUNET_DATASTORE_remove (dsh,
3811                                key,
3812                                size, data,
3813                                -1, -1,
3814                                GNUNET_TIME_UNIT_FOREVER_REL,
3815                                NULL, NULL);
3816       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
3817       return;
3818     }
3819   prq.type = type;
3820   prq.priority = priority;
3821   prq.finished = GNUNET_NO;
3822   prq.request_found = GNUNET_NO;
3823   if ( (old_rf == 0) &&
3824        (pr->results_found == 0) )
3825     update_datastore_delays (pr->start_time);
3826   process_reply (&prq, key, pr);
3827   if (prq.finished == GNUNET_YES)
3828     return;
3829   if (pr->qe == NULL)
3830     return; /* done here */
3831   if (prq.eval == GNUNET_BLOCK_EVALUATION_OK_LAST)
3832     {
3833       pr->local_only = GNUNET_YES; /* do not forward */
3834       GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
3835       return;
3836     }
3837   if ( (pr->client_request_list == NULL) &&
3838        ( (GNUNET_YES == test_get_load_too_high (0)) ||
3839          (pr->results_found > 5 + 2 * pr->priority) ) )
3840     {
3841 #if DEBUG_FS > 2
3842       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3843                   "Load too high, done with request\n");
3844 #endif
3845       GNUNET_STATISTICS_update (stats,
3846                                 gettext_noop ("# processing result set cut short due to load"),
3847                                 1,
3848                                 GNUNET_NO);
3849       GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
3850       return;
3851     }
3852   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
3853 }
3854
3855
3856 /**
3857  * We've received a request with the specified priority.  Bound it
3858  * according to how much we trust the given peer.
3859  *
3860  * @param prio_in requested priority
3861  * @param cp the peer making the request
3862  * @return effective priority
3863  */
3864 static int32_t
3865 bound_priority (uint32_t prio_in,
3866                 struct ConnectedPeer *cp)
3867 {
3868 #define N ((double)128.0)
3869   uint32_t ret;
3870   double rret;
3871   int ld;
3872
3873   ld = test_get_load_too_high (0);
3874   if (ld == GNUNET_SYSERR)
3875     {
3876       GNUNET_STATISTICS_update (stats,
3877                                 gettext_noop ("# requests done for free (low load)"),
3878                                 1,
3879                                 GNUNET_NO);
3880       return 0; /* excess resources */
3881     }
3882   if (prio_in > INT32_MAX)
3883     prio_in = INT32_MAX;
3884   ret = - change_host_trust (cp, - (int) prio_in);
3885   if (ret > 0)
3886     {
3887       if (ret > current_priorities + N)
3888         rret = current_priorities + N;
3889       else
3890         rret = ret;
3891       current_priorities
3892         = (current_priorities * (N-1) + rret)/N;
3893     }
3894   if ( (ld == GNUNET_YES) && (ret > 0) )
3895     {
3896       /* try with charging */
3897       ld = test_get_load_too_high (ret);
3898     }
3899   if (ld == GNUNET_YES)
3900     {
3901       GNUNET_STATISTICS_update (stats,
3902                                 gettext_noop ("# request dropped, priority insufficient"),
3903                                 1,
3904                                 GNUNET_NO);
3905       /* undo charge */
3906       change_host_trust (cp, (int) ret);
3907       return -1; /* not enough resources */
3908     }
3909   else
3910     {
3911       GNUNET_STATISTICS_update (stats,
3912                                 gettext_noop ("# requests done for a price (normal load)"),
3913                                 1,
3914                                 GNUNET_NO);
3915     }
3916 #undef N
3917   return ret;
3918 }
3919
3920
3921 /**
3922  * Iterator over entries in the 'query_request_map' that
3923  * tries to see if we have the same request pending from
3924  * the same peer already.
3925  *
3926  * @param cls closure (our 'struct CheckDuplicateRequestClosure')
3927  * @param key current key code (query, ignored, must match)
3928  * @param value value in the hash map (a 'struct PendingRequest'
3929  *              that already exists)
3930  * @return GNUNET_YES if we should continue to
3931  *         iterate (no match yet)
3932  *         GNUNET_NO if not (match found).
3933  */
3934 static int
3935 check_duplicate_request_peer (void *cls,
3936                               const GNUNET_HashCode * key,
3937                               void *value)
3938 {
3939   struct CheckDuplicateRequestClosure *cdc = cls;
3940   struct PendingRequest *have = value;
3941
3942   if (cdc->pr->target_pid == have->target_pid)
3943     {
3944       cdc->have = have;
3945       return GNUNET_NO;
3946     }
3947   return GNUNET_YES;
3948 }
3949
3950
3951 /**
3952  * Handle P2P "GET" request.
3953  *
3954  * @param cls closure, always NULL
3955  * @param other the other peer involved (sender or receiver, NULL
3956  *        for loopback messages where we are both sender and receiver)
3957  * @param message the actual message
3958  * @param latency reported latency of the connection with 'other'
3959  * @param distance reported distance (DV) to 'other'
3960  * @return GNUNET_OK to keep the connection open,
3961  *         GNUNET_SYSERR to close it (signal serious error)
3962  */
3963 static int
3964 handle_p2p_get (void *cls,
3965                 const struct GNUNET_PeerIdentity *other,
3966                 const struct GNUNET_MessageHeader *message,
3967                 struct GNUNET_TIME_Relative latency,
3968                 uint32_t distance)
3969 {
3970   struct PendingRequest *pr;
3971   struct ConnectedPeer *cp;
3972   struct ConnectedPeer *cps;
3973   struct CheckDuplicateRequestClosure cdc;
3974   struct GNUNET_TIME_Relative timeout;
3975   uint16_t msize;
3976   const struct GetMessage *gm;
3977   unsigned int bits;
3978   const GNUNET_HashCode *opt;
3979   uint32_t bm;
3980   size_t bfsize;
3981   uint32_t ttl_decrement;
3982   int32_t priority;
3983   enum GNUNET_BLOCK_Type type;
3984   int have_ns;
3985
3986   msize = ntohs(message->size);
3987   if (msize < sizeof (struct GetMessage))
3988     {
3989       GNUNET_break_op (0);
3990       return GNUNET_SYSERR;
3991     }
3992   gm = (const struct GetMessage*) message;
3993 #if DEBUG_FS
3994   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3995               "Received request for `%s'\n",
3996               GNUNET_h2s (&gm->query));
3997 #endif
3998   type = ntohl (gm->type);
3999   bm = ntohl (gm->hash_bitmap);
4000   bits = 0;
4001   while (bm > 0)
4002     {
4003       if (1 == (bm & 1))
4004         bits++;
4005       bm >>= 1;
4006     }
4007   if (msize < sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode))
4008     {
4009       GNUNET_break_op (0);
4010       return GNUNET_SYSERR;
4011     }
4012   opt = (const GNUNET_HashCode*) &gm[1];
4013   bfsize = msize - sizeof (struct GetMessage) - bits * sizeof (GNUNET_HashCode);
4014   /* bfsize must be power of 2, check! */
4015   if (0 != ( (bfsize - 1) & bfsize))
4016     {
4017       GNUNET_break_op (0);
4018       return GNUNET_SYSERR;
4019     }
4020   bm = ntohl (gm->hash_bitmap);
4021   bits = 0;
4022   cps = GNUNET_CONTAINER_multihashmap_get (connected_peers,
4023                                            &other->hashPubKey);
4024   if (NULL == cps)
4025     {
4026       /* peer must have just disconnected */
4027       GNUNET_STATISTICS_update (stats,
4028                                 gettext_noop ("# requests dropped due to initiator not being connected"),
4029                                 1,
4030                                 GNUNET_NO);
4031       return GNUNET_SYSERR;
4032     }
4033   if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
4034     cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
4035                                             &opt[bits++]);
4036   else
4037     cp = cps;
4038   if (cp == NULL)
4039     {
4040 #if DEBUG_FS
4041       if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
4042         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4043                     "Failed to find RETURN-TO peer `%4s' in connection set. Dropping query.\n",
4044                     GNUNET_i2s ((const struct GNUNET_PeerIdentity*) &opt[bits-1]));
4045
4046       else
4047         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4048                     "Failed to find peer `%4s' in connection set. Dropping query.\n",
4049                     GNUNET_i2s (other));
4050 #endif
4051       GNUNET_STATISTICS_update (stats,
4052                                 gettext_noop ("# requests dropped due to missing reverse route"),
4053                                 1,
4054                                 GNUNET_NO);
4055      /* FIXME: try connect? */
4056       return GNUNET_OK;
4057     }
4058   /* note that we can really only check load here since otherwise
4059      peers could find out that we are overloaded by not being
4060      disconnected after sending us a malformed query... */
4061   priority = bound_priority (ntohl (gm->priority), cps);
4062   if (priority < 0)
4063     {
4064 #if DEBUG_FS
4065       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4066                   "Dropping query from `%s', this peer is too busy.\n",
4067                   GNUNET_i2s (other));
4068 #endif
4069       return GNUNET_OK;
4070     }
4071 #if DEBUG_FS
4072   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4073               "Received request for `%s' of type %u from peer `%4s' with flags %u\n",
4074               GNUNET_h2s (&gm->query),
4075               (unsigned int) type,
4076               GNUNET_i2s (other),
4077               (unsigned int) bm);
4078 #endif
4079   have_ns = (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE));
4080   pr = GNUNET_malloc (sizeof (struct PendingRequest) +
4081                       (have_ns ? sizeof(GNUNET_HashCode) : 0));
4082   if (have_ns)
4083     {
4084       pr->namespace = (GNUNET_HashCode*) &pr[1];
4085       memcpy (&pr[1], &opt[bits++], sizeof (GNUNET_HashCode));
4086     }
4087   if ( (GNUNET_LOAD_get_load (cp->transmission_delay) > 3 * (1 + priority)) ||
4088        (GNUNET_LOAD_get_average (cp->transmission_delay) >
4089         GNUNET_CONSTANTS_MAX_CORK_DELAY.rel_value * 2 + GNUNET_LOAD_get_average (rt_entry_lifetime)) )
4090     {
4091       /* don't have BW to send to peer, or would likely take longer than we have for it,
4092          so at best indirect the query */
4093       priority = 0;
4094       pr->forward_only = GNUNET_YES;
4095     }
4096   pr->type = type;
4097   pr->mingle = ntohl (gm->filter_mutator);
4098   if (0 != (bm & GET_MESSAGE_BIT_TRANSMIT_TO))
4099     pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &opt[bits++]);
4100   pr->anonymity_level = 1;
4101   pr->priority = (uint32_t) priority;
4102   pr->ttl = bound_ttl (ntohl (gm->ttl), pr->priority);
4103   pr->query = gm->query;
4104   /* decrement ttl (always) */
4105   ttl_decrement = 2 * TTL_DECREMENT +
4106     GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
4107                               TTL_DECREMENT);
4108   if ( (pr->ttl < 0) &&
4109        (((int32_t)(pr->ttl - ttl_decrement)) > 0) )
4110     {
4111 #if DEBUG_FS
4112       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4113                   "Dropping query from `%s' due to TTL underflow (%d - %u).\n",
4114                   GNUNET_i2s (other),
4115                   pr->ttl,
4116                   ttl_decrement);
4117 #endif
4118       GNUNET_STATISTICS_update (stats,
4119                                 gettext_noop ("# requests dropped due TTL underflow"),
4120                                 1,
4121                                 GNUNET_NO);
4122       /* integer underflow => drop (should be very rare)! */
4123       GNUNET_free (pr);
4124       return GNUNET_OK;
4125     }
4126   pr->ttl -= ttl_decrement;
4127   pr->start_time = GNUNET_TIME_absolute_get ();
4128
4129   /* get bloom filter */
4130   if (bfsize > 0)
4131     {
4132       pr->bf = GNUNET_CONTAINER_bloomfilter_init ((const char*) &opt[bits],
4133                                                   bfsize,
4134                                                   BLOOMFILTER_K);
4135       pr->bf_size = bfsize;
4136     }
4137   cdc.have = NULL;
4138   cdc.pr = pr;
4139   GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
4140                                               &gm->query,
4141                                               &check_duplicate_request_peer,
4142                                               &cdc);
4143   if (cdc.have != NULL)
4144     {
4145       if (cdc.have->start_time.abs_value + cdc.have->ttl >=
4146           pr->start_time.abs_value + pr->ttl)
4147         {
4148           /* existing request has higher TTL, drop new one! */
4149           cdc.have->priority += pr->priority;
4150           destroy_pending_request (pr);
4151 #if DEBUG_FS
4152           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4153                       "Have existing request with higher TTL, dropping new request.\n",
4154                       GNUNET_i2s (other));
4155 #endif
4156           GNUNET_STATISTICS_update (stats,
4157                                     gettext_noop ("# requests dropped due to higher-TTL request"),
4158                                     1,
4159                                     GNUNET_NO);
4160           return GNUNET_OK;
4161         }
4162       else
4163         {
4164           /* existing request has lower TTL, drop old one! */
4165           pr->priority += cdc.have->priority;
4166           /* Possible optimization: if we have applicable pending
4167              replies in 'cdc.have', we might want to move those over
4168              (this is a really rare special-case, so it is not clear
4169              that this would be worth it) */
4170           destroy_pending_request (cdc.have);
4171           /* keep processing 'pr'! */
4172         }
4173     }
4174
4175   pr->cp = cp;
4176   GNUNET_break (GNUNET_OK ==
4177                 GNUNET_CONTAINER_multihashmap_put (query_request_map,
4178                                                    &gm->query,
4179                                                    pr,
4180                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
4181   GNUNET_break (GNUNET_OK ==
4182                 GNUNET_CONTAINER_multihashmap_put (peer_request_map,
4183                                                    &other->hashPubKey,
4184                                                    pr,
4185                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
4186
4187   pr->hnode = GNUNET_CONTAINER_heap_insert (requests_by_expiration_heap,
4188                                             pr,
4189                                             pr->start_time.abs_value + pr->ttl);
4190
4191   GNUNET_STATISTICS_update (stats,
4192                             gettext_noop ("# P2P searches received"),
4193                             1,
4194                             GNUNET_NO);
4195   GNUNET_STATISTICS_update (stats,
4196                             gettext_noop ("# P2P searches active"),
4197                             1,
4198                             GNUNET_NO);
4199
4200   /* calculate change in traffic preference */
4201   cps->inc_preference += pr->priority * 1000 + QUERY_BANDWIDTH_VALUE;
4202   /* process locally */
4203   if (type == GNUNET_BLOCK_TYPE_FS_DBLOCK)
4204     type = GNUNET_BLOCK_TYPE_ANY; /* to get on-demand as well */
4205   timeout = GNUNET_TIME_relative_multiply (BASIC_DATASTORE_REQUEST_DELAY,
4206                                            (pr->priority + 1));
4207   if (GNUNET_YES != pr->forward_only)
4208     {
4209 #if DEBUG_FS
4210       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4211                   "Handing request for `%s' to datastore\n",
4212                   GNUNET_h2s (&gm->query));
4213 #endif
4214       pr->qe = GNUNET_DATASTORE_get (dsh,
4215                                      &gm->query,
4216                                      type,
4217                                      pr->priority + 1,
4218                                      MAX_DATASTORE_QUEUE,
4219                                      timeout,
4220                                      &process_local_reply,
4221                                      pr);
4222       if (NULL == pr->qe)
4223         {
4224           GNUNET_STATISTICS_update (stats,
4225                                     gettext_noop ("# requests dropped by datastore (queue length limit)"),
4226                                     1,
4227                                     GNUNET_NO);
4228         }
4229     }
4230   else
4231     {
4232       GNUNET_STATISTICS_update (stats,
4233                                 gettext_noop ("# requests forwarded due to high load"),
4234                                 1,
4235                                 GNUNET_NO);
4236     }
4237
4238   /* Are multiple results possible (and did we look locally)?  If so, start processing remotely now! */
4239   switch (pr->type)
4240     {
4241     case GNUNET_BLOCK_TYPE_FS_DBLOCK:
4242     case GNUNET_BLOCK_TYPE_FS_IBLOCK:
4243       /* only one result, wait for datastore */
4244       if (GNUNET_YES != pr->forward_only)
4245         {
4246           GNUNET_STATISTICS_update (stats,
4247                                     gettext_noop ("# requests not instantly forwarded (waiting for datastore)"),
4248                                     1,
4249                                     GNUNET_NO);
4250           break;
4251         }
4252     default:
4253       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
4254         pr->task = GNUNET_SCHEDULER_add_now (&forward_request_task,
4255                                              pr);
4256     }
4257
4258   /* make sure we don't track too many requests */
4259   if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap) > max_pending_requests)
4260     {
4261       pr = GNUNET_CONTAINER_heap_peek (requests_by_expiration_heap);
4262       GNUNET_assert (pr != NULL);
4263       destroy_pending_request (pr);
4264     }
4265   return GNUNET_OK;
4266 }
4267
4268
4269 /* **************************** CS GET Handling ************************ */
4270
4271
4272 /**
4273  * Handle START_SEARCH-message (search request from client).
4274  *
4275  * @param cls closure
4276  * @param client identification of the client
4277  * @param message the actual message
4278  */
4279 static void
4280 handle_start_search (void *cls,
4281                      struct GNUNET_SERVER_Client *client,
4282                      const struct GNUNET_MessageHeader *message)
4283 {
4284   static GNUNET_HashCode all_zeros;
4285   const struct SearchMessage *sm;
4286   struct ClientList *cl;
4287   struct ClientRequestList *crl;
4288   struct PendingRequest *pr;
4289   uint16_t msize;
4290   unsigned int sc;
4291   enum GNUNET_BLOCK_Type type;
4292
4293   msize = ntohs (message->size);
4294   if ( (msize < sizeof (struct SearchMessage)) ||
4295        (0 != (msize - sizeof (struct SearchMessage)) % sizeof (GNUNET_HashCode)) )
4296     {
4297       GNUNET_break (0);
4298       GNUNET_SERVER_receive_done (client,
4299                                   GNUNET_SYSERR);
4300       return;
4301     }
4302   GNUNET_STATISTICS_update (stats,
4303                             gettext_noop ("# client searches received"),
4304                             1,
4305                             GNUNET_NO);
4306   sc = (msize - sizeof (struct SearchMessage)) / sizeof (GNUNET_HashCode);
4307   sm = (const struct SearchMessage*) message;
4308   type = ntohl (sm->type);
4309 #if DEBUG_FS
4310   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4311               "Received request for `%s' of type %u from local client\n",
4312               GNUNET_h2s (&sm->query),
4313               (unsigned int) type);
4314 #endif
4315   cl = client_list;
4316   while ( (cl != NULL) &&
4317           (cl->client != client) )
4318     cl = cl->next;
4319   if (cl == NULL)
4320     {
4321       cl = GNUNET_malloc (sizeof (struct ClientList));
4322       cl->client = client;
4323       GNUNET_SERVER_client_keep (client);
4324       cl->next = client_list;
4325       client_list = cl;
4326     }
4327   /* detect duplicate KBLOCK requests */
4328   if ( (type == GNUNET_BLOCK_TYPE_FS_KBLOCK) ||
4329        (type == GNUNET_BLOCK_TYPE_FS_NBLOCK) ||
4330        (type == GNUNET_BLOCK_TYPE_ANY) )
4331     {
4332       crl = cl->rl_head;
4333       while ( (crl != NULL) &&
4334               ( (0 != memcmp (&crl->req->query,
4335                               &sm->query,
4336                               sizeof (GNUNET_HashCode))) ||
4337                 (crl->req->type != type) ) )
4338         crl = crl->next;
4339       if (crl != NULL)
4340         {
4341 #if DEBUG_FS
4342           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4343                       "Have existing request, merging content-seen lists.\n");
4344 #endif
4345           pr = crl->req;
4346           /* Duplicate request (used to send long list of
4347              known/blocked results); merge 'pr->replies_seen'
4348              and update bloom filter */
4349           GNUNET_array_grow (pr->replies_seen,
4350                              pr->replies_seen_size,
4351                              pr->replies_seen_off + sc);
4352           memcpy (&pr->replies_seen[pr->replies_seen_off],
4353                   &sm[1],
4354                   sc * sizeof (GNUNET_HashCode));
4355           pr->replies_seen_off += sc;
4356           refresh_bloomfilter (pr);
4357           GNUNET_STATISTICS_update (stats,
4358                                     gettext_noop ("# client searches updated (merged content seen list)"),
4359                                     1,
4360                                     GNUNET_NO);
4361           GNUNET_SERVER_receive_done (client,
4362                                       GNUNET_OK);
4363           return;
4364         }
4365     }
4366   GNUNET_STATISTICS_update (stats,
4367                             gettext_noop ("# client searches active"),
4368                             1,
4369                             GNUNET_NO);
4370   pr = GNUNET_malloc (sizeof (struct PendingRequest) +
4371                       ((type == GNUNET_BLOCK_TYPE_FS_SBLOCK) ? sizeof(GNUNET_HashCode) : 0));
4372   crl = GNUNET_malloc (sizeof (struct ClientRequestList));
4373   memset (crl, 0, sizeof (struct ClientRequestList));
4374   crl->client_list = cl;
4375   GNUNET_CONTAINER_DLL_insert (cl->rl_head,
4376                                cl->rl_tail,
4377                                crl);
4378   crl->req = pr;
4379   pr->type = type;
4380   pr->client_request_list = crl;
4381   GNUNET_array_grow (pr->replies_seen,
4382                      pr->replies_seen_size,
4383                      sc);
4384   memcpy (pr->replies_seen,
4385           &sm[1],
4386           sc * sizeof (GNUNET_HashCode));
4387   pr->replies_seen_off = sc;
4388   pr->anonymity_level = ntohl (sm->anonymity_level);
4389   pr->start_time = GNUNET_TIME_absolute_get ();
4390   refresh_bloomfilter (pr);
4391   pr->query = sm->query;
4392   if (0 == (1 & ntohl (sm->options)))
4393     pr->local_only = GNUNET_NO;
4394   else
4395     pr->local_only = GNUNET_YES;
4396   switch (type)
4397     {
4398     case GNUNET_BLOCK_TYPE_FS_DBLOCK:
4399     case GNUNET_BLOCK_TYPE_FS_IBLOCK:
4400       if (0 != memcmp (&sm->target,
4401                        &all_zeros,
4402                        sizeof (GNUNET_HashCode)))
4403         pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &sm->target);
4404       break;
4405     case GNUNET_BLOCK_TYPE_FS_SBLOCK:
4406       pr->namespace = (GNUNET_HashCode*) &pr[1];
4407       memcpy (&pr[1], &sm->target, sizeof (GNUNET_HashCode));
4408       break;
4409     default:
4410       break;
4411     }
4412   GNUNET_break (GNUNET_OK ==
4413                 GNUNET_CONTAINER_multihashmap_put (query_request_map,
4414                                                    &sm->query,
4415                                                    pr,
4416                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
4417   if (type == GNUNET_BLOCK_TYPE_FS_DBLOCK)
4418     type = GNUNET_BLOCK_TYPE_ANY; /* get on-demand blocks too! */
4419   pr->qe = GNUNET_DATASTORE_get (dsh,
4420                                  &sm->query,
4421                                  type,
4422                                  -3, -1,
4423                                  GNUNET_CONSTANTS_SERVICE_TIMEOUT,
4424                                  &process_local_reply,
4425                                  pr);
4426 }
4427
4428
4429 /* **************************** Startup ************************ */
4430
4431 /**
4432  * Process fs requests.
4433  *
4434  * @param server the initialized server
4435  * @param c configuration to use
4436  */
4437 static int
4438 main_init (struct GNUNET_SERVER_Handle *server,
4439            const struct GNUNET_CONFIGURATION_Handle *c)
4440 {
4441   static const struct GNUNET_CORE_MessageHandler p2p_handlers[] =
4442     {
4443       { &handle_p2p_get,
4444         GNUNET_MESSAGE_TYPE_FS_GET, 0 },
4445       { &handle_p2p_put,
4446         GNUNET_MESSAGE_TYPE_FS_PUT, 0 },
4447       { &handle_p2p_migration_stop,
4448         GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP,
4449         sizeof (struct MigrationStopMessage) },
4450       { NULL, 0, 0 }
4451     };
4452   static const struct GNUNET_SERVER_MessageHandler handlers[] = {
4453     {&GNUNET_FS_handle_index_start, NULL,
4454      GNUNET_MESSAGE_TYPE_FS_INDEX_START, 0},
4455     {&GNUNET_FS_handle_index_list_get, NULL,
4456      GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_GET, sizeof(struct GNUNET_MessageHeader) },
4457     {&GNUNET_FS_handle_unindex, NULL, GNUNET_MESSAGE_TYPE_FS_UNINDEX,
4458      sizeof (struct UnindexMessage) },
4459     {&handle_start_search, NULL, GNUNET_MESSAGE_TYPE_FS_START_SEARCH,
4460      0 },
4461     {NULL, NULL, 0, 0}
4462   };
4463   unsigned long long enc = 128;
4464
4465   cfg = c;
4466   stats = GNUNET_STATISTICS_create ("fs", cfg);
4467   min_migration_delay = GNUNET_TIME_UNIT_SECONDS;
4468   if ( (GNUNET_OK !=
4469         GNUNET_CONFIGURATION_get_value_number (cfg,
4470                                                "fs",
4471                                                "MAX_PENDING_REQUESTS",
4472                                                &max_pending_requests)) ||
4473        (GNUNET_OK !=
4474         GNUNET_CONFIGURATION_get_value_number (cfg,
4475                                                "fs",
4476                                                "EXPECTED_NEIGHBOUR_COUNT",
4477                                                &enc)) ||
4478        (GNUNET_OK !=
4479         GNUNET_CONFIGURATION_get_value_time (cfg,
4480                                              "fs",
4481                                              "MIN_MIGRATION_DELAY",
4482                                              &min_migration_delay)) )
4483     {
4484       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
4485                   _("Configuration fails to specify certain parameters, assuming default values."));
4486     }
4487   connected_peers = GNUNET_CONTAINER_multihashmap_create (enc);
4488   query_request_map = GNUNET_CONTAINER_multihashmap_create (max_pending_requests);
4489   rt_entry_lifetime = GNUNET_LOAD_value_init (GNUNET_TIME_UNIT_FOREVER_REL);
4490   peer_request_map = GNUNET_CONTAINER_multihashmap_create (enc);
4491   requests_by_expiration_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
4492   core = GNUNET_CORE_connect (cfg,
4493                               1, /* larger? */
4494                               GNUNET_TIME_UNIT_FOREVER_REL,
4495                               NULL,
4496                               NULL,
4497                               &peer_connect_handler,
4498                               &peer_disconnect_handler,
4499                               &peer_status_handler,
4500                               NULL, GNUNET_NO,
4501                               NULL, GNUNET_NO,
4502                               p2p_handlers);
4503   if (NULL == core)
4504     {
4505       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
4506                   _("Failed to connect to `%s' service.\n"),
4507                   "core");
4508       GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
4509       connected_peers = NULL;
4510       GNUNET_CONTAINER_multihashmap_destroy (query_request_map);
4511       query_request_map = NULL;
4512       GNUNET_LOAD_value_free (rt_entry_lifetime);
4513       rt_entry_lifetime = NULL;
4514       GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
4515       requests_by_expiration_heap = NULL;
4516       GNUNET_CONTAINER_multihashmap_destroy (peer_request_map);
4517       peer_request_map = NULL;
4518       if (dsh != NULL)
4519         {
4520           GNUNET_DATASTORE_disconnect (dsh, GNUNET_NO);
4521           dsh = NULL;
4522         }
4523       return GNUNET_SYSERR;
4524     }
4525   /* FIXME: distinguish between sending and storing in options? */
4526   if (active_migration)
4527     {
4528       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
4529                   _("Content migration is enabled, will start to gather data\n"));
4530       consider_migration_gathering ();
4531     }
4532   consider_dht_put_gathering (NULL);
4533   GNUNET_SERVER_disconnect_notify (server,
4534                                    &handle_client_disconnect,
4535                                    NULL);
4536   GNUNET_assert (GNUNET_OK ==
4537                  GNUNET_CONFIGURATION_get_value_filename (cfg,
4538                                                           "fs",
4539                                                           "TRUST",
4540                                                           &trustDirectory));
4541   GNUNET_DISK_directory_create (trustDirectory);
4542   GNUNET_SCHEDULER_add_with_priority (GNUNET_SCHEDULER_PRIORITY_HIGH,
4543                                       &cron_flush_trust, NULL);
4544
4545
4546   GNUNET_SERVER_add_handlers (server, handlers);
4547   GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL,
4548                                 &shutdown_task,
4549                                 NULL);
4550   return GNUNET_OK;
4551 }
4552
4553
4554 /**
4555  * Process fs requests.
4556  *
4557  * @param cls closure
4558  * @param server the initialized server
4559  * @param cfg configuration to use
4560  */
4561 static void
4562 run (void *cls,
4563      struct GNUNET_SERVER_Handle *server,
4564      const struct GNUNET_CONFIGURATION_Handle *cfg)
4565 {
4566   active_migration = GNUNET_CONFIGURATION_get_value_yesno (cfg,
4567                                                            "FS",
4568                                                            "ACTIVEMIGRATION");
4569   dsh = GNUNET_DATASTORE_connect (cfg);
4570   if (dsh == NULL)
4571     {
4572       GNUNET_SCHEDULER_shutdown ();
4573       return;
4574     }
4575   datastore_get_load = GNUNET_LOAD_value_init (DATASTORE_LOAD_AUTODECLINE);
4576   datastore_put_load = GNUNET_LOAD_value_init (DATASTORE_LOAD_AUTODECLINE);
4577   block_cfg = GNUNET_CONFIGURATION_create ();
4578   GNUNET_CONFIGURATION_set_value_string (block_cfg,
4579                                          "block",
4580                                          "PLUGINS",
4581                                          "fs");
4582   block_ctx = GNUNET_BLOCK_context_create (block_cfg);
4583   GNUNET_assert (NULL != block_ctx);
4584   dht_handle = GNUNET_DHT_connect (cfg,
4585                                    FS_DHT_HT_SIZE);
4586   if ( (GNUNET_OK != GNUNET_FS_indexing_init (cfg, dsh)) ||
4587        (GNUNET_OK != main_init (server, cfg)) )
4588     {
4589       GNUNET_SCHEDULER_shutdown ();
4590       GNUNET_DATASTORE_disconnect (dsh, GNUNET_NO);
4591       dsh = NULL;
4592       GNUNET_DHT_disconnect (dht_handle);
4593       dht_handle = NULL;
4594       GNUNET_BLOCK_context_destroy (block_ctx);
4595       block_ctx = NULL;
4596       GNUNET_CONFIGURATION_destroy (block_cfg);
4597       block_cfg = NULL;
4598       GNUNET_LOAD_value_free (datastore_get_load);
4599       datastore_get_load = NULL;
4600       GNUNET_LOAD_value_free (datastore_put_load);
4601       datastore_put_load = NULL;
4602       return;
4603     }
4604 }
4605
4606
4607 /**
4608  * The main function for the fs service.
4609  *
4610  * @param argc number of arguments from the command line
4611  * @param argv command line arguments
4612  * @return 0 ok, 1 on error
4613  */
4614 int
4615 main (int argc, char *const *argv)
4616 {
4617   return (GNUNET_OK ==
4618           GNUNET_SERVICE_run (argc,
4619                               argv,
4620                               "fs",
4621                               GNUNET_SERVICE_OPTION_NONE,
4622                               &run, NULL)) ? 0 : 1;
4623 }
4624
4625 /* end of gnunet-service-fs.c */