src/fs/gnunet-service-fs.c

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