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