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