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