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