src/fs/gnunet-service-fs.c

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