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