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