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