src/fs/gnunet-service-fs.c

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