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