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