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