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