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