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