src/fs/gnunet-service-fs.c

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