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       GNUNET_STATISTICS_update (stats,
 671                                 gettext_noop ("# P2P searches active"),
 672                                 -1,
 673                                 GNUNET_NO);
 674     }
 675   else
 676     {
 677       GNUNET_STATISTICS_update (stats,
 678                                 gettext_noop ("# client searches active"),
 679                                 -1,
 680                                 GNUNET_NO);
 681     }
 682   /* might have already been removed from map in 'process_reply' (if
 683      there was a unique reply) or never inserted if it was a
 684      duplicate; hence ignore the return value here */
 685   (void) GNUNET_CONTAINER_multihashmap_remove (query_request_map,
 686                                                &pr->query,
 687                                                pr);
 688   if (pr->drq != NULL)
 689     {
 690       GNUNET_FS_drq_get_cancel (pr->drq);
 691       pr->drq = NULL;
 692     }
 693   if (pr->client_request_list != NULL)
 694     {
 695       GNUNET_CONTAINER_DLL_remove (pr->client_request_list->client_list->rl_head,
 696                                    pr->client_request_list->client_list->rl_tail,
 697                                    pr->client_request_list);
 698       GNUNET_free (pr->client_request_list);
 699       pr->client_request_list = NULL;
 700     }
 701   if (pr->cp != NULL)
 702     {
 703       GNUNET_PEER_resolve (pr->cp->pid,
 704                            &pid);
 705       (void) GNUNET_CONTAINER_multihashmap_remove (peer_request_map,
 706                                                    &pid.hashPubKey,
 707                                                    pr);
 708       pr->cp = NULL;
 709     }
 710   if (pr->bf != NULL)
 711     {
 712       GNUNET_CONTAINER_bloomfilter_free (pr->bf);
 713       pr->bf = NULL;
 714     }
 715   if (pr->irc != NULL)
 716     {
 717       GNUNET_CORE_peer_change_preference_cancel (pr->irc);
 718       pr->irc = NULL;
 719     }
 720   if (pr->replies_seen != NULL)
 721     {
 722       GNUNET_free (pr->replies_seen);
 723       pr->replies_seen = NULL;
 724     }
 725   if (pr->task != GNUNET_SCHEDULER_NO_TASK)
 726     {
 727       GNUNET_SCHEDULER_cancel (sched,
 728                                pr->task);
 729       pr->task = GNUNET_SCHEDULER_NO_TASK;
 730     }
 731   while (NULL != pr->pending_head)
 732     destroy_pending_message_list_entry (pr->pending_head);
 733   GNUNET_PEER_change_rc (pr->target_pid, -1);
 734   if (pr->used_pids != NULL)
 735     {
 736       GNUNET_PEER_decrement_rcs (pr->used_pids, pr->used_pids_off);
 737       GNUNET_free (pr->used_pids);
 738       pr->used_pids_off = 0;
 739       pr->used_pids_size = 0;
 740       pr->used_pids = NULL;
 741     }
 742   GNUNET_free (pr);
 743 }
 744
 745
 746 /**
 747  * Method called whenever a given peer connects.
 748  *
 749  * @param cls closure, not used
 750  * @param peer peer identity this notification is about
 751  * @param latency reported latency of the connection with 'other'
 752  * @param distance reported distance (DV) to 'other'
 753  */
 754 static void
 755 peer_connect_handler (void *cls,
 756                       const struct
 757                       GNUNET_PeerIdentity * peer,
 758                       struct GNUNET_TIME_Relative latency,
 759                       uint32_t distance)
 760 {
 761   struct ConnectedPeer *cp;
 762
 763   cp = GNUNET_malloc (sizeof (struct ConnectedPeer));
 764   cp->pid = GNUNET_PEER_intern (peer);
 765   GNUNET_CONTAINER_multihashmap_put (connected_peers,
 766                                      &peer->hashPubKey,
 767                                      cp,
 768                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
 769 }
 770
 771
 772 /**
 773  * Free (each) request made by the peer.
 774  *
 775  * @param cls closure, points to peer that the request belongs to
 776  * @param key current key code
 777  * @param value value in the hash map
 778  * @return GNUNET_YES (we should continue to iterate)
 779  */
 780 static int
 781 destroy_request (void *cls,
 782                  const GNUNET_HashCode * key,
 783                  void *value)
 784 {
 785   const struct GNUNET_PeerIdentity * peer = cls;
 786   struct PendingRequest *pr = value;
 787
 788   GNUNET_CONTAINER_multihashmap_remove (peer_request_map,
 789                                         &peer->hashPubKey,
 790                                         pr);
 791   destroy_pending_request (pr);
 792   return GNUNET_YES;
 793 }
 794
 795
 796 /**
 797  * Method called whenever a peer disconnects.
 798  *
 799  * @param cls closure, not used
 800  * @param peer peer identity this notification is about
 801  */
 802 static void
 803 peer_disconnect_handler (void *cls,
 804                          const struct
 805                          GNUNET_PeerIdentity * peer)
 806 {
 807   struct ConnectedPeer *cp;
 808   struct PendingMessage *pm;
 809   unsigned int i;
 810
 811   GNUNET_CONTAINER_multihashmap_get_multiple (peer_request_map,
 812                                               &peer->hashPubKey,
 813                                               &destroy_request,
 814                                               (void*) peer);
 815   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
 816                                           &peer->hashPubKey);
 817   if (cp == NULL)
 818     return;
 819   for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
 820     {
 821       if (NULL != cp->last_client_replies[i])
 822         {
 823           GNUNET_SERVER_client_drop (cp->last_client_replies[i]);
 824           cp->last_client_replies[i] = NULL;
 825         }
 826     }
 827   GNUNET_CONTAINER_multihashmap_remove (connected_peers,
 828                                         &peer->hashPubKey,
 829                                         cp);
 830   GNUNET_PEER_change_rc (cp->pid, -1);
 831   GNUNET_PEER_decrement_rcs (cp->last_p2p_replies, P2P_SUCCESS_LIST_SIZE);
 832   if (NULL != cp->cth)
 833     GNUNET_CORE_notify_transmit_ready_cancel (cp->cth);
 834   while (NULL != (pm = cp->pending_messages_head))
 835     destroy_pending_message (pm, 0 /* delivery failed */);
 836   GNUNET_break (0 == cp->pending_requests);
 837   GNUNET_free (cp);
 838 }
 839
 840
 841 /**
 842  * Iterator over hash map entries that removes all occurences
 843  * of the given 'client' from the 'last_client_replies' of the
 844  * given connected peer.
 845  *
 846  * @param cls closure, the 'struct GNUNET_SERVER_Client*' to remove
 847  * @param key current key code (unused)
 848  * @param value value in the hash map (the 'struct ConnectedPeer*' to change)
 849  * @return GNUNET_YES (we should continue to iterate)
 850  */
 851 static int
 852 remove_client_from_last_client_replies (void *cls,
 853                                         const GNUNET_HashCode * key,
 854                                         void *value)
 855 {
 856   struct GNUNET_SERVER_Client *client = cls;
 857   struct ConnectedPeer *cp = value;
 858   unsigned int i;
 859
 860   for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
 861     {
 862       if (cp->last_client_replies[i] == client)
 863         {
 864           GNUNET_SERVER_client_drop (cp->last_client_replies[i]);
 865           cp->last_client_replies[i] = NULL;
 866         }
 867     }
 868   return GNUNET_YES;
 869 }
 870
 871
 872 /**
 873  * A client disconnected.  Remove all of its pending queries.
 874  *
 875  * @param cls closure, NULL
 876  * @param client identification of the client
 877  */
 878 static void
 879 handle_client_disconnect (void *cls,
 880                           struct GNUNET_SERVER_Client
 881                           * client)
 882 {
 883   struct ClientList *pos;
 884   struct ClientList *prev;
 885   struct ClientRequestList *rcl;
 886   struct ClientResponseMessage *creply;
 887
 888   if (client == NULL)
 889     return;
 890   prev = NULL;
 891   pos = client_list;
 892   while ( (NULL != pos) &&
 893           (pos->client != client) )
 894     {
 895       prev = pos;
 896       pos = pos->next;
 897     }
 898   if (pos == NULL)
 899     return; /* no requests pending for this client */
 900   while (NULL != (rcl = pos->rl_head))
 901     {
 902       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 903                   "Destroying pending request `%s' on disconnect\n",
 904                   GNUNET_h2s (&rcl->req->query));
 905       destroy_pending_request (rcl->req);
 906     }
 907   if (prev == NULL)
 908     client_list = pos->next;
 909   else
 910     prev->next = pos->next;
 911   if (pos->th != NULL)
 912     {
 913       GNUNET_CONNECTION_notify_transmit_ready_cancel (pos->th);
 914       pos->th = NULL;
 915     }
 916   while (NULL != (creply = pos->res_head))
 917     {
 918       GNUNET_CONTAINER_DLL_remove (pos->res_head,
 919                                    pos->res_tail,
 920                                    creply);
 921       GNUNET_free (creply);
 922     }
 923   GNUNET_SERVER_client_drop (pos->client);
 924   GNUNET_free (pos);
 925   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
 926                                          &remove_client_from_last_client_replies,
 927                                          client);
 928 }
 929
 930
 931 /**
 932  * Iterator to free peer entries.
 933  *
 934  * @param cls closure, unused
 935  * @param key current key code
 936  * @param value value in the hash map (peer entry)
 937  * @return GNUNET_YES (we should continue to iterate)
 938  */
 939 static int
 940 clean_peer (void *cls,
 941             const GNUNET_HashCode * key,
 942             void *value)
 943 {
 944   peer_disconnect_handler (NULL, (const struct GNUNET_PeerIdentity*) key);
 945   return GNUNET_YES;
 946 }
 947
 948
 949 /**
 950  * Task run during shutdown.
 951  *
 952  * @param cls unused
 953  * @param tc unused
 954  */
 955 static void
 956 shutdown_task (void *cls,
 957                const struct GNUNET_SCHEDULER_TaskContext *tc)
 958 {
 959   while (client_list != NULL)
 960     handle_client_disconnect (NULL,
 961                               client_list->client);
 962   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
 963                                          &clean_peer,
 964                                          NULL);
 965   GNUNET_break (0 == GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap));
 966   GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
 967   requests_by_expiration_heap = 0;
 968   GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
 969   connected_peers = NULL;
 970   GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_size (query_request_map));
 971   GNUNET_CONTAINER_multihashmap_destroy (query_request_map);
 972   query_request_map = NULL;
 973   GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_size (peer_request_map));
 974   GNUNET_CONTAINER_multihashmap_destroy (peer_request_map);
 975   peer_request_map = NULL;
 976   GNUNET_assert (NULL != core);
 977   GNUNET_CORE_disconnect (core);
 978   core = NULL;
 979   if (stats != NULL)
 980     {
 981       GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
 982       stats = NULL;
 983     }
 984   sched = NULL;
 985   cfg = NULL;
 986 }
 987
 988
 989 /* ******************* Utility functions  ******************** */
 990
 991
 992 /**
 993  * Transmit the given message by copying it to the target buffer
 994  * "buf".  "buf" will be NULL and "size" zero if the socket was closed
 995  * for writing in the meantime.  In that case, do nothing
 996  * (the disconnect or shutdown handler will take care of the rest).
 997  * If we were able to transmit messages and there are still more
 998  * pending, ask core again for further calls to this function.
 999  *
1000  * @param cls closure, pointer to the 'struct ConnectedPeer*'
1001  * @param size number of bytes available in buf
1002  * @param buf where the callee should write the message
1003  * @return number of bytes written to buf
1004  */
1005 static size_t
1006 transmit_to_peer (void *cls,
1007                   size_t size, void *buf)
1008 {
1009   struct ConnectedPeer *cp = cls;
1010   char *cbuf = buf;
1011   struct GNUNET_PeerIdentity pid;
1012   struct PendingMessage *pm;
1013   size_t msize;
1014
1015   cp->cth = NULL;
1016   if (NULL == buf)
1017     {
1018 #if DEBUG_FS
1019       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1020                   "Dropping message, core too busy.\n");
1021 #endif
1022       return 0;
1023     }
1024   msize = 0;
1025   while ( (NULL != (pm = cp->pending_messages_head) ) &&
1026           (pm->msize <= size) )
1027     {
1028       memcpy (&cbuf[msize], &pm[1], pm->msize);
1029       msize += pm->msize;
1030       size -= pm->msize;
1031       destroy_pending_message (pm, cp->pid);
1032     }
1033   if (NULL != pm)
1034     {
1035       GNUNET_PEER_resolve (cp->pid,
1036                            &pid);
1037       cp->cth = GNUNET_CORE_notify_transmit_ready (core,
1038                                                    pm->priority,
1039                                                    GNUNET_CONSTANTS_SERVICE_TIMEOUT,
1040                                                    &pid,
1041                                                    pm->msize,
1042                                                    &transmit_to_peer,
1043                                                    cp);
1044     }
1045 #if DEBUG_FS > 2
1046   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1047               "Transmitting %u bytes to peer %u.\n",
1048               msize,
1049               cp->pid);
1050 #endif
1051   return msize;
1052 }
1053
1054
1055 /**
1056  * Add a message to the set of pending messages for the given peer.
1057  *
1058  * @param cp peer to send message to
1059  * @param pm message to queue
1060  * @param pr request on which behalf this message is being queued
1061  */
1062 static void
1063 add_to_pending_messages_for_peer (struct ConnectedPeer *cp,
1064                                   struct PendingMessage *pm,
1065                                   struct PendingRequest *pr)
1066 {
1067   struct PendingMessage *pos;
1068   struct PendingMessageList *pml;
1069   struct GNUNET_PeerIdentity pid;
1070
1071   GNUNET_assert (pm->next == NULL);
1072   GNUNET_assert (pm->pml == NULL);
1073   pml = GNUNET_malloc (sizeof (struct PendingMessageList));
1074   pml->req = pr;
1075   pml->target = cp;
1076   pml->pm = pm;
1077   pm->pml = pml;
1078   GNUNET_CONTAINER_DLL_insert (pr->pending_head,
1079                                pr->pending_tail,
1080                                pml);
1081   pos = cp->pending_messages_head;
1082   while ( (pos != NULL) &&
1083           (pm->priority < pos->priority) )
1084     pos = pos->next;
1085   GNUNET_CONTAINER_DLL_insert_after (cp->pending_messages_head,
1086                                      cp->pending_messages_tail,
1087                                      pos,
1088                                      pm);
1089   cp->pending_requests++;
1090   if (cp->pending_requests > MAX_QUEUE_PER_PEER)
1091     destroy_pending_message (cp->pending_messages_tail, 0);
1092   if (cp->cth == NULL)
1093     {
1094       /* need to schedule transmission */
1095       GNUNET_PEER_resolve (cp->pid, &pid);
1096       cp->cth = GNUNET_CORE_notify_transmit_ready (core,
1097                                                    cp->pending_messages_head->priority,
1098                                                    MAX_TRANSMIT_DELAY,
1099                                                    &pid,
1100                                                    cp->pending_messages_head->msize,
1101                                                    &transmit_to_peer,
1102                                                    cp);
1103     }
1104   if (cp->cth == NULL)
1105     {
1106 #if DEBUG_FS
1107       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1108                   "Failed to schedule transmission with core!\n");
1109 #endif
1110       /* FIXME: call stats (rare, bad case) */
1111     }
1112 }
1113
1114
1115 /**
1116  * Mingle hash with the mingle_number to produce different bits.
1117  */
1118 static void
1119 mingle_hash (const GNUNET_HashCode * in,
1120              int32_t mingle_number,
1121              GNUNET_HashCode * hc)
1122 {
1123   GNUNET_HashCode m;
1124
1125   GNUNET_CRYPTO_hash (&mingle_number,
1126                       sizeof (int32_t),
1127                       &m);
1128   GNUNET_CRYPTO_hash_xor (&m, in, hc);
1129 }
1130
1131
1132 /**
1133  * Test if the load on this peer is too high
1134  * to even consider processing the query at
1135  * all.
1136  *
1137  * @return GNUNET_YES if the load is too high, GNUNET_NO otherwise
1138  */
1139 static int
1140 test_load_too_high ()
1141 {
1142   return GNUNET_NO; // FIXME
1143 }
1144
1145
1146 /* ******************* Pending Request Refresh Task ******************** */
1147
1148
1149
1150 /**
1151  * We use a random delay to make the timing of requests less
1152  * predictable.  This function returns such a random delay.  We add a base
1153  * delay of MAX_CORK_DELAY (1s).
1154  *
1155  * FIXME: make schedule dependent on the specifics of the request?
1156  * Or bandwidth and number of connected peers and load?
1157  *
1158  * @return random delay to use for some request, between 1s and 1000+TTL_DECREMENT ms
1159  */
1160 static struct GNUNET_TIME_Relative
1161 get_processing_delay ()
1162 {
1163   return
1164     GNUNET_TIME_relative_add (GNUNET_CONSTANTS_MAX_CORK_DELAY,
1165                               GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
1166                                                              GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1167                                                                                        TTL_DECREMENT)));
1168 }
1169
1170
1171 /**
1172  * We're processing a GET request from another peer and have decided
1173  * to forward it to other peers.  This function is called periodically
1174  * and should forward the request to other peers until we have all
1175  * possible replies.  If we have transmitted the *only* reply to
1176  * the initiator we should destroy the pending request.  If we have
1177  * many replies in the queue to the initiator, we should delay sending
1178  * out more queries until the reply queue has shrunk some.
1179  *
1180  * @param cls our "struct ProcessGetContext *"
1181  * @param tc unused
1182  */
1183 static void
1184 forward_request_task (void *cls,
1185                       const struct GNUNET_SCHEDULER_TaskContext *tc);
1186
1187
1188 /**
1189  * Function called after we either failed or succeeded
1190  * at transmitting a query to a peer.
1191  *
1192  * @param cls the requests "struct PendingRequest*"
1193  * @param tpid ID of receiving peer, 0 on transmission error
1194  */
1195 static void
1196 transmit_query_continuation (void *cls,
1197                              GNUNET_PEER_Id tpid)
1198 {
1199   struct PendingRequest *pr = cls;
1200
1201   GNUNET_STATISTICS_update (stats,
1202                             gettext_noop ("# queries scheduled for forwarding"),
1203                             -1,
1204                             GNUNET_NO);
1205   if (tpid == 0)
1206     {
1207 #if DEBUG_FS
1208       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1209                   "Transmission of request failed, will try again later.\n");
1210 #endif
1211       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
1212         pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1213                                                  get_processing_delay (),
1214                                                  &forward_request_task,
1215                                                  pr);
1216       return;
1217     }
1218   GNUNET_STATISTICS_update (stats,
1219                             gettext_noop ("# queries forwarded"),
1220                             1,
1221                             GNUNET_NO);
1222   GNUNET_PEER_change_rc (tpid, 1);
1223   if (pr->used_pids_off == pr->used_pids_size)
1224     GNUNET_array_grow (pr->used_pids,
1225                        pr->used_pids_size,
1226                        pr->used_pids_size * 2 + 2);
1227   pr->used_pids[pr->used_pids_off++] = tpid;
1228   if (pr->task == GNUNET_SCHEDULER_NO_TASK)
1229     pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1230                                              get_processing_delay (),
1231                                              &forward_request_task,
1232                                              pr);
1233 }
1234
1235
1236 /**
1237  * How many bytes should a bloomfilter be if we have already seen
1238  * entry_count responses?  Note that BLOOMFILTER_K gives us the number
1239  * of bits set per entry.  Furthermore, we should not re-size the
1240  * filter too often (to keep it cheap).
1241  *
1242  * Since other peers will also add entries but not resize the filter,
1243  * we should generally pick a slightly larger size than what the
1244  * strict math would suggest.
1245  *
1246  * @return must be a power of two and smaller or equal to 2^15.
1247  */
1248 static size_t
1249 compute_bloomfilter_size (unsigned int entry_count)
1250 {
1251   size_t size;
1252   unsigned int ideal = (entry_count * BLOOMFILTER_K) / 4;
1253   uint16_t max = 1 << 15;
1254
1255   if (entry_count > max)
1256     return max;
1257   size = 8;
1258   while ((size < max) && (size < ideal))
1259     size *= 2;
1260   if (size > max)
1261     return max;
1262   return size;
1263 }
1264
1265
1266 /**
1267  * Recalculate our bloom filter for filtering replies.
1268  *
1269  * @param count number of entries we are filtering right now
1270  * @param mingle set to our new mingling value
1271  * @param bf_size set to the size of the bloomfilter
1272  * @param entries the entries to filter
1273  * @return updated bloomfilter, NULL for none
1274  */
1275 static struct GNUNET_CONTAINER_BloomFilter *
1276 refresh_bloomfilter (unsigned int count,
1277                      int32_t * mingle,
1278                      size_t *bf_size,
1279                      const GNUNET_HashCode *entries)
1280 {
1281   struct GNUNET_CONTAINER_BloomFilter *bf;
1282   size_t nsize;
1283   unsigned int i;
1284   GNUNET_HashCode mhash;
1285
1286   if (0 == count)
1287     return NULL;
1288   nsize = compute_bloomfilter_size (count);
1289   *mingle = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, -1);
1290   *bf_size = nsize;
1291   bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
1292                                           nsize,
1293                                           BLOOMFILTER_K);
1294   for (i=0;i<count;i++)
1295     {
1296       mingle_hash (&entries[i], *mingle, &mhash);
1297       GNUNET_CONTAINER_bloomfilter_add (bf, &mhash);
1298     }
1299   return bf;
1300 }
1301
1302
1303 /**
1304  * Function called after we've tried to reserve a certain amount of
1305  * bandwidth for a reply.  Check if we succeeded and if so send our
1306  * query.
1307  *
1308  * @param cls the requests "struct PendingRequest*"
1309  * @param peer identifies the peer
1310  * @param bpm_in set to the current bandwidth limit (receiving) for this peer
1311  * @param bpm_out set to the current bandwidth limit (sending) for this peer
1312  * @param amount set to the amount that was actually reserved or unreserved
1313  * @param preference current traffic preference for the given peer
1314  */
1315 static void
1316 target_reservation_cb (void *cls,
1317                        const struct
1318                        GNUNET_PeerIdentity * peer,
1319                        struct GNUNET_BANDWIDTH_Value32NBO bpm_in,
1320                        struct GNUNET_BANDWIDTH_Value32NBO bpm_out,
1321                        int amount,
1322                        uint64_t preference)
1323 {
1324   struct PendingRequest *pr = cls;
1325   struct ConnectedPeer *cp;
1326   struct PendingMessage *pm;
1327   struct GetMessage *gm;
1328   GNUNET_HashCode *ext;
1329   char *bfdata;
1330   size_t msize;
1331   unsigned int k;
1332   int no_route;
1333   uint32_t bm;
1334
1335   pr->irc = NULL;
1336   if (peer == NULL)
1337     {
1338       /* error in communication with core, try again later */
1339       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
1340         pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1341                                                  get_processing_delay (),
1342                                                  &forward_request_task,
1343                                                  pr);
1344       return;
1345     }
1346   // (3) transmit, update ttl/priority
1347   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
1348                                           &peer->hashPubKey);
1349   if (cp == NULL)
1350     {
1351       /* Peer must have just left */
1352 #if DEBUG_FS
1353       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1354                   "Selected peer disconnected!\n");
1355 #endif
1356       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
1357         pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1358                                                  get_processing_delay (),
1359                                                  &forward_request_task,
1360                                                  pr);
1361       return;
1362     }
1363   no_route = GNUNET_NO;
1364   /* FIXME: check against DBLOCK_SIZE and possibly return
1365      amount to reserve; however, this also needs to work
1366      with testcases which currently start out with a far
1367      too low per-peer bw limit, so they would never send
1368      anything.  Big issue. */
1369   if (amount == 0)
1370     {
1371       if (pr->cp == NULL)
1372         {
1373 #if DEBUG_FS > 1
1374           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1375                       "Failed to reserve bandwidth for reply (got %d/%u bytes only)!\n",
1376                       amount,
1377                       DBLOCK_SIZE);
1378 #endif
1379           GNUNET_STATISTICS_update (stats,
1380                                     gettext_noop ("# reply bandwidth reservation requests failed"),
1381                                     1,
1382                                     GNUNET_NO);
1383           if (pr->task == GNUNET_SCHEDULER_NO_TASK)
1384             pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1385                                                      get_processing_delay (),
1386                                                      &forward_request_task,
1387                                                      pr);
1388           return;  /* this target round failed */
1389         }
1390       /* FIXME: if we are "quite" busy, we may still want to skip
1391          this round; need more load detection code! */
1392       no_route = GNUNET_YES;
1393     }
1394
1395   GNUNET_STATISTICS_update (stats,
1396                             gettext_noop ("# queries scheduled for forwarding"),
1397                             1,
1398                             GNUNET_NO);
1399   /* build message and insert message into priority queue */
1400 #if DEBUG_FS
1401   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1402               "Forwarding request `%s' to `%4s'!\n",
1403               GNUNET_h2s (&pr->query),
1404               GNUNET_i2s (peer));
1405 #endif
1406   k = 0;
1407   bm = 0;
1408   if (GNUNET_YES == no_route)
1409     {
1410       bm |= GET_MESSAGE_BIT_RETURN_TO;
1411       k++;
1412     }
1413   if (pr->namespace != NULL)
1414     {
1415       bm |= GET_MESSAGE_BIT_SKS_NAMESPACE;
1416       k++;
1417     }
1418   if (pr->target_pid != 0)
1419     {
1420       bm |= GET_MESSAGE_BIT_TRANSMIT_TO;
1421       k++;
1422     }
1423   msize = sizeof (struct GetMessage) + pr->bf_size + k * sizeof(GNUNET_HashCode);
1424   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
1425   pm = GNUNET_malloc (sizeof (struct PendingMessage) + msize);
1426   pm->msize = msize;
1427   gm = (struct GetMessage*) &pm[1];
1428   gm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_GET);
1429   gm->header.size = htons (msize);
1430   gm->type = htonl (pr->type);
1431   pr->remaining_priority /= 2;
1432   gm->priority = htonl (pr->remaining_priority);
1433   gm->ttl = htonl (pr->ttl);
1434   gm->filter_mutator = htonl(pr->mingle);
1435   gm->hash_bitmap = htonl (bm);
1436   gm->query = pr->query;
1437   ext = (GNUNET_HashCode*) &gm[1];
1438   k = 0;
1439   if (GNUNET_YES == no_route)
1440     GNUNET_PEER_resolve (pr->cp->pid, (struct GNUNET_PeerIdentity*) &ext[k++]);
1441   if (pr->namespace != NULL)
1442     memcpy (&ext[k++], pr->namespace, sizeof (GNUNET_HashCode));
1443   if (pr->target_pid != 0)
1444     GNUNET_PEER_resolve (pr->target_pid, (struct GNUNET_PeerIdentity*) &ext[k++]);
1445   bfdata = (char *) &ext[k];
1446   if (pr->bf != NULL)
1447     GNUNET_CONTAINER_bloomfilter_get_raw_data (pr->bf,
1448                                                bfdata,
1449                                                pr->bf_size);
1450   pm->cont = &transmit_query_continuation;
1451   pm->cont_cls = pr;
1452   add_to_pending_messages_for_peer (cp, pm, pr);
1453 }
1454
1455
1456 /**
1457  * Closure used for "target_peer_select_cb".
1458  */
1459 struct PeerSelectionContext
1460 {
1461   /**
1462    * The request for which we are selecting
1463    * peers.
1464    */
1465   struct PendingRequest *pr;
1466
1467   /**
1468    * Current "prime" target.
1469    */
1470   struct GNUNET_PeerIdentity target;
1471
1472   /**
1473    * How much do we like this target?
1474    */
1475   double target_score;
1476
1477 };
1478
1479
1480 /**
1481  * Function called for each connected peer to determine
1482  * which one(s) would make good targets for forwarding.
1483  *
1484  * @param cls closure (struct PeerSelectionContext)
1485  * @param key current key code (peer identity)
1486  * @param value value in the hash map (struct ConnectedPeer)
1487  * @return GNUNET_YES if we should continue to
1488  *         iterate,
1489  *         GNUNET_NO if not.
1490  */
1491 static int
1492 target_peer_select_cb (void *cls,
1493                        const GNUNET_HashCode * key,
1494                        void *value)
1495 {
1496   struct PeerSelectionContext *psc = cls;
1497   struct ConnectedPeer *cp = value;
1498   struct PendingRequest *pr = psc->pr;
1499   double score;
1500   unsigned int i;
1501   unsigned int pc;
1502
1503   /* 1) check if we have already (recently) forwarded to this peer */
1504   pc = 0;
1505   for (i=0;i<pr->used_pids_off;i++)
1506     if (pr->used_pids[i] == cp->pid)
1507       {
1508         pc++;
1509         if (0 != GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1510                                            RETRY_PROBABILITY_INV))
1511           {
1512 #if DEBUG_FS
1513             GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1514                         "NOT re-trying query that was previously transmitted %u times\n",
1515                         (unsigned int) pr->used_pids_off);
1516 #endif
1517             return GNUNET_YES; /* skip */
1518           }
1519       }
1520 #if DEBUG_FS
1521   if (0 < pc)
1522     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1523                 "Re-trying query that was previously transmitted %u times to this peer\n",
1524                 (unsigned int) pc);
1525 #endif
1526   // 2) calculate how much we'd like to forward to this peer
1527   score = 42; // FIXME!
1528   // FIXME: also need API to gather data on responsiveness
1529   // of this peer (we have fields for that in 'cp', but
1530   // they are never set!)
1531
1532   /* store best-fit in closure */
1533   if (score > psc->target_score)
1534     {
1535       psc->target_score = score;
1536       psc->target.hashPubKey = *key;
1537     }
1538   return GNUNET_YES;
1539 }
1540
1541
1542 /**
1543  * The priority level imposes a bound on the maximum
1544  * value for the ttl that can be requested.
1545  *
1546  * @param ttl_in requested ttl
1547  * @param prio given priority
1548  * @return ttl_in if ttl_in is below the limit,
1549  *         otherwise the ttl-limit for the given priority
1550  */
1551 static int32_t
1552 bound_ttl (int32_t ttl_in, uint32_t prio)
1553 {
1554   unsigned long long allowed;
1555
1556   if (ttl_in <= 0)
1557     return ttl_in;
1558   allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000;
1559   if (ttl_in > allowed)
1560     {
1561       if (allowed >= (1 << 30))
1562         return 1 << 30;
1563       return allowed;
1564     }
1565   return ttl_in;
1566 }
1567
1568
1569 /**
1570  * We're processing a GET request from another peer and have decided
1571  * to forward it to other peers.  This function is called periodically
1572  * and should forward the request to other peers until we have all
1573  * possible replies.  If we have transmitted the *only* reply to
1574  * the initiator we should destroy the pending request.  If we have
1575  * many replies in the queue to the initiator, we should delay sending
1576  * out more queries until the reply queue has shrunk some.
1577  *
1578  * @param cls our "struct ProcessGetContext *"
1579  * @param tc unused
1580  */
1581 static void
1582 forward_request_task (void *cls,
1583                      const struct GNUNET_SCHEDULER_TaskContext *tc)
1584 {
1585   struct PendingRequest *pr = cls;
1586   struct PeerSelectionContext psc;
1587   struct ConnectedPeer *cp;
1588   struct GNUNET_TIME_Relative delay;
1589
1590   pr->task = GNUNET_SCHEDULER_NO_TASK;
1591   if (pr->irc != NULL)
1592     {
1593 #if DEBUG_FS
1594       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1595                   "Forwarding of query `%s' not attempted due to pending local lookup!\n",
1596                   GNUNET_h2s (&pr->query));
1597 #endif
1598       return; /* already pending */
1599     }
1600   /* (1) select target */
1601   psc.pr = pr;
1602   psc.target_score = DBL_MIN;
1603   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1604                                          &target_peer_select_cb,
1605                                          &psc);
1606   if (psc.target_score == DBL_MIN)
1607     {
1608       delay = get_processing_delay ();
1609 #if DEBUG_FS
1610       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1611                   "No peer selected for forwarding of query `%s', will try again in %llu ms!\n",
1612                   GNUNET_h2s (&pr->query),
1613                   delay.value);
1614 #endif
1615       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1616                                                delay,
1617                                                &forward_request_task,
1618                                                pr);
1619       return; /* nobody selected */
1620     }
1621   /* (3) update TTL/priority */
1622
1623   if (pr->client_request_list != NULL)
1624     {
1625       /* FIXME: use better algorithm!? */
1626       if (0 == GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1627                                          4))
1628         pr->priority++;
1629       /* FIXME: bound priority by "customary" priority used by other peers
1630          at this time! */
1631       pr->ttl = bound_ttl (pr->ttl + TTL_DECREMENT * 2,
1632                            pr->priority);
1633       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1634                   "Trying query `%s' with priority %u and TTL %d.\n",
1635                   GNUNET_h2s (&pr->query),
1636                   pr->priority,
1637                   pr->ttl);
1638     }
1639   else
1640     {
1641       /* FIXME: should we do something here as well!? */
1642     }
1643
1644   /* (3) reserve reply bandwidth */
1645   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
1646                                           &psc.target.hashPubKey);
1647   pr->irc = GNUNET_CORE_peer_change_preference (sched, cfg,
1648                                                 &psc.target,
1649                                                 GNUNET_CONSTANTS_SERVICE_TIMEOUT,
1650                                                 GNUNET_BANDWIDTH_value_init ((uint32_t) -1 /* no limit */),
1651                                                 DBLOCK_SIZE * 2,
1652                                                 (uint64_t) cp->inc_preference,
1653                                                 &target_reservation_cb,
1654                                                 pr);
1655   cp->inc_preference = 0.0;
1656 }
1657
1658
1659 /* **************************** P2P PUT Handling ************************ */
1660
1661
1662 /**
1663  * Function called after we either failed or succeeded
1664  * at transmitting a reply to a peer.
1665  *
1666  * @param cls the requests "struct PendingRequest*"
1667  * @param tpid ID of receiving peer, 0 on transmission error
1668  */
1669 static void
1670 transmit_reply_continuation (void *cls,
1671                              GNUNET_PEER_Id tpid)
1672 {
1673   struct PendingRequest *pr = cls;
1674
1675   switch (pr->type)
1676     {
1677     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
1678     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
1679       /* only one reply expected, done with the request! */
1680       destroy_pending_request (pr);
1681       break;
1682     case GNUNET_DATASTORE_BLOCKTYPE_ANY:
1683     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
1684     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
1685       break;
1686     default:
1687       GNUNET_break (0);
1688       break;
1689     }
1690 }
1691
1692
1693 /**
1694  * Check if the given KBlock is well-formed.
1695  *
1696  * @param kb the kblock data (or at least "dsize" bytes claiming to be one)
1697  * @param dsize size of "kb" in bytes; check for < sizeof(struct KBlock)!
1698  * @param query where to store the query that this block answers
1699  * @return GNUNET_OK if this is actually a well-formed KBlock
1700  */
1701 static int
1702 check_kblock (const struct KBlock *kb,
1703               size_t dsize,
1704               GNUNET_HashCode *query)
1705 {
1706   if (dsize < sizeof (struct KBlock))
1707     {
1708       GNUNET_break_op (0);
1709       return GNUNET_SYSERR;
1710     }
1711   if (dsize - sizeof (struct KBlock) !=
1712       ntohl (kb->purpose.size)
1713       - sizeof (struct GNUNET_CRYPTO_RsaSignaturePurpose)
1714       - sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) )
1715     {
1716       GNUNET_break_op (0);
1717       return GNUNET_SYSERR;
1718     }
1719   if (GNUNET_OK !=
1720       GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_KBLOCK,
1721                                 &kb->purpose,
1722                                 &kb->signature,
1723                                 &kb->keyspace))
1724     {
1725       GNUNET_break_op (0);
1726       return GNUNET_SYSERR;
1727     }
1728   if (query != NULL)
1729     GNUNET_CRYPTO_hash (&kb->keyspace,
1730                         sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
1731                         query);
1732   return GNUNET_OK;
1733 }
1734
1735
1736 /**
1737  * Check if the given NBlock is well-formed.
1738  *
1739  * @param nb the nblock data (or at least "dsize" bytes claiming to be one)
1740  * @param dsize size of "nb" in bytes; check for < sizeof(struct NBlock)!
1741  * @param query where to store the query that this block answers
1742  * @return GNUNET_OK if this is actually a well-formed NBlock
1743  */
1744 static int
1745 check_nblock (const struct NBlock *nb,
1746               size_t dsize,
1747               GNUNET_HashCode *query)
1748 {
1749   if (dsize < sizeof (struct NBlock))
1750     {
1751       GNUNET_break_op (0);
1752       return GNUNET_SYSERR;
1753     }
1754   if (dsize - sizeof (struct NBlock) !=
1755       ntohl (nb->ns_purpose.size)
1756       - sizeof (struct GNUNET_CRYPTO_RsaSignaturePurpose)
1757       - sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) )
1758     {
1759       GNUNET_break_op (0);
1760       return GNUNET_SYSERR;
1761     }
1762   if (dsize !=
1763       ntohl (nb->ksk_purpose.size) + sizeof (struct GNUNET_CRYPTO_RsaSignature))
1764     {
1765       GNUNET_break_op (0);
1766       return GNUNET_SYSERR;
1767     }
1768   if (GNUNET_OK !=
1769       GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_NBLOCK_KSIG,
1770                                 &nb->ksk_purpose,
1771                                 &nb->ksk_signature,
1772                                 &nb->keyspace))
1773     {
1774       GNUNET_break_op (0);
1775       return GNUNET_SYSERR;
1776     }
1777   if (GNUNET_OK !=
1778       GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_NBLOCK,
1779                                 &nb->ns_purpose,
1780                                 &nb->ns_signature,
1781                                 &nb->subspace))
1782     {
1783       GNUNET_break_op (0);
1784       return GNUNET_SYSERR;
1785     }
1786   if (query != NULL)
1787     GNUNET_CRYPTO_hash (&nb->keyspace,
1788                         sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
1789                         query);
1790   return GNUNET_OK;
1791 }
1792
1793
1794 /**
1795  * Check if the given SBlock is well-formed.
1796  *
1797  * @param sb the sblock data (or at least "dsize" bytes claiming to be one)
1798  * @param dsize size of "kb" in bytes; check for < sizeof(struct SBlock)!
1799  * @param query where to store the query that this block answers
1800  * @param namespace where to store the namespace that this block belongs to
1801  * @return GNUNET_OK if this is actually a well-formed SBlock
1802  */
1803 static int
1804 check_sblock (const struct SBlock *sb,
1805               size_t dsize,
1806               GNUNET_HashCode *query,
1807               GNUNET_HashCode *namespace)
1808 {
1809   if (dsize < sizeof (struct SBlock))
1810     {
1811       GNUNET_break_op (0);
1812       return GNUNET_SYSERR;
1813     }
1814   if (dsize !=
1815       ntohl (sb->purpose.size) + sizeof (struct GNUNET_CRYPTO_RsaSignature))
1816     {
1817       GNUNET_break_op (0);
1818       return GNUNET_SYSERR;
1819     }
1820   if (GNUNET_OK !=
1821       GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_SBLOCK,
1822                                 &sb->purpose,
1823                                 &sb->signature,
1824                                 &sb->subspace))
1825     {
1826       GNUNET_break_op (0);
1827       return GNUNET_SYSERR;
1828     }
1829   if (query != NULL)
1830     *query = sb->identifier;
1831   if (namespace != NULL)
1832     GNUNET_CRYPTO_hash (&sb->subspace,
1833                         sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
1834                         namespace);
1835   return GNUNET_OK;
1836 }
1837
1838
1839 /**
1840  * Transmit the given message by copying it to the target buffer
1841  * "buf".  "buf" will be NULL and "size" zero if the socket was closed
1842  * for writing in the meantime.  In that case, do nothing
1843  * (the disconnect or shutdown handler will take care of the rest).
1844  * If we were able to transmit messages and there are still more
1845  * pending, ask core again for further calls to this function.
1846  *
1847  * @param cls closure, pointer to the 'struct ClientList*'
1848  * @param size number of bytes available in buf
1849  * @param buf where the callee should write the message
1850  * @return number of bytes written to buf
1851  */
1852 static size_t
1853 transmit_to_client (void *cls,
1854                   size_t size, void *buf)
1855 {
1856   struct ClientList *cl = cls;
1857   char *cbuf = buf;
1858   struct ClientResponseMessage *creply;
1859   size_t msize;
1860
1861   cl->th = NULL;
1862   if (NULL == buf)
1863     {
1864 #if DEBUG_FS
1865       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1866                   "Not sending reply, client communication problem.\n");
1867 #endif
1868       return 0;
1869     }
1870   msize = 0;
1871   while ( (NULL != (creply = cl->res_head) ) &&
1872           (creply->msize <= size) )
1873     {
1874       memcpy (&cbuf[msize], &creply[1], creply->msize);
1875       msize += creply->msize;
1876       size -= creply->msize;
1877       GNUNET_CONTAINER_DLL_remove (cl->res_head,
1878                                    cl->res_tail,
1879                                    creply);
1880       GNUNET_free (creply);
1881     }
1882   if (NULL != creply)
1883     cl->th = GNUNET_SERVER_notify_transmit_ready (cl->client,
1884                                                   creply->msize,
1885                                                   GNUNET_TIME_UNIT_FOREVER_REL,
1886                                                   &transmit_to_client,
1887                                                   cl);
1888 #if DEBUG_FS
1889   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1890               "Transmitted %u bytes to client\n",
1891               (unsigned int) msize);
1892 #endif
1893   return msize;
1894 }
1895
1896
1897 /**
1898  * Closure for "process_reply" function.
1899  */
1900 struct ProcessReplyClosure
1901 {
1902   /**
1903    * The data for the reply.
1904    */
1905   const void *data;
1906
1907   // FIXME: add 'struct ConnectedPeer' to track 'last_xxx_replies' here!
1908
1909   /**
1910    * When the reply expires.
1911    */
1912   struct GNUNET_TIME_Absolute expiration;
1913
1914   /**
1915    * Size of data.
1916    */
1917   size_t size;
1918
1919   /**
1920    * Namespace that this reply belongs to
1921    * (if it is of type SBLOCK).
1922    */
1923   GNUNET_HashCode namespace;
1924
1925   /**
1926    * Type of the block.
1927    */
1928   uint32_t type;
1929
1930   /**
1931    * How much was this reply worth to us?
1932    */
1933   uint32_t priority;
1934 };
1935
1936
1937 /**
1938  * We have received a reply; handle it!
1939  *
1940  * @param cls response (struct ProcessReplyClosure)
1941  * @param key our query
1942  * @param value value in the hash map (info about the query)
1943  * @return GNUNET_YES (we should continue to iterate)
1944  */
1945 static int
1946 process_reply (void *cls,
1947                const GNUNET_HashCode * key,
1948                void *value)
1949 {
1950   struct ProcessReplyClosure *prq = cls;
1951   struct PendingRequest *pr = value;
1952   struct PendingMessage *reply;
1953   struct ClientResponseMessage *creply;
1954   struct ClientList *cl;
1955   struct PutMessage *pm;
1956   struct ConnectedPeer *cp;
1957   GNUNET_HashCode chash;
1958   GNUNET_HashCode mhash;
1959   size_t msize;
1960
1961 #if DEBUG_FS
1962   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1963               "Matched result (type %u) for query `%s' with pending request\n",
1964               (unsigned int) prq->type,
1965               GNUNET_h2s (key));
1966 #endif
1967   GNUNET_STATISTICS_update (stats,
1968                             gettext_noop ("# replies received and matched"),
1969                             1,
1970                             GNUNET_NO);
1971   GNUNET_CRYPTO_hash (prq->data,
1972                       prq->size,
1973                       &chash);
1974   switch (prq->type)
1975     {
1976     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
1977     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
1978       /* only possible reply, stop requesting! */
1979       while (NULL != pr->pending_head)
1980         destroy_pending_message_list_entry (pr->pending_head);
1981       if (pr->drq != NULL)
1982         {
1983           if (pr->client_request_list != NULL)
1984             GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
1985                                         GNUNET_YES);
1986           GNUNET_FS_drq_get_cancel (pr->drq);
1987           pr->drq = NULL;
1988         }
1989       pr->do_remove = GNUNET_YES;
1990       if (pr->task != GNUNET_SCHEDULER_NO_TASK)
1991         {
1992           GNUNET_SCHEDULER_cancel (sched,
1993                                    pr->task);
1994           pr->task = GNUNET_SCHEDULER_NO_TASK;
1995         }
1996       GNUNET_break (GNUNET_YES ==
1997                     GNUNET_CONTAINER_multihashmap_remove (query_request_map,
1998                                                           key,
1999                                                           pr));
2000       break;
2001     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
2002       if (pr->namespace == NULL)
2003         {
2004           GNUNET_break (0);
2005           return GNUNET_YES;
2006         }
2007       if (0 != memcmp (pr->namespace,
2008                        &prq->namespace,
2009                        sizeof (GNUNET_HashCode)))
2010         {
2011           GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
2012                       _("Reply mismatched in terms of namespace.  Discarded.\n"));
2013           return GNUNET_YES; /* wrong namespace */
2014         }
2015       /* then: fall-through! */
2016     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
2017     case GNUNET_DATASTORE_BLOCKTYPE_NBLOCK:
2018       if (pr->bf != NULL)
2019         {
2020           mingle_hash (&chash, pr->mingle, &mhash);
2021           if (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (pr->bf,
2022                                                                &mhash))
2023             {
2024               GNUNET_STATISTICS_update (stats,
2025                                         gettext_noop ("# duplicate replies discarded (bloomfilter)"),
2026                                         1,
2027                                         GNUNET_NO);
2028 #if DEBUG_FS
2029               GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2030                           "Duplicate response `%s', discarding.\n",
2031                           GNUNET_h2s (&mhash));
2032 #endif
2033               return GNUNET_YES; /* duplicate */
2034             }
2035 #if DEBUG_FS
2036           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2037                       "New response `%s', adding to filter.\n",
2038                       GNUNET_h2s (&mhash));
2039 #endif
2040           GNUNET_CONTAINER_bloomfilter_add (pr->bf,
2041                                             &mhash);
2042         }
2043       if (pr->client_request_list != NULL)
2044         {
2045           if (pr->replies_seen_size == pr->replies_seen_off)
2046             {
2047               GNUNET_array_grow (pr->replies_seen,
2048                                  pr->replies_seen_size,
2049                                  pr->replies_seen_size * 2 + 4);
2050               if (pr->bf != NULL)
2051                 GNUNET_CONTAINER_bloomfilter_free (pr->bf);
2052               pr->bf = refresh_bloomfilter (pr->replies_seen_off,
2053                                             &pr->mingle,
2054                                             &pr->bf_size,
2055                                             pr->replies_seen);
2056             }
2057             pr->replies_seen[pr->replies_seen_off++] = chash;
2058         }
2059       break;
2060     default:
2061       GNUNET_break (0);
2062       return GNUNET_YES;
2063     }
2064   prq->priority += pr->remaining_priority;
2065   pr->remaining_priority = 0;
2066   if (pr->client_request_list != NULL)
2067     {
2068       GNUNET_STATISTICS_update (stats,
2069                                 gettext_noop ("# replies received for local clients"),
2070                                 1,
2071                                 GNUNET_NO);
2072       cl = pr->client_request_list->client_list;
2073       msize = sizeof (struct PutMessage) + prq->size;
2074       creply = GNUNET_malloc (msize + sizeof (struct ClientResponseMessage));
2075       creply->msize = msize;
2076       creply->client_list = cl;
2077       GNUNET_CONTAINER_DLL_insert_after (cl->res_head,
2078                                          cl->res_tail,
2079                                          cl->res_tail,
2080                                          creply);
2081       pm = (struct PutMessage*) &creply[1];
2082       pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
2083       pm->header.size = htons (msize);
2084       pm->type = htonl (prq->type);
2085       pm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
2086       memcpy (&pm[1], prq->data, prq->size);
2087       if (NULL == cl->th)
2088         {
2089 #if DEBUG_FS
2090           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2091                       "Transmitting result for query `%s' to client\n",
2092                       GNUNET_h2s (key));
2093 #endif
2094           cl->th = GNUNET_SERVER_notify_transmit_ready (cl->client,
2095                                                         msize,
2096                                                         GNUNET_TIME_UNIT_FOREVER_REL,
2097                                                         &transmit_to_client,
2098                                                         cl);
2099         }
2100       GNUNET_break (cl->th != NULL);
2101       if (pr->do_remove)
2102         destroy_pending_request (pr);
2103     }
2104   else
2105     {
2106       cp = pr->cp;
2107 #if DEBUG_FS
2108       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2109                   "Transmitting result for query `%s' to other peer (PID=%u)\n",
2110                   GNUNET_h2s (key),
2111                   (unsigned int) cp->pid);
2112 #endif
2113       GNUNET_STATISTICS_update (stats,
2114                                 gettext_noop ("# replies received for other peers"),
2115                                 1,
2116                                 GNUNET_NO);
2117       msize = sizeof (struct PutMessage) + prq->size;
2118       reply = GNUNET_malloc (msize + sizeof (struct PendingMessage));
2119       reply->cont = &transmit_reply_continuation;
2120       reply->cont_cls = pr;
2121       reply->msize = msize;
2122       reply->priority = (uint32_t) -1; /* send replies first! */
2123       pm = (struct PutMessage*) &reply[1];
2124       pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
2125       pm->header.size = htons (msize);
2126       pm->type = htonl (prq->type);
2127       pm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
2128       memcpy (&pm[1], prq->data, prq->size);
2129       add_to_pending_messages_for_peer (cp, reply, pr);
2130     }
2131   // FIXME: implement hot-path routing statistics keeping!
2132   return GNUNET_YES;
2133 }
2134
2135
2136 /**
2137  * Handle P2P "PUT" message.
2138  *
2139  * @param cls closure, always NULL
2140  * @param other the other peer involved (sender or receiver, NULL
2141  *        for loopback messages where we are both sender and receiver)
2142  * @param message the actual message
2143  * @param latency reported latency of the connection with 'other'
2144  * @param distance reported distance (DV) to 'other'
2145  * @return GNUNET_OK to keep the connection open,
2146  *         GNUNET_SYSERR to close it (signal serious error)
2147  */
2148 static int
2149 handle_p2p_put (void *cls,
2150                 const struct GNUNET_PeerIdentity *other,
2151                 const struct GNUNET_MessageHeader *message,
2152                 struct GNUNET_TIME_Relative latency,
2153                 uint32_t distance)
2154 {
2155   const struct PutMessage *put;
2156   uint16_t msize;
2157   size_t dsize;
2158   uint32_t type;
2159   struct GNUNET_TIME_Absolute expiration;
2160   GNUNET_HashCode query;
2161   struct ProcessReplyClosure prq;
2162
2163   msize = ntohs (message->size);
2164   if (msize < sizeof (struct PutMessage))
2165     {
2166       GNUNET_break_op(0);
2167       return GNUNET_SYSERR;
2168     }
2169   put = (const struct PutMessage*) message;
2170   dsize = msize - sizeof (struct PutMessage);
2171   type = ntohl (put->type);
2172   expiration = GNUNET_TIME_absolute_ntoh (put->expiration);
2173
2174   /* first, validate! */
2175   switch (type)
2176     {
2177     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2178     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2179       GNUNET_CRYPTO_hash (&put[1], dsize, &query);
2180       break;
2181     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
2182       if (GNUNET_OK !=
2183           check_kblock ((const struct KBlock*) &put[1],
2184                         dsize,
2185                         &query))
2186         return GNUNET_SYSERR;
2187       break;
2188     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
2189       if (GNUNET_OK !=
2190           check_sblock ((const struct SBlock*) &put[1],
2191                         dsize,
2192                         &query,
2193                         &prq.namespace))
2194         return GNUNET_SYSERR;
2195       break;
2196     case GNUNET_DATASTORE_BLOCKTYPE_NBLOCK:
2197       if (GNUNET_OK !=
2198           check_nblock ((const struct NBlock*) &put[1],
2199                         dsize,
2200                         &query))
2201         return GNUNET_SYSERR;
2202       return GNUNET_OK;
2203     default:
2204       /* unknown block type */
2205       GNUNET_break_op (0);
2206       return GNUNET_SYSERR;
2207     }
2208
2209 #if DEBUG_FS
2210   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2211               "Received result for query `%s' from peer `%4s'\n",
2212               GNUNET_h2s (&query),
2213               GNUNET_i2s (other));
2214 #endif
2215   GNUNET_STATISTICS_update (stats,
2216                             gettext_noop ("# replies received (overall)"),
2217                             1,
2218                             GNUNET_NO);
2219   /* now, lookup 'query' */
2220   prq.data = (const void*) &put[1];
2221   prq.size = dsize;
2222   prq.type = type;
2223   prq.expiration = expiration;
2224   prq.priority = 0;
2225   GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
2226                                               &query,
2227                                               &process_reply,
2228                                               &prq);
2229   // FIXME: if migration is on and load is low,
2230   // queue to store data in datastore;
2231   // use "prq.priority" for that!
2232   return GNUNET_OK;
2233 }
2234
2235
2236 /* **************************** P2P GET Handling ************************ */
2237
2238
2239 /**
2240  * Closure for 'check_duplicate_request_{peer,client}'.
2241  */
2242 struct CheckDuplicateRequestClosure
2243 {
2244   /**
2245    * The new request we should check if it already exists.
2246    */
2247   const struct PendingRequest *pr;
2248
2249   /**
2250    * Existing request found by the checker, NULL if none.
2251    */
2252   struct PendingRequest *have;
2253 };
2254
2255
2256 /**
2257  * Iterator over entries in the 'query_request_map' that
2258  * tries to see if we have the same request pending from
2259  * the same client already.
2260  *
2261  * @param cls closure (our 'struct CheckDuplicateRequestClosure')
2262  * @param key current key code (query, ignored, must match)
2263  * @param value value in the hash map (a 'struct PendingRequest'
2264  *              that already exists)
2265  * @return GNUNET_YES if we should continue to
2266  *         iterate (no match yet)
2267  *         GNUNET_NO if not (match found).
2268  */
2269 static int
2270 check_duplicate_request_client (void *cls,
2271                                 const GNUNET_HashCode * key,
2272                                 void *value)
2273 {
2274   struct CheckDuplicateRequestClosure *cdc = cls;
2275   struct PendingRequest *have = value;
2276
2277   if (have->client_request_list == NULL)
2278     return GNUNET_YES;
2279   if ( (cdc->pr->client_request_list->client_list->client == have->client_request_list->client_list->client) &&
2280        (cdc->pr != have) )
2281     {
2282       cdc->have = have;
2283       return GNUNET_NO;
2284     }
2285   return GNUNET_YES;
2286 }
2287
2288
2289 /**
2290  * We're processing (local) results for a search request
2291  * from another peer.  Pass applicable results to the
2292  * peer and if we are done either clean up (operation
2293  * complete) or forward to other peers (more results possible).
2294  *
2295  * @param cls our closure (struct LocalGetContext)
2296  * @param key key for the content
2297  * @param size number of bytes in data
2298  * @param data content stored
2299  * @param type type of the content
2300  * @param priority priority of the content
2301  * @param anonymity anonymity-level for the content
2302  * @param expiration expiration time for the content
2303  * @param uid unique identifier for the datum;
2304  *        maybe 0 if no unique identifier is available
2305  */
2306 static void
2307 process_local_reply (void *cls,
2308                      const GNUNET_HashCode * key,
2309                      uint32_t size,
2310                      const void *data,
2311                      uint32_t type,
2312                      uint32_t priority,
2313                      uint32_t anonymity,
2314                      struct GNUNET_TIME_Absolute
2315                      expiration,
2316                      uint64_t uid)
2317 {
2318   struct PendingRequest *pr = cls;
2319   struct ProcessReplyClosure prq;
2320   struct CheckDuplicateRequestClosure cdrc;
2321   GNUNET_HashCode dhash;
2322   GNUNET_HashCode mhash;
2323   GNUNET_HashCode query;
2324
2325   if (NULL == key)
2326     {
2327 #if DEBUG_FS > 1
2328       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2329                   "Done processing local replies, forwarding request to other peers.\n");
2330 #endif
2331       pr->drq = NULL;
2332       if (pr->client_request_list != NULL)
2333         {
2334           GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
2335                                       GNUNET_YES);
2336           /* Figure out if this is a duplicate request and possibly
2337              merge 'struct PendingRequest' entries */
2338           cdrc.have = NULL;
2339           cdrc.pr = pr;
2340           GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
2341                                                       &pr->query,
2342                                                       &check_duplicate_request_client,
2343                                                       &cdrc);
2344           if (cdrc.have != NULL)
2345             {
2346 #if DEBUG_FS
2347               GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2348                           "Received request for block `%s' twice from client, will only request once.\n",
2349                           GNUNET_h2s (&pr->query));
2350 #endif
2351
2352               destroy_pending_request (pr);
2353               return;
2354             }
2355         }
2356
2357       /* no more results */
2358       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
2359         pr->task = GNUNET_SCHEDULER_add_now (sched,
2360                                              &forward_request_task,
2361                                              pr);
2362       return;
2363     }
2364 #if DEBUG_FS
2365   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2366               "New local response to `%s' of type %u.\n",
2367               GNUNET_h2s (key),
2368               type);
2369 #endif
2370   if (type == GNUNET_DATASTORE_BLOCKTYPE_ONDEMAND)
2371     {
2372 #if DEBUG_FS
2373       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2374                   "Found ONDEMAND block, performing on-demand encoding\n");
2375 #endif
2376       GNUNET_STATISTICS_update (stats,
2377                                 gettext_noop ("# on-demand blocks matched requests"),
2378                                 1,
2379                                 GNUNET_NO);
2380       if (GNUNET_OK !=
2381           GNUNET_FS_handle_on_demand_block (key, size, data, type, priority,
2382                                             anonymity, expiration, uid,
2383                                             &process_local_reply,
2384                                             pr))
2385         GNUNET_FS_drq_get_next (GNUNET_YES);
2386       return;
2387     }
2388   /* check for duplicates */
2389   GNUNET_CRYPTO_hash (data, size, &dhash);
2390   mingle_hash (&dhash,
2391                pr->mingle,
2392                &mhash);
2393   if ( (pr->bf != NULL) &&
2394        (GNUNET_YES ==
2395         GNUNET_CONTAINER_bloomfilter_test (pr->bf,
2396                                            &mhash)) )
2397     {
2398 #if DEBUG_FS
2399       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2400                   "Result from datastore filtered by bloomfilter (duplicate).\n");
2401 #endif
2402       GNUNET_STATISTICS_update (stats,
2403                                 gettext_noop ("# results filtered by query bloomfilter"),
2404                                 1,
2405                                 GNUNET_NO);
2406       GNUNET_FS_drq_get_next (GNUNET_YES);
2407       return;
2408     }
2409 #if DEBUG_FS
2410   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2411               "Found result for query `%s' in local datastore\n",
2412               GNUNET_h2s (key));
2413 #endif
2414   GNUNET_STATISTICS_update (stats,
2415                             gettext_noop ("# results found locally"),
2416                             1,
2417                             GNUNET_NO);
2418   pr->results_found++;
2419   if ( (pr->type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
2420        (pr->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ||
2421        (pr->type == GNUNET_DATASTORE_BLOCKTYPE_NBLOCK) )
2422     {
2423       if (pr->bf == NULL)
2424         {
2425           pr->bf_size = 32;
2426           pr->bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
2427                                                       pr->bf_size,
2428                                                       BLOOMFILTER_K);
2429         }
2430 #if DEBUG_FS
2431       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2432                   "New local response `%s', adding to filter.\n",
2433                   GNUNET_h2s (&mhash));
2434 #endif
2435 #if 0
2436       /* this would break stuff since we will check the bf later
2437          again (and would then discard the reply!) */
2438       GNUNET_CONTAINER_bloomfilter_add (pr->bf,
2439                                         &mhash);
2440 #endif
2441     }
2442   memset (&prq, 0, sizeof (prq));
2443   prq.data = data;
2444   prq.expiration = expiration;
2445   prq.size = size;
2446   if ( (type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) &&
2447        (GNUNET_OK != check_sblock ((const struct SBlock*) data,
2448                                    size,
2449                                    &query,
2450                                    &prq.namespace)) )
2451     {
2452       GNUNET_break (0);
2453       /* FIXME: consider removing the block? */
2454       GNUNET_FS_drq_get_next (GNUNET_YES);
2455       return;
2456     }
2457   prq.type = type;
2458   prq.priority = priority;
2459   process_reply (&prq, key, pr);
2460
2461   if ( (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK) ||
2462        (type == GNUNET_DATASTORE_BLOCKTYPE_IBLOCK) )
2463     {
2464       GNUNET_FS_drq_get_next (GNUNET_NO);
2465       return;
2466     }
2467   if ( (pr->client_request_list == NULL) &&
2468        ( (GNUNET_YES == test_load_too_high()) ||
2469          (pr->results_found > 5 + 2 * pr->priority) ) )
2470     {
2471 #if DEBUG_FS > 2
2472       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2473                   "Load too high, done with request\n");
2474 #endif
2475       GNUNET_STATISTICS_update (stats,
2476                                 gettext_noop ("# processing result set cut short due to load"),
2477                                 1,
2478                                 GNUNET_NO);
2479       GNUNET_FS_drq_get_next (GNUNET_NO);
2480       return;
2481     }
2482   GNUNET_FS_drq_get_next (GNUNET_YES);
2483 }
2484
2485
2486 /**
2487  * We've received a request with the specified priority.  Bound it
2488  * according to how much we trust the given peer.
2489  *
2490  * @param prio_in requested priority
2491  * @param cp the peer making the request
2492  * @return effective priority
2493  */
2494 static uint32_t
2495 bound_priority (uint32_t prio_in,
2496                 struct ConnectedPeer *cp)
2497 {
2498   return 0; // FIXME!
2499 }
2500
2501
2502 /**
2503  * Iterator over entries in the 'query_request_map' that
2504  * tries to see if we have the same request pending from
2505  * the same peer already.
2506  *
2507  * @param cls closure (our 'struct CheckDuplicateRequestClosure')
2508  * @param key current key code (query, ignored, must match)
2509  * @param value value in the hash map (a 'struct PendingRequest'
2510  *              that already exists)
2511  * @return GNUNET_YES if we should continue to
2512  *         iterate (no match yet)
2513  *         GNUNET_NO if not (match found).
2514  */
2515 static int
2516 check_duplicate_request_peer (void *cls,
2517                               const GNUNET_HashCode * key,
2518                               void *value)
2519 {
2520   struct CheckDuplicateRequestClosure *cdc = cls;
2521   struct PendingRequest *have = value;
2522
2523   if (cdc->pr->target_pid == have->target_pid)
2524     {
2525       cdc->have = have;
2526       return GNUNET_NO;
2527     }
2528   return GNUNET_YES;
2529 }
2530
2531
2532 /**
2533  * Handle P2P "GET" request.
2534  *
2535  * @param cls closure, always NULL
2536  * @param other the other peer involved (sender or receiver, NULL
2537  *        for loopback messages where we are both sender and receiver)
2538  * @param message the actual message
2539  * @param latency reported latency of the connection with 'other'
2540  * @param distance reported distance (DV) to 'other'
2541  * @return GNUNET_OK to keep the connection open,
2542  *         GNUNET_SYSERR to close it (signal serious error)
2543  */
2544 static int
2545 handle_p2p_get (void *cls,
2546                 const struct GNUNET_PeerIdentity *other,
2547                 const struct GNUNET_MessageHeader *message,
2548                 struct GNUNET_TIME_Relative latency,
2549                 uint32_t distance)
2550 {
2551   struct PendingRequest *pr;
2552   struct ConnectedPeer *cp;
2553   struct ConnectedPeer *cps;
2554   struct CheckDuplicateRequestClosure cdc;
2555   struct GNUNET_TIME_Relative timeout;
2556   uint16_t msize;
2557   const struct GetMessage *gm;
2558   unsigned int bits;
2559   const GNUNET_HashCode *opt;
2560   uint32_t bm;
2561   size_t bfsize;
2562   uint32_t ttl_decrement;
2563   uint32_t type;
2564   double preference;
2565   int have_ns;
2566
2567   msize = ntohs(message->size);
2568   if (msize < sizeof (struct GetMessage))
2569     {
2570       GNUNET_break_op (0);
2571       return GNUNET_SYSERR;
2572     }
2573   gm = (const struct GetMessage*) message;
2574   type = ntohl (gm->type);
2575   switch (type)
2576     {
2577     case GNUNET_DATASTORE_BLOCKTYPE_ANY:
2578     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2579     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2580     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
2581     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
2582       break;
2583     default:
2584       GNUNET_break_op (0);
2585       return GNUNET_SYSERR;
2586     }
2587   bm = ntohl (gm->hash_bitmap);
2588   bits = 0;
2589   while (bm > 0)
2590     {
2591       if (1 == (bm & 1))
2592         bits++;
2593       bm >>= 1;
2594     }
2595   if (msize < sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode))
2596     {
2597       GNUNET_break_op (0);
2598       return GNUNET_SYSERR;
2599     }
2600   opt = (const GNUNET_HashCode*) &gm[1];
2601   bfsize = msize - sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode);
2602   bm = ntohl (gm->hash_bitmap);
2603   if ( (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE)) &&
2604        (type != GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) )
2605     {
2606       GNUNET_break_op (0);
2607       return GNUNET_SYSERR;
2608     }
2609   bits = 0;
2610   cps = GNUNET_CONTAINER_multihashmap_get (connected_peers,
2611                                            &other->hashPubKey);
2612   if (NULL == cps)
2613     {
2614       /* peer must have just disconnected */
2615       GNUNET_STATISTICS_update (stats,
2616                                 gettext_noop ("# requests dropped due to initiator not being connected"),
2617                                 1,
2618                                 GNUNET_NO);
2619       return GNUNET_SYSERR;
2620     }
2621   if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
2622     cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
2623                                             &opt[bits++]);
2624   else
2625     cp = cps;
2626   if (cp == NULL)
2627     {
2628 #if DEBUG_FS
2629       if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
2630         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2631                     "Failed to find RETURN-TO peer `%4s' in connection set. Dropping query.\n",
2632                     GNUNET_i2s ((const struct GNUNET_PeerIdentity*) &opt[bits-1]));
2633
2634       else
2635         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2636                     "Failed to find peer `%4s' in connection set. Dropping query.\n",
2637                     GNUNET_i2s (other));
2638 #endif
2639       GNUNET_STATISTICS_update (stats,
2640                                 gettext_noop ("# requests dropped due to missing reverse route"),
2641                                 1,
2642                                 GNUNET_NO);
2643      /* FIXME: try connect? */
2644       return GNUNET_OK;
2645     }
2646   /* note that we can really only check load here since otherwise
2647      peers could find out that we are overloaded by not being
2648      disconnected after sending us a malformed query... */
2649   if (GNUNET_YES == test_load_too_high ())
2650     {
2651 #if DEBUG_FS
2652       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2653                   "Dropping query from `%s', this peer is too busy.\n",
2654                   GNUNET_i2s (other));
2655 #endif
2656       GNUNET_STATISTICS_update (stats,
2657                                 gettext_noop ("# requests dropped due to high load"),
2658                                 1,
2659                                 GNUNET_NO);
2660       return GNUNET_OK;
2661     }
2662
2663 #if DEBUG_FS
2664   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2665               "Received request for `%s' of type %u from peer `%4s' with flags %u\n",
2666               GNUNET_h2s (&gm->query),
2667               (unsigned int) type,
2668               GNUNET_i2s (other),
2669               (unsigned int) bm);
2670 #endif
2671   have_ns = (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE));
2672   pr = GNUNET_malloc (sizeof (struct PendingRequest) +
2673                       (have_ns ? sizeof(GNUNET_HashCode) : 0));
2674   if (have_ns)
2675     pr->namespace = (GNUNET_HashCode*) &pr[1];
2676   pr->type = type;
2677   pr->mingle = ntohl (gm->filter_mutator);
2678   if (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE))
2679     memcpy (&pr[1], &opt[bits++], sizeof (GNUNET_HashCode));
2680   if (0 != (bm & GET_MESSAGE_BIT_TRANSMIT_TO))
2681     pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &opt[bits++]);
2682
2683   pr->anonymity_level = 1;
2684   pr->priority = bound_priority (ntohl (gm->priority), cps);
2685   pr->ttl = bound_ttl (ntohl (gm->ttl), pr->priority);
2686   pr->query = gm->query;
2687   /* decrement ttl (always) */
2688   ttl_decrement = 2 * TTL_DECREMENT +
2689     GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2690                               TTL_DECREMENT);
2691   if ( (pr->ttl < 0) &&
2692        (((int32_t)(pr->ttl - ttl_decrement)) > 0) )
2693     {
2694 #if DEBUG_FS
2695       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2696                   "Dropping query from `%s' due to TTL underflow (%d - %u).\n",
2697                   GNUNET_i2s (other),
2698                   pr->ttl,
2699                   ttl_decrement);
2700 #endif
2701       GNUNET_STATISTICS_update (stats,
2702                                 gettext_noop ("# requests dropped due TTL underflow"),
2703                                 1,
2704                                 GNUNET_NO);
2705       /* integer underflow => drop (should be very rare)! */
2706       GNUNET_free (pr);
2707       return GNUNET_OK;
2708     }
2709   pr->ttl -= ttl_decrement;
2710   pr->start_time = GNUNET_TIME_absolute_get ();
2711
2712   /* get bloom filter */
2713   if (bfsize > 0)
2714     {
2715       pr->bf = GNUNET_CONTAINER_bloomfilter_init ((const char*) &opt[bits],
2716                                                   bfsize,
2717                                                   BLOOMFILTER_K);
2718       pr->bf_size = bfsize;
2719     }
2720
2721   cdc.have = NULL;
2722   cdc.pr = pr;
2723   GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
2724                                               &gm->query,
2725                                               &check_duplicate_request_peer,
2726                                               &cdc);
2727   if (cdc.have != NULL)
2728     {
2729       if (cdc.have->start_time.value + cdc.have->ttl >=
2730           pr->start_time.value + pr->ttl)
2731         {
2732           /* existing request has higher TTL, drop new one! */
2733           cdc.have->priority += pr->priority;
2734           destroy_pending_request (pr);
2735 #if DEBUG_FS
2736           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2737                       "Have existing request with higher TTL, dropping new request.\n",
2738                       GNUNET_i2s (other));
2739 #endif
2740           GNUNET_STATISTICS_update (stats,
2741                                     gettext_noop ("# requests dropped due to existing request with higher TTL"),
2742                                     1,
2743                                     GNUNET_NO);
2744           return GNUNET_OK;
2745         }
2746       else
2747         {
2748           /* existing request has lower TTL, drop old one! */
2749           pr->priority += cdc.have->priority;
2750           /* Possible optimization: if we have applicable pending
2751              replies in 'cdc.have', we might want to move those over
2752              (this is a really rare special-case, so it is not clear
2753              that this would be worth it) */
2754           destroy_pending_request (cdc.have);
2755           /* keep processing 'pr'! */
2756         }
2757     }
2758
2759   pr->cp = cp;
2760   GNUNET_CONTAINER_multihashmap_put (query_request_map,
2761                                      &gm->query,
2762                                      pr,
2763                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
2764   GNUNET_CONTAINER_multihashmap_put (peer_request_map,
2765                                      &other->hashPubKey,
2766                                      pr,
2767                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
2768
2769   pr->hnode = GNUNET_CONTAINER_heap_insert (requests_by_expiration_heap,
2770                                             pr,
2771                                             pr->start_time.value + pr->ttl);
2772
2773   GNUNET_STATISTICS_update (stats,
2774                             gettext_noop ("# P2P searches received"),
2775                             1,
2776                             GNUNET_NO);
2777   GNUNET_STATISTICS_update (stats,
2778                             gettext_noop ("# P2P searches active"),
2779                             1,
2780                             GNUNET_NO);
2781
2782   /* calculate change in traffic preference */
2783   preference = (double) pr->priority;
2784   if (preference < QUERY_BANDWIDTH_VALUE)
2785     preference = QUERY_BANDWIDTH_VALUE;
2786   cps->inc_preference += preference;
2787
2788   /* process locally */
2789   if (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK)
2790     type = GNUNET_DATASTORE_BLOCKTYPE_ANY; /* to get on-demand as well */
2791   timeout = GNUNET_TIME_relative_multiply (BASIC_DATASTORE_REQUEST_DELAY,
2792                                            (pr->priority + 1));
2793   pr->drq = GNUNET_FS_drq_get (&gm->query,
2794                                type,
2795                                &process_local_reply,
2796                                pr,
2797                                timeout,
2798                                GNUNET_NO);
2799
2800   /* Are multiple results possible?  If so, start processing remotely now! */
2801   switch (pr->type)
2802     {
2803     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2804     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2805       /* only one result, wait for datastore */
2806       break;
2807     default:
2808       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
2809         pr->task = GNUNET_SCHEDULER_add_now (sched,
2810                                              &forward_request_task,
2811                                              pr);
2812     }
2813
2814   /* make sure we don't track too many requests */
2815   if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap) > max_pending_requests)
2816     {
2817       pr = GNUNET_CONTAINER_heap_peek (requests_by_expiration_heap);
2818       destroy_pending_request (pr);
2819     }
2820   return GNUNET_OK;
2821 }
2822
2823
2824 /* **************************** CS GET Handling ************************ */
2825
2826
2827 /**
2828  * Handle START_SEARCH-message (search request from client).
2829  *
2830  * @param cls closure
2831  * @param client identification of the client
2832  * @param message the actual message
2833  */
2834 static void
2835 handle_start_search (void *cls,
2836                      struct GNUNET_SERVER_Client *client,
2837                      const struct GNUNET_MessageHeader *message)
2838 {
2839   static GNUNET_HashCode all_zeros;
2840   const struct SearchMessage *sm;
2841   struct ClientList *cl;
2842   struct ClientRequestList *crl;
2843   struct PendingRequest *pr;
2844   uint16_t msize;
2845   unsigned int sc;
2846   uint32_t type;
2847
2848   msize = ntohs (message->size);
2849   if ( (msize < sizeof (struct SearchMessage)) ||
2850        (0 != (msize - sizeof (struct SearchMessage)) % sizeof (GNUNET_HashCode)) )
2851     {
2852       GNUNET_break (0);
2853       GNUNET_SERVER_receive_done (client,
2854                                   GNUNET_SYSERR);
2855       return;
2856     }
2857   GNUNET_STATISTICS_update (stats,
2858                             gettext_noop ("# client searches received"),
2859                             1,
2860                             GNUNET_NO);
2861   sc = (msize - sizeof (struct SearchMessage)) / sizeof (GNUNET_HashCode);
2862   sm = (const struct SearchMessage*) message;
2863
2864   cl = client_list;
2865   while ( (cl != NULL) &&
2866           (cl->client != client) )
2867     cl = cl->next;
2868   if (cl == NULL)
2869     {
2870       cl = GNUNET_malloc (sizeof (struct ClientList));
2871       cl->client = client;
2872       GNUNET_SERVER_client_keep (client);
2873       cl->next = client_list;
2874       client_list = cl;
2875     }
2876   type = ntohl (sm->type);
2877 #if DEBUG_FS
2878   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2879               "Received request for `%s' of type %u from local client\n",
2880               GNUNET_h2s (&sm->query),
2881               (unsigned int) type);
2882 #endif
2883   switch (type)
2884     {
2885     case GNUNET_DATASTORE_BLOCKTYPE_ANY:
2886     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2887     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2888     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
2889     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
2890     case GNUNET_DATASTORE_BLOCKTYPE_NBLOCK:
2891       break;
2892     default:
2893       GNUNET_break (0);
2894       GNUNET_SERVER_receive_done (client,
2895                                   GNUNET_SYSERR);
2896       return;
2897     }
2898
2899   /* detect duplicate KBLOCK requests */
2900   if ( (type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
2901        (type == GNUNET_DATASTORE_BLOCKTYPE_NBLOCK) ||
2902        (type == GNUNET_DATASTORE_BLOCKTYPE_ANY) )
2903     {
2904       crl = cl->rl_head;
2905       while ( (crl != NULL) &&
2906               ( (0 != memcmp (&crl->req->query,
2907                               &sm->query,
2908                               sizeof (GNUNET_HashCode))) ||
2909                 (crl->req->type != type) ) )
2910         crl = crl->next;
2911       if (crl != NULL)
2912         {
2913 #if DEBUG_FS
2914           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2915                       "Have existing request, merging content-seen lists.\n");
2916 #endif
2917           pr = crl->req;
2918           /* Duplicate request (used to send long list of
2919              known/blocked results); merge 'pr->replies_seen'
2920              and update bloom filter */
2921           GNUNET_array_grow (pr->replies_seen,
2922                              pr->replies_seen_size,
2923                              pr->replies_seen_off + sc);
2924           memcpy (&pr->replies_seen[pr->replies_seen_off],
2925                   &sm[1],
2926                   sc * sizeof (GNUNET_HashCode));
2927           pr->replies_seen_off += sc;
2928           if (pr->bf != NULL)
2929             GNUNET_CONTAINER_bloomfilter_free (pr->bf);
2930           pr->bf = refresh_bloomfilter (pr->replies_seen_off,
2931                                         &pr->mingle,
2932                                         &pr->bf_size,
2933                                         pr->replies_seen);
2934           GNUNET_STATISTICS_update (stats,
2935                                     gettext_noop ("# client searches updated (merged content seen list)"),
2936                                     1,
2937                                     GNUNET_NO);
2938           GNUNET_SERVER_receive_done (client,
2939                                       GNUNET_OK);
2940           return;
2941         }
2942     }
2943   GNUNET_STATISTICS_update (stats,
2944                             gettext_noop ("# client searches active"),
2945                             1,
2946                             GNUNET_NO);
2947   pr = GNUNET_malloc (sizeof (struct PendingRequest) +
2948                       ((type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK)?sizeof(GNUNET_HashCode):0));
2949   crl = GNUNET_malloc (sizeof (struct ClientRequestList));
2950   memset (crl, 0, sizeof (struct ClientRequestList));
2951   crl->client_list = cl;
2952   GNUNET_CONTAINER_DLL_insert (cl->rl_head,
2953                                cl->rl_tail,
2954                                crl);
2955   crl->req = pr;
2956   pr->type = type;
2957   pr->client_request_list = crl;
2958   GNUNET_array_grow (pr->replies_seen,
2959                      pr->replies_seen_size,
2960                      sc);
2961   memcpy (pr->replies_seen,
2962           &sm[1],
2963           sc * sizeof (GNUNET_HashCode));
2964   pr->replies_seen_off = sc;
2965   pr->anonymity_level = ntohl (sm->anonymity_level);
2966   pr->bf = refresh_bloomfilter (pr->replies_seen_off,
2967                                 &pr->mingle,
2968                                 &pr->bf_size,
2969                                 pr->replies_seen);
2970   pr->query = sm->query;
2971   switch (type)
2972     {
2973     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2974     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2975       if (0 != memcmp (&sm->target,
2976                        &all_zeros,
2977                        sizeof (GNUNET_HashCode)))
2978         pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &sm->target);
2979       break;
2980     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
2981       pr->namespace = (GNUNET_HashCode*) &pr[1];
2982       memcpy (&pr[1], &sm->target, sizeof (GNUNET_HashCode));
2983       break;
2984     default:
2985       break;
2986     }
2987   GNUNET_CONTAINER_multihashmap_put (query_request_map,
2988                                      &sm->query,
2989                                      pr,
2990                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
2991   if (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK)
2992     type = GNUNET_DATASTORE_BLOCKTYPE_ANY; /* get on-demand blocks too! */
2993   pr->drq = GNUNET_FS_drq_get (&sm->query,
2994                                type,
2995                                &process_local_reply,
2996                                pr,
2997                                GNUNET_CONSTANTS_SERVICE_TIMEOUT,
2998                                GNUNET_YES);
2999 }
3000
3001
3002 /* **************************** Startup ************************ */
3003
3004
3005 /**
3006  * List of handlers for P2P messages
3007  * that we care about.
3008  */
3009 static struct GNUNET_CORE_MessageHandler p2p_handlers[] =
3010   {
3011     { &handle_p2p_get,
3012       GNUNET_MESSAGE_TYPE_FS_GET, 0 },
3013     { &handle_p2p_put,
3014       GNUNET_MESSAGE_TYPE_FS_PUT, 0 },
3015     { NULL, 0, 0 }
3016   };
3017
3018
3019 /**
3020  * List of handlers for the messages understood by this
3021  * service.
3022  */
3023 static struct GNUNET_SERVER_MessageHandler handlers[] = {
3024   {&GNUNET_FS_handle_index_start, NULL,
3025    GNUNET_MESSAGE_TYPE_FS_INDEX_START, 0},
3026   {&GNUNET_FS_handle_index_list_get, NULL,
3027    GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_GET, sizeof(struct GNUNET_MessageHeader) },
3028   {&GNUNET_FS_handle_unindex, NULL, GNUNET_MESSAGE_TYPE_FS_UNINDEX,
3029    sizeof (struct UnindexMessage) },
3030   {&handle_start_search, NULL, GNUNET_MESSAGE_TYPE_FS_START_SEARCH,
3031    0 },
3032   {NULL, NULL, 0, 0}
3033 };
3034
3035
3036 /**
3037  * Process fs requests.
3038  *
3039  * @param s scheduler to use
3040  * @param server the initialized server
3041  * @param c configuration to use
3042  */
3043 static int
3044 main_init (struct GNUNET_SCHEDULER_Handle *s,
3045            struct GNUNET_SERVER_Handle *server,
3046            const struct GNUNET_CONFIGURATION_Handle *c)
3047 {
3048   sched = s;
3049   cfg = c;
3050   stats = GNUNET_STATISTICS_create (sched, "fs", cfg);
3051   connected_peers = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
3052   query_request_map = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
3053   peer_request_map = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
3054   requests_by_expiration_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
3055   core = GNUNET_CORE_connect (sched,
3056                               cfg,
3057                               GNUNET_TIME_UNIT_FOREVER_REL,
3058                               NULL,
3059                               NULL,
3060                               NULL,
3061                               &peer_connect_handler,
3062                               &peer_disconnect_handler,
3063                               NULL, GNUNET_NO,
3064                               NULL, GNUNET_NO,
3065                               p2p_handlers);
3066   if (NULL == core)
3067     {
3068       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
3069                   _("Failed to connect to `%s' service.\n"),
3070                   "core");
3071       GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
3072       connected_peers = NULL;
3073       GNUNET_CONTAINER_multihashmap_destroy (query_request_map);
3074       query_request_map = NULL;
3075       GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
3076       requests_by_expiration_heap = NULL;
3077       GNUNET_CONTAINER_multihashmap_destroy (peer_request_map);
3078       peer_request_map = NULL;
3079
3080       return GNUNET_SYSERR;
3081     }
3082   GNUNET_SERVER_disconnect_notify (server,
3083                                    &handle_client_disconnect,
3084                                    NULL);
3085   GNUNET_SERVER_add_handlers (server, handlers);
3086   GNUNET_SCHEDULER_add_delayed (sched,
3087                                 GNUNET_TIME_UNIT_FOREVER_REL,
3088                                 &shutdown_task,
3089                                 NULL);
3090   return GNUNET_OK;
3091 }
3092
3093
3094 /**
3095  * Process fs requests.
3096  *
3097  * @param cls closure
3098  * @param sched scheduler to use
3099  * @param server the initialized server
3100  * @param cfg configuration to use
3101  */
3102 static void
3103 run (void *cls,
3104      struct GNUNET_SCHEDULER_Handle *sched,
3105      struct GNUNET_SERVER_Handle *server,
3106      const struct GNUNET_CONFIGURATION_Handle *cfg)
3107 {
3108   if ( (GNUNET_OK != GNUNET_FS_drq_init (sched, cfg)) ||
3109        (GNUNET_OK != GNUNET_FS_indexing_init (sched, cfg)) ||
3110        (GNUNET_OK != main_init (sched, server, cfg)) )
3111     {
3112       GNUNET_SCHEDULER_shutdown (sched);
3113       return;
3114     }
3115 }
3116
3117
3118 /**
3119  * The main function for the fs service.
3120  *
3121  * @param argc number of arguments from the command line
3122  * @param argv command line arguments
3123  * @return 0 ok, 1 on error
3124  */
3125 int
3126 main (int argc, char *const *argv)
3127 {
3128   return (GNUNET_OK ==
3129           GNUNET_SERVICE_run (argc,
3130                               argv,
3131                               "fs",
3132                               GNUNET_SERVICE_OPTION_NONE,
3133                               &run, NULL)) ? 0 : 1;
3134 }
3135
3136 /* end of gnunet-service-fs.c */