src/fs/gnunet-service-fs.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009 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 program that provides the file-sharing service
  24  * @author Christian Grothoff
  25  *
  26  * TODO:
  27  * - fix gazillion of minor FIXME's
  28  * - possible major issue: we may queue "gazillions" of (K|S)Blocks for the
  29  *   core to transmit to another peer; need to make sure this is bounded overall...
  30  * - randomly delay processing for improved anonymity (can wait)
  31  * - content migration (put in local DS) (can wait)
  32  * - handle some special cases when forwarding replies based on tracked requests (can wait)
  33  * - tracking of success correlations for hot-path routing (can wait)
  34  * - various load-based actions (can wait)
  35  * - validation of KSBLOCKS (can wait)
  36  * - remove on-demand blocks if they keep failing (can wait)
  37  * - check that we decrement PIDs always where necessary (can wait)
  38  * - find out how to use core-pulling instead of pushing (at least for some cases)
  39  */
  40 #include "platform.h"
  41 #include <float.h>
  42 #include "gnunet_core_service.h"
  43 #include "gnunet_datastore_service.h"
  44 #include "gnunet_peer_lib.h"
  45 #include "gnunet_protocols.h"
  46 #include "gnunet_signatures.h"
  47 #include "gnunet_util_lib.h"
  48 #include "fs.h"
  49
  50 #define DEBUG_FS GNUNET_NO
  51
  52
  53 /**
  54  * In-memory information about indexed files (also available
  55  * on-disk).
  56  */
  57 struct IndexInfo
  58 {
  59
  60   /**
  61    * This is a linked list.
  62    */
  63   struct IndexInfo *next;
  64
  65   /**
  66    * Name of the indexed file.  Memory allocated
  67    * at the end of this struct (do not free).
  68    */
  69   const char *filename;
  70
  71   /**
  72    * Context for transmitting confirmation to client,
  73    * NULL if we've done this already.
  74    */
  75   struct GNUNET_SERVER_TransmitContext *tc;
  76
  77   /**
  78    * Hash of the contents of the file.
  79    */
  80   GNUNET_HashCode file_id;
  81
  82 };
  83
  84
  85 /**
  86  * Signature of a function that is called whenever a datastore
  87  * request can be processed (or an entry put on the queue times out).
  88  *
  89  * @param cls closure
  90  * @param ok GNUNET_OK if DS is ready, GNUNET_SYSERR on timeout
  91  */
  92 typedef void (*RequestFunction)(void *cls,
  93                                 int ok);
  94
  95
  96 /**
  97  * Doubly-linked list of our requests for the datastore.
  98  */
  99 struct DatastoreRequestQueue
 100 {
 101
 102   /**
 103    * This is a doubly-linked list.
 104    */
 105   struct DatastoreRequestQueue *next;
 106
 107   /**
 108    * This is a doubly-linked list.
 109    */
 110   struct DatastoreRequestQueue *prev;
 111
 112   /**
 113    * Function to call (will issue the request).
 114    */
 115   RequestFunction req;
 116
 117   /**
 118    * Closure for req.
 119    */
 120   void *req_cls;
 121
 122   /**
 123    * When should this request time-out because we don't care anymore?
 124    */
 125   struct GNUNET_TIME_Absolute timeout;
 126
 127   /**
 128    * ID of task used for signaling timeout.
 129    */
 130   GNUNET_SCHEDULER_TaskIdentifier task;
 131
 132 };
 133
 134
 135 /**
 136  * Closure for processing START_SEARCH messages from a client.
 137  */
 138 struct LocalGetContext
 139 {
 140
 141   /**
 142    * This is a doubly-linked list.
 143    */
 144   struct LocalGetContext *next;
 145
 146   /**
 147    * This is a doubly-linked list.
 148    */
 149   struct LocalGetContext *prev;
 150
 151   /**
 152    * Client that initiated the search.
 153    */
 154   struct GNUNET_SERVER_Client *client;
 155
 156   /**
 157    * Array of results that we've already received
 158    * (can be NULL).
 159    */
 160   GNUNET_HashCode *results;
 161
 162   /**
 163    * Bloomfilter over all results (for fast query construction);
 164    * NULL if we don't have any results.
 165    *
 166    * FIXME: this member is not used, is that OK? If so, it should
 167    * be removed!
 168    */
 169   struct GNUNET_CONTAINER_BloomFilter *results_bf;
 170
 171   /**
 172    * DS request associated with this operation.
 173    */
 174   struct DatastoreRequestQueue *req;
 175
 176   /**
 177    * Current result message to transmit to client (or NULL).
 178    */
 179   struct ContentMessage *result;
 180
 181   /**
 182    * Type of the content that we're looking for.
 183    * 0 for any.
 184    */
 185   uint32_t type;
 186
 187   /**
 188    * Desired anonymity level.
 189    */
 190   uint32_t anonymity_level;
 191
 192   /**
 193    * Number of results actually stored in the results array.
 194    */
 195   unsigned int results_used;
 196
 197   /**
 198    * Size of the results array in memory.
 199    */
 200   unsigned int results_size;
 201
 202   /**
 203    * Size (in bytes) of the 'results_bf' bloomfilter.
 204    *
 205    * FIXME: this member is not used, is that OK? If so, it should
 206    * be removed!
 207    */
 208   size_t results_bf_size;
 209
 210   /**
 211    * If the request is for a DBLOCK or IBLOCK, this is the identity of
 212    * the peer that is known to have a response.  Set to all-zeros if
 213    * such a target is not known (note that even if OUR anonymity
 214    * level is >0 we may happen to know the responder's identity;
 215    * nevertheless, we should probably not use it for a DHT-lookup
 216    * or similar blunt actions in order to avoid exposing ourselves).
 217    */
 218   struct GNUNET_PeerIdentity target;
 219
 220   /**
 221    * If the request is for an SBLOCK, this is the identity of the
 222    * pseudonym to which the SBLOCK belongs.
 223    */
 224   GNUNET_HashCode namespace;
 225
 226   /**
 227    * Hash of the keyword (aka query) for KBLOCKs; Hash of
 228    * the CHK-encoded block for DBLOCKS and IBLOCKS (aka query)
 229    * and hash of the identifier XORed with the target for
 230    * SBLOCKS (aka query).
 231    */
 232   GNUNET_HashCode query;
 233
 234 };
 235
 236
 237 /**
 238  * Possible routing policies for an FS-GET request.
 239  */
 240 enum RoutingPolicy
 241   {
 242     /**
 243      * Simply drop the request.
 244      */
 245     ROUTING_POLICY_NONE = 0,
 246
 247     /**
 248      * Answer it if we can from local datastore.
 249      */
 250     ROUTING_POLICY_ANSWER = 1,
 251
 252     /**
 253      * Forward the request to other peers (if possible).
 254      */
 255     ROUTING_POLICY_FORWARD = 2,
 256
 257     /**
 258      * Forward to other peers, and ask them to route
 259      * the response via ourselves.
 260      */
 261     ROUTING_POLICY_INDIRECT = 6,
 262
 263     /**
 264      * Do everything we could possibly do (that would
 265      * make sense).
 266      */
 267     ROUTING_POLICY_ALL = 7
 268   };
 269
 270
 271 /**
 272  * Internal context we use for our initial processing
 273  * of a GET request.
 274  */
 275 struct ProcessGetContext
 276 {
 277   /**
 278    * The search query (used for datastore lookup).
 279    */
 280   GNUNET_HashCode query;
 281
 282   /**
 283    * Which peer we should forward the response to.
 284    */
 285   struct GNUNET_PeerIdentity reply_to;
 286
 287   /**
 288    * Namespace for the result (only set for SKS requests)
 289    */
 290   GNUNET_HashCode namespace;
 291
 292   /**
 293    * Peer that we should forward the query to if possible
 294    * (since that peer likely has the content).
 295    */
 296   struct GNUNET_PeerIdentity prime_target;
 297
 298   /**
 299    * When did we receive this request?
 300    */
 301   struct GNUNET_TIME_Absolute start_time;
 302
 303   /**
 304    * Our entry in the DRQ (non-NULL while we wait for our
 305    * turn to interact with the local database).
 306    */
 307   struct DatastoreRequestQueue *drq;
 308
 309   /**
 310    * Filter used to eliminate duplicate results.  Can be NULL if we
 311    * are not yet filtering any results.
 312    */
 313   struct GNUNET_CONTAINER_BloomFilter *bf;
 314
 315   /**
 316    * Bitmap describing which of the optional
 317    * hash codes / peer identities were given to us.
 318    */
 319   uint32_t bm;
 320
 321   /**
 322    * Desired block type.
 323    */
 324   uint32_t type;
 325
 326   /**
 327    * Priority of the request.
 328    */
 329   uint32_t priority;
 330
 331   /**
 332    * Size of the 'bf' (in bytes).
 333    */
 334   size_t bf_size;
 335
 336   /**
 337    * In what ways are we going to process
 338    * the request?
 339    */
 340   enum RoutingPolicy policy;
 341
 342   /**
 343    * Time-to-live for the request (value
 344    * we use).
 345    */
 346   int32_t ttl;
 347
 348   /**
 349    * Number to mingle hashes for bloom-filter
 350    * tests with.
 351    */
 352   int32_t mingle;
 353
 354   /**
 355    * Number of results that were found so far.
 356    */
 357   unsigned int results_found;
 358 };
 359
 360
 361 /**
 362  * Information we keep for each pending reply.
 363  */
 364 struct PendingReply
 365 {
 366   /**
 367    * This is a linked list.
 368    */
 369   struct PendingReply *next;
 370
 371   /**
 372    * Size of the reply; actual reply message follows
 373    * at the end of this struct.
 374    */
 375   size_t msize;
 376
 377 };
 378
 379
 380 /**
 381  * All requests from a client are kept in a doubly-linked list.
 382  */
 383 struct ClientRequestList;
 384
 385
 386 /**
 387  * Information we keep for each pending request.  We should try to
 388  * keep this struct as small as possible since its memory consumption
 389  * is key to how many requests we can have pending at once.
 390  */
 391 struct PendingRequest
 392 {
 393
 394   /**
 395    * ID of a client making a request, NULL if this entry is for a
 396    * peer.
 397    */
 398   struct GNUNET_SERVER_Client *client;
 399
 400   /**
 401    * If this request was made by a client,
 402    * this is our entry in the client request
 403    * list; otherwise NULL.
 404    */
 405   struct ClientRequestList *crl_entry;
 406
 407   /**
 408    * If this is a namespace query, pointer to the hash of the public
 409    * key of the namespace; otherwise NULL.
 410    */
 411   GNUNET_HashCode *namespace;
 412
 413   /**
 414    * Bloomfilter we use to filter out replies that we don't care about
 415    * (anymore).  NULL as long as we are interested in all replies.
 416    */
 417   struct GNUNET_CONTAINER_BloomFilter *bf;
 418
 419   /**
 420    * Replies that we have received but were unable to forward yet
 421    * (typically non-null only if we have a pending transmission
 422    * request with the client or the respective peer).
 423    */
 424   struct PendingReply *replies_pending;
 425
 426   /**
 427    * Pending transmission request with the core service for the target
 428    * peer (for processing of 'replies_pending') or Handle for a
 429    * pending query-request for P2P-transmission with the core service.
 430    * If non-NULL, this request must be cancelled should this struct be
 431    * destroyed!
 432    */
 433   struct GNUNET_CORE_TransmitHandle *cth;
 434
 435   /**
 436    * Pending transmission request for the target client (for processing of
 437    * 'replies_pending').
 438    */
 439   struct GNUNET_CONNECTION_TransmitHandle *th;
 440
 441   /**
 442    * Hash code of all replies that we have seen so far (only valid
 443    * if client is not NULL since we only track replies like this for
 444    * our own clients).
 445    */
 446   GNUNET_HashCode *replies_seen;
 447
 448   /**
 449    * When did we first see this request (form this peer), or, if our
 450    * client is initiating, when did we last initiate a search?
 451    */
 452   struct GNUNET_TIME_Absolute start_time;
 453
 454   /**
 455    * The query that this request is for.
 456    */
 457   GNUNET_HashCode query;
 458
 459   /**
 460    * The task responsible for transmitting queries
 461    * for this request.
 462    */
 463   GNUNET_SCHEDULER_TaskIdentifier task;
 464
 465   /**
 466    * (Interned) Peer identifier (only valid if "client" is NULL)
 467    * that identifies a peer that gave us this request.
 468    */
 469   GNUNET_PEER_Id source_pid;
 470
 471   /**
 472    * (Interned) Peer identifier that identifies a preferred target
 473    * for requests.
 474    */
 475   GNUNET_PEER_Id target_pid;
 476
 477   /**
 478    * (Interned) Peer identifiers of peers that have already
 479    * received our query for this content.
 480    */
 481   GNUNET_PEER_Id *used_pids;
 482
 483   /**
 484    * Size of the 'bf' (in bytes).
 485    */
 486   size_t bf_size;
 487
 488   /**
 489    * Desired anonymity level; only valid for requests from a local client.
 490    */
 491   uint32_t anonymity_level;
 492
 493   /**
 494    * How many entries in "used_pids" are actually valid?
 495    */
 496   unsigned int used_pids_off;
 497
 498   /**
 499    * How long is the "used_pids" array?
 500    */
 501   unsigned int used_pids_size;
 502
 503   /**
 504    * How many entries in "replies_seen" are actually valid?
 505    */
 506   unsigned int replies_seen_off;
 507
 508   /**
 509    * How long is the "replies_seen" array?
 510    */
 511   unsigned int replies_seen_size;
 512
 513   /**
 514    * Priority with which this request was made.  If one of our clients
 515    * made the request, then this is the current priority that we are
 516    * using when initiating the request.  This value is used when
 517    * we decide to reward other peers with trust for providing a reply.
 518    */
 519   uint32_t priority;
 520
 521   /**
 522    * Priority points left for us to spend when forwarding this request
 523    * to other peers.
 524    */
 525   uint32_t remaining_priority;
 526
 527   /**
 528    * Number to mingle hashes for bloom-filter
 529    * tests with.
 530    */
 531   int32_t mingle;
 532
 533   /**
 534    * TTL with which we saw this request (or, if we initiated, TTL that
 535    * we used for the request).
 536    */
 537   int32_t ttl;
 538
 539   /**
 540    * Type of the content that this request is for.
 541    */
 542   uint32_t type;
 543
 544 };
 545
 546
 547 /**
 548  * All requests from a client are kept in a doubly-linked list.
 549  */
 550 struct ClientRequestList
 551 {
 552   /**
 553    * This is a doubly-linked list.
 554    */
 555   struct ClientRequestList *next;
 556
 557   /**
 558    * This is a doubly-linked list.
 559    */
 560   struct ClientRequestList *prev;
 561
 562   /**
 563    * A request from this client.
 564    */
 565   struct PendingRequest *req;
 566
 567   /**
 568    * Client list with the head and tail of this DLL.
 569    */
 570   struct ClientList *cl;
 571 };
 572
 573
 574 /**
 575  * Linked list of all clients that we are
 576  * currently processing requests for.
 577  */
 578 struct ClientList
 579 {
 580
 581   /**
 582    * This is a linked list.
 583    */
 584   struct ClientList *next;
 585
 586   /**
 587    * What client is this entry for?
 588    */
 589   struct GNUNET_SERVER_Client* client;
 590
 591   /**
 592    * Head of the DLL of requests from this client.
 593    */
 594   struct ClientRequestList *head;
 595
 596   /**
 597    * Tail of the DLL of requests from this client.
 598    */
 599   struct ClientRequestList *tail;
 600
 601 };
 602
 603
 604 /**
 605  * Closure for "process_reply" function.
 606  */
 607 struct ProcessReplyClosure
 608 {
 609   /**
 610    * The data for the reply.
 611    */
 612   const void *data;
 613
 614   /**
 615    * When the reply expires.
 616    */
 617   struct GNUNET_TIME_Absolute expiration;
 618
 619   /**
 620    * Size of data.
 621    */
 622   size_t size;
 623
 624   /**
 625    * Namespace that this reply belongs to
 626    * (if it is of type SBLOCK).
 627    */
 628   GNUNET_HashCode namespace;
 629
 630   /**
 631    * Type of the block.
 632    */
 633   uint32_t type;
 634
 635   /**
 636    * How much was this reply worth to us?
 637    */
 638   uint32_t priority;
 639 };
 640
 641
 642 /**
 643  * Information about a peer that we are connected to.
 644  * We track data that is useful for determining which
 645  * peers should receive our requests.
 646  */
 647 struct ConnectedPeer
 648 {
 649
 650   /**
 651    * List of the last clients for which this peer
 652    * successfully answered a query.
 653    */
 654   struct GNUNET_SERVER_Client *last_client_replies[CS2P_SUCCESS_LIST_SIZE];
 655
 656   /**
 657    * List of the last PIDs for which
 658    * this peer successfully answered a query;
 659    * We use 0 to indicate no successful reply.
 660    */
 661   GNUNET_PEER_Id last_p2p_replies[P2P_SUCCESS_LIST_SIZE];
 662
 663   /**
 664    * Average delay between sending the peer a request and
 665    * getting a reply (only calculated over the requests for
 666    * which we actually got a reply).   Calculated
 667    * as a moving average: new_delay = ((n-1)*last_delay+curr_delay) / n
 668    */
 669   struct GNUNET_TIME_Relative avg_delay;
 670
 671   /**
 672    * Average priority of successful replies.  Calculated
 673    * as a moving average: new_avg = ((n-1)*last_avg+curr_prio) / n
 674    */
 675   double avg_priority;
 676
 677   /**
 678    * The peer's identity.
 679    */
 680   GNUNET_PEER_Id pid;
 681
 682   /**
 683    * Number of requests we have currently pending
 684    * with this peer (that is, requests that were
 685    * transmitted so recently that we would not retransmit
 686    * them right now).
 687    */
 688   unsigned int pending_requests;
 689
 690   /**
 691    * Which offset in "last_p2p_replies" will be updated next?
 692    * (we go round-robin).
 693    */
 694   unsigned int last_p2p_replies_woff;
 695
 696   /**
 697    * Which offset in "last_client_replies" will be updated next?
 698    * (we go round-robin).
 699    */
 700   unsigned int last_client_replies_woff;
 701
 702 };
 703
 704
 705 /**
 706  * Our connection to the datastore.
 707  */
 708 static struct GNUNET_DATASTORE_Handle *dsh;
 709
 710 /**
 711  * Our scheduler.
 712  */
 713 static struct GNUNET_SCHEDULER_Handle *sched;
 714
 715 /**
 716  * Our configuration.
 717  */
 718 const struct GNUNET_CONFIGURATION_Handle *cfg;
 719
 720 /**
 721  * Handle to the core service (NULL until we've connected to it).
 722  */
 723 struct GNUNET_CORE_Handle *core;
 724
 725 /**
 726  * Head of doubly-linked LGC list.
 727  */
 728 static struct LocalGetContext *lgc_head;
 729
 730 /**
 731  * Tail of doubly-linked LGC list.
 732  */
 733 static struct LocalGetContext *lgc_tail;
 734
 735 /**
 736  * Head of request queue for the datastore, sorted by timeout.
 737  */
 738 static struct DatastoreRequestQueue *drq_head;
 739
 740 /**
 741  * Tail of request queue for the datastore.
 742  */
 743 static struct DatastoreRequestQueue *drq_tail;
 744
 745 /**
 746  * Linked list of indexed files.
 747  */
 748 static struct IndexInfo *indexed_files;
 749
 750 /**
 751  * Maps hash over content of indexed files to the respective filename.
 752  * The filenames are pointers into the indexed_files linked list and
 753  * do not need to be freed.
 754  */
 755 static struct GNUNET_CONTAINER_MultiHashMap *ifm;
 756
 757 /**
 758  * Map of query hash codes to requests.
 759  */
 760 static struct GNUNET_CONTAINER_MultiHashMap *requests_by_query;
 761
 762 /**
 763  * Map of peer IDs to requests (for those requests coming
 764  * from other peers).
 765  */
 766 static struct GNUNET_CONTAINER_MultiHashMap *requests_by_peer;
 767
 768 /**
 769  * Linked list of all of our clients and their requests.
 770  */
 771 static struct ClientList *clients;
 772
 773 /**
 774  * Heap with the request that will expire next at the top.  Contains
 775  * pointers of type "struct PendingRequest*"; these will *also* be
 776  * aliased from the "requests_by_peer" data structures and the
 777  * "requests_by_query" table.  Note that requests from our clients
 778  * don't expire and are thus NOT in the "requests_by_expiration"
 779  * (or the "requests_by_peer" tables).
 780  */
 781 static struct GNUNET_CONTAINER_Heap *requests_by_expiration;
 782
 783 /**
 784  * Map of peer identifiers to "struct ConnectedPeer" (for that peer).
 785  */
 786 static struct GNUNET_CONTAINER_MultiHashMap *connected_peers;
 787
 788 /**
 789  * Maximum number of requests (from other peers) that we're
 790  * willing to have pending at any given point in time.
 791  * FIXME: set from configuration (and 32 is a tiny value for testing only).
 792  */
 793 static uint64_t max_pending_requests = 32;
 794
 795
 796 /**
 797  * Write the current index information list to disk.
 798  */
 799 static void
 800 write_index_list ()
 801 {
 802   struct GNUNET_BIO_WriteHandle *wh;
 803   char *fn;
 804   struct IndexInfo *pos;
 805
 806   if (GNUNET_OK !=
 807       GNUNET_CONFIGURATION_get_value_filename (cfg,
 808                                                "FS",
 809                                                "INDEXDB",
 810                                                &fn))
 811     {
 812       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 813                   _("Configuration option `%s' in section `%s' missing.\n"),
 814                   "INDEXDB",
 815                   "FS");
 816       return;
 817     }
 818   wh = GNUNET_BIO_write_open (fn);
 819   if (NULL == wh)
 820     {
 821       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 822                   _("Could not open `%s'.\n"),
 823                   fn);
 824       GNUNET_free (fn);
 825       return;
 826     }
 827   pos = indexed_files;
 828   while (pos != NULL)
 829     {
 830       if ( (GNUNET_OK !=
 831             GNUNET_BIO_write (wh,
 832                               &pos->file_id,
 833                               sizeof (GNUNET_HashCode))) ||
 834            (GNUNET_OK !=
 835             GNUNET_BIO_write_string (wh,
 836                                      pos->filename)) )
 837         break;
 838       pos = pos->next;
 839     }
 840   if (GNUNET_OK !=
 841       GNUNET_BIO_write_close (wh))
 842     {
 843       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 844                   _("Error writing `%s'.\n"),
 845                   fn);
 846       GNUNET_free (fn);
 847       return;
 848     }
 849   GNUNET_free (fn);
 850 }
 851
 852
 853 /**
 854  * Read index information from disk.
 855  */
 856 static void
 857 read_index_list ()
 858 {
 859   struct GNUNET_BIO_ReadHandle *rh;
 860   char *fn;
 861   struct IndexInfo *pos;
 862   char *fname;
 863   GNUNET_HashCode hc;
 864   size_t slen;
 865   char *emsg;
 866
 867   if (GNUNET_OK !=
 868       GNUNET_CONFIGURATION_get_value_filename (cfg,
 869                                                "FS",
 870                                                "INDEXDB",
 871                                                &fn))
 872     {
 873       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 874                   _("Configuration option `%s' in section `%s' missing.\n"),
 875                   "INDEXDB",
 876                   "FS");
 877       return;
 878     }
 879   if (GNUNET_NO == GNUNET_DISK_file_test (fn))
 880     {
 881       /* no index info  yet */
 882       GNUNET_free (fn);
 883       return;
 884     }
 885   rh = GNUNET_BIO_read_open (fn);
 886   if (NULL == rh)
 887     {
 888       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 889                   _("Could not open `%s'.\n"),
 890                   fn);
 891       GNUNET_free (fn);
 892       return;
 893     }
 894
 895   while ( (GNUNET_OK ==
 896            GNUNET_BIO_read (rh,
 897                             "Hash of indexed file",
 898                             &hc,
 899                             sizeof (GNUNET_HashCode))) &&
 900           (GNUNET_OK ==
 901            GNUNET_BIO_read_string (rh,
 902                                    "Name of indexed file",
 903                                    &fname,
 904                                    1024 * 16)) )
 905     {
 906       slen = strlen (fname) + 1;
 907       pos = GNUNET_malloc (sizeof (struct IndexInfo) + slen);
 908       pos->file_id = hc;
 909       pos->filename = (const char *) &pos[1];
 910       memcpy (&pos[1], fname, slen);
 911       if (GNUNET_SYSERR ==
 912           GNUNET_CONTAINER_multihashmap_put (ifm,
 913                                              &hc,
 914                                              (void*) pos->filename,
 915                                              GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
 916         {
 917           GNUNET_free (pos);
 918         }
 919       else
 920         {
 921           pos->next = indexed_files;
 922           indexed_files = pos;
 923         }
 924       GNUNET_free (fname);
 925     }
 926   if (GNUNET_OK !=
 927       GNUNET_BIO_read_close (rh, &emsg))
 928     GNUNET_free (emsg);
 929   GNUNET_free (fn);
 930 }
 931
 932
 933 /**
 934  * We've validated the hash of the file we're about to
 935  * index.  Signal success to the client and update
 936  * our internal data structures.
 937  *
 938  * @param ii the index info entry for the request
 939  */
 940 static void
 941 signal_index_ok (struct IndexInfo *ii)
 942 {
 943   if (GNUNET_SYSERR ==
 944       GNUNET_CONTAINER_multihashmap_put (ifm,
 945                                          &ii->file_id,
 946                                          (void*) ii->filename,
 947                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
 948     {
 949       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 950                   _("Index request received for file `%s' is indexed as `%s'.  Permitting anyway.\n"),
 951                   ii->filename,
 952                   (const char*) GNUNET_CONTAINER_multihashmap_get (ifm,
 953                                                                    &ii->file_id));
 954       GNUNET_SERVER_transmit_context_append (ii->tc,
 955                                              NULL, 0,
 956                                              GNUNET_MESSAGE_TYPE_FS_INDEX_START_OK);
 957       GNUNET_SERVER_transmit_context_run (ii->tc,
 958                                           GNUNET_TIME_UNIT_MINUTES);
 959       GNUNET_free (ii);
 960       return;
 961     }
 962   ii->next = indexed_files;
 963   indexed_files = ii;
 964   write_index_list ();
 965   GNUNET_SERVER_transmit_context_append (ii->tc,
 966                                          NULL, 0,
 967                                          GNUNET_MESSAGE_TYPE_FS_INDEX_START_OK);
 968   GNUNET_SERVER_transmit_context_run (ii->tc,
 969                                       GNUNET_TIME_UNIT_MINUTES);
 970   ii->tc = NULL;
 971 }
 972
 973
 974 /**
 975  * Function called once the hash computation over an
 976  * indexed file has completed.
 977  *
 978  * @param cls closure, our publishing context
 979  * @param res resulting hash, NULL on error
 980  */
 981 static void
 982 hash_for_index_val (void *cls,
 983                     const GNUNET_HashCode *
 984                     res)
 985 {
 986   struct IndexInfo *ii = cls;
 987
 988   if ( (res == NULL) ||
 989        (0 != memcmp (res,
 990                      &ii->file_id,
 991                      sizeof(GNUNET_HashCode))) )
 992     {
 993       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 994                   _("Hash mismatch trying to index file `%s' which has hash `%s'\n"),
 995                   ii->filename,
 996                   GNUNET_h2s (res));
 997 #if DEBUG_FS
 998       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 999                   "Wanted `%s'\n",
1000                   GNUNET_h2s (&ii->file_id));
1001 #endif
1002       GNUNET_SERVER_transmit_context_append (ii->tc,
1003                                              NULL, 0,
1004                                              GNUNET_MESSAGE_TYPE_FS_INDEX_START_FAILED);
1005       GNUNET_SERVER_transmit_context_run (ii->tc,
1006                                           GNUNET_TIME_UNIT_MINUTES);
1007       GNUNET_free (ii);
1008       return;
1009     }
1010   signal_index_ok (ii);
1011 }
1012
1013
1014 /**
1015  * Handle INDEX_START-message.
1016  *
1017  * @param cls closure
1018  * @param client identification of the client
1019  * @param message the actual message
1020  */
1021 static void
1022 handle_index_start (void *cls,
1023                     struct GNUNET_SERVER_Client *client,
1024                     const struct GNUNET_MessageHeader *message)
1025 {
1026   const struct IndexStartMessage *ism;
1027   const char *fn;
1028   uint16_t msize;
1029   struct IndexInfo *ii;
1030   size_t slen;
1031   uint32_t dev;
1032   uint64_t ino;
1033   uint32_t mydev;
1034   uint64_t myino;
1035
1036   msize = ntohs(message->size);
1037   if ( (msize <= sizeof (struct IndexStartMessage)) ||
1038        ( ((const char *)message)[msize-1] != '\0') )
1039     {
1040       GNUNET_break (0);
1041       GNUNET_SERVER_receive_done (client,
1042                                   GNUNET_SYSERR);
1043       return;
1044     }
1045   ism = (const struct IndexStartMessage*) message;
1046   fn = (const char*) &ism[1];
1047   dev = ntohl (ism->device);
1048   ino = GNUNET_ntohll (ism->inode);
1049   ism = (const struct IndexStartMessage*) message;
1050   slen = strlen (fn) + 1;
1051   ii = GNUNET_malloc (sizeof (struct IndexInfo) + slen);
1052   ii->filename = (const char*) &ii[1];
1053   memcpy (&ii[1], fn, slen);
1054   ii->file_id = ism->file_id;
1055   ii->tc = GNUNET_SERVER_transmit_context_create (client);
1056   if ( ( (dev != 0) ||
1057          (ino != 0) ) &&
1058        (GNUNET_OK == GNUNET_DISK_file_get_identifiers (fn,
1059                                                        &mydev,
1060                                                        &myino)) &&
1061        ( (dev == mydev) &&
1062          (ino == myino) ) )
1063     {
1064       /* fast validation OK! */
1065       signal_index_ok (ii);
1066       return;
1067     }
1068 #if DEBUG_FS
1069   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1070               "Mismatch in file identifiers (%llu != %llu or %u != %u), need to hash.\n",
1071               (unsigned long long) ino,
1072               (unsigned long long) myino,
1073               (unsigned int) dev,
1074               (unsigned int) mydev);
1075 #endif
1076   /* slow validation, need to hash full file (again) */
1077   GNUNET_CRYPTO_hash_file (sched,
1078                            GNUNET_SCHEDULER_PRIORITY_IDLE,
1079                            fn,
1080                            HASHING_BLOCKSIZE,
1081                            &hash_for_index_val,
1082                            ii);
1083 }
1084
1085
1086 /**
1087  * Handle INDEX_LIST_GET-message.
1088  *
1089  * @param cls closure
1090  * @param client identification of the client
1091  * @param message the actual message
1092  */
1093 static void
1094 handle_index_list_get (void *cls,
1095                        struct GNUNET_SERVER_Client *client,
1096                        const struct GNUNET_MessageHeader *message)
1097 {
1098   struct GNUNET_SERVER_TransmitContext *tc;
1099   struct IndexInfoMessage *iim;
1100   char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE];
1101   size_t slen;
1102   const char *fn;
1103   struct GNUNET_MessageHeader *msg;
1104   struct IndexInfo *pos;
1105
1106   tc = GNUNET_SERVER_transmit_context_create (client);
1107   iim = (struct IndexInfoMessage*) buf;
1108   msg = &iim->header;
1109   pos = indexed_files;
1110   while (NULL != pos)
1111     {
1112       iim->reserved = 0;
1113       iim->file_id = pos->file_id;
1114       fn = pos->filename;
1115       slen = strlen (fn) + 1;
1116       if (slen + sizeof (struct IndexInfoMessage) >
1117           GNUNET_SERVER_MAX_MESSAGE_SIZE)
1118         {
1119           GNUNET_break (0);
1120           break;
1121         }
1122       memcpy (&iim[1], fn, slen);
1123       GNUNET_SERVER_transmit_context_append
1124         (tc,
1125          &msg[1],
1126          sizeof (struct IndexInfoMessage)
1127          - sizeof (struct GNUNET_MessageHeader) + slen,
1128          GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_ENTRY);
1129       pos = pos->next;
1130     }
1131   GNUNET_SERVER_transmit_context_append (tc,
1132                                          NULL, 0,
1133                                          GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_END);
1134   GNUNET_SERVER_transmit_context_run (tc,
1135                                       GNUNET_TIME_UNIT_MINUTES);
1136 }
1137
1138
1139 /**
1140  * Handle UNINDEX-message.
1141  *
1142  * @param cls closure
1143  * @param client identification of the client
1144  * @param message the actual message
1145  */
1146 static void
1147 handle_unindex (void *cls,
1148                 struct GNUNET_SERVER_Client *client,
1149                 const struct GNUNET_MessageHeader *message)
1150 {
1151   const struct UnindexMessage *um;
1152   struct IndexInfo *pos;
1153   struct IndexInfo *prev;
1154   struct IndexInfo *next;
1155   struct GNUNET_SERVER_TransmitContext *tc;
1156   int found;
1157
1158   um = (const struct UnindexMessage*) message;
1159   found = GNUNET_NO;
1160   prev = NULL;
1161   pos = indexed_files;
1162   while (NULL != pos)
1163     {
1164       next = pos->next;
1165       if (0 == memcmp (&pos->file_id,
1166                        &um->file_id,
1167                        sizeof (GNUNET_HashCode)))
1168         {
1169           if (prev == NULL)
1170             indexed_files = pos->next;
1171           else
1172             prev->next = pos->next;
1173           GNUNET_free (pos);
1174           found = GNUNET_YES;
1175         }
1176       else
1177         {
1178           prev = pos;
1179         }
1180       pos = next;
1181     }
1182 #if DEBUG_FS
1183   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1184               "Client requested unindexing of file `%s': %s\n",
1185               GNUNET_h2s (&um->file_id),
1186               found ? "found" : "not found");
1187 #endif
1188   if (GNUNET_YES == found)
1189     write_index_list ();
1190   tc = GNUNET_SERVER_transmit_context_create (client);
1191   GNUNET_SERVER_transmit_context_append (tc,
1192                                          NULL, 0,
1193                                          GNUNET_MESSAGE_TYPE_FS_UNINDEX_OK);
1194   GNUNET_SERVER_transmit_context_run (tc,
1195                                       GNUNET_TIME_UNIT_MINUTES);
1196 }
1197
1198
1199 /**
1200  * Run the next DS request in our
1201  * queue, we're done with the current one.
1202  */
1203 static void
1204 next_ds_request ()
1205 {
1206   struct DatastoreRequestQueue *e;
1207
1208   while (NULL != (e = drq_head))
1209     {
1210       if (0 != GNUNET_TIME_absolute_get_remaining (e->timeout).value)
1211         break;
1212       if (e->task != GNUNET_SCHEDULER_NO_TASK)
1213         GNUNET_SCHEDULER_cancel (sched, e->task);
1214       GNUNET_CONTAINER_DLL_remove (drq_head, drq_tail, e);
1215       e->req (e->req_cls, GNUNET_NO);
1216       GNUNET_free (e);
1217     }
1218   if (e == NULL)
1219     return;
1220   if (e->task != GNUNET_SCHEDULER_NO_TASK)
1221     GNUNET_SCHEDULER_cancel (sched, e->task);
1222   e->task = GNUNET_SCHEDULER_NO_TASK;
1223   e->req (e->req_cls, GNUNET_YES);
1224   GNUNET_CONTAINER_DLL_remove (drq_head, drq_tail, e);
1225   GNUNET_free (e);
1226 }
1227
1228
1229 /**
1230  * A datastore request had to be timed out.
1231  *
1232  * @param cls closure (of type "struct DatastoreRequestQueue*")
1233  * @param tc task context, unused
1234  */
1235 static void
1236 timeout_ds_request (void *cls,
1237                     const struct GNUNET_SCHEDULER_TaskContext *tc)
1238 {
1239   struct DatastoreRequestQueue *e = cls;
1240
1241   e->task = GNUNET_SCHEDULER_NO_TASK;
1242   GNUNET_CONTAINER_DLL_remove (drq_head, drq_tail, e);
1243   e->req (e->req_cls, GNUNET_NO);
1244   GNUNET_free (e);
1245 }
1246
1247
1248 /**
1249  * Queue a request for the datastore.
1250  *
1251  * @param deadline by when the request should run
1252  * @param fun function to call once the request can be run
1253  * @param fun_cls closure for fun
1254  */
1255 static struct DatastoreRequestQueue *
1256 queue_ds_request (struct GNUNET_TIME_Relative deadline,
1257                   RequestFunction fun,
1258                   void *fun_cls)
1259 {
1260   struct DatastoreRequestQueue *e;
1261   struct DatastoreRequestQueue *bef;
1262
1263   if (drq_head == NULL)
1264     {
1265       /* no other requests pending, run immediately */
1266       fun (fun_cls, GNUNET_OK);
1267       return NULL;
1268     }
1269   e = GNUNET_malloc (sizeof (struct DatastoreRequestQueue));
1270   e->timeout = GNUNET_TIME_relative_to_absolute (deadline);
1271   e->req = fun;
1272   e->req_cls = fun_cls;
1273   if (deadline.value == GNUNET_TIME_UNIT_FOREVER_REL.value)
1274     {
1275       /* local request, highest prio, put at head of queue
1276          regardless of deadline */
1277       bef = NULL;
1278     }
1279   else
1280     {
1281       bef = drq_tail;
1282       while ( (NULL != bef) &&
1283               (e->timeout.value < bef->timeout.value) )
1284         bef = bef->prev;
1285     }
1286   GNUNET_CONTAINER_DLL_insert_after (drq_head, drq_tail, bef, e);
1287   if (deadline.value == GNUNET_TIME_UNIT_FOREVER_REL.value)
1288     return e;
1289   e->task = GNUNET_SCHEDULER_add_delayed (sched,
1290                                           deadline,
1291                                           &timeout_ds_request,
1292                                           e);
1293   return e;
1294 }
1295
1296
1297 /**
1298  * Free the state associated with a local get context.
1299  *
1300  * @param lgc the lgc to free
1301  */
1302 static void
1303 local_get_context_free (struct LocalGetContext *lgc)
1304 {
1305   GNUNET_CONTAINER_DLL_remove (lgc_head, lgc_tail, lgc);
1306   GNUNET_SERVER_client_drop (lgc->client);
1307   GNUNET_free_non_null (lgc->results);
1308   if (lgc->results_bf != NULL)
1309     GNUNET_CONTAINER_bloomfilter_free (lgc->results_bf);
1310   if (lgc->req != NULL)
1311     {
1312       if (lgc->req->task != GNUNET_SCHEDULER_NO_TASK)
1313         GNUNET_SCHEDULER_cancel (sched, lgc->req->task);
1314       GNUNET_CONTAINER_DLL_remove (lgc_head, lgc_tail, lgc);
1315       GNUNET_free (lgc->req);
1316     }
1317   GNUNET_free (lgc);
1318 }
1319
1320
1321 /**
1322  * We're able to transmit the next (local) result to the client.
1323  * Do it and ask the datastore for more.  Or, on error, tell
1324  * the datastore to stop giving us more.
1325  *
1326  * @param cls our closure (struct LocalGetContext)
1327  * @param max maximum number of bytes we can transmit
1328  * @param buf where to copy our message
1329  * @return number of bytes copied to buf
1330  */
1331 static size_t
1332 transmit_local_result (void *cls,
1333                        size_t max,
1334                        void *buf)
1335 {
1336   struct LocalGetContext *lgc = cls;
1337   uint16_t msize;
1338
1339   if (NULL == buf)
1340     {
1341 #if DEBUG_FS
1342       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1343                   "Failed to transmit result to local client, aborting datastore iteration.\n");
1344 #endif
1345       /* error, abort! */
1346       GNUNET_free (lgc->result);
1347       lgc->result = NULL;
1348       GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
1349       return 0;
1350     }
1351   msize = ntohs (lgc->result->header.size);
1352 #if DEBUG_FS
1353   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1354               "Transmitting %u bytes of result to local client.\n",
1355               msize);
1356 #endif
1357   GNUNET_assert (max >= msize);
1358   memcpy (buf, lgc->result, msize);
1359   GNUNET_free (lgc->result);
1360   lgc->result = NULL;
1361   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1362   return msize;
1363 }
1364
1365
1366 /**
1367  * Continuation called from datastore's remove
1368  * function.
1369  *
1370  * @param cls unused
1371  * @param success did the deletion work?
1372  * @param msg error message
1373  */
1374 static void
1375 remove_cont (void *cls,
1376              int success,
1377              const char *msg)
1378 {
1379   if (GNUNET_OK != success)
1380     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1381                 _("Failed to delete bogus block: %s\n"),
1382                 msg);
1383   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1384 }
1385
1386
1387 /**
1388  * Mingle hash with the mingle_number to
1389  * produce different bits.
1390  */
1391 static void
1392 mingle_hash (const GNUNET_HashCode * in,
1393              int32_t mingle_number,
1394              GNUNET_HashCode * hc)
1395 {
1396   GNUNET_HashCode m;
1397
1398   GNUNET_CRYPTO_hash (&mingle_number,
1399                       sizeof (int32_t),
1400                       &m);
1401   GNUNET_CRYPTO_hash_xor (&m, in, hc);
1402 }
1403
1404
1405 /**
1406  * We've received an on-demand encoded block
1407  * from the datastore.  Attempt to do on-demand
1408  * encoding and (if successful), call the
1409  * continuation with the resulting block.  On
1410  * error, clean up and ask the datastore for
1411  * more results.
1412  *
1413  * @param key key for the content
1414  * @param size number of bytes in data
1415  * @param data content stored
1416  * @param type type of the content
1417  * @param priority priority of the content
1418  * @param anonymity anonymity-level for the content
1419  * @param expiration expiration time for the content
1420  * @param uid unique identifier for the datum;
1421  *        maybe 0 if no unique identifier is available
1422  * @param cont function to call with the actual block
1423  * @param cont_cls closure for cont
1424  */
1425 static void
1426 handle_on_demand_block (const GNUNET_HashCode * key,
1427                         uint32_t size,
1428                         const void *data,
1429                         uint32_t type,
1430                         uint32_t priority,
1431                         uint32_t anonymity,
1432                         struct GNUNET_TIME_Absolute
1433                         expiration, uint64_t uid,
1434                         GNUNET_DATASTORE_Iterator cont,
1435                         void *cont_cls)
1436 {
1437   const struct OnDemandBlock *odb;
1438   GNUNET_HashCode nkey;
1439   struct GNUNET_CRYPTO_AesSessionKey skey;
1440   struct GNUNET_CRYPTO_AesInitializationVector iv;
1441   GNUNET_HashCode query;
1442   ssize_t nsize;
1443   char ndata[DBLOCK_SIZE];
1444   char edata[DBLOCK_SIZE];
1445   const char *fn;
1446   struct GNUNET_DISK_FileHandle *fh;
1447   uint64_t off;
1448
1449   if (size != sizeof (struct OnDemandBlock))
1450     {
1451       GNUNET_break (0);
1452       GNUNET_DATASTORE_remove (dsh,
1453                                key,
1454                                size,
1455                                data,
1456                                &remove_cont,
1457                                NULL,
1458                                GNUNET_TIME_UNIT_FOREVER_REL);
1459       return;
1460     }
1461   odb = (const struct OnDemandBlock*) data;
1462   off = GNUNET_ntohll (odb->offset);
1463   fn = (const char*) GNUNET_CONTAINER_multihashmap_get (ifm,
1464                                                         &odb->file_id);
1465   fh = NULL;
1466   if ( (NULL == fn) ||
1467        (NULL == (fh = GNUNET_DISK_file_open (fn,
1468                                              GNUNET_DISK_OPEN_READ,
1469                                              GNUNET_DISK_PERM_NONE))) ||
1470        (off !=
1471         GNUNET_DISK_file_seek (fh,
1472                                off,
1473                                GNUNET_DISK_SEEK_SET)) ||
1474        (-1 ==
1475         (nsize = GNUNET_DISK_file_read (fh,
1476                                         ndata,
1477                                         sizeof (ndata)))) )
1478     {
1479       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1480                   _("Could not access indexed file `%s' at offset %llu: %s\n"),
1481                   GNUNET_h2s (&odb->file_id),
1482                   (unsigned long long) off,
1483                   STRERROR (errno));
1484       if (fh != NULL)
1485         GNUNET_DISK_file_close (fh);
1486       /* FIXME: if this happens often, we need
1487          to remove the OnDemand block from the DS! */
1488       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1489       return;
1490     }
1491   GNUNET_DISK_file_close (fh);
1492   GNUNET_CRYPTO_hash (ndata,
1493                       nsize,
1494                       &nkey);
1495   GNUNET_CRYPTO_hash_to_aes_key (&nkey, &skey, &iv);
1496   GNUNET_CRYPTO_aes_encrypt (ndata,
1497                              nsize,
1498                              &skey,
1499                              &iv,
1500                              edata);
1501   GNUNET_CRYPTO_hash (edata,
1502                       nsize,
1503                       &query);
1504   if (0 != memcmp (&query,
1505                    key,
1506                    sizeof (GNUNET_HashCode)))
1507     {
1508       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1509                   _("Indexed file `%s' changed at offset %llu\n"),
1510                   fn,
1511                   (unsigned long long) off);
1512       /* FIXME: if this happens often, we need
1513          to remove the OnDemand block from the DS! */
1514       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1515       return;
1516     }
1517   cont (cont_cls,
1518         key,
1519         nsize,
1520         edata,
1521         GNUNET_DATASTORE_BLOCKTYPE_DBLOCK,
1522         priority,
1523         anonymity,
1524         expiration,
1525         uid);
1526 }
1527
1528
1529 /**
1530  * How many bytes should a bloomfilter be if we have already seen
1531  * entry_count responses?  Note that BLOOMFILTER_K gives us the number
1532  * of bits set per entry.  Furthermore, we should not re-size the
1533  * filter too often (to keep it cheap).
1534  *
1535  * Since other peers will also add entries but not resize the filter,
1536  * we should generally pick a slightly larger size than what the
1537  * strict math would suggest.
1538  *
1539  * @return must be a power of two and smaller or equal to 2^15.
1540  */
1541 static size_t
1542 compute_bloomfilter_size (unsigned int entry_count)
1543 {
1544   size_t size;
1545   unsigned int ideal = (entry_count * BLOOMFILTER_K) / 4;
1546   uint16_t max = 1 << 15;
1547
1548   if (entry_count > max)
1549     return max;
1550   size = 8;
1551   while ((size < max) && (size < ideal))
1552     size *= 2;
1553   if (size > max)
1554     return max;
1555   return size;
1556 }
1557
1558
1559 /**
1560  * Recalculate our bloom filter for filtering replies.
1561  *
1562  * @param count number of entries we are filtering right now
1563  * @param mingle set to our new mingling value
1564  * @param bf_size set to the size of the bloomfilter
1565  * @param entries the entries to filter
1566  * @return updated bloomfilter, NULL for none
1567  */
1568 static struct GNUNET_CONTAINER_BloomFilter *
1569 refresh_bloomfilter (unsigned int count,
1570                      int32_t * mingle,
1571                      size_t *bf_size,
1572                      const GNUNET_HashCode *entries)
1573 {
1574   struct GNUNET_CONTAINER_BloomFilter *bf;
1575   size_t nsize;
1576   unsigned int i;
1577   GNUNET_HashCode mhash;
1578
1579   if (0 == count)
1580     return NULL;
1581   nsize = compute_bloomfilter_size (count);
1582   *mingle = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, -1);
1583   *bf_size = nsize;
1584   bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
1585                                           nsize,
1586                                           BLOOMFILTER_K);
1587   for (i=0;i<count;i++)
1588     {
1589       mingle_hash (&entries[i], *mingle, &mhash);
1590       GNUNET_CONTAINER_bloomfilter_add (bf, &mhash);
1591     }
1592   return bf;
1593 }
1594
1595
1596 /**
1597  * Closure used for "target_peer_select_cb".
1598  */
1599 struct PeerSelectionContext
1600 {
1601   /**
1602    * The request for which we are selecting
1603    * peers.
1604    */
1605   struct PendingRequest *pr;
1606
1607   /**
1608    * Current "prime" target.
1609    */
1610   struct GNUNET_PeerIdentity target;
1611
1612   /**
1613    * How much do we like this target?
1614    */
1615   double target_score;
1616
1617 };
1618
1619
1620 /**
1621  * Function called for each connected peer to determine
1622  * which one(s) would make good targets for forwarding.
1623  *
1624  * @param cls closure (struct PeerSelectionContext)
1625  * @param key current key code (peer identity)
1626  * @param value value in the hash map (struct ConnectedPeer)
1627  * @return GNUNET_YES if we should continue to
1628  *         iterate,
1629  *         GNUNET_NO if not.
1630  */
1631 static int
1632 target_peer_select_cb (void *cls,
1633                        const GNUNET_HashCode * key,
1634                        void *value)
1635 {
1636   struct PeerSelectionContext *psc = cls;
1637   struct ConnectedPeer *cp = value;
1638   struct PendingRequest *pr = psc->pr;
1639   double score;
1640   unsigned int i;
1641
1642   /* 1) check if we have already (recently) forwarded to this peer */
1643   for (i=0;i<pr->used_pids_off;i++)
1644     if (pr->used_pids[i] == cp->pid)
1645       return GNUNET_YES; /* skip */
1646   // 2) calculate how much we'd like to forward to this peer
1647   score = 0; // FIXME!
1648
1649   /* store best-fit in closure */
1650   if (score > psc->target_score)
1651     {
1652       psc->target_score = score;
1653       psc->target.hashPubKey = *key;
1654     }
1655   return GNUNET_YES;
1656 }
1657
1658
1659 /**
1660  * We use a random delay to make the timing of requests
1661  * less predictable.  This function returns such a random
1662  * delay.
1663  *
1664  * @return random delay to use for some request, between 0 and TTL_DECREMENT ms
1665  */
1666 static struct GNUNET_TIME_Relative
1667 get_processing_delay ()
1668 {
1669   return GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
1670                                         GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1671                                                                   TTL_DECREMENT));
1672 }
1673
1674
1675 /**
1676  * Task that is run for each request with the
1677  * goal of forwarding the associated query to
1678  * other peers.  The task should re-schedule
1679  * itself to be re-run once the TTL has expired.
1680  * (or at a later time if more peers should
1681  * be queried earlier).
1682  *
1683  * @param cls the requests "struct PendingRequest*"
1684  * @param tc task context (unused)
1685  */
1686 static void
1687 forward_request_task (void *cls,
1688                       const struct GNUNET_SCHEDULER_TaskContext *tc);
1689
1690
1691 /**
1692  * We've selected a peer for forwarding of a query.
1693  * Construct the message and then re-schedule the
1694  * task to forward again to (other) peers.
1695  *
1696  * @param cls closure
1697  * @param size number of bytes available in buf
1698  * @param buf where the callee should write the message
1699  * @return number of bytes written to buf
1700  */
1701 static size_t
1702 transmit_request_cb (void *cls,
1703                      size_t size,
1704                      void *buf)
1705 {
1706   struct PendingRequest *pr = cls;
1707   struct GetMessage *gm;
1708   GNUNET_HashCode *ext;
1709   char *bfdata;
1710   size_t msize;
1711   unsigned int k;
1712
1713   pr->cth = NULL;
1714   /* (1) check for timeout */
1715   if (NULL == buf)
1716     {
1717       /* timeout, try another peer immediately again */
1718       pr->task = GNUNET_SCHEDULER_add_with_priority (sched,
1719                                                      GNUNET_SCHEDULER_PRIORITY_IDLE,
1720                                                      &forward_request_task,
1721                                                      pr);
1722       return 0;
1723     }
1724   /* (2) build query message */
1725   k = 0; // FIXME: count hash codes!
1726   msize = sizeof (struct GetMessage) + pr->bf_size + k * sizeof(GNUNET_HashCode);
1727   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
1728   gm = (struct GetMessage*) buf;
1729   gm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_GET);
1730   gm->header.size = htons (msize);
1731   gm->type = htonl (pr->type);
1732   pr->remaining_priority /= 2;
1733   gm->priority = htonl (pr->remaining_priority);
1734   gm->ttl = htonl (pr->ttl);
1735   gm->filter_mutator = htonl(pr->mingle);
1736   gm->hash_bitmap = htonl (42);
1737   gm->query = pr->query;
1738   ext = (GNUNET_HashCode*) &gm[1];
1739   // FIXME: setup "ext[0]..[k-1]"
1740   bfdata = (char *) &ext[k];
1741   if (pr->bf != NULL)
1742     GNUNET_CONTAINER_bloomfilter_get_raw_data (pr->bf,
1743                                                bfdata,
1744                                                pr->bf_size);
1745
1746   /* (3) schedule job to do it again (or another peer, etc.) */
1747   pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1748                                            get_processing_delay (), // FIXME!
1749                                            &forward_request_task,
1750                                            pr);
1751
1752   return msize;
1753 }
1754
1755
1756 /**
1757  * Function called after we've tried to reserve
1758  * a certain amount of bandwidth for a reply.
1759  * Check if we succeeded and if so send our query.
1760  *
1761  * @param cls the requests "struct PendingRequest*"
1762  * @param peer identifies the peer
1763  * @param latency current latency estimate, "FOREVER" if we have been
1764  *                disconnected
1765  * @param bpm_in set to the current bandwidth limit (receiving) for this peer
1766  * @param bpm_out set to the current bandwidth limit (sending) for this peer
1767  * @param amount set to the amount that was actually reserved or unreserved
1768  * @param preference current traffic preference for the given peer
1769  */
1770 static void
1771 target_reservation_cb (void *cls,
1772                        const struct
1773                        GNUNET_PeerIdentity * peer,
1774                        unsigned int bpm_in,
1775                        unsigned int bpm_out,
1776                        struct GNUNET_TIME_Relative
1777                        latency, int amount,
1778                        unsigned long long preference)
1779 {
1780   struct PendingRequest *pr = cls;
1781   uint32_t priority;
1782   uint16_t size;
1783   struct GNUNET_TIME_Relative maxdelay;
1784
1785   GNUNET_assert (peer != NULL);
1786   if ( (amount != DBLOCK_SIZE) ||
1787        (pr->cth != NULL) )
1788     {
1789       /* try again later; FIXME: we may need to un-reserve "amount"? */
1790       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1791                                                get_processing_delay (), // FIXME: longer?
1792                                                &forward_request_task,
1793                                                pr);
1794       return;
1795     }
1796   // (2) transmit, update ttl/priority
1797   // FIXME: calculate priority, maxdelay, size properly!
1798   priority = 0;
1799   size = 60000;
1800   maxdelay = GNUNET_CONSTANTS_SERVICE_TIMEOUT;
1801   pr->cth = GNUNET_CORE_notify_transmit_ready (core,
1802                                                priority,
1803                                                maxdelay,
1804                                                peer,
1805                                                size,
1806                                                &transmit_request_cb,
1807                                                pr);
1808   if (pr->cth == NULL)
1809     {
1810       /* try again later */
1811       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1812                                                get_processing_delay (), // FIXME: longer?
1813                                                &forward_request_task,
1814                                                pr);
1815     }
1816 }
1817
1818
1819 /**
1820  * Task that is run for each request with the
1821  * goal of forwarding the associated query to
1822  * other peers.  The task should re-schedule
1823  * itself to be re-run once the TTL has expired.
1824  * (or at a later time if more peers should
1825  * be queried earlier).
1826  *
1827  * @param cls the requests "struct PendingRequest*"
1828  * @param tc task context (unused)
1829  */
1830 static void
1831 forward_request_task (void *cls,
1832                       const struct GNUNET_SCHEDULER_TaskContext *tc)
1833 {
1834   struct PendingRequest *pr = cls;
1835   struct PeerSelectionContext psc;
1836
1837   pr->task = GNUNET_SCHEDULER_NO_TASK;
1838   if (pr->cth != NULL)
1839     {
1840       /* we're busy transmitting a result, wait a bit */
1841       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1842                                                get_processing_delay (),
1843                                                &forward_request_task,
1844                                                pr);
1845       return;
1846     }
1847   /* (1) select target */
1848   psc.pr = pr;
1849   psc.target_score = DBL_MIN;
1850   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1851                                          &target_peer_select_cb,
1852                                          &psc);
1853   if (psc.target_score == DBL_MIN)
1854     {
1855       /* no possible target found, wait some time */
1856       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1857                                                get_processing_delay (), // FIXME: exponential back-off? or at least wait longer...
1858                                                &forward_request_task,
1859                                                pr);
1860       return;
1861     }
1862   /* (2) reserve reply bandwidth */
1863   // FIXME: need a way to cancel; this
1864   // async operation is problematic (segv-problematic)
1865   // if "pr" is destroyed while it happens!
1866   GNUNET_CORE_peer_configure (core,
1867                               &psc.target,
1868                               GNUNET_CONSTANTS_SERVICE_TIMEOUT,
1869                               -1,
1870                               DBLOCK_SIZE, // FIXME: make dependent on type?
1871                               0,
1872                               &target_reservation_cb,
1873                               pr);
1874 }
1875
1876
1877 /**
1878  * We're processing (local) results for a search request
1879  * from a (local) client.  Pass applicable results to the
1880  * client and if we are done either clean up (operation
1881  * complete) or switch to P2P search (more results possible).
1882  *
1883  * @param cls our closure (struct LocalGetContext)
1884  * @param key key for the content
1885  * @param size number of bytes in data
1886  * @param data content stored
1887  * @param type type of the content
1888  * @param priority priority of the content
1889  * @param anonymity anonymity-level for the content
1890  * @param expiration expiration time for the content
1891  * @param uid unique identifier for the datum;
1892  *        maybe 0 if no unique identifier is available
1893  */
1894 static void
1895 process_local_get_result (void *cls,
1896                           const GNUNET_HashCode * key,
1897                           uint32_t size,
1898                           const void *data,
1899                           uint32_t type,
1900                           uint32_t priority,
1901                           uint32_t anonymity,
1902                           struct GNUNET_TIME_Absolute
1903                           expiration,
1904                           uint64_t uid)
1905 {
1906   struct LocalGetContext *lgc = cls;
1907   struct PendingRequest *pr;
1908   struct ClientRequestList *crl;
1909   struct ClientList *cl;
1910   size_t msize;
1911   unsigned int i;
1912
1913   if (key == NULL)
1914     {
1915 #if DEBUG_FS
1916       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1917                   "Received last result for `%s' from local datastore, deciding what to do next.\n",
1918                   GNUNET_h2s (&lgc->query));
1919 #endif
1920       /* no further results from datastore; continue
1921          processing further requests from the client and
1922          allow the next task to use the datastore; also,
1923          switch to P2P requests or clean up our state. */
1924       next_ds_request ();
1925       GNUNET_SERVER_receive_done (lgc->client,
1926                                   GNUNET_OK);
1927       if ( (lgc->results_used == 0) ||
1928            (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
1929            (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ||
1930            (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK) )
1931         {
1932 #if DEBUG_FS
1933           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1934                       "Forwarding query for `%s' to network.\n",
1935                       GNUNET_h2s (&lgc->query));
1936 #endif
1937           cl = clients;
1938           while ( (NULL != cl) &&
1939                   (cl->client != lgc->client) )
1940             cl = cl->next;
1941           if (cl == NULL)
1942             {
1943               cl = GNUNET_malloc (sizeof (struct ClientList));
1944               cl->client = lgc->client;
1945               cl->next = clients;
1946               clients = cl;
1947             }
1948           crl = GNUNET_malloc (sizeof (struct ClientRequestList));
1949           crl->cl = cl;
1950           GNUNET_CONTAINER_DLL_insert (cl->head, cl->tail, crl);
1951           pr = GNUNET_malloc (sizeof (struct PendingRequest));
1952           pr->client = lgc->client;
1953           GNUNET_SERVER_client_keep (pr->client);
1954           pr->crl_entry = crl;
1955           crl->req = pr;
1956           if (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK)
1957             {
1958               pr->namespace = GNUNET_malloc (sizeof (GNUNET_HashCode));
1959               *pr->namespace = lgc->namespace;
1960             }
1961           pr->replies_seen = lgc->results;
1962           lgc->results = NULL;
1963           pr->start_time = GNUNET_TIME_absolute_get ();
1964           pr->query = lgc->query;
1965           pr->target_pid = GNUNET_PEER_intern (&lgc->target);
1966           pr->replies_seen_off = lgc->results_used;
1967           pr->replies_seen_size = lgc->results_size;
1968           lgc->results_size = 0;
1969           pr->type = lgc->type;
1970           pr->anonymity_level = lgc->anonymity_level;
1971           pr->bf = refresh_bloomfilter (pr->replies_seen_off,
1972                                         &pr->mingle,
1973                                         &pr->bf_size,
1974                                         pr->replies_seen);
1975           GNUNET_CONTAINER_multihashmap_put (requests_by_query,
1976                                              &pr->query,
1977                                              pr,
1978                                              GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1979           pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1980                                                    get_processing_delay (),
1981                                                    &forward_request_task,
1982                                                    pr);
1983           local_get_context_free (lgc);
1984           return;
1985         }
1986       /* got all possible results, clean up! */
1987 #if DEBUG_FS
1988       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1989                   "Found all possible results for query for `%s', done!\n",
1990                   GNUNET_h2s (&lgc->query));
1991 #endif
1992       local_get_context_free (lgc);
1993       return;
1994     }
1995   if (type == GNUNET_DATASTORE_BLOCKTYPE_ONDEMAND)
1996     {
1997 #if DEBUG_FS
1998       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1999                   "Received on-demand block for `%s' from local datastore, fetching data.\n",
2000                   GNUNET_h2s (&lgc->query));
2001 #endif
2002       handle_on_demand_block (key, size, data, type, priority,
2003                               anonymity, expiration, uid,
2004                               &process_local_get_result,
2005                               lgc);
2006       return;
2007     }
2008   if ( (type != lgc->type) &&
2009        (lgc->type != GNUNET_DATASTORE_BLOCKTYPE_ANY) )
2010     {
2011       /* this should be virtually impossible to reach (DBLOCK
2012          query hash being identical to KBLOCK/SBLOCK query hash);
2013          nevertheless, if it happens, the correct thing is to
2014          simply skip the result. */
2015 #if DEBUG_FS
2016       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2017                   "Received block of unexpected type (%u, want %u) for `%s' from local datastore, ignoring.\n",
2018                   type,
2019                   lgc->type,
2020                   GNUNET_h2s (&lgc->query));
2021 #endif
2022       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2023       return;
2024     }
2025   /* check if this is a result we've alredy
2026      received */
2027   for (i=0;i<lgc->results_used;i++)
2028     if (0 == memcmp (key,
2029                      &lgc->results[i],
2030                      sizeof (GNUNET_HashCode)))
2031       {
2032 #if DEBUG_FS
2033         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2034                     "Received duplicate result for `%s' from local datastore, ignoring.\n",
2035                     GNUNET_h2s (&lgc->query));
2036 #endif
2037         GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2038         return;
2039       }
2040   if (lgc->results_used == lgc->results_size)
2041     GNUNET_array_grow (lgc->results,
2042                        lgc->results_size,
2043                        lgc->results_size * 2 + 2);
2044   GNUNET_CRYPTO_hash (data,
2045                       size,
2046                       &lgc->results[lgc->results_used++]);
2047   msize = size + sizeof (struct ContentMessage);
2048   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
2049   lgc->result = GNUNET_malloc (msize);
2050   lgc->result->header.size = htons (msize);
2051   lgc->result->header.type = htons (GNUNET_MESSAGE_TYPE_FS_CONTENT);
2052   lgc->result->type = htonl (type);
2053   lgc->result->expiration = GNUNET_TIME_absolute_hton (expiration);
2054   memcpy (&lgc->result[1],
2055           data,
2056           size);
2057 #if DEBUG_FS
2058   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2059               "Received new result for `%s' from local datastore, passing to client.\n",
2060               GNUNET_h2s (&lgc->query));
2061 #endif
2062   GNUNET_SERVER_notify_transmit_ready (lgc->client,
2063                                        msize,
2064                                        GNUNET_TIME_UNIT_FOREVER_REL,
2065                                        &transmit_local_result,
2066                                        lgc);
2067 }
2068
2069
2070 /**
2071  * We're processing a search request from a local
2072  * client.  Now it is our turn to query the datastore.
2073  *
2074  * @param cls our closure (struct LocalGetContext)
2075  * @param tc unused
2076  */
2077 static void
2078 transmit_local_get (void *cls,
2079                     const struct GNUNET_SCHEDULER_TaskContext *tc)
2080 {
2081   struct LocalGetContext *lgc = cls;
2082   uint32_t type;
2083
2084   type = lgc->type;
2085   if (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK)
2086     type = GNUNET_DATASTORE_BLOCKTYPE_ANY; /* to get on-demand as well */
2087   GNUNET_DATASTORE_get (dsh,
2088                         &lgc->query,
2089                         type,
2090                         &process_local_get_result,
2091                         lgc,
2092                         GNUNET_TIME_UNIT_FOREVER_REL);
2093 }
2094
2095
2096 /**
2097  * We're processing a search request from a local
2098  * client.  Now it is our turn to query the datastore.
2099  *
2100  * @param cls our closure (struct LocalGetContext)
2101  * @param ok did we succeed to queue for datastore access, should always be GNUNET_OK
2102  */
2103 static void
2104 transmit_local_get_ready (void *cls,
2105                           int ok)
2106 {
2107   struct LocalGetContext *lgc = cls;
2108
2109   GNUNET_assert (GNUNET_OK == ok);
2110   GNUNET_SCHEDULER_add_continuation (sched,
2111                                      &transmit_local_get,
2112                                      lgc,
2113                                      GNUNET_SCHEDULER_REASON_PREREQ_DONE);
2114 }
2115
2116
2117 /**
2118  * Handle START_SEARCH-message (search request from client).
2119  *
2120  * @param cls closure
2121  * @param client identification of the client
2122  * @param message the actual message
2123  */
2124 static void
2125 handle_start_search (void *cls,
2126                      struct GNUNET_SERVER_Client *client,
2127                      const struct GNUNET_MessageHeader *message)
2128 {
2129   const struct SearchMessage *sm;
2130   struct LocalGetContext *lgc;
2131   uint16_t msize;
2132   unsigned int sc;
2133
2134   msize = ntohs (message->size);
2135   if ( (msize < sizeof (struct SearchMessage)) ||
2136        (0 != (msize - sizeof (struct SearchMessage)) % sizeof (GNUNET_HashCode)) )
2137     {
2138       GNUNET_break (0);
2139       GNUNET_SERVER_receive_done (client,
2140                                   GNUNET_SYSERR);
2141       return;
2142     }
2143   sc = (msize - sizeof (struct SearchMessage)) / sizeof (GNUNET_HashCode);
2144   sm = (const struct SearchMessage*) message;
2145   GNUNET_SERVER_client_keep (client);
2146   lgc = GNUNET_malloc (sizeof (struct LocalGetContext));
2147   if  (sc > 0)
2148     {
2149       lgc->results_used = sc;
2150       GNUNET_array_grow (lgc->results,
2151                          lgc->results_size,
2152                          sc * 2);
2153       memcpy (lgc->results,
2154               &sm[1],
2155               sc * sizeof (GNUNET_HashCode));
2156     }
2157   lgc->client = client;
2158   lgc->type = ntohl (sm->type);
2159   lgc->anonymity_level = ntohl (sm->anonymity_level);
2160   switch (lgc->type)
2161     {
2162     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2163     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2164       lgc->target.hashPubKey = sm->target;
2165       break;
2166     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
2167       lgc->namespace = sm->target;
2168       break;
2169     default:
2170       break;
2171     }
2172   lgc->query = sm->query;
2173   GNUNET_CONTAINER_DLL_insert (lgc_head, lgc_tail, lgc);
2174   lgc->req = queue_ds_request (GNUNET_TIME_UNIT_FOREVER_REL,
2175                                &transmit_local_get_ready,
2176                                lgc);
2177 }
2178
2179
2180 /**
2181  * List of handlers for the messages understood by this
2182  * service.
2183  */
2184 static struct GNUNET_SERVER_MessageHandler handlers[] = {
2185   {&handle_index_start, NULL,
2186    GNUNET_MESSAGE_TYPE_FS_INDEX_START, 0},
2187   {&handle_index_list_get, NULL,
2188    GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_GET, sizeof(struct GNUNET_MessageHeader) },
2189   {&handle_unindex, NULL, GNUNET_MESSAGE_TYPE_FS_UNINDEX,
2190    sizeof (struct UnindexMessage) },
2191   {&handle_start_search, NULL, GNUNET_MESSAGE_TYPE_FS_START_SEARCH,
2192    0 },
2193   {NULL, NULL, 0, 0}
2194 };
2195
2196
2197 /**
2198  * Clean up the memory used by the PendingRequest structure (except
2199  * for the client or peer list that the request may be part of).
2200  *
2201  * @param pr request to clean up
2202  */
2203 static void
2204 destroy_pending_request (struct PendingRequest *pr)
2205 {
2206   struct PendingReply *reply;
2207   struct ClientList *cl;
2208
2209   GNUNET_CONTAINER_multihashmap_remove (requests_by_query,
2210                                         &pr->query,
2211                                         pr);
2212   // FIXME: not sure how this can work (efficiently)
2213   // also, what does the return value mean?
2214   if (pr->client == NULL)
2215     {
2216       GNUNET_CONTAINER_heap_remove_node (requests_by_expiration,
2217                                          pr);
2218     }
2219   else
2220     {
2221       cl = pr->crl_entry->cl;
2222       GNUNET_CONTAINER_DLL_remove (cl->head,
2223                                    cl->tail,
2224                                    pr->crl_entry);
2225     }
2226   if (GNUNET_SCHEDULER_NO_TASK != pr->task)
2227     GNUNET_SCHEDULER_cancel (sched, pr->task);
2228   if (NULL != pr->cth)
2229     GNUNET_CORE_notify_transmit_ready_cancel (pr->cth);
2230   if (NULL != pr->bf)
2231     GNUNET_CONTAINER_bloomfilter_free (pr->bf);
2232   if (NULL != pr->th)
2233     GNUNET_CONNECTION_notify_transmit_ready_cancel (pr->th);
2234   while (NULL != (reply = pr->replies_pending))
2235     {
2236       pr->replies_pending = reply->next;
2237       GNUNET_free (reply);
2238     }
2239   GNUNET_PEER_change_rc (pr->source_pid, -1);
2240   GNUNET_PEER_change_rc (pr->target_pid, -1);
2241   GNUNET_PEER_decrement_rcs (pr->used_pids, pr->used_pids_off);
2242   GNUNET_free_non_null (pr->used_pids);
2243   GNUNET_free_non_null (pr->replies_seen);
2244   GNUNET_free_non_null (pr->namespace);
2245   GNUNET_free (pr);
2246 }
2247
2248
2249 /**
2250  * A client disconnected.  Remove all of its pending queries.
2251  *
2252  * @param cls closure, NULL
2253  * @param client identification of the client
2254  */
2255 static void
2256 handle_client_disconnect (void *cls,
2257                           struct GNUNET_SERVER_Client
2258                           * client)
2259 {
2260   struct LocalGetContext *lgc;
2261   struct ClientList *cpos;
2262   struct ClientList *cprev;
2263   struct ClientRequestList *rl;
2264
2265   lgc = lgc_head;
2266   while ( (NULL != lgc) &&
2267           (lgc->client != client) )
2268     lgc = lgc->next;
2269   if (lgc != NULL)
2270     local_get_context_free (lgc);
2271   cprev = NULL;
2272   cpos = clients;
2273   while ( (NULL != cpos) &&
2274           (clients->client != client) )
2275     {
2276       cprev = cpos;
2277       cpos = cpos->next;
2278     }
2279   if (cpos != NULL)
2280     {
2281       if (cprev == NULL)
2282         clients = cpos->next;
2283       else
2284         cprev->next = cpos->next;
2285       while (NULL != (rl = cpos->head))
2286         {
2287           cpos->head = rl->next;
2288           destroy_pending_request (rl->req);
2289           GNUNET_free (rl);
2290         }
2291       GNUNET_free (cpos);
2292     }
2293 }
2294
2295
2296 /**
2297  * Iterator over entries in the "requests_by_query" map
2298  * that frees all the entries.
2299  *
2300  * @param cls closure, NULL
2301  * @param key current key code (the query, unused)
2302  * @param value value in the hash map, of type "struct PendingRequest*"
2303  * @return GNUNET_YES (we should continue to  iterate)
2304  */
2305 static int
2306 destroy_pending_request_cb (void *cls,
2307                             const GNUNET_HashCode * key,
2308                             void *value)
2309 {
2310   struct PendingRequest *pr = value;
2311
2312   destroy_pending_request (pr);
2313   return GNUNET_YES;
2314 }
2315
2316
2317 /**
2318  * Task run during shutdown.
2319  *
2320  * @param cls unused
2321  * @param tc unused
2322  */
2323 static void
2324 shutdown_task (void *cls,
2325                const struct GNUNET_SCHEDULER_TaskContext *tc)
2326 {
2327   struct IndexInfo *pos;
2328
2329   if (NULL != core)
2330     {
2331       GNUNET_CORE_disconnect (core);
2332       core = NULL;
2333     }
2334   if (NULL != dsh)
2335     {
2336       GNUNET_DATASTORE_disconnect (dsh,
2337                                    GNUNET_NO);
2338       dsh = NULL;
2339     }
2340   GNUNET_CONTAINER_multihashmap_iterate (requests_by_query,
2341                                          &destroy_pending_request_cb,
2342                                          NULL);
2343   while (clients != NULL)
2344     handle_client_disconnect (NULL,
2345                               clients->client);
2346   GNUNET_CONTAINER_multihashmap_destroy (requests_by_query);
2347   requests_by_query = NULL;
2348   GNUNET_CONTAINER_multihashmap_destroy (requests_by_peer);
2349   requests_by_peer = NULL;
2350   GNUNET_CONTAINER_heap_destroy (requests_by_expiration);
2351   requests_by_expiration = NULL;
2352   // FIXME: iterate over entries and free individually?
2353   // (or do we get disconnect notifications?)
2354   GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
2355   connected_peers = NULL;
2356   GNUNET_CONTAINER_multihashmap_destroy (ifm);
2357   ifm = NULL;
2358   while (NULL != (pos = indexed_files))
2359     {
2360       indexed_files = pos->next;
2361       GNUNET_free (pos);
2362     }
2363 }
2364
2365
2366 /**
2367  * Free (each) request made by the peer.
2368  *
2369  * @param cls closure, points to peer that the request belongs to
2370  * @param key current key code
2371  * @param value value in the hash map
2372  * @return GNUNET_YES (we should continue to iterate)
2373  */
2374 static int
2375 destroy_request (void *cls,
2376                  const GNUNET_HashCode * key,
2377                  void *value)
2378 {
2379   const struct GNUNET_PeerIdentity * peer = cls;
2380   struct PendingRequest *pr = value;
2381
2382   GNUNET_CONTAINER_multihashmap_remove (requests_by_peer,
2383                                         &peer->hashPubKey,
2384                                         pr);
2385   destroy_pending_request (pr);
2386   return GNUNET_YES;
2387 }
2388
2389
2390
2391 /**
2392  * Method called whenever a given peer connects.
2393  *
2394  * @param cls closure, not used
2395  * @param peer peer identity this notification is about
2396  */
2397 static void
2398 peer_connect_handler (void *cls,
2399                       const struct
2400                       GNUNET_PeerIdentity * peer)
2401 {
2402   struct ConnectedPeer *cp;
2403
2404   cp = GNUNET_malloc (sizeof (struct ConnectedPeer));
2405   cp->pid = GNUNET_PEER_intern (peer);
2406   GNUNET_CONTAINER_multihashmap_put (connected_peers,
2407                                      &peer->hashPubKey,
2408                                      cp,
2409                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
2410 }
2411
2412
2413 /**
2414  * Method called whenever a peer disconnects.
2415  *
2416  * @param cls closure, not used
2417  * @param peer peer identity this notification is about
2418  */
2419 static void
2420 peer_disconnect_handler (void *cls,
2421                          const struct
2422                          GNUNET_PeerIdentity * peer)
2423 {
2424   struct ConnectedPeer *cp;
2425
2426   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
2427                                           &peer->hashPubKey);
2428   GNUNET_PEER_change_rc (cp->pid, -1);
2429   GNUNET_PEER_decrement_rcs (cp->last_p2p_replies, P2P_SUCCESS_LIST_SIZE);
2430   GNUNET_free (cp);
2431   GNUNET_CONTAINER_multihashmap_get_multiple (requests_by_peer,
2432                                               &peer->hashPubKey,
2433                                               &destroy_request,
2434                                               (void*) peer);
2435 }
2436
2437
2438 /**
2439  * We're processing a GET request from
2440  * another peer and have decided to forward
2441  * it to other peers.
2442  *
2443  * @param cls our "struct ProcessGetContext *"
2444  * @param tc unused
2445  */
2446 static void
2447 forward_get_request (void *cls,
2448                      const struct GNUNET_SCHEDULER_TaskContext *tc)
2449 {
2450   struct ProcessGetContext *pgc = cls;
2451   struct PendingRequest *pr;
2452   struct PendingRequest *eer;
2453   struct GNUNET_PeerIdentity target;
2454
2455   pr = GNUNET_malloc (sizeof (struct PendingRequest));
2456   if (GET_MESSAGE_BIT_SKS_NAMESPACE == (GET_MESSAGE_BIT_SKS_NAMESPACE & pgc->bm))
2457     {
2458       pr->namespace = GNUNET_malloc (sizeof(GNUNET_HashCode));
2459       *pr->namespace = pgc->namespace;
2460     }
2461   pr->bf = pgc->bf;
2462   pr->bf_size = pgc->bf_size;
2463   pgc->bf = NULL;
2464   pr->start_time = pgc->start_time;
2465   pr->query = pgc->query;
2466   pr->source_pid = GNUNET_PEER_intern (&pgc->reply_to);
2467   if (GET_MESSAGE_BIT_TRANSMIT_TO == (GET_MESSAGE_BIT_TRANSMIT_TO & pgc->bm))
2468     pr->target_pid = GNUNET_PEER_intern (&pgc->prime_target);
2469   pr->anonymity_level = 1; /* default */
2470   pr->priority = pgc->priority;
2471   pr->remaining_priority = pr->priority;
2472   pr->mingle = pgc->mingle;
2473   pr->ttl = pgc->ttl;
2474   pr->type = pgc->type;
2475   GNUNET_CONTAINER_multihashmap_put (requests_by_query,
2476                                      &pr->query,
2477                                      pr,
2478                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
2479   GNUNET_CONTAINER_multihashmap_put (requests_by_peer,
2480                                      &pgc->reply_to.hashPubKey,
2481                                      pr,
2482                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
2483   GNUNET_CONTAINER_heap_insert (requests_by_expiration,
2484                                 pr,
2485                                 pr->start_time.value + pr->ttl);
2486   if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration) > max_pending_requests)
2487     {
2488       /* expire oldest request! */
2489       eer = GNUNET_CONTAINER_heap_peek (requests_by_expiration);
2490       GNUNET_PEER_resolve (eer->source_pid,
2491                            &target);
2492       GNUNET_CONTAINER_multihashmap_remove (requests_by_peer,
2493                                             &target.hashPubKey,
2494                                             eer);
2495       destroy_pending_request (eer);
2496     }
2497   pr->task = GNUNET_SCHEDULER_add_delayed (sched,
2498                                            get_processing_delay (),
2499                                            &forward_request_task,
2500                                            pr);
2501   GNUNET_free (pgc);
2502 }
2503 /**
2504  * Transmit the given message by copying it to
2505  * the target buffer "buf".  "buf" will be
2506  * NULL and "size" zero if the socket was closed for
2507  * writing in the meantime.  In that case, only
2508
2509  * free the message
2510  *
2511  * @param cls closure, pointer to the message
2512  * @param size number of bytes available in buf
2513  * @param buf where the callee should write the message
2514  * @return number of bytes written to buf
2515  */
2516 static size_t
2517 transmit_message (void *cls,
2518                   size_t size, void *buf)
2519 {
2520   struct GNUNET_MessageHeader *msg = cls;
2521   uint16_t msize;
2522
2523   if (NULL == buf)
2524     {
2525 #if DEBUG_FS
2526       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2527                   "Dropping reply, core too busy.\n");
2528 #endif
2529       GNUNET_free (msg);
2530       return 0;
2531     }
2532   msize = ntohs (msg->size);
2533   GNUNET_assert (size >= msize);
2534   memcpy (buf, msg, msize);
2535   GNUNET_free (msg);
2536   return msize;
2537 }
2538
2539
2540 /**
2541  * Test if the load on this peer is too high
2542  * to even consider processing the query at
2543  * all.
2544  *
2545  * @return GNUNET_YES if the load is too high, GNUNET_NO otherwise
2546  */
2547 static int
2548 test_load_too_high ()
2549 {
2550   return GNUNET_NO; // FIXME
2551 }
2552
2553
2554 /**
2555  * We're processing (local) results for a search request
2556  * from another peer.  Pass applicable results to the
2557  * peer and if we are done either clean up (operation
2558  * complete) or forward to other peers (more results possible).
2559  *
2560  * @param cls our closure (struct LocalGetContext)
2561  * @param key key for the content
2562  * @param size number of bytes in data
2563  * @param data content stored
2564  * @param type type of the content
2565  * @param priority priority of the content
2566  * @param anonymity anonymity-level for the content
2567  * @param expiration expiration time for the content
2568  * @param uid unique identifier for the datum;
2569  *        maybe 0 if no unique identifier is available
2570  */
2571 static void
2572 process_p2p_get_result (void *cls,
2573                         const GNUNET_HashCode * key,
2574                         uint32_t size,
2575                         const void *data,
2576                         uint32_t type,
2577                         uint32_t priority,
2578                         uint32_t anonymity,
2579                         struct GNUNET_TIME_Absolute
2580                         expiration,
2581                         uint64_t uid)
2582 {
2583   struct ProcessGetContext *pgc = cls;
2584   GNUNET_HashCode dhash;
2585   GNUNET_HashCode mhash;
2586   struct PutMessage *reply;
2587
2588   if (NULL == key)
2589     {
2590       /* no more results */
2591       if ( ( (pgc->policy & ROUTING_POLICY_FORWARD) ==  ROUTING_POLICY_FORWARD) &&
2592            ( (0 == pgc->results_found) ||
2593              (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
2594              (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ||
2595              (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK) ) )
2596         {
2597           GNUNET_SCHEDULER_add_continuation (sched,
2598                                              &forward_get_request,
2599                                              pgc,
2600                                              GNUNET_SCHEDULER_REASON_PREREQ_DONE);
2601         }
2602       else
2603         {
2604           if (pgc->bf != NULL)
2605             GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2606           GNUNET_free (pgc);
2607         }
2608       next_ds_request ();
2609       return;
2610     }
2611   if (type == GNUNET_DATASTORE_BLOCKTYPE_ONDEMAND)
2612     {
2613       handle_on_demand_block (key, size, data, type, priority,
2614                               anonymity, expiration, uid,
2615                               &process_p2p_get_result,
2616                               pgc);
2617       return;
2618     }
2619   /* check for duplicates */
2620   GNUNET_CRYPTO_hash (data, size, &dhash);
2621   mingle_hash (&dhash,
2622                pgc->mingle,
2623                &mhash);
2624   if ( (pgc->bf != NULL) &&
2625        (GNUNET_YES ==
2626         GNUNET_CONTAINER_bloomfilter_test (pgc->bf,
2627                                            &mhash)) )
2628     {
2629 #if DEBUG_FS
2630       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2631                   "Result from datastore filtered by bloomfilter.\n");
2632 #endif
2633       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2634       return;
2635     }
2636   pgc->results_found++;
2637   if ( (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
2638        (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ||
2639        (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK) )
2640     {
2641       if (pgc->bf == NULL)
2642         {
2643           pgc->bf_size = 32;
2644           pgc->bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
2645                                                        pgc->bf_size,
2646                                                        BLOOMFILTER_K);
2647         }
2648       GNUNET_CONTAINER_bloomfilter_add (pgc->bf,
2649                                         &mhash);
2650     }
2651
2652   reply = GNUNET_malloc (sizeof (struct PutMessage) + size);
2653   reply->header.size = htons (sizeof (struct PutMessage) + size);
2654   reply->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
2655   reply->type = htonl (type);
2656   reply->expiration = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining (expiration));
2657   memcpy (&reply[1], data, size);
2658   GNUNET_CORE_notify_transmit_ready (core,
2659                                      pgc->priority,
2660                                      ACCEPTABLE_REPLY_DELAY,
2661                                      &pgc->reply_to,
2662                                      sizeof (struct PutMessage) + size,
2663                                      &transmit_message,
2664                                      reply);
2665   if ( (GNUNET_YES == test_load_too_high()) ||
2666        (pgc->results_found > 5 + 2 * pgc->priority) )
2667     {
2668       GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
2669       pgc->policy &= ~ ROUTING_POLICY_FORWARD;
2670       return;
2671     }
2672   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2673 }
2674
2675
2676 /**
2677  * We're processing a GET request from another peer.  Give it to our
2678  * local datastore.
2679  *
2680  * @param cls our "struct ProcessGetContext"
2681  * @param ok did we get a datastore slice or not?
2682  */
2683 static void
2684 ds_get_request (void *cls,
2685                 int ok)
2686 {
2687   struct ProcessGetContext *pgc = cls;
2688   uint32_t type;
2689   struct GNUNET_TIME_Relative timeout;
2690
2691   if (GNUNET_OK != ok)
2692     {
2693       /* no point in doing P2P stuff if we can't even do local */
2694       GNUNET_free (dsh);
2695       return;
2696     }
2697   type = pgc->type;
2698   if (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK)
2699     type = GNUNET_DATASTORE_BLOCKTYPE_ANY; /* to get on-demand as well */
2700   timeout = GNUNET_TIME_relative_multiply (BASIC_DATASTORE_REQUEST_DELAY,
2701                                            (pgc->priority + 1));
2702   GNUNET_DATASTORE_get (dsh,
2703                         &pgc->query,
2704                         type,
2705                         &process_p2p_get_result,
2706                         pgc,
2707                         timeout);
2708 }
2709
2710
2711 /**
2712  * The priority level imposes a bound on the maximum
2713  * value for the ttl that can be requested.
2714  *
2715  * @param ttl_in requested ttl
2716  * @param prio given priority
2717  * @return ttl_in if ttl_in is below the limit,
2718  *         otherwise the ttl-limit for the given priority
2719  */
2720 static int32_t
2721 bound_ttl (int32_t ttl_in, uint32_t prio)
2722 {
2723   unsigned long long allowed;
2724
2725   if (ttl_in <= 0)
2726     return ttl_in;
2727   allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000;
2728   if (ttl_in > allowed)
2729     {
2730       if (allowed >= (1 << 30))
2731         return 1 << 30;
2732       return allowed;
2733     }
2734   return ttl_in;
2735 }
2736
2737
2738 /**
2739  * We've received a request with the specified
2740  * priority.  Bound it according to how much
2741  * we trust the given peer.
2742  *
2743  * @param prio_in requested priority
2744  * @param peer the peer making the request
2745  * @return effective priority
2746  */
2747 static uint32_t
2748 bound_priority (uint32_t prio_in,
2749                 const struct GNUNET_PeerIdentity *peer)
2750 {
2751   return 0; // FIXME!
2752 }
2753
2754
2755 /**
2756  * Handle P2P "GET" request.
2757  *
2758  * @param cls closure, always NULL
2759  * @param other the other peer involved (sender or receiver, NULL
2760  *        for loopback messages where we are both sender and receiver)
2761  * @param message the actual message
2762  * @return GNUNET_OK to keep the connection open,
2763  *         GNUNET_SYSERR to close it (signal serious error)
2764  */
2765 static int
2766 handle_p2p_get (void *cls,
2767                 const struct GNUNET_PeerIdentity *other,
2768                 const struct GNUNET_MessageHeader *message)
2769 {
2770   uint16_t msize;
2771   const struct GetMessage *gm;
2772   unsigned int bits;
2773   const GNUNET_HashCode *opt;
2774   struct ProcessGetContext *pgc;
2775   uint32_t bm;
2776   size_t bfsize;
2777   uint32_t ttl_decrement;
2778   double preference;
2779   int net_load_up;
2780   int net_load_down;
2781
2782   msize = ntohs(message->size);
2783   if (msize < sizeof (struct GetMessage))
2784     {
2785       GNUNET_break_op (0);
2786       return GNUNET_SYSERR;
2787     }
2788   gm = (const struct GetMessage*) message;
2789   bm = ntohl (gm->hash_bitmap);
2790   bits = 0;
2791   while (bm > 0)
2792     {
2793       if (1 == (bm & 1))
2794         bits++;
2795       bm >>= 1;
2796     }
2797   if (msize < sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode))
2798     {
2799       GNUNET_break_op (0);
2800       return GNUNET_SYSERR;
2801     }
2802   opt = (const GNUNET_HashCode*) &gm[1];
2803   bfsize = msize - sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode);
2804   pgc = GNUNET_malloc (sizeof (struct ProcessGetContext));
2805   if (bfsize > 0)
2806     {
2807       pgc->bf = GNUNET_CONTAINER_bloomfilter_init ((const char*) &pgc[1],
2808                                                    bfsize,
2809                                                    BLOOMFILTER_K);
2810       pgc->bf_size = bfsize;
2811     }
2812   pgc->type = ntohl (gm->type);
2813   pgc->bm = ntohl (gm->hash_bitmap);
2814   pgc->mingle = gm->filter_mutator;
2815   bits = 0;
2816   if (0 != (pgc->bm & GET_MESSAGE_BIT_RETURN_TO))
2817     pgc->reply_to.hashPubKey = opt[bits++];
2818   else
2819     pgc->reply_to = *other;
2820   if (0 != (pgc->bm & GET_MESSAGE_BIT_SKS_NAMESPACE))
2821     pgc->namespace = opt[bits++];
2822   else if (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK)
2823     {
2824       GNUNET_break_op (0);
2825       GNUNET_free (pgc);
2826       return GNUNET_SYSERR;
2827     }
2828   if (0 != (pgc->bm & GET_MESSAGE_BIT_TRANSMIT_TO))
2829     pgc->prime_target.hashPubKey = opt[bits++];
2830   /* note that we can really only check load here since otherwise
2831      peers could find out that we are overloaded by being disconnected
2832      after sending us a malformed query... */
2833   if (GNUNET_YES == test_load_too_high ())
2834     {
2835       if (NULL != pgc->bf)
2836         GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2837       GNUNET_free (pgc);
2838 #if DEBUG_FS
2839       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2840                   "Dropping query from `%s', this peer is too busy.\n",
2841                   GNUNET_i2s (other));
2842 #endif
2843       return GNUNET_OK;
2844     }
2845   net_load_up = 50; // FIXME
2846   net_load_down = 50; // FIXME
2847   pgc->policy = ROUTING_POLICY_NONE;
2848   if ( (net_load_up < IDLE_LOAD_THRESHOLD) &&
2849        (net_load_down < IDLE_LOAD_THRESHOLD) )
2850     {
2851       pgc->policy |= ROUTING_POLICY_ALL;
2852       pgc->priority = 0; /* no charge */
2853     }
2854   else
2855     {
2856       pgc->priority = bound_priority (ntohl (gm->priority), other);
2857       if ( (net_load_up <
2858             IDLE_LOAD_THRESHOLD + pgc->priority * pgc->priority) &&
2859            (net_load_down <
2860             IDLE_LOAD_THRESHOLD + pgc->priority * pgc->priority) )
2861         {
2862           pgc->policy |= ROUTING_POLICY_ALL;
2863         }
2864       else
2865         {
2866           // FIXME: is this sound?
2867           if (net_load_up < 90 + 10 * pgc->priority)
2868             pgc->policy |= ROUTING_POLICY_FORWARD;
2869           if (net_load_down < 90 + 10 * pgc->priority)
2870             pgc->policy |= ROUTING_POLICY_ANSWER;
2871         }
2872     }
2873   if (pgc->policy == ROUTING_POLICY_NONE)
2874     {
2875 #if DEBUG_FS
2876       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2877                   "Dropping query from `%s', network saturated.\n",
2878                   GNUNET_i2s (other));
2879 #endif
2880       if (NULL != pgc->bf)
2881         GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2882       GNUNET_free (pgc);
2883       return GNUNET_OK;     /* drop */
2884     }
2885   if ((pgc->policy & ROUTING_POLICY_INDIRECT) != ROUTING_POLICY_INDIRECT)
2886     pgc->priority = 0;  /* kill the priority (we cannot benefit) */
2887   pgc->ttl = bound_ttl (ntohl (gm->ttl), pgc->priority);
2888   /* decrement ttl (always) */
2889   ttl_decrement = 2 * TTL_DECREMENT +
2890     GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2891                               TTL_DECREMENT);
2892   if ( (pgc->ttl < 0) &&
2893        (pgc->ttl - ttl_decrement > 0) )
2894     {
2895 #if DEBUG_FS
2896       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2897                   "Dropping query from `%s' due to TTL underflow.\n",
2898                   GNUNET_i2s (other));
2899 #endif
2900       /* integer underflow => drop (should be very rare)! */
2901       if (NULL != pgc->bf)
2902         GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2903       GNUNET_free (pgc);
2904       return GNUNET_OK;
2905     }
2906   pgc->ttl -= ttl_decrement;
2907   pgc->start_time = GNUNET_TIME_absolute_get ();
2908   preference = (double) pgc->priority;
2909   if (preference < QUERY_BANDWIDTH_VALUE)
2910     preference = QUERY_BANDWIDTH_VALUE;
2911   // FIXME: also reserve bandwidth for reply?
2912   GNUNET_CORE_peer_configure (core,
2913                               other,
2914                               GNUNET_TIME_UNIT_FOREVER_REL,
2915                               0, 0, preference, NULL, NULL);
2916   if (0 != (pgc->policy & ROUTING_POLICY_ANSWER))
2917     pgc->drq = queue_ds_request (BASIC_DATASTORE_REQUEST_DELAY,
2918                                  &ds_get_request,
2919                                  pgc);
2920   else
2921     GNUNET_SCHEDULER_add_continuation (sched,
2922                                        &forward_get_request,
2923                                        pgc,
2924                                        GNUNET_SCHEDULER_REASON_PREREQ_DONE);
2925   return GNUNET_OK;
2926 }
2927
2928
2929 /**
2930  * Function called to notify us that we can now transmit a reply to a
2931  * client or peer.  "buf" will be NULL and "size" zero if the socket was
2932  * closed for writing in the meantime.
2933  *
2934  * @param cls closure, points to a "struct PendingRequest*" with
2935  *            one or more pending replies
2936  * @param size number of bytes available in buf
2937  * @param buf where the callee should write the message
2938  * @return number of bytes written to buf
2939  */
2940 static size_t
2941 transmit_result (void *cls,
2942                  size_t size,
2943                  void *buf)
2944 {
2945   struct PendingRequest *pr = cls;
2946   char *cbuf = buf;
2947   struct PendingReply *reply;
2948   size_t ret;
2949
2950   ret = 0;
2951   while (NULL != (reply = pr->replies_pending))
2952     {
2953       if ( (reply->msize + ret < ret) ||
2954            (reply->msize + ret > size) )
2955         break;
2956       pr->replies_pending = reply->next;
2957       memcpy (&cbuf[ret], &reply[1], reply->msize);
2958       ret += reply->msize;
2959       GNUNET_free (reply);
2960     }
2961   return ret;
2962 }
2963
2964
2965 /**
2966  * Iterator over pending requests.
2967  *
2968  * @param cls response (struct ProcessReplyClosure)
2969  * @param key our query
2970  * @param value value in the hash map (meta-info about the query)
2971  * @return GNUNET_YES (we should continue to iterate)
2972  */
2973 static int
2974 process_reply (void *cls,
2975                const GNUNET_HashCode * key,
2976                void *value)
2977 {
2978   struct ProcessReplyClosure *prq = cls;
2979   struct PendingRequest *pr = value;
2980   struct PendingRequest *eer;
2981   struct PendingReply *reply;
2982   struct PutMessage *pm;
2983   struct ContentMessage *cm;
2984   GNUNET_HashCode chash;
2985   GNUNET_HashCode mhash;
2986   struct GNUNET_PeerIdentity target;
2987   size_t msize;
2988   uint32_t prio;
2989   struct GNUNET_TIME_Relative max_delay;
2990
2991   GNUNET_CRYPTO_hash (prq->data,
2992                       prq->size,
2993                       &chash);
2994   switch (prq->type)
2995     {
2996     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2997     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2998       break;
2999     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
3000       /* FIXME: does prq->namespace match our expectations? */
3001       /* then: fall-through??? */
3002     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
3003       if (pr->bf != NULL)
3004         {
3005           mingle_hash (&chash, pr->mingle, &mhash);
3006           if (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (pr->bf,
3007                                                                &mhash))
3008             return GNUNET_YES; /* duplicate */
3009           GNUNET_CONTAINER_bloomfilter_add (pr->bf,
3010                                             &mhash);
3011         }
3012       break;
3013     case GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK:
3014       // FIXME: any checks against duplicates for SKBlocks?
3015       break;
3016     }
3017   prio = pr->priority;
3018   prq->priority += pr->remaining_priority;
3019   pr->remaining_priority = 0;
3020   if (pr->client != NULL)
3021     {
3022       if (pr->replies_seen_size == pr->replies_seen_off)
3023         GNUNET_array_grow (pr->replies_seen,
3024                            pr->replies_seen_size,
3025                            pr->replies_seen_size * 2 + 4);
3026       pr->replies_seen[pr->replies_seen_off++] = chash;
3027       // FIXME: possibly recalculate BF!
3028     }
3029   if (pr->client == NULL)
3030     {
3031       msize = sizeof (struct ContentMessage) + prq->size;
3032       reply = GNUNET_malloc (msize + sizeof (struct PendingReply));
3033       reply->msize = msize;
3034       cm = (struct ContentMessage*) &reply[1];
3035       cm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_CONTENT);
3036       cm->header.size = htons (msize);
3037       cm->type = htonl (prq->type);
3038       cm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
3039       reply->next = pr->replies_pending;
3040       pr->replies_pending = reply;
3041       memcpy (&reply[1], prq->data, prq->size);
3042       if (pr->cth != NULL)
3043         return GNUNET_YES;
3044       max_delay = GNUNET_TIME_UNIT_FOREVER_REL;
3045       if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration) >= max_pending_requests)
3046         {
3047           /* estimate expiration time from time difference between
3048              first request that will be discarded and this request */
3049           eer = GNUNET_CONTAINER_heap_peek (requests_by_expiration);
3050           max_delay = GNUNET_TIME_absolute_get_difference (pr->start_time,
3051                                                            eer->start_time);
3052         }
3053       GNUNET_PEER_resolve (pr->source_pid,
3054                            &target);
3055       pr->cth = GNUNET_CORE_notify_transmit_ready (core,
3056                                                    prio,
3057                                                    max_delay,
3058                                                    &target,
3059                                                    msize,
3060                                                    &transmit_result,
3061                                                    pr);
3062       if (NULL == pr->cth)
3063         {
3064           // FIXME: now what? discard?
3065         }
3066     }
3067   else
3068     {
3069       msize = sizeof (struct PutMessage) + prq->size;
3070       reply = GNUNET_malloc (msize + sizeof (struct PendingReply));
3071       reply->msize = msize;
3072       reply->next = pr->replies_pending;
3073       pm = (struct PutMessage*) &reply[1];
3074       pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
3075       pm->header.size = htons (msize);
3076       pm->type = htonl (prq->type);
3077       pm->expiration = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining (prq->expiration));
3078       pr->replies_pending = reply;
3079       memcpy (&reply[1], prq->data, prq->size);
3080       if (pr->th != NULL)
3081         return GNUNET_YES;
3082       pr->th = GNUNET_SERVER_notify_transmit_ready (pr->client,
3083                                                     msize,
3084                                                     GNUNET_TIME_UNIT_FOREVER_REL,
3085                                                     &transmit_result,
3086                                                     pr);
3087       if (pr->th == NULL)
3088         {
3089           // FIXME: need to try again later (not much
3090           // to do here specifically, but we need to
3091           // check somewhere else to handle this case!)
3092         }
3093     }
3094   // FIXME: implement hot-path routing statistics keeping!
3095   return GNUNET_YES;
3096 }
3097
3098
3099 /**
3100  * Check if the given KBlock is well-formed.
3101  *
3102  * @param kb the kblock data (or at least "dsize" bytes claiming to be one)
3103  * @param dsize size of "kb" in bytes; check for < sizeof(struct KBlock)!
3104  * @param query where to store the query that this block answers
3105  * @return GNUNET_OK if this is actually a well-formed KBlock
3106  */
3107 static int
3108 check_kblock (const struct KBlock *kb,
3109               size_t dsize,
3110               GNUNET_HashCode *query)
3111 {
3112   if (dsize < sizeof (struct KBlock))
3113     {
3114       GNUNET_break_op (0);
3115       return GNUNET_SYSERR;
3116     }
3117   if (dsize - sizeof (struct KBlock) !=
3118       ntohs (kb->purpose.size)
3119       - sizeof (struct GNUNET_CRYPTO_RsaSignaturePurpose)
3120       - sizeof(struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) )
3121     {
3122       GNUNET_break_op (0);
3123       return GNUNET_SYSERR;
3124     }
3125   if (GNUNET_OK !=
3126       GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_KBLOCK,
3127                                 &kb->purpose,
3128                                 &kb->signature,
3129                                 &kb->keyspace))
3130     {
3131       GNUNET_break_op (0);
3132       return GNUNET_SYSERR;
3133     }
3134   if (query != NULL)
3135     GNUNET_CRYPTO_hash (&kb->keyspace,
3136                         sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
3137                         query);
3138   return GNUNET_OK;
3139 }
3140
3141
3142 /**
3143  * Check if the given SBlock is well-formed.
3144  *
3145  * @param sb the sblock data (or at least "dsize" bytes claiming to be one)
3146  * @param dsize size of "kb" in bytes; check for < sizeof(struct SBlock)!
3147  * @param query where to store the query that this block answers
3148  * @param namespace where to store the namespace that this block belongs to
3149  * @return GNUNET_OK if this is actually a well-formed SBlock
3150  */
3151 static int
3152 check_sblock (const struct SBlock *sb,
3153               size_t dsize,
3154               GNUNET_HashCode *query,
3155               GNUNET_HashCode *namespace)
3156 {
3157   if (dsize < sizeof (struct SBlock))
3158     {
3159       GNUNET_break_op (0);
3160       return GNUNET_SYSERR;
3161     }
3162   if (dsize !=
3163       ntohs (sb->purpose.size) + sizeof (struct GNUNET_CRYPTO_RsaSignature))
3164     {
3165       GNUNET_break_op (0);
3166       return GNUNET_SYSERR;
3167     }
3168   if (GNUNET_OK !=
3169       GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_SBLOCK,
3170                                 &sb->purpose,
3171                                 &sb->signature,
3172                                 &sb->subspace))
3173     {
3174       GNUNET_break_op (0);
3175       return GNUNET_SYSERR;
3176     }
3177   if (query != NULL)
3178     *query = sb->identifier;
3179   if (namespace != NULL)
3180     GNUNET_CRYPTO_hash (&sb->subspace,
3181                         sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
3182                         namespace);
3183   return GNUNET_OK;
3184 }
3185
3186
3187 /**
3188  * Handle P2P "PUT" request.
3189  *
3190  * @param cls closure, always NULL
3191  * @param other the other peer involved (sender or receiver, NULL
3192  *        for loopback messages where we are both sender and receiver)
3193  * @param message the actual message
3194  * @return GNUNET_OK to keep the connection open,
3195  *         GNUNET_SYSERR to close it (signal serious error)
3196  */
3197 static int
3198 handle_p2p_put (void *cls,
3199                 const struct GNUNET_PeerIdentity *other,
3200                 const struct GNUNET_MessageHeader *message)
3201 {
3202   const struct PutMessage *put;
3203   uint16_t msize;
3204   size_t dsize;
3205   uint32_t type;
3206   struct GNUNET_TIME_Absolute expiration;
3207   GNUNET_HashCode query;
3208   struct ProcessReplyClosure prq;
3209
3210   msize = ntohs (message->size);
3211   if (msize < sizeof (struct PutMessage))
3212     {
3213       GNUNET_break_op(0);
3214       return GNUNET_SYSERR;
3215     }
3216   put = (const struct PutMessage*) message;
3217   dsize = msize - sizeof (struct PutMessage);
3218   type = ntohl (put->type);
3219   expiration = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_relative_ntoh (put->expiration));
3220
3221   /* first, validate! */
3222   switch (type)
3223     {
3224     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
3225     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
3226       GNUNET_CRYPTO_hash (&put[1], dsize, &query);
3227       break;
3228     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
3229       if (GNUNET_OK !=
3230           check_kblock ((const struct KBlock*) &put[1],
3231                         dsize,
3232                         &query))
3233         return GNUNET_SYSERR;
3234       break;
3235     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
3236       if (GNUNET_OK !=
3237           check_sblock ((const struct SBlock*) &put[1],
3238                         dsize,
3239                         &query,
3240                         &prq.namespace))
3241         return GNUNET_SYSERR;
3242       break;
3243     case GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK:
3244       // FIXME -- validate SKBLOCK!
3245       GNUNET_break (0);
3246       return GNUNET_OK;
3247     default:
3248       /* unknown block type */
3249       GNUNET_break_op (0);
3250       return GNUNET_SYSERR;
3251     }
3252
3253   /* now, lookup 'query' */
3254   prq.data = (const void*) &put[1];
3255   prq.size = dsize;
3256   prq.type = type;
3257   prq.expiration = expiration;
3258   prq.priority = 0;
3259   GNUNET_CONTAINER_multihashmap_get_multiple (requests_by_query,
3260                                               &query,
3261                                               &process_reply,
3262                                               &prq);
3263   // FIXME: if migration is on and load is low,
3264   // queue to store data in datastore;
3265   // use "prq.priority" for that!
3266   return GNUNET_OK;
3267 }
3268
3269
3270 /**
3271  * List of handlers for P2P messages
3272  * that we care about.
3273  */
3274 static struct GNUNET_CORE_MessageHandler p2p_handlers[] =
3275   {
3276     { &handle_p2p_get,
3277       GNUNET_MESSAGE_TYPE_FS_GET, 0 },
3278     { &handle_p2p_put,
3279       GNUNET_MESSAGE_TYPE_FS_PUT, 0 },
3280     { NULL, 0, 0 }
3281   };
3282
3283
3284 /**
3285  * Process fs requests.
3286  *
3287  * @param cls closure
3288  * @param s scheduler to use
3289  * @param server the initialized server
3290  * @param c configuration to use
3291  */
3292 static void
3293 run (void *cls,
3294      struct GNUNET_SCHEDULER_Handle *s,
3295      struct GNUNET_SERVER_Handle *server,
3296      const struct GNUNET_CONFIGURATION_Handle *c)
3297 {
3298   sched = s;
3299   cfg = c;
3300
3301   ifm = GNUNET_CONTAINER_multihashmap_create (128);
3302   requests_by_query = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
3303   requests_by_peer = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
3304   connected_peers = GNUNET_CONTAINER_multihashmap_create (64);
3305   requests_by_expiration = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
3306   read_index_list ();
3307   dsh = GNUNET_DATASTORE_connect (cfg,
3308                                   sched);
3309   core = GNUNET_CORE_connect (sched,
3310                               cfg,
3311                               GNUNET_TIME_UNIT_FOREVER_REL,
3312                               NULL,
3313                               NULL,
3314                               &peer_connect_handler,
3315                               &peer_disconnect_handler,
3316                               NULL,
3317                               NULL, GNUNET_NO,
3318                               NULL, GNUNET_NO,
3319                               p2p_handlers);
3320
3321   GNUNET_SERVER_disconnect_notify (server,
3322                                    &handle_client_disconnect,
3323                                    NULL);
3324   GNUNET_SERVER_add_handlers (server, handlers);
3325   GNUNET_SCHEDULER_add_delayed (sched,
3326                                 GNUNET_TIME_UNIT_FOREVER_REL,
3327                                 &shutdown_task,
3328                                 NULL);
3329   if (NULL == dsh)
3330     {
3331       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
3332                   _("Failed to connect to `%s' service.\n"),
3333                   "datastore");
3334       GNUNET_SCHEDULER_shutdown (sched);
3335       return;
3336     }
3337   if (NULL == core)
3338     {
3339       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
3340                   _("Failed to connect to `%s' service.\n"),
3341                   "core");
3342       GNUNET_SCHEDULER_shutdown (sched);
3343       return;
3344     }
3345 }
3346
3347
3348 /**
3349  * The main function for the fs service.
3350  *
3351  * @param argc number of arguments from the command line
3352  * @param argv command line arguments
3353  * @return 0 ok, 1 on error
3354  */
3355 int
3356 main (int argc, char *const *argv)
3357 {
3358   return (GNUNET_OK ==
3359           GNUNET_SERVICE_run (argc,
3360                               argv,
3361                               "fs",
3362                               GNUNET_SERVICE_OPTION_NONE,
3363                               &run, NULL)) ? 0 : 1;
3364 }
3365
3366 /* end of gnunet-service-fs.c */